diff --git a/README.md b/README.md index 9692b70..51062d6 100644 --- a/README.md +++ b/README.md @@ -201,6 +201,8 @@ token printed to the console) that talks to your WeChat gateway so you can: to every message; Prefix = only when a message starts with `/ask `); - **pick which groups the bot may answer in** (checkboxes → `allowed_groups`; none checked = every group), choose the provider, and enable/disable the instance. +- **sign in by QR** — if the account is signed out, the portal shows the gateway's + login QR and continues automatically once you scan it. Saving writes `channels.toml` — restart `channels start` to apply. The gateway token stays server-side and never reaches the browser. Flags: `--port` (default diff --git a/coding_bridge/channels/portal.py b/coding_bridge/channels/portal.py index c639437..2d0dd17 100644 --- a/coding_bridge/channels/portal.py +++ b/coding_bridge/channels/portal.py @@ -31,6 +31,7 @@ import tempfile import threading import time +from concurrent.futures import ThreadPoolExecutor from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import TYPE_CHECKING, Any @@ -195,6 +196,9 @@ def __init__(self, settings: Settings, *, client: httpx.Client | None = None) -> self._owns_client = client is None self._contacts_cache: dict[str, tuple[float, list[dict[str, Any]]]] = {} self._lock = threading.Lock() + # Per-instance single-flight lock so overlapping searches (and the + # background warm) share ONE fetch instead of each paging 4k+ rows. + self._fetch_locks: dict[str, threading.Lock] = {} def close(self) -> None: if self._owns_client: @@ -255,19 +259,24 @@ def _instance(self, instance_id: str) -> WeChatInstanceConfig: # ---- WeChat gateway proxy ------------------------------------------- # - def _gateway_get(self, inst: WeChatInstanceConfig, path: str, params: dict[str, Any]) -> Any: + def _gateway_request( + self, inst: WeChatInstanceConfig, method: str, path: str, params: dict[str, Any] + ) -> httpx.Response: try: token = inst.resolve_token() except ConfigError as exc: raise PortalError(400, str(exc)) from None url = f"{inst.base_url}{path}" try: - resp = self._client.get( - url, params=params, headers={"Authorization": f"Bearer {token}"} + return self._client.request( + method, url, params=params, headers={"Authorization": f"Bearer {token}"} ) except httpx.HTTPError as exc: raise PortalError(502, f"gateway unreachable: {exc.__class__.__name__}") from None - if resp.status_code == 401 or resp.status_code == 403: + + def _gateway_get(self, inst: WeChatInstanceConfig, path: str, params: dict[str, Any]) -> Any: + resp = self._gateway_request(inst, "GET", path, params) + if resp.status_code in (401, 403): raise PortalError(502, "gateway rejected the token (check token_env/token_file)") if resp.status_code >= 400: raise PortalError(502, f"gateway returned {resp.status_code}") @@ -282,6 +291,56 @@ def account(self, instance_id: str) -> dict[str, Any]: def status(self, instance_id: str) -> dict[str, Any]: return self._gateway_get(self._instance(instance_id), "/api/auth/status", {}) + def qr(self, instance_id: str) -> dict[str, Any]: + """Fetch a login QR (base64 PNG), resolving the gateway's async UI task. + + The gateway's ``/api/auth/qr`` returns a queued ``UiTaskOut``; the image + lands in ``result`` once the task finishes, so we poll ``/api/tasks/{id}`` + server-side and hand the browser a ready-to-render data URL. Returns 409 + (surfaced as such) when the account is already logged in. + """ + inst = self._instance(instance_id) + resp = self._gateway_request(inst, "GET", "/api/auth/qr", {"type": "base64"}) + if resp.status_code == 409: + raise PortalError(409, "already logged in") + if resp.status_code in (401, 403): + raise PortalError(502, "gateway rejected the token (check token_env/token_file)") + if resp.status_code >= 400: + raise PortalError(502, f"gateway returned {resp.status_code}") + try: + task = resp.json() + except ValueError: + raise PortalError(502, "gateway returned non-JSON") from None + result = self._resolve_ui_task(inst, task) + b64 = result.get("base64") if isinstance(result, dict) else None + if not b64 or not isinstance(b64, str): + raise PortalError(502, "gateway did not return a QR image") + return {"base64": b64} + + def _resolve_ui_task( + self, inst: WeChatInstanceConfig, task: Any, *, attempts: int = 25, delay: float = 0.4 + ) -> Any: + """Return a UI task's ``result``, polling ``/api/tasks/{id}`` until it lands.""" + if not isinstance(task, dict): + raise PortalError(502, "gateway returned a malformed task") + if task.get("result"): + return task["result"] + task_id = task.get("id") + if not task_id or not isinstance(task_id, str): + raise PortalError(502, "gateway task has no id") + for _ in range(attempts): + time.sleep(delay) + polled = self._gateway_get(inst, f"/api/tasks/{task_id}", {}) + if not isinstance(polled, dict): + continue + if polled.get("result"): + return polled["result"] + err = polled.get("error") + if err: + msg = err.get("message") if isinstance(err, dict) else None + raise PortalError(502, f"gateway task failed: {msg or 'error'}") + raise PortalError(504, "gateway QR task timed out") + def groups(self, instance_id: str) -> list[dict[str, Any]]: inst = self._instance(instance_id) data = self._gateway_get(inst, "/api/conversations", {"limit": 200}) @@ -308,45 +367,86 @@ def search_contacts(self, instance_id: str, query: str, limit: int) -> list[dict ] def _contacts(self, inst: WeChatInstanceConfig) -> list[dict[str, Any]]: - now = time.monotonic() - with self._lock: + def _fresh() -> list[dict[str, Any]] | None: cached = self._contacts_cache.get(inst.instance_id) - if cached and now - cached[0] < _CONTACTS_TTL_SECONDS: + if cached and time.monotonic() - cached[0] < _CONTACTS_TTL_SECONDS: return cached[1] - contacts = self._fetch_all_contacts(inst) - with self._lock: - self._contacts_cache[inst.instance_id] = (now, contacts) - return contacts + return None - def _fetch_all_contacts(self, inst: WeChatInstanceConfig) -> list[dict[str, Any]]: + with self._lock: + hit = _fresh() + if hit is not None: + return hit + fetch_lock = self._fetch_locks.get(inst.instance_id) + if fetch_lock is None: + fetch_lock = threading.Lock() + self._fetch_locks[inst.instance_id] = fetch_lock + # Single-flight: the first caller fetches; concurrent callers block here + # and then reuse the just-populated cache instead of refetching. + with fetch_lock: + with self._lock: + hit = _fresh() + if hit is not None: + return hit + contacts = self._fetch_all_contacts(inst) + with self._lock: + self._contacts_cache[inst.instance_id] = (time.monotonic(), contacts) + return contacts + + @staticmethod + def _normalize_contacts(rows: Any) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] - offset = 0 - while offset < _CONTACTS_MAX: + if not isinstance(rows, list): + return out + for c in rows: + if not isinstance(c, dict): + continue + wxid = c.get("wechat_id") or "" + nickname = c.get("nickname") or "" + remark = c.get("remark") or "" + alias = c.get("alias") or "" + out.append( + { + "wechat_id": wxid, + "nickname": nickname, + "remark": remark, + "avatar_url": c.get("avatar_url"), + "_haystack": f"{wxid} {nickname} {remark} {alias}".lower(), + } + ) + return out + + def _fetch_page(self, inst: WeChatInstanceConfig, offset: int) -> list[dict[str, Any]]: + try: data = self._gateway_get( - inst, "/api/contacts", {"limit": _CONTACTS_PAGE, "offset": offset} + inst, + "/api/contacts", + {"limit": _CONTACTS_PAGE, "offset": offset, "type": "friend"}, ) - rows = data.get("contacts", []) if isinstance(data, dict) else [] - if not rows: - break - for c in rows: - if not isinstance(c, dict): - continue - wxid = c.get("wechat_id") or "" - nickname = c.get("nickname") or "" - remark = c.get("remark") or "" - alias = c.get("alias") or "" - out.append( - { - "wechat_id": wxid, - "nickname": nickname, - "remark": remark, - "avatar_url": c.get("avatar_url"), - "_haystack": f"{wxid} {nickname} {remark} {alias}".lower(), - } - ) - if len(rows) < _CONTACTS_PAGE: - break - offset += _CONTACTS_PAGE + except PortalError: + # Best-effort: a single transient page failure shouldn't fail the + # whole search — the target contact is likely on another page. + return [] + return self._normalize_contacts(data.get("contacts") if isinstance(data, dict) else None) + + def _fetch_all_contacts(self, inst: WeChatInstanceConfig) -> list[dict[str, Any]]: + first = self._gateway_get( + inst, "/api/contacts", {"limit": _CONTACTS_PAGE, "offset": 0, "type": "friend"} + ) + rows0 = first.get("contacts") if isinstance(first, dict) else None + out = self._normalize_contacts(rows0) + if not isinstance(rows0, list) or len(rows0) < _CONTACTS_PAGE: + return out + total = first.get("total") if isinstance(first, dict) else None + hi = total if isinstance(total, int) and 0 < total <= _CONTACTS_MAX else _CONTACTS_MAX + offsets = list(range(_CONTACTS_PAGE, hi, _CONTACTS_PAGE)) + if not offsets: + return out + # Fetch the remaining pages concurrently — the gateway caps limit at 200, + # so a few thousand friends is otherwise ~20 serial round-trips. + with ThreadPoolExecutor(max_workers=6) as pool: + for rows in pool.map(lambda o: self._fetch_page(inst, o), offsets): + out.extend(rows) return out @@ -497,6 +597,8 @@ def _handle_api_get(self, path: str, query: dict[str, list[str]]) -> None: self._send_json(200, service.account(instance)) elif path == "/api/wechat/status": self._send_json(200, service.status(instance)) + elif path == "/api/wechat/qr": + self._send_json(200, service.qr(instance)) elif path == "/api/wechat/groups": self._send_json(200, {"groups": service.groups(instance)}) elif path == "/api/wechat/contacts": diff --git a/coding_bridge/channels/portal_html.py b/coding_bridge/channels/portal_html.py index 64b7355..cab5e2d 100644 --- a/coding_bridge/channels/portal_html.py +++ b/coding_bridge/channels/portal_html.py @@ -143,16 +143,26 @@ } let STATE={instances:[],idx:0,contacts:new Map()}; +let qrPollTimer=null; +function stopQrPoll(){ if(qrPollTimer){ clearInterval(qrPollTimer); qrPollTimer=null; } } function current(){return STATE.instances[STATE.idx];} async function load(){ + stopQrPoll(); const cfg=await api("/api/config"); STATE.instances=cfg.instances; STATE.idx=0; $("#cfgpath").textContent=cfg.config_path; renderInstSelect(); if(!STATE.instances.length){ $("#app").innerHTML=""; $("#app").append(noInstances()); $("#bar").style.display="none"; return; } $("#bar").style.display="block"; render(); - loadAccount(); loadGroups(); + loadAccount(); loadGroups(); warmContacts(); +} + +function warmContacts(){ + const it=current(); if(!it||!it.token_resolvable) return; + // Prime the server-side contact cache in the background so the first search + // is fast instead of blocking ~seconds on a cold 4k-row fetch. + api("/api/wechat/contacts?instance="+encodeURIComponent(it.instance_id)+"&q=&limit=1").catch(()=>{}); } function noInstances(){ @@ -164,13 +174,14 @@ function renderInstSelect(){ const w=$("#instwrap"); w.innerHTML=""; if(STATE.instances.length<=1) return; - const sel=el("select",{id:"inst",onchange:e=>{STATE.idx=+e.target.value;render();loadAccount();loadGroups();}}); + const sel=el("select",{id:"inst",onchange:e=>{stopQrPoll();STATE.idx=+e.target.value;render();loadAccount();loadGroups();warmContacts();}}); STATE.instances.forEach((it,i)=>sel.append(el("option",{value:i},it.instance_id))); w.append(sel); } function render(){ const it=current(); const app=$("#app"); app.innerHTML=""; + app.append(el("div",{id:"onboarding"})); // account card const acct=el("div",{class:"card"}, @@ -254,7 +265,8 @@ async function doSearch(q){ const it=current(); const box=$("#results"); - if(!it.token_resolvable){ box.innerHTML=""; box.append(el("div",{class:"opt muted"},"token not set — cannot load contacts")); box.classList.add("show"); return; } + if(!it.token_resolvable){ box.replaceChildren(el("div",{class:"opt muted"},"token not set — cannot load contacts")); box.classList.add("show"); return; } + box.replaceChildren(el("div",{class:"opt muted"},"searching…")); box.classList.add("show"); try{ const d=await api("/api/wechat/contacts?instance="+encodeURIComponent(it.instance_id)+"&q="+encodeURIComponent(q)+"&limit=25"); box.innerHTML=""; @@ -277,16 +289,52 @@ } async function loadAccount(){ - const it=current(); if(!it||!it.token_resolvable) return; + const it=current(); const onb=$("#onboarding"); if(!it) return; + if(!it.token_resolvable){ if(onb) onb.innerHTML=""; return; } + let st=null; + try{ st=await api("/api/wechat/status?instance="+encodeURIComponent(it.instance_id)); } + catch(e){ if(onb) onb.innerHTML=""; return; } + if(st && st.logged_in===false){ renderOnboarding(it, st); return; } + stopQrPoll(); if(onb) onb.innerHTML=""; try{ const a=await api("/api/wechat/account?instance="+encodeURIComponent(it.instance_id)); - const line=$("#acctline"); const acct=$("#acct"); + const acct=$("#acct"); if(acct){ acct.replaceChildren(avatar(a.avatar_url,a.nickname||it.instance_id), el("div",{}, el("div",{style:"font-weight:650"}, a.nickname||a.wechat_id||it.instance_id), el("div",{class:"pill"}, el("span",{class:"dot"+((a.status==="online")?"":" off")}), (a.wechat_id||"")+" · "+it.base_url))); } }catch(e){ /* leave the fallback header */ } } +function renderOnboarding(it, st){ + const onb=$("#onboarding"); if(!onb) return; + onb.innerHTML=""; + onb.append(el("div",{class:"card",style:"text-align:center"}, + el("h2",{style:"text-align:left"},"Connect WeChat"), + el("p",{class:"hint",style:"text-align:left"},"This account is signed out"+((st&&st.page)?(" ("+st.page+")"):"")+". Scan the QR below with the WeChat app to sign in."), + el("div",{id:"qrbox",style:"margin:10px auto"}, el("div",{class:"empty"},"loading QR…")), + el("button",{class:"btn ghost",onclick:()=>loadQR(it)},"Refresh QR"))); + loadQR(it); + stopQrPoll(); + qrPollTimer=setInterval(async ()=>{ + try{ const s=await api("/api/wechat/status?instance="+encodeURIComponent(it.instance_id)); + if(s && s.logged_in){ stopQrPoll(); toast("WeChat connected"); load(); } }catch(e){} + }, 3000); +} + +async function loadQR(it){ + const box=$("#qrbox"); if(!box) return; + box.replaceChildren(el("div",{class:"empty"},"loading QR…")); + try{ + const d=await api("/api/wechat/qr?instance="+encodeURIComponent(it.instance_id)); + box.replaceChildren(el("img",{src:"data:image/png;base64,"+d.base64,alt:"WeChat login QR", + style:"width:220px;height:220px;border-radius:12px;border:1px solid var(--border)"})); + }catch(e){ + const msg=String(e.message||""); + if(msg.indexOf("already logged in")>=0){ stopQrPoll(); load(); return; } + box.replaceChildren(el("div",{class:"empty"},"QR error: "+msg)); + } +} + async function loadGroups(){ const it=current(); const box=$("#groups"); if(!box) return; if(!it.token_resolvable){ box.innerHTML=""; box.append(el("div",{class:"empty"},"token not set")); return; } diff --git a/tests/test_channels_portal.py b/tests/test_channels_portal.py index 39ceabf..e894f87 100644 --- a/tests/test_channels_portal.py +++ b/tests/test_channels_portal.py @@ -87,6 +87,20 @@ def handler(request: httpx.Request) -> httpx.Response: return httpx.Response(200, json={"total": 2, "contacts": []}) if p == "/api/auth/status": return httpx.Response(200, json={"logged_in": True}) + if p == "/api/auth/qr": + return httpx.Response( + 202, + json={ + "id": "t1", + "kind": "auth.qr", + "status": "completed", + "result": {"base64": "QRDATA"}, + }, + ) + if p.startswith("/api/tasks/"): + return httpx.Response( + 200, json={"id": "t1", "status": "completed", "result": {"base64": "QRDATA"}} + ) return httpx.Response(404, json={"error": "nope"}) return httpx.MockTransport(handler), calls @@ -254,6 +268,53 @@ def test_account_and_groups(tmp_path, monkeypatch): svc.close() +def test_qr_returns_base64_inline_result(tmp_path, monkeypatch): + svc, _ = _service(tmp_path, monkeypatch) + assert svc.qr("beijing")["base64"] == "QRDATA" + svc.close() + + +def test_qr_polls_until_result(tmp_path, monkeypatch): + monkeypatch.setenv("WT", "gw-secret") + s = _settings(tmp_path) + (tmp_path / "channels.toml").write_text( + dump_channels_toml(ChannelsConfig(wechat=(_inst(),))), encoding="utf-8" + ) + + def handler(request: httpx.Request) -> httpx.Response: + p = request.url.path + if p == "/api/auth/qr": + return httpx.Response(202, json={"id": "t2", "status": "queued", "result": None}) + if p == "/api/tasks/t2": + return httpx.Response( + 200, json={"id": "t2", "status": "completed", "result": {"base64": "POLLED"}} + ) + return httpx.Response(404, json={}) + + svc = PortalService(s, client=httpx.Client(transport=httpx.MockTransport(handler))) + assert svc.qr("beijing")["base64"] == "POLLED" + svc.close() + + +def test_qr_already_logged_in_is_409(tmp_path, monkeypatch): + monkeypatch.setenv("WT", "gw-secret") + s = _settings(tmp_path) + (tmp_path / "channels.toml").write_text( + dump_channels_toml(ChannelsConfig(wechat=(_inst(),))), encoding="utf-8" + ) + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/auth/qr": + return httpx.Response(409, json={"error": "already logged in"}) + return httpx.Response(404, json={}) + + svc = PortalService(s, client=httpx.Client(transport=httpx.MockTransport(handler))) + with pytest.raises(PortalError) as ei: + svc.qr("beijing") + assert ei.value.status == 409 + svc.close() + + def test_contacts_search_and_cache(tmp_path, monkeypatch): svc, calls = _service(tmp_path, monkeypatch) hits = svc.search_contacts("beijing", "崔", 10)