#!/usr/bin/env python3
"""
Stable daily dataflow failed job analysis.

This wrapper keeps scheduling and notification deterministic while still using
OpenClaw for the root-cause analysis step.
"""

import datetime
from datetime import timedelta
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
from typing import Any, Dict, List, Tuple

import requests

# Add parent directory to path for imports
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from base import APP_ID, APP_SECRET, FAILED_JOBS_APP_TOKEN, FAILED_JOBS_TABLE_ID, get_feishu_token, send_feishu_card

REPORT_WIKI_SPACE = "7594383226739674063"
REPORT_WIKI_PARENT = "SiFYwRwZyiCvSYkMA5Qcl526nwg"
REPORT_WIKI_BASE_URL = "https://pingcap.feishu.cn/wiki"
AI_CONFIG_REPO = os.environ.get("AI_CONFIG_REPO", "/home/openclaw/github/data-service-infra-ai-configurations")
MD2FEISHU_SCRIPT = os.path.join(AI_CONFIG_REPO, "skills/team-common/md2feishu/scripts/md2feishu.py")
OPENCLAW_TIMEOUT_SECONDS = int(os.environ.get("OPENCLAW_ANALYSIS_TIMEOUT", "600"))
DEFAULT_OPENCLAW_CANDIDATES = (
    "/usr/local/bin/openclaw",
    "/usr/bin/openclaw",
    "/home/ubuntu/.npm-global/bin/openclaw",
    "/home/openclaw/.npm-global/bin/openclaw",
)
DEFAULT_NOTIFY_CHAT_ID = os.environ.get("ONCALL_NOTIFY_CHAT_ID") or os.environ.get("ONCALL_METRICS_CHAT_ID")
MAX_CARD_ITEMS = 5
SUPPORTED_COMPONENTS = ("Import", "DM", "Backup", "BR")


def is_true(value: str) -> bool:
    return str(value or "").strip().lower() in {"1", "true", "yes", "on"}


def parse_notify_targets(raw_value: str | None) -> List[str]:
    values = re.split(r"[\s,;]+", str(raw_value or "").strip())
    return [item for item in values if item]


def parse_target_date(date_str: str = None) -> datetime.date:
    if date_str:
        return datetime.datetime.strptime(date_str, "%Y%m%d").date()
    return datetime.date.today() - timedelta(days=1)


def normalize_text(value: Any, default: str = "N/A") -> str:
    if value is None:
        return default
    text = str(value).strip()
    return text or default


def shorten(text: str, limit: int = 220) -> str:
    text = re.sub(r"\s+", " ", normalize_text(text, ""))
    if len(text) <= limit:
        return text
    return text[: limit - 3].rstrip() + "..."


def resolve_openclaw_bin() -> str:
    configured = os.environ.get("OPENCLAW_BIN")
    if configured:
        return configured

    detected = shutil.which("openclaw")
    if detected:
        return detected

    for candidate in DEFAULT_OPENCLAW_CANDIDATES:
        if os.path.exists(candidate):
            return candidate

    return DEFAULT_OPENCLAW_CANDIDATES[0]


