From 626f83fb9cfc2f99bc567affb12a656ca07fb0b6 Mon Sep 17 00:00:00 2001 From: Steven Date: Fri, 6 Nov 2026 00:54:21 +0800 Subject: [PATCH] =?UTF-8?q?feat(wechat):=20=E5=BE=AE=E4=BF=A1=E9=87=8C?= =?UTF-8?q?=E5=8F=91=E4=B8=80=E6=9D=A1=E9=93=BE=E6=8E=A5=EF=BC=8C=E6=AD=A3?= =?UTF-8?q?=E6=96=87=E5=92=8C=E9=85=8D=E5=9B=BE=E6=8A=93=E4=B8=8B=E6=9D=A5?= =?UTF-8?q?=E5=AD=98=E6=88=90=E7=9F=A5=E8=AF=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 公众号的文章也能抓：走 mp.weixin.qq.com 的 DOM（#js_content）， 按 data-src 取图，不靠通用抽取器——它会把公众号的图全丢掉。 正文里的外链图下载进我们自己的存储（最多 12 张、单张 8 MB）， 地址改成 /attachments/{id}/raw?签名：公众号那边一换图，外链就是一排红叉。 抓取入口谁都能触发，所以只允许公网地址（内网段、回环、链路本地、 保留段、解析不出的域名一律拒绝），否则就是个 SSRF 跳板。 抓不到要说人话："这个链接没抓下来（原因）。想直接记成任务的话， 加「任务：」前缀再发一次。" 反爬验证页、SPA 空壳都走这条路。 - 新增 services/article.py：extract_url / parse / fetch / store_images / compose_body - wechat_intake：整条消息是链接时走抓取，落知识草稿，回「已存成知识草稿：标题」 - 依赖 trafilatura==2.0.0（装法见 requirements.txt 注释：必须逐个装再 --no-deps） - 测试 9 例：链接判定、公众号解析、验证页、通用页裁导航、SSRF 拒绝、图片本地化、失败提示 --- 2.Design/API.md | 19 + 2.Design/ROADMAP-M0-M1.md | 19 +- backend/app/services/article.py | 513 ++++++++++++++++++++++++++ backend/app/services/wechat_intake.py | 52 +++ backend/requirements.txt | 12 + backend/tests/test_wechat_article.py | 244 ++++++++++++ web/src/pages/Integrations.tsx | 1 + 7 files changed, 859 insertions(+), 1 deletion(-) create mode 100644 backend/app/services/article.py create mode 100644 backend/tests/test_wechat_article.py diff --git a/2.Design/API.md b/2.Design/API.md index 114a709..f3c8fdd 100644 --- a/2.Design/API.md +++ b/2.Design/API.md @@ -750,6 +750,25 @@ Query： 图片／文件／视频目前只在 `bot_message.raw` 里留痕，正文不搬；语音如果服务端给了转写就用转写文本。 +### 6.5.2.1 发一条链接：抓正文存成知识 + +整个消息就是一条 `http://` / `https://` 链接时（前后带标点不算），不当任务，而是去把文章抓下来： + +| 环节 | 行为 | +|---|---| +| 判断 | `extract_url()`：去掉首尾标点后只剩一个链接；带了别的话（"看下 https://… 这个"）仍按任务处理 | +| 抓取 | `httpx` 取页面（25 秒超时、8 MB 上限），`trafilatura` 抽正文；标题取 `og:title`／`

