qwen_agent/skills/auto-daily-summary/scripts/summary_cli.py

#!/usr/bin/env python3
import argparse
import json
import sys
from datetime import datetime, UTC

from summary_core import build_summary, validate_payload


ERROR_TEMPLATE = {
    "success": False,
    "code": "invalid_input",
    "message": "",
    "data": {},
    "meta": {},
    "errors": [],
}


def _now_iso() -> str:
    return datetime.now(UTC).isoformat()


def _emit_json(data: dict, pretty: bool, stream=None):
    print(json.dumps(data, ensure_ascii=False, indent=2 if pretty else None), file=stream or sys.stdout)


def _error_response(code: str, message: str, errors: list[str] | None = None) -> dict:
    return {
        **ERROR_TEMPLATE,
        "code": code,
        "message": message,
        "meta": {"generated_at": _now_iso()},
        "errors": errors or [],
    }


def _parse_bool(value: str | None) -> bool | None:
    if value is None:
        return None
    lowered = value.lower()
    if lowered in {"1", "true", "yes", "y"}:
        return True
    if lowered in {"0", "false", "no", "n"}:
        return False
    raise ValueError(f"invalid boolean value: {value}")


def _load_payload(raw: str) -> dict:
    return json.loads(raw)


def _apply_overrides(payload: dict, args: argparse.Namespace) -> dict:
    payload.setdefault("data", {})
    if args.lang:
        payload["language"] = args.lang
    if args.style:
        payload["data"]["style"] = args.style
    if args.length:
        payload["data"]["length"] = args.length
    if hasattr(args, "extract_actions"):
        extract_actions = _parse_bool(getattr(args, "extract_actions", None))
        if extract_actions is not None:
            payload["data"]["extract_actions"] = extract_actions
    if hasattr(args, "extract_risks"):
        extract_risks = _parse_bool(getattr(args, "extract_risks", None))
        if extract_risks is not None:
            payload["data"]["extract_risks"] = extract_risks
    return payload


def cmd_validate(args: argparse.Namespace):
    payload = _apply_overrides(_load_payload(args.input_json), args)
    errors = validate_payload(payload)
    result = {
        "success": not errors,
        "code": "ok" if not errors else "invalid_input",
        "message": "payload valid" if not errors else "payload invalid",
        "data": {"valid": not errors},
        "meta": {"generated_at": _now_iso()},
        "errors": errors,
    }
    target_stream = sys.stdout if not errors else sys.stderr
    _emit_json(result, args.pretty, target_stream)
    if errors:
        raise SystemExit(1)


def cmd_run(args: argparse.Namespace):
    payload = _apply_overrides(_load_payload(args.input_json), args)
    errors = validate_payload(payload)
    if errors:
        _emit_json(_error_response("invalid_input", "payload invalid", errors), args.pretty, sys.stderr)
        raise SystemExit(1)
    data = build_summary(payload)
    if args.output == "markdown":
        print(data["markdown"])
        return
    result = {
        "success": True,
        "code": "ok",
        "message": "summary generated",
        "data": data,
        "meta": {
            "generated_at": _now_iso(),
            "source_count": len(payload.get("data", {}).get("sources", [])),
        },
        "errors": [],
    }
    _emit_json(result, args.pretty)


def cmd_plan_recurring(args: argparse.Namespace):
    payload = _apply_overrides(_load_payload(args.input_json), args)
    errors = validate_payload(payload)
    if errors:
        _emit_json(_error_response("invalid_input", "payload invalid", errors), args.pretty, sys.stderr)
        raise SystemExit(1)
    data = build_summary(payload)
    result = {
        "success": True,
        "code": "ok",
        "message": "recurring plan generated",
        "data": {"schedule_payload": data["schedule_payload"]},
        "meta": {"generated_at": _now_iso()},
        "errors": [],
    }
    _emit_json(result, args.pretty)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Generate structured summaries")
    subparsers = parser.add_subparsers(dest="command", required=True)
    for name in ["validate", "run", "plan-recurring"]:
        sub = subparsers.add_parser(name)
        sub.add_argument("--input-json", required=True)
        sub.add_argument("--lang")
        sub.add_argument("--style")
        sub.add_argument("--length")
        sub.add_argument("--extract-actions")
        sub.add_argument("--extract-risks")
        sub.add_argument("--pretty", action="store_true")
        if name == "run":
            sub.add_argument("--output", choices=["json", "markdown"], default="json")
    subparsers.choices["validate"].set_defaults(func=cmd_validate)
    subparsers.choices["run"].set_defaults(func=cmd_run)
    subparsers.choices["plan-recurring"].set_defaults(func=cmd_plan_recurring)
    return parser


if __name__ == "__main__":
    parser = build_parser()
    args = parser.parse_args()
    try:
        args.func(args)
    except json.JSONDecodeError as exc:
        _emit_json(_error_response("invalid_input", f"invalid json: {exc}", [str(exc)]), getattr(args, "pretty", False), sys.stderr)
        raise SystemExit(1)
    except ValueError as exc:
        _emit_json(_error_response("invalid_input", str(exc), [str(exc)]), getattr(args, "pretty", False), sys.stderr)
        raise SystemExit(1)
    except Exception as exc:
        _emit_json(_error_response("internal_error", "unexpected error", [str(exc)]), getattr(args, "pretty", False), sys.stderr)
        raise SystemExit(1)