wuling-overseas-dealer-expansion-skill/scripts/workflows/run_dealer_pipeline.py

from __future__ import annotations

import argparse
import json
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Sequence


SCRIPT_DIR = Path(__file__).resolve().parent
SKILL_ROOT = SCRIPT_DIR.parents[1]
COMMON_DIR = SKILL_ROOT / "scripts" / "common"
if str(COMMON_DIR) not in sys.path:
    sys.path.insert(0, str(COMMON_DIR))

from artifact_manager import new_run_id, resolve_artifact_path, run_dir, write_json  # type: ignore  # noqa: E402
from workbook_resolver import resolve_workbook_path  # type: ignore  # noqa: E402


SUMMARY_SHEET = "\u5ba2\u6237\u4fe1\u606f\u6c47\u603b\u8868"
DEFAULT_CONFIG_NAME = "workflow_config.json"
FEISHU_CONFIG_NAME = "feishu_sync_config.json"


DEFAULT_CONFIG: Dict[str, Any] = {
    "enabled": True,
    "country": "Morocco",
    "profile_id": "",
    "excel": "",
    "summary_sheet_name": SUMMARY_SHEET,
    "daily_outreach_target": 20,
    "default_stages": ["summary", "dashboard", "feishu_check"],
    "refresh_summary": True,
    "refresh_dashboard": True,
    "dashboard_latest_dir": "dashboards/latest",
    "sync_feishu": True,
    "prepare_email_preview": False,
    "prepare_facebook_outreach": False,
    "collect_facebook_replies": False,
    "send_email": False,
    "send_facebook_dm": False,
    "human_confirmation_required": [
        "send_email",
        "send_facebook_dm",
        "linkedin_outreach",
        "write_facebook_conversations",
    ],
}


def read_json(path: Path) -> Dict[str, Any]:
    with path.open("r", encoding="utf-8") as fh:
        data = json.load(fh)
    if not isinstance(data, dict):
        raise ValueError(f"Config must be a JSON object: {path}")
    return data


def load_config(raw_path: str) -> Dict[str, Any]:
    config = dict(DEFAULT_CONFIG)
    root_config = Path.cwd() / DEFAULT_CONFIG_NAME
    if raw_path:
        path = Path(raw_path).expanduser()
        if not path.is_absolute():
            path = (Path.cwd() / path).resolve()
    elif root_config.exists():
        path = root_config.resolve()
    else:
        example = SKILL_ROOT / "assets" / "workflow_config.example.json"
        path = example.resolve() if example.exists() else root_config.resolve()

    loaded = read_json(path) if path.exists() else {}
    config.update(loaded)
    config["_config_path"] = str(path) if path.exists() else ""
    config["_config_found"] = path.exists()
    return config


def parse_stages(raw: str, config: Dict[str, Any]) -> List[str]:
    if raw:
        return [item.strip() for item in raw.split(",") if item.strip()]
    stages = config.get("default_stages") or DEFAULT_CONFIG["default_stages"]
    return [str(item).strip() for item in stages if str(item).strip()]


def run_command(args: List[str], step_name: str, output_dir: Path) -> Dict[str, Any]:
    started = datetime.now().isoformat(timespec="seconds")
    proc = subprocess.run(
        args,
        cwd=str(SKILL_ROOT),
        text=True,
        encoding="utf-8",
        errors="replace",
        capture_output=True,
    )
    log_path = output_dir / f"{step_name}.log"
    log_path.write_text(
        "COMMAND: " + " ".join(args) + "\n\nSTDOUT:\n" + proc.stdout + "\n\nSTDERR:\n" + proc.stderr,
        encoding="utf-8",
    )
    parsed: Optional[Any] = None
    stdout = proc.stdout.strip()
    if stdout:
        try:
            parsed = json.loads(stdout)
        except Exception:
            parsed = None
    return {
        "step": step_name,
        "command": args,
        "returncode": proc.returncode,
        "started_at": started,
        "finished_at": datetime.now().isoformat(timespec="seconds"),
        "log_path": str(log_path),
        "json": parsed,
        "ok": proc.returncode == 0,
    }


def feishu_config_status(config: Dict[str, Any]) -> Dict[str, Any]:
    project_cfg = Path.cwd() / FEISHU_CONFIG_NAME
    if not project_cfg.exists():
        return {
            "configured": False,
            "enabled": False,
            "required": bool(config.get("sync_feishu", True)),
            "message": "Local workbook updated; Feishu sync is waiting for feishu_sync_config.json.",
        }
    try:
        data = read_json(project_cfg)
    except Exception as exc:
        return {
            "configured": True,
            "enabled": False,
            "required": True,
            "error": str(exc),
            "message": "Feishu config exists but cannot be parsed.",
        }
    enabled = bool(data.get("enabled"))
    return {
        "configured": True,
        "enabled": enabled,
        "required": enabled,
        "config_path": str(project_cfg.resolve()),
        "summary_sheet_name": data.get("summary_sheet_name", SUMMARY_SHEET),
        "spreadsheet_url_present": bool(data.get("spreadsheet_url") or data.get("spreadsheet_token")),
        "message": "Agent must call the WorkBuddy lark-sheets plugin after local writeback." if enabled else "Feishu sync disabled by config.",
    }


