app.py

import json
import re
import textwrap
from datetime import datetime

import gradio as gr


def _now() -> str:
    return datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")


SENSITIVE_KEYS = {
    "apikey",
    "token",
    "password",
    "secret",
    "authorization",
    "bearer",
    "access_token",
    "refresh_token",
}


def _looks_sensitive_key(k: str) -> bool:
    lk = k.lower()
    return any(s in lk for s in SENSITIVE_KEYS)


def _mask_value(v):
    if v is None:
        return v
    if isinstance(v, (int, float, bool)):
        return v
    s = str(v).strip()
    if not s:
        return s
    if len(s) <= 8:
        return "***"
    return s[:2] + "***" + s[-2:]


def redact(obj):
    if isinstance(obj, dict):
        out = {}
        for k, v in obj.items():
            if _looks_sensitive_key(str(k)):
                out[k] = _mask_value(v)
            else:
                out[k] = redact(v)
        return out
    if isinstance(obj, list):
        return [redact(x) for x in obj]
    return obj


def explain_config(raw: str):
    raw = (raw or "").strip()
    if not raw:
        return "请粘贴 openclaw.json 内容。", "", ""

    try:
        cfg = json.loads(raw)
    except Exception as e:
        return f"JSON 解析失败：{e}", "", ""

    red = redact(cfg)
    red_json = json.dumps(red, ensure_ascii=False, indent=2)

    tips = [f"体检时间：{_now()}"]

    def has(path):
        cur = cfg
        for p in path:
            if not isinstance(cur, dict) or p not in cur:
                return False
            cur = cur[p]
        return True

    if has(["gateway", "auth", "token"]) and has(["gateway", "auth", "password"]) and not has(
        ["gateway", "auth", "mode"]
    ):
        tips.append(
            "风险：同时配置了 gateway.auth.token 和 gateway.auth.password，但未显式设置 gateway.auth.mode；可能导致鉴权行为不符合预期。"
        )
        tips.append("建议：明确写 gateway.auth.mode（按你的实际需求选 token/password/both）。")

    talk = cfg.get("talk", {}) if isinstance(cfg, dict) else {}
    if isinstance(talk, dict) and "silenceTimeoutMs" in talk:
        tips.append(f"Talk：silenceTimeoutMs={talk.get('silenceTimeoutMs')}（可控制静音多久后自动发送）。")

    provider_hits = []

    def walk(o, prefix=""):
        if isinstance(o, dict):
            for k, v in o.items():
                np = f"{prefix}.{k}" if prefix else str(k)
                if str(k).lower() in ("baseurl", "endpoint", "url") and isinstance(v, str):
                    provider_hits.append((np, v))
                walk(v, np)
        elif isinstance(o, list):
            for i, x in enumerate(o):
                walk(x, f"{prefix}[{i}]")

    walk(cfg)
    if provider_hits:
        tips.append("检测到可能的接口地址（原样显示；注意不要公开你的 key）：")
        for p, v in provider_hits[:10]:
            tips.append(f"- {p} = {v}")
        if len(provider_hits) > 10:
            tips.append(f"- ... 共 {len(provider_hits)} 条（仅展示前 10）")

    cmds = textwrap.dedent(
        """
        # 常用排障命令（复制到终端执行）
        openclaw status
        openclaw gateway status
        openclaw gateway probe
        openclaw logs --follow
        openclaw doctor
        openclaw channels status --probe
        """
    ).strip()

    report = "\n".join(f"- {t}" for t in tips)
    return report, red_json, cmds


def explain_logs(raw: str):
    raw = (raw or "").strip()
    if not raw:
        return "请粘贴日志片段。", ""

    lines = raw.splitlines()
    score = {"auth": 0, "rpc": 0, "channel": 0, "provider": 0, "other": 0}

    patterns = {
        "auth": [r"401", r"unauthorized", r"forbidden", r"auth", r"token"],
        "rpc": [r"rpc", r"timeout", r"ws://", r"websocket"],
        "channel": [r"telegram", r"signal", r"discord", r"channel"],
        "provider": [r"openai", r"zhipu", r"siliconflow", r"baseurl", r"api"],
    }

    for ln in lines[:2000]:
        low = ln.lower()
        matched = False
        for k, ps in patterns.items():
            if any(re.search(p, low) for p in ps):
                score[k] += 1
                matched = True
        if not matched:
            score["other"] += 1

    kind = sorted(score.items(), key=lambda x: x[1], reverse=True)[0][0]

    advice = [f"体检时间：{_now()}", f"粗分类：{kind}（仅基于关键词，供快速定位）"]
    next_cmds = ["openclaw status", "openclaw gateway probe"]

    if kind == "auth":
        advice.append("关注点：鉴权配置/allowedOrigins/上游 token。")
        next_cmds += ["openclaw channels status --probe"]
    elif kind == "rpc":
        advice.append("关注点：gateway 重启后的短暂恢复窗口、网络、RPC 超时。")
        next_cmds += ["openclaw gateway status", "openclaw logs --follow"]
    elif kind == "channel":
        advice.append("关注点：通道连接/回调、provider 配置是否影响发送。")
        next_cmds += ["openclaw channels status --probe", "openclaw logs --follow"]
    elif kind == "provider":
        advice.append("关注点：baseUrl、API key、模型名、限流/401。")
        next_cmds += ["openclaw logs --follow"]
    else:
        advice.append("关注点：先用 openclaw doctor + logs 定位模块。")
        next_cmds += ["openclaw doctor", "openclaw logs --follow"]

    return "\n".join(f"- {x}" for x in advice), "\n".join(next_cmds)


with gr.Blocks(title="OpenClaw 运维工具站（中文）") as demo:
    gr.Markdown(
        """
# OpenClaw 运维工具站（中文）

- **配置体检**：粘贴 `openclaw.json` → 输出中文风险提示 + 脱敏后的配置
- **日志体检**：粘贴日志片段 → 快速归类 + 给出下一步排障命令

说明：本工具会对疑似敏感字段做脱敏显示，但**请不要在公开场合粘贴真实密钥**。
        """.strip()
    )

    with gr.Tab("配置体检"):
        cfg_in = gr.Textbox(
            label="粘贴 openclaw.json（完整 JSON）",
            lines=16,
            placeholder="{\n  \"gateway\": { ... }\n}\n",
        )
        btn = gr.Button("开始体检")
        report = gr.Textbox(label="中文结论", lines=10)
        redacted = gr.Textbox(label="脱敏后的配置（可分享）", lines=16)
        cmds = gr.Textbox(label="下一步命令（复制执行）", lines=8)
        btn.click(explain_config, inputs=[cfg_in], outputs=[report, redacted, cmds])

    with gr.Tab("日志体检"):
        log_in = gr.Textbox(label="粘贴日志片段", lines=16)
        btn2 = gr.Button("开始体检")
        report2 = gr.Textbox(label="中文结论", lines=10)
        cmds2 = gr.Textbox(label="下一步命令（复制执行）", lines=8)
        btn2.click(explain_logs, inputs=[log_in], outputs=[report2, cmds2])

    gr.Markdown(
        """
### 你可以怎么用它
1) 先跑 `openclaw status`，把关键报错贴到“日志体检”。
2) 想检查配置，就把 `openclaw.json` 全文贴到“配置体检”。

### 后续计划
- 配置项中文解释更细
- 常见错误库（按版本/通道/Provider）
- 一键生成备份/恢复脚本（脱敏）
        """.strip()
    )


demo.queue()


def main():
    demo.launch(server_name="0.0.0.0", server_port=7860)


if __name__ == "__main__":
    main()