Agent/examples/auto_put_ad_mini/import_material_strategy_learning.py

"""将本地素材策略学习分析导入数据库。

使用默认 run_id 时本脚本幂等；只写内部学习表，不产生腾讯或飞书副作用。
"""

from __future__ import annotations

import argparse
import logging
import sys
from datetime import date
from pathlib import Path

from dotenv import load_dotenv

_HERE = Path(__file__).parent
load_dotenv(_HERE / ".env")
sys.path.insert(0, str(_HERE))

from tools.material_strategy_learning import (  # noqa: E402
    ensure_strategy_learning_tables,
    import_current_material_analysis,
)


DEFAULT_RUN_ID = "material_30d_20260607_20260706_top5000"
DEFAULT_ANALYSIS_DIR = _HERE / "outputs/data/material_analysis_20260708"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="导入高消耗素材策略学习结果")
    parser.add_argument("--run-id", default=DEFAULT_RUN_ID)
    parser.add_argument("--window-start", default="2026-06-07")
    parser.add_argument("--window-end", default="2026-07-06")
    parser.add_argument(
        "--performance-csv",
        type=Path,
        default=_HERE / "outputs/data/high_consumption_materials_30d_20260707_223644.csv",
    )
    parser.add_argument(
        "--performance-summary-json",
        type=Path,
        default=_HERE / "outputs/data/high_consumption_materials_30d_20260707_223644_summary.json",
    )
    parser.add_argument(
        "--visual-annotations-csv",
        type=Path,
        default=DEFAULT_ANALYSIS_DIR / "top100_visual_annotations.csv",
    )
    parser.add_argument(
        "--visual-summary-json",
        type=Path,
        default=DEFAULT_ANALYSIS_DIR / "top100_visual_annotation_summary.json",
    )
    parser.add_argument(
        "--report-path",
        type=Path,
        default=DEFAULT_ANALYSIS_DIR / "top100_visual_annotation_report.md",
    )
    parser.add_argument(
        "--sql-file",
        default="examples/auto_put_ad_mini/sql/high_consumption_materials_30d.sql",
    )
    parser.add_argument("--top-n", type=int, default=5000)
    parser.add_argument("--annotation-version", default="top100_rule_v1")
    parser.add_argument("--dry-run", action="store_true", help="只检查文件和建表,不导入数据")
    return parser.parse_args()


def _require_file(path: Path) -> None:
    if not path.exists():
        raise FileNotFoundError(f"文件不存在:{path}")


def main() -> int:
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    args = parse_args()

    for path in (
        args.performance_csv,
        args.performance_summary_json,
        args.visual_annotations_csv,
        args.visual_summary_json,
        args.report_path,
    ):
        _require_file(path)

    ensure_strategy_learning_tables()
    if args.dry_run:
        print("OK: 策略学习表已确认存在,输入文件检查通过")
        return 0

    result = import_current_material_analysis(
        run_id=args.run_id,
        window_start=date.fromisoformat(args.window_start),
        window_end=date.fromisoformat(args.window_end),
        performance_csv=args.performance_csv,
        performance_summary_json=args.performance_summary_json,
        visual_annotations_csv=args.visual_annotations_csv,
        visual_summary_json=args.visual_summary_json,
        report_path=args.report_path,
        sql_file=args.sql_file,
        top_n=args.top_n,
        annotation_version=args.annotation_version,
    )
    print(
        "OK: 导入完成 "
        f"run_id={result.run_id} "
        f"snapshot_rows={result.snapshot_rows} "
        f"annotation_rows={result.annotation_rows} "
        f"report_rows={result.report_rows}"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())