import csv
|
import hashlib
|
from collections import Counter
|
from datetime import datetime, timezone, timedelta
|
from pathlib import Path
|
|
import fitz
|
|
|
TZ = timezone(timedelta(hours=8))
|
NOW = datetime.now(TZ).strftime("%Y-%m-%dT%H:%M:%S+08:00")
|
CASE_ID = "ANA-YS-INDUSTRY-001"
|
RUN_ID = "RUN-ANA-YS-PDF-TABLE-RECOVERY-019"
|
|
|
def read_csv(path):
|
with path.open("r", encoding="utf-8-sig", newline="") as f:
|
return list(csv.DictReader(f))
|
|
|
def write_csv(path, rows):
|
path.parent.mkdir(parents=True, exist_ok=True)
|
with path.open("w", encoding="utf-8-sig", newline="") as f:
|
writer = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
|
writer.writeheader()
|
writer.writerows(rows)
|
|
|
def main():
|
queue_path = next(Path("ana-data/cases").glob("*/evidence/key_fact_pdf_table_review_queue_pass004.csv"))
|
case_root = queue_path.parents[1]
|
industry_name = queue_path.parts[2]
|
case_dir = case_root / CASE_ID
|
outputs_dir = case_dir / "outputs"
|
evidence_dir = case_root / "evidence"
|
case_evidence_dir = case_dir / "evidence"
|
img_dir = Path("ana-data/img") / industry_name / CASE_ID / RUN_ID
|
for path in (outputs_dir, evidence_dir, case_evidence_dir, img_dir):
|
path.mkdir(parents=True, exist_ok=True)
|
|
queue_rows = read_csv(queue_path)
|
recovery_rows = []
|
for row in queue_rows:
|
rec = dict(row)
|
rec["recovery_run_id"] = RUN_ID
|
rec["recovery_status"] = ""
|
rec["recovery_image_path"] = ""
|
rec["recovery_error"] = ""
|
rec["manual_review_required"] = "YES"
|
rec["formal_upgrade_status"] = "DRAFT_PAGE_IMAGE_READY_NEEDS_MANUAL_TABLE_REVIEW"
|
rec["updated_at"] = NOW
|
|
if row.get("pdf_header_ok") == "YES" and row.get("page_known") == "YES":
|
raw = Path(row["raw_file_path"])
|
try:
|
page_no = int(row["precheck_page_no"])
|
doc = fitz.open(str(raw))
|
idx = max(0, min(page_no - 1, len(doc) - 1))
|
page = doc.load_page(idx)
|
pix = page.get_pixmap(matrix=fitz.Matrix(1.8, 1.8), alpha=False)
|
out = img_dir / f"页图_{row['pdf_table_review_id']}_第{page_no}页.png"
|
pix.save(str(out))
|
rec["recovery_status"] = "PDF_PAGE_IMAGE_RENDERED"
|
rec["recovery_image_path"] = out.as_posix()
|
rec["rendered_page_index"] = str(idx)
|
rec["pdf_page_count"] = str(len(doc))
|
rec["image_sha256"] = hashlib.sha256(out.read_bytes()).hexdigest()
|
rec["image_bytes"] = str(out.stat().st_size)
|
except Exception as exc:
|
rec["recovery_status"] = "RENDER_FAILED"
|
rec["recovery_error"] = repr(exc)
|
rec["formal_upgrade_status"] = "HELD_BY_RENDER_ERROR"
|
elif row.get("pdf_header_ok") != "YES":
|
rec["recovery_status"] = "HELD_BY_NOT_PDF_HEADER"
|
rec["formal_upgrade_status"] = "HELD_BY_SOURCE_FILE_TYPE"
|
else:
|
rec["recovery_status"] = "HELD_BY_PAGE_UNKNOWN"
|
rec["formal_upgrade_status"] = "HELD_BY_PAGE_UNKNOWN"
|
recovery_rows.append(rec)
|
|
recovery_path = evidence_dir / "原页表格复核恢复表_pass019.csv"
|
write_csv(recovery_path, recovery_rows)
|
|
manifest_path = case_evidence_dir / "原页表格复核恢复manifest_pass019.csv"
|
manifest_rows = [{
|
"case_id": CASE_ID,
|
"batch_id": "BATCH-003",
|
"run_id": RUN_ID,
|
"artifact_type": "pdf_page_table_recovery",
|
"artifact_path": recovery_path.as_posix(),
|
"row_count": str(len(recovery_rows)),
|
"rendered_count": str(sum(1 for r in recovery_rows if r["recovery_status"] == "PDF_PAGE_IMAGE_RENDERED")),
|
"held_by_page_unknown": str(sum(1 for r in recovery_rows if r["recovery_status"] == "HELD_BY_PAGE_UNKNOWN")),
|
"held_by_not_pdf_header": str(sum(1 for r in recovery_rows if r["recovery_status"] == "HELD_BY_NOT_PDF_HEADER")),
|
"image_dir": img_dir.as_posix(),
|
"review_status": "DRAFT_FOR_REVIEW",
|
"created_at": NOW,
|
}]
|
write_csv(manifest_path, manifest_rows)
|
|
cross_path = case_root / "evidence/key_fact_external_crosscheck_pass006.csv"
|
cross_rows = read_csv(cross_path)
|
image_by_evidence_card = {r.get("evidence_card_id"): r for r in recovery_rows if r.get("evidence_card_id")}
|
price_rows = [r for r in cross_rows if r.get("evidence_kind") == "PRICE_OR_MARKET_DATA"]
|
status_counter = Counter(r.get("cross_check_status") for r in price_rows)
|
metal_counter = Counter(r.get("metal_tags") or "未标注" for r in price_rows)
|
|
lines = [
|
"# 原研报价格库存线索表",
|
"",
|
"状态：DRAFT_FOR_REVIEW",
|
f"生成时间：{NOW}",
|
f"关联案例：{CASE_ID}",
|
f"关联运行：{RUN_ID}",
|
"",
|
"## 使用边界",
|
"",
|
"1. 本文件优先整理已提供 PDF/converted text 中出现的价格、库存、供需、成本和市场变量线索，用于回应“价格库存是否已在研报中提供”。",
|
"2. 本文件不是正式价格数据库，也不是正式指标证据；正式输出前仍需核对报告日期、原始表格行、单位、口径和必要外部来源。",
|
"3. 已生成页图的记录只表示 PDF 原页可查看，不等于表头、单位、日期和数值一致性已经人工通过。",
|
"",
|
"## 汇总",
|
"",
|
f"- 价格/库存/市场数据候选记录：{len(price_rows)} 条。",
|
f"- 外部来源已有候选映射：{status_counter.get('PRICE_INVENTORY_SOURCE_MATCHED', 0)} 条。",
|
f"- 仍需价格库存外部来源或日期/表格行补核：{status_counter.get('NEEDS_PRICE_INVENTORY_EXTERNAL_SOURCE', 0)} 条。",
|
f"- 本轮 PDF 原页页图已恢复：{sum(1 for r in recovery_rows if r['recovery_status'] == 'PDF_PAGE_IMAGE_RENDERED')} 张。",
|
"",
|
"### 金属分布",
|
"",
|
"| 金属标签 | 线索条数 |",
|
"|---|---:|",
|
]
|
for metal, count in metal_counter.most_common():
|
lines.append(f"| {metal} | {count} |")
|
|
lines.extend([
|
"",
|
"### 证据状态分布",
|
"",
|
"| 状态 | 条数 |",
|
"|---|---:|",
|
])
|
for status, count in status_counter.most_common():
|
lines.append(f"| {status or '未标注'} | {count} |")
|
|
lines.extend([
|
"",
|
"## 明细",
|
"",
|
"| 序号 | doc_id | 金属 | 主题 | 页码/位置 | PDF页图状态 | 价格库存线索 | 数值/单位候选 | 日期候选 | 后续动作 |",
|
"|---:|---|---|---|---|---|---|---|---|---|",
|
])
|
for index, row in enumerate(price_rows, 1):
|
rec = image_by_evidence_card.get(row.get("evidence_card_id"), {})
|
img_status = rec.get("recovery_status") or "未进入PASS-004页图队列"
|
if rec.get("recovery_image_path"):
|
img_status = f"[已生成页图]({rec['recovery_image_path']})"
|
indicators = (row.get("matched_price_inventory_indicators") or "").replace("|", "<br>")
|
values = (row.get("value_unit_candidates") or "").replace("|", "<br>")
|
dates = (row.get("date_candidates") or "").replace("|", "<br>")
|
loc = f"第{row.get('precheck_page_no')}页 / {row.get('precheck_location')}"
|
action = row.get("next_action") or ""
|
lines.append(
|
f"| {index} | {row.get('doc_id')} | {row.get('metal_tags')} | {row.get('theme_tags')} | "
|
f"{loc} | {img_status} | {indicators} | {values} | {dates} | {action} |"
|
)
|
|
price_doc_path = outputs_dir / "原研报价格库存线索表.md"
|
price_doc_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
summary_path = outputs_dir / "原页表格复核恢复摘要.md"
|
summary_lines = [
|
"# 原页表格复核恢复摘要",
|
"",
|
"状态：DRAFT_FOR_REVIEW",
|
f"生成时间：{NOW}",
|
f"关联案例：{CASE_ID}",
|
f"关联运行：{RUN_ID}",
|
"",
|
"## 恢复原因",
|
"",
|
"用户要求重新尝试安装 PyMuPDF。分析员确认此前失败原因是本机代理环境变量指向 `127.0.0.1:1080` 但端口无服务；临时清除代理变量后，PyMuPDF 安装和 `fitz` 导入成功。该轮属于“环境恢复且可一次性解决阻断”的例外。",
|
"",
|
"## 执行结果",
|
"",
|
f"- PASS-004 输入表格候选：{len(recovery_rows)} 条。",
|
f"- PDF 头正常且页码已知，已生成页图：{sum(1 for r in recovery_rows if r['recovery_status'] == 'PDF_PAGE_IMAGE_RENDERED')} 条。",
|
f"- 页码未知仍保持 HELD：{sum(1 for r in recovery_rows if r['recovery_status'] == 'HELD_BY_PAGE_UNKNOWN')} 条。",
|
f"- 文件头不是 PDF 仍保持 HELD：{sum(1 for r in recovery_rows if r['recovery_status'] == 'HELD_BY_NOT_PDF_HEADER')} 条。",
|
f"- 页图目录：`{img_dir.as_posix()}`。",
|
"",
|
"## 输出入口",
|
"",
|
f"1. 恢复表：`{recovery_path.as_posix()}`",
|
f"2. manifest：`{manifest_path.as_posix()}`",
|
f"3. 价格库存线索表：`{price_doc_path.as_posix()}`",
|
"",
|
"## 仍保留的边界",
|
"",
|
"1. 页图生成不等于正式表格证据通过；表头、单位、日期、来源和数值一致性仍需人工核验。",
|
"2. 页码未知和非 PDF 头文件不在本轮扩大处理范围内。",
|
"3. 本轮不输出正式行业结论、正式公司结论、正式指标、交易指令或收益承诺。",
|
]
|
summary_path.write_text("\n".join(summary_lines) + "\n", encoding="utf-8")
|
|
print(f"recovery_path={recovery_path}")
|
print(f"manifest_path={manifest_path}")
|
print(f"summary_path={summary_path}")
|
print(f"price_doc_path={price_doc_path}")
|
print(f"image_dir={img_dir}")
|
print(f"rendered={sum(1 for r in recovery_rows if r['recovery_status'] == 'PDF_PAGE_IMAGE_RENDERED')}")
|
print(f"held_page_unknown={sum(1 for r in recovery_rows if r['recovery_status'] == 'HELD_BY_PAGE_UNKNOWN')}")
|
print(f"held_not_pdf={sum(1 for r in recovery_rows if r['recovery_status'] == 'HELD_BY_NOT_PDF_HEADER')}")
|
print(f"price_rows={len(price_rows)}")
|
|
|
if __name__ == "__main__":
|
main()
|
