run.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path
from typing import Any, Dict

from src.project_ai.services import ProjectAIPipeline


def main() -> None:
    parser = argparse.ArgumentParser(description="ProjectAI 管控工具")
    subparsers = parser.add_subparsers(dest="command")

    run_parser = subparsers.add_parser("run", help="执行完整流程")
    run_parser.add_argument("--config", required=True, help="配置文件路径")

    args = parser.parse_args()
    if args.command != "run":
        parser.print_help()
        sys.exit(1)

    config_path = Path(args.config).expanduser()
    if not config_path.exists():
        print(f"配置文件不存在: {config_path}", file=sys.stderr)
        sys.exit(2)

    config = _load_config(config_path)

    pipeline = ProjectAIPipeline(config)
    result = pipeline.run()

    print(f"日报已生成: {result['report_path']}")
    print(f"风险数量: {result['risk_count']}")

    notifications = result.get("notifications", [])
    if notifications:
        print("\n待发送的提醒:")
        for message in notifications:
            print("----")
            print(message)
    else:
        print("\n暂无需要发送的高风险提醒。")


def _load_config(path: Path) -> Dict[str, Any]:
    config_text = path.read_text(encoding="utf-8")
    raw = json.loads(config_text)
    return _expand_env(raw)


def _expand_env(node: Any) -> Any:
    if isinstance(node, dict):
        return {key: _expand_env(value) for key, value in node.items()}
    if isinstance(node, list):
        return [_expand_env(item) for item in node]
    if isinstance(node, str):
        if node.startswith("${") and node.endswith("}"):
            key = node[2:-1]
            return os.environ.get(key, "")
        return node
    return node


if __name__ == "__main__":
    main()