fnos-agent-checkin/AgentCheckin/app/backend/api.py

#!/usr/bin/env python3
"""AgentCheckin CGI API server (invoked by app/ui/index.cgi).

All endpoints are JSON. The fnOS web server validates the NAS login session
before calling the CGI, so no extra auth is implemented here.
"""
import json
import os
import sys
import time
import urllib.parse

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from common import (load_accounts, load_settings, save_settings, load_state,
                    save_state, log, fmt_now, now_ts, PID_FILE, TRIGGER_FILE,
                    ensure_dirs, mask_token, public_account)
import engine
import workbuddy

MAX_BODY = 256 * 1024
OAUTH_TTL = 10 * 60  # seconds a login link stays valid


# ---------------------------------------------------------------------------
# CGI plumbing
# ---------------------------------------------------------------------------

def cgi_env():
    return {
        "method": os.environ.get("REQUEST_METHOD", "GET").upper(),
        "uri": os.environ.get("REQUEST_URI", ""),
        "query": os.environ.get("QUERY_STRING", ""),
    }


def read_body():
    try:
        length = int(os.environ.get("CONTENT_LENGTH") or 0)
    except ValueError:
        length = 0
    if length <= 0:
        return {}
    if length > MAX_BODY:
        return None
    raw = sys.stdin.buffer.read(length)
    try:
        data = json.loads(raw.decode("utf-8"))
        return data if isinstance(data, dict) else {}
    except (ValueError, json.JSONDecodeError):
        return {}


def respond(obj, status="200 OK"):
    body = json.dumps(obj, ensure_ascii=False)
    out = sys.stdout
    out.write("Status: %s\r\n" % status)
    out.write("Content-Type: application/json; charset=utf-8\r\n")
    out.write("Cache-Control: no-store\r\n")
    out.write("Content-Length: %d\r\n" % len(body.encode("utf-8")))
    out.write("\r\n")
    out.write(body)


def error(msg, status="400 Bad Request"):
    respond({"error": msg}, status)


def route_path(uri):
    path = uri.split("?", 1)[0]
    idx = path.find("index.cgi")
    if idx >= 0:
        path = path[idx + len("index.cgi"):]
    return path.rstrip("/") or "/"


def query_params(qs):
    return {k: v[0] for k, v in urllib.parse.parse_qs(qs, keep_blank_values=True).items()}


def daemon_alive():
    try:
        with open(PID_FILE) as f:
            pid = int(f.read().strip())
        os.kill(pid, 0)
        return True
    except (OSError, ValueError):
        return False


# ---------------------------------------------------------------------------
# Handlers
# ---------------------------------------------------------------------------

def h_state(_params, _body):
    data = engine.dashboard()
    data["daemon_alive"] = daemon_alive()
    return data


def h_checkin(_params, body):
    account_id = body.get("account_id")
    accounts = load_accounts()
    if account_id and not any(a.get("id") == account_id for a in accounts):
        return ("error", "账号不存在")
    if daemon_alive():
        ensure_dirs()
        with open(TRIGGER_FILE, "w") as f:
            f.write(json.dumps({"account_id": account_id, "ts": now_ts()}))
        return {"ok": True, "mode": "trigger", "message": "已触发签到，守护进程正在执行"}
    # daemon not running: run synchronously as a fallback
    results = engine.run_all(account_id=account_id, reason="manual-cgi")
    return {"ok": True, "mode": "sync", "results": results}


def h_accounts(_params, _body):
    return {"accounts": [public_account(a) for a in load_accounts()]}


def h_add_wb_manual(_params, body):
    token = (body.get("access_token") or "").strip()
    if not token:
        return ("error", "access_token 不能为空")
    if len(token) > 8192:
        return ("error", "access_token 过长")
    acc = engine.add_workbuddy_manual(
        name=(body.get("name") or "").strip()[:60],
        access_token=token,
        refresh_token=(body.get("refresh_token") or "").strip(),
        domain=(body.get("domain") or "codebuddy.cn").strip(),
    )
    return {"ok": True, "account": public_account(acc)}


def h_add_trae(_params, body):
    token = (body.get("access_token") or "").strip()
    if not token:
        return ("error", "access_token 不能为空")
    if len(token) > 8192:
        return ("error", "access_token 过长")
    acc = engine.add_trae(
        name=(body.get("name") or "").strip()[:60],
        access_token=token,
        refresh_token=(body.get("refresh_token") or "").strip(),
        device_id_str=(body.get("device_id") or "").strip(),
    )
    return {"ok": True, "account": public_account(acc)}



