import argparse
|
import csv
|
import hashlib
|
import re
|
from collections import Counter
|
from datetime import datetime, timedelta, timezone
|
from pathlib import Path
|
|
|
PAGE_RE = re.compile(r"\[\[PAGE\s+(\d+)\]\]")
|
VALUE_RE = re.compile(
|
r"[-+]?\d+(?:\.\d+)?\s*(?:%|\u4e07\u5428|\u5428|\u4ebf\u5143|\u5143/\u5428|\u7f8e\u5143/\u5428|GWh)"
|
)
|
TABLE_HINTS = [
|
"\u8868",
|
"\u56fe",
|
"\u8d44\u6599\u6765\u6e90",
|
"\u5355\u4f4d\uff1a",
|
"\u5e93\u5b58",
|
"\u4ef7\u683c",
|
"\u4ea7\u91cf",
|
"\u4f9b\u7ed9",
|
]
|
|
|
def read_csv(path):
|
with path.open("r", encoding="utf-8-sig", newline="") as handle:
|
return list(csv.DictReader(handle))
|
|
|
def write_csv(path, fieldnames, rows):
|
path.parent.mkdir(parents=True, exist_ok=True)
|
with path.open("w", encoding="utf-8-sig", newline="") as handle:
|
writer = csv.DictWriter(handle, fieldnames=fieldnames)
|
writer.writeheader()
|
writer.writerows(rows)
|
|
|
def sha256_file(path):
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
def normalize(text):
|
return re.sub(r"\s+", "", text or "")
|
|
|
def read_lines(path):
|
try:
|
return path.read_text(encoding="utf-8", errors="replace").splitlines()
|
except FileNotFoundError:
|
return []
|
|
|
def page_marks(lines):
|
marks = []
|
for line_no, line in enumerate(lines, start=1):
|
match = PAGE_RE.search(line)
|
if match:
|
marks.append((line_no, match.group(1)))
|
return marks
|
|
|
def page_for_line(marks, line_no):
|
current = ""
|
for mark_line, page_no in marks:
|
if mark_line <= line_no:
|
current = page_no
|
else:
|
break
|
return current
|
|
|
def table_score(lines):
|
score = 0
|
joined = "\n".join(lines)
|
if VALUE_RE.search(joined):
|
score += 1
|
if sum(1 for line in lines if len(re.findall(r"\s{2,}", line)) >= 2) >= 2:
|
score += 1
|
if any(hint in joined for hint in TABLE_HINTS):
|
score += 1
|
return score
|
|
|
def find_line(lines, evidence, declared_page):
|
needle = normalize(evidence)
|
if needle:
|
for line_no, line in enumerate(lines, start=1):
|
if needle in normalize(line):
|
return line_no, "EXACT_NORMALIZED_LINE"
|
prefix = needle[:80]
|
if prefix:
|
for idx in range(len(lines)):
|
window = "".join(normalize(x) for x in lines[max(0, idx - 2) : min(len(lines), idx + 3)])
|
if prefix in window:
|
return idx + 1, "WINDOW_PREFIX_MATCH"
|
if declared_page and declared_page.isdigit():
|
for line_no, page_no in page_marks(lines):
|
if page_no == declared_page:
|
return line_no, "PAGE_MARK_FALLBACK"
|
return None, "NO_MATCH"
|
|
|
def main():
|
parser = argparse.ArgumentParser()
|
parser.add_argument("--project-root", default=".")
|
parser.add_argument("--input", required=True)
|
parser.add_argument("--output", required=True)
|
parser.add_argument("--manifest", required=True)
|
parser.add_argument("--summary", required=True)
|
parser.add_argument("--run-id", default="RUN-ANA-YS-FACT-LOCATION-003")
|
args = parser.parse_args()
|
|
project_root = Path(args.project_root).resolve()
|
cards = read_csv(project_root / args.input)
|
created_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds")
|
rows = []
|
for card in cards:
|
converted_rel = card.get("converted_text_path", "")
|
lines = read_lines(project_root / converted_rel)
|
marks = page_marks(lines)
|
declared_page = card.get("matched_page_no") or card.get("declared_page_no")
|
line_no, match_method = find_line(lines, card.get("evidence_text", ""), declared_page)
|
if line_no:
|
context = lines[max(0, line_no - 4) : min(len(lines), line_no + 5)]
|
location_status = "LOCATED_IN_CONVERTED_TEXT"
|
location = f"line:{line_no}"
|
precheck_page_no = page_for_line(marks, line_no) or declared_page
|
location_type = "TABLE_CANDIDATE" if table_score(context) >= 2 else "PARAGRAPH_CANDIDATE"
|
else:
|
context = []
|
location_status = "NEEDS_MANUAL_SOURCE_PAGE_REVIEW"
|
location = ""
|
precheck_page_no = declared_page
|
location_type = "UNLOCATED"
|
rows.append(
|
{
|
"location_precheck_id": f"YS-FACT-LOC-003-{len(rows) + 1:04d}",
|
"case_id": card.get("case_id", ""),
|
"batch_id": card.get("batch_id", ""),
|
"run_id": args.run_id,
|
"evidence_card_id": card.get("evidence_card_id", ""),
|
"fact_id": card.get("fact_id", ""),
|
"doc_id": card.get("doc_id", ""),
|
"converted_text_path": converted_rel,
|
"declared_page_no": card.get("declared_page_no", ""),
|
"matched_page_no_pass002": card.get("matched_page_no", ""),
|
"precheck_page_no": precheck_page_no,
|
"precheck_location": location,
|
"match_method": match_method,
|
"location_type_candidate": location_type,
|
"location_status": location_status,
|
"evidence_text": card.get("evidence_text", ""),
|
"location_context": "\\n".join(context)[:1200],
|
"external_cross_check_status": card.get("external_cross_check_status", ""),
|
"review_status": "DRAFT_FOR_REVIEW",
|
"created_at": created_at,
|
}
|
)
|
|
output_path = project_root / args.output
|
fieldnames = list(rows[0].keys()) if rows else [
|
"location_precheck_id",
|
"case_id",
|
"batch_id",
|
"run_id",
|
"review_status",
|
]
|
write_csv(output_path, fieldnames, rows)
|
output_sha = sha256_file(output_path)
|
|
manifest_path = project_root / args.manifest
|
manifest_rows = [
|
{
|
"case_id": "ANA-YS-INDUSTRY-001",
|
"batch_id": "BATCH-001+BATCH-003",
|
"run_id": args.run_id,
|
"artifact_type": "key_fact_location_precheck",
|
"artifact_path": output_path.relative_to(project_root).as_posix(),
|
"row_count": str(len(rows)),
|
"sha256": output_sha,
|
"review_status": "DRAFT_FOR_REVIEW",
|
"created_at": created_at,
|
}
|
]
|
write_csv(
|
manifest_path,
|
["case_id", "batch_id", "run_id", "artifact_type", "artifact_path", "row_count", "sha256", "review_status", "created_at"],
|
manifest_rows,
|
)
|
|
status_counts = Counter(row["location_status"] for row in rows)
|
type_counts = Counter(row["location_type_candidate"] for row in rows)
|
summary_path = project_root / args.summary
|
summary_lines = [
|
"# \u5173\u952e\u4e8b\u5b9e\u5b9a\u4f4d\u590d\u6838\u5305 PASS-003 \u6458\u8981",
|
"",
|
"\u72b6\u6001\uff1aDRAFT_FOR_REVIEW",
|
f"\u751f\u6210\u65f6\u95f4\uff1a{created_at}",
|
"",
|
"## \u8f93\u51fa",
|
"",
|
f"- \u5b9a\u4f4d\u590d\u6838\u8868\uff1a`{output_path.relative_to(project_root).as_posix()}`",
|
f"- manifest\uff1a`{manifest_path.relative_to(project_root).as_posix()}`",
|
f"- \u8bb0\u5f55\u6570\uff1a{len(rows)}",
|
f"- sha256\uff1a`{output_sha}`",
|
"",
|
"## \u5b9a\u4f4d\u72b6\u6001",
|
"",
|
"| \u72b6\u6001 | \u6570\u91cf |",
|
"|---|---:|",
|
]
|
for key, count in status_counts.most_common():
|
summary_lines.append(f"| {key} | {count} |")
|
summary_lines.extend(["", "## \u5b9a\u4f4d\u7c7b\u578b\u5019\u9009", "", "| \u7c7b\u578b | \u6570\u91cf |", "|---|---:|"])
|
for key, count in type_counts.most_common():
|
summary_lines.append(f"| {key} | {count} |")
|
summary_lines.extend(
|
[
|
"",
|
"## \u8fb9\u754c",
|
"",
|
"\u672c\u8f6e\u53ea\u5b8c\u6210 converted text \u5c42\u7684\u9875\u7801\u3001\u884c\u53f7\u548c\u4e0a\u4e0b\u6587\u5b9a\u4f4d\u9884\u590d\u6838\u3002\u8868\u683c\u5019\u9009\u4e0d\u7b49\u4e8e PDF \u539f\u9875\u8868\u683c\u5df2\u590d\u6838\uff1b\u6b63\u5f0f\u8bc1\u636e\u5347\u7ea7\u524d\u4ecd\u9700\u8981\u4eba\u5de5\u590d\u6838\u548c\u5916\u90e8\u4ea4\u53c9\u9a8c\u8bc1\u3002",
|
]
|
)
|
summary_path.parent.mkdir(parents=True, exist_ok=True)
|
summary_path.write_text("\n".join(summary_lines) + "\n", encoding="utf-8")
|
|
print(f"rows={len(rows)}")
|
print(f"located={status_counts.get('LOCATED_IN_CONVERTED_TEXT', 0)}")
|
print(f"unlocated={status_counts.get('NEEDS_MANUAL_SOURCE_PAGE_REVIEW', 0)}")
|
print(f"output={output_path.relative_to(project_root).as_posix()}")
|
print(f"manifest={manifest_path.relative_to(project_root).as_posix()}")
|
print(f"summary={summary_path.relative_to(project_root).as_posix()}")
|
|
|
if __name__ == "__main__":
|
main()
|
