基于 n8n 与 Web Unlocker 的自动化资讯采集推送实战 | 极客日志

基于 n8n 与 Web Unlocker 的自动化资讯采集推送实战 | 极客日志

# 忽略 SSL 警告
import warnings
warnings.filterwarnings('ignore', message='Unverified HTTPS request')

# 您的 Bright Data 凭证（根据配置页面更新）
proxy_host = "brd.superproxy.io"
proxy_port = "33335"
username = "brd-customer-hl_d8430cdd-zone-keyanshuju1"
password = "yj4qaotb7edx"

# 代理设置 - 支持多种格式
proxy_url = f"{proxy_host}:{proxy_port}"
proxy_auth = f"{username}:{password}"

# 主要代理配置
proxies = {
    "http": f"http://{proxy_auth}@{proxy_url}",
    "https": f"http://{proxy_auth}@{proxy_url}"
}

def parse_36kr_news(html_content, url):
    """解析 36 氪新闻内容"""
    soup = BeautifulSoup(html_content, 'html.parser')
    articles = []
    # 多种可能的新闻项选择器
    selectors = [
        'a.article-item-title', '.article-item-title', 'a[class*="title"]',
        '.news-item a', '.article-title a', 'h3 a', 'h2 a',
        '.item-title a', 'a[href*="/p/"]', 'a[href*="36kr.com"]'
    ]
    print(f"正在解析页面：{url}")
    for selector in selectors:
        try:
            elements = soup.select(selector)
            if elements:
                print(f"使用选择器 '{selector}' 找到 {len(elements)} 个元素")
                for element in elements:
                    title = element.get_text(strip=True)
                    link = element.get("href")
                    if title and len(title) > 5:
                        # 过滤掉太短的标题
                        # 处理相对链接
                        if link:
                            if link.startswith("/"):
                                full_link = f"https://36kr.com{link}"
                            elif not link.startswith("http"):
                                full_link = f"https://36kr.com/{link}"
                            else:
                                full_link = link
                        else:
                            full_link = "链接不可用"
                        articles.append({'title': title, 'link': full_link, 'source': url})
                if articles:
                    break
        except Exception as e:
            print(f"选择器 '{selector}' 解析失败：{e}")

{
  "role": "system",
  "content": "你是一名资深 IT 行业媒体编辑，请对以下新闻生成简洁明了的摘要，要求语言客观、中立、不重复标题，字数控制在 150 字以内。"
}

{
  "role": "user",
  "content": "标题：小鹏'瘦身'，蔚来学得会吗？\n链接：https://36kr.com/p/3305015892810496"
}

{
  "choices": [{
    "message": {
      "role": "assistant",
      "content": "依靠爆款策略、成本控制和管理重构，小鹏蹚出一条路，朝着盈利更进一步，但还不能'半场开香槟'。同样亟需扭亏的蔚来，试图复制小鹏'逆袭'，战略大方向没有错，但在具体战术上仍有众多待解难题。"
    }
  }]
}

{
  "msgtype": "markdown",
  "markdown": {
    "content": "【{{$json[\"title\"]}}】\n> {{$json[\"summary\"]}}\n\n[阅读全文]({{$json["url"]}})"
  }
}

def wxPusher_send_message_post(message):
    import requests
    import json
    app_token = 'xxx'
    UID = 'xxx'
    data = {
        "appToken": app_token,
        "content": message,
        "summary": message,
        "contentType": 1,
        "topicIds": [123],
        "uids": [UID],
        "url": "https://wxpusher.zjiecode.com",
        "verifyPay": False
    }
    json_data = json.dumps(data)
    url = "https://wxpusher.zjiecode.com/api/send/message"
    headers = {'Content-Type': "application/json",}
    request = requests.post(url, data=json_data, headers=headers)
    print("request:", request)
    return request

wxPusher_send_message_post('好的')