def h_add_qoder(_params, body):
    token = (body.get("access_token") or "").strip()
    if not token:
        return ("error", "access_token 不能为空")
    if len(token) > 8192:
        return ("error", "access_token 过长")
    device = body.get("device") or {}
    if not isinstance(device, dict):
        return ("error", "device 必须是对象")
    acc = engine.add_qoder(
        name=(body.get("name") or "").strip()[:60],
        access_token=token,
        refresh_token=(body.get("refresh_token") or "").strip(),
        device=device,
        domain=(body.get("domain") or "openapi.qoder.com.cn").strip(),
    )
    return {"ok": True, "account": public_account(acc)}

def h_update_account(_params, body):
    acc_id = body.get("id")
    if not acc_id:
        return ("error", "缺少 id")
    fields = {}
    if "name" in body:
        fields["name"] = str(body["name"])[:60]
    if "enabled" in body:
        fields["enabled"] = bool(body["enabled"])
    acc = engine.update_account(acc_id, **fields)
    if not acc:
        return ("error", "账号不存在")
    return {"ok": True, "account": public_account(acc)}



def h_fix_qoder_domain(_params, body):
    """Probe both Qoder sites with the account token and switch to the one
    that authenticates (the two sites' tokens are not interchangeable)."""
    acc_id = body.get("id")
    if not acc_id:
        return ("error", "缺少 id")
    accounts = load_accounts()
    acc = engine.find_account(accounts, acc_id)
    if not acc or acc.get("platform") != "qoder":
        return ("error", "账号不存在或不是 Qoder 账号")
    import qoder as _q
    probe = _q.probe_domains(acc)
    for dom, r in probe.items():
        if r.get("auth"):
            acc.setdefault("auth", {})["domain"] = dom
            engine.save_accounts(accounts)
            log("qoder account %s domain switched to %s" % (acc_id, dom))
            return {"ok": True, "domain": dom, "probe": probe}
    return {"ok": False, "probe": probe,
            "error": "两个站点均无法认证（Token 可能已过期），请重新获取 Token"}


def h_umid_status(_params, _body):
    """Qoder 风控身份（UMID runtime-info）安装状态。"""
    import umid as _u
    return _u.status()


def h_umid_install(_params, _body):
    """下载官方 qodercli 并提取 runtime-info（约 30MB，耗时 1-3 分钟）。"""
    import umid as _u
    ok, msg = _u.install(force=bool((_body or {}).get("force")))
    return {"ok": ok, "message": msg, "status": _u.status()}


def h_qoder_refresh_risk(_params, body):
    """为某个 Qoder 账号重新生成风控身份。"""
    acc_id = body.get("id")
    if not acc_id:
        return ("error", "缺少 id")
    accounts = load_accounts()
    acc = engine.find_account(accounts, acc_id)
    if not acc or acc.get("platform") != "qoder":
        return ("error", "账号不存在或不是 Qoder 账号")
    import qoder as _q
    tok, code, mtype, err = _q.risk_identity(acc, force=True)
    if err:
        return {"ok": False, "error": err}
    engine.save_accounts([acc if a.get("id") == acc_id else a for a in accounts])
    return {"ok": True, "machineToken": tok[:12] + "…", "machineCode": code, "machineType": mtype}


def h_delete_account(_params, body):
    acc_id = body.get("id")
    if not acc_id:
        return ("error", "缺少 id")
    ok = engine.delete_account(acc_id)
    if not ok:
        return ("error", "账号不存在")
    return {"ok": True}


def h_test_account(_params, body):
    acc_id = body.get("id")
    if not acc_id:
        return ("error", "缺少 id")
    return engine.test_account(acc_id)


def h_wb_oauth_begin(_params, _body):
    state, login_url, err = workbuddy.begin_oauth()
    if err:
        return ("error", err)
    st = load_state()
    sessions = st.setdefault("oauth_sessions", {})
    # drop expired sessions
    now = now_ts()
    for k in list(sessions.keys()):
        if now - sessions[k].get("created", 0) > OAUTH_TTL:
            del sessions[k]
    sessions[state] = {"created": now}
    save_state(st)
    log("workbuddy oauth session started")
    return {"ok": True, "state": state, "login_url": login_url,
            "ttl": OAUTH_TTL}


