kg-agent/scripts/fetch_dms_production.py

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