import argparse
|
import csv
|
import hashlib
|
from collections import Counter, defaultdict
|
from datetime import datetime, timedelta, timezone
|
from pathlib import Path
|
|
|
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 join_unique(values, limit=8):
|
out = []
|
for value in values:
|
if value and value not in out:
|
out.append(value)
|
return "|".join(out[:limit])
|
|
|
def evidence_strength(status, location_status):
|
if status in {"PRICE_INVENTORY_SOURCE_MATCHED", "COMPANY_SOURCE_MATCHED"} and location_status == "LOCATED_IN_CONVERTED_TEXT":
|
return "MEDIUM_DRAFT"
|
if status.endswith("MATCHED"):
|
return "LOW_DRAFT"
|
return "GAP_REVIEW"
|
|
|
def main():
|
parser = argparse.ArgumentParser()
|
parser.add_argument("--project-root", default=".")
|
parser.add_argument("--cards", required=True)
|
parser.add_argument("--locations", required=True)
|
parser.add_argument("--price-supplement", required=True)
|
parser.add_argument("--source-manifest", required=True)
|
parser.add_argument("--company-sources", 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-CROSSCHECK-006")
|
args = parser.parse_args()
|
|
project_root = Path(args.project_root).resolve()
|
cards = read_csv(project_root / args.cards)
|
locations = {row.get("evidence_card_id"): row for row in read_csv(project_root / args.locations)}
|
price_rows = read_csv(project_root / args.price_supplement)
|
source_rows = read_csv(project_root / args.source_manifest)
|
company_rows = read_csv(project_root / args.company_sources)
|
created_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds")
|
|
price_by_metal = defaultdict(list)
|
for row in price_rows:
|
price_by_metal[row.get("metal", "")].append(row)
|
|
company_by_metal = defaultdict(list)
|
for row in company_rows:
|
for tag in split_tags(row.get("primary_metal_tags", "")):
|
company_by_metal[tag].append(row)
|
|
source_by_name = {row.get("source_name", ""): row for row in source_rows}
|
output_rows = []
|
for card in cards:
|
metal_tags = split_tags(card.get("metal_tags", ""))
|
theme_tags = set(split_tags(card.get("theme_tags", "")))
|
loc = locations.get(card.get("evidence_card_id"), {})
|
matched_prices = []
|
matched_companies = []
|
for metal in metal_tags:
|
matched_prices.extend(price_by_metal.get(metal, []))
|
matched_companies.extend(company_by_metal.get(metal, []))
|
|
evidence_kind = card.get("evidence_kind", "")
|
if evidence_kind in {"PRICE_OR_MARKET_DATA", "INVENTORY"} or "price" in theme_tags or "inventory" in theme_tags:
|
if matched_prices:
|
status = "PRICE_INVENTORY_SOURCE_MATCHED"
|
next_action = "compare value/unit/date against matched external price or inventory source"
|
else:
|
status = "NEEDS_PRICE_INVENTORY_EXTERNAL_SOURCE"
|
next_action = "supplement exchange/SMM/industry-association source for the metal and indicator"
|
elif {"supply", "company", "project"} & theme_tags or evidence_kind == "SUPPLY_CAPACITY_OR_OUTPUT":
|
if matched_companies:
|
status = "COMPANY_SOURCE_MATCHED"
|
next_action = "verify capacity/output/project wording against company official source"
|
else:
|
status = "NEEDS_COMPANY_OFFICIAL_SOURCE"
|
next_action = "supplement annual report, exchange filing, or company IR source"
|
else:
|
status = "NEEDS_MANUAL_EXTERNAL_CROSSCHECK"
|
next_action = "classify external source requirement manually"
|
|
source_names = [row.get("source_name", "") for row in matched_prices]
|
source_urls = [row.get("source_url", "") for row in matched_prices]
|
price_indicators = [
|
f"{row.get('metal','')}:{row.get('indicator','')}:{row.get('value','')} {row.get('unit','')}"
|
for row in matched_prices
|
]
|
company_names = [row.get("company_name", "") for row in matched_companies]
|
company_urls = [row.get("source_url", "") for row in matched_companies]
|
location_status = loc.get("location_status", "")
|
strength = evidence_strength(status, location_status)
|
output_rows.append(
|
{
|
"crosscheck_id": f"YS-CROSSCHECK-006-{len(output_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", ""),
|
"metal_tags": card.get("metal_tags", ""),
|
"theme_tags": card.get("theme_tags", ""),
|
"evidence_kind": evidence_kind,
|
"location_status": location_status,
|
"precheck_page_no": loc.get("precheck_page_no", ""),
|
"precheck_location": loc.get("precheck_location", ""),
|
"cross_check_status": status,
|
"evidence_strength_draft": strength,
|
"matched_price_inventory_count": str(len(matched_prices)),
|
"matched_price_inventory_indicators": join_unique(price_indicators),
|
"matched_price_inventory_sources": join_unique(source_names),
|
"matched_price_inventory_urls": join_unique(source_urls, limit=5),
|
"matched_company_source_count": str(len(matched_companies)),
|
"matched_company_names": join_unique(company_names),
|
"matched_company_source_urls": join_unique(company_urls, limit=5),
|
"source_manifest_count": str(len(source_by_name)),
|
"value_unit_candidates": card.get("value_unit_candidates", ""),
|
"date_candidates": card.get("date_candidates", ""),
|
"next_action": next_action,
|
"review_status": "DRAFT_FOR_REVIEW",
|
"created_at": created_at,
|
}
|
)
|
|
output_path = project_root / args.output
|
fields = list(output_rows[0].keys()) if output_rows else ["crosscheck_id", "case_id", "run_id", "review_status"]
|
write_csv(output_path, fields, output_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_external_crosscheck",
|
"artifact_path": output_path.relative_to(project_root).as_posix(),
|
"row_count": str(len(output_rows)),
|
"sha256": output_sha,
|
"review_status": "DRAFT_FOR_REVIEW",
|
"created_at": created_at,
|
}
|
],
|
)
|
|
status_counts = Counter(row["cross_check_status"] for row in output_rows)
|
strength_counts = Counter(row["evidence_strength_draft"] for row in output_rows)
|
summary_path = project_root / args.summary
|
lines = [
|
"# \u5173\u952e\u4e8b\u5b9e\u5916\u90e8\u4ea4\u53c9\u9a8c\u8bc1\u6620\u5c04 PASS-006 \u6458\u8981",
|
"",
|
"\u72b6\u6001\uff1aDRAFT_FOR_REVIEW",
|
f"\u751f\u6210\u65f6\u95f4\uff1a{created_at}",
|
"",
|
"## \u8f93\u51fa",
|
"",
|
f"- \u4ea4\u53c9\u9a8c\u8bc1\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(output_rows)}",
|
f"- sha256\uff1a`{output_sha}`",
|
"",
|
"## \u4ea4\u53c9\u9a8c\u8bc1\u72b6\u6001",
|
"",
|
"| \u72b6\u6001 | \u6570\u91cf |",
|
"|---|---:|",
|
]
|
for key, count in status_counts.most_common():
|
lines.append(f"| {key} | {count} |")
|
lines.extend(["", "## \u8bc1\u636e\u5f3a\u5ea6\u8349\u5224", "", "| \u72b6\u6001 | \u6570\u91cf |", "|---|---:|"])
|
for key, count in strength_counts.most_common():
|
lines.append(f"| {key} | {count} |")
|
lines.extend(
|
[
|
"",
|
"## \u8fb9\u754c",
|
"",
|
"\u672c\u8f6e\u53ea\u628a\u5173\u952e\u4e8b\u5b9e\u5019\u9009\u4e0e\u5df2\u843d\u5730\u7684\u4ef7\u683c\u5e93\u5b58\u8865\u6570\u3001\u516c\u53f8\u5b98\u65b9\u6765\u6e90\u5361\u548c converted text \u5b9a\u4f4d\u7ed3\u679c\u5efa\u7acb\u6620\u5c04\u3002\u547d\u4e2d\u5916\u90e8\u6765\u6e90\u4e0d\u7b49\u4e8e\u6570\u503c\u3001\u5355\u4f4d\u3001\u53e3\u5f84\u5df2\u4eba\u5de5\u6838\u5b9e\uff1b\u6b63\u5f0f\u8bc1\u636e\u5347\u7ea7\u524d\u4ecd\u9700\u590d\u6838\u6765\u6e90\u65e5\u671f\u3001\u539f\u6587\u8868\u683c\u548c\u5916\u90e8\u6570\u503c\u4e00\u81f4\u6027\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(output_rows)}")
|
print(dict(status_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()
|