def h_wb_oauth_poll(params, _body):
    state = params.get("state", "")
    if not state:
        return ("error", "缺少 state")
    st = load_state()
    sessions = st.get("oauth_sessions") or {}
    sess = sessions.get(state)
    if not sess:
        return ("error", "授权会话不存在或已过期，请重新生成链接")
    if now_ts() - sess.get("created", 0) > OAUTH_TTL:
        del sessions[state]
        save_state(st)
        return ("error", "授权链接已过期，请重新生成")
    status, payload, err = workbuddy.poll_oauth(state)
    if status == "done":
        acc = engine.add_workbuddy_oauth(payload)
        del sessions[state]
        save_state(st)
        return {"ok": True, "status": "done", "account": public_account(acc)}
    return {"ok": True, "status": "waiting"}


def h_settings_get(_params, _body):
    return {"settings": load_settings()}


def h_settings_post(_params, body):
    settings = load_settings()
    if "schedule_times" in body:
        times = body["schedule_times"]
        if not isinstance(times, list):
            return ("error", "schedule_times 必须是数组")
        clean = []
        for t in times:
            t = str(t)
            if len(t) == 5 and t[2] == ":" and t[:2].isdigit() and t[3:].isdigit():
                hh, mm = int(t[:2]), int(t[3:])
                if 0 <= hh < 24 and 0 <= mm < 60:
                    clean.append(t)
        if not clean:
            return ("error", "至少需要一个有效的签到时间（HH:MM）")
        settings["schedule_times"] = sorted(set(clean))
    if "webhook_url" in body:
        url = str(body["webhook_url"]).strip()
        if url and not url.startswith(("http://", "https://")):
            return ("error", "webhook_url 必须以 http(s):// 开头")
        settings["webhook_url"] = url[:500]
    if "webhook_type" in body:
        wt = str(body["webhook_type"])
        if wt in engine.WEBHOOK_TYPES:
            settings["webhook_type"] = wt
        else:
            return ("error", "不支持的 webhook_type")
    if "notify_on" in body:
        nf = body["notify_on"]
        if isinstance(nf, list):
            settings["notify_on"] = [n for n in nf if n in ("success", "fail")]
    save_settings(settings)
    return {"ok": True, "settings": settings}


def h_logs(params, _body):
    try:
        lines = int(params.get("lines", "200"))
    except ValueError:
        lines = 200
    lines = max(10, min(lines, 2000))
    from common import LOG_FILE
    if not os.path.exists(LOG_FILE):
        return {"lines": []}
    try:
        with open(LOG_FILE, encoding="utf-8") as f:
            content = f.read().splitlines()
        return {"lines": content[-lines:]}
    except OSError as e:
        return {"lines": [], "error": str(e)}


def h_qoder_probe(_params, body):
    """诊断：转储 Qoder 接口原始响应（含一次幂等领取）。"""
    import qoder as _q
    acc_id = body.get("id")
    acc = None
    for a in load_accounts():
        if a.get("id") == acc_id:
            acc = a
            break
    if acc is None:
        return ("error", "账号不存在")
    if acc.get("platform") != "qoder":
        return ("error", "该账号不是 Qoder")
    return _q.raw_probe(acc)


def h_qoder_experiment(_params, body):
    """诊断：定位「每日领 100 Credits」活动下发条件的对照实验（只读）。"""
    import qoder as _q
    acc = None
    for a in load_accounts():
        if a.get("id") == body.get("id"):
            acc = a
            break
    if acc is None:
        return ("error", "账号不存在")
    if acc.get("platform") != "qoder":
        return ("error", "该账号不是 Qoder")
    return _q.checkin_experiment(acc)


def h_qoder_growth(_params, body):
    """诊断：扒活动 iframe 页面/JS，定位每日签到真实接口。"""
    import qoder as _q
    acc = None
    for a in load_accounts():
        if a.get("id") == body.get("id"):
            acc = a
            break
    if acc is None:
        return ("error", "账号不存在")
    return _q.growth_probe(acc)


def h_qoder_matrix(_params, body):
    """诊断：umid 调用方式 × 请求头组合的对照矩阵。"""
    import qoder as _q
    acc = None
    for a in load_accounts():
        if a.get("id") == body.get("id"):
            acc = a
            break
    if acc is None:
        return ("error", "账号不存在")
    return _q.identity_matrix_probe(acc)


