From b2feaeddb492cad9c4f7f5f0ae3357c1c2043b35 Mon Sep 17 00:00:00 2001 From: wangwei Date: Thu, 6 Aug 2026 11:08:46 +0800 Subject: [PATCH] update for mcp --- .env | 24 +- .env.example | 11 +- backend/app/api/routes/perception.py | 30 +- .../app/application/agent/hyde_expander.py | 2 +- .../application/perception/crawl_service.py | 122 +++++- backend/app/config/settings.py | 20 +- .../perception/base_notification_store.py | 47 +++ .../perception/crawlers/base.py | 45 ++- .../perception/crawlers/catarc_crawler.py | 7 +- .../perception/crawlers/eurlex_crawler.py | 7 +- .../perception/crawlers/guobiao_crawler.py | 8 +- .../infrastructure/perception/llm_pipeline.py | 164 ++++---- .../perception/mock_notification_store.py | 66 ++++ .../perception/postgres_event_store.py | 13 +- .../perception/postgres_notification_store.py | 157 ++++++++ .../perception/regulation_differ.py | 224 +++++++++++ .../app/infrastructure/tasks/celery_app.py | 13 +- .../infrastructure/tasks/perception_tasks.py | 63 ++++ backend/app/services/llm/llm_factory.py | 4 +- backend/app/services/llm/qwen_client.py | 4 +- backend/app/shared/bootstrap.py | 21 ++ backend/requirements.txt | 5 + .../tests/perception/test_crawl_service.py | 356 +++++++++++++++++- backend/tests/perception/test_llm_pipeline.py | 155 ++++++-- .../perception/test_notification_store.py | 83 ++++ .../test_perception_notification_routes.py | 72 ++++ .../tests/perception/test_perception_tasks.py | 77 ++++ .../perception/test_regulation_differ.py | 157 ++++++++ .../backend-project-architecture.md | 25 ++ frontend/src/api/perception.ts | 33 ++ frontend/src/components/layout/Sidebar.tsx | 23 +- frontend/src/contexts/PageStateContext.tsx | 17 +- frontend/src/contexts/index.ts | 1 + frontend/src/contexts/pageStateDefaults.ts | 27 ++ .../src/pages/Compliance/CompliancePage.tsx | 2 +- .../pages/Compliance/useComplianceAnalysis.ts | 19 +- frontend/src/pages/Login/LoginPage.tsx | 2 +- .../src/pages/Perception/PerceptionPage.tsx | 66 +++- pyproject.toml | 2 + tests/test_celery_tasks.py | 14 + 40 files changed, 1986 insertions(+), 202 deletions(-) create mode 100644 backend/app/infrastructure/perception/base_notification_store.py create mode 100644 backend/app/infrastructure/perception/mock_notification_store.py create mode 100644 backend/app/infrastructure/perception/postgres_notification_store.py create mode 100644 backend/app/infrastructure/perception/regulation_differ.py create mode 100644 backend/app/infrastructure/tasks/perception_tasks.py create mode 100644 backend/tests/perception/test_notification_store.py create mode 100644 backend/tests/perception/test_perception_notification_routes.py create mode 100644 backend/tests/perception/test_perception_tasks.py create mode 100644 backend/tests/perception/test_regulation_differ.py create mode 100644 frontend/src/contexts/pageStateDefaults.ts diff --git a/.env b/.env index 6a3fae5..83e59f1 100644 --- a/.env +++ b/.env @@ -55,9 +55,16 @@ DOCUMENT_REPOSITORY_BACKEND=postgres USE_CELERY_WORKER=false # ===== 法规感知爬取配置 ===== +# 单次 HTTP 请求超时（秒），含正文抓取（fetch_full_text）。 PERCEPTION_CRAWL_TIMEOUT_SECONDS=120 +# 每个数据源单次爬取的最大条目数。 PERCEPTION_MAX_EVENTS_PER_SOURCE=100 -PERCEPTION_DIFF_SIMILARITY_THRESHOLD=0.85 +# 变更判定的次要闸门：段落改动字符占比达到该阈值才送 LLM 分类。 +# 数字变化（如 30米->20米）或情态词变化（应当/宜/不得等）无视此阈值，始终判定为显著变更。 +PERCEPTION_DIFF_MIN_CHANGE_RATIO=0.02 +# 定时全量爬取的执行间隔（秒），默认 21600 = 6 小时。 +# 仅当 Celery Beat 进程在运行时才生效（./dev.sh start beat），Beat 未启动则完全不会自动爬取。 +PERCEPTION_CRAWL_INTERVAL_SECONDS=21600 # ===== API配置 ===== API_HOST=0.0.0.0 @@ -125,5 +132,18 @@ CORS_ALLOW_ORIGINS=http://localhost:5173 HYDE_ENABLED=true HYDE_MAX_TOKENS=200 HYDE_LLM_PROVIDER=qwen -HYDE_LLM_MODEL=qwen3.5-flash +HYDE_LLM_MODEL=qwen3.6-flash + +# ===== MCP 服务配置 ===== +# MCP SDK 在传输层绑定回环地址时会自动启用 DNS 重绑定防护：Host 头不在下表内的 +# 请求一律返回 HTTP 421，且发生在进入工具逻辑之前。部署在 6.86.80.9 必须显式列出 +# 该地址，否则所有远程 MCP 客户端（Claude Desktop / IDE 等）100% 连不上。 +# 语法：`:*` 后缀匹配任意端口；填 `*` 表示彻底关闭该防护（不推荐）。 +MCP_ALLOWED_HOSTS=6.86.80.9:*,127.0.0.1:*,localhost:*,[::1]:* + +# 系统状态页 MCP 卡片展示、以及"复制接入配置"按钮写入的对外访问地址。 +# 留空则由后端从请求 Host 头推导；但前端经 Vite 代理（changeOrigin: true）转发后 +# Host 会被改写成 API_HOST:API_PORT，推导结果是 0.0.0.0/127.0.0.1，客户端无法使用， +# 因此远程部署必须显式指定。结尾的斜杠不能省略。 +MCP_PUBLIC_URL=http://6.86.80.9:8000/mcp/ \ No newline at end of file diff --git a/.env.example b/.env.example index 69721fd..bd1bb25 100644 --- a/.env.example +++ b/.env.example @@ -60,9 +60,16 @@ DOCUMENT_REPOSITORY_BACKEND=json USE_CELERY_WORKER=false # ===== 法规感知爬取配置 ===== +# 单次 HTTP 请求超时（秒），含正文抓取（fetch_full_text）。 PERCEPTION_CRAWL_TIMEOUT_SECONDS=120 +# 每个数据源单次爬取的最大条目数。 PERCEPTION_MAX_EVENTS_PER_SOURCE=100 -PERCEPTION_DIFF_SIMILARITY_THRESHOLD=0.85 +# 变更判定的次要闸门：段落改动字符占比达到该阈值才送 LLM 分类。 +# 数字变化（如 30米->20米）或情态词变化（应当/宜/不得等）无视此阈值，始终判定为显著变更。 +PERCEPTION_DIFF_MIN_CHANGE_RATIO=0.02 +# 定时全量爬取的执行间隔（秒），默认 21600 = 6 小时。 +# 仅当 Celery Beat 进程在运行时才生效（./dev.sh start beat），Beat 未启动则完全不会自动爬取。 +PERCEPTION_CRAWL_INTERVAL_SECONDS=21600 # ===== 阿里云文档解析 ===== ALIBABA_ACCESS_KEY_ID=your_aliyun_access_key_id @@ -147,7 +154,7 @@ HYDE_ENABLED=true HYDE_MAX_TOKENS=200 # ?????? LLM;??????????????? HYDE_LLM_PROVIDER=qwen -HYDE_LLM_MODEL=qwen3.5-flash +HYDE_LLM_MODEL=qwen3.6-flash # ===== Agentic RAG 配置 (P0-1) ===== # 以下参数控制 /api/v1/agent/agentic/stream 多步推理管线 diff --git a/backend/app/api/routes/perception.py b/backend/app/api/routes/perception.py index e398e49..a5af52f 100644 --- a/backend/app/api/routes/perception.py +++ b/backend/app/api/routes/perception.py @@ -7,7 +7,12 @@ import json from fastapi import APIRouter, Depends, Query from fastapi.responses import StreamingResponse -from app.shared.bootstrap import get_crawl_service, get_event_store, get_perception_service +from app.shared.bootstrap import ( + get_crawl_service, + get_event_store, + get_notification_store, + get_perception_service, +) from app.api.dependencies.auth import get_current_user from app.domain.auth.models import UserClaims from app.shared.async_utils import iter_in_thread @@ -141,3 +146,26 @@ async def get_event_diff(event_id: str): "previous_hash": event.get("previous_hash"), "content_hash": event.get("content_hash"), } + + +@router.get("/notifications") +async def list_notifications( + limit: int = Query(default=20, ge=1, le=100), + current_user: UserClaims = Depends(get_current_user), +): + """Return the newest in-app notifications plus this user's unread count. + + Every logged-in user sees the same broadcast feed — there is no per-role + or per-topic subscription. "read" per item and the aggregate unread_count + both reflect only the calling user's own read receipts. + """ + store = get_notification_store() + items = store.list_for_user(current_user.user_id, limit=limit) + return {"items": items, "unread_count": store.unread_count(current_user.user_id)} + + +@router.post("/notifications/read") +async def mark_notifications_read(current_user: UserClaims = Depends(get_current_user)): + """Mark every currently-unread notification read for the calling user.""" + marked = get_notification_store().mark_all_read(current_user.user_id) + return {"marked": marked} diff --git a/backend/app/application/agent/hyde_expander.py b/backend/app/application/agent/hyde_expander.py index 8b19e28..8312d6a 100644 --- a/backend/app/application/agent/hyde_expander.py +++ b/backend/app/application/agent/hyde_expander.py @@ -73,7 +73,7 @@ class HyDEExpander: return query # Use the dedicated HyDE model when configured; fall back to main LLM. - # A lightweight model (e.g. qwen3.5-flash) is sufficient for generating + # A lightweight model (e.g. qwen3.6-flash) is sufficient for generating # a short hypothetical passage and significantly reduces cost + latency. provider = settings.hyde_llm_provider or settings.llm_provider model = settings.hyde_llm_model or settings.llm_model diff --git a/backend/app/application/perception/crawl_service.py b/backend/app/application/perception/crawl_service.py index afcc452..6b9ce48 100644 --- a/backend/app/application/perception/crawl_service.py +++ b/backend/app/application/perception/crawl_service.py @@ -7,9 +7,13 @@ from typing import Any, Generator from loguru import logger +from app.config.settings import settings +from app.domain.documents import ParsedDocument from app.infrastructure.perception.base_event_store import BaseEventStore +from app.infrastructure.perception.base_notification_store import BaseNotificationStore from app.infrastructure.perception.crawlers.base import BaseCrawler, RawEvent from app.infrastructure.perception.llm_pipeline import LlmPipeline +from app.infrastructure.parser.local_chunk_builder import LocalRegulationChunkBuilder def _event_id(source: str, standard_code: str) -> str: @@ -21,7 +25,68 @@ def _content_hash(raw_text: str) -> str: return hashlib.sha256(raw_text.encode()).hexdigest() -def _raw_to_dict(raw: RawEvent, event_id: str, content_hash: str) -> dict: +def _is_significant(changed_sections: list[dict]) -> bool: + """Report whether any changed section is worth notifying every user about. + + changed_sections legitimately includes cosmetic edits — the differ + (subproject 1) still reports a fixed typo or a dropped trailing period as + a change, it just doesn't send those to the LLM. Broadcasting a + notification for every cosmetic edit would train people to ignore it, so + this reuses the same significance test the differ's own LLM gate applies: + a numeric or deontic change, or a whole paragraph added or removed. + """ + return any( + section.get("numeric_changed") + or section.get("deontic_changed") + or section.get("change_type") in ("added", "removed") + for section in changed_sections + ) + + +def _index_in_knowledge_base(event: dict, *, embedding_provider: Any, vector_index: Any) -> None: + """Chunk, embed, and upsert a regulation's text into the shared knowledge base. + + Always uses the local markdown chunker, never get_chunk_builder() — that + bootstrap function resolves to AliyunVectorChunkBuilder when + settings.chunk_backend == "aliyun" (the deployed value), which consumes + Aliyun DocMind's structured parse output. Crawled text has no such parse + output; it is already plain text (trafilatura, subproject 1), which is + exactly what LocalRegulationChunkBuilder chunks directly. + + delete_by_document runs unconditionally before upsert — a no-op for a + brand-new event, and the only way to keep a changed regulation from + leaving its superseded text retrievable alongside the new version. + """ + vector_index.delete_by_document(event["id"]) + + parsed = ParsedDocument( + doc_id=event["id"], + doc_name=event.get("title", ""), + structure_nodes=[], + semantic_blocks=[], + vector_chunks=[], + parser_name="perception_crawl", + raw_text=event.get("raw_text") or "", + ) + builder = LocalRegulationChunkBuilder( + chunk_size=settings.chunk_size, chunk_overlap=settings.chunk_overlap, + ) + chunks = builder.build( + parsed_document=parsed, + # regulation_type/version fill the same slots a manually uploaded + # document's form fields would, so the two intake paths are + # indistinguishable to retrieval and compliance analysis. + regulation_type=event.get("category", ""), + version=event.get("standard_code", ""), + ) + if not chunks: + return + + vectors = embedding_provider.embed_texts([c.embedding_text for c in chunks]) + vector_index.upsert(chunks, vectors) + + +def _raw_to_dict(raw: RawEvent, event_id: str, content_hash: str, raw_text: str) -> dict: return { "id": event_id, "source": raw.source, @@ -36,6 +101,10 @@ def _raw_to_dict(raw: RawEvent, event_id: str, content_hash: str) -> dict: "effective_at": raw.effective_at, "category": raw.category, "tags": raw.tags, + # Persisted so the next crawl has a baseline to diff against. Without + # this the change detector has nothing to compare and every update + # looks like a first sighting. + "raw_text": raw_text, "content_hash": content_hash, "previous_hash": None, } @@ -50,11 +119,17 @@ class CrawlService: event_store: BaseEventStore, llm_pipeline: LlmPipeline, retrieval_service: Any, + notification_store: BaseNotificationStore, + embedding_provider: Any, + vector_index: Any, ) -> None: self._crawlers = crawlers self._store = event_store self._pipeline = llm_pipeline self._retrieval = retrieval_service + self._notifications = notification_store + self._embedding_provider = embedding_provider + self._vector_index = vector_index def run_crawl( self, sources: list[str] | None = None @@ -72,7 +147,7 @@ class CrawlService: yield {"event": "progress", "data": {"source": source_key, "stage": "fetching"}} try: - raw_events = crawler.fetch(limit=100) + raw_events = crawler.fetch(limit=settings.perception_max_events_per_source) except Exception as exc: logger.exception("Crawler failed source={}", source_key) yield {"event": "error", "data": {"source": source_key, "message": str(exc)}} @@ -88,17 +163,21 @@ class CrawlService: for raw in raw_events: eid = _event_id(raw.source, raw.standard_code) - new_hash = _content_hash(raw.raw_text or raw.title) + # List pages carry only a code and a title, which is not enough + # to detect a change in the regulation itself. Fetch the body, + # degrading to whatever the list page gave us if that fails. + body_text = crawler.fetch_full_text(raw.full_text_url) or raw.raw_text or raw.title + new_hash = _content_hash(body_text) existing = self._store.get(eid) if existing and existing.get("content_hash") == new_hash: continue is_update = existing is not None - old_text = existing.get("summary", "") if is_update else "" + old_body = existing.get("raw_text") or "" if is_update else "" previous_hash = existing.get("content_hash") if is_update else None - event_dict = _raw_to_dict(raw, eid, new_hash) + event_dict = _raw_to_dict(raw, eid, new_hash, body_text) event_dict["previous_hash"] = previous_hash try: @@ -113,9 +192,11 @@ class CrawlService: except Exception as exc: logger.warning("Impact assessment failed id={} err={}", eid, exc) - if is_update and old_text and raw.raw_text: + # Events stored before raw_text was persisted have no baseline, + # so they are treated as a first sighting and establish one now. + if is_update and old_body and body_text: try: - diff = self._pipeline.compute_diff(old_text, raw.raw_text) + diff = self._pipeline.compute_diff(old_body, body_text) event_dict["change_summary"] = diff.get("change_summary") event_dict["changed_sections"] = diff.get("changed_sections") except Exception as exc: @@ -123,6 +204,33 @@ class CrawlService: self._store.upsert(event_dict) + should_index = not is_update or _is_significant(event_dict.get("changed_sections") or []) + + try: + if not is_update: + self._notifications.create( + event_id=eid, kind="new", title=raw.title, + impact_level=event_dict.get("impact_level"), summary=None, + ) + elif should_index: # significant change, already computed above + self._notifications.create( + event_id=eid, kind="changed", title=raw.title, + impact_level=event_dict.get("impact_level"), + summary=event_dict.get("change_summary"), + ) + except Exception as exc: + logger.warning("Notification create failed id={} err={}", eid, exc) + + if should_index: + try: + _index_in_knowledge_base( + event_dict, + embedding_provider=self._embedding_provider, + vector_index=self._vector_index, + ) + except Exception as exc: + logger.warning("Knowledge base indexing failed id={} err={}", eid, exc) + if is_update: updated_count += 1 else: diff --git a/backend/app/config/settings.py b/backend/app/config/settings.py index dec2344..771ff64 100644 --- a/backend/app/config/settings.py +++ b/backend/app/config/settings.py @@ -94,9 +94,21 @@ class Settings(BaseSettings): perception_max_events_per_source: int = Field( default=100, description="Maximum events fetched per source per crawl run." ) - perception_diff_similarity_threshold: float = Field( - default=0.85, - description="Cosine similarity below which a paragraph is flagged as changed.", + perception_diff_min_change_ratio: float = Field( + default=0.02, + description=( + "Fraction of characters that must differ before an otherwise " + "unremarkable paragraph edit is worth an LLM classification call. " + "Numeric and deontic changes bypass this gate entirely." + ), + ) + perception_crawl_interval_seconds: int = Field( + default=21600, + description=( + "How often Celery Beat runs the scheduled crawl-all-sources task, " + "in seconds. Default 21600 = 6 hours. Only takes effect when a " + "Beat process is running (./dev.sh start beat)." + ), ) # Keep configuration setup explicit so runtime behavior is easy to reason about. @@ -117,7 +129,7 @@ class Settings(BaseSettings): # Keep configuration setup explicit so runtime behavior is easy to reason about. qwen_api_key: str = Field(default="", description="Qwen API密钥") qwen_base_url: str = Field(default="http://6.86.80.4:30080/v1", description="Qwen API地址") - qwen_model: str = Field(default="qwen3.5-flash", description="Qwen文本模型") + qwen_model: str = Field(default="qwen3.6-flash", description="Qwen文本模型") qwen_vl_model: str = Field(default="qwen3-vl-plus", description="Qwen视觉模型") # Keep configuration setup explicit so runtime behavior is easy to reason about. diff --git a/backend/app/infrastructure/perception/base_notification_store.py b/backend/app/infrastructure/perception/base_notification_store.py new file mode 100644 index 0000000..59b6026 --- /dev/null +++ b/backend/app/infrastructure/perception/base_notification_store.py @@ -0,0 +1,47 @@ +"""Abstract base class for in-app regulatory-signal notifications. + +A notification is created once per triggering event (a brand-new regulation, +or a significant change to an existing one) and broadcast to every logged-in +user. There is no per-user subscription targeting — see the design doc for why. +Per-user "read" state is tracked separately from the notification itself, so +one notification row serves every user rather than being fanned out on create. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod + + +class BaseNotificationStore(ABC): + """Port interface for perception notification persistence.""" + + @abstractmethod + def create( + self, + *, + event_id: str, + kind: str, + title: str, + impact_level: str | None, + summary: str | None, + ) -> None: + """Record a new notification. kind is 'new' or 'changed'.""" + + @abstractmethod + def list_for_user(self, user_id: str, limit: int = 20) -> list[dict]: + """Return the most recent notifications, newest first. + + Each item includes a "read" boolean reflecting whether `user_id` has + marked it read. + """ + + @abstractmethod + def unread_count(self, user_id: str) -> int: + """Return how many notifications `user_id` has not yet read.""" + + @abstractmethod + def mark_all_read(self, user_id: str) -> int: + """Mark every currently-unread notification read for `user_id`. + + Returns the number of notifications newly marked. + """ diff --git a/backend/app/infrastructure/perception/crawlers/base.py b/backend/app/infrastructure/perception/crawlers/base.py index b359a5c..b34edb4 100644 --- a/backend/app/infrastructure/perception/crawlers/base.py +++ b/backend/app/infrastructure/perception/crawlers/base.py @@ -5,6 +5,12 @@ from __future__ import annotations from abc import ABC, abstractmethod from dataclasses import dataclass, field +import httpx +import trafilatura +from loguru import logger + +from app.config.settings import settings + @dataclass class RawEvent: @@ -21,7 +27,10 @@ class RawEvent: effective_at: str | None category: str tags: list[str] = field(default_factory=list) - raw_text: str = "" # full crawled text for hashing + LLM + # Whatever text the list page yields. CrawlService upgrades this by calling + # fetch_full_text(full_text_url); this value is the fallback when that + # fails. Used for change hashing and for the version diff. + raw_text: str = "" class BaseCrawler(ABC): @@ -30,3 +39,37 @@ class BaseCrawler(ABC): @abstractmethod def fetch(self, limit: int = 50) -> list[RawEvent]: """Fetch up to `limit` recent events from the data source.""" + + def fetch_full_text(self, url: str) -> str: + """Download a regulation detail page and extract its body text. + + Change detection is only as good as the text it compares, and list + pages carry nothing but a standard code and a title. This default + implementation serves all current sources; a source that needs PDF + extraction or authentication overrides this one method. + + Returns an empty string on any failure rather than raising, so one + unreachable page cannot abort a whole crawl run. The caller decides how + to degrade. + """ + if not url: + return "" + try: + response = httpx.get( + url, + timeout=settings.perception_crawl_timeout_seconds, + follow_redirects=True, + ) + response.raise_for_status() + except Exception as exc: # noqa: BLE001 - any transport error degrades the same way + logger.warning("Full-text fetch failed url={} err={}", url, exc) + return "" + + # trafilatura scores 0.92 F1 on government pages against 0.78 for + # readability-lxml, and handles CJK content; include_tables matters + # because regulatory limits are frequently tabulated. + extracted = trafilatura.extract(response.text, include_tables=True) + if not extracted: + logger.warning("Full-text extraction returned nothing url={}", url) + return "" + return extracted.strip() diff --git a/backend/app/infrastructure/perception/crawlers/catarc_crawler.py b/backend/app/infrastructure/perception/crawlers/catarc_crawler.py index 3ff5dd9..0683889 100644 --- a/backend/app/infrastructure/perception/crawlers/catarc_crawler.py +++ b/backend/app/infrastructure/perception/crawlers/catarc_crawler.py @@ -8,6 +8,7 @@ import httpx from bs4 import BeautifulSoup from loguru import logger +from app.config.settings import settings from app.infrastructure.perception.crawlers.base import BaseCrawler, RawEvent from ._utils import extract_tags, parse_date @@ -33,7 +34,11 @@ class CatarcCrawler(BaseCrawler): while len(events) < limit and page <= max_pages: url = f"{_BASE_URL}?page={page}" try: - resp = httpx.get(url, timeout=30, follow_redirects=True) + resp = httpx.get( + url, + timeout=settings.perception_crawl_timeout_seconds, + follow_redirects=True, + ) resp.raise_for_status() except Exception as exc: logger.warning("CATARC fetch failed page={} err={}", page, exc) diff --git a/backend/app/infrastructure/perception/crawlers/eurlex_crawler.py b/backend/app/infrastructure/perception/crawlers/eurlex_crawler.py index 3f5fdd2..5658e11 100644 --- a/backend/app/infrastructure/perception/crawlers/eurlex_crawler.py +++ b/backend/app/infrastructure/perception/crawlers/eurlex_crawler.py @@ -9,6 +9,7 @@ import httpx from bs4 import BeautifulSoup from loguru import logger +from app.config.settings import settings from app.infrastructure.perception.crawlers.base import BaseCrawler, RawEvent from ._utils import parse_date @@ -53,7 +54,11 @@ class EurlexCrawler(BaseCrawler): if len(events) >= limit: break try: - resp = httpx.get(rss_url, timeout=30, follow_redirects=True) + resp = httpx.get( + rss_url, + timeout=settings.perception_crawl_timeout_seconds, + follow_redirects=True, + ) resp.raise_for_status() except Exception as exc: logger.warning("EUR-Lex RSS fetch failed url={} err={}", rss_url, exc) diff --git a/backend/app/infrastructure/perception/crawlers/guobiao_crawler.py b/backend/app/infrastructure/perception/crawlers/guobiao_crawler.py index 77c5b7b..d30178b 100644 --- a/backend/app/infrastructure/perception/crawlers/guobiao_crawler.py +++ b/backend/app/infrastructure/perception/crawlers/guobiao_crawler.py @@ -5,6 +5,7 @@ from __future__ import annotations import httpx from loguru import logger +from app.config.settings import settings from app.infrastructure.perception.crawlers.base import BaseCrawler, RawEvent from ._utils import extract_tags, parse_date @@ -22,7 +23,12 @@ def _fetch_page(std_type: int, page: int, page_size: int) -> list[dict]: "p.p7": page_size, } try: - resp = httpx.get(_BASE_URL, params=params, headers=_HEADERS, timeout=30) + resp = httpx.get( + _BASE_URL, + params=params, + headers=_HEADERS, + timeout=settings.perception_crawl_timeout_seconds, + ) resp.raise_for_status() data = resp.json() return data.get("rows", []) or [] diff --git a/backend/app/infrastructure/perception/llm_pipeline.py b/backend/app/infrastructure/perception/llm_pipeline.py index 37cdce5..629d804 100644 --- a/backend/app/infrastructure/perception/llm_pipeline.py +++ b/backend/app/infrastructure/perception/llm_pipeline.py @@ -3,14 +3,14 @@ from __future__ import annotations import json -import math from typing import Any from loguru import logger from app.config.settings import settings -from app.infrastructure.embedding.openai_compatible_embedding_provider import ( - OpenAICompatibleEmbeddingProvider, +from app.infrastructure.perception.regulation_differ import ( + ParagraphChange, + RegulationDiffer, ) from app.services.llm.llm_factory import get_llm_client @@ -27,21 +27,31 @@ _ASSESS_SYSTEM = ( ) _DIFF_SYSTEM = ( - "You are a regulatory change analyst. Given an old and new version of a regulation paragraph, " - "classify the type of change and summarise it. " - "Return JSON only: {\"change_type\": \"tightened|relaxed|added|removed\", \"summary\": \"...\"}" + "You are a regulatory change analyst. You are given the OLD and NEW version of " + "one regulation paragraph, with the exact edits marked removed and " + "added. Classify the legal effect of the change. " + "Return JSON only: {\"change_type\": \"tightened|relaxed|numeric|clarified|scope\", " + "\"legal_effect\": \"one sentence on what this means for compliance\"}" ) -_SIMILARITY_THRESHOLD = 0.85 +def _marked_diff(change: ParagraphChange) -> str: + """Render a paragraph change with the exact edits marked for the model. -def _cosine(a: list[float], b: list[float]) -> float: - dot = sum(x * y for x, y in zip(a, b)) - norm_a = math.sqrt(sum(x * x for x in a)) - norm_b = math.sqrt(sum(x * x for x in b)) - if norm_a == 0 or norm_b == 0: - return 0.0 - return dot / (norm_a * norm_b) + The model is shown where the edit is rather than being asked to find it, + and is never asked to reproduce the changed text — the differ already + computed those spans exactly, so there is nothing for the model to + hallucinate. + """ + marked = "".join( + text if op == 0 else (f"{text}" if op < 0 else f"{text}") + for op, text in change.diff_spans + ) + return ( + f"OLD: {change.old_text[:500]}\n" + f"NEW: {change.new_text[:500]}\n" + f"MARKED: {marked[:800]}" + ) def _llm_json(client: Any, messages: list[dict]) -> Any: @@ -67,7 +77,9 @@ class LlmPipeline: provider=settings.llm_provider, model=settings.llm_model, ) - self._embedder = OpenAICompatibleEmbeddingProvider() + # Change detection is deterministic; the differ needs no model and no + # network, so the pipeline no longer constructs an embedding provider. + self._differ = RegulationDiffer() # ------------------------------------------------------------------ # Step 1: Structure extraction @@ -166,76 +178,68 @@ For each document, assess impact and recommend action. Return JSON array: return doc_excerpts # ------------------------------------------------------------------ - # Step 3: Semantic diff + # Step 3: Deterministic diff with gated LLM classification # ------------------------------------------------------------------ def compute_diff(self, old_text: str, new_text: str) -> dict: - """Compare old and new regulation text; return changed sections and summary.""" - old_paras = [p.strip() for p in old_text.split("\n") if p.strip()] - new_paras = [p.strip() for p in new_text.split("\n") if p.strip()] + """Compare old and new regulation text; return changed sections and summary. - if not old_paras or not new_paras: - return {"changed_sections": [], "change_summary": "No comparable text."} + Detection is deterministic — see regulation_differ for why embedding + similarity was removed. The LLM is called only for paragraphs the + differ marked significant, and only to explain the legal effect of a + change that has already been located exactly. + """ + changes = self._differ.diff(old_text, new_text) + if not changes: + return { + "changed_sections": [], + "change_summary": "No substantive changes detected between versions.", + } - all_paras = old_paras + new_paras - try: - all_embeddings = self._embedder.embed_texts(all_paras) - except Exception as exc: - logger.warning("Embedding for diff failed: {}", exc) - return {"changed_sections": [], "change_summary": "Diff unavailable (embedding error)."} + changed_sections = [self._describe(change) for change in changes] - old_embeddings = all_embeddings[: len(old_paras)] - new_embeddings = all_embeddings[len(old_paras):] - - changed_sections: list[dict] = [] - max_len = max(len(old_paras), len(new_paras)) - - for i in range(max_len): - if i >= len(old_paras): - # New paragraph added - changed_sections.append({ - "old_text": "", - "new_text": new_paras[i][:300], - "similarity": 0.0, - "change_type": "added", - "summary": "New paragraph added.", - }) - continue - if i >= len(new_paras): - # Old paragraph removed - changed_sections.append({ - "old_text": old_paras[i][:300], - "new_text": "", - "similarity": 0.0, - "change_type": "removed", - "summary": "Paragraph removed.", - }) - continue - # Both exist — compare via embeddings - sim = _cosine(old_embeddings[i], new_embeddings[i]) - if sim < _SIMILARITY_THRESHOLD: - messages = [ - {"role": "system", "content": _DIFF_SYSTEM}, - {"role": "user", "content": f"OLD: {old_paras[i][:500]}\nNEW: {new_paras[i][:500]}"}, - ] - classification = _llm_json(self._client, messages) or {} - changed_sections.append({ - "old_text": old_paras[i][:300], - "new_text": new_paras[i][:300], - "similarity": round(sim, 3), - "change_type": classification.get("change_type", "modified"), - "summary": classification.get("summary", ""), - }) - - if not changed_sections: - change_summary = "No substantive changes detected between versions." - else: - types = [s["change_type"] for s in changed_sections] - change_summary = ( - f"{len(changed_sections)} paragraph(s) changed: " - + ", ".join(f"{t}" for t in set(types)) - + ". " - + (changed_sections[0].get("summary", "") if changed_sections else "") - ) + types = sorted({section["change_type"] for section in changed_sections}) + gated = sum(1 for change in changes if change.needs_llm) + change_summary = ( + f"{len(changed_sections)} paragraph(s) changed ({', '.join(types)}); " + f"{gated} significant. " + + (changed_sections[0].get("summary") or "") + ).strip() return {"changed_sections": changed_sections, "change_summary": change_summary} + + def _describe(self, change: ParagraphChange) -> dict: + """Turn one detected change into the API payload, classifying if warranted.""" + section = { + "old_text": change.old_text[:300], + "new_text": change.new_text[:300], + "change_type": change.change_type, + "change_ratio": round(change.change_ratio, 3), + "numeric_changed": change.numeric_changed, + "deontic_changed": change.deontic_changed, + "summary": "", + } + + if not change.needs_llm: + return section + + classification = _llm_json( + self._client, + [ + {"role": "system", "content": _DIFF_SYSTEM}, + {"role": "user", "content": _marked_diff(change)}, + ], + ) + if isinstance(classification, dict): + section["change_type"] = classification.get("change_type") or change.change_type + section["summary"] = classification.get("legal_effect") or "" + # A failed or malformed model response must not discard a change that + # deterministic analysis already proved real; the section keeps its + # spans, flags, and alignment-derived type with an empty summary. + + if change.numeric_changed: + # Models routinely label a changed threshold as "clarified". The + # deterministic pass already knows a number moved, so it wins. + section["change_type"] = "numeric" + + return section diff --git a/backend/app/infrastructure/perception/mock_notification_store.py b/backend/app/infrastructure/perception/mock_notification_store.py new file mode 100644 index 0000000..083c78d --- /dev/null +++ b/backend/app/infrastructure/perception/mock_notification_store.py @@ -0,0 +1,66 @@ +"""In-memory notification store used when Postgres is not configured. + +Mirrors MockEventStore's role for BaseEventStore: keeps the feature usable in +local dev and in tests without a live database, and matches +DOCUMENT_REPOSITORY_BACKEND's existing Mock/Postgres split. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from app.infrastructure.perception.base_notification_store import BaseNotificationStore + + +class MockNotificationStore(BaseNotificationStore): + """Dict-backed notification store. Data does not survive a process restart.""" + + def __init__(self) -> None: + """Start with an empty feed and no read receipts.""" + self._notifications: list[dict] = [] + self._next_id = 1 + # (notification_id, user_id) pairs — presence means read. + self._reads: set[tuple[int, str]] = set() + + def create( + self, + *, + event_id: str, + kind: str, + title: str, + impact_level: str | None, + summary: str | None, + ) -> None: + """Append a notification with an auto-incrementing id.""" + self._notifications.append({ + "id": self._next_id, + "event_id": event_id, + "kind": kind, + "title": title, + "impact_level": impact_level, + "summary": summary, + "created_at": datetime.now(UTC).isoformat(), + }) + self._next_id += 1 + + def list_for_user(self, user_id: str, limit: int = 20) -> list[dict]: + """Return the newest `limit` notifications with this user's read state.""" + ordered = sorted(self._notifications, key=lambda n: n["id"], reverse=True) + return [ + {**n, "read": (n["id"], user_id) in self._reads} + for n in ordered[:limit] + ] + + def unread_count(self, user_id: str) -> int: + """Count notifications this user has not yet read.""" + return sum(1 for n in self._notifications if (n["id"], user_id) not in self._reads) + + def mark_all_read(self, user_id: str) -> int: + """Add a read receipt for every currently-unread notification.""" + marked = 0 + for n in self._notifications: + key = (n["id"], user_id) + if key not in self._reads: + self._reads.add(key) + marked += 1 + return marked diff --git a/backend/app/infrastructure/perception/postgres_event_store.py b/backend/app/infrastructure/perception/postgres_event_store.py index 4782ae0..67453a6 100644 --- a/backend/app/infrastructure/perception/postgres_event_store.py +++ b/backend/app/infrastructure/perception/postgres_event_store.py @@ -40,7 +40,8 @@ CREATE TABLE IF NOT EXISTS regulation_events ( affected_docs JSONB, crawled_at TIMESTAMPTZ DEFAULT now(), processed_at TIMESTAMPTZ, - raw_storage_key TEXT + raw_storage_key TEXT, + raw_text TEXT ); CREATE INDEX IF NOT EXISTS reg_events_source_date ON regulation_events (source, published_at DESC); @@ -48,12 +49,16 @@ CREATE INDEX IF NOT EXISTS reg_events_impact_date ON regulation_events (impact_level, published_at DESC); """ +_ADD_COLUMNS = """ +ALTER TABLE regulation_events ADD COLUMN IF NOT EXISTS raw_text TEXT; +""" + _ALL_COLUMNS = ( "id", "source", "source_label", "standard_code", "title", "summary", "full_text_url", "status", "impact_level", "published_at", "effective_at", "category", "tags", "obligations", "deadlines", "scope", "penalties", "content_hash", "previous_hash", "change_summary", "changed_sections", - "affected_docs", "crawled_at", "processed_at", "raw_storage_key", + "affected_docs", "crawled_at", "processed_at", "raw_storage_key", "raw_text", ) @@ -97,6 +102,10 @@ class PostgresEventStore(BaseEventStore): try: with conn.cursor() as cur: cur.execute(_CREATE_TABLE) + # CREATE TABLE IF NOT EXISTS is a no-op on deployments that + # already have this table, so new columns must be added + # explicitly or existing installations silently lack them. + cur.execute(_ADD_COLUMNS) conn.commit() except Exception: conn.rollback() diff --git a/backend/app/infrastructure/perception/postgres_notification_store.py b/backend/app/infrastructure/perception/postgres_notification_store.py new file mode 100644 index 0000000..55d508f --- /dev/null +++ b/backend/app/infrastructure/perception/postgres_notification_store.py @@ -0,0 +1,157 @@ +"""PostgreSQL-backed notification store. + +One row per triggering event, shared by every user; a separate read-receipt +table tracks per-user read state so broadcasting to everyone needs no fan-out +insert per user. See base_notification_store.py for the port contract and the +design doc for why this shape was chosen over per-user subscriptions. +""" + +from __future__ import annotations + +from contextlib import contextmanager +from typing import Any + +import psycopg2 +import psycopg2.extras +from psycopg2.pool import ThreadedConnectionPool + +from app.config.settings import settings +from app.infrastructure.perception.base_notification_store import BaseNotificationStore + +_CREATE_TABLES = """ +CREATE TABLE IF NOT EXISTS perception_notifications ( + id SERIAL PRIMARY KEY, + event_id TEXT NOT NULL REFERENCES regulation_events(id) ON DELETE CASCADE, + kind TEXT NOT NULL, + title TEXT NOT NULL, + impact_level TEXT, + summary TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +CREATE TABLE IF NOT EXISTS perception_notification_reads ( + notification_id INTEGER NOT NULL REFERENCES perception_notifications(id) ON DELETE CASCADE, + user_id TEXT NOT NULL, + read_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (notification_id, user_id) +); +CREATE INDEX IF NOT EXISTS perception_notif_created + ON perception_notifications (created_at DESC); +""" + + +def _row_to_dict(row: dict[str, Any]) -> dict: + """Convert a psycopg2 RealDictRow to a plain dict with an ISO timestamp.""" + d = dict(row) + if d.get("created_at") is not None: + d["created_at"] = d["created_at"].isoformat() + return d + + +class PostgresNotificationStore(BaseNotificationStore): + """Notification store backed by PostgreSQL.""" + + def __init__(self) -> None: + """Open a connection pool and ensure both tables exist.""" + self._pool = ThreadedConnectionPool( + minconn=1, + maxconn=5, + host=settings.postgres_host, + port=settings.postgres_port, + user=settings.postgres_user, + password=settings.postgres_password, + dbname=settings.postgres_db, + ) + self._ensure_schema() + + def _ensure_schema(self) -> None: + with self._conn() as conn: + try: + with conn.cursor() as cur: + cur.execute(_CREATE_TABLES) + conn.commit() + except Exception: + conn.rollback() + raise + + @contextmanager + def _conn(self): + conn = None + try: + conn = self._pool.getconn() + yield conn + finally: + if conn is not None: + self._pool.putconn(conn) + + def create( + self, + *, + event_id: str, + kind: str, + title: str, + impact_level: str | None, + summary: str | None, + ) -> None: + """Insert one notification row for the triggering event.""" + with self._conn() as conn: + with conn.cursor() as cur: + cur.execute( + "INSERT INTO perception_notifications " + "(event_id, kind, title, impact_level, summary) " + "VALUES (%s, %s, %s, %s, %s)", + (event_id, kind, title, impact_level, summary), + ) + conn.commit() + + def list_for_user(self, user_id: str, limit: int = 20) -> list[dict]: + """Return the newest notifications with this user's read state joined in.""" + with self._conn() as conn: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute( + """ + SELECT n.*, (r.user_id IS NOT NULL) AS read + FROM perception_notifications n + LEFT JOIN perception_notification_reads r + ON r.notification_id = n.id AND r.user_id = %s + ORDER BY n.created_at DESC + LIMIT %s + """, + (user_id, limit), + ) + return [_row_to_dict(r) for r in cur.fetchall()] + + def unread_count(self, user_id: str) -> int: + """Count notifications with no read receipt for this user.""" + with self._conn() as conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT COUNT(*) FROM perception_notifications n + WHERE NOT EXISTS ( + SELECT 1 FROM perception_notification_reads r + WHERE r.notification_id = n.id AND r.user_id = %s + ) + """, + (user_id,), + ) + return cur.fetchone()[0] + + def mark_all_read(self, user_id: str) -> int: + """Insert a read receipt for every notification this user hasn't read.""" + with self._conn() as conn: + with conn.cursor() as cur: + cur.execute( + """ + INSERT INTO perception_notification_reads (notification_id, user_id) + SELECT n.id, %s FROM perception_notifications n + WHERE NOT EXISTS ( + SELECT 1 FROM perception_notification_reads r + WHERE r.notification_id = n.id AND r.user_id = %s + ) + ON CONFLICT (notification_id, user_id) DO NOTHING + """, + (user_id, user_id), + ) + marked = cur.rowcount + conn.commit() + return marked diff --git a/backend/app/infrastructure/perception/regulation_differ.py b/backend/app/infrastructure/perception/regulation_differ.py new file mode 100644 index 0000000..4eb984c --- /dev/null +++ b/backend/app/infrastructure/perception/regulation_differ.py @@ -0,0 +1,224 @@ +"""Deterministic change detection between two versions of a regulation. + +This module deliberately contains no LLM call, no network access, and no +embedding lookup. It exists because the previous implementation decided whether +a paragraph had changed by comparing embedding cosine similarity against a 0.85 +threshold, which is blind to exactly the edits that matter in regulation. +Measured against the deployed text-embedding-v3 gateway, tightening a braking +limit from 30米 to 20米 scores 0.9153 and relaxing 应当 to 宜 scores 0.9162 — +both far above the threshold, both undetected — while an entirely unrelated +clause scores 0.6862 and is the only thing that fires. Cosine is scale +invariant, so it cannot represent a change in magnitude or certainty +(arXiv:2403.05440, ACM Web Conference 2024); no threshold recovers the signal. + +The replacement is the production consensus for legal text: align paragraphs +with a longest-common-subsequence matcher, run a literal character diff on the +aligned pairs, and let cheap deterministic rules decide whether a change is +significant enough to spend an LLM call classifying. +""" + +from __future__ import annotations + +import re +import unicodedata +from dataclasses import dataclass, field + +from diff_match_patch import diff_match_patch +from difflib import SequenceMatcher + +from app.config.settings import settings + +# Chinese regulatory drafting uses a small, near-unambiguous set of deontic +# markers, so a regex pre-pass identifies legally significant edits without an +# LLM. Adding or removing any of these changes what the provision compels. +_DEONTIC_PATTERN = re.compile(r"应当|须|禁止|不得|可以|允许|宜") + +# Matches digit runs including decimals, so "30" -> "20" and "0.85" -> "0.9" +# are both treated as numeric changes. +_NUMBER_PATTERN = re.compile(r"\d+(?:\.\d+)?") + +# diff_match_patch operation codes. +_DMP_DELETE = -1 +_DMP_INSERT = 1 +_DMP_EQUAL = 0 + + +@dataclass(frozen=True) +class ParagraphChange: + """One detected difference between the old and new version of a regulation. + + `needs_llm` is the gate: it records whether this change is worth the cost of + an LLM classification call. The deterministic flags that drive it are kept + on the record so downstream code can act on them even when the LLM call + fails or is skipped. + """ + + change_type: str + old_text: str + new_text: str + numeric_changed: bool + deontic_changed: bool + change_ratio: float + needs_llm: bool + # (op, text) pairs from diff_match_patch, for rendering a redline view. + diff_spans: list[tuple[int, str]] = field(default_factory=list) + + +def _split_paragraphs(text: str) -> list[str]: + """Split regulation text into comparable units, dropping blank lines. + + ponytail: newline splitting, not clause parsing. Upgrade to 第X条 / X.X.X + segmentation only if paragraph granularity proves too coarse in practice. + """ + return [line.strip() for line in (text or "").split("\n") if line.strip()] + + +def _numbers_differ(old: str, new: str) -> bool: + """Report whether the two spans contain a different sequence of numbers.""" + return _NUMBER_PATTERN.findall(old) != _NUMBER_PATTERN.findall(new) + + +def _deontic_differs(old: str, new: str) -> bool: + """Report whether obligation markers were added, removed, or swapped.""" + return sorted(_DEONTIC_PATTERN.findall(old)) != sorted(_DEONTIC_PATTERN.findall(new)) + + +def _is_cosmetic(spans: list[tuple[int, str]]) -> bool: + """Report whether the edit touched nothing but punctuation and whitespace. + + A change ratio alone cannot answer this for Chinese regulation text. Clauses + run 20-60 characters, so deleting a single 。 is a 4% change and clears any + threshold low enough to still catch real edits in longer paragraphs. Testing + what actually changed is both cheaper and exact. + """ + changed = "".join(text for op, text in spans if op != _DMP_EQUAL) + # Unicode categories P (punctuation), Z (separator) and C (control) cover + # Chinese and ASCII punctuation plus every flavour of whitespace. + return all(unicodedata.category(char)[0] in {"P", "Z", "C"} for char in changed) + + +class RegulationDiffer: + """Align two regulation versions and classify what changed, without an LLM.""" + + def __init__(self, min_change_ratio: float | None = None) -> None: + """Store the gate threshold, defaulting to the configured value. + + The explicit argument exists so tests never depend on the deployed .env. + """ + self._min_change_ratio = ( + settings.perception_diff_min_change_ratio + if min_change_ratio is None + else min_change_ratio + ) + self._dmp = diff_match_patch() + + def diff(self, old_text: str, new_text: str) -> list[ParagraphChange]: + """Return every changed paragraph between two versions. + + Unchanged paragraphs are not returned. An empty old version means there + is no baseline to compare against — the caller's first crawl — so no + changes are reported rather than the whole document being called new. + """ + old_paras = _split_paragraphs(old_text) + new_paras = _split_paragraphs(new_text) + + if not old_paras or not new_paras: + return [] + + # autojunk=False is load-bearing: the default treats any element + # appearing in over 1% of a sequence of 200+ items as junk, and + # regulations repeat boilerplate paragraphs that alignment depends on + # as anchors. + matcher = SequenceMatcher(None, old_paras, new_paras, autojunk=False) + + changes: list[ParagraphChange] = [] + for tag, i1, i2, j1, j2 in matcher.get_opcodes(): + if tag == "equal": + continue + if tag == "insert": + changes.extend(self._added(p) for p in new_paras[j1:j2]) + elif tag == "delete": + changes.extend(self._removed(p) for p in old_paras[i1:i2]) + elif tag == "replace": + changes.extend(self._replaced(old_paras[i1:i2], new_paras[j1:j2])) + + return changes + + def _added(self, paragraph: str) -> ParagraphChange: + """Build a record for a provision present only in the new version.""" + return ParagraphChange( + change_type="added", + old_text="", + new_text=paragraph, + numeric_changed=False, + deontic_changed=bool(_DEONTIC_PATTERN.search(paragraph)), + change_ratio=1.0, + # A new provision always carries new obligations, so it is always + # worth classifying. + needs_llm=True, + diff_spans=[(_DMP_INSERT, paragraph)], + ) + + def _removed(self, paragraph: str) -> ParagraphChange: + """Build a record for a provision dropped from the new version.""" + return ParagraphChange( + change_type="removed", + old_text=paragraph, + new_text="", + numeric_changed=False, + deontic_changed=bool(_DEONTIC_PATTERN.search(paragraph)), + change_ratio=1.0, + needs_llm=True, + diff_spans=[(_DMP_DELETE, paragraph)], + ) + + def _replaced(self, old_block: list[str], new_block: list[str]) -> list[ParagraphChange]: + """Compare a run of rewritten paragraphs pairwise, reporting the remainder. + + SequenceMatcher emits `replace` for a whole run at once, and the two + sides may differ in length. Pairing by position within the run is safe + here because alignment has already established that this run as a whole + corresponds; any surplus on either side is a genuine insertion or + deletion. + """ + results: list[ParagraphChange] = [] + for index in range(max(len(old_block), len(new_block))): + if index >= len(old_block): + results.append(self._added(new_block[index])) + elif index >= len(new_block): + results.append(self._removed(old_block[index])) + else: + results.append(self._modified(old_block[index], new_block[index])) + return results + + def _modified(self, old: str, new: str) -> ParagraphChange: + """Character-diff an aligned pair and decide whether it warrants an LLM call.""" + spans = self._dmp.diff_main(old, new) + # Merges single-character edits into human-meaningful chunks so the + # redline view and the change ratio both reflect real edits. + self._dmp.diff_cleanupSemantic(spans) + + changed_chars = sum(len(text) for op, text in spans if op != _DMP_EQUAL) + denominator = max(len(old), len(new), 1) + change_ratio = changed_chars / denominator + + numeric_changed = _numbers_differ(old, new) + deontic_changed = _deontic_differs(old, new) + + # A changed limit or obligation marker is always significant no matter + # how few characters moved. Everything else must be substantive and + # clear the ratio gate to be worth a model call. + significant = numeric_changed or deontic_changed or ( + not _is_cosmetic(spans) and change_ratio >= self._min_change_ratio + ) + + return ParagraphChange( + change_type="modified", + old_text=old, + new_text=new, + numeric_changed=numeric_changed, + deontic_changed=deontic_changed, + change_ratio=change_ratio, + needs_llm=significant, + diff_spans=[(op, text) for op, text in spans], + ) diff --git a/backend/app/infrastructure/tasks/celery_app.py b/backend/app/infrastructure/tasks/celery_app.py index 0e86664..aa13002 100644 --- a/backend/app/infrastructure/tasks/celery_app.py +++ b/backend/app/infrastructure/tasks/celery_app.py @@ -28,7 +28,10 @@ celery_app = Celery( "compliance_hub", broker=_BROKER, backend=_BACKEND, - include=["app.infrastructure.tasks.document_tasks"], + include=[ + "app.infrastructure.tasks.document_tasks", + "app.infrastructure.tasks.perception_tasks", + ], ) celery_app.conf.update( @@ -42,4 +45,12 @@ celery_app.conf.update( task_reject_on_worker_lost=True, # Keep results for 1 hour for status polling. result_expires=3600, + # Scheduled counterpart to the Perception page's manual "Refresh" button. + # Only takes effect while a Beat process is running (./dev.sh start beat). + beat_schedule={ + "crawl-regulations-periodic": { + "task": "app.infrastructure.tasks.perception_tasks.crawl_regulations_task", + "schedule": settings.perception_crawl_interval_seconds, + }, + }, ) diff --git a/backend/app/infrastructure/tasks/perception_tasks.py b/backend/app/infrastructure/tasks/perception_tasks.py new file mode 100644 index 0000000..072795a --- /dev/null +++ b/backend/app/infrastructure/tasks/perception_tasks.py @@ -0,0 +1,63 @@ +"""Celery task for scheduled regulatory source crawling. + +This is the scheduled counterpart to the Perception page's manual "Refresh" +button (POST /perception/crawl). Every architecture reference document +describes source monitoring as continuous ("定时爬取"), not operator-triggered, +so this task is what Celery Beat runs on a fixed interval once an operator +starts a Beat process. +""" + +from __future__ import annotations + +from loguru import logger + +from app.infrastructure.tasks.celery_app import celery_app + + +@celery_app.task( + name="app.infrastructure.tasks.perception_tasks.crawl_regulations_task", + bind=True, +) +def crawl_regulations_task(self) -> dict: + """Crawl every registered regulatory source and enrich new/changed events. + + Drains CrawlService.run_crawl(), which already isolates each source's + fetch and each event's enrichment behind its own try/except — a source + outage or a single bad event yields an "error" progress item and the + generator continues. Re-catching those here would only hide problems the + service has already handled, so this task's job is limited to counting + them and logging a summary. + + No automatic retry is configured. An exception escaping run_crawl itself + means something broke in a way the service's own error handling did not + anticipate; the next scheduled tick already provides a retry within + settings.perception_crawl_interval_seconds, so an immediate retry against + the same failure is not worth the added complexity. + + ponytail: relies on a single worker process to serialize scheduled runs + (Celery's default concurrency processes one task at a time, so a run that + outlasts the interval delays the next tick rather than overlapping it). + Add a Redis-based lock (e.g. SETNX on a per-task key) if this queue is + ever served by more than one worker. + """ + from app.shared.bootstrap import get_crawl_service + + error_count = 0 + new_count = 0 + updated_count = 0 + + for item in get_crawl_service().run_crawl(): + event = item.get("event") + if event == "error": + error_count += 1 + logger.warning("Scheduled crawl source error: {}", item.get("data")) + elif event == "done": + data = item.get("data") or {} + new_count = data.get("total_new", 0) + updated_count = data.get("total_updated", 0) + + logger.info( + "Scheduled crawl finished: new={} updated={} source_errors={}", + new_count, updated_count, error_count, + ) + return {"new": new_count, "updated": updated_count, "source_errors": error_count} diff --git a/backend/app/services/llm/llm_factory.py b/backend/app/services/llm/llm_factory.py index 7b7e118..8d25af1 100644 --- a/backend/app/services/llm/llm_factory.py +++ b/backend/app/services/llm/llm_factory.py @@ -16,7 +16,7 @@ from app.shared.model_usage_tracker import get_model_usage_tracker # Keep provider-specific behavior explicit so debugging stays straightforward. DEFAULT_MODELS = { LLMProvider.DEEPSEEK: "deepseek-v4-flash", - LLMProvider.QWEN: "qwen3.5-flash", + LLMProvider.QWEN: "qwen3.6-flash", LLMProvider.QWEN_VL: "qwen3-vl-plus" } @@ -101,6 +101,8 @@ class LLMFactory: "qwen-max": LLMProvider.QWEN, "qwen3.5-flash": LLMProvider.QWEN, "qwen3.5-plus": LLMProvider.QWEN, + "qwen3.6-flash": LLMProvider.QWEN, + "qwen3.6-plus": LLMProvider.QWEN, "qwen_vl": LLMProvider.QWEN_VL, "qwen-vl": LLMProvider.QWEN_VL, "qwen-vl-plus": LLMProvider.QWEN_VL, diff --git a/backend/app/services/llm/qwen_client.py b/backend/app/services/llm/qwen_client.py index aa184f7..065793c 100644 --- a/backend/app/services/llm/qwen_client.py +++ b/backend/app/services/llm/qwen_client.py @@ -27,6 +27,8 @@ class QwenClient(BaseLLMClient): "qwen-long", "qwen3.5-flash", "qwen3.5-plus", + "qwen3.6-flash", + "qwen3.6-plus", "qwen3-plus", "qwen2.5-72b-instruct", "qwen2.5-32b-instruct", @@ -371,7 +373,7 @@ class QwenVLClient(BaseLLMClient): def create_qwen_client( api_key: str, - model: str = "qwen3.5-flash", + model: str = "qwen3.6-flash", base_url: str = "http://6.86.80.4:30080/v1", **kwargs ) -> QwenClient: diff --git a/backend/app/shared/bootstrap.py b/backend/app/shared/bootstrap.py index 74836fd..94e1d5f 100644 --- a/backend/app/shared/bootstrap.py +++ b/backend/app/shared/bootstrap.py @@ -23,8 +23,10 @@ from app.infrastructure.parser.local_chunk_builder import LocalRegulationChunkBu from app.infrastructure.parser.local_document_parser import LocalDocumentParser from app.infrastructure.parser.vector_chunk_builder import AliyunVectorChunkBuilder from app.infrastructure.perception.mock_event_store import MockEventStore +from app.infrastructure.perception.mock_notification_store import MockNotificationStore from app.application.perception.crawl_service import CrawlService from app.infrastructure.perception.base_event_store import BaseEventStore +from app.infrastructure.perception.base_notification_store import BaseNotificationStore from app.infrastructure.perception.crawlers.catarc_crawler import CatarcCrawler from app.infrastructure.perception.crawlers.guobiao_crawler import ( GuobiaoMandatoryCrawler, @@ -327,6 +329,22 @@ def get_event_store() -> BaseEventStore: return MockEventStore() +@lru_cache +def get_notification_store() -> BaseNotificationStore: + """Return notification store selected by DOCUMENT_REPOSITORY_BACKEND setting. + + Mirrors get_event_store()'s gate: Mock in-memory when Postgres isn't + configured, so the feature works in local dev and tests without a + database. + """ + if settings.document_repository_backend == "postgres": + from app.infrastructure.perception.postgres_notification_store import ( + PostgresNotificationStore, + ) + return PostgresNotificationStore() + return MockNotificationStore() + + @lru_cache def get_compliance_repository() -> ComplianceRepository: """Return the compliance analysis repository. @@ -370,6 +388,9 @@ def get_crawl_service() -> CrawlService: event_store=get_event_store(), llm_pipeline=LlmPipeline(), retrieval_service=get_retrieval_service(), + notification_store=get_notification_store(), + embedding_provider=get_embedding_provider(), + vector_index=get_vector_index(), ) diff --git a/backend/requirements.txt b/backend/requirements.txt index 058d4f4..c6e03d1 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -17,6 +17,11 @@ beautifulsoup4>=4.12.0 lxml>=5.0.0 tiktoken>=0.5.0 tenacity>=8.2.0 +# Regulatory signal crawling (backend/app/infrastructure/perception/) — main-content +# extraction from crawled regulation detail pages and character-level diff for change +# detection. Import name for diff-match-patch is diff_match_patch (underscored). +trafilatura>=2.0.0 +diff-match-patch>=20241021 # ── Auth ────────────────────────────────────────────────────────────────────── python-jose[cryptography]>=3.3.0 diff --git a/backend/tests/perception/test_crawl_service.py b/backend/tests/perception/test_crawl_service.py index 50dec69..589286d 100644 --- a/backend/tests/perception/test_crawl_service.py +++ b/backend/tests/perception/test_crawl_service.py @@ -6,6 +6,7 @@ import pytest from app.infrastructure.perception.crawlers.base import RawEvent from app.infrastructure.perception.mock_event_store import MockEventStore +from app.infrastructure.perception.mock_notification_store import MockNotificationStore def _make_raw_event(code="TST-001"): @@ -17,11 +18,18 @@ def _make_raw_event(code="TST-001"): ) +def _make_crawler(raw_events, full_text="full body text"): + """Build a mock crawler. `full_text=""` simulates a failed detail fetch.""" + mock_crawler = MagicMock() + mock_crawler.fetch.return_value = raw_events + mock_crawler.fetch_full_text.return_value = full_text + return mock_crawler + + def _make_service(raw_events): from app.application.perception.crawl_service import CrawlService - mock_crawler = MagicMock() - mock_crawler.fetch.return_value = raw_events + mock_crawler = _make_crawler(raw_events) mock_pipeline = MagicMock() mock_pipeline.extract_structure.return_value = { @@ -41,6 +49,9 @@ def _make_service(raw_events): event_store=store, llm_pipeline=mock_pipeline, retrieval_service=mock_retrieval, + notification_store=MockNotificationStore(), + embedding_provider=MagicMock(), + vector_index=MagicMock(), ) @@ -54,8 +65,7 @@ def test_crawl_yields_progress_and_done(): def test_crawl_upserts_to_store(): store = MockEventStore() from app.application.perception.crawl_service import CrawlService - mock_crawler = MagicMock() - mock_crawler.fetch.return_value = [_make_raw_event("NEW-001")] + mock_crawler = _make_crawler([_make_raw_event("NEW-001")]) mock_pipeline = MagicMock() mock_pipeline.extract_structure.return_value = { "obligations": [], "deadlines": [], "scope": "", @@ -70,6 +80,9 @@ def test_crawl_upserts_to_store(): event_store=store, llm_pipeline=mock_pipeline, retrieval_service=MagicMock(), + notification_store=MockNotificationStore(), + embedding_provider=MagicMock(), + vector_index=MagicMock(), ) list(svc.run_crawl()) result = store.get_by_standard_code("NEW-001") @@ -80,7 +93,8 @@ def test_crawl_upserts_to_store(): def test_crawl_skips_unchanged_events(): store = MockEventStore() raw = _make_raw_event("SKIP-001") - content_hash = hashlib.sha256(raw.raw_text.encode()).hexdigest() + body = "full body text" + content_hash = hashlib.sha256(body.encode()).hexdigest() store.upsert({ "id": hashlib.sha256(f"TEST-SKIP-001".encode()).hexdigest()[:12], "standard_code": "SKIP-001", @@ -99,13 +113,341 @@ def test_crawl_skips_unchanged_events(): }) mock_pipeline = MagicMock() from app.application.perception.crawl_service import CrawlService - mock_crawler = MagicMock() - mock_crawler.fetch.return_value = [raw] + mock_crawler = _make_crawler([raw], full_text=body) svc = CrawlService( crawlers={"TEST": mock_crawler}, event_store=store, llm_pipeline=mock_pipeline, retrieval_service=MagicMock(), + notification_store=MockNotificationStore(), + embedding_provider=MagicMock(), + vector_index=MagicMock(), ) list(svc.run_crawl()) mock_pipeline.extract_structure.assert_not_called() + + +def test_crawl_stores_the_fetched_body_for_the_next_diff(): + """The body must be persisted, or the next crawl has no baseline to compare.""" + store = MockEventStore() + from app.application.perception.crawl_service import CrawlService + mock_pipeline = MagicMock() + mock_pipeline.extract_structure.return_value = {} + mock_pipeline.assess_impact.return_value = [] + svc = CrawlService( + crawlers={"TEST": _make_crawler([_make_raw_event("BODY-001")], full_text="第一条 正文内容。")}, + event_store=store, + llm_pipeline=mock_pipeline, + retrieval_service=MagicMock(), + notification_store=MockNotificationStore(), + embedding_provider=MagicMock(), + vector_index=MagicMock(), + ) + list(svc.run_crawl()) + + stored = store.get_by_standard_code("BODY-001") + assert stored["raw_text"] == "第一条 正文内容。" + + +def test_crawl_falls_back_when_full_text_fetch_fails(): + """An unreachable detail page degrades to the list-page text, never crashes.""" + store = MockEventStore() + from app.application.perception.crawl_service import CrawlService + mock_pipeline = MagicMock() + mock_pipeline.extract_structure.return_value = {} + mock_pipeline.assess_impact.return_value = [] + svc = CrawlService( + crawlers={"TEST": _make_crawler([_make_raw_event("FALL-001")], full_text="")}, + event_store=store, + llm_pipeline=mock_pipeline, + retrieval_service=MagicMock(), + notification_store=MockNotificationStore(), + embedding_provider=MagicMock(), + vector_index=MagicMock(), + ) + list(svc.run_crawl()) + + stored = store.get_by_standard_code("FALL-001") + assert stored is not None + assert stored["raw_text"] == "full text" + + +def test_crawl_skips_diff_when_no_previous_body_exists(): + """Rows stored before raw_text was persisted must not be diffed against nothing.""" + store = MockEventStore() + from app.application.perception.crawl_service import CrawlService + event_id = hashlib.sha256(b"TEST-OLD-001").hexdigest()[:12] + store.upsert({ + "id": event_id, + "standard_code": "OLD-001", + "source": "TEST", + "title": "Test OLD-001", + "summary": "legacy row", + "impact_level": "low", + "published_at": "2026-01-01", + "tags": [], + "content_hash": "stale-hash-from-before-this-change", + }) + mock_pipeline = MagicMock() + mock_pipeline.extract_structure.return_value = {} + mock_pipeline.assess_impact.return_value = [] + svc = CrawlService( + crawlers={"TEST": _make_crawler([_make_raw_event("OLD-001")], full_text="第一条 新正文。")}, + event_store=store, + llm_pipeline=mock_pipeline, + retrieval_service=MagicMock(), + notification_store=MockNotificationStore(), + embedding_provider=MagicMock(), + vector_index=MagicMock(), + ) + list(svc.run_crawl()) + + mock_pipeline.compute_diff.assert_not_called() + assert store.get(event_id)["raw_text"] == "第一条 新正文。" + + +def test_new_event_creates_a_new_notification(): + """A brand-new event must produce exactly one kind='new' notification.""" + from app.application.perception.crawl_service import CrawlService + + store = MockEventStore() + notifications = MockNotificationStore() + mock_pipeline = MagicMock() + mock_pipeline.extract_structure.return_value = {} + mock_pipeline.assess_impact.return_value = [] + svc = CrawlService( + crawlers={"TEST": _make_crawler([_make_raw_event("NOTIF-NEW")])}, + event_store=store, + llm_pipeline=mock_pipeline, + retrieval_service=MagicMock(), + notification_store=notifications, + embedding_provider=MagicMock(), + vector_index=MagicMock(), + ) + list(svc.run_crawl()) + + items = notifications.list_for_user("any-user") + assert len(items) == 1 + assert items[0]["kind"] == "new" + + +def test_significant_change_creates_a_changed_notification(): + """A numeric or deontic change must produce a kind='changed' notification.""" + from app.application.perception.crawl_service import CrawlService + + store = MockEventStore() + event_id = hashlib.sha256(b"TEST-SIG-001").hexdigest()[:12] + store.upsert({ + "id": event_id, "standard_code": "SIG-001", "source": "TEST", + "title": "Test SIG-001", "summary": "", "impact_level": "medium", + "published_at": "2026-01-01", "tags": [], + "content_hash": "old-hash", "raw_text": "old body", + }) + notifications = MockNotificationStore() + mock_pipeline = MagicMock() + mock_pipeline.extract_structure.return_value = {} + mock_pipeline.assess_impact.return_value = [] + mock_pipeline.compute_diff.return_value = { + "changed_sections": [{"change_type": "modified", "numeric_changed": True, "deontic_changed": False}], + "change_summary": "1 paragraph changed (numeric).", + } + svc = CrawlService( + crawlers={"TEST": _make_crawler([_make_raw_event("SIG-001")], full_text="new body")}, + event_store=store, + llm_pipeline=mock_pipeline, + retrieval_service=MagicMock(), + notification_store=notifications, + embedding_provider=MagicMock(), + vector_index=MagicMock(), + ) + list(svc.run_crawl()) + + items = notifications.list_for_user("any-user") + assert len(items) == 1 + assert items[0]["kind"] == "changed" + + +def test_cosmetic_only_change_creates_no_notification(): + """A change with no numeric/deontic/added/removed section must not notify.""" + from app.application.perception.crawl_service import CrawlService + + store = MockEventStore() + event_id = hashlib.sha256(b"TEST-COS-001").hexdigest()[:12] + store.upsert({ + "id": event_id, "standard_code": "COS-001", "source": "TEST", + "title": "Test COS-001", "summary": "", "impact_level": "low", + "published_at": "2026-01-01", "tags": [], + "content_hash": "old-hash", "raw_text": "old body.", + }) + notifications = MockNotificationStore() + mock_pipeline = MagicMock() + mock_pipeline.extract_structure.return_value = {} + mock_pipeline.assess_impact.return_value = [] + mock_pipeline.compute_diff.return_value = { + "changed_sections": [{"change_type": "modified", "numeric_changed": False, "deontic_changed": False}], + "change_summary": "cosmetic only", + } + svc = CrawlService( + crawlers={"TEST": _make_crawler([_make_raw_event("COS-001")], full_text="old body")}, + event_store=store, + llm_pipeline=mock_pipeline, + retrieval_service=MagicMock(), + notification_store=notifications, + embedding_provider=MagicMock(), + vector_index=MagicMock(), + ) + list(svc.run_crawl()) + + assert notifications.list_for_user("any-user") == [] + + +def test_notification_store_failure_does_not_abort_the_crawl(): + """A broken notification store must not stop the crawl or raise.""" + from app.application.perception.crawl_service import CrawlService + + store = MockEventStore() + broken_notifications = MagicMock() + broken_notifications.create.side_effect = RuntimeError("notification db down") + mock_pipeline = MagicMock() + mock_pipeline.extract_structure.return_value = {} + mock_pipeline.assess_impact.return_value = [] + svc = CrawlService( + crawlers={"TEST": _make_crawler([_make_raw_event("BROKEN-001")])}, + event_store=store, + llm_pipeline=mock_pipeline, + retrieval_service=MagicMock(), + notification_store=broken_notifications, + embedding_provider=MagicMock(), + vector_index=MagicMock(), + ) + events = list(svc.run_crawl()) + + assert any(e.get("event") == "done" for e in events) + assert store.get_by_standard_code("BROKEN-001") is not None + + +def test_new_event_is_indexed_in_the_knowledge_base(): + """A brand-new event must be chunked, embedded, and upserted into Milvus.""" + from app.application.perception.crawl_service import CrawlService + + body = "第一条 本标准规定了车辆制动系统的技术要求。\n第二条 车辆制动系统应在时速50公里条件下于30米内完全停止。" + embedding_provider = MagicMock() + embedding_provider.embed_texts.return_value = [[0.1] * 8] + vector_index = MagicMock() + mock_pipeline = MagicMock() + mock_pipeline.extract_structure.return_value = {} + mock_pipeline.assess_impact.return_value = [] + svc = CrawlService( + crawlers={"TEST": _make_crawler([_make_raw_event("IDX-NEW")], full_text=body)}, + event_store=MockEventStore(), + llm_pipeline=mock_pipeline, + retrieval_service=MagicMock(), + notification_store=MockNotificationStore(), + embedding_provider=embedding_provider, + vector_index=vector_index, + ) + list(svc.run_crawl()) + + event_id = hashlib.sha256(b"TEST-IDX-NEW").hexdigest()[:12] + vector_index.delete_by_document.assert_called_once_with(event_id) + vector_index.upsert.assert_called_once() + chunks_arg = vector_index.upsert.call_args.args[0] + assert len(chunks_arg) > 0 + + +def test_significant_change_reindexes_the_knowledge_base(): + """A numeric/deontic change must delete the stale chunks and upsert new ones.""" + from app.application.perception.crawl_service import CrawlService + + store = MockEventStore() + event_id = hashlib.sha256(b"TEST-IDX-SIG").hexdigest()[:12] + store.upsert({ + "id": event_id, "standard_code": "IDX-SIG", "source": "TEST", + "title": "Test IDX-SIG", "summary": "", "impact_level": "medium", + "published_at": "2026-01-01", "tags": [], + "content_hash": "old-hash", "raw_text": "old body", + }) + embedding_provider = MagicMock() + embedding_provider.embed_texts.return_value = [[0.1] * 8] + vector_index = MagicMock() + mock_pipeline = MagicMock() + mock_pipeline.extract_structure.return_value = {} + mock_pipeline.assess_impact.return_value = [] + mock_pipeline.compute_diff.return_value = { + "changed_sections": [{"change_type": "modified", "numeric_changed": True, "deontic_changed": False}], + "change_summary": "numeric change", + } + svc = CrawlService( + crawlers={"TEST": _make_crawler([_make_raw_event("IDX-SIG")], full_text="new body with a number 20米")}, + event_store=store, + llm_pipeline=mock_pipeline, + retrieval_service=MagicMock(), + notification_store=MockNotificationStore(), + embedding_provider=embedding_provider, + vector_index=vector_index, + ) + list(svc.run_crawl()) + + vector_index.delete_by_document.assert_called_once_with(event_id) + vector_index.upsert.assert_called_once() + + +def test_cosmetic_only_change_does_not_touch_the_knowledge_base(): + """A punctuation-only edit must not trigger embedding or a Milvus write.""" + from app.application.perception.crawl_service import CrawlService + + store = MockEventStore() + event_id = hashlib.sha256(b"TEST-IDX-COS").hexdigest()[:12] + store.upsert({ + "id": event_id, "standard_code": "IDX-COS", "source": "TEST", + "title": "Test IDX-COS", "summary": "", "impact_level": "low", + "published_at": "2026-01-01", "tags": [], + "content_hash": "old-hash", "raw_text": "old body.", + }) + embedding_provider = MagicMock() + vector_index = MagicMock() + mock_pipeline = MagicMock() + mock_pipeline.extract_structure.return_value = {} + mock_pipeline.assess_impact.return_value = [] + mock_pipeline.compute_diff.return_value = { + "changed_sections": [{"change_type": "modified", "numeric_changed": False, "deontic_changed": False}], + "change_summary": "cosmetic only", + } + svc = CrawlService( + crawlers={"TEST": _make_crawler([_make_raw_event("IDX-COS")], full_text="old body")}, + event_store=store, + llm_pipeline=mock_pipeline, + retrieval_service=MagicMock(), + notification_store=MockNotificationStore(), + embedding_provider=embedding_provider, + vector_index=vector_index, + ) + list(svc.run_crawl()) + + embedding_provider.embed_texts.assert_not_called() + vector_index.upsert.assert_not_called() + + +def test_vector_index_failure_does_not_abort_the_crawl(): + """A broken vector index must not stop the crawl or raise.""" + from app.application.perception.crawl_service import CrawlService + + broken_vector_index = MagicMock() + broken_vector_index.upsert.side_effect = RuntimeError("milvus unreachable") + store = MockEventStore() + mock_pipeline = MagicMock() + mock_pipeline.extract_structure.return_value = {} + mock_pipeline.assess_impact.return_value = [] + svc = CrawlService( + crawlers={"TEST": _make_crawler([_make_raw_event("IDX-FAIL")], full_text="第一条 正文内容。")}, + event_store=store, + llm_pipeline=mock_pipeline, + retrieval_service=MagicMock(), + notification_store=MockNotificationStore(), + embedding_provider=MagicMock(embed_texts=MagicMock(return_value=[[0.1] * 8])), + vector_index=broken_vector_index, + ) + events = list(svc.run_crawl()) + + assert any(e.get("event") == "done" for e in events) + assert store.get_by_standard_code("IDX-FAIL") is not None diff --git a/backend/tests/perception/test_llm_pipeline.py b/backend/tests/perception/test_llm_pipeline.py index f828f01..d8df9c2 100644 --- a/backend/tests/perception/test_llm_pipeline.py +++ b/backend/tests/perception/test_llm_pipeline.py @@ -1,28 +1,34 @@ -"""Unit tests for LlmPipeline — mock LLM client and embedding provider.""" +"""Unit tests for LlmPipeline with a mocked LLM client. + +The pipeline no longer constructs an embedding provider: change detection moved +to the deterministic RegulationDiffer, and the LLM is called only to explain +changes that determinism already located. These tests pin that gating contract. +""" from __future__ import annotations + from unittest.mock import MagicMock, patch -import json -import pytest -def _make_pipeline(): - with patch("app.infrastructure.perception.llm_pipeline.get_llm_client") as mock_llm_fn, \ - patch("app.infrastructure.perception.llm_pipeline.OpenAICompatibleEmbeddingProvider") as mock_emb_cls: - +def _make_pipeline(content: str | None = None): + """Build a pipeline whose LLM client is a mock returning `content`.""" + default = ( + '{"obligations":[{"text":"test obligation","deontic":"must","subject":"OEM",' + '"object":"system","condition":""}],"deadlines":[{"date":"2026-07-01",' + '"description":"实施截止"}],"scope":"适用于M1类车辆","penalties":"罚款",' + '"impact_level":"high"}' + ) + with patch("app.infrastructure.perception.llm_pipeline.get_llm_client") as mock_llm_fn: mock_client = MagicMock() - mock_client.chat.return_value = MagicMock(content='{"obligations":[{"text":"test obligation","deontic":"must","subject":"OEM","object":"system","condition":""}],"deadlines":[{"date":"2026-07-01","description":"实施截止"}],"scope":"适用于M1类车辆","penalties":"罚款","impact_level":"high"}') + mock_client.chat.return_value = MagicMock(content=content or default) mock_llm_fn.return_value = mock_client - mock_emb = MagicMock() - mock_emb.embed_texts.return_value = [[0.1] * 1024, [0.9] * 1024] - mock_emb_cls.return_value = mock_emb - from app.infrastructure.perception.llm_pipeline import LlmPipeline - return LlmPipeline(), mock_client, mock_emb + return LlmPipeline(), mock_client def test_extract_structure_returns_dict(): - pipeline, mock_client, _ = _make_pipeline() + """Structure extraction still returns the enrichment keys callers expect.""" + pipeline, _ = _make_pipeline() event = { "id": "evt-001", "standard_code": "GB 18384-2025", @@ -38,8 +44,11 @@ def test_extract_structure_returns_dict(): def test_assess_impact_returns_list(): - pipeline, mock_client, _ = _make_pipeline() - mock_client.chat.return_value = MagicMock(content='[{"doc_id":"d1","doc_name":"Safety Manual","score":0.85,"key_clauses":"§4.2","recommendation":"更新第4章"}]') + """Impact assessment still returns a list of affected documents.""" + pipeline, _ = _make_pipeline( + '[{"doc_id":"d1","doc_name":"Safety Manual","score":0.85,' + '"key_clauses":"§4.2","recommendation":"更新第4章"}]' + ) mock_retrieval = MagicMock() chunk = MagicMock() chunk.doc_id = "d1" @@ -53,25 +62,103 @@ def test_assess_impact_returns_list(): "title": "电动汽车安全要求", "obligations": [{"text": "OEM shall comply"}], } - result = pipeline.assess_impact(event, mock_retrieval) - assert isinstance(result, list) + assert isinstance(pipeline.assess_impact(event, mock_retrieval), list) -def test_compute_diff_no_change(): - pipeline, _, mock_emb = _make_pipeline() - mock_emb.embed_texts.return_value = [[0.5] * 1024, [0.5] * 1024] - result = pipeline.compute_diff("paragraph one", "paragraph one") - assert isinstance(result, dict) - assert "changed_sections" in result - assert "change_summary" in result +def test_compute_diff_no_change_costs_no_llm_call(): + """Identical text must short-circuit before reaching the model.""" + pipeline, mock_client = _make_pipeline() + mock_client.chat.reset_mock() + + result = pipeline.compute_diff("第一条 保持不变的条款。", "第一条 保持不变的条款。") + + assert result["changed_sections"] == [] + assert "No substantive changes" in result["change_summary"] + mock_client.chat.assert_not_called() -def test_compute_diff_detects_change(): - pipeline, mock_client, mock_emb = _make_pipeline() - mock_emb.embed_texts.return_value = [ - [1.0] + [0.0] * 1023, - [0.0] + [1.0] + [0.0] * 1022, - ] - mock_client.chat.return_value = MagicMock(content='{"change_type":"tightened","summary":"Requirement tightened"}') - result = pipeline.compute_diff("old paragraph text", "new tighter requirement text") - assert isinstance(result["changed_sections"], list) +def test_compute_diff_classifies_a_real_change(): + """A gated change is classified and the model's legal_effect is surfaced.""" + pipeline, _ = _make_pipeline( + '{"change_type":"tightened","legal_effect":"Requirement tightened."}' + ) + result = pipeline.compute_diff( + "第三条 生产企业应当每年开展一次安全评估。", + "第三条 生产企业宜每年开展一次安全评估。", + ) + + sections = result["changed_sections"] + assert len(sections) == 1 + assert sections[0]["change_type"] == "tightened" + assert sections[0]["summary"] == "Requirement tightened." + + +def test_numeric_change_overrides_the_model_label(): + """A moved number wins over the model, which routinely calls it 'clarified'.""" + pipeline, _ = _make_pipeline( + '{"change_type":"clarified","legal_effect":"Minor wording update."}' + ) + result = pipeline.compute_diff( + "第二条 车辆制动系统应在30米内完全停止。", + "第二条 车辆制动系统应在20米内完全停止。", + ) + + section = result["changed_sections"][0] + assert section["numeric_changed"] is True + assert section["change_type"] == "numeric" + + +def test_cosmetic_change_is_never_sent_to_the_model(): + """Punctuation-only edits are recorded but must not cost a model call.""" + pipeline, mock_client = _make_pipeline() + mock_client.chat.reset_mock() + + result = pipeline.compute_diff( + "第五条 本标准由全国汽车标准化技术委员会归口管理。", + "第五条 本标准由全国汽车标准化技术委员会归口管理", + ) + + assert len(result["changed_sections"]) == 1 + mock_client.chat.assert_not_called() + + +def test_llm_failure_preserves_the_deterministic_record(): + """A model error must not discard a change deterministic analysis proved real.""" + pipeline, mock_client = _make_pipeline() + mock_client.chat.side_effect = RuntimeError("gateway down") + + result = pipeline.compute_diff( + "第二条 车辆制动系统应在30米内完全停止。", + "第二条 车辆制动系统应在20米内完全停止。", + ) + + section = result["changed_sections"][0] + assert section["numeric_changed"] is True + assert section["change_type"] == "numeric" + assert section["summary"] == "" + assert "第二条" in section["old_text"] + + +def test_only_gated_paragraphs_reach_the_model(): + """One significant change among cosmetic ones yields exactly one model call.""" + pipeline, mock_client = _make_pipeline( + '{"change_type":"tightened","legal_effect":"Tighter limit."}' + ) + mock_client.chat.reset_mock() + + old = "\n".join([ + "第一条 本标准规定了车辆制动系统的技术要求。", + "第二条 车辆制动系统应在30米内完全停止。", + "第三条 本标准由全国汽车标准化技术委员会归口管理。", + ]) + new = "\n".join([ + "第一条 本标准规定了车辆制动系统的技术要求。", + "第二条 车辆制动系统应在20米内完全停止。", + "第三条 本标准由全国汽车标准化技术委员会归口管理", + ]) + + result = pipeline.compute_diff(old, new) + + # Two paragraphs changed; only the numeric one clears the gate. + assert len(result["changed_sections"]) == 2 + assert mock_client.chat.call_count == 1 diff --git a/backend/tests/perception/test_notification_store.py b/backend/tests/perception/test_notification_store.py new file mode 100644 index 0000000..4d55ca7 --- /dev/null +++ b/backend/tests/perception/test_notification_store.py @@ -0,0 +1,83 @@ +"""Tests for the notification store's per-user read-state contract. + +These pin the core property that makes broadcast-to-everyone work without a +subscription model: one notification row is shared by all users, and each +user's read state is tracked independently against it. +""" +from __future__ import annotations + +from app.infrastructure.perception.mock_notification_store import MockNotificationStore + + +def _store_with_one_notification() -> MockNotificationStore: + store = MockNotificationStore() + store.create( + event_id="evt-001", + kind="new", + title="《电动汽车安全要求》国家标准第三版正式发布", + impact_level="high", + summary=None, + ) + return store + + +def test_a_new_notification_is_unread_for_everyone(): + """Nobody has read it yet, so unread_count is 1 for any user.""" + store = _store_with_one_notification() + assert store.unread_count("user-a") == 1 + assert store.unread_count("user-b") == 1 + + +def test_mark_all_read_zeroes_the_count_for_that_user(): + """Reading clears the count for the user who read it.""" + store = _store_with_one_notification() + marked = store.mark_all_read("user-a") + assert marked == 1 + assert store.unread_count("user-a") == 0 + + +def test_one_users_read_state_does_not_affect_another(): + """The whole point of read receipts over per-user fan-out: independence.""" + store = _store_with_one_notification() + store.mark_all_read("user-a") + assert store.unread_count("user-a") == 0 + assert store.unread_count("user-b") == 1 + + +def test_list_for_user_reports_the_read_flag_correctly(): + """The list endpoint must reflect this user's own read state per item.""" + store = _store_with_one_notification() + store.mark_all_read("user-a") + + items_a = store.list_for_user("user-a") + items_b = store.list_for_user("user-b") + + assert len(items_a) == 1 + assert items_a[0]["read"] is True + assert items_b[0]["read"] is False + + +def test_list_for_user_orders_newest_first(): + """Notifications appear most-recent-first regardless of creation order.""" + store = MockNotificationStore() + store.create(event_id="evt-1", kind="new", title="first", impact_level="low", summary=None) + store.create(event_id="evt-2", kind="new", title="second", impact_level="low", summary=None) + + items = store.list_for_user("user-a") + + assert [item["title"] for item in items] == ["second", "first"] + + +def test_mark_all_read_is_a_no_op_on_an_empty_feed(): + """Reading an empty feed must not raise and reports zero marked.""" + store = MockNotificationStore() + assert store.mark_all_read("user-a") == 0 + + +def test_mark_all_read_does_not_recount_already_read_notifications(): + """Calling mark_all_read twice must not double-count as newly marked.""" + store = _store_with_one_notification() + first = store.mark_all_read("user-a") + second = store.mark_all_read("user-a") + assert first == 1 + assert second == 0 diff --git a/backend/tests/perception/test_perception_notification_routes.py b/backend/tests/perception/test_perception_notification_routes.py new file mode 100644 index 0000000..02215cd --- /dev/null +++ b/backend/tests/perception/test_perception_notification_routes.py @@ -0,0 +1,72 @@ +"""Tests for the notification API routes. + +Follows the same direct-call convention as backend/tests/mcp/test_mcp_status.py: +patch the bootstrap singleton getter where the route module imports it, then +call the async route function directly with asyncio.run rather than standing +up a FastAPI TestClient — this repo has no existing TestClient harness, and +these route handlers are thin enough that exercising them directly tests the +same logic without inventing a new testing convention for two pass-through +endpoints. +""" + +from __future__ import annotations + +import asyncio +from unittest.mock import patch + +from app.domain.auth.models import UserClaims, UserRole +from app.infrastructure.perception.mock_notification_store import MockNotificationStore + + +def _user(user_id: str) -> UserClaims: + return UserClaims(user_id=user_id, username=user_id, role=UserRole.READONLY) + + +def _list_notifications(store, user_id: str, limit: int = 20) -> dict: + from app.api.routes.perception import list_notifications + + with patch("app.api.routes.perception.get_notification_store", return_value=store): + return asyncio.run(list_notifications(limit=limit, current_user=_user(user_id))) + + +def _mark_read(store, user_id: str) -> dict: + from app.api.routes.perception import mark_notifications_read + + with patch("app.api.routes.perception.get_notification_store", return_value=store): + return asyncio.run(mark_notifications_read(current_user=_user(user_id))) + + +def test_a_fresh_user_sees_the_notification_as_unread(): + """GET .../notifications reports the caller's own unread_count.""" + store = MockNotificationStore() + store.create(event_id="evt-1", kind="new", title="新法规发布", impact_level="high", summary=None) + + result = _list_notifications(store, "user-a") + + assert result["unread_count"] == 1 + assert len(result["items"]) == 1 + assert result["items"][0]["read"] is False + + +def test_mark_read_zeroes_a_subsequent_get(): + """POST .../read must clear unread_count for the next GET by the same user.""" + store = MockNotificationStore() + store.create(event_id="evt-1", kind="new", title="新法规发布", impact_level="high", summary=None) + + marked = _mark_read(store, "user-a") + result = _list_notifications(store, "user-a") + + assert marked == {"marked": 1} + assert result["unread_count"] == 0 + assert result["items"][0]["read"] is True + + +def test_mark_read_for_one_user_does_not_affect_another(): + """Read state is per-user — the whole point of the read-receipt design.""" + store = MockNotificationStore() + store.create(event_id="evt-1", kind="new", title="新法规发布", impact_level="high", summary=None) + + _mark_read(store, "user-a") + result_b = _list_notifications(store, "user-b") + + assert result_b["unread_count"] == 1 diff --git a/backend/tests/perception/test_perception_tasks.py b/backend/tests/perception/test_perception_tasks.py new file mode 100644 index 0000000..30abef1 --- /dev/null +++ b/backend/tests/perception/test_perception_tasks.py @@ -0,0 +1,77 @@ +"""Tests for the scheduled crawl Celery task. + +These pin the draining contract: the task must consume every item from +CrawlService.run_crawl(), tally per-source errors without raising on them, and +let a genuine whole-crawl exception propagate rather than swallowing it. + +Patches target app.shared.bootstrap.get_crawl_service — not +perception_tasks.get_crawl_service — because the task imports it inside its +own function body (see perception_tasks.py's docstring for why), so there is +no module-level name in perception_tasks to intercept. +""" +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + + +def _fake_crawl_service(events): + """Build a fake whose run_crawl() yields the given fixed event sequence.""" + service = MagicMock() + service.run_crawl.return_value = iter(events) + return service + + +def test_task_drains_generator_and_summarizes_errors(): + """One source error among two must not stop the run or raise.""" + events = [ + {"event": "progress", "data": {"source": "CATARC", "stage": "fetching"}}, + {"event": "error", "data": {"source": "CATARC", "message": "timeout"}}, + {"event": "progress", "data": {"source": "EUR-Lex", "stage": "fetching"}}, + {"event": "done", "data": {"total_new": 2, "total_updated": 1}}, + ] + with patch( + "app.shared.bootstrap.get_crawl_service", + return_value=_fake_crawl_service(events), + ): + from app.infrastructure.tasks.perception_tasks import crawl_regulations_task + result = crawl_regulations_task() + + assert result == {"new": 2, "updated": 1, "source_errors": 1} + + +def test_task_reports_zero_errors_on_a_clean_run(): + """A run with no source errors must report source_errors: 0.""" + events = [ + {"event": "progress", "data": {"source": "CATARC", "stage": "fetching"}}, + {"event": "done", "data": {"total_new": 0, "total_updated": 0}}, + ] + with patch( + "app.shared.bootstrap.get_crawl_service", + return_value=_fake_crawl_service(events), + ): + from app.infrastructure.tasks.perception_tasks import crawl_regulations_task + result = crawl_regulations_task() + + assert result == {"new": 0, "updated": 0, "source_errors": 0} + + +def test_whole_crawl_exception_is_not_swallowed(): + """A failure below run_crawl's own error handling must propagate. + + Per-source failures are already handled inside run_crawl and never raise; + an exception escaping the generator entirely means something unexpected + broke, and Celery's own failure handling — not a silent catch here — is + the intended backstop. + """ + broken_service = MagicMock() + broken_service.run_crawl.side_effect = RuntimeError("event store unreachable") + + with patch( + "app.shared.bootstrap.get_crawl_service", + return_value=broken_service, + ): + from app.infrastructure.tasks.perception_tasks import crawl_regulations_task + with pytest.raises(RuntimeError, match="event store unreachable"): + crawl_regulations_task() diff --git a/backend/tests/perception/test_regulation_differ.py b/backend/tests/perception/test_regulation_differ.py new file mode 100644 index 0000000..37752f5 --- /dev/null +++ b/backend/tests/perception/test_regulation_differ.py @@ -0,0 +1,157 @@ +"""Tests for the deterministic regulation differ. + +These tests pin the behaviour that the previous cosine-similarity implementation +could not deliver: real regulatory edits (numeric limits, deontic modals) must be +detected, and inserting a paragraph must not report unrelated paragraphs as +changed. Everything here runs offline — no LLM, no network, no embeddings. +""" + +from __future__ import annotations + +from app.infrastructure.perception.regulation_differ import RegulationDiffer + + +def _differ() -> RegulationDiffer: + """Build a differ with an explicit ratio so tests never depend on .env.""" + return RegulationDiffer(min_change_ratio=0.02) + + +def test_identical_documents_report_no_changes(): + """An unchanged regulation must produce an empty change list.""" + text = "第一条 车辆制动系统应在时速50公里条件下于30米内完全停止。\n第二条 驾驶员座椅面料的阻燃性能应符合附录B的规定。" + assert _differ().diff(text, text) == [] + + +def test_numeric_tightening_is_detected(): + """A changed numeric limit is the case cosine similarity scored 0.9153 and missed.""" + old = "第一条 车辆制动系统应在时速50公里条件下于30米内完全停止。" + new = "第一条 车辆制动系统应在时速50公里条件下于20米内完全停止。" + + changes = _differ().diff(old, new) + + assert len(changes) == 1 + change = changes[0] + assert change.change_type == "modified" + assert change.numeric_changed is True + assert change.needs_llm is True + + +def test_deontic_relaxation_is_detected(): + """Weakening 应当 to 宜 changes the legal force and must be flagged.""" + old = "第三条 生产企业应当每年开展一次安全评估。" + new = "第三条 生产企业宜每年开展一次安全评估。" + + changes = _differ().diff(old, new) + + assert len(changes) == 1 + assert changes[0].deontic_changed is True + assert changes[0].needs_llm is True + + +def test_prohibition_removal_is_detected(): + """Dropping 不得 flips a prohibition into a permission.""" + old = "第四条 车辆不得使用未经认证的电池组。" + new = "第四条 车辆可以使用经备案的电池组。" + + changes = _differ().diff(old, new) + + assert len(changes) == 1 + assert changes[0].deontic_changed is True + + +def test_inserted_paragraph_does_not_shift_the_rest(): + """Regression test for positional alignment. + + The previous implementation compared old[i] to new[i], so inserting one + paragraph at the top reported every following paragraph as changed. With + sequence alignment only the inserted paragraph is new. + """ + old = "\n".join([ + "第一条 本标准规定了车辆制动系统的技术要求。", + "第二条 车辆制动系统应在时速50公里条件下于30米内完全停止。", + "第三条 驾驶员座椅面料的阻燃性能应符合附录B的规定。", + ]) + new = "\n".join([ + "第零条 本标准适用于所有M1类车辆。", + "第一条 本标准规定了车辆制动系统的技术要求。", + "第二条 车辆制动系统应在时速50公里条件下于30米内完全停止。", + "第三条 驾驶员座椅面料的阻燃性能应符合附录B的规定。", + ]) + + changes = _differ().diff(old, new) + + assert len(changes) == 1, f"expected only the inserted paragraph, got {changes}" + assert changes[0].change_type == "added" + assert "第零条" in changes[0].new_text + assert changes[0].old_text == "" + + +def test_deleted_paragraph_is_reported_once(): + """Removing a provision yields exactly one 'removed' record.""" + old = "\n".join([ + "第一条 本标准规定了车辆制动系统的技术要求。", + "第二条 车辆制动系统应在时速50公里条件下于30米内完全停止。", + "第三条 驾驶员座椅面料的阻燃性能应符合附录B的规定。", + ]) + new = "\n".join([ + "第一条 本标准规定了车辆制动系统的技术要求。", + "第三条 驾驶员座椅面料的阻燃性能应符合附录B的规定。", + ]) + + changes = _differ().diff(old, new) + + assert len(changes) == 1 + assert changes[0].change_type == "removed" + assert "第二条" in changes[0].old_text + assert changes[0].new_text == "" + assert changes[0].needs_llm is True + + +def test_trivial_edit_is_not_sent_to_the_llm(): + """A cosmetic edit with no number or modal change must not cost an LLM call.""" + old = "第五条 本标准由全国汽车标准化技术委员会归口管理。" + new = "第五条 本标准由全国汽车标准化技术委员会归口管理" + + changes = _differ().diff(old, new) + + for change in changes: + assert change.numeric_changed is False + assert change.deontic_changed is False + assert change.needs_llm is False, f"trivial edit was gated to the LLM: {change}" + + +def test_large_rewrite_is_sent_to_the_llm(): + """A substantial rewrite clears the change-ratio gate even without numbers or modals.""" + old = "第六条 本标准参考了国际同类标准的相关内容。" + new = "第六条 本条款描述了完全不同的主题内容，涉及整车认证流程与型式试验的组织安排。" + + changes = _differ().diff(old, new) + + assert len(changes) == 1 + assert changes[0].change_ratio >= 0.02 + assert changes[0].needs_llm is True + + +def test_empty_old_text_yields_no_changes(): + """A first crawl has no baseline, so there is nothing to diff.""" + assert _differ().diff("", "第一条 任意内容。") == [] + + +def test_unchanged_paragraphs_are_never_returned(): + """Only changed paragraphs appear; equal ones are dropped.""" + old = "\n".join([ + "第一条 保持不变的条款。", + "第二条 车辆制动距离不得超过30米。", + "第三条 另一条保持不变的条款。", + ]) + new = "\n".join([ + "第一条 保持不变的条款。", + "第二条 车辆制动距离不得超过20米。", + "第三条 另一条保持不变的条款。", + ]) + + changes = _differ().diff(old, new) + + assert len(changes) == 1 + assert changes[0].numeric_changed is True + assert "第二条" in changes[0].old_text diff --git a/docs/architecture/backend-project-architecture.md b/docs/architecture/backend-project-architecture.md index 1bef908..4668b82 100644 --- a/docs/architecture/backend-project-architecture.md +++ b/docs/architecture/backend-project-architecture.md @@ -337,6 +337,31 @@ backend/app/ - 当前实现见 `backend/app/mcp/server.py`，只暴露 `search_regulations` 一个 tool，内部直接调用 `get_agent_conversation_service()`。 - `api/routes/status.py` 的 `GET /status/mcp` 是薄适配层：它只负责解析对外可达的 URL（`settings.mcp_public_url` 或从请求推导），其余全部交给 `mcp.server.get_mcp_status()`，路由不得直接读取 MCP 内部结构。 +### 4.7 `perception` + +职责： + +- 爬取外部法规源（CATARC、国标委强制性/推荐性、EUR-Lex）的列表页与正文（`infrastructure/perception/crawlers/`） +- 基于内容哈希的变更检测入口，以及**确定性**的段落级差异分析（`regulation_differ.py`：对齐 + 字符级 diff + 数字/情态词/新增删除闸门），只有通过闸门的段落才调用 LLM 分类（`llm_pipeline.py`） +- 站内通知的广播存储与每用户已读状态（`base_notification_store.py` 及 Mock/Postgres 实现）——广播给所有登录用户，不做订阅/角色过滤 +- 通过 Celery Beat 定时调度全量爬取（`infrastructure/tasks/perception_tasks.py`），调度间隔由 `perception_crawl_interval_seconds` 配置 +- 新事件或"显著变更"（数字/情态词变化，或整段增删）自动写入知识库：本地 markdown 分块（`LocalRegulationChunkBuilder`，与 `chunk_backend=aliyun` 的上传流程无关）→ 复用既有 `embedding_provider`/`vector_index` → 写入与 `/documents` 上传管线**同一个** Milvus collection + +非职责： + +- 不维护第二套知识库或第二套向量索引——爬取入库与手动上传共用 `get_embedding_provider()` / `get_vector_index()` 这两个端口实现，二者在检索侧不可区分 +- 不做外部推送渠道（Email/Teams/飞书/钉钉）——当前部署无 SMTP/Webhook 凭据，做了也无法验证；只做站内通知 +- 不做订阅/偏好引擎——所有登录用户收到同一份广播，按用户区分的只有"已读"状态 +- 不做整改任务追踪（责任人、期限、验收、证据归档）——PPT 原文将其标注为"扩展功能"，规模上属独立子项目，尚未开始 +- 变更检测判据不依赖 embedding 余弦相似度——该方法被证明无法区分数值/情态词变化（如"30米"→"20米"、"应当"→"宜"），已被字符级 diff + 语言学规则取代 + +说明： + +- `application/perception/crawl_service.py` 的 `CrawlService.run_crawl()` 是本模块的核心编排：单个事件的每一步（结构抽取、影响评估、diff、通知、知识库索引）各自 try/except 包裹，任一步失败只记警告、不中断整次爬取。 +- `_is_significant()` 与 `should_index` 是同一份判据，被通知创建和知识库索引两处复用，避免出现"值得通知"和"值得入库"两套互相漂移的标准。 +- `postgres_event_store.py` / `postgres_notification_store.py` 与对应的 Mock 实现共享同一个开关 `settings.document_repository_backend == "postgres"`，与文档处理模块的 backend 切换方式一致。 +- MinIO 上的 `raw_storage_key` 字段（schema 中声明）目前无人写入；正文改为直接存 `regulation_events.raw_text` 列，供下次爬取做基线对比。 + ## 5. Module Responsibilities ### 5.1 `api` diff --git a/frontend/src/api/perception.ts b/frontend/src/api/perception.ts index ac550bd..9b57875 100644 --- a/frontend/src/api/perception.ts +++ b/frontend/src/api/perception.ts @@ -52,6 +52,39 @@ export interface AnalysisSSEMessage { text?: string; } +export interface PerceptionNotification { + id: number; + event_id: string; + kind: 'new' | 'changed'; + title: string; + impact_level: string | null; + summary: string | null; + created_at: string; + read: boolean; +} + +export interface NotificationListResponse { + items: PerceptionNotification[]; + unread_count: number; +} + +/** Broadcast feed shared by every logged-in user; read state is per-caller. */ +export async function getNotifications(limit = 20): Promise { + const res = await fetch(`${PERCEPTION_API_BASE}/perception/notifications?limit=${limit}`, { headers: authHeader() }); + if (!res.ok) throw new Error(`notifications failed: ${res.status}`); + return res.json() as Promise; +} + +/** Marks every currently-unread notification read for the calling user. */ +export async function markNotificationsRead(): Promise<{ marked: number }> { + const res = await fetch(`${PERCEPTION_API_BASE}/perception/notifications/read`, { + method: 'POST', + headers: authHeader(), + }); + if (!res.ok) throw new Error(`mark read failed: ${res.status}`); + return res.json() as Promise<{ marked: number }>; +} + export async function getPerceptionStats(): Promise { const res = await fetch(`${PERCEPTION_API_BASE}/perception/stats`, { headers: authHeader() }); if (!res.ok) throw new Error(`stats failed: ${res.status}`); diff --git a/frontend/src/components/layout/Sidebar.tsx b/frontend/src/components/layout/Sidebar.tsx index 5f0f9a1..0ffeb15 100644 --- a/frontend/src/components/layout/Sidebar.tsx +++ b/frontend/src/components/layout/Sidebar.tsx @@ -1,3 +1,4 @@ +import { useEffect, useState } from 'react'; import { NavLink } from 'react-router-dom'; import { LayoutDashboard, Radio, Monitor, FileText, @@ -6,6 +7,12 @@ import { import { useTheme } from '../../contexts/ThemeContext'; import { useAuth } from '../../contexts/AuthContext'; import { useLanguage } from '../../contexts/LanguageContext'; +import { getNotifications } from '../../api/perception'; + +// How often the sidebar re-checks the unread count. A plain UI refresh +// cadence, not an infrastructure setting — unlike the crawl interval, this +// never needs to be tuned per deployment. +const UNREAD_POLL_MS = 60_000; interface NavItem { to: string; @@ -47,10 +54,24 @@ export function Sidebar() { const { theme, toggleTheme } = useTheme(); const { user, logout } = useAuth(); const { lang, t, toggleLang } = useLanguage(); + const [unreadSignals, setUnreadSignals] = useState(0); + + // Sidebar only mounts inside RequireAuth, so a token always exists here. + // Polling (not push) keeps this simple — at one crawl every 6 hours, a + // 60s badge refresh is more than fast enough to feel current. + useEffect(() => { + let cancelled = false; + function poll() { + getNotifications().then(r => { if (!cancelled) setUnreadSignals(r.unread_count); }).catch(() => {}); + } + poll(); + const timer = setInterval(poll, UNREAD_POLL_MS); + return () => { cancelled = true; clearInterval(timer); }; + }, []); const mainNav: NavItem[] = [ { to: '/', icon: , label: t.nav.overview }, - { to: '/signals', icon: , label: t.nav.signals }, + { to: '/signals', icon: , label: t.nav.signals, badge: unreadSignals }, { to: '/status', icon: , label: t.nav.status }, ]; diff --git a/frontend/src/contexts/PageStateContext.tsx b/frontend/src/contexts/PageStateContext.tsx index b9127ba..e0b1529 100644 --- a/frontend/src/contexts/PageStateContext.tsx +++ b/frontend/src/contexts/PageStateContext.tsx @@ -12,6 +12,7 @@ */ import React, { createContext, useContext, useState, useCallback, useRef } from 'react'; +import { COMPLIANCE_INIT } from './pageStateDefaults'; // ── RagChat types ───────────────────────────────────────────────────────────── @@ -122,22 +123,6 @@ export interface ComplianceState { conflicts: ComplianceConflict[]; } -const COMPLIANCE_INIT: ComplianceState = { - status: 'idle', - stageLabel: '', - stageKey: '', - meta: null, - sources: [], - findings: [], - done: null, - errorText: '', - analysisId: null, - isReadOnly: false, - activeFindingId: null, - progress: null, - conflicts: [], -}; - // ── Perception types ────────────────────────────────────────────────────────── export interface PerceptionSignal { diff --git a/frontend/src/contexts/index.ts b/frontend/src/contexts/index.ts index 89a9b59..d6663f2 100644 --- a/frontend/src/contexts/index.ts +++ b/frontend/src/contexts/index.ts @@ -2,6 +2,7 @@ export { ThemeProvider, useTheme } from './ThemeContext'; export { AuthProvider, useAuth } from './AuthContext'; export type { AuthUser } from './AuthContext'; export { PageStateProvider, usePageState } from './PageStateContext'; +export { COMPLIANCE_INIT } from './pageStateDefaults'; export { LanguageProvider, useLanguage } from './LanguageContext'; export type { Lang } from './LanguageContext'; export type { diff --git a/frontend/src/contexts/pageStateDefaults.ts b/frontend/src/contexts/pageStateDefaults.ts new file mode 100644 index 0000000..a6bfa85 --- /dev/null +++ b/frontend/src/contexts/pageStateDefaults.ts @@ -0,0 +1,27 @@ +/** + * Default values for PageStateContext slices. + * + * These live outside PageStateContext.tsx because that file exports React + * components, and `react-refresh/only-export-components` requires shared + * constants to sit in their own module. Keeping the defaults here also gives + * consumers a single canonical initial state to spread from, instead of each + * page maintaining its own copy that silently drifts when a field is added. + */ + +import type { ComplianceState } from './PageStateContext'; + +export const COMPLIANCE_INIT: ComplianceState = { + status: 'idle', + stageLabel: '', + stageKey: '', + meta: null, + sources: [], + findings: [], + done: null, + errorText: '', + analysisId: null, + isReadOnly: false, + activeFindingId: null, + progress: null, + conflicts: [], +}; diff --git a/frontend/src/pages/Compliance/CompliancePage.tsx b/frontend/src/pages/Compliance/CompliancePage.tsx index dc6a717..6966417 100644 --- a/frontend/src/pages/Compliance/CompliancePage.tsx +++ b/frontend/src/pages/Compliance/CompliancePage.tsx @@ -7,7 +7,7 @@ import { useComplianceAnalysis } from './useComplianceAnalysis'; import { usePageState } from '../../contexts'; import { HistoryRail } from './HistoryRail'; import { FindingChatDrawer } from './FindingChatDrawer'; -import type { FindingEvent, SourceEvent, AnalysisMeta } from './useComplianceAnalysis'; +import type { FindingEvent, SourceEvent } from './useComplianceAnalysis'; const TOKEN_KEY = 'auth_token'; function authHeader(): Record { diff --git a/frontend/src/pages/Compliance/useComplianceAnalysis.ts b/frontend/src/pages/Compliance/useComplianceAnalysis.ts index 4a0c110..f830834 100644 --- a/frontend/src/pages/Compliance/useComplianceAnalysis.ts +++ b/frontend/src/pages/Compliance/useComplianceAnalysis.ts @@ -7,7 +7,7 @@ */ import { useCallback } from 'react'; -import { usePageState } from '../../contexts'; +import { usePageState, COMPLIANCE_INIT } from '../../contexts'; import type { ComplianceMeta, ComplianceState, @@ -28,21 +28,6 @@ function authHeader(): Record { return t ? { Authorization: `Bearer ${t}` } : {}; } -const INITIAL_STATE: ComplianceState = { - status: 'idle', - stageLabel: '', - stageKey: '', - meta: null, - sources: [], - findings: [], - done: null, - errorText: '', - analysisId: null, - isReadOnly: false, - progress: null, - conflicts: [], -}; - export function useComplianceAnalysis() { const { complianceState: state, setComplianceState: setState, complianceAbortRef, resetCompliance: reset } = usePageState(); @@ -51,7 +36,7 @@ export function useComplianceAnalysis() { const ctrl = new AbortController(); complianceAbortRef.current = ctrl; - setState({ ...INITIAL_STATE, status: 'streaming', stageLabel: 'Starting…', meta }); + setState({ ...COMPLIANCE_INIT, status: 'streaming', stageLabel: 'Starting…', meta }); try { const res = await fetch('/api/v1/compliance/analyze-stream', { diff --git a/frontend/src/pages/Login/LoginPage.tsx b/frontend/src/pages/Login/LoginPage.tsx index 96287c8..241c9a4 100644 --- a/frontend/src/pages/Login/LoginPage.tsx +++ b/frontend/src/pages/Login/LoginPage.tsx @@ -1,4 +1,4 @@ -import React, { FormEvent, useState } from 'react'; +import { useState, type FormEvent } from 'react'; import { useAuth } from '../../contexts'; export function LoginPage() { diff --git a/frontend/src/pages/Perception/PerceptionPage.tsx b/frontend/src/pages/Perception/PerceptionPage.tsx index 4d68830..4d9ae62 100644 --- a/frontend/src/pages/Perception/PerceptionPage.tsx +++ b/frontend/src/pages/Perception/PerceptionPage.tsx @@ -1,4 +1,4 @@ -import { useState, useEffect, useRef } from 'react'; +import { useState, useEffect } from 'react'; import { Topbar } from '../../components/layout/Topbar'; import { RefreshCw, Play, Square, ExternalLink } from 'lucide-react'; import { usePageState } from '../../contexts'; @@ -15,23 +15,24 @@ interface Stats { total: number; high_impact: number; medium_impact: number; - last_90_days: number; + recent_90d: number; } -const SOURCES = ['All', 'MIIT', 'UN-ECE', 'ISO', 'GB Comm.', 'EUR-Lex', 'IATF']; const IMPACTS = ['All', 'High', 'Medium', 'Low']; // Backend event → Signal function mapEvent(e: Record): PerceptionSignal { const impact = String(e.impact_level ?? '').toLowerCase(); + // The backend publishes a lifecycle stage, not a severity. Mapping it through + // an impact-level vocabulary sent every real value to the default branch, + // which renders as "已发布" — so consultation drafts were labelled as enacted. const backendStatus = String(e.status ?? '').toLowerCase(); return { id: String(e.id ?? e.event_id ?? ''), source: String(e.source ?? ''), standard: String(e.standard ?? e.standard_code ?? e.regulation_id ?? ''), - status: backendStatus === 'high' || backendStatus === 'urgent' ? 'risk' - : backendStatus === 'medium' || backendStatus === 'draft' ? 'warn' - : backendStatus === 'low' || backendStatus === 'final' ? 'ok' + status: backendStatus === 'enacted' ? 'ok' + : backendStatus === 'draft' || backendStatus === 'consultation' ? 'warn' : 'info', title: String(e.title ?? ''), summary: String(e.summary ?? e.description ?? ''), @@ -80,7 +81,13 @@ export function PerceptionPage() { fetch('/api/v1/perception/stats', { headers: authHeader() }) .then(r => r.json()) .then(setStats) - .catch(() => setStats({ total: 47, high_impact: 7, medium_impact: 18, last_90_days: 14 })); + .catch(() => setStats({ total: 47, high_impact: 7, medium_impact: 18, recent_90d: 14 })); + }, []); + + // Landing on this page is the acknowledgement — clear the sidebar badge by + // marking every currently-unread notification read. No dismiss UI needed. + useEffect(() => { + fetch('/api/v1/perception/notifications/read', { method: 'POST', headers: authHeader() }).catch(() => {}); }, []); // Fetch signal list on first mount only (if empty), otherwise preserve context state @@ -114,6 +121,17 @@ export function PerceptionPage() { const selected = signals.find(s => s.id === selectedId) ?? null; + // Derived from the loaded data rather than hardcoded. The previous fixed list + // was written against the mock fixtures, so the two sources the crawlers + // actually produce — CATARC and 国标委 — had no chip and could never be + // filtered. Deriving them also means a new crawler needs no frontend change. + // sourceFilter survives navigation in PageStateContext, so a filter chosen + // against an earlier dataset is kept in the list; dropping it would strand + // the user on an empty list with no chip to click their way out of. + const sources = ['All', ...Array.from( + new Set([...signals.map(s => s.source), sourceFilter].filter(s => s && s !== 'All')), + ).sort()]; + const filtered = signals.filter(s => { if (sourceFilter !== 'All' && s.source !== sourceFilter) return false; if (impactFilter !== 'All' && s.impact !== impactFilter) return false; @@ -178,6 +196,11 @@ export function PerceptionPage() { } async function runCrawl() { + // A crawl already in flight is superseded — cancel it so its SSE reader + // stops writing status text for a run the user has replaced. + perceptionCrawlAbortRef.current?.abort(); + const ctrl = new AbortController(); + perceptionCrawlAbortRef.current = ctrl; setCrawling(true); setPerceptionState(s => ({ ...s, crawlStatus: t.signals.statusConnecting })); try { @@ -185,6 +208,7 @@ export function PerceptionPage() { method: 'POST', headers: { 'Content-Type': 'application/json', ...authHeader() }, body: JSON.stringify({}), + signal: ctrl.signal, }); if (!res.body) { setPerceptionState(s => ({ ...s, crawlStatus: 'No stream' })); @@ -232,10 +256,14 @@ export function PerceptionPage() { } } } catch (e: unknown) { - setPerceptionState(s => ({ - ...s, - crawlStatus: t.signals.statusConnFailed.replace('{message}', e instanceof Error ? e.message : String(e)), - })); + // An abort is a deliberate supersede, not a backend failure — leaving the + // status untouched avoids reporting "connection failed" to the user. + if (!(e instanceof DOMException && e.name === 'AbortError')) { + setPerceptionState(s => ({ + ...s, + crawlStatus: t.signals.statusConnFailed.replace('{message}', e instanceof Error ? e.message : String(e)), + })); + } } setCrawling(false); } @@ -293,14 +321,14 @@ export function PerceptionPage() { {t.signals.statMedium}
- {stats?.last_90_days ?? '—'} + {stats?.recent_90d ?? '—'} {t.signals.statLast90}
- {SOURCES.map(s => ( + {sources.map(s => (
@@ -411,9 +439,9 @@ export function PerceptionPage() {

{(selectedFull?.scope as string) || selected.summary}

- {selectedFull?.penalties && ( + {Boolean(selectedFull?.penalties) && (

- ⚠ {selectedFull.penalties as string} + ⚠ {selectedFull?.penalties as string}

)}
@@ -486,8 +514,8 @@ export function PerceptionPage() { {String(d.doc_name || '')} {String(d.key_clauses || d.clause || '')} - {d.snippet &&
{String(d.snippet)}
} - {d.recommendation && ( + {Boolean(d.snippet) &&
{String(d.snippet)}
} + {Boolean(d.recommendation) && (
→ {String(d.recommendation)}
)} @@ -523,7 +551,7 @@ export function PerceptionPage() { {String(s.new_text || '')} - {s.summary &&

{String(s.summary)}

} + {Boolean(s.summary) &&

{String(s.summary)}

} )); })()} diff --git a/pyproject.toml b/pyproject.toml index c518e0c..2a9e78e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,6 +26,8 @@ dependencies = [ "httpx>=0.24.0", "beautifulsoup4>=4.12.0", "lxml>=5.0.0", + "trafilatura>=2.0.0", + "diff-match-patch>=20241021", "alibabacloud-docmind-api20220711>=1.0.6", "alibabacloud-tea-openapi>=0.3.11", "alibabacloud-tea-util>=0.3.13", diff --git a/tests/test_celery_tasks.py b/tests/test_celery_tasks.py index 9498939..804e62a 100644 --- a/tests/test_celery_tasks.py +++ b/tests/test_celery_tasks.py @@ -33,6 +33,20 @@ def test_process_document_task_is_registered(): ) +def test_crawl_regulations_task_is_registered(): + """crawl_regulations_task must be discoverable in the Celery task registry. + + This is the scheduled counterpart to the manual "Refresh" button — Beat + has nothing to run on its interval unless this task is registered. + """ + import app.infrastructure.tasks.perception_tasks # noqa: F401 — triggers task registration + from app.infrastructure.tasks.celery_app import celery_app + registered = list(celery_app.tasks.keys()) + assert any("crawl_regulations_task" in name for name in registered), ( + f"crawl_regulations_task not found in {registered}" + ) + + def test_document_command_service_has_process_document(): """DocumentCommandService must expose _process_document method.""" from app.application.documents.services import DocumentCommandService