builder.py

"""
builder.py — 核心构建引擎
URL 匹配 → 工序链执行 → 产物收集 → 通知
"""
import os
import re
import time
import uuid
import logging
import subprocess
from typing import Optional, Dict, Any

from notifier import send_notification
from ssh_client import run_ssh_command

log = logging.getLogger("builder")


class _SafeDict(dict):
    """str.format_map 用：未知占位符原样保留，避免 KeyError"""
    def __missing__(self, key):
        return "{" + key + "}"

# 内存任务存储（生产可换 SQLite/Redis）
_tasks: Dict[str, dict] = {}
_tasks_lock = None  # 延迟导入 threading.RLock

# 去重：正在运行的 pipeline 防并发覆盖
_running: Dict[str, str] = {}  # pipeline_id → task_id


# ════════════════════════════════════════════════════════
# URL 匹配
# ════════════════════════════════════════════════════════
def match_url_rule(url_rules: list, path: str) -> Optional[dict]:
    """匹配 URL 规则，支持 /* 通配"""
    for rule in url_rules:
        if not rule.get("enabled", True):
            continue
        pattern = rule.get("path_pattern", "")
        if not pattern:
            continue
        # 转正则：/* → /.*,  /*/ → /.*/
        regex = re.escape(pattern).replace(r"\*", ".*")
        if re.fullmatch(regex, path):
            return rule
    return None


def match_and_run(config: dict, path: str, event: dict) -> Optional[dict]:
    """根据 path 匹配规则 → 找到 pipeline → 执行"""
    url_rules = config.get("url_rules", [])
    rule = match_url_rule(url_rules, path)
    if not rule:
        log.warning(f"No URL rule matched: {path}")
        return None

    pipeline_id = rule.get("pipeline_id")
    pipelines = config.get("pipelines", [])
    pipeline = next((p for p in pipelines if p.get("id") == pipeline_id), None)
    if not pipeline:
        log.error(f"Pipeline {pipeline_id} not found")
        return None

    return run_pipeline(config, pipeline, event)


# ════════════════════════════════════════════════════════
# 工序链执行
# ════════════════════════════════════════════════════════
def run_pipeline(config: dict, pipeline: dict, event: dict = None) -> dict:
    """执行一条工序链"""
    global _tasks_lock
    if _tasks_lock is None:
        import threading
        _tasks_lock = threading.RLock()

    pid = pipeline.get("id")
    pname = pipeline.get("name", f"pipeline-{pid}")

    # 去重检查
    if pid in _running:
        log.info(f"Pipeline {pid} already running, skip")
        return _tasks[_running[pid]]

    task_id = str(uuid.uuid4())[:8]
    task = {
        "id": task_id,
        "pipeline_id": pid,
        "pipeline_name": pname,
        "status": "running",
        "started_at": time.time(),
        "finished_at": 0,
        "steps": [],
        "output_files": [],
        "event": event or {},
    }

    with _tasks_lock:
        _tasks[task_id] = task
        _running[pid] = task_id

    log.info(f"▶ Pipeline [{pname}] task={task_id} started")

    # 执行步骤
    steps = pipeline.get("steps", [])
    on_failure = pipeline.get("on_failure", "stop")  # stop | continue

    for i, step_cfg in enumerate(steps):
        step_result = _run_step(config, step_cfg, i, task, event)
        task["steps"].append(step_result)

        if step_result["exit_code"] != 0 and on_failure == "stop":
            log.warning(f"  Step {i+1} FAILED, stopping pipeline")
            task["status"] = "failed"
            break
    else:
        task["status"] = "success"

    # 收尾
    task["finished_at"] = time.time()
    task["duration_sec"] = round(task["finished_at"] - task["started_at"], 2)

    # 发通知
    notif_cfg = pipeline.get("notification", {})
    if notif_cfg.get("enabled"):
        send_notification(notif_cfg, task)

    with _tasks_lock:
        if pid in _running:
            del _running[pid]

    log.info(f"◼ Pipeline [{pname}] task={task_id} {task['status']} ({task['duration_sec']}s)")
    return task


