Spaces:
Sleeping
Sleeping
File size: 4,220 Bytes
300df0f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """
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
# ------------------------------------------------------------------
@staticmethod
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)
@staticmethod
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,
)
|