74 lines
2.8 KiB
Python
74 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
|
|
from review_agent.models import FileSummaryBatch
|
|
from review_agent.regulatory_review.schemas import Finding
|
|
|
|
|
|
def run_completeness_check(
|
|
batch: FileSummaryBatch,
|
|
rule_set: dict,
|
|
progress_callback: Callable[[dict[str, object]], None] | None = None,
|
|
) -> list[Finding]:
|
|
items = list(batch.items.order_by("file_index"))
|
|
findings: list[Finding] = []
|
|
requirements = [
|
|
requirement
|
|
for requirement in rule_set.get("requirements", [])
|
|
if requirement.get("type") in {"required", "conditional", "recommended", "chapter", "directory"}
|
|
]
|
|
total = len(requirements)
|
|
for index, requirement in enumerate(requirements, start=1):
|
|
if requirement.get("type") not in {"required", "conditional", "recommended", "chapter", "directory"}:
|
|
continue
|
|
matched = [
|
|
item
|
|
for item in items
|
|
if _matches_item(
|
|
item.file_name,
|
|
item.relative_path,
|
|
item.directory_level,
|
|
[*requirement.get("file_keywords", []), *requirement.get("aliases", [])],
|
|
)
|
|
]
|
|
if not matched:
|
|
findings.append(
|
|
Finding(
|
|
rule_code=requirement["code"],
|
|
category=requirement.get("category", "completeness"),
|
|
severity=requirement.get("severity", "medium"),
|
|
title=f"缺少{_numbered_title(requirement)}",
|
|
detail=f"当前文件汇总批次未发现{_numbered_title(requirement)}。",
|
|
suggestion=requirement.get("suggestion", ""),
|
|
evidence={
|
|
"requirement_type": requirement.get("type"),
|
|
"matched_files": [],
|
|
"searched_keywords": requirement.get("file_keywords", []),
|
|
"searched_fields": ["file_name", "relative_path", "directory_level"],
|
|
},
|
|
)
|
|
)
|
|
if progress_callback:
|
|
progress_callback(
|
|
{
|
|
"processed": index,
|
|
"total": total,
|
|
"label": _numbered_title(requirement),
|
|
"finding_count": len(findings),
|
|
}
|
|
)
|
|
return findings
|
|
|
|
|
|
def _matches_item(file_name: str, relative_path: str, directory_level: str, keywords: list[str]) -> bool:
|
|
haystack = f"{file_name} {relative_path} {directory_level}".lower()
|
|
return any(str(keyword).lower() in haystack for keyword in keywords)
|
|
|
|
|
|
def _numbered_title(requirement: dict) -> str:
|
|
attachment4_code = requirement.get("attachment4_code")
|
|
if not attachment4_code:
|
|
return requirement["title"]
|
|
return f"{attachment4_code}{requirement['title']}"
|