from __future__ import annotations import csv import hashlib import json import re from datetime import datetime, timezone, timedelta from pathlib import Path import pandas as pd RUN_ID = "RUN-ANA-WUJI-V1-STRICT-NOTE-FULL-RERUN-20260614-001" ROOT = Path(__file__).resolve().parents[1] TZ = timezone(timedelta(hours=8)) DECISION_SOURCE_PREFIX = "CASE_ANALYSIS_ANALYST_MANUAL_BUY_POINT_CHART_REVIEW_EXTERNAL_DRAFT_BATCH" VALID_ACTIONS = {"BUY", "REVIEW_HELD"} def sha256_file(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def read_lines(path: Path) -> list[str]: return path.read_text(encoding="utf-8").splitlines() def parse_header(lines: list[str], path: Path) -> dict: header: dict[str, str] = {} for line in lines: if line.startswith("## EXT-BUY-") or line.startswith("| external_decision_id "): break if not line.startswith("- "): continue key_value = line[2:].split(":", 1) if len(key_value) != 2: continue key = key_value[0].strip() value = key_value[1].strip() if key in {"decision_operator", "decision_source", "batch_id"}: header[key] = value for key in ["decision_operator", "decision_source"]: if not header.get(key): raise RuntimeError(f"draft header missing {key}: {path.name}") if not header["decision_source"].startswith(DECISION_SOURCE_PREFIX): raise RuntimeError(f"unexpected decision_source in {path.name}: {header['decision_source']}") return header def parse_table_row(line: str, path: Path) -> dict | None: if not line.startswith("| EXT-BUY-"): return None parts = [part.strip() for part in line.strip().strip("|").split("|")] if len(parts) != 5: raise RuntimeError(f"bad table row in {path.name}: {line}") return { "external_decision_id": parts[0], "human_decision_action": parts[1], "decision_time": parts[2], "accept_code_suggestion_flag": parts[3].lower(), "human_decision_reason_cn": parts[4], "reviewer_notes": "manual chart review draft row; script parsed only explicit human fields", } def parse_long_blocks(lines: list[str], path: Path) -> list[dict]: rows: list[dict] = [] current: dict | None = None for line in lines: if line.startswith("## EXT-BUY-"): if current: rows.append(current) current = {"external_decision_id": line.replace("## ", "").strip()} continue if current is None or not line.startswith("- "): continue key_value = line[2:].split(":", 1) if len(key_value) != 2: continue key = key_value[0].strip() value = key_value[1].strip() if key == "action": current["human_decision_action"] = value elif key == "reason": current["human_decision_reason_cn"] = value elif key == "decision_time": current["decision_time"] = value elif key == "accept_code_suggestion_flag": current["accept_code_suggestion_flag"] = value.lower() elif key == "reviewer_notes": current["reviewer_notes"] = value if current: rows.append(current) for row in rows: row.setdefault("reviewer_notes", "manual chart review draft block; script parsed only explicit human fields") return rows def parse_draft(path: Path) -> list[dict]: lines = read_lines(path) header = parse_header(lines, path) table_rows = [row for line in lines if (row := parse_table_row(line, path)) is not None] rows = table_rows if table_rows else parse_long_blocks(lines, path) draft_hash = sha256_file(path) rel_path = path.relative_to(ROOT).as_posix() for row in rows: row["decision_operator"] = header["decision_operator"] row["decision_source"] = header["decision_source"] row["manual_draft_path"] = rel_path row["manual_draft_sha256"] = draft_hash return rows def validate(decisions: pd.DataFrame, template: pd.DataFrame) -> None: required = [ "external_decision_id", "human_decision_action", "human_decision_reason_cn", "decision_operator", "decision_time", "decision_source", "accept_code_suggestion_flag", "manual_draft_path", "manual_draft_sha256", ] for col in required: if col not in decisions.columns: raise RuntimeError(f"missing parsed column: {col}") if decisions[col].isna().any() or decisions[col].astype(str).str.strip().eq("").any(): raise RuntimeError(f"blank parsed column: {col}") if len(decisions) != len(template): raise RuntimeError(f"manual decision row count mismatch: {len(decisions)} != {len(template)}") if decisions["external_decision_id"].duplicated().any(): raise RuntimeError("duplicate external_decision_id") if set(decisions["external_decision_id"]) != set(template["external_decision_id"]): raise RuntimeError("external_decision_id set does not match blank template") if not decisions["human_decision_action"].isin(VALID_ACTIONS).all(): bad = sorted(set(decisions["human_decision_action"]) - VALID_ACTIONS) raise RuntimeError(f"invalid action values: {bad}") if not decisions["accept_code_suggestion_flag"].isin({"true", "false"}).all(): raise RuntimeError("accept_code_suggestion_flag must be true/false") if not decisions["decision_source"].str.startswith(DECISION_SOURCE_PREFIX).all(): raise RuntimeError("unexpected decision_source prefix") def main() -> None: draft_paths = sorted(ROOT.glob("manual_buy_decision_external_draft_batch*.md")) if not draft_paths: raise RuntimeError("no manual draft batch files found") parsed: list[dict] = [] for path in draft_paths: parsed.extend(parse_draft(path)) decisions = pd.DataFrame(parsed) template = pd.read_csv(ROOT / "manual_buy_decision_external_template.csv", encoding="utf-8-sig") validate(decisions, template) merged = template.drop( columns=[ "human_decision_action", "human_decision_reason_cn", "decision_operator", "decision_time", "decision_source", "accept_code_suggestion_flag", "reviewer_notes", ], errors="ignore", ).merge(decisions, on="external_decision_id", how="left") if len(merged) != len(template): raise RuntimeError("merge changed row count") for col in ["human_decision_action", "human_decision_reason_cn", "decision_time", "decision_source"]: if merged[col].isna().any(): raise RuntimeError(f"missing merged manual field: {col}") source_path = ROOT / "manual_buy_decision_external_source_ledger.csv" merged.to_csv(source_path, index=False, encoding="utf-8-sig", quoting=csv.QUOTE_MINIMAL) summary = { "run_id": RUN_ID, "generated_at": datetime.now(TZ).isoformat(timespec="seconds"), "source": "manual draft batch files parsed without deriving decisions from candidate fields", "draft_files": [path.relative_to(ROOT).as_posix() for path in draft_paths], "manual_decisions": int(len(merged)), "buy": int(merged["human_decision_action"].eq("BUY").sum()), "review_held": int(merged["human_decision_action"].eq("REVIEW_HELD").sum()), "decision_source_count": int(merged["decision_source"].nunique()), } (ROOT / "manual_buy_decision_external_summary.json").write_text( json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8", ) (ROOT / "manual_buy_decision_external_draft.md").write_text( "# Manual BUY Decision External Draft Index\n\n" f"- run_id: {RUN_ID}\n" f"- generated_at: {summary['generated_at']}\n" "- source: batch draft files written before this conversion step\n" f"- rows: {summary['manual_decisions']}\n" f"- BUY: {summary['buy']}\n" f"- REVIEW_HELD: {summary['review_held']}\n\n" "## Batch files\n" + "\n".join(f"- {path.relative_to(ROOT).as_posix()} sha256={sha256_file(path)}" for path in draft_paths) + "\n", encoding="utf-8", ) if __name__ == "__main__": main()