def sort_jobs(jobs: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    return sorted(
        jobs,
        key=lambda item: (
            normalize_text(item.get("component"), ""),
            normalize_text(item.get("job_id"), ""),
            normalize_text(item.get("cluster"), ""),
            normalize_text(item.get("error_type"), ""),
            normalize_text(item.get("message"), ""),
        ),
    )


def get_failed_jobs_from_bitable(date_str: str = None) -> Dict[str, List[Dict[str, Any]]]:
    """Get failed dataflow jobs from Bitable for a specific date."""
    target_date = parse_target_date(date_str)
    start_ts = int(datetime.datetime.combine(target_date, datetime.datetime.min.time()).timestamp() * 1000)
    end_ts = int(datetime.datetime.combine(target_date + timedelta(days=1), datetime.datetime.min.time()).timestamp() * 1000)

    token = get_feishu_token()
    if not token:
        raise RuntimeError("Failed to get Feishu token")

    grouped_jobs: Dict[str, List[Dict[str, Any]]] = {component: [] for component in SUPPORTED_COMPONENTS}
    page_token = None

    while True:
        url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{FAILED_JOBS_APP_TOKEN}/tables/{FAILED_JOBS_TABLE_ID}/records"
        headers = {"Authorization": f"Bearer {token}"}
        params = {
            "page_size": 100,
            "field_names": '["Job ID", "Component", "Failed Tasks", "Failed Reason", "Cluster", "Time"]',
        }
        if page_token:
            params["page_token"] = page_token

        response = requests.get(url, headers=headers, params=params, timeout=30)
        data = response.json()
        if data.get("code") != 0:
            raise RuntimeError(f"Error fetching Bitable: {data.get('msg')}")

        for item in data.get("data", {}).get("items", []):
            fields = item.get("fields", {})
            record_time = fields.get("Time", 0)
            if not (start_ts <= record_time < end_ts):
                continue

            component = normalize_text(fields.get("Component"), "Unknown")
            job = {
                "job_id": normalize_text(fields.get("Job ID")),
                "component": component,
                "cluster": normalize_text(fields.get("Cluster")),
                "error_type": normalize_text(fields.get("Failed Tasks"), "Unknown"),
                "message": normalize_text(fields.get("Failed Reason")),
                "record_id": item.get("record_id"),
            }
            if component in grouped_jobs:
                grouped_jobs[component].append(job)

        page_token = data.get("data", {}).get("page_token")
        if not page_token:
            break

    return {component.lower(): sort_jobs(jobs) for component, jobs in grouped_jobs.items()}


def build_default_analysis(all_jobs: List[Dict[str, Any]]) -> Dict[str, Any]:
    jobs = []
    for job in all_jobs:
        jobs.append(
            {
                "job_id": job["job_id"],
                "component": job["component"],
                "cluster": job["cluster"],
                "error_type": job["error_type"],
                "root_cause": "Insufficient evidence from automation context.",
                "evidence": shorten(job["message"]),
                "recommendation": "Open the report and inspect logs manually.",
                "confidence": "low",
            }
        )
    return {
        "summary": {
            "overview": "OpenClaw analysis was skipped.",
            "risk_level": "low" if not jobs else "medium",
        },
        "jobs": jobs,
    }


def extract_json_blob(text: str) -> str:
    start = text.find("{")
    end = text.rfind("}")
    if start == -1 or end == -1 or end < start:
        raise ValueError("No JSON object found")
    return text[start : end + 1]


def parse_openclaw_cli_output(stdout: str) -> Dict[str, Any]:
    cli_payload = json.loads(extract_json_blob(stdout))
    result_payload = cli_payload.get("result") if isinstance(cli_payload, dict) else None
    payload_root = result_payload if isinstance(result_payload, dict) else cli_payload
    payloads = payload_root.get("payloads") or []
    reply_text = "\n".join(item.get("text", "") for item in payloads if item.get("text"))
    if not reply_text:
        raise ValueError("OpenClaw returned no text payload")
    return json.loads(extract_json_blob(reply_text))


def run_openclaw_analysis(date_str: str, grouped_jobs: Dict[str, List[Dict[str, Any]]]) -> Tuple[Dict[str, Any], str]:
    all_jobs = sort_jobs([job for jobs in grouped_jobs.values() for job in jobs])
    if not all_jobs:
        return (
            {
                "summary": {
                    "overview": "No failed dataflow jobs were found for the target day.",
                    "risk_level": "low",
                },
                "jobs": [],
            },
            "not-needed",
        )

    if is_true(os.environ.get("SKIP_OPENCLAW_ANALYSIS")):
        return build_default_analysis(all_jobs), "skipped"

    input_payload = {
        "date": parse_target_date(date_str).strftime("%Y-%m-%d"),
        "jobs_by_component": grouped_jobs,
    }
    with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as handle:
        json.dump(input_payload, handle, ensure_ascii=False, indent=2)
        input_path = handle.name

    prompt = f"""
Read the JSON file at {input_path}.
Analyze only the failed Import, DM, Backup, and BR jobs in that file.
Do not send messages. Do not edit files. Do not use markdown.
Return exactly one JSON object with this schema:
{{
  "summary": {{
    "overview": "string",
    "risk_level": "low|medium|high"
  }},
  "jobs": [
    {{
      "job_id": "string",
      "component": "Import|DM|Backup|BR",
      "cluster": "string",
      "error_type": "string",
      "root_cause": "string",
      "evidence": "string",
      "recommendation": "string",
      "confidence": "low|medium|high"
    }}
  ]
}}
Rules:
- Keep the jobs array in the same order as the input.
- If evidence is insufficient, say so explicitly instead of guessing.
- Keep each string concise and operational.
- Base your reasoning on the provided job metadata and any logs or tools you can access.
""".strip()

    with tempfile.NamedTemporaryFile(mode="w", suffix=".prompt", delete=False) as prompt_handle:
        prompt_handle.write(prompt)
        prompt_path = prompt_handle.name

    with open(prompt_path, encoding="utf-8") as prompt_handle:
        prompt_text = prompt_handle.read()

    command = [
        resolve_openclaw_bin(),
        "agent",
        "--agent",
        "main",
        "--json",
        "--thinking",
        "minimal",
        "--timeout",
        str(OPENCLAW_TIMEOUT_SECONDS),
        "--message",
        prompt_text,
    ]

    try:
        result = subprocess.run(command, capture_output=True, text=True, timeout=OPENCLAW_TIMEOUT_SECONDS + 30, env=os.environ)
        if result.returncode != 0:
            raise RuntimeError(result.stderr.strip() or result.stdout.strip() or "openclaw agent failed")
        analysis = parse_openclaw_cli_output(result.stdout)
        if not isinstance(analysis.get("jobs"), list) or not isinstance(analysis.get("summary"), dict):
            raise ValueError("OpenClaw returned an invalid analysis schema")
        return analysis, "ok"
    except Exception as exc:
        fallback = build_default_analysis(all_jobs)
        fallback["summary"]["overview"] = f"OpenClaw analysis failed: {shorten(str(exc), 180)}"
        fallback["summary"]["risk_level"] = "medium"
        return fallback, "failed"
    finally:
        for path_to_remove in (input_path, prompt_path):
            try:
                os.unlink(path_to_remove)
            except OSError:
                pass


def format_analysis_section(analysis: Dict[str, Any]) -> List[str]:
    lines = ["## Analysis", ""]
    summary = analysis.get("summary", {})
    lines.append(f"- **Overview**: {normalize_text(summary.get('overview'))}")
    lines.append(f"- **Risk Level**: {normalize_text(summary.get('risk_level'), 'unknown')}")
    lines.append("")

    jobs = analysis.get("jobs") or []
    if not jobs:
        lines.append("No failed jobs require analysis.")
        lines.append("")
        return lines

    for index, job in enumerate(jobs, 1):
        lines.append(f"### {index}. Job {normalize_text(job.get('job_id'))}")
        lines.append(f"- **Component**: {normalize_text(job.get('component'))}")
        lines.append(f"- **Cluster**: {normalize_text(job.get('cluster'))}")
        lines.append(f"- **Error Type**: {normalize_text(job.get('error_type'), 'Unknown')}")
        lines.append(f"- **Root Cause**: {normalize_text(job.get('root_cause'))}")
        lines.append(f"- **Evidence**: {normalize_text(job.get('evidence'))}")
        lines.append(f"- **Recommendation**: {normalize_text(job.get('recommendation'))}")
        lines.append(f"- **Confidence**: {normalize_text(job.get('confidence'), 'unknown')}")
        lines.append("")
    return lines


def build_component_section(title: str, jobs: List[Dict[str, Any]], empty_label: str) -> List[str]:
    lines = [title, ""]
    if jobs:
        for index, job in enumerate(jobs, 1):
            lines.append(f"### {index}. Job {job['job_id']}")
            lines.append(f"- **Cluster**: {job['cluster']}")
            lines.append(f"- **Error Type**: {job['error_type']}")
            lines.append(f"- **Failed Reason**: {job['message']}")
            lines.append("")
    else:
        lines.append(empty_label)
        lines.append("")
    return lines


def get_backup_br_jobs(grouped_jobs: Dict[str, List[Dict[str, Any]]]) -> List[Dict[str, Any]]:
    return sort_jobs(grouped_jobs["backup"] + grouped_jobs["br"])


def generate_report_markdown(date_str: str, grouped_jobs: Dict[str, List[Dict[str, Any]]], analysis: Dict[str, Any]) -> str:
    display_date = parse_target_date(date_str).strftime("%Y-%m-%d")
    backup_br_jobs = get_backup_br_jobs(grouped_jobs)
    lines = [f"# Dataflow Failed Jobs Report - {display_date}", "", "## Summary", ""]
    lines.append(f"- **Import Failed**: {len(grouped_jobs['import'])}")
    lines.append(f"- **DM Failed**: {len(grouped_jobs['dm'])}")
    lines.append(f"- **Backup/BR Failed**: {len(backup_br_jobs)}")
    lines.append("")

    lines.extend(build_component_section("## Import Failed Jobs", grouped_jobs["import"], "No failed Import jobs"))
    lines.extend(build_component_section("## DM Failed Jobs", grouped_jobs["dm"], "No failed DM jobs"))
    lines.extend(build_component_section("## Backup/BR Failed Jobs", backup_br_jobs, "No failed Backup/BR jobs"))

    lines.extend(format_analysis_section(analysis))
    return "\n".join(lines)


def create_wiki_node(title: str) -> Tuple[str, str]:
    token = get_feishu_token()
    response = requests.post(
        f"https://open.feishu.cn/open-apis/wiki/v2/spaces/{REPORT_WIKI_SPACE}/nodes",
        json={
            "obj_type": "docx",
            "parent_node_token": REPORT_WIKI_PARENT,
            "node_type": "origin",
            "title": title,
        },
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        timeout=30,
    )
    result = response.json()
    if result.get("code") != 0:
        raise RuntimeError(f"Failed to create Wiki node: {result}")
    node = result["data"]["node"]
    return node["obj_token"], node["node_token"]


def upload_markdown_to_doc(doc_token: str, title: str, content: str) -> None:
    with tempfile.NamedTemporaryFile(mode="w", suffix=".md", delete=False) as handle:
        handle.write(content)
        temp_file = handle.name
    try:
        command = [
            "/usr/bin/python3",
            MD2FEISHU_SCRIPT,
            "--app-id",
            APP_ID,
            "--app-secret",
            APP_SECRET,
            "--file",
            temp_file,
            "--title",
            title,
            "--doc-id",
            doc_token,
        ]
        result = subprocess.run(command, capture_output=True, text=True, timeout=600, env=os.environ)
        if result.returncode != 0:
            raise RuntimeError(result.stderr.strip() or "md2feishu failed")
    finally:
        try:
            os.unlink(temp_file)
        except OSError:
            pass


def create_wiki_report(date_str: str, content: str) -> str:
    display_date = parse_target_date(date_str).strftime("%Y-%m-%d")
    title = f"Dataflow Failed Jobs Report - {display_date}"
    doc_token, wiki_token = create_wiki_node(title)
    upload_markdown_to_doc(doc_token, title, content)
    return f"{REPORT_WIKI_BASE_URL}/{wiki_token}"


def build_notification_card(date_str: str, grouped_jobs: Dict[str, List[Dict[str, Any]]], analysis: Dict[str, Any], wiki_url: str, analysis_status: str) -> Dict[str, Any]:
    display_date = parse_target_date(date_str).strftime("%Y-%m-%d")
    summary = analysis.get("summary", {})
    backup_br_jobs = get_backup_br_jobs(grouped_jobs)
    items = []
    for job in (analysis.get("jobs") or [])[:MAX_CARD_ITEMS]:
        items.append(
            "- **{job_id}** ({component})\\n  Root cause: {root_cause}\\n  Recommendation: {recommendation}".format(
                job_id=normalize_text(job.get("job_id")),
                component=normalize_text(job.get("component")),
                root_cause=shorten(normalize_text(job.get("root_cause")), 120),
                recommendation=shorten(normalize_text(job.get("recommendation")), 120),
            )
        )
    detail_block = "\n".join(items) if items else "No failed jobs found for the target day."

    return {
        "config": {"wide_screen_mode": True, "enable_forward": True},
        "header": {
            "template": "blue",
            "title": {"tag": "plain_text", "content": f"Dataflow Failed Job Analysis | {display_date}"},
        },
        "elements": [
            {
                "tag": "column_set",
                "flex_mode": "none",
                "columns": [
                    {
                        "tag": "column",
                        "width": "weighted",
                        "weight": 1,
                        "elements": [{"tag": "markdown", "content": f"**Import Failed**\n{len(grouped_jobs['import'])}"}],
                    },
                    {
                        "tag": "column",
                        "width": "weighted",
                        "weight": 1,
                        "elements": [{"tag": "markdown", "content": f"**DM Failed**\n{len(grouped_jobs['dm'])}"}],
                    },
                    {
                        "tag": "column",
                        "width": "weighted",
                        "weight": 1,
                        "elements": [{"tag": "markdown", "content": f"**Backup/BR Failed**\n{len(backup_br_jobs)}"}],
                    },
                    {
                        "tag": "column",
                        "width": "weighted",
                        "weight": 1,
                        "elements": [{"tag": "markdown", "content": f"**Risk Level**\n{normalize_text(summary.get('risk_level'), 'unknown')}"}],
                    },
                ],
            },
            {"tag": "hr"},
            {
                "tag": "markdown",
                "content": "**Analysis status**: {status}\n**Overview**: {overview}".format(
                    status=analysis_status,
                    overview=shorten(normalize_text(summary.get("overview")), 220),
                ),
            },
            {"tag": "markdown", "content": "**Key Findings**\n" + detail_block},
            {
                "tag": "action",
                "actions": [
                    {
                        "tag": "button",
                        "text": {"tag": "plain_text", "content": "Open Wiki Report"},
                        "type": "primary",
                        "url": wiki_url,
                    }
                ],
            },
        ],
    }


def send_notification(card: Dict[str, Any]) -> str:
    if is_true(os.environ.get("SKIP_SEND_NOTIFICATION")):
        return "skipped"
    targets = parse_notify_targets(DEFAULT_NOTIFY_CHAT_ID)
    if not targets:
        return "missing-chat-id"
    results = [send_feishu_card(target, card) for target in targets]
    return "sent" if all(results) else "partial-failed"


def main(date_arg: str = None) -> Dict[str, Any]:
    date_str = date_arg or parse_target_date().strftime("%Y%m%d")
    grouped_jobs = get_failed_jobs_from_bitable(date_str)

    analysis, analysis_status = run_openclaw_analysis(date_str, grouped_jobs)
    report_content = generate_report_markdown(date_str, grouped_jobs, analysis)
    wiki_url = create_wiki_report(date_str, report_content)
    card = build_notification_card(date_str, grouped_jobs, analysis, wiki_url, analysis_status)
    notification_status = send_notification(card)

    result = {
        "date": parse_target_date(date_str).strftime("%Y-%m-%d"),
        "import_failed": len(grouped_jobs["import"]),
        "dm_failed": len(grouped_jobs["dm"]),
        "backup_failed": len(grouped_jobs["backup"]),
        "br_failed": len(grouped_jobs["br"]),
        "analysis_status": analysis_status,
        "notification_status": notification_status,
        "wiki_url": wiki_url,
        "risk_level": normalize_text(analysis.get("summary", {}).get("risk_level"), "unknown"),
        "overview": normalize_text(analysis.get("summary", {}).get("overview")),
    }
    print(json.dumps(result, ensure_ascii=False, sort_keys=True))
    return result


if __name__ == "__main__":
    arg = sys.argv[1] if len(sys.argv) > 1 else None
    main(arg)