From 8f2a73c5d4e610a3a114192ab488fe02309fcb3d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Tue, 20 Oct 2026 20:57:50 +0800 Subject: [PATCH] =?UTF-8?q?feat(site-import):=20=E6=8A=93=E5=8F=96?= =?UTF-8?q?=E7=BC=96=E6=8E=92=E4=B8=8E=E5=86=85=E5=AD=98=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E8=A1=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit crawler：BFS ≤10 页（深度 ≤2）+ 入口双重门（非 html 零页/SPA 空壳）+ 薄子页跳过 + CSS/logo 15s 追抓（MIME+魔数双验 ≤2MB → base64 data URL） + 总文本 <1K 内容不足 + 主色默认 #066fc0 回退；logo 相对路径对入口绝对化。 task：(slug, task_id) 联合索引内存表 + 30 分钟惰性清理 + 强引用防 GC + 全平台并发 ≤2（复活同占名额）+ 频控 3 次/15min（新建与 retry 同计， SlidingWindowCounter 同款独立实例）+ 7 分钟总超时兜底（生成段超时标 generate 可 retry=generate 复活不重抓）+ 内容不足三触发失败文案映射。 conftest 增 autouse 复位夹具（任务表/频控单例跨测试隔离）。 Co-Authored-By: Claude Code --- app/platform/site_import/crawler.py | 191 +++++++++++ app/platform/site_import/task.py | 243 ++++++++++++++ tests/conftest.py | 11 + tests/test_site_import_task.py | 481 ++++++++++++++++++++++++++++ 4 files changed, 926 insertions(+) create mode 100644 app/platform/site_import/crawler.py create mode 100644 app/platform/site_import/task.py create mode 100644 tests/test_site_import_task.py diff --git a/app/platform/site_import/crawler.py b/app/platform/site_import/crawler.py new file mode 100644 index 0000000..3c8b84f --- /dev/null +++ b/app/platform/site_import/crawler.py @@ -0,0 +1,191 @@ +"""site_import 抓取编排（spec §3.1/§3.3 的 crawl 段）。 + +BFS 选页 ≤10（深度 ≤2）→ safe_fetch 逐页 → extract 清洗抽取 → +CSS/logo 追抓（15s 超时）→ 组装 crawl_result dict 供生成器。 + +入口页双重门（spec §3.1/§6）：非 text/html → 零页内容不足；清洗后 +<200 字符 → SPA 空壳（子页过薄只跳过不失败——JS 渲染的产品子页常见， +整站不该陪葬）。单页 >500KB 截断解析（clean_html 未闭合块二遍清洗兜底）。 +discover_links 会回吐入口自身（href="#"/"/"）——visited 预置 start URL +去重，防浪费 ≤10 页名额（Task 3 备忘）。 +""" +from __future__ import annotations + +import base64 +import re +from collections import deque +from urllib.parse import urljoin, urlsplit, urlunsplit + +from app.platform.site_import import safe_fetch +from app.platform.site_import.extract import ( + clean_html, discover_links, extract_color, extract_contacts, + extract_logo_url, is_thin, +) +from app.platform.site_import.safe_fetch import FetchError + +MAX_PAGES = 10 +MAX_DEPTH = 2 +DEFAULT_COLOR = "#066fc0" # extract_color None 的函数层不兜底原则（Task 3 备忘） +PAGE_TIMEOUT = 10.0 +ASSET_TIMEOUT = 15.0 +MAX_PAGE_CHARS = 500 * 1024 # 单页 500KB 上限（截断解析） +LOGO_MAX_BYTES = 2 * 1024 * 1024 +LOGO_OK_MIME = {"image/png", "image/jpeg", "image/webp"} +INSUFFICIENT_TOTAL_CHARS = 1000 + +SPA_MESSAGE = "该网站需要浏览器渲染才能看到内容，暂不支持自动导入——请在配置中心手动配置" +INSUFFICIENT_MESSAGE = "网站内容不足以生成配置，请在配置中心手动配置" + +# 首个外链样式表发现（link 标签内 rel 含 stylesheet → 提取 href） +_RE_LINK_TAG = re.compile(r"]*>", re.IGNORECASE) +_RE_REL_STYLESHEET = re.compile(r"""rel\s*=\s*["']?\s*stylesheet\b""", + re.IGNORECASE) +_RE_HREF_IN_TAG = re.compile(r"""href\s*=\s*["']([^"']+)["']""", + re.IGNORECASE) + + +class CrawlError(Exception): + """抓取阶段不可恢复失败（SPA 空壳/内容不足）——phase=crawl，不可 retry。""" + + +def _sniff_image(data: bytes) -> str | None: + """魔数指纹（与 Content-Type 双验——防伪造头）。""" + if data.startswith(b"\x89PNG\r\n\x1a\n"): + return "image/png" + if data.startswith(b"\xff\xd8\xff"): + return "image/jpeg" + if len(data) >= 12 and data[:4] == b"RIFF" and data[8:12] == b"WEBP": + return "image/webp" + return None + + +def _normalize(url: str) -> str: + """同源同径去重键（与 discover_links 的清洗口径对齐：去查询去锚点）。""" + parts = urlsplit(url) + return urlunsplit((parts.scheme.lower(), parts.netloc.lower(), + parts.path, "", "")) + + +def _html_text(resp) -> str | None: + """Content-Type 非 text/html → None（调用方跳页/判零页）；500KB 截断。""" + ct = resp.headers.get("content-type", "").lower() + if "text/html" not in ct: + return None + return resp.text[:MAX_PAGE_CHARS] + + +def _first_css_url(base_url: str, html_text: str) -> str | None: + for m in _RE_LINK_TAG.finditer(html_text): + tag = m.group(0) + if _RE_REL_STYLESHEET.search(tag): + hm = _RE_HREF_IN_TAG.search(tag) + if hm: + return urljoin(base_url, hm.group(1).strip()) + return None + + +async def _fetch_logo(url: str, warnings: list[str]) -> str | None: + """logo 下载校验（spec §3.1）：MIME 白名单 + 魔数双验 + ≤2MB → + base64 data URL；不合格 → None + warning 注记（审核页无头像预览， + 保存链路头像项空置——零新后端端点）。""" + try: + resp = await safe_fetch.fetch(url, ASSET_TIMEOUT) + except FetchError: + warnings.append("logo 下载失败，已跳过头像") + return None + data = resp.content + if len(data) > LOGO_MAX_BYTES: + warnings.append("logo 超过 2MB 上限，已跳过头像") + return None + mime = resp.headers.get("content-type", "").split(";", 1)[0].strip().lower() + sniffed = _sniff_image(data) + if mime not in LOGO_OK_MIME or sniffed is None or sniffed != mime: + warnings.append("logo 为不支持的格式，已跳过头像") + return None + return f"data:{sniffed};base64," + base64.b64encode(data).decode("ascii") + + +async def crawl(start_url: str) -> dict: + """整站抓取编排，返回 crawl_result dict。 + + FetchError 不在此包装——交任务层 fail（白话化 + phase 归档）。 + """ + warnings: list[str] = [] + + # 1) 入口页：非 text/html → 零页（内容不足三触发之一） + entry = await safe_fetch.fetch(start_url, PAGE_TIMEOUT) + entry_html = _html_text(entry) + if entry_html is None: + raise CrawlError(INSUFFICIENT_MESSAGE) + entry_clean = clean_html(entry_html) + if is_thin(entry_clean): + raise CrawlError(SPA_MESSAGE) + + # 2) BFS：入口 depth 0 → 子页 ≤MAX_DEPTH；关键词优先序即抓取优先级 + pages_text = entry_clean + htmls = [entry_html] + fetched = 1 + visited = {_normalize(start_url)} # discover_links 含入口自身——预置去重 + queue: deque[tuple[str, int]] = deque( + (u, 1) for u in discover_links(start_url, entry_html)) + + while queue and fetched < MAX_PAGES: + url, depth = queue.popleft() + norm = _normalize(url) + if norm in visited: + continue + visited.add(norm) + try: + resp = await safe_fetch.fetch(url, PAGE_TIMEOUT) + except FetchError: + warnings.append(f"子页抓取失败已跳过：{url}") + continue + html = _html_text(resp) + if html is None: + warnings.append(f"子页非 HTML 已跳过：{url}") + continue + clean = clean_html(html) + if is_thin(clean): + warnings.append(f"子页内容过薄已跳过：{url}") + continue + pages_text += "\n" + clean + htmls.append(html) + fetched += 1 + if depth < MAX_DEPTH: + queue.extend((u, depth + 1) for u in discover_links(url, html)) + + # 3) 总文本 <1K（内容不足三触发之一） + if len(pages_text) < INSUFFICIENT_TOTAL_CHARS: + raise CrawlError(INSUFFICIENT_MESSAGE) + + # 4) 主色：内联 style（全页 html 集）+ 首个外链 CSS（15s 追抓，失败 + # 静默回退内联/默认色——样式表属增益项不陪葬） + css_text: str | None = None + css_url = _first_css_url(start_url, entry_html) + if css_url: + try: + css_resp = await safe_fetch.fetch(css_url, ASSET_TIMEOUT) + css_text = css_resp.text[:MAX_PAGE_CHARS] + except FetchError: + pass + combined_html = "\n".join(htmls) + color = extract_color(combined_html, css_text) or DEFAULT_COLOR + + contacts = extract_contacts(combined_html) + + logo_data_url = None + logo_url = extract_logo_url(combined_html) + if logo_url: + # src 可能是相对路径（WordPress uploads 常见）——对入口页绝对化 + logo_data_url = await _fetch_logo(urljoin(start_url, logo_url), + warnings) + + return { + "pages_text": pages_text, + "color": color, + "whatsapp": contacts["whatsapp"], + "email": contacts["email"], + "logo_data_url": logo_data_url, + "warnings": warnings, + "pages_fetched": fetched, + } diff --git a/app/platform/site_import/task.py b/app/platform/site_import/task.py new file mode 100644 index 0000000..f3ecd68 --- /dev/null +++ b/app/platform/site_import/task.py @@ -0,0 +1,243 @@ +"""site_import 内存任务表（spec §3.3）——单进程 asyncio，(slug, task_id) 索引。 + +内存铁律：30 分钟惰性清理（查询/新建时顺带扫）；强引用防 GC（裸 +create_task 会被回收——asyncio 官方坑，模块级 dict 存 task 对象）； +全平台并发 ≤2（复活任务占用同一名额）；频控 3 次/15 分钟（新建与 +retry 同计）。异常类型：ImportRateLimited→429 / ConcurrentLimit→409 / +LookupError→404 / ValueError→409——API 层（Task 7）映射。 + +7 分钟总超时兜底：超时落在生成段（抓取产物已在）标 failed_phase= +"generate"——retry=generate 是 spec §3.3 的预期恢复路径（不重抓）； +落在抓取段则无产物，重开任务。 +""" +from __future__ import annotations + +import asyncio +import time +import uuid +from collections import defaultdict + +from app.platform.site_import import crawler, generator +from app.platform.site_import.crawler import CrawlError, INSUFFICIENT_MESSAGE +from app.platform.site_import.generator import GenerateError + +TASK_TTL = 30 * 60 +TOTAL_TIMEOUT = 7 * 60 +MAX_CONCURRENT = 2 +IMPORT_WINDOW = 15 * 60 +IMPORT_MAX = 3 + + +class ImportRateLimited(Exception): + """频控超限（15 分钟 3 次）——API 层映射 429。""" + + +class ConcurrentLimit(Exception): + """全平台并发上限（≤2，复活任务同占名额）——API 层映射 409。""" + + +class _SlidingWindow: + """(kind, slug) 键滑动窗计数——config_routes.SlidingWindowCounter 同款 + 风格的独立实例（勿共用 email/push 测试端点的预算）。同款语义：超限后 + allow 守卫先于一切预检；计数只落在真正发起的这次（校验失败不消耗）。""" + + def __init__(self, window_seconds: float = IMPORT_WINDOW, + max_calls: int = IMPORT_MAX) -> None: + self.window_seconds = window_seconds + self.max_calls = max_calls + self._calls: dict[tuple[str, str], list[float]] = defaultdict(list) + + def allow(self, key: tuple[str, str]) -> bool: + self._prune(key, time.time()) + return len(self._calls.get(key, ())) < self.max_calls + + def record(self, key: tuple[str, str]) -> None: + now = time.time() + self._prune(key, now) + self._calls[key].append(now) + + def _prune(self, key: tuple[str, str], now: float) -> None: + calls = self._calls.get(key) + if calls is None: + return + kept = [t for t in calls if t > now - self.window_seconds] + if kept: + self._calls[key] = kept + else: + self._calls.pop(key, None) + + +# 模块级单例：导入频控（键 ("import", slug)，全平台各租户独立预算） +_import_limiter = _SlidingWindow() + +# 内存任务表：(slug, task_id) 联合索引——跨租户持他人 id 查询即 miss +_tasks: dict[tuple[str, str], "ImportTask"] = {} +_lock = asyncio.Lock() + + +class TaskState: + PENDING = "pending" + CRAWLING = "crawling" + GENERATING = "generating" + DONE = "done" + FAILED = "failed" + + ACTIVE = (PENDING, CRAWLING, GENERATING) + + +class ImportTask: + """一条导入任务：状态机 + 阶段产物 + 强引用的后台 aio task。 + + failed_phase："generate" 可 retry=generate 复活（复用 crawl_result， + 不重抓）；"crawl"（SPA/内容不足/抓取失败）与 None（超时落在抓取段/ + 未知异常）不可复活——重开任务。 + """ + + def __init__(self, slug: str, task_id: str, url: str | None) -> None: + self.slug = slug + self.task_id = task_id + self.url = url + self.state = TaskState.PENDING + self.phase: str | None = None # 运行中阶段（crawling/generating） + self.pages_fetched = 0 + self.draft: dict | None = None + self.error: str | None = None + self.created_at = time.time() + self.finished_at: float | None = None + self.failed_phase: str | None = None + self.crawl_result: dict | None = None # retry=generate 复用 + self.need_crawl = True + self._aio_task: asyncio.Task | None = None # 强引用防 GC + + def fail(self, message: str, *, phase: str | None) -> None: + self.state = TaskState.FAILED + self.error = message + self.failed_phase = phase + self.phase = None + self.finished_at = time.time() + + def _finish(self, draft: dict) -> None: + self.draft = draft + self.state = TaskState.DONE + self.phase = None + self.finished_at = time.time() + + +def _prune_expired(now: float | None = None) -> None: + """30 分钟惰性清理（查询/新建顺带扫）。运行中任务防御性取消—— + 7 分钟总超时下正常不会发生（30 >> 7），仅兜异常态。""" + now = time.time() if now is None else now + deadline = now - TASK_TTL + for key in [k for k, t in _tasks.items() if t.created_at < deadline]: + stale = _tasks.pop(key) + if stale._aio_task is not None and not stale._aio_task.done(): + stale._aio_task.cancel() + + +def _count_active() -> int: + return sum(1 for t in _tasks.values() if t.state in TaskState.ACTIVE) + + +def active_count() -> int: + """全平台进行中任务数（pending/crawling/generating）。""" + _prune_expired() + return _count_active() + + +def get(slug: str, task_id: str) -> ImportTask | None: + """(slug, task_id) 联合索引查询 + 惰性清理（跨租户 → None → 404）。""" + _prune_expired() + return _tasks.get((slug, task_id)) + + +async def start(slug: str, url: str | None, *, + retry_task_id: str | None = None) -> str: + """新建或复活（retry=generate）导入任务，返回 task_id。 + + 频控守卫最先（超限后 429 优先于一切预检）；record 只落在真正发起的 + 这次——LookupError/ValueError/ConcurrentLimit 不消耗预算。 + """ + _prune_expired() + key = ("import", slug) + if not _import_limiter.allow(key): + raise ImportRateLimited( + f"导入太频繁，{IMPORT_WINDOW // 60} 分钟内最多 {IMPORT_MAX} 次，" + "稍后再试") + async with _lock: + if retry_task_id is not None: + target = _tasks.get((slug, retry_task_id)) + if target is None: + raise LookupError( + f"task {retry_task_id!r} not found for tenant {slug!r}") + if (target.state != TaskState.FAILED + or target.failed_phase != "generate"): + raise ValueError("任务当前状态不可重试" + "（仅 generate 阶段失败可复活）") + # 并发名额对新建与 retry 复活同样生效（spec §3.3） + if _count_active() >= MAX_CONCURRENT: + raise ConcurrentLimit("已有导入任务进行中，稍后再试") + if retry_task_id is not None: + # 复活同 task_id：复用已抓产物，不重抓 + target.state = TaskState.PENDING + target.error = None + target.failed_phase = None + target.finished_at = None + target.need_crawl = False + else: + target = ImportTask(slug, uuid.uuid4().hex, url) + _tasks[(slug, target.task_id)] = target + _import_limiter.record(key) + target._aio_task = asyncio.create_task(_pipeline(target)) + return target.task_id + + +async def _pipeline(task_obj: ImportTask) -> None: + """后台管线：7 分钟总超时兜底 + 异常归档为白话失败。""" + try: + await asyncio.wait_for(_run(task_obj), TOTAL_TIMEOUT) + except asyncio.TimeoutError: + task_obj.fail( + "导入超时，请重试", + phase="generate" if task_obj.crawl_result is not None else None) + except CrawlError as exc: + task_obj.fail(str(exc), phase="crawl") # SPA/内容不足——不可 retry + except GenerateError as exc: + task_obj.fail(str(exc), phase="generate") # retry=generate 复活 + except Exception as exc: # 防御：未知异常不裸抛（任务表对象完整性优先） + task_obj.fail(f"导入失败：{exc}", phase=None) + + +async def _run(task_obj: ImportTask) -> None: + if task_obj.need_crawl: + task_obj.state = TaskState.CRAWLING + task_obj.phase = "crawling" + task_obj.crawl_result = await crawler.crawl(task_obj.url) + task_obj.pages_fetched = int( + task_obj.crawl_result.get("pages_fetched") or 0) + task_obj.state = TaskState.GENERATING + task_obj.phase = "generating" + provider = generator.make_provider(None) # resolver 三级链（平台设置/env） + # source_text 传 analyze 实际消费的截断后文本（24K 同款——Task 5 备忘： + # 否则截断段数字会被 build 的数字校验误剔除） + source = ((task_obj.crawl_result or {}).get("pages_text") + or "")[:generator.TOTAL_INPUT_CAP] + analysis = await generator.analyze(provider, source) + if not (analysis.get("product_lines") or []): + # 内容不足三触发之一（调用 1 产品线为空）——同 crawl 类不可复活 + task_obj.fail(INSUFFICIENT_MESSAGE, phase="crawl") + return + gen = await generator.generate(provider, analysis) + task_obj._finish( + generator.build(gen, analysis, task_obj.crawl_result, source)) + + +def _reset_for_test() -> None: + """测试隔离（conftest autouse 前后各跑）：任务表清空（运行中 aio + 任务防御性取消）+ 频控窗计数/参数复位。""" + for t in _tasks.values(): + if t._aio_task is not None and not t._aio_task.done(): + t._aio_task.cancel() + _tasks.clear() + _import_limiter._calls.clear() + _import_limiter.window_seconds = IMPORT_WINDOW + _import_limiter.max_calls = IMPORT_MAX diff --git a/tests/conftest.py b/tests/conftest.py index 7c96ef9..c6dd1d6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -115,6 +115,17 @@ def _reset_test_endpoint_limiter(): _reset_test_limiter_for_test() +@pytest.fixture(autouse=True) +def _reset_site_import_tasks(): + """site-import Task 6: 导入任务表/频控器模块级单例前后复位——(slug, + task_id) 表与 15 分钟窗计数不隔离会串测试(前一脚打满的频控预算顶掉 + 后一脚的 start 断言);运行中后台 aio 任务一并防御性取消。""" + from app.platform.site_import.task import _reset_for_test + _reset_for_test() + yield + _reset_for_test() + + # ---- Phase 2 Task 10: FakeAI 公共确定性桩(清理单 #1) -------------------------- # 收敛自四份拷贝(test_lead_email_trigger / test_cc_integration / # test_channel_hygiene / test_lead_push_wiring,接口取超集: 前两份全量, diff --git a/tests/test_site_import_task.py b/tests/test_site_import_task.py new file mode 100644 index 0000000..7fa3597 --- /dev/null +++ b/tests/test_site_import_task.py @@ -0,0 +1,481 @@ +"""site_import 任务编排与内存任务表（Task 6）。 + +crawler 段全打桩 safe_fetch.fetch（构造真 httpx.Response，不连网）； +task 段打桩 crawler.crawl + generator 三函数与 make_provider。频控/清理 +的时间推进照 test_auth.py 先例直改内部计数或 created_at。 +""" +import asyncio +import base64 +import weakref + +import httpx +import pytest + +from app.platform.site_import import crawler, generator, safe_fetch, task + + +def _resp(url, html, status=200, content_type="text/html; charset=utf-8"): + return httpx.Response( + status, text=html, headers={"content-type": content_type}, + request=httpx.Request("GET", url)) + + +def _bin_resp(url, data, content_type): + return httpx.Response( + 200, content=data, headers={"content-type": content_type}, + request=httpx.Request("GET", url)) + + +def _fake_fetch(pages: dict, log: list | None = None): + """url → Response 桩；未登记 url 抛 FetchError（子页 404 形态）。""" + async def fetch(url, timeout=10.0): + if log is not None: + log.append((url, timeout)) + if url not in pages: + raise safe_fetch.FetchError(f"fetch failed for {url!r}: 404") + return pages[url] + return fetch + + +def _page(text: str) -> str: + return f"