def build_plan(config: Dict[str, Any], stages: List[str], workbook_info: Dict[str, Any]) -> Dict[str, Any]:
    blocked_external: List[str] = []
    for key in ("send_email", "send_facebook_dm"):
        if config.get(key):
            blocked_external.append(key)
    return {
        "country": config.get("country"),
        "profile_id": config.get("profile_id"),
        "daily_outreach_target": config.get("daily_outreach_target"),
        "stages": stages,
        "workbook": {key: str(value) if isinstance(value, Path) else value for key, value in workbook_info.items()},
        "blocked_external_actions": blocked_external,
        "human_confirmation_required": config.get("human_confirmation_required", []),
        "notes": [
            "This workflow orchestrates safe local post-processing by default.",
            "Email sending and Facebook/LinkedIn outbound actions still require preview and explicit user confirmation.",
            "Any browser stage must use AdsPower + Playwright and must not close the browser.",
        ],
    }


def main(argv: Optional[Sequence[str]] = None) -> int:
    parser = argparse.ArgumentParser(description="Run the Wuling dealer-expansion workflow orchestrator.")
    parser.add_argument("--config", default="", help="Project workflow config JSON. Default: ./workflow_config.json or skill example.")
    parser.add_argument("--excel", default="", help="Workbook path. Overrides config.excel.")
    parser.add_argument("--country", default="", help="Target country. Overrides config.country.")
    parser.add_argument("--profile-id", default="", help="AdsPower profile ID for browser stages. Overrides config.profile_id.")
    parser.add_argument("--stages", default="", help="Comma-separated stages. Default from config: summary,dashboard,feishu_check.")
    parser.add_argument("--run-id", default="", help="Run ID for artifacts.")
    parser.add_argument("--review-only", action="store_true", help="Only emit plan; do not write workbook or generate dashboard.")
    parser.add_argument("--write-workbook", action="store_true", help="Allow local workbook write stages such as summary rebuild.")
    parser.add_argument("--output", default="", help="Optional JSON report path.")
    args = parser.parse_args(argv)

    config = load_config(args.config)
    if args.excel:
        config["excel"] = args.excel
    if args.country:
        config["country"] = args.country
    if args.profile_id:
        config["profile_id"] = args.profile_id

    run_id = args.run_id or new_run_id("dealer_pipeline")
    out_dir = run_dir(run_id)
    out_dir.mkdir(parents=True, exist_ok=True)
    stages = parse_stages(args.stages, config)

    workbook_info = resolve_workbook_path(
        str(config.get("excel") or ""),
        create_from_template=bool(args.write_workbook),
        template_root=SKILL_ROOT,
    )
    plan = build_plan(config, stages, workbook_info)
    write_json(out_dir / "workflow-config-snapshot.json", config)
    write_json(out_dir / "workflow-plan.json", plan)

    report: Dict[str, Any] = {
        "run_id": run_id,
        "run_dir": str(out_dir),
        "config_found": config.get("_config_found", False),
        "config_path": config.get("_config_path", ""),
        "plan": plan,
        "steps": [],
        "feishu": {},
        "review_only": bool(args.review_only),
    }

    if workbook_info.get("source") == "missing":
        report["ok"] = False
        report["error"] = "No local workbook found. Use --excel, add a project workbook, or rerun with --write-workbook to copy the skill blank template."
    elif args.review_only:
        report["ok"] = True
    else:
        workbook_path = str(workbook_info["path"])
        if "summary" in stages:
            summary_report_path = resolve_artifact_path(
                "",
                kind="workflow_summary",
                default_name="summary-report.json",
                run_id=run_id,
            )
            cmd = [
                sys.executable,
                str(SKILL_ROOT / "scripts" / "common" / "build_customer_summary.py"),
                "--excel",
                workbook_path,
                "--output",
                str(summary_report_path),
                "--run-id",
                run_id,
            ]
            if args.write_workbook or bool(config.get("refresh_summary", True)):
                cmd.append("--write-summary")
            else:
                cmd.append("--dry-run")
            report["steps"].append(run_command(cmd, "summary", out_dir))

        if "dashboard" in stages:
            dashboard_html = resolve_artifact_path(
                "",
                kind="workflow_dashboard",
                default_name="customer_dashboard.html",
                run_id=run_id,
            )
            dashboard_json = resolve_artifact_path(
                "",
                kind="workflow_dashboard",
                default_name="customer_dashboard_data.json",
                run_id=run_id,
            )
            cmd = [
                sys.executable,
                str(SKILL_ROOT / "scripts" / "dashboard" / "build_dashboard.py"),
                "--excel",
                workbook_path,
                "--output-html",
                str(dashboard_html),
                "--output-json",
                str(dashboard_json),
                "--run-id",
                run_id,
                "--latest-dir",
                str(config.get("dashboard_latest_dir") or "dashboards/latest"),
            ]
            report["steps"].append(run_command(cmd, "dashboard", out_dir))

        if "facebook_outreach_preview" in stages:
            if not config.get("profile_id"):
                report["steps"].append({
                    "step": "facebook_outreach_preview",
                    "ok": False,
                    "skipped": True,
                    "reason": "profile_id is required for Facebook review planning.",
                })
            else:
                cmd = [
                    sys.executable,
                    str(SKILL_ROOT / "scripts" / "social" / "run_facebook_follow_dm.py"),
                    "--excel",
                    workbook_path,
                    "--profile-id",
                    str(config.get("profile_id")),
                    "--review-only",
                    "--run-id",
                    run_id,
                ]
                report["steps"].append(run_command(cmd, "facebook_outreach_preview", out_dir))

        if "feishu_check" in stages:
            report["feishu"] = feishu_config_status(config)

        report["ok"] = all(step.get("ok") or step.get("skipped") for step in report["steps"])

    output_path = resolve_artifact_path(
        args.output,
        kind="workflow_report",
        default_name="dealer-pipeline-report.json",
        run_id=run_id,
    )
    write_json(output_path, report)
    report["report_path"] = str(output_path)
    print(json.dumps(report, ensure_ascii=False, indent=2))
    return 0 if report.get("ok") else 1


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