"""Fetch DMS lifecycle data according to metadata templates.""" from __future__ import annotations import argparse import sys from pathlib import Path from step1_data_aggregation.data_analysis import analyze_production_data_from_env from step1_data_aggregation.dms_pipeline import build_production_data_from_env ROOT = Path(__file__).resolve().parents[1] def main() -> None: if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") parser = argparse.ArgumentParser( description="\u6309 metadata \u6a21\u677f\u4ece DMS \u8349\u7a3f\u7bb1\u751f\u6210\u751f\u4ea7\u6570\u636e Excel" ) parser.add_argument( "--templates", type=Path, help="metadata template directory (overrides METADATA_TEMPLATE_DIR)", ) parser.add_argument( "--output", type=Path, default=ROOT / "data" / "production", help="production workbook output directory", ) parser.add_argument( "--page-size", type=int, default=200, help="DMS page size (default: 200)", ) parser.add_argument( "--skip-analysis", action="store_true", help="only export DMS data; skip LLM field analysis and quality assessment", ) args = parser.parse_args() results = build_production_data_from_env( dotenv_path=ROOT / ".env", template_dir=args.templates, output_dir=args.output, page_size=args.page_size, ) print("\u5df2\u751f\u6210 DMS \u72b6\u6001 0/1/2/3 \u6570\u636e\uff1a") for result in results: print( f"- {result.template}: {result.content_rows} \u884c, " f"{result.fields} \u5b57\u6bb5\uff0c\u5206\u72b6\u6001 {result.state_rows} -> {result.output_file}" ) if not args.skip_analysis: analyses = analyze_production_data_from_env( dotenv_path=ROOT / ".env", template_dir=args.templates, production_dir=args.output, analysis_dir=args.output / "analysis", ) print("已生成字段分析、质量评估与 QA 上下文:") for analysis in analyses: print( f"- {analysis.template}: 质量分 {analysis.quality_score:g} " f"-> {analysis.report_file}" ) if __name__ == "__main__": main()