{text}

" + + +def _fat(marker: str, repeat: int = 40) -> str: + return _page(f"{marker} " + "guardrail highway safety manufacturer " * repeat) + + +PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"\x00" * 64 +JPEG_BYTES = b"\xff\xd8\xff\xe0" + b"\x00" * 64 + +HOME_FULL = """ +Acme + + + +
+logo +WhatsApp +Mail +

Acme builds guardrail systems for highways since 1999.

+

Acme manufactures guardrail, safety barriers and crash cushions for +highway contractors worldwide. Certified production, galvanized finish, +custom lengths available for export projects. Acme manufactures guardrail, +safety barriers and crash cushions for highway contractors worldwide. +Certified production, galvanized finish, custom lengths available.

+
copyright noise
+ +""" + +CSS_FULL = "body{color:#bb3344} a{color:#bb3344} h1{color:#bb3344}" + + +def _full_site_pages() -> dict: + base = "https://acme.com" + return { + f"{base}/": _resp(f"{base}/", HOME_FULL), + f"{base}/about": _resp(f"{base}/about", _fat("About us marker")), + f"{base}/products": _resp(f"{base}/products", _fat("Products marker")), + f"{base}/style.css": _resp(f"{base}/style.css", CSS_FULL, + content_type="text/css; charset=utf-8"), + f"{base}/uploads/2024/05/logo-acme.png": _bin_resp( + f"{base}/uploads/2024/05/logo-acme.png", PNG_BYTES, "image/png"), + } + + +class TestCrawl: + + async def test_首页关于产品页全抓并清洗(self, monkeypatch): + pages = _full_site_pages() + monkeypatch.setattr(crawler.safe_fetch, "fetch", _fake_fetch(pages)) + result = await crawler.crawl("https://acme.com/") + # 三页正文都在，script/footer 噪音不在 + assert "About us marker" in result["pages_text"] + assert "Products marker" in result["pages_text"] + assert "Acme builds guardrail" in result["pages_text"] + assert "var x=1" not in result["pages_text"] + assert "copyright noise" not in result["pages_text"] + # CSS #bb3344×3 压过内联 #aa1122×1；联系方式/logo 就位 + assert result["color"] == "#bb3344" + assert result["whatsapp"] == "+8613800138000" + assert result["email"] == "sales@acme.com" + assert result["logo_data_url"].startswith("data:image/png;base64,") + decoded = base64.b64decode(result["logo_data_url"].split(",", 1)[1]) + assert decoded == PNG_BYTES + assert result["pages_fetched"] == 3 + assert result["warnings"] == [] + + async def test_深度与页数上限(self, monkeypatch): + base = "https://acme.com" + links = "".join(f'page {i}' for i in range(15)) + home = _page("Acme industrial guardrail home. " + + "highway safety barrier manufacturer " * 20 + links) + pages = {f"{base}/": _resp(f"{base}/", home)} + for i in range(15): + pages[f"{base}/p{i}"] = _resp(f"{base}/p{i}", _fat(f"P{i} marker")) + log: list = [] + monkeypatch.setattr(crawler.safe_fetch, "fetch", _fake_fetch(pages, log)) + result = await crawler.crawl(f"{base}/") + assert result["pages_fetched"] == crawler.MAX_PAGES + assert len(log) == crawler.MAX_PAGES # 入口 + 9 个子页，其余不再抓 + assert "P8 marker" in result["pages_text"] + assert "P9 marker" not in result["pages_text"] + + async def test_薄子页跳过不失败(self, monkeypatch): + base = "https://acme.com" + home = _page("Acme guardrail home. " + + "highway safety barrier manufacturer " * 20 + + ' abouttiny') + pages = { + f"{base}/": _resp(f"{base}/", home), + f"{base}/about": _resp(f"{base}/about", _fat("About marker")), + f"{base}/tiny": _resp(f"{base}/tiny", _page("thin page")), + } + monkeypatch.setattr(crawler.safe_fetch, "fetch", _fake_fetch(pages)) + result = await crawler.crawl(f"{base}/") + assert "About marker" in result["pages_text"] + assert "thin page" not in result["pages_text"] + assert result["pages_fetched"] == 2 + assert any("薄" in w for w in result["warnings"]) + + async def test_入口页SPA空壳抛CrawlError(self, monkeypatch): + base = "https://acme.com" + pages = {f"{base}/": _resp(f"{base}/", _page("bare shell"))} + monkeypatch.setattr(crawler.safe_fetch, "fetch", _fake_fetch(pages)) + with pytest.raises(crawler.CrawlError) as ei: + await crawler.crawl(f"{base}/") + assert "浏览器渲染" in str(ei.value) + + async def test_内容不足_总文本1K以下抛CrawlError(self, monkeypatch): + base = "https://acme.com" + # 清洗后 ≥200（过 SPA 门）但总文本 <1K（无子页） + entry = _page("Acme guardrail maker. " * 12) + assert 200 <= len(entry.split("

")[1].split("

")[0]) < 1000 + pages = {f"{base}/": _resp(f"{base}/", entry)} + monkeypatch.setattr(crawler.safe_fetch, "fetch", _fake_fetch(pages)) + with pytest.raises(crawler.CrawlError) as ei: + await crawler.crawl(f"{base}/") + assert str(ei.value) == crawler.INSUFFICIENT_MESSAGE + + async def test_入口页非html抛CrawlError(self, monkeypatch): + base = "https://acme.com" + pages = {f"{base}/": _resp(f"{base}/", "%PDF-1.7 fake", + content_type="application/pdf")} + log: list = [] + monkeypatch.setattr(crawler.safe_fetch, "fetch", _fake_fetch(pages, log)) + with pytest.raises(crawler.CrawlError) as ei: + await crawler.crawl(f"{base}/") + assert str(ei.value) == crawler.INSUFFICIENT_MESSAGE + assert len(log) == 1 # 零子页追抓 + + async def test_logo下载_格式大小校验(self, monkeypatch): + base = "https://acme.com" + fat_home = _page( + 'logo ' + + "Acme industrial guardrail manufacturer. " * 60) + + def _pages(logo_resp) -> dict: + return { + f"{base}/": _resp(f"{base}/", fat_home), + f"{base}/logo.png": logo_resp, + } + + # png 通过 → data URL + monkeypatch.setattr(crawler.safe_fetch, "fetch", _fake_fetch(_pages( + _bin_resp(f"{base}/logo.png", PNG_BYTES, "image/png")))) + result = await crawler.crawl(f"{base}/") + assert result["logo_data_url"].startswith("data:image/png;base64,") + + # svg → None + warning + monkeypatch.setattr(crawler.safe_fetch, "fetch", _fake_fetch(_pages( + _bin_resp(f"{base}/logo.png", b"", + "image/svg+xml")))) + result = await crawler.crawl(f"{base}/") + assert result["logo_data_url"] is None + assert any("logo" in w and "格式" in w for w in result["warnings"]) + + # >2MB → None + warning + monkeypatch.setattr(crawler.safe_fetch, "fetch", _fake_fetch(_pages( + _bin_resp(f"{base}/logo.png", + PNG_BYTES + b"\x00" * (2 * 1024 * 1024), "image/png")))) + result = await crawler.crawl(f"{base}/") + assert result["logo_data_url"] is None + assert any("logo" in w and "2MB" in w for w in result["warnings"]) + + # Content-Type 与魔数不符（头谎报 png 实为 jpeg）→ None + warning + monkeypatch.setattr(crawler.safe_fetch, "fetch", _fake_fetch(_pages( + _bin_resp(f"{base}/logo.png", JPEG_BYTES, "image/png")))) + result = await crawler.crawl(f"{base}/") + assert result["logo_data_url"] is None + assert any("logo" in w for w in result["warnings"]) + + async def test_默认色回退(self, monkeypatch): + base = "https://acme.com" + pages = {f"{base}/": _resp(f"{base}/", _fat("Acme home", 60))} + monkeypatch.setattr(crawler.safe_fetch, "fetch", _fake_fetch(pages)) + result = await crawler.crawl(f"{base}/") + assert result["color"] == crawler.DEFAULT_COLOR + + async def test_CSS追抓15s超时不阻塞(self, monkeypatch): + base = "https://acme.com" + broken_css = dict(_full_site_pages()) + broken_css[f"{base}/style.css"] = _resp( + f"{base}/style.css", "", content_type="text/css; charset=utf-8") + + async def fetch(url, timeout=10.0): + if url == f"{base}/style.css": + raise safe_fetch.FetchError( + f"fetch failed for {url!r}: timed out") + return broken_css[url] + + log: list = [] + original = fetch + + async def fetch_logged(url, timeout=10.0): + log.append((url, timeout)) + return await original(url, timeout) + + monkeypatch.setattr(crawler.safe_fetch, "fetch", fetch_logged) + # CSS 失败不阻塞：抓取完成，内联色兜底生效 + result = await crawler.crawl(f"{base}/") + assert result["color"] == "#aa1122" + # CSS 追抓用的是 15s 资产超时（页 10s） + assert (f"{base}/style.css", crawler.ASSET_TIMEOUT) in log + + +# ---- task 段 --------------------------------------------------------------- + +_DEFAULT_CRAWL = { + "pages_text": "Acme guardrail manufacturer since 1999 certified. " * 40, + "color": "#123456", + "whatsapp": "+8613800138000", + "email": "sales@acme.com", + "logo_data_url": None, + "warnings": [], + "pages_fetched": 3, +} + + +class _Stubs: + """crawler/generator 全量确定性桩 + 调用计数。 + + *_errors 为异常队列出列（首错只撞一次），delay 模拟慢阶段。 + """ + + def __init__(self): + self.calls = {"crawl": 0, "analyze": 0, "generate": 0, "build": 0} + self.crawl_result = dict(_DEFAULT_CRAWL) + self.analysis = {"company_name": "Acme", + "product_lines": [{"name": "guardrail"}]} + self.gen_out = {"身份描述": "外贸销售"} + self.draft = {"brand": {"title": "Acme", "color": "#123456"}} + self.crawl_errors: list = [] + self.generate_errors: list = [] + self.delay = 0.0 + + async def _slow(self): + if self.delay: + await asyncio.sleep(self.delay) + + async def crawl_(self, url): + self.calls["crawl"] += 1 + await self._slow() + if self.crawl_errors: + raise self.crawl_errors.pop(0) + return dict(self.crawl_result) + + async def analyze_(self, provider, source): + self.calls["analyze"] += 1 + await self._slow() + return dict(self.analysis) + + async def generate_(self, provider, analysis): + self.calls["generate"] += 1 + await self._slow() + if self.generate_errors: + raise self.generate_errors.pop(0) + return dict(self.gen_out) + + def build_(self, gen, analysis, crawl, source): + self.calls["build"] += 1 + assert "pages_text" in crawl # build 消费 crawl_result 原形 + return dict(self.draft) + + def install(self, monkeypatch): + monkeypatch.setattr(crawler, "crawl", self.crawl_) + monkeypatch.setattr(generator, "analyze", self.analyze_) + monkeypatch.setattr(generator, "generate", self.generate_) + monkeypatch.setattr(generator, "build", self.build_) + monkeypatch.setattr(generator, "make_provider", + lambda config_data=None: object()) + return self + + +async def _wait(slug: str, tid: str) -> "task.ImportTask": + t = task.get(slug, tid) + assert t is not None + await t._aio_task + return t + + +class TestTasks: + + async def test_状态机流转到done(self, monkeypatch): + stubs = _Stubs().install(monkeypatch) + tid = await task.start("t1", "https://acme.com/") + t = await _wait("t1", tid) + assert t.state == "done" + assert t.draft == stubs.draft + assert t.error is None + assert t.pages_fetched == 3 + assert t.finished_at is not None + assert stubs.calls == {"crawl": 1, "analyze": 1, "generate": 1, + "build": 1} + + async def test_slug与taskId联合索引_跨租户查404(self, monkeypatch): + _Stubs().install(monkeypatch) + tid = await task.start("t1", "https://acme.com/") + await _wait("t1", tid) + assert task.get("t1", tid) is not None + assert task.get("other", tid) is None + + async def test_全平台并发2_第三个拒绝(self, monkeypatch): + stubs = _Stubs().install(monkeypatch) + stubs.delay = 0.3 + ta = await task.start("t1", "https://a.com/") + tb = await task.start("t2", "https://b.com/") # 全平台（跨租户同算） + assert task.active_count() == 2 + with pytest.raises(task.ConcurrentLimit): + await task.start("t3", "https://c.com/") + await _wait("t1", ta) + await _wait("t2", tb) + + async def test_频控3次每15分钟_第4次429(self, monkeypatch): + _Stubs().install(monkeypatch) + for i in range(3): + tid = await task.start("t1", f"https://a.com/{i}") + await _wait("t1", tid) # 顺序完成，不撞并发上限 + with pytest.raises(task.ImportRateLimited) as ei: + await task.start("t1", "https://a.com/again") + assert "3 次" in str(ei.value) + + async def test_频控新建与retry同计(self, monkeypatch): + stubs = _Stubs().install(monkeypatch) + stubs.generate_errors.append( + generator.GenerateError("内容生成：输出无法解析")) + tid = await task.start("t1", "https://a.com/") + await _wait("t1", tid) # 计数 1（新建） + tid2 = await task.start("t1", None, retry_task_id=tid) + assert tid2 == tid + await _wait("t1", tid) # 计数 2（retry 同计） + tid_b = await task.start("t1", "https://a.com/b") + await _wait("t1", tid_b) # 计数 3 + with pytest.raises(task.ImportRateLimited): + await task.start("t1", "https://a.com/c") + + async def test_7分钟总超时强failed(self, monkeypatch): + stubs = _Stubs().install(monkeypatch) + monkeypatch.setattr(task, "TOTAL_TIMEOUT", 0.05) + # 超时落在抓取段：无产物 → 不可 retry + stubs.delay = 1.0 + tid = await task.start("t1", "https://a.com/") + t = await _wait("t1", tid) + assert t.state == "failed" + assert "导入超时" in t.error + assert t.failed_phase is None + # 超时落在生成段（抓取产物已在）→ 标 generate，retry=generate 复活 + stubs.delay = 0.0 + stubs.crawl_result["pages_fetched"] = 5 + async def slow_analyze(provider, source): + await asyncio.sleep(1.0) + return {"product_lines": [{"name": "x"}]} + monkeypatch.setattr(generator, "analyze", slow_analyze) + tid2 = await task.start("t1", "https://b.com/") + t2 = await _wait("t1", tid2) + assert t2.state == "failed" + assert "导入超时" in t2.error + assert t2.failed_phase == "generate" + + async def test_generate失败可retry同task复活且不重抓(self, monkeypatch): + stubs = _Stubs().install(monkeypatch) + stubs.generate_errors.append( + generator.GenerateError("内容生成：AI 调用超时（120s）")) + tid = await task.start("t1", "https://a.com/") + t = await _wait("t1", tid) + assert t.state == "failed" + assert t.failed_phase == "generate" + assert stubs.calls["crawl"] == 1 + tid2 = await task.start("t1", None, retry_task_id=tid) + assert tid2 == tid + t = await _wait("t1", tid2) + assert t.state == "done" + assert t.draft == stubs.draft + assert stubs.calls["crawl"] == 1 # 未重抓 + assert stubs.calls["generate"] == 2 + assert stubs.calls["build"] == 1 + + async def test_retry校验_不存在404与非法态409(self, monkeypatch): + stubs = _Stubs().install(monkeypatch) + with pytest.raises(LookupError): + await task.start("t1", None, retry_task_id="nope") + tid = await task.start("t1", "https://a.com/") + await _wait("t1", tid) + with pytest.raises(ValueError): # done 态不可 retry + await task.start("t1", None, retry_task_id=tid) + # 跨租户 retry 他人 task → LookupError（404 语义） + with pytest.raises(LookupError): + await task.start("t2", None, retry_task_id=tid) + # crawl 失败态（不可复活）→ ValueError + stubs.crawl_errors.append(crawler.CrawlError("抓取失败")) + tid_b = await task.start("t1", "https://b.com/") + await _wait("t1", tid_b) + with pytest.raises(ValueError): + await task.start("t1", None, retry_task_id=tid_b) + + async def test_30分钟惰性清理(self, monkeypatch): + _Stubs().install(monkeypatch) + tid = await task.start("t1", "https://a.com/") + t = await _wait("t1", tid) + assert task.get("t1", tid) is not None + t.created_at -= task.TASK_TTL + 1 # 拨到 30 分钟窗外（免真实等待） + assert task.get("t1", tid) is None + assert ("t1", tid) not in task._tasks + # 新建路径同样顺带扫 + tid2 = await task.start("t1", "https://b.com/") + await _wait("t1", tid2) + assert ("t1", tid) not in task._tasks + + async def test_强引用_任务对象在表内(self, monkeypatch): + stubs = _Stubs().install(monkeypatch) + stubs.delay = 0.3 + tid = await task.start("t1", "https://a.com/") + t = task.get("t1", tid) + ref = weakref.ref(t) + assert task.get("t1", tid) is t + assert ref() is not None # 表内强引用存活（未被 GC） + assert isinstance(t._aio_task, asyncio.Task) + await t._aio_task + + async def test_内容不足三触发映射失败文案(self, monkeypatch): + # 触发 1：入口页非 text/html（crawler 层零页 → CrawlError） + stubs = _Stubs().install(monkeypatch) + stubs.crawl_errors.append( + crawler.CrawlError(crawler.INSUFFICIENT_MESSAGE)) + tid = await task.start("t1", "https://a.com/") + t = await _wait("t1", tid) + assert t.state == "failed" + assert t.error == crawler.INSUFFICIENT_MESSAGE + assert t.failed_phase == "crawl" # 内容不足类不可 retry + # 触发 2：SPA 空壳（专属文案） + stubs.crawl_errors.append(crawler.CrawlError(crawler.SPA_MESSAGE)) + tid = await task.start("t1", "https://b.com/") + t = await _wait("t1", tid) + assert "浏览器渲染" in t.error + assert t.failed_phase == "crawl" + # 触发 3：调用 1 产品线为空（任务层映射内容不足文案） + stubs.analysis = {"company_name": "Acme", "product_lines": []} + tid = await task.start("t1", "https://c.com/") + t = await _wait("t1", tid) + assert t.error == crawler.INSUFFICIENT_MESSAGE + assert t.failed_phase == "crawl"