joy-cli/scripts/manhuang/actions-to-board.sh

#!/usr/bin/env bash
# 蛮荒可行动 actions + chain_v2 breaks → Task Board（family=observe）
# 用法: bash actions-to-board.sh [--force] [--dry-run]
set -euo pipefail

ROOT="$(cd "$(dirname "$0")/../.." && pwd)"
CACHE="${HOME}/.cache/manhuang"
STAMP_DIR="${CACHE}/board-actions"
LOG="${HOME}/logs/manhuang-quotes.log"
DASH="${DASH:-/tmp/joy-dashboard.json}"
NET="${HOME}/.cache/manhuang/net-path/status.json"
CHAIN="${HOME}/.cache/manhuang/chain/breaks.json"
THROTTLE_S="${MANHUANG_BOARD_THROTTLE_S:-43200}"
FORCE=0
DRY=0
for a in "$@"; do
  case "$a" in
    --force) FORCE=1 ;;
    --dry-run|--dry) DRY=1 ;;
  esac
done

mkdir -p "$STAMP_DIR" "$(dirname "$LOG")"
log() { echo "[$(date '+%Y-%m-%d %H:%M:%S')] actions-to-board: $*" | tee -a "$LOG"; }

BP=""
if [[ -x "${HOME}/Projects/linuxjoy/scripts/cluster/board-publish" ]]; then
  BP="${HOME}/Projects/linuxjoy/scripts/cluster/board-publish"
elif command -v board-publish >/dev/null 2>&1; then
  BP="$(command -v board-publish)"
fi
if [[ -z "$BP" ]]; then
  log "skip: board-publish not found"
  exit 0
fi

export PATH="/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:${HOME}/bin:${PATH:-}"
export DASH NET CHAIN FORCE THROTTLE_S STAMP_DIR DRY BP ROOT

python3 - <<'PY'
import hashlib, json, os, subprocess, time
from pathlib import Path

dash = Path(os.environ.get("DASH", "/tmp/joy-dashboard.json"))
netp = Path(os.environ.get("NET", ""))
chainp = Path(os.environ.get("CHAIN", ""))
stamp_dir = Path(os.environ["STAMP_DIR"])
throttle = int(os.environ.get("THROTTLE_S", "43200"))
force = os.environ.get("FORCE") == "1"
dry = os.environ.get("DRY") == "1"
bp = os.environ["BP"]
now = time.time()

# crit 断点更短节流；warn 用默认
CRIT_THROTTLE = min(throttle, 6 * 3600)


def log(msg: str) -> None:
    line = time.strftime("%Y-%m-%d %H:%M:%S") + f" actions-to-board: {msg}"
    print(line, flush=True)
    try:
        with open(Path.home() / "logs/manhuang-quotes.log", "a") as f:
            f.write(line + "\n")
    except OSError:
        pass


def break_to_action(b: dict, src: str) -> dict | None:
    if not isinstance(b, dict):
        return None
    reason = str(b.get("reason") or "").strip()
    title = str(b.get("title") or reason or "").strip()
    if not title and not reason:
        return None
    level = str(b.get("level") or "warn")
    if level in ("err",):
        level = "crit"
    # info/ok 不断
    if level not in ("crit", "warn", "err"):
        return None
    # 常态 info 类（系统代理 ON）不发 Board
    if reason in ("system_proxy_on_expected",):
        return None
    ring = str(b.get("ring") or "")
    t = title if title.startswith("[链断]") else f"[链断] {title}"
    return {
        "title": t[:120],
        "level": level,
        "kind": "chain_break",
        "detail": str(b.get("detail") or "")[:300],
        "next": str(b.get("next") or "")[:300],
        "tab": b.get("tab") or "mh-overview",
        "priority": 1 if level == "crit" else 2,
        "actionable": True,
        "reason": reason,
        "ring": ring,
        "owner": b.get("owner") or "linuxjoy",
        "sla_hint": b.get("sla_hint"),
        "path_id": b.get("evidence", {}).get("path_id") if isinstance(b.get("evidence"), dict) else None,
        "_src": src,
        "_break_id": b.get("id") or reason,
    }


actions: list[dict] = []

# 1) chain breaks 缓存（collect 写出）
if chainp.is_file():
    try:
        ch = json.loads(chainp.read_text())
        for b in ch.get("breaks") or []:
            a = break_to_action(b, "chain_cache")
            if a:
                if ch.get("path_id") and not a.get("path_id"):
                    a["path_id"] = ch.get("path_id")
                actions.append(a)
        # top_break 若未在列表
        tb = ch.get("top_break")
        if isinstance(tb, dict):
            a = break_to_action(tb, "chain_top")
            if a:
                actions.append(a)
    except Exception as e:
        log(f"chain_cache read err: {e}")