def h_qoder_desktop(_params, body):
    """诊断：模拟官方桌面端出站头，寻找能拿到 CLAIM_BENEFIT 的组合。"""
    import qoder as _q
    acc = None
    for a in load_accounts():
        if a.get("id") == body.get("id"):
            acc = a
            break
    if acc is None:
        return ("error", "账号不存在")
    return _q.desktop_emulation_probe(acc)


def h_qoder_login_start(_params, body):
    """发起 Qoder OAuth 设备授权登录（一键登录）。"""
    import qoder as _q
    realm = body.get("realm") or "intl"
    if realm not in ("intl", "cn"):
        return ("error", "realm 只能是 intl / cn")
    info = _q.device_login_start(realm)
    st = load_state()
    logins = st.setdefault("qoder_logins", {})
    now = now_ts()
    for k in list(logins.keys()):
        if now - logins[k].get("created", 0) > OAUTH_TTL:
            del logins[k]
    state = "qd%d" % int(now * 1000)
    info["created"] = now
    logins[state] = info
    save_state(st)
    log("qoder device login started (realm=%s)" % realm)
    return {"ok": True, "state": state, "auth_url": info["auth_url"],
            "realm": realm, "ttl": OAUTH_TTL}


def h_qoder_login_poll(params, body):
    """轮询 Qoder 设备授权结果，成功后直接入池。"""
    import qoder as _q
    state = params.get("state", "")
    st = load_state()
    info = (st.get("qoder_logins") or {}).get(state)
    if not info:
        return ("error", "登录会话不存在或已过期，请重新发起")
    res = _q.device_login_poll(info.get("realm", "intl"), info.get("verifier", ""),
                               info.get("nonce", ""), info.get("machine_id", ""))
    if res.get("status") != "ok":
        return res
    dev = {"machineId": info.get("machine_id", "")}
    uid = res.get("uid") or ""
    name = (body or {}).get("name") or ("Qoder %s"
           % ("国内版" if info.get("realm") == "cn" else "国际版"))
    # 同一个 uid 已经存在则更新凭证（把设备重新绑到本机），不新增重复账号
    existing = None
    for a in load_accounts():
        if a.get("platform") == "qoder" and uid and a.get("uid") == uid:
            existing = a
            break
    if existing is not None:
        accounts = load_accounts()
        acc = existing
        for a in accounts:
            if a.get("id") == existing["id"]:
                a.setdefault("auth", {})["accessToken"] = res["accessToken"]
                a["auth"]["refreshToken"] = res["refreshToken"]
                a["auth"]["domain"] = res["domain"]
                a["auth"]["expiresAt"] = res.get("expiresAt") or 0
                dev2 = dict(a.get("device") or {})
                dev2.update({k: v for k, v in dev.items() if v})
                a["device"] = dev2
                a["uid"] = uid
                a["need_reauth"] = False
                acc = a
        engine.save_accounts(accounts)
        log("qoder device login refreshed %s (uid=%s)"
            % (existing["id"], uid[:8]))
        msg = "登录成功：已把账号重新绑定到本机设备"
    else:
        acc = engine.add_qoder(
            name=name,
            access_token=res["accessToken"],
            refresh_token=res["refreshToken"],
            device={k: v for k, v in dev.items() if v},
            domain=res["domain"],
        )
        if uid:
            accounts = load_accounts()
            for a in accounts:
                if a.get("id") == acc["id"]:
                    a["uid"] = uid
            engine.save_accounts(accounts)
        msg = "登录成功，已加入账号列表"
    st = load_state()
    (st.get("qoder_logins") or {}).pop(state, None)
    save_state(st)
    log("qoder device login ok -> %s" % acc["id"])
    return {"status": "ok", "account": public_account(acc),
            "message": "登录成功，已加入账号列表"}


