diff --git a/gateway/platforms/wecom.py b/gateway/platforms/wecom.py index 8cfc5c2c6..aced2bb1e 100644 --- a/gateway/platforms/wecom.py +++ b/gateway/platforms/wecom.py @@ -1464,3 +1464,122 @@ class WeComAdapter(BasePlatformAdapter): "name": chat_id, "type": "group" if chat_id and chat_id.lower().startswith("group") else "dm", } + + +# ------------------------------------------------------------------ +# QR code scan flow for obtaining bot credentials +# ------------------------------------------------------------------ + +_QR_GENERATE_URL = "https://work.weixin.qq.com/ai/qc/generate" +_QR_QUERY_URL = "https://work.weixin.qq.com/ai/qc/query_result" +_QR_CODE_PAGE = "https://work.weixin.qq.com/ai/qc/gen?source=hermes&scode=" +_QR_POLL_INTERVAL = 3 # seconds +_QR_POLL_TIMEOUT = 300 # 5 minutes + + +def qr_scan_for_bot_info( + *, + timeout_seconds: int = _QR_POLL_TIMEOUT, +) -> Optional[Dict[str, str]]: + """Run the WeCom QR scan flow to obtain bot_id and secret. + + Fetches a QR code from WeCom, renders it in the terminal, and polls + until the user scans it or the timeout expires. + + Returns ``{"bot_id": ..., "secret": ...}`` on success, ``None`` on + failure or timeout. + """ + try: + import urllib.request + import urllib.parse + except ImportError: # pragma: no cover + logger.error("urllib is required for WeCom QR scan") + return None + + generate_url = f"{_QR_GENERATE_URL}?source=hermes" + + # ── Step 1: Fetch QR code ── + print(" Connecting to WeCom...", end="", flush=True) + try: + req = urllib.request.Request(generate_url, headers={"User-Agent": "HermesAgent/1.0"}) + with urllib.request.urlopen(req, timeout=15) as resp: + raw = json.loads(resp.read().decode("utf-8")) + except Exception as exc: + logger.error("WeCom QR: failed to fetch QR code: %s", exc) + print(f" failed: {exc}") + return None + + data = raw.get("data") or {} + scode = str(data.get("scode") or "").strip() + auth_url = str(data.get("auth_url") or "").strip() + + if not scode or not auth_url: + logger.error("WeCom QR: unexpected response format: %s", raw) + print(" failed: unexpected response format") + return None + + print(" done.") + + # ── Step 2: Render QR code in terminal ── + print() + qr_rendered = False + try: + import qrcode as _qrcode + qr = _qrcode.QRCode() + qr.add_data(auth_url) + qr.make(fit=True) + qr.print_ascii(invert=True) + qr_rendered = True + except ImportError: + pass + except Exception: + pass + + page_url = f"{_QR_CODE_PAGE}{urllib.parse.quote(scode)}" + if qr_rendered: + print(f"\n Scan the QR code above, or open this URL directly:\n {page_url}") + else: + print(f" Open this URL in WeCom on your phone:\n\n {page_url}\n") + print(" Tip: pip install qrcode to display a scannable QR code here next time") + print() + print(" Fetching configuration results...", end="", flush=True) + + # ── Step 3: Poll for result ── + import time + deadline = time.time() + timeout_seconds + query_url = f"{_QR_QUERY_URL}?scode={urllib.parse.quote(scode)}" + poll_count = 0 + + while time.time() < deadline: + try: + req = urllib.request.Request(query_url, headers={"User-Agent": "HermesAgent/1.0"}) + with urllib.request.urlopen(req, timeout=10) as resp: + result = json.loads(resp.read().decode("utf-8")) + except Exception as exc: + logger.debug("WeCom QR poll error: %s", exc) + time.sleep(_QR_POLL_INTERVAL) + continue + + poll_count += 1 + if poll_count % 6 == 0: + print(".", end="", flush=True) + + result_data = result.get("data") or {} + status = str(result_data.get("status") or "").lower() + + if status == "success": + print() # newline after "Fetching configuration results..." dots + bot_info = result_data.get("bot_info") or {} + bot_id = str(bot_info.get("botid") or bot_info.get("bot_id") or "").strip() + secret = str(bot_info.get("secret") or "").strip() + if bot_id and secret: + return {"bot_id": bot_id, "secret": secret} + logger.warning("WeCom QR: success but missing bot_info: %s", result_data) + print(" QR scan succeeded but bot info was not returned") + return None + + time.sleep(_QR_POLL_INTERVAL) + + print() # newline after dots + print(f" QR scan timed out ({timeout_seconds // 60} minutes). Please try again.") + return None diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index f7c9cfff8..481566f9d 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -2639,9 +2639,120 @@ def _setup_dingtalk(): def _setup_wecom(): - """Configure WeCom (Enterprise WeChat) via the standard platform setup.""" - wecom_platform = next(p for p in _PLATFORMS if p["key"] == "wecom") - _setup_standard_platform(wecom_platform) + """Interactive setup for WeCom — scan QR code or manual credential input.""" + print() + print(color(" ─── 💬 WeCom (Enterprise WeChat) Setup ───", Colors.CYAN)) + + existing_bot_id = get_env_value("WECOM_BOT_ID") + existing_secret = get_env_value("WECOM_SECRET") + if existing_bot_id and existing_secret: + print() + print_success("WeCom is already configured.") + if not prompt_yes_no(" Reconfigure WeCom?", False): + return + + # ── Choose setup method ── + print() + method_choices = [ + "Scan QR code to obtain Bot ID and Secret automatically (recommended)", + "Enter existing Bot ID and Secret manually", + ] + method_idx = prompt_choice(" How would you like to set up WeCom?", method_choices, 0) + + bot_id = None + secret = None + + if method_idx == 0: + # ── QR scan flow ── + try: + from gateway.platforms.wecom import qr_scan_for_bot_info + except Exception as exc: + print_error(f" WeCom QR scan import failed: {exc}") + qr_scan_for_bot_info = None + + if qr_scan_for_bot_info is not None: + try: + credentials = qr_scan_for_bot_info() + except KeyboardInterrupt: + print() + print_warning(" WeCom setup cancelled.") + return + except Exception as exc: + print_warning(f" QR scan failed: {exc}") + credentials = None + if credentials: + bot_id = credentials.get("bot_id", "") + secret = credentials.get("secret", "") + print_success(" ✔ QR scan successful! Bot ID and Secret obtained.") + + if not bot_id or not secret: + print_info(" QR scan did not complete. Continuing with manual input.") + bot_id = None + secret = None + + # ── Manual credential input ── + if not bot_id or not secret: + print() + print_info(" 1. Go to WeCom Application → Workspace → Smart Robot -> Create smart robots") + print_info(" 2. Select API Mode") + print_info(" 3. Copy the Bot ID and Secret from the bot's credentials info") + print_info(" 4. The bot connects via WebSocket — no public endpoint needed") + print() + bot_id = prompt(" Bot ID", password=False) + if not bot_id: + print_warning(" Skipped — WeCom won't work without a Bot ID.") + return + secret = prompt(" Secret", password=True) + if not secret: + print_warning(" Skipped — WeCom won't work without a Secret.") + return + + # ── Save core credentials ── + save_env_value("WECOM_BOT_ID", bot_id) + save_env_value("WECOM_SECRET", secret) + + # ── Allowed users (deny-by-default security) ── + print() + print_info(" The gateway DENIES all users by default for security.") + print_info(" Enter user IDs to create an allowlist, or leave empty.") + allowed = prompt(" Allowed user IDs (comma-separated, or empty)", password=False) + if allowed: + cleaned = allowed.replace(" ", "") + save_env_value("WECOM_ALLOWED_USERS", cleaned) + print_success(" Saved — only these users can interact with the bot.") + else: + print() + access_choices = [ + "Enable open access (anyone can message the bot)", + "Use DM pairing (unknown users request access, you approve with 'hermes pairing approve')", + "Disable direct messages", + "Skip for now (bot will deny all users until configured)", + ] + access_idx = prompt_choice(" How should unauthorized users be handled?", access_choices, 1) + if access_idx == 0: + save_env_value("WECOM_DM_POLICY", "open") + save_env_value("GATEWAY_ALLOW_ALL_USERS", "true") + print_warning(" Open access enabled — anyone can use your bot!") + elif access_idx == 1: + save_env_value("WECOM_DM_POLICY", "pairing") + print_success(" DM pairing mode — users will receive a code to request access.") + print_info(" Approve with: hermes pairing approve ") + elif access_idx == 2: + save_env_value("WECOM_DM_POLICY", "disabled") + print_warning(" Direct messages disabled.") + else: + print_info(" Skipped — configure later with 'hermes gateway setup'") + + # ── Home channel (optional) ── + print() + print_info(" Chat ID for scheduled results and notifications.") + home = prompt(" Home chat ID (optional, for cron/notifications)", password=False) + if home: + save_env_value("WECOM_HOME_CHANNEL", home) + print_success(f" Home channel set to {home}") + + print() + print_success("💬 WeCom configured!") def _is_service_installed() -> bool: @@ -3390,6 +3501,8 @@ def gateway_setup(): _setup_feishu() elif platform["key"] == "qqbot": _setup_qqbot() + elif platform["key"] == "wecom": + _setup_wecom() else: _setup_standard_platform(platform)