cfc/运营文案/opencode_client.py

# -*- coding: utf-8 -*-
"""
opencode_client.py — 通过本机常驻 opencode 服务的 HTTP API 进行会话交互

负责：
    1. 认证（basic auth）
    2. chat_id → session_id 映射的持久化（JSON 文件，每用户/群一个 opencode 会话）
    3. 确保某 chat 的会话存在（build agent）
    4. 发消息 + 轮询取 assistant 最终文本回复

权限策略：
    - 允许 bash/shell（可执行 git 等只读命令，满足「检查同步」类需求）
    - 仍禁用 write/edit/apply_patch 等写文件操作与任务派发
    - 因此 opencode 可以分析、执行只读命令、生成回复，但无法改动本机文件
"""
import io
import json
import os
import sys
import time
import urllib.parse
import urllib.request
import base64

SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
OPERATION_DIR = SCRIPT_DIR
CFC_ROOT = os.path.dirname(OPERATION_DIR)

# opencode 常驻服务
OPCODE_BASE = "http://127.0.0.1:4090"
OPCODE_USER = "opencode"
OPCODE_PASSWORD = os.environ.get("OPENCODE_SERVER_PASSWORD", "IwinTrue@123")

# 会话映射文件（chat_id -> session_id）
SESSION_MAP_PATH = os.path.join(OPERATION_DIR, "opencode_sessions.json")

# opencode 会话工作目录（项目根，含素材库/脚本，open code 可直接只读分析）
WORK_DIR = CFC_ROOT

# build agent 默认工作于该 project 的模型取服务端默认
AGENT = "build"

# 默认模型。opencode 内置 big-pickle 免费额度有限流，显式指定可用的 axon/deepseek。
MODEL = {"providerID": "axon", "modelID": "deepseek"}

# 只读工具白名单：其余工具（bash 执行、写文件等）一律禁用
# 此处列出 opencode 内置常用只读工具；未列出的默认不可用（服务端按 tools 参数过滤）
READONLY_TOOLS = []

# 危险的需授权工具，显式禁用（防御性，即便服务端宽松也不给）
FORBIDDEN_TOOLS = [
    "write", "edit", "apply_patch",
    "task", "dispatch", "webfetch_post", "chrome_launch",
]


def _auth_header() -> str:
    token = base64.b64encode(
        f"{OPCODE_USER}:{OPCODE_PASSWORD}".encode("utf-8")
    ).decode("ascii")
    return f"Basic {token}"