def h_qoder_import_device(_params, body):
    """导入桌面客户端的设备身份（machineToken/Code/Type）。"""
    acc_id = body.get("id")
    if not any(a.get("id") == acc_id for a in load_accounts()):
        return ("error", "账号不存在")
    tok = (body.get("machineToken") or "").strip()
    code = (body.get("machineCode") or "").strip()
    mtype = (body.get("machineType") or "").strip()
    if not (tok and code and mtype):
        return ("error", "machineToken / machineCode / machineType 都不能为空")
    accounts = load_accounts()
    for a in accounts:
        if a.get("id") == acc_id:
            dev = dict(a.get("device") or {})
            dev.update({"machineToken": tok, "machineCode": code,
                        "machineType": mtype, "uid": a.get("uid") or ""})
            a["device"] = dev
            break
    engine.save_accounts(accounts)
    log("qoder account %s device identity imported" % acc_id)
    return {"ok": True, "message": "已导入桌面端设备身份，下次签到会使用它"}


def h_debug(_params, _body):
    """Diagnostics: resolved paths, env, daemon liveness (no secrets)."""
    import common as _c
    info = {
        "data_dir": _c.DATA_DIR,
        "accounts_file": _c.ACCOUNTS_FILE,
        "state_file": _c.STATE_FILE,
        "pid_file": _c.PID_FILE,
        "log_file": _c.LOG_FILE,
        "env": {
            "TRIM_PKGVAR": os.environ.get("TRIM_PKGVAR", "<unset>"),
            "TRIM_APPDEST": os.environ.get("TRIM_APPDEST", "<unset>"),
            "AGENTCHECKIN_DEV": os.environ.get("AGENTCHECKIN_DEV", "<unset>"),
        },
        "file": __file__,
    }
    info["state_file_exists"] = os.path.exists(_c.STATE_FILE)
    if info["state_file_exists"]:
        st = _c.load_state()
        info["state_daemon"] = st.get("daemon")
        info["state_last_run"] = st.get("last_run")
    info["pid_file_exists"] = os.path.exists(_c.PID_FILE)
    info["daemon_alive"] = daemon_alive()
    return info


ROUTES = {
    ("GET", "/api/state"): h_state,
    ("GET", "/api/debug"): h_debug,
    ("POST", "/api/checkin"): h_checkin,
    ("GET", "/api/accounts"): h_accounts,
    ("POST", "/api/accounts/add_wb_manual"): h_add_wb_manual,
    ("POST", "/api/accounts/add_trae"): h_add_trae,
    ("POST", "/api/accounts/add_qoder"): h_add_qoder,
    ("POST", "/api/accounts/update"): h_update_account,
    ("POST", "/api/accounts/delete"): h_delete_account,
    ("POST", "/api/accounts/fix_domain"): h_fix_qoder_domain,
    ("POST", "/api/accounts/test"): h_test_account,
    ("POST", "/api/wb_oauth_begin"): h_wb_oauth_begin,
    ("GET", "/api/wb_oauth_poll"): h_wb_oauth_poll,
    ("GET", "/api/umid/status"): h_umid_status,
    ("POST", "/api/umid/install"): h_umid_install,
    ("POST", "/api/accounts/refresh_risk"): h_qoder_refresh_risk,
    ("POST", "/api/qoder/probe"): h_qoder_probe,
    ("POST", "/api/qoder/experiment"): h_qoder_experiment,
    ("POST", "/api/qoder/growth"): h_qoder_growth,
    ("POST", "/api/qoder/matrix"): h_qoder_matrix,
    ("POST", "/api/qoder/desktop"): h_qoder_desktop,
    ("POST", "/api/qoder/login/start"): h_qoder_login_start,
    ("GET", "/api/qoder/login/poll"): h_qoder_login_poll,
    ("POST", "/api/qoder/import_device"): h_qoder_import_device,
    ("GET", "/api/settings"): h_settings_get,
    ("POST", "/api/settings"): h_settings_post,
    ("GET", "/api/logs"): h_logs,
}


def main():
    env = cgi_env()
    path = route_path(env["uri"])
    params = query_params(env["query"])
    handler = ROUTES.get((env["method"], path))
    if not handler:
        return error("not found: %s %s" % (env["method"], path), "404 Not Found")
    body = read_body() if env["method"] == "POST" else {}
    if body is None:
        return error("payload too large")
    try:
        result = handler(params, body)
    except Exception as e:
        log("api error on %s %s: %s" % (env["method"], path, e), "ERROR")
        return error("internal error: %s" % e, "500 Internal Server Error")
    if isinstance(result, tuple) and len(result) == 2 and result[0] == "error":
        return error(result[1])
    respond(result if isinstance(result, dict) else {"data": result})


if __name__ == "__main__":
    main()