# 2) net_path actions
if netp.is_file():
    try:
        np = json.loads(netp.read_text())
        for a in np.get("actions") or []:
            if isinstance(a, dict) and a.get("actionable"):
                a = dict(a)
                a["_src"] = "net_path"
                actions.append(a)
    except Exception as e:
        log(f"net_path read err: {e}")

# 3) dashboard manhuang.actions + chain.breaks
if dash.is_file():
    try:
        m = json.loads(dash.read_text()).get("manhuang") or {}
        for b in (m.get("chain") or {}).get("breaks") or []:
            a = break_to_action(b, "dashboard_chain")
            if a:
                actions.append(a)
        for a in m.get("actions") or []:
            if not isinstance(a, dict):
                continue
            if a.get("level") not in ("crit", "warn", "err"):
                continue
            if a.get("actionable") is False:
                continue
            kind = str(a.get("kind") or "")
            if kind not in ("net_path", "chain", "chain_break", "miling", "brand_ioc", "lookalike", "kev_stack"):
                if int(a.get("priority") or 9) > 2:
                    continue
            a = dict(a)
            a["_src"] = "dashboard"
            actions.append(a)
    except Exception as e:
        log(f"dashboard read err: {e}")

# 排序：crit 先 · chain_break 优先 · priority
def _sort_key(a: dict):
    lv = str(a.get("level") or "")
    crit = 0 if lv in ("crit", "err") else 1
    src = 0 if a.get("kind") == "chain_break" else (1 if a.get("_src") == "net_path" else 2)
    return (crit, src, int(a.get("priority") or 9))


actions.sort(key=_sort_key)

seen = set()
published = 0
for a in actions:
    t = str(a.get("title") or "").strip()
    # 去重键：break_id 或 title
    key = str(a.get("_break_id") or a.get("reason") or t)
    if not t or key in seen:
        continue
    seen.add(key)
    h = hashlib.sha256(key.encode()).hexdigest()[:16]
    stamp = stamp_dir / f"{h}.ts"
    thr = CRIT_THROTTLE if str(a.get("level")) in ("crit", "err") else throttle
    if not force and stamp.is_file():
        try:
            age = now - float(stamp.read_text().strip() or "0")
            if age < thr:
                log(f"throttle skip: {t[:60]}")
                continue
        except Exception:
            pass
    kind = str(a.get("kind") or "net_path")
    cap = "security_ops" if kind in ("brand_ioc", "lookalike", "kev_stack") else "observability_ops"
    body_lines = [
        f"[蛮荒自动] kind={kind} level={a.get('level')} src={a.get('_src')}",
        f"reason: {a.get('reason') or '—'}",
        f"ring: {a.get('ring') or '—'}",
        f"detail: {a.get('detail') or ''}",
        f"next: {a.get('next') or ''}",
        f"tab: {a.get('tab') or ''}",
        f"owner: {a.get('owner') or 'linuxjoy'}",
        f"sla_hint: {a.get('sla_hint') or '—'}",
    ]
    if a.get("path_id"):
        body_lines.append(f"path_id: {a.get('path_id')}")
    body_lines.extend([
        "family=observe · 只读排查 · 勿自动改生产 · 勿关系统代理",
        "规则: configs/manhuang/clash-rules-shenzhou-direct.snippet.yaml",
        "链缓存: ~/.cache/manhuang/chain/breaks.json",
    ])
    body = "\n".join(body_lines)
    title = t[:120]
    log(f"publish: {title}")
    if dry:
        log(f"dry-run skip hash={h} thr={thr}s")
        published += 1  # dry 计数为「将发布」
        continue
    cmd = [
        bp, "task",
        "--project", "P-manhuang-net",
        "--title", title,
        "--capability", cap,
        "--family", "observe",
        "--priority", "high" if str(a.get("level")) in ("crit", "err") else "normal",
        "--task-text", body,
        "--timeout", "120",
    ]
    try:
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=90)
        out = ((r.stdout or "") + (r.stderr or "")).strip()
        if r.returncode == 0:
            stamp.write_text(str(int(time.time())))
            published += 1
            last = out.splitlines()[-1] if out else "ok"
            log(f"ok: {last}")
        else:
            log(f"fail rc={r.returncode}: {out[-300:]}")
    except Exception as e:
        log(f"fail: {e}")

log(f"done published={published} candidates={len(actions)} unique={len(seen)}")
PY