From 121bf33bf415b0ef5927b39b4d9133d731fbcc33 Mon Sep 17 00:00:00 2001 From: yuqei Date: Wed, 20 May 2026 11:53:10 +0800 Subject: [PATCH] fix: resolve legal article refs by source aliases --- backend/app/services/review_service.py | 41 +++++++++++++++++++++++++- backend/tests/test_review_service.py | 32 +++++++++++++++++++- 2 files changed, 71 insertions(+), 2 deletions(-) diff --git a/backend/app/services/review_service.py b/backend/app/services/review_service.py index 63d52f9b..64f25ead 100644 --- a/backend/app/services/review_service.py +++ b/backend/app/services/review_service.py @@ -1715,6 +1715,7 @@ class ReviewService: source_titles, article_nos, keywords = LegalArticleService._extract_runtime_search_terms(risk_text) candidates: List[Any] = [] try: + candidates.extend(self._search_rule_target_legal_article_candidates(risk_text, tenant_id)) if source_titles or article_nos: candidates.extend( self.legal_article_repository.search_published_for_runtime( @@ -1745,6 +1746,26 @@ class ReviewService: return [] return self._merge_legal_article_candidates(candidates) + def _search_rule_target_legal_article_candidates(self, risk_text: str, tenant_id: str) -> List[Any]: + candidates: List[Any] = [] + for rule in LEGAL_ARTICLE_REFERENCE_RULES: + matched_trigger = next((trigger for trigger in rule["triggers"] if trigger in risk_text), None) + if not matched_trigger: + continue + for source_id, article_no in rule["targets"]: + source_titles = LEGAL_SOURCE_ALIAS_MAP.get(source_id, []) + if not source_titles or not article_no: + continue + candidates.extend( + self.legal_article_repository.search_published_for_runtime( + tenant_id, + source_titles=source_titles, + article_nos=[article_no], + limit=10, + ) + ) + return self._merge_legal_article_candidates(candidates) + @staticmethod def _merge_legal_article_candidates(*groups: List[Any]) -> List[Any]: merged: List[Any] = [] @@ -1761,6 +1782,7 @@ class ReviewService: def _resolve_legal_article_refs_for_risk(self, item: Dict[str, Any], candidates: List[Any], by_source_article: Dict[tuple, Any]) -> List[Dict[str, Any]]: risk_text = self._build_legal_reference_search_text(item) selected: List[tuple[Any, str]] = [] + by_alias_article = self._build_legal_article_alias_index(candidates) for article in candidates: if self._is_direct_legal_article_match(risk_text, article): @@ -1771,7 +1793,7 @@ class ReviewService: if not matched_trigger: continue for target in rule["targets"]: - article = by_source_article.get(target) + article = by_source_article.get(target) or by_alias_article.get(target) if article is not None: selected.append((article, f"命中风险关键词：{matched_trigger}")) @@ -1789,6 +1811,23 @@ class ReviewService: break return refs + def _build_legal_article_alias_index(self, candidates: List[Any]) -> Dict[tuple, Any]: + by_alias_article: Dict[tuple, Any] = {} + for article in candidates: + source_title = str(getattr(article, "source_title", "") or "").lower() + article_no = getattr(article, "article_no", None) + if not source_title or not article_no: + continue + for source_id, aliases in LEGAL_SOURCE_ALIAS_MAP.items(): + names = [source_id, *(aliases or [])] + if any(name and self._is_same_legal_source_name(source_title, str(name).lower()) for name in names): + by_alias_article[(source_id, article_no)] = article + return by_alias_article + + @staticmethod + def _is_same_legal_source_name(source_title: str, alias: str) -> bool: + return bool(alias and (alias in source_title or source_title in alias)) + def _build_legal_reference_search_text(self, item: Dict[str, Any]) -> str: evidence = item.get("evidence") or {} parts = [ diff --git a/backend/tests/test_review_service.py b/backend/tests/test_review_service.py index 65252e76..df644826 100644 --- a/backend/tests/test_review_service.py +++ b/backend/tests/test_review_service.py @@ -8,7 +8,7 @@ from reportlab.pdfbase import pdfmetrics from reportlab.pdfbase.cidfonts import UnicodeCIDFont from reportlab.pdfgen import canvas -from app.services.review_service import ReviewService +from app.services.review_service import LEGAL_ARTICLE_REFERENCE_RULES, LEGAL_SOURCE_ALIAS_MAP, ReviewService class StubRepository: @@ -366,6 +366,36 @@ class ReviewServiceTestCase(unittest.TestCase): self.assertEqual(enriched[0]["legal_article_refs"][0]["detail_path"], "/legal-articles?article_id=lart-cyber-001") self.assertIn("第一条", enriched[0]["legal_article_refs"][0]["article_no"]) + def test_legal_article_refs_resolve_imported_source_alias_targets(self): + source_id, article_no = LEGAL_ARTICLE_REFERENCE_RULES[0]["targets"][0] + trigger = LEGAL_ARTICLE_REFERENCE_RULES[0]["triggers"][0] + source_title = LEGAL_SOURCE_ALIAS_MAP[source_id][-1] + self.service.legal_article_repository.reference_candidate_count = 0 + self.service.legal_article_repository.items = [ + StubLegalArticle( + "lart-civil-585-full", + "FLK-CIVIL-CODE-FULL", + source_title, + article_no, + ["official-law"], + f"{trigger} full article text", + ) + ] + risk_item = { + "title": f"{trigger} risk needs review", + "level": "high", + "description": trigger, + "suggestion": "complete clause", + "legal_basis": "", + "evidence": {"quote": trigger, "clause_title": "liability"}, + } + + enriched = self.service._attach_legal_article_refs([risk_item], self.current_user["tenant_id"]) + + self.assertEqual(enriched[0]["legal_article_refs"][0]["id"], "lart-civil-585-full") + self.assertEqual(enriched[0]["legal_article_refs"][0]["source_id"], "FLK-CIVIL-CODE-FULL") + self.assertEqual(enriched[0]["legal_article_refs"][0]["article_no"], article_no) + def test_report_metadata_and_content_match_review(self): created = self.service.create_review_task( tenant_id=self.current_user["tenant_id"],