import argparse
|
import csv
|
from collections import Counter, defaultdict
|
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 norm_path(value):
|
return (value or "").replace("\\", "/").strip()
|
|
|
def rel_to_project(path, project_root):
|
p = Path(path)
|
if p.is_absolute():
|
try:
|
return p.relative_to(project_root).as_posix()
|
except ValueError:
|
return p.as_posix()
|
return p.as_posix()
|
|
|
def main():
|
parser = argparse.ArgumentParser()
|
parser.add_argument("--project-root", default=".")
|
parser.add_argument("--status", required=True)
|
parser.add_argument("--source-document", required=True)
|
parser.add_argument("--manifest-dir", required=True)
|
parser.add_argument("--outputs-dir", required=True)
|
args = parser.parse_args()
|
|
project_root = Path(args.project_root).resolve()
|
status_path = project_root / args.status
|
source_path = project_root / args.source_document
|
manifest_dir = project_root / args.manifest_dir
|
outputs_dir = project_root / args.outputs_dir
|
|
status_rows = read_csv(status_path)
|
source_rows = read_csv(source_path)
|
|
source_by_path = {norm_path(row.get("raw_file_path")): row for row in source_rows}
|
source_by_hash = {row.get("file_sha256", ""): row for row in source_rows if row.get("file_sha256")}
|
|
enriched = []
|
for row in status_rows:
|
raw_path = norm_path(row.get("raw_file_path"))
|
source = source_by_path.get(raw_path) or source_by_hash.get(row.get("raw_sha256", ""))
|
enriched_row = dict(row)
|
if source:
|
enriched_row["source_doc_id"] = source.get("doc_id", "")
|
enriched_row["source_title"] = source.get("title", "")
|
enriched_row["source_processing_status"] = source.get("processing_status", "")
|
else:
|
enriched_row["source_doc_id"] = row.get("source_doc_id", "")
|
enriched_row["source_title"] = ""
|
enriched_row["source_processing_status"] = ""
|
enriched.append(enriched_row)
|
|
enriched_fields = list(status_rows[0].keys()) if status_rows else []
|
for extra in ["source_title", "source_processing_status"]:
|
if extra not in enriched_fields:
|
enriched_fields.append(extra)
|
enriched_path = manifest_dir / "conversion_status_batch003_enriched.csv"
|
write_csv(enriched_path, enriched_fields, enriched)
|
|
by_sub = defaultdict(list)
|
for row in enriched:
|
by_sub[row.get("sub_batch_id", "")].append(row)
|
|
summary_rows = []
|
for sub_batch_id in sorted(by_sub):
|
rows = by_sub[sub_batch_id]
|
status_counts = Counter(row.get("conversion_status", "") for row in rows)
|
detected_counts = Counter(row.get("detected_type", "") for row in rows)
|
indexes = []
|
for row in rows:
|
conv_id = row.get("conversion_id", "")
|
try:
|
indexes.append(int(conv_id.rsplit("-", 1)[-1]))
|
except ValueError:
|
pass
|
summary_rows.append(
|
{
|
"case_id": rows[0].get("case_id", ""),
|
"batch_id": rows[0].get("batch_id", ""),
|
"sub_batch_id": sub_batch_id,
|
"run_id": rows[0].get("run_id", ""),
|
"input_count": len(rows),
|
"raw_inventory_index_min": min(indexes) if indexes else "",
|
"raw_inventory_index_max": max(indexes) if indexes else "",
|
"text_converted_count": status_counts.get("TEXT_CONVERTED", 0),
|
"failed_count": status_counts.get("FAILED", 0),
|
"zip_or_office_count": detected_counts.get("ZIP_OR_OFFICE", 0),
|
"pdf_count": detected_counts.get("PDF", 0),
|
"review_status": "DRAFT_FOR_REVIEW",
|
}
|
)
|
|
sub_path = manifest_dir / f"conversion_status_batch003_{sub_batch_id}.csv"
|
write_csv(sub_path, enriched_fields, rows)
|
|
summary_fields = [
|
"case_id",
|
"batch_id",
|
"sub_batch_id",
|
"run_id",
|
"input_count",
|
"raw_inventory_index_min",
|
"raw_inventory_index_max",
|
"text_converted_count",
|
"failed_count",
|
"zip_or_office_count",
|
"pdf_count",
|
"review_status",
|
]
|
sub_manifest_path = manifest_dir / "sub_batch_manifest_batch003.csv"
|
write_csv(sub_manifest_path, summary_fields, summary_rows)
|
|
failed_rows = [row for row in enriched if row.get("conversion_status") != "TEXT_CONVERTED"]
|
gap_fields = [
|
"case_id",
|
"batch_id",
|
"sub_batch_id",
|
"run_id",
|
"source_doc_id",
|
"raw_file_path",
|
"raw_sha256",
|
"detected_type",
|
"conversion_status",
|
"error_message",
|
"review_status",
|
]
|
gap_path = manifest_dir / "conversion_gap_batch003.csv"
|
write_csv(gap_path, gap_fields, [{field: row.get(field, "") for field in gap_fields} for row in failed_rows])
|
|
total_counts = Counter(row.get("conversion_status", "") for row in enriched)
|
type_counts = Counter(row.get("detected_type", "") for row in enriched)
|
md_lines = [
|
"# BATCH-003 转换摘要",
|
"",
|
"状态：DRAFT_FOR_REVIEW",
|
"",
|
"## 总览",
|
"",
|
f"- 输入记录：{len(enriched)}",
|
f"- TEXT_CONVERTED：{total_counts.get('TEXT_CONVERTED', 0)}",
|
f"- FAILED：{total_counts.get('FAILED', 0)}",
|
f"- PDF：{type_counts.get('PDF', 0)}",
|
f"- ZIP_OR_OFFICE：{type_counts.get('ZIP_OR_OFFICE', 0)}",
|
"",
|
"## sub_batch 汇总",
|
"",
|
"| sub_batch_id | 输入 | 成功 | 失败 | PDF | ZIP_OR_OFFICE | raw_inventory 行号范围 |",
|
"|---|---:|---:|---:|---:|---:|---|",
|
]
|
for row in summary_rows:
|
md_lines.append(
|
f"| {row['sub_batch_id']} | {row['input_count']} | {row['text_converted_count']} | "
|
f"{row['failed_count']} | {row['pdf_count']} | {row['zip_or_office_count']} | "
|
f"{row['raw_inventory_index_min']}-{row['raw_inventory_index_max']} |"
|
)
|
md_lines.extend(
|
[
|
"",
|
"说明：raw_inventory 行号范围来自 `conversion_id`，中间可能跳号；跳号通常是因为原始盘点中已有登记文件被本轮未登记转换流程跳过。",
|
"",
|
"## 输出文件",
|
"",
|
f"- `{rel_to_project(enriched_path, project_root)}`",
|
f"- `{rel_to_project(sub_manifest_path, project_root)}`",
|
f"- `{rel_to_project(gap_path, project_root)}`",
|
"- `ana-data/cases/有色案例/manifest/conversion_status_batch003_SB*.csv`",
|
"",
|
"## 限制",
|
"",
|
"本摘要只证明 raw 到文本的转换状态，不构成正式行业结论。失败文件进入转换缺口清单，后续需要重试、替代解析或人工复核。",
|
]
|
)
|
summary_path = outputs_dir / "batch003_conversion_summary.md"
|
summary_path.parent.mkdir(parents=True, exist_ok=True)
|
summary_path.write_text("\n".join(md_lines) + "\n", encoding="utf-8")
|
|
print(f"enriched={rel_to_project(enriched_path, project_root)}")
|
print(f"sub_manifest={rel_to_project(sub_manifest_path, project_root)}")
|
print(f"gap={rel_to_project(gap_path, project_root)}")
|
print(f"summary={rel_to_project(summary_path, project_root)}")
|
|
|
if __name__ == "__main__":
|
main()
|
