Spaces:
Sleeping
Sleeping
| """ | |
| Validity computer for Articles and EffectiveArticles — T3.5 (Người B) | |
| Determines is_current based on: | |
| 1. Document.tinh_trang_hieu_luc | |
| 2. Incoming SUPERSEDES / PARTIALLY_SUPERSEDES doc-level relationships (from T1.7, Người A) | |
| 3. Article-level BAI_BO modifications (from T2.3 MODIFIES edges) | |
| Input : Neo4j graph | |
| Output: ValidityReport (pure data) → passed to EffectiveArticleWriter.write_validity() | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| from typing import TYPE_CHECKING | |
| from .models import ValidityReport, ArticleValidity, ValidityStatus | |
| if TYPE_CHECKING: | |
| from neo4j import Driver | |
| logger = logging.getLogger(__name__) | |
| # Map from Vietnamese tinh_trang_hieu_luc to ValidityStatus enum | |
| _TINH_TRANG_MAP = { | |
| "Còn hiệu lực": ValidityStatus.CON_HIEU_LUC, | |
| "Hết hiệu lực toàn bộ": ValidityStatus.HET_HIEU_LUC, | |
| "Hết hiệu lực một phần": ValidityStatus.HET_HIEU_LUC_MOT_PHAN, | |
| "Ngưng hiệu lực": ValidityStatus.NGUNG_HIEU_LUC, | |
| } | |
| class CurrentStatusComputer: | |
| """ | |
| Computes is_current for all Article nodes by walking Neo4j graph. | |
| Usage | |
| ----- | |
| computer = CurrentStatusComputer(driver) | |
| report = computer.compute_all() | |
| # report.decisions contains one ArticleValidity per article | |
| # Then persist: | |
| writer.write_validity(report) | |
| """ | |
| def __init__(self, driver: "Driver") -> None: | |
| self._driver = driver | |
| # ------------------------------------------------------------------ | |
| # Public API | |
| # ------------------------------------------------------------------ | |
| def compute_all(self) -> ValidityReport: | |
| """ | |
| Compute is_current for every Article in Neo4j. | |
| Algorithm (in priority order): | |
| 1. Doc is "Hết hiệu lực toàn bộ" or "Ngưng hiệu lực" | |
| → ALL its Articles: is_current=False, reason="doc_het_hieu_luc" | |
| 2. Doc has incoming [:SUPERSEDES] edge from another doc | |
| → ALL its Articles: is_current=False, reason="superseded" | |
| 3. Doc has incoming [:PARTIALLY_SUPERSEDES] from another doc | |
| → SPECIFIC Khoản/Điểm: is_current=False, reason="partially_superseded" | |
| 4. Article has MODIFIES edge with action="bai_bo" targeting it | |
| → That Article: is_current=False, reason="bai_bo" | |
| 5. Default: is_current=True, reason="con_hieu_luc" | |
| Returns | |
| ------- | |
| ValidityReport | |
| decisions: one ArticleValidity per Article | |
| TODO (T3.5): implement this method. | |
| Suggested query to get doc-level validity: | |
| MATCH (d:Document) | |
| OPTIONAL MATCH (sup:Document)-[:SUPERSEDES]->(d) | |
| RETURN d.id, d.tinh_trang_hieu_luc, count(sup) > 0 AS is_superseded | |
| Then for each doc, if HET_HIEU_LUC or is_superseded: | |
| MATCH (d:Document {id: $doc_id})-[:HAS_ARTICLE]->(a:Article) | |
| RETURN a.uid, d.loai_van_ban | |
| """ | |
| raise NotImplementedError("T3.5: implement compute_all()") | |
| def compute_for_doc(self, doc_id: str) -> list[ArticleValidity]: | |
| """ | |
| Compute is_current for all Articles of one document. | |
| Useful for incremental updates when a new amendment is processed. | |
| TODO (T3.5): implement. | |
| """ | |
| raise NotImplementedError("T3.5: implement compute_for_doc()") | |
| # ------------------------------------------------------------------ | |
| # Internal helpers | |
| # ------------------------------------------------------------------ | |
| def _tinh_trang_to_status(raw: str) -> ValidityStatus: | |
| """Map Document.tinh_trang_hieu_luc string to ValidityStatus enum.""" | |
| return _TINH_TRANG_MAP.get(raw.strip() if raw else "", ValidityStatus.UNKNOWN) | |
| def _make_decision( | |
| article_uid: str, | |
| doc_id: str, | |
| loai_van_ban: str, | |
| is_current: bool, | |
| reason: str, | |
| ) -> ArticleValidity: | |
| return ArticleValidity( | |
| article_uid=article_uid, | |
| doc_id=doc_id, | |
| loai_van_ban=loai_van_ban, | |
| is_current=is_current, | |
| reason=reason, | |
| ) | |