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()
|