def _api(method: str, path: str, body=None, timeout: float = 60):
    """请求 opencode HTTP API，返回 (status, decoded_body)。"""
    url = OPCODE_BASE + path
    data = None
    if body is not None:
        data = json.dumps(body, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(url, data=data, method=method)
    req.add_header("Authorization", _auth_header())
    if body is not None:
        req.add_header("Content-Type", "application/json; charset=utf-8")
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        raw = resp.read().decode("utf-8", errors="replace")
        return resp.status, raw


def _load_session_map() -> dict:
    if not os.path.exists(SESSION_MAP_PATH):
        return {}
    try:
        with open(SESSION_MAP_PATH, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return {}


def _save_session_map(mapping: dict):
    with open(SESSION_MAP_PATH, "w", encoding="utf-8") as f:
        json.dump(mapping, f, ensure_ascii=False, indent=2)


def _create_session(chat_id: str) -> str:
    """为某 chat 新建一个 build agent 会话，返回 session_id。"""
    dir_q = urllib.parse.quote(WORK_DIR)
    body = {
        "title": f"xhs-bot-{chat_id[:24]}",
        "agent": AGENT,
    }
    st, raw = _api("POST", f"/session?directory={dir_q}", body=body, timeout=30)
    if st not in (200, 201):
        raise RuntimeError(f"创建会话失败 status={st}: {raw[:300]}")
    data = json.loads(raw)
    sid = data.get("id")
    if not sid:
        raise RuntimeError(f"创建会话未返回 session_id: {raw[:300]}")
    return sid


def ensure_session(chat_id: str) -> str:
    """确保某 chat 有对应 opencode 会话，返回 session_id。"""
    mapping = _load_session_map()
    sid = mapping.get(chat_id)
    if sid:
        return sid
    sid = _create_session(chat_id)
    mapping[chat_id] = sid
    _save_session_map(mapping)
    return sid


def send_prompt(session_id: str, text: str, timeout: float = 30):
    """向指定会话异步发送一条用户消息。返回 HTTP 状态。"""
    parts = [{"type": "text", "text": text}]
    body = {
        "parts": parts,
        "model": MODEL,
        "tools": {t: False for t in FORBIDDEN_TOOLS},
    }
    st, raw = _api("POST", f"/session/{session_id}/prompt_async",
                   body=body, timeout=timeout)
    return st, raw


def _extract_assistant_text(msgs) -> str:
    """从 v1 message 列表里提取最后一条 assistant 的最终可见文本。"""
    last = msgs[-1] if msgs else None
    if not last:
        return ""
    if last.get("info", {}).get("role") != "assistant":
        return ""
    texts = [p.get("text", "") for p in last.get("parts", [])
             if p.get("type") == "text" and p.get("text")]
    return texts[-1] if texts else ""


def _extract_progress(msgs, sent_ts: float = 0.0):
    """从消息列表提取当前进展摘要（含思考 reasoning、工具调用、输出文本）。

    sent_ts：仅提取该时刻之后（本次 prompt 之后）新增的 assistant 消息，
    防止复用会话时把旧回复当成新进展上报。
    """
    items = []
    for m in reversed(msgs):
        info = m.get("info", {})
        if info.get("role") != "assistant":
            continue
        m_created = (info.get("time") or {}).get("created", 0)
        if m_created < sent_ts:
            continue
        for p in reversed(m.get("parts", [])):
            t = p.get("type")
            if t == "reasoning":
                txt = (p.get("text") or "").strip()
                if txt:
                    items.append(("思考", txt))
            elif t == "text":
                txt = (p.get("text") or "").strip()
                if txt:
                    items.append(("输出", txt))
            elif t == "tool":
                tool = p.get("tool") or {}
                name = tool.get("name") if isinstance(tool, dict) else str(tool)
                if name:
                    items.append(("工具", str(name)))
            if len(items) >= 3:
                break
        if len(items) >= 3:
            break
    if not items:
        return "(处理中...)"
    return " | ".join(f"[{k}] {v[:120]}" for k, v in items)


def wait_for_reply(session_id: str, timeout: float = 180,
                   poll_interval: float = 5.0,
                   progress_callback=None, progress_interval: float = 120.0,
                   sent_ts: float = None) -> str:
    """等待会话最新 assistant 回复完成，返回最终文本。

    处理两种情形：
      1) assistant 回复极快，进入本函数时已生成 → 直接看最后一条 assistant 文本
      2) 正常流式：等待「发消息后新增」的 assistant 带文本出现

    超长任务支持（progress_callback 非 None 时启用）：
      - 每 progress_interval 秒提取进展（含 reasoning）回调给调用方
      - 若两次回调间无任何进展，向 opencode 提交「在干什么」推动执行

    sent_ts：发送 prompt 时的毫秒时间戳。所有返回路径只接受此时刻之后的
    assistant 消息，防止在复用会话中错误返回旧消息。

    timeout 为 None 时表示无限等待，直到 assistant 产生最终回复才返回。
    """


    _sent_ts = sent_ts or 0.0

    try:
        _, raw0 = _api("GET", f"/session/{session_id}/message", timeout=30)
        before = len(json.loads(raw0)) if raw0 else 0
    except Exception:
        before = 0

    deadline = (time.time() + timeout) if timeout is not None else None
    last_report_at = time.time()
    # 跟踪最后一条 assistant 消息的 created 时间戳，只有它不更新才算真卡住
    last_asst_ts = 0
    # nudge 冷却：同一会话 300s 内只允许触发一次
    last_nudge_at = 0
    while deadline is None or time.time() < deadline:
        time.sleep(poll_interval)
        try:
            st, raw = _api("GET", f"/session/{session_id}/message", timeout=30)
            if st != 200 or not raw:
                continue
            msgs = json.loads(raw)
        except Exception:
            continue
        if not msgs:
            continue

        # 快速路径：最后一条消息是新 assistant 且有文本 → 直接返回
        last = msgs[-1]
        if last.get("info", {}).get("role") == "assistant":
            last_created = (last.get("info", {}).get("time") or {}).get("created", 0)
            if last_created >= _sent_ts:
                lt = [p.get("text", "") for p in last.get("parts", [])
                      if p.get("type") == "text" and (p.get("text") or "").strip()]
                if lt:
                    return lt[-1]

        # 常规路径：遍历消息找第一个（最新）满足条件的 assistant 文本
        if len(msgs) > before:
            for m in reversed(msgs):
                info = m.get("info", {})
                if info.get("role") != "assistant":
                    continue
                m_created = (info.get("time") or {}).get("created", 0)
                if m_created < _sent_ts:
                    continue
                txts = [p.get("text", "") for p in m.get("parts", [])
                        if p.get("type") == "text" and (p.get("text") or "").strip()]
                if txts:
                    return txts[-1]

        # 超长任务进展上报（含卡住检测）
        if progress_callback and (time.time() - last_report_at) >= progress_interval:
            last_report_at = time.time()
            progress = _extract_progress(msgs, sent_ts=_sent_ts)

            # 找到最新的 assistant 消息的 created 时间戳
            latest_asst_ts = 0
            for m in msgs:
                info = m.get("info", {})
                if info.get("role") == "assistant":
                    ts = (info.get("time") or {}).get("created", 0)
                    if ts > latest_asst_ts:
                        latest_asst_ts = ts

            if latest_asst_ts == last_asst_ts and latest_asst_ts > _sent_ts:
                # 真正的卡住：assistant 消息时间戳没有变化
                now = time.time()
                if now - last_nudge_at >= 300:
                    last_nudge_at = now
                    progress_callback("⏳ 2 分钟无新进展，已向 opencode 询问「在干什么」…")
                    try:
                        send_prompt(session_id, "在干什么")
                    except Exception:
                        pass
            else:
                last_asst_ts = latest_asst_ts
                progress_callback(f"⏳ 进展：{progress}")

    raise TimeoutError(f"等待 opencode 回复超时（{timeout}s）")


def ask(chat_id: str, text: str, timeout: float = None,
        progress_callback=None, progress_interval: float = 120.0) -> str:
    """高层封装：确保会话 → 发消息 → 等回复 → 返回最终文本。

    timeout 为 None（默认）时无限等待，直到任务结束。
    """
    sid = ensure_session(chat_id)
    # 记录发送时刻（先于发送，保证新消息 created >= sent_ts，旧消息被过滤）
    sent_ts = time.time() * 1000
    st, raw = send_prompt(sid, text)
    if st != 204:
        raise RuntimeError(f"发送到 opencode 失败 status={st}: {raw[:300]}")
    return wait_for_reply(sid, timeout=timeout,
                          progress_callback=progress_callback,
                          progress_interval=progress_interval,
                          sent_ts=sent_ts)


if __name__ == "__main__":

    if not sys.stdout.isatty():
        try:
            sys.stdout.reconfigure(encoding='utf-8', errors='replace')
        except Exception:
            pass
    test_chat = "test-opencode-client"
    reply = ask(test_chat, "请用一句话回复：收到消息了吗？")
    print(f"REPLY: {reply}")