feat: expose Douyin login QR in admin

cmd/docker_gateway/douyin.py

@@ -27,6 +27,10 @@ from .docker_client import RUNTIME_ID_RE
 LOG = logging.getLogger("creatorhub.douyin")
 ORIGIN = "https://www.douyin.com"
 ORIGIN_URL = ORIGIN + "/"
+LOGIN_ORIGINS = frozenset(
+    {ORIGIN, "https://sso.douyin.com", "https://verify.snssdk.com", "https://verify.bytedance.com"}
+)
+LOGIN_SCREENSHOT_LIMIT = 8 << 20
 IDENTITY_URL = (
     ORIGIN + "/aweme/v1/web/user/profile/self/?aid=6383&device_platform=webapp"
 )
@@ -74,6 +78,13 @@ class BrowserMediaResponse:
     body_base64: str
 
 
+@dataclass(frozen=True)
+class BrowserLoginQRResponse:
+    content_type: str
+    body_base64: str
+    qr_detected: bool
+
+
 class CDPConnection:
     def __init__(self, socket: websocket.WebSocket) -> None:
         self.socket = socket
@@ -385,6 +396,83 @@ class DouyinBrowser:
             raise DouyinError("restricted browser fetch returned invalid body")
         return BrowserResponse(status, body, detect_challenge(status, body))
 
+    def login_qr(self, alias: str) -> BrowserLoginQRResponse:
+        with self.connection(alias) as cdp:
+            navigation = cdp.command("Page.navigate", {"url": ORIGIN_URL})
+            if (
+                not isinstance(navigation, dict)
+                or not isinstance(navigation.get("frameId"), str)
+                or navigation.get("errorText")
+            ):
+                raise DouyinError("Douyin login page navigation failed")
+            time.sleep(0.5)
+            opened = cdp.evaluate(
+                """(() => {
+                  const text = value => String(value || '').replace(/\\s+/g, '');
+                  const candidate = [...document.querySelectorAll('button,a,[role="button"]')]
+                    .find(element => /登录|扫码登录/.test(text(element.innerText || element.getAttribute('aria-label'))));
+                  if (!candidate) return false;
+                  candidate.click();
+                  return true;
+                })()"""
+            )
+            if isinstance(opened, bool) and opened:
+                time.sleep(0.5)
+            page = cdp.evaluate(
+                """(() => {
+                  const visible = element => {
+                    const rect = element.getBoundingClientRect();
+                    const style = getComputedStyle(element);
+                    return rect.width >= 120 && rect.height >= 120 &&
+                      style.visibility !== 'hidden' && style.display !== 'none';
+                  };
+                  const qrElement = [...document.querySelectorAll('img,canvas')].find(element => {
+                    if (!visible(element)) return false;
+                    const rect = element.getBoundingClientRect();
+                    const label = `${element.alt || ''} ${element.title || ''} ${element.getAttribute('aria-label') || ''}`;
+                    return /二维码|qr.?code/i.test(label) ||
+                      (element.tagName === 'CANVAS' && Math.abs(rect.width - rect.height) < 24);
+                  });
+                  if (!qrElement) return {origin: location.origin, qr_detected: false, clip: null};
+                  const rect = qrElement.getBoundingClientRect();
+                  const padding = 16;
+                  const x = Math.max(0, Math.min(rect.x - padding, innerWidth - 1));
+                  const y = Math.max(0, Math.min(rect.y - padding, innerHeight - 1));
+                  return {
+                    origin: location.origin,
+                    qr_detected: true,
+                    clip: {
+                      x, y,
+                      width: Math.min(innerWidth - x, rect.width + padding * 2),
+                      height: Math.min(innerHeight - y, rect.height + padding * 2),
+                      scale: 1,
+                    },
+                  };
+                })()"""
+            )
+            if (
+                not isinstance(page, dict)
+                or page.get("origin") not in LOGIN_ORIGINS
+                or not isinstance(page.get("qr_detected"), bool)
+            ):
+                raise DouyinError("Douyin login page origin is not allowed")
+            screenshot_params = {"format": "png", "fromSurface": True}
+            if isinstance(page.get("clip"), dict):
+                screenshot_params["clip"] = page["clip"]
+            screenshot = cdp.command("Page.captureScreenshot", screenshot_params)
+        if not isinstance(screenshot, dict):
+            raise DouyinError("Douyin login screenshot is invalid")
+        body = screenshot.get("data")
+        if not isinstance(body, str) or not body:
+            raise DouyinError("Douyin login screenshot is invalid")
+        try:
+            decoded_size = len(base64.b64decode(body, validate=True))
+        except (ValueError, binascii.Error) as exc:
+            raise DouyinError("Douyin login screenshot is invalid") from exc
+        if decoded_size > LOGIN_SCREENSHOT_LIMIT:
+            raise DouyinError("Douyin login screenshot is too large")
+        return BrowserLoginQRResponse("image/png", body, bool(page["qr_detected"]))
+
     def get_media(self, alias: str, target: str) -> BrowserMediaResponse:
         if not self.media_validator(target):
             raise DouyinError("restricted browser media target is invalid")

