import argparse
|
import csv
|
import hashlib
|
from collections import Counter
|
from datetime import datetime, timedelta, timezone
|
from pathlib import Path
|
|
|
CORE_METALS_HIGH = {"\u94dc", "\u94dd", "\u91d1", "\u94f6", "\u9502", "\u7a00\u571f"}
|
CORE_METALS_MEDIUM = {"\u9521", "\u954d", "\u94b4", "\u950c", "\u94c5", "\u94a8", "\u9511", "\u94bc", "\u9511"}
|
|
|
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 split_tags(value):
|
return [part.strip() for part in (value or "").split("|") if part.strip()]
|
|
|
def priority(metals, status, has_value, located):
|
metal_set = set(metals)
|
if metal_set & CORE_METALS_HIGH and has_value and located:
|
return "HIGH"
|
if status == "NEEDS_COMPANY_OFFICIAL_SOURCE" and metal_set & CORE_METALS_HIGH:
|
return "HIGH"
|
if metal_set & CORE_METALS_MEDIUM or has_value:
|
return "MEDIUM"
|
return "LOW"
|
|
|
def source_type(status, themes):
|
theme_set = set(themes)
|
if status == "NEEDS_PRICE_INVENTORY_EXTERNAL_SOURCE":
|
if "inventory" in theme_set:
|
return "exchange_inventory_or_industry_inventory"
|
return "exchange_or_SMM_or_industry_price"
|
if status == "NEEDS_COMPANY_OFFICIAL_SOURCE":
|
if "project" in theme_set:
|
return "annual_report_exchange_filing_or_project_announcement"
|
return "annual_report_exchange_filing_or_company_ir"
|
return "manual_source_classification"
|
|
|
def question(status, metals, themes, value_candidates):
|
metal_text = "|".join(metals) or "UNKNOWN_METAL"
|
theme_text = "|".join(themes) or "UNKNOWN_THEME"
|
if status == "NEEDS_PRICE_INVENTORY_EXTERNAL_SOURCE":
|
return f"\u4e3a {metal_text} / {theme_text} \u8865\u5145\u53ef\u590d\u6838\u7684\u4ef7\u683c\u3001\u5e93\u5b58\u6216\u4ea4\u6613\u6240/\u884c\u4e1a\u6765\u6e90\uff0c\u6838\u5bf9\u5019\u9009\u6570\u503c {value_candidates or 'NA'}"
|
if status == "NEEDS_COMPANY_OFFICIAL_SOURCE":
|
return f"\u4e3a {metal_text} / {theme_text} \u8865\u5145\u5e74\u62a5\u3001\u4ea4\u6613\u6240\u516c\u544a\u6216\u516c\u53f8 IR \u6765\u6e90\uff0c\u6838\u5bf9\u4ea7\u80fd/\u9879\u76ee/\u4f9b\u7ed9\u8868\u8ff0"
|
return f"\u4eba\u5de5\u5224\u65ad {metal_text} / {theme_text} \u6240\u9700\u5916\u90e8\u6765\u6e90"
|
|
|
def main():
|
parser = argparse.ArgumentParser()
|
parser.add_argument("--project-root", default=".")
|
parser.add_argument("--crosscheck", required=True)
|
parser.add_argument("--cards", required=True)
|
parser.add_argument("--locations", 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-GAP-PRIORITY-007")
|
args = parser.parse_args()
|
|
project_root = Path(args.project_root).resolve()
|
crosscheck = read_csv(project_root / args.crosscheck)
|
cards = {row.get("evidence_card_id"): row for row in read_csv(project_root / args.cards)}
|
locations = {row.get("evidence_card_id"): row for row in read_csv(project_root / args.locations)}
|
targets = [
|
row
|
for row in crosscheck
|
if row.get("evidence_strength_draft") == "GAP_REVIEW" or row.get("cross_check_status", "").startswith("NEEDS_")
|
]
|
created_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds")
|
rows = []
|
for item in targets:
|
card = cards.get(item.get("evidence_card_id"), {})
|
loc = locations.get(item.get("evidence_card_id"), {})
|
metals = split_tags(item.get("metal_tags", ""))
|
themes = split_tags(item.get("theme_tags", ""))
|
has_value = bool(item.get("value_unit_candidates") or card.get("value_unit_candidates"))
|
located = loc.get("location_status") == "LOCATED_IN_CONVERTED_TEXT"
|
status = item.get("cross_check_status", "")
|
pr = priority(metals, status, has_value, located)
|
stype = source_type(status, themes)
|
rows.append(
|
{
|
"gap_priority_id": f"YS-GAP-PRIORITY-007-{len(rows) + 1:04d}",
|
"case_id": item.get("case_id", ""),
|
"batch_id": item.get("batch_id", ""),
|
"run_id": args.run_id,
|
"evidence_card_id": item.get("evidence_card_id", ""),
|
"fact_id": item.get("fact_id", ""),
|
"doc_id": item.get("doc_id", ""),
|
"metal_tags": item.get("metal_tags", ""),
|
"theme_tags": item.get("theme_tags", ""),
|
"cross_check_status": status,
|
"gap_type": "PRICE_INVENTORY_SOURCE_GAP" if "PRICE" in status else "COMPANY_OFFICIAL_SOURCE_GAP" if "COMPANY" in status else "MANUAL_SOURCE_GAP",
|
"priority": pr,
|
"preferred_source_type": stype,
|
"supplement_question": question(status, metals, themes, item.get("value_unit_candidates", "")),
|
"located_in_converted_text": "YES" if located else "NO",
|
"precheck_page_no": loc.get("precheck_page_no", ""),
|
"precheck_location": loc.get("precheck_location", ""),
|
"value_unit_candidates": item.get("value_unit_candidates", ""),
|
"date_candidates": item.get("date_candidates", ""),
|
"next_action": "supplement_now" if pr == "HIGH" else "queue_for_next_pass" if pr == "MEDIUM" else "defer_low_priority",
|
"review_status": "DRAFT_FOR_REVIEW",
|
"created_at": created_at,
|
}
|
)
|
|
output_path = project_root / args.output
|
fields = list(rows[0].keys()) if rows else ["gap_priority_id", "case_id", "run_id", "review_status"]
|
write_csv(output_path, fields, rows)
|
output_sha = sha256_file(output_path)
|
manifest_path = project_root / args.manifest
|
write_csv(
|
manifest_path,
|
["case_id", "batch_id", "run_id", "artifact_type", "artifact_path", "row_count", "sha256", "review_status", "created_at"],
|
[
|
{
|
"case_id": "ANA-YS-INDUSTRY-001",
|
"batch_id": "BATCH-001+BATCH-003",
|
"run_id": args.run_id,
|
"artifact_type": "key_fact_gap_priority",
|
"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,
|
}
|
],
|
)
|
|
priority_counts = Counter(row["priority"] for row in rows)
|
gap_counts = Counter(row["gap_type"] for row in rows)
|
action_counts = Counter(row["next_action"] for row in rows)
|
summary_path = project_root / args.summary
|
lines = [
|
"# \u5173\u952e\u4e8b\u5b9e\u5916\u90e8\u6765\u6e90\u7f3a\u53e3\u4f18\u5148\u7ea7 PASS-007 \u6458\u8981",
|
"",
|
"\u72b6\u6001\uff1aDRAFT_FOR_REVIEW",
|
f"\u751f\u6210\u65f6\u95f4\uff1a{created_at}",
|
"",
|
"## \u8f93\u51fa",
|
"",
|
f"- \u7f3a\u53e3\u961f\u5217\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}`",
|
"",
|
"## \u4f18\u5148\u7ea7",
|
"",
|
"| \u4f18\u5148\u7ea7 | \u6570\u91cf |",
|
"|---|---:|",
|
]
|
for key, count in priority_counts.most_common():
|
lines.append(f"| {key} | {count} |")
|
lines.extend(["", "## \u7f3a\u53e3\u7c7b\u578b", "", "| \u7c7b\u578b | \u6570\u91cf |", "|---|---:|"])
|
for key, count in gap_counts.most_common():
|
lines.append(f"| {key} | {count} |")
|
lines.extend(["", "## \u4e0b\u4e00\u6b65", "", "| \u52a8\u4f5c | \u6570\u91cf |", "|---|---:|"])
|
for key, count in action_counts.most_common():
|
lines.append(f"| {key} | {count} |")
|
lines.extend(
|
[
|
"",
|
"## \u8fb9\u754c",
|
"",
|
"\u672c\u8f6e\u53ea\u662f\u5bf9 PASS-006 \u7f3a\u53e3\u505a\u4f18\u5148\u7ea7\u62c6\u5206\uff0c\u4e0d\u8868\u793a\u7f3a\u53e3\u5df2\u8865\u9f50\uff0c\u4e0d\u4f5c\u4e3a\u6b63\u5f0f\u8bc1\u636e\u6216\u6b63\u5f0f\u6307\u6807\u3002",
|
]
|
)
|
summary_path.parent.mkdir(parents=True, exist_ok=True)
|
summary_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
print(f"rows={len(rows)}")
|
print(dict(priority_counts))
|
print(dict(gap_counts))
|
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()
|
