107 lines
3.1 KiB
Python
107 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
from django.conf import settings
|
|
|
|
from review_agent.models import RegulatoryRuleVersion
|
|
|
|
|
|
DEFAULT_RULE_CODE = "nmpa_ivd_registration_v1"
|
|
DEFAULT_RULE_PATH = (
|
|
Path(settings.BASE_DIR)
|
|
/ "review_agent"
|
|
/ "regulatory_review"
|
|
/ "rules"
|
|
/ "nmpa_ivd_registration_v1.yaml"
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RuleVersionCheck:
|
|
status: str
|
|
code: str
|
|
path: Path
|
|
current_hash: str
|
|
database_hash: str = ""
|
|
record: RegulatoryRuleVersion | None = None
|
|
|
|
|
|
def compute_file_sha256(path: str | Path) -> str:
|
|
file_path = Path(path)
|
|
digest = hashlib.sha256()
|
|
with file_path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def load_rule_file(path: str | Path | None = None) -> dict:
|
|
rule_path = Path(path) if path else DEFAULT_RULE_PATH
|
|
with rule_path.open("r", encoding="utf-8") as handle:
|
|
payload = yaml.safe_load(handle) or {}
|
|
if payload.get("code") != DEFAULT_RULE_CODE:
|
|
raise ValueError(f"规则 code 必须为 {DEFAULT_RULE_CODE}")
|
|
if not isinstance(payload.get("requirements"), list) or not payload["requirements"]:
|
|
raise ValueError("规则文件必须包含 requirements 列表。")
|
|
return payload
|
|
|
|
|
|
def check_rule_version(
|
|
*,
|
|
path: str | Path | None = None,
|
|
update_missing: bool = True,
|
|
) -> RuleVersionCheck:
|
|
rule_path = Path(path) if path else DEFAULT_RULE_PATH
|
|
rule_set = load_rule_file(rule_path)
|
|
current_hash = compute_file_sha256(rule_path)
|
|
record = RegulatoryRuleVersion.objects.filter(code=rule_set["code"]).first()
|
|
yaml_path = str(rule_path.relative_to(settings.BASE_DIR))
|
|
|
|
if record is None:
|
|
if not update_missing:
|
|
return RuleVersionCheck(
|
|
status="missing",
|
|
code=rule_set["code"],
|
|
path=rule_path,
|
|
current_hash=current_hash,
|
|
)
|
|
record = RegulatoryRuleVersion.objects.create(
|
|
code=rule_set["code"],
|
|
name=rule_set.get("name") or rule_set["code"],
|
|
yaml_path=yaml_path,
|
|
yaml_hash=current_hash,
|
|
rag_collection=rule_set.get("rag_collection", ""),
|
|
status=RegulatoryRuleVersion.Status.ACTIVE,
|
|
)
|
|
return RuleVersionCheck(
|
|
status="created",
|
|
code=record.code,
|
|
path=rule_path,
|
|
current_hash=current_hash,
|
|
database_hash=record.yaml_hash,
|
|
record=record,
|
|
)
|
|
|
|
if record.yaml_hash != current_hash:
|
|
return RuleVersionCheck(
|
|
status="mismatch",
|
|
code=record.code,
|
|
path=rule_path,
|
|
current_hash=current_hash,
|
|
database_hash=record.yaml_hash,
|
|
record=record,
|
|
)
|
|
|
|
return RuleVersionCheck(
|
|
status="ok",
|
|
code=record.code,
|
|
path=rule_path,
|
|
current_hash=current_hash,
|
|
database_hash=record.yaml_hash,
|
|
record=record,
|
|
)
|