def _run_step(config: dict, step_cfg: dict, index: int, task: dict, event: dict) -> dict:
    """执行单个步骤"""
    toolchain_id = step_cfg.get("toolchain_id")
    toolchains = config.get("toolchains", [])
    tc = next((t for t in toolchains if t.get("id") == toolchain_id), None)

    step = {
        "index": index,
        "name": step_cfg.get("name", f"step-{index+1}"),
        "toolchain_id": toolchain_id,
        "toolchain_name": tc.get("name", "") if tc else "",
        "started_at": time.time(),
        "finished_at": 0,
        "exit_code": -1,
        "stdout": "",
        "stderr": "",
    }

    if not tc:
        step["stderr"] = f"Toolchain {toolchain_id} not found"
        step["finished_at"] = time.time()
        return step

    tc_type = tc.get("type", "")

    # 渲染参数
    project = step_cfg.get("project", "")
    output_dir = step_cfg.get("output_dir", "D:/fw/output")
    os.makedirs(output_dir, exist_ok=True)

    # 替换变量
    board = (event or {}).get("body", {}).get("board", "") or (event or {}).get("query", {}).get("board", "")
    # 用于 str.replace（键带花括号）
    var_map = {
        "{project}": project,
        "{output_dir}": output_dir,
        "{board}": board,
    }
    # 用于 str.format（键为合法标识符）
    fmt_map = {
        "project": project,
        "output_dir": output_dir,
        "board": board,
    }

    args_tpl = tc.get("args_template", "")
    for k, v in var_map.items():
        args_tpl = args_tpl.replace(k, str(v))

    # ── SSH 类型 ──
    if tc_type == "ssh":
        env = tc.get("env_vars", {})
        host = env.get("ssh_host", "")
        port = int(env.get("ssh_port", 22))
        user = env.get("ssh_user", "")
        key = env.get("ssh_key", "")
        cmd = env.get("ssh_command", "").format_map(_SafeDict(fmt_map))

        log.info(f"  [SSH] {step['name']}: {user}@{host} → {cmd}")
        rc, out, err = run_ssh_command(host, port, user, key, cmd)
        step["stdout"] = out
        step["stderr"] = err
        step["exit_code"] = rc

    # ── 本地命令 ──
    else:
        executable = tc.get("executable", "")
        env_vars = tc.get("env_vars", {})
        cmd = f'"{executable}" {args_tpl}'.strip()

        log.info(f"  [LOCAL] {step['name']}: {cmd}")
        try:
            proc = subprocess.run(
                cmd,
                shell=True,
                capture_output=True,
                text=True,
                timeout=step_cfg.get("timeout", 600),
                cwd=step_cfg.get("cwd", None),
                env={**os.environ, **env_vars} if env_vars else None,
            )
            step["stdout"] = proc.stdout[-5000:]  # 截断
            step["stderr"] = proc.stderr[-5000:]
            step["exit_code"] = proc.returncode
        except subprocess.TimeoutExpired:
            step["stderr"] = f"Timeout after {step_cfg.get('timeout', 600)}s"
            step["exit_code"] = 124
        except Exception as e:
            step["stderr"] = str(e)
            step["exit_code"] = 1

    step["finished_at"] = time.time()
    step["duration_sec"] = round(step["finished_at"] - step["started_at"], 2)

    # 收集产物
    if step["exit_code"] == 0:
        for f in step_cfg.get("output_files", []):
            f = f.format_map(_SafeDict(fmt_map))
            if os.path.exists(f):
                task["output_files"].append(f)

    return step


# ════════════════════════════════════════════════════════
# 任务查询
# ════════════════════════════════════════════════════════
def get_task(task_id: str) -> Optional[dict]:
    return _tasks.get(task_id)


def list_tasks(limit: int = 50) -> list:
    tasks = sorted(_tasks.values(), key=lambda t: t["started_at"], reverse=True)
    return tasks[:limit]


def get_pipeline_tasks(pipeline_id: int) -> list:
    return [t for t in _tasks.values() if t["pipeline_id"] == pipeline_id]