cmd/docker_gateway/gateway.py

@@ -700,6 +700,27 @@ class Gateway:
                 )
         return identity
 
+    def douyin_login_qr(self, alias: str, input: dict) -> dict:
+        if not valid_douyin_generation(input):
+            raise RequestError("invalid Douyin login QR request", 400)
+        with self._alias_lock(alias):
+            self._require_douyin_generation(alias, input)
+            try:
+                screen = self.browser.login_qr(alias)
+                self._require_douyin_generation(alias, input)
+            except DouyinError as exc:
+                LOG.warning(
+                    "Douyin login screen capture failed alias=%s reason=%s",
+                    alias,
+                    str(exc),
+                )
+                raise RequestError("Douyin login screen could not be captured") from exc
+        return {
+            "content_type": screen.content_type,
+            "body_base64": screen.body_base64,
+            "qr_detected": screen.qr_detected,
+        }
+
     def get_xiaohongshu(self, alias: str, input: dict) -> dict:
         target = input.get("url", "")
         if not valid_xiaohongshu_generation(input) or not valid_xiaohongshu_url(target):
@@ -1402,7 +1423,7 @@ class GatewayHandler(BaseHTTPRequestHandler):
             if action == "identity" and method == "POST":
                 return gateway.xiaohongshu_identity(alias, body)
         match = re.fullmatch(
-            r"/v1/browsers/([a-z0-9][a-z0-9-]{0,31})/douyin/(get|media|identity|action|messages|events)",
+            r"/v1/browsers/([a-z0-9][a-z0-9-]{0,31})/douyin/(get|media|identity|login-qr|action|messages|events)",
             path,
         )
         if match:
@@ -1413,6 +1434,8 @@ class GatewayHandler(BaseHTTPRequestHandler):
                 return gateway.get_douyin_media(alias, body)
             if action == "identity" and method == "POST":
                 return gateway.douyin_identity(alias, body)
+            if action == "login-qr" and method == "POST":
+                return gateway.douyin_login_qr(alias, body)
             if action == "action" and method == "POST":
                 return gateway.douyin_action(alias, body)
             if action == "messages" and method == "POST":

cmd/docker_gateway/test_gateway.py

@@ -267,6 +267,11 @@ class GatewayValidationTests(unittest.TestCase):
         gateway = Mock()
         gateway.list_browsers.return_value = []
         gateway.douyin_identity.return_value = {"uid": "123"}
+        gateway.douyin_login_qr.return_value = {
+            "content_type": "image/png",
+            "body_base64": "cG5n",
+            "qr_detected": True,
+        }
         gateway.douyin_action.return_value = {"status": "succeeded"}
         gateway.douyin_message_history.return_value = {"status": "succeeded"}
         gateway.poll_douyin_events.return_value = []
@@ -295,6 +300,11 @@ class GatewayValidationTests(unittest.TestCase):
             handler._route("POST", "/v1/browsers/safe/douyin/cookies", {}, {})
         handler._route("POST", "/v1/browsers/safe/douyin/get", {}, {})
         handler._route("POST", "/v1/browsers/safe/douyin/identity", {}, {})
+        self.assertEqual(
+            handler._route("POST", "/v1/browsers/safe/douyin/login-qr", {}, {}),
+            gateway.douyin_login_qr.return_value,
+        )
+        gateway.douyin_login_qr.assert_called_once_with("safe", {})
         handler._route("POST", "/v1/browsers/safe/douyin/action", {}, {})
         self.assertEqual(
             handler._route("POST", "/v1/browsers/safe/douyin/messages", {}, {}),
@@ -719,6 +729,30 @@ class BrowserTests(unittest.TestCase):
         self.assertEqual(response.status, 200)
         self.assertNotIn("Network.setCookies", [method for method, _ in cdp.commands])
 
+    def test_login_qr_captures_a_browser_screen_without_credentials(self) -> None:
+        screenshot = base64.b64encode(b"png-bytes").decode("ascii")
+        cdp = Mock()
+        cdp.evaluate.side_effect = [
+            True,
+            {"origin": "https://www.douyin.com", "qr_detected": True},
+        ]
+        cdp.command.side_effect = [
+            {"frameId": "frame-1"},
+            {"data": screenshot},
+        ]
+        browser = DouyinBrowser()
+        self._with_connection(browser, cast(BrowserCDP, cdp))
+        with patch.object(douyin_module.time, "sleep"):
+            response = browser.login_qr("safe")
+        self.assertEqual(response.content_type, "image/png")
+        self.assertEqual(response.body_base64, screenshot)
+        self.assertTrue(response.qr_detected)
+        self.assertEqual(
+            [call.args[0] for call in cdp.command.call_args_list],
+            ["Page.navigate", "Page.captureScreenshot"],
+        )
+        self.assertFalse(any("cookie" in expression.lower() for expression in cdp.evaluate.call_args.args))
+
     def test_connect_selects_configured_target_from_large_mixed_list(self) -> None:
         targets = [
             {"type": "service", "url": "http://127.0.0.1:9222/json"} for _ in range(40)