`／`` | +| 公众号 | `mp.weixin.qq.com` 单独走 DOM：标题 `#activity-name`、作者 `#js_name`、正文 `#js_content`，图片取 `data-src`（通用抽取器会把公众号的图丢光） | +| 清洗 | 导航/页脚按行边界裁掉（`_trim_chrome`）；`trafilatura` 结果太短（<120 字）就退到容器打分兜底 | +| 配图 | 正文里的外链图下载进我们自己的存储，最多 12 张、单张 8 MB，正文改写成 `/attachments/{id}/raw?签名`——公众号一换图，外链就是一排红叉 | +| 落库 | 知识草稿（`status="draft"`，`doc_type="note"`），正文首行是 `> 原文：<url>` | +| 回话 | `已存成知识草稿：标题`（存了图再加「，连图 N 张」） | +| 失败 | `这个链接没抓下来（原因）。想直接记成任务的话，加「任务：」前缀再发一次。` | + +安全：抓取入口对任何能给机器人发消息的人开放，所以**只允许公网地址**——`127.0.0.1`、内网段、 +链路本地、保留地址、解析不出来的域名一律拒绝（`ArticleError`），否则就是个 SSRF 跳板。 +反爬/需登录的页面（微信"环境异常"验证页、Cloudflare 302、SPA 空壳）会回上面那句失败提示，不静默吞掉。 + ### 6.5.3 错误码 | 场景 | code | HTTP | diff --git a/2.Design/ROADMAP-M0-M1.md b/2.Design/ROADMAP-M0-M1.md index 87ce332..7696b71 100644 --- a/2.Design/ROADMAP-M0-M1.md +++ b/2.Design/ROADMAP-M0-M1.md @@ -840,7 +840,24 @@ **教训：接入第三方协议，先去已有实现里找协议，别从直觉倒推。** 那一版的迁移 `0012` 保留 （`bot_channel` / `bot_message` 两张表还在用），`token_hash` 列留着但不再使用。 -### 13.8 还没做 +### 13.8 发一条链接，抓成知识 + +需求原话：「如果发送给机器人的是一个链接，那么就需要把链接内的文章内容给抓取下来，然后保存为知识； +就算是微信公众号的链接也需要抓取下来」。 + +- 选型 `trafilatura==2.0.0`（中文正文抽取最稳）+ `lxml`；公众号单独走 DOM（`#js_content`、图片在 + `data-src`），因为通用抽取器会把公众号的图全丢掉。 +- 抓下来的图下载进自己的存储（最多 12 张、单张 8 MB），正文改写成本地 `raw?签名` 地址—— + 外链图在公众号侧一换就全红叉。 +- 只允许公网地址：谁都能给机器人发消息，不拦住内网就是 SSRF 跳板（`127.0.0.1`、`192.168.x`、链路本地、 + 保留段、解析不出的域名一律拒绝）。 +- 失败要说话：反爬验证页、Cloudflare 302、SPA 空壳统一回「这个链接没抓下来（原因）。想直接记成任务 + 的话，加「任务：」前缀再发一次。」 +- **装依赖的坑**：`pip install trafilatura` 在本机会卡死（依赖解析完就挂住），要逐个装 + `lxml / courlan / htmldate / jusText / lxml_html_clean / …` 再 `pip install --no-deps trafilatura`； + `lxml` 6 起 `lxml.html.clean` 拆成 `lxml_html_clean`，不装就 `ImportError`。 + +### 13.9 还没做 - 只做了"进"（收消息建事项），没做"出"（任务有进展回一条微信通知）。 - 消息不认人：按通道创建者记账；要认人得先做微信身份绑定。 diff --git a/backend/app/services/article.py b/backend/app/services/article.py new file mode 100644 index 0000000..e1a93eb --- /dev/null +++ b/backend/app/services/article.py @@ -0,0 +1,513 @@ +"""把一条链接抓成一篇知识。 + +微信里发一条链接，多半就是想"把这篇文章存下来"。这里负责三件事： + +1. **取正文**：用 trafilatura 的正文抽取（开源社区里在中文站上表现最稳的一个）， + 微信公众号单独走一条路——它的正文在 `#js_content` 里、图片放在 `data-src`， + 通用抽取器会把图丢掉。 +2. **留来源**：正文开头记一行原文地址与作者，日后要核对时不用猜。 +3. **图片转存**：文章里的图片不是躺在我们的存储里，公众号一改就变红叉。 + 所以抓下来的图逐张存成附件，正文里的地址换成我们自己的签名直链。 + +抓不下来不硬扛：调用方拿到 `ArticleError` 就把原链接当普通文本存成任务， +用户至少不会丢东西。 +""" + +from __future__ import annotations + +import ipaddress +import re +import socket +import uuid +from dataclasses import dataclass, field +from urllib.parse import urlparse + +import httpx +from sqlalchemy.ext.asyncio import AsyncSession +from starlette.concurrency import run_in_threadpool + +# 先让 app.models 加载完：它在文件末尾反向导入各域的租户表清单， +# 直接 import models_attachment 会撞上"部分初始化"的循环导入。 +from .. import models as _models # noqa: F401 +from ..config import get_settings +from ..models_attachment import Attachment +from . import signing, storage + +# 一条消息里只认第一个链接；带别的字就当普通任务（"看下 https://… 这个" 不该被吞成收藏） +URL_RE = re.compile(r"https?://[^\s<>\"'）)】。，；]+", re.I) + +MAX_ARTICLES_BYTES = 8 * 1024 * 1024 +MAX_IMAGES = 12 +MAX_IMAGE_BYTES = 8 * 1024 * 1024 +MAX_BODY_CHARS = 200_000 +FETCH_TIMEOUT = 25.0 + +DESKTOP_UA = ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" + " (KHTML, like Gecko) Chrome/126.0 Safari/537.36" +) +# 公众号对"非微信浏览器"会压缩内容、图片懒加载，用微信 UA 拿到的才是完整一篇。 +WECHAT_UA = ( + "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15" + " (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.44(0x18002c2e)" + " NetType/WIFI Language/zh_CN" +) + +IMAGE_EXTS = {"jpeg": "jpg", "jpg": "jpg", "png": "png", "gif": "gif", "webp": "webp"} + + +class ArticleError(RuntimeError): + """抓取/解析没成功。调用方据此决定降级方式。""" + + +@dataclass +class Article: + url: str + title: str + body_md: str + site: str = "" + author: str = "" + images: list[str] = field(default_factory=list) + + +def extract_url(text: str) -> str | None: + """这一整句话就是个链接吗？ + + 允许链接前后有点标点、空格（"看下：https://x" 不算）；带了别的内容就当普通文本， + 免得用户随口提一句网址，心情是"记条任务"，结果被存成一篇文章。 + """ + stripped = text.strip() + match = URL_RE.search(stripped) + if not match: + return None + rest = (stripped[: match.start()] + stripped[match.end() :]).strip() + rest = rest.strip(" \t\r\n:：，。；;、!！?？（）()【】[]<>「」") + if rest: + return None + return match.group(0) + + +def is_wechat_article(url: str) -> bool: + host = (urlparse(url).hostname or "").lower() + return host == "mp.weixin.qq.com" + + +async def _assert_public(url: str) -> None: + """只抓公网地址。 + + 这个入口能由工作区以外的人（微信里的任何人）触发，放任它就是一台 SSRF 跳板： + `http://127.0.0.1:8200/version`、`http://169.254.169.254/` 之类都能被打。 + """ + parsed = urlparse(url) + if parsed.scheme not in ("http", "https"): + raise ArticleError("只支持 http/https 链接") + host = parsed.hostname or "" + if not host: + raise ArticleError("链接里没有主机名") + try: + infos = await _resolve(host) + except OSError as exc: # 解析不出来就别猜 + raise ArticleError(f"解析不了这个域名：{exc}") from exc + for _family, _type, _proto, _canon, sockaddr in infos: + ip = ipaddress.ip_address(sockaddr[0]) + if ( + ip.is_private + or ip.is_loopback + or ip.is_link_local + or ip.is_reserved + or ip.is_multicast + or ip.is_unspecified + ): + raise ArticleError("这个地址在内网，不抓") + + +async def _resolve(host: str): + import asyncio + + def lookup(): + return socket.getaddrinfo(host, None) + + return await asyncio.to_thread(lookup) + + +def _pick_ua(url: str) -> str: + return WECHAT_UA if is_wechat_article(url) else DESKTOP_UA + + +def _looks_blocked(html: str) -> bool: + probes = ( + "环境异常", + "完成验证后即可继续访问", + "请在微信客户端打开链接", + "该内容已被发布者删除", + "此内容因违规无法查看", + ) + return any(p in html for p in probes) + + +def _clean_text(value: str) -> str: + return re.sub(r"[ \t\u00a0]+", " ", (value or "").replace("\r", "")).strip() + + +def _wechat_meta(html: str) -> tuple[str, str, list[tuple[str, str]]]: + """从公众号页面里取 (标题, 作者/公众号名, [(图片地址, alt)]) 与正文节点。""" + from lxml import html as lxml_html + + tree = lxml_html.fromstring(html) + title = "" + for xpath in ( + '//*[@id="activity-name"]//text()', + '//meta[@property="og:title"]/@content', + "//title/text()", + ): + parts = [_clean_text(t) for t in tree.xpath(xpath)] + title = " ".join(p for p in parts if p) + if title: + break + author = "" + for xpath in ( + '//*[@id="js_name"]//text()', + '//meta[@name="author"]/@content', + '//*[@id="meta_content"]//*[@class="rich_media_meta_text"]//text()', + ): + parts = [_clean_text(t) for t in tree.xpath(xpath)] + author = " ".join(p for p in parts if p) + if author: + break + return title, author, tree + + +def _wechat_body(html: str) -> str: + """`#js_content` → markdown。 + + 公众号正文是一堆嵌套 `section` 拼版，段落文字常被拆成多个 `span`； + 这里只取"没有块级子节点"的元素，避免同一个句子被输出好几遍。 + """ + from lxml import html as lxml_html + + tree = lxml_html.fromstring(html) + nodes = tree.xpath('//*[@id="js_content"]') + if not nodes: + return "" + content = nodes[0] + + block_tags = { + "p", + "section", + "div", + "blockquote", + "li", + "h1", + "h2", + "h3", + "h4", + "h5", + "h6", + "pre", + } + lines: list[str] = [] + for node in content.iter(): + tag = node.tag if isinstance(node.tag, str) else "" + if tag == "img": + src = node.get("data-src") or node.get("src") or "" + if src.startswith("http"): + lines.append(f"![]({src})") + continue + if tag == "br": + lines.append("") + continue + if tag not in block_tags: + continue + # 还有块级子孙的容器交给子节点输出，自己不出文字，否则整段重复 + if any( + (child.tag if isinstance(child.tag, str) else "") in block_tags + for child in node.iterdescendants() + ): + continue + text = _clean_text("".join(node.itertext())) + if not text: + continue + if tag.startswith("h") and len(tag) == 2: + text = f"## {text}" + elif tag == "li": + text = f"- {text}" + elif tag == "blockquote": + text = f"> {text}" + lines.append(text) + + # 相邻空行压成一个，段间留空行 + out: list[str] = [] + for line in lines: + if not line and (not out or not out[-1]): + continue + out.append(line) + return "\n\n".join(s.strip() for s in "\n".join(out).split("\n\n") if s.strip()) + + +def _block_markdown(html: str) -> str: + """兜底的正文提取：找出"装着最多段文字"的那个容器，再按段落拼成 markdown。 + + 为什么还要这一手：trafilatura 在结构清楚的资讯站上很准，但遇到"整页一个 div、 + 导航和正文混在一起"的老站点（政府站、企业站）时会连导航一起端出来。 + 这里用最笨但稳的判据——哪个容器里的 <p> 文字最多，正文就在哪个容器里。 + """ + from lxml import html as lxml_html + + tree = lxml_html.fromstring(html) + best = None + best_len = 0 + for node in tree.iter("div", "article", "section", "main", "td"): + total = sum(len("".join(p.itertext()).strip()) for p in node.iter("p")) + if total > best_len: + best, best_len = node, total + if best is None or best_len < 200: + return "" + + parts: list[str] = [] + for node in best.iter(): + tag = node.tag if isinstance(node.tag, str) else "" + if tag == "img": + src = node.get("data-src") or node.get("src") or "" + if src.startswith("http"): + parts.append(f"![]({src})") + continue + if tag not in {"p", "h1", "h2", "h3", "h4", "blockquote", "li"}: + continue + text = _clean_text("".join(node.itertext())) + if not text: + continue + if tag.startswith("h") and len(tag) == 2: + text = f"## {text}" + elif tag == "li": + text = f"- {text}" + elif tag == "blockquote": + text = f"> {text}" + parts.append(text) + return "\n\n".join(parts).strip() + + +def _body_score(body: str) -> int: + """给一份抽取结果打分：有多少段"像人写的段落"。 + + 只用来在两个候选之间挑一个，不需要绝对准确——比较的是同一篇文章的两种切法。 + """ + lines = [line.strip() for line in body.splitlines() if line.strip()] + return sum(1 for line in lines if len(line) >= 40) + + +def _trim_chrome(body: str) -> str: + """去掉抽取器一并带出来的导航与页脚。 + + 按行看，找出第一行和最后一行"像正文的"（够长、或者有句读），把两头那些短标签 + （"Java教程"、"Sign In"）切掉。按行而不是按段落，是因为导航往往是一行一个、 + 中间不空行，按段落切会把它和正文粘成一整块，判不出边界。 + """ + lines = body.splitlines() + if not lines: + return "" + + def substantive(line: str) -> bool: + text = line.strip() + if not text or text.startswith(("![", "> ")): + return False + plain = text.lstrip("#-* 　") + return len(plain) >= 20 or bool(re.search(r"[。！？；]", plain)) + + first = next((i for i, line in enumerate(lines) if substantive(line)), None) + if first is None: + return body.strip() + last = max(i for i, line in enumerate(lines) if substantive(line)) + # 图片通常紧贴在正文前后，边界外紧邻的图保留 + while first > 0 and lines[first - 1].strip().startswith("!["): + first -= 1 + while last + 1 < len(lines) and lines[last + 1].strip().startswith("!["): + last += 1 + return "\n".join(lines[first : last + 1]).strip() + + +def _generic_article(html: str, url: str) -> Article: + """非公众号：交给 trafilatura，它在正文/噪声的判别上比手搓规则靠谱得多。 + + 先按 `favor_precision` 抠最干净的一块；它经常因为"看着都不像正文"而返回空， + 再退回默认召回模式。两头再削一刀导航。 + """ + import trafilatura + + metadata = trafilatura.extract_metadata(html, default_url=url) + common = { + "url": url, + "output_format": "markdown", + "include_images": True, + "include_links": True, + "include_formatting": True, + } + body = trafilatura.extract(html, favor_precision=True, **common) or "" + if len(body.strip()) < 120: + body = trafilatura.extract(html, **common) or "" + if not body.strip(): + # 再退一步：trafilatura 认不出结构时，退到纯文本 + body = trafilatura.extract(html, url=url, output_format="txt") or "" + body = _trim_chrome(body) + + # 老站点（政府站、企业站）常把导航和正文塞进同一个 div，trafilatura 会连导航一起端出来。 + # 用"哪个容器里段落文字最多"另取一份，两份按"像样段落的条数"挑更干净的那个。 + block = _trim_chrome(_block_markdown(html)) + if _body_score(block) > _body_score(body): + body = block + if not body.strip(): + raise ArticleError("这个页面里没找到正文（可能是动态渲染或需要登录）") + title = _clean_text(getattr(metadata, "title", "") or "") + return Article( + url=url, + title=title, + body_md=body.strip(), + site=_clean_text(getattr(metadata, "sitename", "") or ""), + author=_clean_text(getattr(metadata, "author", "") or ""), + ) + + +def parse_article(html: str, url: str) -> Article: + if _looks_blocked(html): + raise ArticleError("微信把这篇拦下来了（需要人工验证或链接已失效）") + if is_wechat_article(url): + title, author, _tree = _wechat_meta(html) + body = _wechat_body(html) + if not body.strip(): + # 公众号偶尔会换结构，退回通用抽取，宁可少图也别丢正文 + article = _generic_article(html, url) + article.author = author or article.author + return article + return Article( + url=url, + title=title, + body_md=body.strip(), + site="微信公众号", + author=author, + ) + return _generic_article(html, url) + + +async def fetch_article(url: str) -> Article: + await _assert_public(url) + headers = { + "User-Agent": _pick_ua(url), + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.6", + } + try: + async with httpx.AsyncClient( + follow_redirects=True, timeout=FETCH_TIMEOUT, headers=headers + ) as client: + resp = await client.get(url) + resp.raise_for_status() + raw = resp.content + except httpx.HTTPStatusError as exc: + raise ArticleError(f"对方站点返回 {exc.response.status_code}") from exc + except httpx.HTTPError as exc: + raise ArticleError(f"取不到这个页面：{type(exc).__name__}") from exc + if len(raw) > MAX_ARTICLES_BYTES: + raise ArticleError("页面太大了，不像是篇文章") + html = raw.decode(resp.encoding or "utf-8", errors="replace") + article = parse_article(html, url) + article.title = (article.title or urlparse(url).hostname or "未命名")[:180] + return article + + +async def _download_image(client: httpx.AsyncClient, url: str) -> tuple[bytes, str] | None: + try: + resp = await client.get(url) + resp.raise_for_status() + except httpx.HTTPError: + return None + if len(resp.content) > MAX_IMAGE_BYTES: + return None + ctype = (resp.headers.get("content-type") or "").split(";")[0].strip().lower() + if not ctype.startswith("image/"): + # 公众号图片地址带 wx_fmt 参数，content-type 偶尔是 octet-stream，按扩展名兜一下 + fmt = urlparse(url).query or "" + m = re.search(r"wx_fmt=(\w+)", fmt) + guess = IMAGE_EXTS.get((m.group(1) if m else "").lower(), "") + if not guess: + return None + ctype = f"image/{'jpeg' if guess == 'jpg' else guess}" + return resp.content, ctype + + +async def store_images( + session: AsyncSession, + *, + ws_id: uuid.UUID, + item_id: uuid.UUID, + uploader_id: uuid.UUID, + body_md: str, + limit: int = MAX_IMAGES, +) -> tuple[str, list[Attachment]]: + """把正文里引用的远程图片存成我们自己的附件，返回改写后的正文与附件行。 + + 失败就当没这张图：一张打不开的配图不该让整篇收藏失败。 + """ + from io import BytesIO + + urls = re.findall(r"!\[[^\]]*\]\((https?://[^)\s]+)\)", body_md) + unique: list[str] = [] + for url in urls: + if url not in unique: + unique.append(url) + if len(unique) >= limit: + break + if not unique: + return body_md, [] + + settings = get_settings() + saved: list[Attachment] = [] + mapping: dict[str, str] = {} + headers = {"User-Agent": DESKTOP_UA, "Referer": "https://mp.weixin.qq.com/"} + async with httpx.AsyncClient( + follow_redirects=True, timeout=FETCH_TIMEOUT, headers=headers + ) as client: + for index, remote in enumerate(unique, start=1): + got = await _download_image(client, remote) + if got is None: + continue + payload, content_type = got + ext = IMAGE_EXTS.get(content_type.split("/")[-1], "jpg") + # 落盘是阻塞 IO：一篇公众号十几张图，别把事件循环按住 + stored = await run_in_threadpool( + storage.save, BytesIO(payload), workspace_id=ws_id, item_id=item_id + ) + if stored.size > settings.max_upload_mb * 1024 * 1024: + storage.remove(stored.key) + continue + row = Attachment( + workspace_id=ws_id, + item_id=item_id, + uploader_id=uploader_id, + filename=f"图{index}.{ext}", + content_type=content_type[:120], + size_bytes=stored.size, + sha256=stored.sha256, + storage_key=stored.key, + ) + session.add(row) + await session.flush() + mapping[remote] = ( + f"{settings.api_prefix}/workspaces/{ws_id}" + f"/attachments/{row.id}/raw?{signing.sign(ws_id, row.id)}" + ) + saved.append(row) + + for remote, local in mapping.items(): + body_md = body_md.replace(remote, local) + return body_md, saved + + +def compose_body(article: Article) -> str: + """正文 = 来源行 + 原文。来源要留在最上面，日后核对不用翻聊天记录。""" + bits = [f"> 原文：<{article.url}>"] + if article.site or article.author: + who = " · ".join(x for x in (article.site, article.author) if x) + bits[0] += f"（{who}）" + body = "\n".join(bits) + "\n\n" + article.body_md + if len(body) > MAX_BODY_CHARS: + body = body[:MAX_BODY_CHARS] + "\n\n…（原文太长，后面的内容没有保存）" + return body diff --git a/backend/app/services/wechat_intake.py b/backend/app/services/wechat_intake.py index 310b32c..72f6f7e 100644 --- a/backend/app/services/wechat_intake.py +++ b/backend/app/services/wechat_intake.py @@ -17,6 +17,7 @@ from ..models import Member, Workspace from ..models_integrations import BotChannel from ..models_knowledge import KnowledgeDoc from ..models_task import Item +from . import article as article_service MAX_TITLE = 180 MAX_BODY = 20000 @@ -30,6 +31,7 @@ HELP_TEXT = ( "· 直接说　→ 记成任务\n" "· 知识：内容　→ 记成知识草稿（也可以写「笔记：」）\n" "· 任务：内容　→ 明确记成任务\n" + "· 一条链接　→ 抓正文存成知识（公众号也行）\n" "首行当标题，后面几行当正文；标题里带 #标签 会自动加上。" ) @@ -177,6 +179,53 @@ def reply_text(kind: str, item: Item) -> str: return f"已保存任务：{item.title}" +async def _collect_article( + session: AsyncSession, + channel: BotChannel, + url: str, +) -> dict[str, Any]: + """一条链接 → 一篇知识草稿（正文 + 转存后的图）。 + + 抓不到就如实说，并把原链接留着：用户多半还会想手动点开看看， + 回一句"失败了"再让他重发一遍是最糟的体验。 + """ + creator = await creator_id(session, channel) + try: + fetched = await article_service.fetch_article(url) + except article_service.ArticleError as exc: + return { + "action": "ignored", + "item_id": None, + "reply": f"这个链接没抓下来（{exc}）。想直接记成任务的话，加「任务：」前缀再发一次。", + } + + item = await create_knowledge( + session, + ws_id=channel.workspace_id, + creator=creator, + title=fetched.title, + body=article_service.compose_body(fetched), + tags=[], + ) + body_md, saved = await article_service.store_images( + session, + ws_id=channel.workspace_id, + item_id=item.id, + uploader_id=creator, + body_md=item.body_md, + ) + item.body_md = body_md + await session.flush() + + # 不写"已收藏"，写清楚抓到了什么：有没有正文、有没有图，一眼能判断值不值得去补。 + tail = f"，连图 {len(saved)} 张" if saved else "" + return { + "action": "knowledge", + "item_id": item.id, + "reply": f"已存成知识草稿：{item.title}{tail}", + } + + async def handle_text( session: AsyncSession, channel: BotChannel, @@ -188,6 +237,9 @@ async def handle_text( kind, body = split_command(text.strip()) if kind == "help": return {"action": "help", "item_id": None, "reply": HELP_TEXT} + link = article_service.extract_url(body) + if link: + return await _collect_article(session, channel, link) body, tags = split_tags(body) title, rest = split_title(body) if not title: diff --git a/backend/requirements.txt b/backend/requirements.txt index 919730e..8ea45b4 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -21,4 +21,16 @@ httpx==0.28.1 # 微信登录态加密（Fernet）与二维码渲染 cryptography==50.0.2 qrcode==8.2 +# 微信里发来链接时抓正文：trafilatura 的正文抽取在中文站上最稳，lxml 是它的解析底座； +# lxml_html_clean 是 jusText（trafilatura 的回退抽取）还要的模块，lxml 6 起被拆成独立包。 +trafilatura==2.0.0 +lxml==6.1.3 +lxml_html_clean==0.4.6 +courlan==1.4.0 +htmldate==1.11.0 +jusText==3.0.2 +charset-normalizer==3.5.2 +urllib3==2.8.0 +tld==0.13.2 +dateparser==1.5.0 ruff==0.8.4 diff --git a/backend/tests/test_wechat_article.py b/backend/tests/test_wechat_article.py new file mode 100644 index 0000000..a15d463 --- /dev/null +++ b/backend/tests/test_wechat_article.py @@ -0,0 +1,244 @@ +"""微信里发链接 → 抓成知识。 + +不联网：`fetch_article` 与 `store_images` 用 monkeypatch 顶掉，验的是解析规则与落库； +真抓取（trafilatura / 公众号 DOM）在 `parse_article` 的用例里用固定 HTML 覆盖。 +""" + +from __future__ import annotations + +import uuid +from typing import Any + +import pytest +from httpx import AsyncClient +from sqlalchemy import select + +from app.db import SessionFactory, bind_tenant +from app.models_integrations import BotChannel +from app.models_knowledge import KnowledgeDoc +from app.models_task import Item +from app.services import article as article_service +from app.services.wechat_intake import handle_text + +from .conftest import create_workspace, login + +WECHAT_HTML = """ +<html><head> +<meta property="og:title" content="公众号标题兜底"> +</head><body> +<h1 id="activity-name"> 回滚前先停灰度：一次线上事故的复盘 </h1> +<a id="js_name">达道技术</a> +<div id="js_content"> + <section><p>事故发生在周五晚高峰，灰度流量上涨后错误率飙升。</p></section> + <section><p>第一步先停灰度，把流量打回稳定版本。</p> + <p><img data-src="https://mmbiz.qpic.cn/mmbiz_png/abc/640?wx_fmt=png" /></p> + </section> + <section><section><p>第二步核对回滚后的核心指标。</p></section></section> +</div> +<div id="js_pc_qr_code">扫码关注</div> +</body></html> +""" + +BLOCKED_HTML = """ +<html><body><div class="weui-msg"><p>环境异常</p><p>完成验证后即可继续访问</p></div></body></html> +""" + + +def test_extract_url_only_when_message_is_a_link() -> None: + assert article_service.extract_url("https://mp.weixin.qq.com/s/abc") == ( + "https://mp.weixin.qq.com/s/abc" + ) + # 前后有点标点也算，手机上复制链接常带标点 + assert article_service.extract_url(" https://a.com/x 。") == "https://a.com/x" + # 带了别的内容就不是"发链接"，它是想记条任务 + assert article_service.extract_url("看下 https://a.com/x 这个方案") is None + assert article_service.extract_url("明天下午三点开会") is None + + +def test_wechat_article_parses_title_body_and_images() -> None: + got = article_service.parse_article(WECHAT_HTML, "https://mp.weixin.qq.com/s/abc") + + assert got.title == "回滚前先停灰度：一次线上事故的复盘" + assert got.author == "达道技术" + assert got.site == "微信公众号" + assert "灰度流量上涨后错误率飙升" in got.body_md + # 图片取 data-src，不能丢 + assert "https://mmbiz.qpic.cn/mmbiz_png/abc/640?wx_fmt=png" in got.body_md + # 正文不能重复输出（嵌套 section 是最容易重复的地方） + assert got.body_md.count("第一步先停灰度") == 1 + assert got.body_md.count("第二步核对回滚后的核心指标") == 1 + # 页面底部二维码之类的噪声不该进来 + assert "扫码关注" not in got.body_md + + +def test_blocked_wechat_page_is_reported() -> None: + with pytest.raises(article_service.ArticleError): + article_service.parse_article(BLOCKED_HTML, "https://mp.weixin.qq.com/s/abc") + + +def test_generic_page_gets_chrome_trimmed() -> None: + html = """<html><body> + <nav><a>首页</a><a>产品</a><a>关于我们</a></nav> + <article> + <h1>怎么把知识管起来</h1> + <p>先把重复回答的问题记下来，这一步最便宜，也最容易坚持。</p> + <p>第二步才是建目录，目录只服务于找得到，不要为了好看去分二十个类。</p> + </article> + <footer><a>Sign In</a><a>Privacy</a></footer> + </body></html>""" + got = article_service.parse_article(html, "https://example.com/post") + assert "先把重复回答的问题记下来" in got.body_md + assert "Sign In" not in got.body_md + + +def test_compose_body_keeps_source_line() -> None: + got = article_service.Article( + url="https://mp.weixin.qq.com/s/abc", + title="标题", + body_md="正文", + site="微信公众号", + author="达道技术", + ) + body = article_service.compose_body(got) + assert body.startswith("> 原文：<https://mp.weixin.qq.com/s/abc>") + assert "微信公众号 · 达道技术" in body + assert body.endswith("正文") + + +async def test_private_address_is_refused() -> None: + """这个入口谁都能触发，不能当 SSRF 跳板。""" + for url in ("http://127.0.0.1:8200/version", "http://192.168.8.188:8200/api"): + with pytest.raises(article_service.ArticleError): + await article_service.fetch_article(url) + + +async def test_link_message_becomes_knowledge( + client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + user = await login(client) + ws = await create_workspace(client, user, "收链接") + + async def fake_fetch(url: str) -> article_service.Article: + return article_service.Article( + url=url, + title="回滚前先停灰度", + body_md="先停流量，再退镜像版本。", + site="微信公众号", + author="达道技术", + ) + + async def fake_store(session: Any, **kwargs: Any) -> tuple[str, list[Any]]: + return kwargs["body_md"], [] + + monkeypatch.setattr(article_service, "fetch_article", fake_fetch) + monkeypatch.setattr(article_service, "store_images", fake_store) + + ws_id = uuid.UUID(ws["id"]) + async with SessionFactory() as session: + await bind_tenant(session, ws_id, uuid.UUID(user["user_id"])) + channel = BotChannel( + workspace_id=ws_id, + kind="wechat_clawbot", + name="微信", + status="active", + created_by=uuid.UUID(user["user_id"]), + ) + session.add(channel) + await session.flush() + + result = await handle_text( + session, + channel, + "https://mp.weixin.qq.com/s/abc", + base_url="http://site", + ) + await session.commit() + + assert result["action"] == "knowledge" + assert result["reply"] == "已存成知识草稿：回滚前先停灰度" + async with SessionFactory() as session: + await bind_tenant(session, ws_id, uuid.UUID(user["user_id"])) + item = await session.get(Item, result["item_id"]) + assert item is not None + assert item.kind == "knowledge" + assert item.status == "draft" + assert "> 原文：<https://mp.weixin.qq.com/s/abc>" in (item.body_md or "") + doc = ( + await session.execute(select(KnowledgeDoc).where(KnowledgeDoc.item_id == item.id)) + ).scalar_one() + assert doc.status_hint == "draft" + + +async def test_link_failure_is_reported_not_silent( + client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + user = await login(client) + ws = await create_workspace(client, user, "收链接") + + async def fake_fetch(url: str) -> article_service.Article: # noqa: ARG001 + raise article_service.ArticleError("对方站点返回 403") + + monkeypatch.setattr(article_service, "fetch_article", fake_fetch) + + ws_id = uuid.UUID(ws["id"]) + async with SessionFactory() as session: + await bind_tenant(session, ws_id, uuid.UUID(user["user_id"])) + channel = BotChannel( + workspace_id=ws_id, + kind="wechat_clawbot", + name="微信", + status="active", + created_by=uuid.UUID(user["user_id"]), + ) + session.add(channel) + await session.flush() + + result = await handle_text( + session, channel, "https://example.com/a", base_url="http://site" + ) + await session.commit() + + assert result["action"] == "ignored" + assert "没抓下来" in result["reply"] + assert "任务：" in result["reply"] + + +async def test_images_are_copied_into_our_storage( + client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """文章里的图要落到我们自己的存储里：公众号一改图，正文就是一排红叉。""" + user = await login(client) + ws = await create_workspace(client, user, "存图") + ws_id = uuid.UUID(ws["id"]) + + async def fake_download(client: Any, url: str) -> tuple[bytes, str]: # noqa: ARG001 + return b"\x89PNG\r\n\x1a\nfake", "image/png" + + monkeypatch.setattr(article_service, "_download_image", fake_download) + + body = "看图\n\n![](https://mmbiz.qpic.cn/mmbiz_png/abc/640?wx_fmt=png)" + async with SessionFactory() as session: + await bind_tenant(session, ws_id, uuid.UUID(user["user_id"])) + item = Item( + workspace_id=ws_id, + kind="knowledge", + title="带图的一篇", + body_md=body, + status="draft", + creator_id=uuid.UUID(user["user_id"]), + ) + session.add(item) + await session.flush() + rewritten, rows = await article_service.store_images( + session, + ws_id=ws_id, + item_id=item.id, + uploader_id=uuid.UUID(user["user_id"]), + body_md=body, + ) + await session.commit() + + assert len(rows) == 1 + assert rows[0].filename.startswith("图1.") + assert "attachment" in rewritten and "/raw?" in rewritten + assert "mmbiz.qpic.cn" not in rewritten diff --git a/web/src/pages/Integrations.tsx b/web/src/pages/Integrations.tsx index bfc7e02..d97e6b6 100644 --- a/web/src/pages/Integrations.tsx +++ b/web/src/pages/Integrations.tsx @@ -216,6 +216,7 @@ export function IntegrationsPage() { <li>点右上「扫码连接微信」，用手机微信扫出现的二维码并在手机上确认。</li> <li>连上之后直接给这个微信机器人发消息：默认记成任务，「知识：」开头记成知识草稿。</li> <li>第一行当标题，其余当正文；标题或正文里带 #标签 会一起加上。</li> + <li>直接发一条链接（公众号文章也行），正文和配图会抓下来存成知识草稿。</li> </ol> </section>