公众号的文章也能抓：走 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 拒绝、图片本地化、失败提示
This commit is contained in:
Steven
@@ -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"
|
||||
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
|
||||
Reference in New Issue
Block a user