diff --git a/cmd/control-plane/creator.go b/cmd/control-plane/creator.go index a013ce9..d9cfadd 100644 --- a/cmd/control-plane/creator.go +++ b/cmd/control-plane/creator.go @@ -743,6 +743,66 @@ func registerCreatorWithServices(app *fiber.App, store *creator.Store, phaseASto } return c.JSON(items) }) + app.Post("/api/creator/conversations/:id/sync", func(c fiber.Ctx) error { + conversation, err := store.GetConversation(c.Context(), c.Params("id")) + if err != nil { + return creatorError(c, err) + } + if conversation.Platform != creator.PlatformDouyin { + return creatorError(c, creator.ErrUnavailable) + } + limit := 100 + if value := c.Query("limit"); value != "" { + limit, err = strconv.Atoi(value) + if err != nil || limit < 1 || limit > 200 { + return creatorError(c, creator.ErrInvalid) + } + } + profile, err := store.GetAccountProfile(c.Context(), conversation.AccountID) + if err != nil { + return creatorError(c, err) + } + account, err := phaseAStore.GetAccount(c.Context(), conversation.AccountID) + if err != nil { + return creatorError(c, err) + } + if account.Platform != creator.PlatformDouyin || account.AuthorizationStatus != "authorized" || + profile.Platform != creator.PlatformDouyin || profile.BusinessStatus != "normal" || profile.LoginStatus != "logged_in" || + profile.PlatformAccountKey == "" || account.PlatformAccountKey != profile.PlatformAccountKey { + return creatorError(c, creator.ErrConflict) + } + environment, err := hubStore.GetEnvironmentContextForAccount(c.Context(), conversation.AccountID) + if err != nil { + return creatorError(c, err) + } + gateway, err := hubStore.GetGateway(c.Context(), environment.Gateway) + if err != nil { + return creatorError(c, err) + } + browser := creatorGatewayBrowser{gateway: gateway, environment: environment} + accountUID, err := browser.Identity(c.Context(), profile.PlatformAccountKey) + if err != nil { + return creatorError(c, fmt.Errorf("%w: account identity verification failed: %v", creator.ErrConflict, err)) + } + history, err := browser.MessageHistory(c.Context(), accountUID, conversation.PeerUID, limit) + if err != nil { + return creatorError(c, err) + } + if history.AccountUID != accountUID { + return creatorError(c, creator.ErrConflict) + } + inserted, err := persistDouyinMessageHistory(c.Context(), store, conversation, accountUID, history.Messages) + if err != nil { + return creatorError(c, err) + } + creatorUpdates.publish() + return c.JSON(map[string]any{ + "conversation_id": conversation.ID, + "messages": inserted, + "history_source": history.HistorySource, + "history_has_more": history.HistoryHasMore, + }) + }) app.Post("/api/creator/messages/send", func(c fiber.Ctx) error { var input creator.MessageInput if err := decodeCreator(c, &input); err != nil { @@ -795,6 +855,69 @@ func registerCreatorWithServices(app *fiber.App, store *creator.Store, phaseASto }) } +func persistDouyinMessageHistory(ctx context.Context, store *creator.Store, conversation creator.Conversation, accountUID string, messages []douyinHistoryMessage) (int, error) { + if store == nil || conversation.Platform != creator.PlatformDouyin || strings.TrimSpace(accountUID) == "" || strings.TrimSpace(conversation.PeerUID) == "" { + return 0, creator.ErrInvalid + } + inserted := 0 + for _, item := range messages { + if strings.TrimSpace(item.ServerID) == "" || strings.TrimSpace(item.SenderUID) == "" { + return inserted, creator.ErrInvalid + } + messageType := creator.MessageTypeUnknown + text := "" + if len(item.Content) > 0 && string(item.Content) != "null" { + var payload struct { + Text string `json:"text"` + } + var encoded string + if err := json.Unmarshal(item.Content, &encoded); err == nil { + if err := json.Unmarshal([]byte(encoded), &payload); err != nil { + return inserted, creator.ErrInvalid + } + } else if err := json.Unmarshal(item.Content, &payload); err != nil { + return inserted, creator.ErrInvalid + } + text = payload.Text + if text != "" { + messageType = creator.MessageTypeText + } + } + var messageAt *time.Time + if item.CreatedAt != "" { + milliseconds, err := strconv.ParseInt(item.CreatedAt, 10, 64) + if err != nil || milliseconds <= 0 { + return inserted, creator.ErrInvalid + } + value := time.UnixMilli(milliseconds).UTC() + messageAt = &value + } + direction, state := "inbound", "received" + if item.SenderUID == accountUID { + direction, state = "outbound", "succeeded" + } + _, wasInserted, err := store.SaveMessage(ctx, creator.MessageInput{ + Platform: creator.PlatformDouyin, + AccountID: conversation.AccountID, + PeerUID: conversation.PeerUID, + PeerName: conversation.PeerName, + PlatformMessageKey: item.ServerID, + Direction: direction, + MessageType: messageType, + Text: text, + SentState: state, + MessageAt: messageAt, + }) + if err != nil { + return inserted, err + } + if wasInserted { + inserted++ + } + } + return inserted, nil +} + func setStrategyEnabled(c fiber.Ctx, store *creator.Store, enabled bool) error { item, err := store.SetStrategyEnabled(c.Context(), c.Params("id"), enabled) if err != nil { @@ -916,6 +1039,9 @@ func (executor creatorGatewayActionExecutor) Execute(ctx context.Context, reques // the verified platform keys and the comment's owning work key. if request.TargetCommentID != "" { comment, targetErr := executor.store.GetComment(ctx, request.TargetCommentID) + if errors.Is(targetErr, creator.ErrNotFound) { + comment, targetErr = executor.store.GetCommentByKey(ctx, request.Platform, request.TargetCommentID) + } if targetErr != nil { return creator.ActionResult{}, targetErr } @@ -926,6 +1052,9 @@ func (executor creatorGatewayActionExecutor) Execute(ctx context.Context, reques } if request.TargetWorkID != "" { work, targetErr := executor.store.GetWork(ctx, request.TargetWorkID) + if errors.Is(targetErr, creator.ErrNotFound) { + work, targetErr = executor.store.GetWorkByKey(ctx, request.Platform, request.TargetWorkID) + } if targetErr != nil { return creator.ActionResult{}, targetErr } @@ -1012,6 +1141,25 @@ func (browser creatorGatewayBrowser) Identity(ctx context.Context, expectedKey s return identity.UID, nil } +func (browser creatorGatewayBrowser) MessageHistory(ctx context.Context, expectedUID, targetUID string, limit int) (douyinMessageHistory, error) { + if strings.TrimSpace(expectedUID) == "" || strings.TrimSpace(targetUID) == "" || limit < 1 || limit > 200 { + return douyinMessageHistory{}, errors.New("invalid message history request") + } + payload := gatewayGenerationPayload(browser.environment) + payload["expected_uid"] = expectedUID + payload["target_uid"] = targetUID + payload["limit"] = limit + status, body, err := gatewayCall(ctx, browser.gateway, http.MethodPost, "/v1/browsers/"+url.PathEscape(browser.environment.Alias)+"/douyin/messages", payload, 30*time.Second) + if err != nil || status != http.StatusOK { + return douyinMessageHistory{}, errors.New("restricted message history operation failed") + } + var response douyinMessageHistory + if json.Unmarshal(body, &response) != nil || response.Status != "succeeded" || response.HistorySource == "" { + return douyinMessageHistory{}, errors.New("restricted message history response is invalid") + } + return response, nil +} + func verifyCreatorAccount(ctx context.Context, store *creator.Store, phaseAStore *phasea.Store, hubStore *hub.Store, accountID string) (creator.LoginResult, error) { if store == nil || phaseAStore == nil || hubStore == nil || strings.TrimSpace(accountID) == "" { return creator.LoginResult{}, creator.ErrUnavailable diff --git a/cmd/control-plane/creator_events.go b/cmd/control-plane/creator_events.go index 163a57d..abb78e5 100644 --- a/cmd/control-plane/creator_events.go +++ b/cmd/control-plane/creator_events.go @@ -19,13 +19,14 @@ import ( const creatorEventReconcileInterval = 10 * time.Second type creatorGatewayEvent struct { - DeliveryID string `json:"delivery_id,omitempty"` - Kind string `json:"kind"` - Reason string `json:"reason,omitempty"` - Continuity string `json:"continuity,omitempty"` - BoundaryAt string `json:"boundary_at,omitempty"` - Baseline bool `json:"baseline,omitempty"` - Notice *creatorGatewayEventNotice `json:"notice,omitempty"` + DeliveryID string `json:"delivery_id,omitempty"` + Kind string `json:"kind"` + Reason string `json:"reason,omitempty"` + Continuity string `json:"continuity,omitempty"` + BoundaryAt string `json:"boundary_at,omitempty"` + BoundarySource string `json:"boundary_source,omitempty"` + Baseline bool `json:"baseline,omitempty"` + Notice *creatorGatewayEventNotice `json:"notice,omitempty"` } type creatorGatewayEventNotice struct { @@ -290,23 +291,9 @@ func runCreatorEventListener(ctx context.Context, store *creator.Store, binding } for _, event := range events { if event.Kind == "baseline" { - if event.BoundaryAt != "" { - parsedBoundary, boundaryErr := time.Parse(time.RFC3339Nano, event.BoundaryAt) - if boundaryErr != nil { - logrus.WithError(boundaryErr).WithField("account_id", binding.accountID).Warn("creator event baseline timestamp is invalid") - ready = false - boundaryAt = time.Time{} - } else { - boundaryAt = parsedBoundary.UTC() - ready = true - } - } else { - // A marker without a platform-verified boundary cannot prove - // continuity. Keep all timestamped notices non-actionable. - ready = false - boundaryAt = time.Time{} - } - status, reason := "gap", "平台边界无效" + parsedBoundary, boundaryReady, reason := creatorGatewayBoundary(event) + boundaryAt, ready = parsedBoundary, boundaryReady + status := "gap" if ready { status, reason = "ready", "" } @@ -474,6 +461,20 @@ func handleCreatorGatewayEvent(ctx context.Context, store *creator.Store, bindin }() } +func creatorGatewayBoundary(event creatorGatewayEvent) (time.Time, bool, string) { + if event.BoundarySource != "douyin_identity_extra_now" { + return time.Time{}, false, "平台边界来源未验证" + } + if event.BoundaryAt == "" { + return time.Time{}, false, "平台边界无效" + } + parsedBoundary, err := time.Parse(time.RFC3339Nano, event.BoundaryAt) + if err != nil { + return time.Time{}, false, "平台边界无效" + } + return parsedBoundary.UTC(), true, "" +} + func creatorEventBeforeBoundary(event creatorGatewayEvent, boundaryAt time.Time) bool { if boundaryAt.IsZero() || event.Notice == nil { return false diff --git a/cmd/control-plane/creator_events_test.go b/cmd/control-plane/creator_events_test.go index d24a078..93af0fa 100644 --- a/cmd/control-plane/creator_events_test.go +++ b/cmd/control-plane/creator_events_test.go @@ -52,6 +52,19 @@ func TestCreatorGatewayEventNeedsBaseline(t *testing.T) { } } +func TestCreatorGatewayBoundary(t *testing.T) { + boundary, ready, reason := creatorGatewayBoundary(creatorGatewayEvent{ + BoundaryAt: "2024-01-01T00:00:00Z", + BoundarySource: "douyin_identity_extra_now", + }) + if !ready || !boundary.Equal(time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)) || reason != "" { + t.Fatalf("unexpected valid boundary: %v %v %q", boundary, ready, reason) + } + if _, ready, reason := creatorGatewayBoundary(creatorGatewayEvent{BoundaryAt: "2024-01-01T00:00:00Z"}); ready || reason != "平台边界来源未验证" { + t.Fatalf("unverified boundary must stay blocked: %v %q", ready, reason) + } +} + func TestCreatorEventBeforeBoundary(t *testing.T) { boundary := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) before := creatorGatewayEvent{Notice: &creatorGatewayEventNotice{PlatformEventAt: "2023-12-31T23:59:59Z"}} diff --git a/cmd/control-plane/douyin.go b/cmd/control-plane/douyin.go index 8f65c6d..a52a420 100644 --- a/cmd/control-plane/douyin.go +++ b/cmd/control-plane/douyin.go @@ -49,6 +49,24 @@ func (browser douyinGatewayBrowser) Get(ctx context.Context, target string) (dou return douyin.Response{Status: response.Status, Body: []byte(response.Body), Challenge: response.Challenge}, nil } +type douyinHistoryMessage struct { + ServerID string `json:"server_id"` + SenderUID string `json:"sender_uid"` + MessageType string `json:"message_type"` + Content json.RawMessage `json:"content"` + CreatedAt string `json:"created_at"` + ServerState *int `json:"server_status"` +} + +type douyinMessageHistory struct { + Status string `json:"status"` + HistorySource string `json:"history_source"` + HistoryHasMore bool `json:"history_has_more"` + AccountUID string `json:"account_uid"` + Conversation map[string]any `json:"conversation"` + Messages []douyinHistoryMessage `json:"messages"` +} + func (browser douyinGatewayBrowser) request() (douyinGatewayRequest, error) { environment := browser.environment if browser.gateway.Endpoint == "" || browser.gateway.Token == "" || !accountRunnable(environment) || diff --git a/cmd/control-plane/main.go b/cmd/control-plane/main.go index cfc54ee..777f5c6 100644 --- a/cmd/control-plane/main.go +++ b/cmd/control-plane/main.go @@ -286,6 +286,30 @@ func newHandlerWithCreatorAndAI(webDirectory, username, password string, phaseAS c.Status(fiber.StatusNoContent) return nil }) + app.Get("/readyz", func(c fiber.Ctx) error { + if phaseAStore == nil || hubStore == nil || creatorStore == nil { + return c.Status(fiber.StatusServiceUnavailable).JSON(fiber.Map{"error": "service is not ready"}) + } + ctx, cancel := context.WithTimeout(c.Context(), 2*time.Second) + defer cancel() + checks := []struct { + name string + fn func(context.Context) error + }{ + {"phase_a", phaseAStore.Ping}, + {"hub", hubStore.Ping}, + {"creator", creatorStore.Ping}, + {"creator_schema", creatorStore.EnsureSchema}, + } + for _, check := range checks { + if err := check.fn(ctx); err != nil { + logrus.WithError(err).WithField("check", check.name).Warn("control plane readiness check failed") + return c.Status(fiber.StatusServiceUnavailable).JSON(fiber.Map{"error": "service is not ready"}) + } + } + c.Status(fiber.StatusNoContent) + return nil + }) app.Use(authenticate(username, password)) if hubStore != nil { registerHub(app, hubStore) @@ -304,10 +328,25 @@ func newHandlerWithCreatorAndAI(webDirectory, username, password string, phaseAS registerCreatorWithServices(app, creatorStore, phaseAStore, hubStore, executor, generator, analyzer) } } + app.Use(func(c fiber.Ctx) error { + if isControlPlaneAPIPath(c.Path()) { + return c.Status(fiber.StatusNotFound).JSON(fiber.Map{"error": "not found"}) + } + return c.Next() + }) app.Get("/*", spaHandler(webDirectory)) return app } +func isControlPlaneAPIPath(path string) bool { + for _, prefix := range []string{"/api", "/phase-a", "/gateways", "/browser-images", "/browsers", "/network-exits"} { + if path == prefix || strings.HasPrefix(path, prefix+"/") { + return true + } + } + return false +} + func authenticate(username, password string) fiber.Handler { wantUser, wantPassword := sha256.Sum256([]byte(username)), sha256.Sum256([]byte(password)) return func(c fiber.Ctx) error { diff --git a/cmd/control-plane/main_test.go b/cmd/control-plane/main_test.go index b129d4f..59386c4 100644 --- a/cmd/control-plane/main_test.go +++ b/cmd/control-plane/main_test.go @@ -134,6 +134,23 @@ func TestControlPlaneAuthentication(t *testing.T) { } health.Body.Close() + ready, err := app.Test(httptest.NewRequest(http.MethodGet, "/readyz", nil)) + if err != nil || ready.StatusCode != http.StatusServiceUnavailable { + t.Fatalf("readiness must fail without stores: status=%d err=%v", ready.StatusCode, err) + } + ready.Body.Close() + + unknown := httptest.NewRequest(http.MethodGet, "/api/not-registered", nil) + unknown.SetBasicAuth("operator", "unit-test-password") + unknownResponse, err := app.Test(unknown) + if err != nil || unknownResponse.StatusCode != http.StatusNotFound { + t.Fatalf("unknown API must be JSON 404: status=%d err=%v", unknownResponse.StatusCode, err) + } + if got := unknownResponse.Header.Get("Content-Type"); !strings.HasPrefix(got, "application/json") { + t.Fatalf("unknown API content type=%q", got) + } + unknownResponse.Body.Close() + for _, path := range []string{ "/", "/api/phase-a/accounts", "/api/browsers", "/api/network-exits", "/api/phase-a/tasks", "/api/phase-a/audit", } { @@ -262,8 +279,8 @@ func TestControlPlaneRegisteredRouteMatrix(t *testing.T) { method, path string want int }{ - {http.MethodPost, "/api/not-registered", http.StatusMethodNotAllowed}, - {http.MethodDelete, "/api/not-registered", http.StatusMethodNotAllowed}, + {http.MethodPost, "/api/not-registered", http.StatusNotFound}, + {http.MethodDelete, "/api/not-registered", http.StatusNotFound}, } { t.Run("unregistered "+route.method, func(t *testing.T) { if response := do(app, route.method, route.path, ""); response.Code != http.StatusUnauthorized { diff --git a/cmd/control-plane/xiaohongshu.go b/cmd/control-plane/xiaohongshu.go index c2c4875..2944844 100644 --- a/cmd/control-plane/xiaohongshu.go +++ b/cmd/control-plane/xiaohongshu.go @@ -211,10 +211,11 @@ func newCreatorCollector(ctx context.Context, platform string, gateway hub.Gatew return nil, "", err } collector := douyinCollector(browser, targetAccountKey, sourceType, sourceID) - if _, err := collector.CanonicalSecUID(ctx, viewerAccountKey); err != nil { + canonicalKey, err := collector.CanonicalTargetSecUID(ctx, targetAccountKey) + if err != nil { return nil, "", err } - return &collector, targetAccountKey, nil + return &collector, canonicalKey, nil case creator.PlatformXiaohongshu: if homepageURL != "" { if err := xiaohongshu.ValidateSourceURL(homepageURL, targetAccountKey); err != nil { diff --git a/cmd/control-plane/xiaohongshu_test.go b/cmd/control-plane/xiaohongshu_test.go index 89592a8..27d20d7 100644 --- a/cmd/control-plane/xiaohongshu_test.go +++ b/cmd/control-plane/xiaohongshu_test.go @@ -51,6 +51,34 @@ func TestXiaohongshuGatewayBrowserResolvesShareLinks(t *testing.T) { } } +func TestNewDouyinCollectorResolvesTargetIdentitySeparately(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + var body map[string]any + if json.NewDecoder(request.Body).Decode(&body) != nil { + t.Fatal("invalid gateway request") + } + switch request.URL.Path { + case "/v1/browsers/account-a/douyin/identity": + if body["expected_account_key"] != "viewer-1" { + t.Fatalf("viewer identity key=%#v", body["expected_account_key"]) + } + _ = json.NewEncoder(response).Encode(map[string]any{"uid": "viewer-1", "sec_uid": "viewer-sec", "unique_id": "viewer"}) + case "/v1/browsers/account-a/douyin/get": + if body["url"] != "https://www.douyin.com/aweme/v1/web/user/profile/other/?aid=6383&device_platform=webapp&user_id=2328120603967913" { + t.Fatalf("target profile URL=%#v", body["url"]) + } + _ = json.NewEncoder(response).Encode(map[string]any{"status": 200, "body": `{"status_code":0,"user":{"uid":"2328120603967913","sec_uid":"target-sec","unique_id":"target"}}`, "challenge": ""}) + default: + t.Fatalf("unexpected path: %s", request.URL.Path) + } + })) + defer server.Close() + collector, target, err := newCreatorCollector(context.Background(), creator.PlatformDouyin, hub.Gateway{Endpoint: server.URL, Token: "gateway-token-1"}, readyDouyinEnvironment(), "viewer-1", "2328120603967913", "", creator.SourceCompetitor, "source-1") + if err != nil || collector == nil || target != "target-sec" { + t.Fatalf("new Douyin collector: collector=%#v target=%q err=%v", collector, target, err) + } +} + func TestNewXiaohongshuCollectorKeepsViewerAndTargetSeparate(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { if request.URL.Path != "/v1/browsers/account-a/xiaohongshu/identity" { diff --git a/cmd/docker_gateway/douyin.py b/cmd/docker_gateway/douyin.py index b6a6da8..219e88f 100644 --- a/cmd/docker_gateway/douyin.py +++ b/cmd/docker_gateway/douyin.py @@ -35,6 +35,7 @@ COMMENTS_PATH = "/aweme/v1/web/comment/list/" RESPONSE_LIMIT = 1 << 20 MEDIA_RESPONSE_LIMIT = 32 << 20 CONTROL_TIMEOUT = 15.0 +MEDIA_SOURCE_WAIT_MS = 15000 UID_RE = re.compile(r"^[1-9][0-9]{0,19}$") ID_RE = re.compile(r"^[1-9][0-9]{0,63}$") ACCOUNT_KEY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,127}$") @@ -56,6 +57,7 @@ LISTENER_ERRORS = ( ValueError, websocket.WebSocketException, ) +LISTENER_START_TIMEOUT = 30.0 @dataclass(frozen=True) @@ -230,6 +232,7 @@ class DouyinBrowser: url_validator: Callable[[object], bool] | None = None, media_validator: Callable[[object], bool] | None = None, media_selector: str = "video", + target_id: str = "", ) -> None: self.endpoint = endpoint or ( lambda alias: f"http://creatorhub-browser-{alias}:9222" @@ -238,6 +241,7 @@ class DouyinBrowser: self.url_validator = url_validator or is_douyin_url self.media_validator = media_validator or is_douyin_media_url self.media_selector = media_selector + self.target_id = target_id.strip() @contextmanager def connection(self, alias: str): @@ -287,22 +291,29 @@ class DouyinBrowser: targets = json.loads(payload) except json.JSONDecodeError as exc: raise DouyinError("browser target discovery response is invalid") from exc - if not isinstance(targets, list) or len(targets) > 32: + if not isinstance(targets, list): raise DouyinError("browser target discovery response is invalid") page_targets = [ target for target in targets if isinstance(target, dict) and target.get("type") == "page" ] - matching_targets = [ - target - for target in page_targets - if self.url_validator(target.get("url", "")) - ] - if len(matching_targets) > 1 or ( - not matching_targets and len(page_targets) > 1 - ): - raise DouyinError("browser has more than one page target") + if self.target_id: + matching_targets = [ + target for target in page_targets if target.get("id") == self.target_id + ] + if len(matching_targets) != 1: + raise DouyinError("configured browser page target is unavailable") + else: + matching_targets = [ + target + for target in page_targets + if self.url_validator(target.get("url", "")) + ] + if len(matching_targets) > 1 or ( + not matching_targets and len(page_targets) > 1 + ): + raise DouyinError("browser has more than one page target") target = ( matching_targets[0] if matching_targets @@ -338,7 +349,7 @@ class DouyinBrowser: socket = websocket.create_connection( page_target, timeout=CONTROL_TIMEOUT, - origin="devtools://devtools", + suppress_origin=True, enable_multithread=True, ) except (OSError, websocket.WebSocketException) as exc: @@ -382,24 +393,35 @@ class DouyinBrowser: frame_id = navigation.get("frameId") if not isinstance(frame_id, str) or navigation.get("errorText"): raise DouyinError("Douyin media page navigation failed") - cdp.wait_event( - "Page.frameNavigated", - lambda params: ( - params.get("frame", {}).get("id") == frame_id - and self.url_validator(params.get("frame", {}).get("url")) - ), - ) + target_path = urlsplit(target).path deadline = time.monotonic() + CONTROL_TIMEOUT while time.monotonic() < deadline: - if cdp.evaluate("document.readyState") in {"interactive", "complete"}: + try: + page = cdp.evaluate( + "({url: location.href, readyState: document.readyState})" + ) + except LISTENER_ERRORS: + time.sleep(0.1) + continue + if ( + isinstance(page, dict) + and self.url_validator(page.get("url", "")) + and urlsplit(page["url"]).path == target_path + and page.get("readyState") in {"interactive", "complete"} + ): break time.sleep(0.1) else: raise DouyinError("Douyin media page did not load") result = cdp.evaluate( f"""(async()=>{{ - const media=document.querySelector({json.dumps(self.media_selector)}); - const source=media?.currentSrc||media?.src||''; + const deadline=Date.now()+{MEDIA_SOURCE_WAIT_MS};let source=''; + while(Date.now()media.currentSrc||media.src||'') + .find(value=>value&&!value.includes('/obj/douyin-pc-web/uuu_265.mp4'))||''; + if(!source)await new Promise(resolve=>setTimeout(resolve,100)); + }} if(!source)return {{error:'media_source_unavailable'}}; const r=await fetch(source,{{credentials:'include',redirect:'error'}}); if(!r.body)return {{status:r.status,content_type:r.headers.get('content-type')||'',body:''}}; @@ -415,12 +437,10 @@ class DouyinBrowser: return {{status:r.status,content_type:r.headers.get('content-type')||'',body:btoa(binary)}}; }})()""" ) - if ( - not isinstance(result, dict) - or result.get("too_large") - or result.get("error") - ): + if not isinstance(result, dict) or result.get("too_large"): raise DouyinError("Douyin media download failed") + if result.get("error"): + raise DouyinError(f"Douyin media download failed: {result['error']}") status = result.get("status") content_type = result.get("content_type") body = result.get("body") @@ -463,13 +483,26 @@ class DouyinBrowser: raise DouyinError("Douyin identity response is invalid") if expected_uid and uid != expected_uid: raise DouyinError("Douyin identity does not match the expected account") - return { + result = { "uid": uid, "sec_uid": sec_uid, "unique_id": unique_id, "nickname": user.get("nickname", "") if isinstance(user, dict) else "", "short_id": user.get("short_id", "") if isinstance(user, dict) else "", } + extra = payload.get("extra") if isinstance(payload, dict) else None + server_now = extra.get("now") if isinstance(extra, dict) else None + if isinstance(server_now, (int, float)) and not isinstance(server_now, bool): + try: + server_now_float = float(server_now) + if not math.isfinite(server_now_float) or server_now_float <= 0: + raise ValueError("server clock is not finite") + result["platform_now"] = datetime.fromtimestamp( + server_now_float / 1000, timezone.utc + ).isoformat() + except (OverflowError, OSError, ValueError, TypeError) as exc: + raise DouyinError("Douyin platform clock is invalid") from exc + return result def action( self, @@ -570,6 +603,24 @@ class DouyinBrowser: exc.uncertain = True raise + def message_history( + self, alias: str, expected_uid: str, target_uid: str, limit: int = 100 + ) -> dict: + if not UID_RE.fullmatch(expected_uid) or not UID_RE.fullmatch(target_uid): + raise DouyinError("message history UID is invalid") + if not 1 <= limit <= 200: + raise DouyinError("message history limit is invalid") + self.identity(alias, expected_uid) + value = self._evaluate( + alias, + message_history_expression( + {"uid": target_uid, "limit": limit}, expected_uid + ), + ) + if not isinstance(value, dict): + raise DouyinError("Douyin message history response is invalid") + return value + class DouyinSubscription: def __init__( @@ -597,18 +648,19 @@ class DouyinSubscription: max_workers=4, thread_name_prefix=f"creatorhub-notice-details-{alias}" ) try: - self.connection = self._open_listener() + self.connection, boundary_at = self._open_listener_until_ready() except Exception: self._detail_pool.shutdown(wait=False, cancel_futures=True) raise - # Events observed before the consumer establishes its durable boundary + # Events observed before the consumer establishes its baseline boundary # are explicitly classified as baseline and must never trigger writes. self._put( { "kind": "baseline", "reason": "listener_start", "uid": self.uid, - "boundary_at": datetime.now(timezone.utc).isoformat(), + "boundary_at": boundary_at, + "boundary_source": "douyin_identity_extra_now", } ) self.thread = threading.Thread( @@ -616,7 +668,24 @@ class DouyinSubscription: ) self.thread.start() - def _open_listener(self) -> CDPConnection: + def _open_listener_until_ready(self) -> tuple[CDPConnection, str]: + deadline = time.monotonic() + LISTENER_START_TIMEOUT + delay = 0.5 + while True: + try: + return self._open_listener() + except LISTENER_ERRORS as exc: + if time.monotonic() >= deadline: + raise + LOG.info( + "Douyin notification listener is not ready; retrying", + extra={"alias": self.alias, "uid": self.uid, "reason": str(exc)}, + ) + if self.stopped.wait(delay): + raise DouyinError("Douyin notification listener stopped") from exc + delay = min(delay * 2, 5.0) + + def _open_listener(self) -> tuple[CDPConnection, str]: connection = self.browser._connect(self.alias) try: result = connection.evaluate(install_expression(self.key, self.uid)) @@ -625,15 +694,22 @@ class DouyinSubscription: state = json.loads(result) if not isinstance(state, dict) or not state.get("connected"): raise DouyinError("Douyin notification connection is not ready") - return connection - except ( - DouyinError, - OSError, - TypeError, - ValueError, - websocket.WebSocketException, - ): - connection.close() + identity = self.browser.identity(self.alias, self.uid) + boundary_at = identity.get("platform_now") + if not isinstance(boundary_at, str) or not boundary_at: + raise DouyinError("Douyin platform event boundary is unavailable") + return connection, boundary_at + except LISTENER_ERRORS: + try: + connection.evaluate(dispose_expression(self.key)) + except LISTENER_ERRORS: + LOG.debug( + "failed to dispose a partially installed Douyin listener", + extra={"alias": self.alias, "uid": self.uid}, + exc_info=True, + ) + finally: + connection.close() raise def _get_connection(self) -> CDPConnection: @@ -679,7 +755,7 @@ class DouyinSubscription: old = self.connection try: old.evaluate(dispose_expression(self.key)) - except DouyinError: + except (DouyinError, OSError, websocket.WebSocketException): LOG.debug( "old notification listener disposal was not available", exc_info=True, @@ -692,7 +768,7 @@ class DouyinSubscription: if self.stopped.wait(delay): return try: - connection = self._open_listener() + connection, boundary_at = self._open_listener() except ( DouyinError, OSError, @@ -714,7 +790,8 @@ class DouyinSubscription: "kind": "baseline", "reason": "listener_reconnected", "uid": self.uid, - "boundary_at": datetime.now(timezone.utc).isoformat(), + "boundary_at": boundary_at, + "boundary_source": "douyin_identity_extra_now", } ) return @@ -723,7 +800,7 @@ class DouyinSubscription: connection = self._get_connection() try: connection.evaluate(dispose_expression(self.key)) - except DouyinError: + except (DouyinError, OSError, websocket.WebSocketException): LOG.debug("notification listener disposal was not available", exc_info=True) finally: connection.close() @@ -1297,13 +1374,14 @@ XHS_ALLOWED_HOSTS = frozenset( class XiaohongshuBrowser(DouyinBrowser): - def __init__(self, endpoint=None) -> None: + def __init__(self, endpoint=None, *, target_id: str = "") -> None: super().__init__( endpoint, origin=XHS_ORIGIN, url_validator=is_xiaohongshu_url, media_validator=is_xiaohongshu_media_url, media_selector="video, img.note-slider-img", + target_id=target_id, ) def post(self, alias: str, target: str, body: bytes) -> BrowserResponse: @@ -1585,6 +1663,12 @@ def im_expression(params: dict, expected_uid: str) -> str: ) +def message_history_expression(params: dict, expected_uid: str) -> str: + return MESSAGE_HISTORY_SCRIPT.replace( + "EXPECTED_UID_VALUE", json.dumps(expected_uid) + ).replace("PARAMS_VALUE", json.dumps(params, ensure_ascii=True)) + + def action_expression(params: dict) -> str: return ACTION_SCRIPT.replace("PARAMS_VALUE", json.dumps(params, ensure_ascii=True)) @@ -1597,7 +1681,7 @@ INSTALL_SCRIPT = r"""(async()=>{try{ const codec=entries.find(([,f])=>{const s=String(f);return s.includes(".decodedFrame=")&&s.includes(".encodeFrame=");}); if(!entry||!codec)throw Error("SDK_CHANGED");const C=req(entry[0]).NoticeFrontier,decode=req(codec[0]).decodedFrame,f=C.frontierInstance; if(!f||String(f._options.deviceID)!==uid)throw Error("SOCKET_NOT_READY"); - const state={queue:carry,wake:null,C,f,uid,dropped:0,nextDelivery:0};const emit=e=>{const item={...e,delivered:false,delivery_id:"browser-"+(++state.nextDelivery)};if(state.queue.length>=1000){state.queue.shift();state.dropped++;state.queue.push({kind:"error",reason:"QUEUE_OVERFLOW",continuity:"gap",dropped:state.dropped,delivered:false,delivery_id:"browser-"+(++state.nextDelivery)});}else state.queue.push(item);if(state.wake)state.wake();}; + const state={queue:carry,wake:null,C,f,uid,dropped:0};const delivery=()=>"browser-"+crypto.randomUUID();const emit=e=>{const item={...e,delivered:false,delivery_id:delivery()};if(state.queue.length>=1000){state.queue.shift();state.dropped++;state.queue.push({kind:"error",reason:"QUEUE_OVERFLOW",continuity:"gap",dropped:state.dropped,delivered:false,delivery_id:delivery()});}else state.queue.push(item);if(state.wake)state.wake();}; const message=e=>{try{const frame=decode(new Uint8Array(e.data));if(frame.service===20313||frame.service===20003)emit({kind:"push",service:frame.service,payload:new TextDecoder().decode(frame.payload)});}catch(_){emit({kind:"error",reason:"FRAME_DECODE_FAILED",continuity:"gap"});}}; const open=()=>emit({kind:"open"}),close=()=>emit({kind:"close"});f.addEventListener("message",message);f.addEventListener("open",open);f.addEventListener("close",close); state.dispose=()=>{f.removeEventListener("message",message);f.removeEventListener("open",open);f.removeEventListener("close",close);if(state.wake)state.wake();};window[key]=state; @@ -1625,4 +1709,6 @@ FOLLOW_SCRIPT = r"""(async()=>{const p=PARAMS_VALUE;let sent=false;try{if(locati ACTION_SCRIPT = r"""(async()=>{const p=PARAMS_VALUE;let sent=false;const fail=(code,definitive=false)=>{const e=Error(code);e.definitive=definitive;throw e;};try{if(location.origin!=="https://www.douyin.com")fail("WRONG_ORIGIN");const qs="device_platform=webapp&aid=6383&channel=channel_pc_web";const get=async path=>{const r=await fetch(path,{credentials:"include",redirect:"error",signal:AbortSignal.timeout(15000)});let v;try{v=await r.json();}catch(_){fail("INVALID_RESPONSE");}if(!r.ok||v.status_code!==0)fail("READ_FAILED");return v;};const post=async(path,data)=>{const r=await fetch(path,{method:"POST",credentials:"include",redirect:"error",headers:{"Content-Type":"application/x-www-form-urlencoded;charset=UTF-8"},body:new URLSearchParams(data),signal:AbortSignal.timeout(10000)});let v;try{v=await r.json();}catch(_){fail(r.ok?"INVALID_RESPONSE":"POST_UNCERTAIN");}if(!r.ok)fail("POST_UNCERTAIN");if(v.status_code===undefined)fail("POST_UNCONFIRMED");if(v.status_code!==0)fail("BUSINESS_REJECTED",true);return v;};const self=await get("/aweme/v1/web/user/profile/self/?"+qs);if(String(self.user?.uid)!==p.expected)fail("IDENTITY_MISMATCH");if((p.action==="follow"||p.action==="dm")&&p.target===p.expected)fail("SELF_TARGET");const detail=async()=>get("/aweme/v1/web/aweme/detail/?"+qs+"&aweme_id="+encodeURIComponent(p.work));const findComment=async()=>{let cursor=0;for(let page=0;page<100;page++){const response=await get("/aweme/v1/web/comment/list/?"+qs+"&aweme_id="+encodeURIComponent(p.work)+"&cursor="+cursor+"&count=50");if(!Array.isArray(response.comments)||typeof response.has_more!=="boolean")fail("READ_FAILED");const list=response.comments;const comment=list.find(c=>String(c?.cid||c?.comment_id||"")===p.comment);if(comment){if(String(comment.aweme_id||comment.item_id||p.work)!==p.work||String(comment.user?.uid||comment.user_id||"")!==p.target)fail("TARGET_MISMATCH");return comment;}if(!response.has_more)break;const next=Number(response.cursor);if(!Number.isSafeInteger(next)||next<=cursor)fail("PAGINATION_INVALID");cursor=next;}fail("TARGET_NOT_FOUND");};if(p.action==="like_work"){const before=(await detail()).aweme_detail;if(String(before?.aweme_id)!==p.work)fail("TARGET_MISMATCH");if(Number(before.user_digged)===1)return {status:"succeeded",action:"already_liked",evidence:{work_id:p.work,user_digged:"1"}};sent=true;const result=await post("/aweme/v1/web/commit/item/digg/?"+qs,{aweme_id:p.work,type:"1",item_type:"0"});if(Number(result.is_digg)!==1)fail("POST_UNCONFIRMED");const after=(await detail()).aweme_detail;return {status:Number(after?.user_digged)===1?"succeeded":"unknown",action:"liked",evidence:{work_id:p.work,user_digged:String(after?.user_digged??"")}};}if(p.action==="like_comment"){const comment=await findComment();if(Number(comment.user_digged)===1)return {status:"succeeded",action:"already_liked",evidence:"comment.user_digged"};sent=true;await post("/aweme/v1/web/comment/digg?"+qs,{cid:p.comment,aweme_id:p.work,digg_type:"1",channel_id:"0",app_name:"aweme",item_type:"0",level:"1"});const after=await findComment();return {status:Number(after.user_digged)===1?"succeeded":"unknown",action:"liked_comment",evidence:{comment_id:p.comment,work_id:p.work,user_digged:String(after.user_digged??"")}};}if(p.action==="reply_comment"){await findComment();sent=true;const result=await post("/aweme/v1/web/comment/publish?"+qs,{app_name:"aweme",enter_from:"pc_web",previous_page:"video",reply_id:p.comment,reply_to_reply_id:"0",aweme_id:p.work,text:p.text,text_extra:"[]",comment_send_celltime:"0",comment_video_celltime:"0"});const posted=result.comment||result.comment_info||result.data?.comment;const postedID=String(posted?.cid||posted?.comment_id||"");const postedWork=String(posted?.aweme_id||posted?.item_id||"");const postedAuthor=String(posted?.user?.uid||posted?.user_id||"");const postedText=String(posted?.text??posted?.content??"");if(!posted||!postedID||postedWork!==p.work||postedAuthor!==p.expected||postedText!==p.text)fail("UNCONFIRMED");return {status:"succeeded",action:"replied",evidence:{comment_id:postedID,work_id:postedWork,author_uid:postedAuthor,text:postedText}};}if(p.action==="repost"){if(!p.text)fail("TEXT_REQUIRED");fail("REPOST_TEXT_UNSUPPORTED");}fail("ACTION_NOT_IMPLEMENTED");}catch(e){const code=String(e.message||"REQUEST_FAILED");return {status:sent?(e.definitive?"failed":"unknown"):"failed",code};}})()""" +MESSAGE_HISTORY_SCRIPT = r"""(async()=>{const p=PARAMS_VALUE;const fail=code=>{throw Error(code);};try{if(location.origin!=="https://www.douyin.com")fail("WRONG_ORIGIN");const response=await fetch("/aweme/v1/web/user/profile/self/?device_platform=webapp&aid=6383",{credentials:"include",signal:AbortSignal.timeout(15000)});if(!response.ok)fail("LOGIN_CHECK_FAILED");const profile=await response.json();if(profile.status_code!==0||String(profile.user?.uid)!==EXPECTED_UID_VALUE)fail("IDENTITY_MISMATCH");if(String(profile.user.uid)===p.uid)fail("SELF_TARGET");let service;for(const key of Object.keys(window).filter(k=>k.startsWith("@pc-im/im:"))){const chunks=window[key];if(!Array.isArray(chunks))continue;let req;chunks.push([["creatorhub-history-"+Date.now()],{},r=>{req=r;}]);chunks.pop();for(const [id,module] of Object.entries(req?.c||{})){if(!String(req.m?.[id]||"").includes("getOrCreatePrivateConversationByUid"))continue;for(const exported of Object.values(module.exports||{}))if(exported?.instance?.imSdkService)service=exported.instance.imSdkService;}if(service)break;}if(!service)fail("IM_SDK_NOT_READY");const sdk=service.imSdkManager.getImSdkInstance();if(!sdk)fail("IM_SDK_NOT_READY");const conversation=sdk.getConversationList().find(c=>c.type===1&&String(c.toParticipantUserId)===p.uid);if(!conversation)fail("CONVERSATION_NOT_FOUND");const meta=c=>({id:String(c.id),short_id:String(c.shortId),uid:String(c.toParticipantUserId),type:c.type});let cursor,previousCursor="",hasMore=false;for(let page=0;page<20;page++){const request={conversation,limit:Math.min(50,p.limit)};if(cursor!==undefined)request.cursor=cursor;const result=await sdk.getMessagesByConversation(request);if(!result||!Array.isArray(result.messages))fail("MESSAGE_HISTORY_INVALID");hasMore=Boolean(result.hasMore);if(!hasMore)break;const next=result.cursor,key=String(next?.toString?.()??next??"");if(!key||key===previousCursor)fail("MESSAGE_HISTORY_CURSOR_INVALID");previousCursor=key;cursor=next;}const messages=conversation.getMessageList();if(!Array.isArray(messages))fail("MESSAGE_HISTORY_INVALID");const pack=m=>{const ext=m.ext||{};const rawTime=ext["s:server_message_create_time"]||"";return {server_id:String(m.serverId||""),sender_uid:String(m.sender||""),message_type:String(m.type??""),content:typeof m.content==="string"?m.content.slice(0,100000):m.content,created_at:/^[0-9]+$/.test(String(rawTime))?String(rawTime):null,server_status:m.serverStatus??null};};return {status:"succeeded",action:"history",history_source:"im_sdk_pull",history_has_more:hasMore,account_uid:String(profile.user.uid),conversation:meta(conversation),messages:messages.slice(-p.limit).map(pack)};}catch(e){const known=["WRONG_ORIGIN","LOGIN_CHECK_FAILED","IDENTITY_MISMATCH","SELF_TARGET","IM_SDK_NOT_READY","CONVERSATION_NOT_FOUND","MESSAGE_HISTORY_INVALID","MESSAGE_HISTORY_CURSOR_INVALID"];return {status:"failed",code:known.includes(e.message)?e.message:"SDK_REQUEST_FAILED"};}})()""" + IM_SCRIPT = r"""(async()=>{const p=PARAMS_VALUE;let sent=false;const fail=(code,definitive=false)=>{const e=Error(code);e.definitive=definitive;throw e;};try{if(location.origin!=="https://www.douyin.com")fail("WRONG_ORIGIN");const response=await fetch("/aweme/v1/web/user/profile/self/?device_platform=webapp&aid=6383",{credentials:"include",signal:AbortSignal.timeout(15000)});if(!response.ok)fail("LOGIN_CHECK_FAILED");const profile=await response.json();if(profile.status_code!==0||String(profile.user?.uid)!==EXPECTED_UID_VALUE)fail("IDENTITY_MISMATCH");if(String(profile.user.uid)===p.uid)fail("SELF_TARGET");let service;for(const key of Object.keys(window).filter(k=>k.startsWith("@pc-im/im:"))){const chunks=window[key];if(!Array.isArray(chunks))continue;let req;chunks.push([["creatorhub-im-"+Date.now()],{},r=>{req=r;}]);for(const [id,module] of Object.entries(req?.c||{})){if(!String(req.m[id]).includes("getOrCreatePrivateConversationByUid"))continue;for(const exported of Object.values(module.exports||{}))if(exported?.instance?.imSdkService)service=exported.instance.imSdkService;}}if(!service)fail("IM_SDK_NOT_READY");const sdk=service.imSdkManager.getImSdkInstance();if(!sdk)fail("IM_SDK_NOT_READY");const meta=c=>({id:String(c.id),short_id:String(c.shortId),uid:String(c.toParticipantUserId),type:c.type});const pack=m=>({server_id:String(m.serverId||""),client_id:m.clientId||null,sender:String(m.sender),type:m.type,content:m.content,created_at:m.createdAt,server_status:m.serverStatus});let conversation=sdk.getConversationList().find(c=>c.type===1&&String(c.toParticipantUserId)===p.uid);if(!conversation&&p.action==="send")conversation=await service.conversationManager.getOrCreatePrivateConversationByUid(p.uid);if(!conversation)fail("CONVERSATION_NOT_FOUND");if(conversation.type!==1||String(conversation.toParticipantUserId)!==p.uid)fail("TARGET_MISMATCH");if(p.action!=="send"||!p.confirm)return {status:"preview",action:"preview",sender_uid:String(profile.user.uid),uid:p.uid,text:p.text,conversation:meta(conversation)};const message=await sdk.createMessage({conversation,type:7,content:JSON.stringify({aweType:700,type:0,richTextInfos:[],text:p.text})});if(!message||typeof message.sendFunc!=="function")fail("MESSAGE_BUILD_FAILED");sent=true;const result=await Promise.race([sdk.sendMessage({message}),new Promise((_,reject)=>setTimeout(()=>reject(Error("MESSAGE_UNCONFIRMED")),10000))]);if(result?.success===false)fail("MESSAGE_REJECTED",true);if(result?.success!==true)fail("MESSAGE_UNCONFIRMED");const packed=pack(message);if(!packed.server_id&&!packed.client_id)fail("MESSAGE_UNCONFIRMED");return {status:"succeeded",action:"send",success:true,status_code:result?.statusCode??null,check_code:String(result?.checkCode??""),conversation:meta(conversation),message:packed,evidence:{conversation_id:String(conversation.id),message_server_id:packed.server_id,message_client_id:String(packed.client_id??"")}};}catch(e){const known=["WRONG_ORIGIN","LOGIN_CHECK_FAILED","IDENTITY_MISMATCH","SELF_TARGET","IM_SDK_NOT_READY","CONVERSATION_NOT_FOUND","TARGET_MISMATCH","MESSAGE_BUILD_FAILED","MESSAGE_REJECTED","MESSAGE_UNCONFIRMED"];return {status:sent?(e.definitive?"failed":"unknown"):"failed",code:known.includes(e.message)?e.message:"SDK_REQUEST_FAILED"};}})()""" diff --git a/cmd/docker_gateway/gateway.py b/cmd/docker_gateway/gateway.py index a58f288..a6686d3 100644 --- a/cmd/docker_gateway/gateway.py +++ b/cmd/docker_gateway/gateway.py @@ -13,7 +13,7 @@ import socket import threading import time from collections.abc import Mapping -from contextlib import suppress +from contextlib import nullcontext, suppress from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import cast from urllib.parse import parse_qs, quote, urlsplit @@ -63,6 +63,7 @@ EXIT_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$") DOUYIN_ACCOUNT_KEY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@-]{0,127}$") DOUYIN_ORIGIN = "https://www.douyin.com" DOUYIN_IDENTITY_PATH = "/aweme/v1/web/user/profile/self/" +DOUYIN_PROFILE_OTHER_PATH = "/aweme/v1/web/user/profile/other/" DOUYIN_IDENTITY_URL = IDENTITY_URL DOUYIN_WORKS_PATH = WORKS_PATH DOUYIN_COMMENTS_PATH = COMMENTS_PATH @@ -94,11 +95,13 @@ class Gateway: self_name: str, browser: DouyinBrowser | None = None, xiaohongshu_browser: XiaohongshuBrowser | None = None, + external_cdp: Mapping[str, object] | None = None, ) -> None: self.docker = docker self.network = network self.token = token self.self_name = self_name + self.external_cdp = dict(external_cdp or {}) self.browser = browser or DouyinBrowser(self._browser_endpoint) self.xiaohongshu_browser = xiaohongshu_browser or XiaohongshuBrowser( self._browser_endpoint @@ -110,6 +113,10 @@ class Gateway: self._uncertain_actions: dict[str, float] = {} def _browser_endpoint(self, alias: str) -> str: + if self.external_cdp: + if alias != self.external_cdp["alias"]: + raise GenerationConflict("external browser alias does not match") + return cast(str, self.external_cdp["url"]) container_id, labels = self.docker.managed_container(alias) network_id = labels.get(NETWORK_ID_LABEL, "") if not isinstance(network_id, str) or not network_id: @@ -118,6 +125,21 @@ class Gateway: return f"http://{address}:9222" def list_browsers(self) -> list[dict]: + if self.external_cdp: + return [ + { + "id": self.external_cdp["runtime_id"], + "alias": self.external_cdp["alias"], + "name": self.external_cdp["alias"], + "state": "external", + "status": "external CDP", + "endpoint": self.external_cdp["url"], + "binding_version": self.external_cdp["binding_version"], + "network_exit_id": "", + "network_id": self.external_cdp["network_id"], + "proxy_ready": True, + } + ] filters = quote( json.dumps({"label": [f"{MANAGED_LABEL}=true"]}, separators=(",", ":")), safe="", @@ -808,6 +830,37 @@ class Gateway: ) return identity + def douyin_message_history(self, alias: str, input: dict) -> dict: + expected_uid = input.get("expected_uid", "") + target_uid = input.get("target_uid", "") + limit = input.get("limit", 100) + if ( + not valid_douyin_generation(input) + or not isinstance(expected_uid, str) + or not isinstance(target_uid, str) + or type(limit) is not int + or not UID_RE.fullmatch(expected_uid) + or not UID_RE.fullmatch(target_uid) + or expected_uid == target_uid + or not 1 <= limit <= 200 + ): + raise RequestError("invalid Douyin message history request", 400) + with self._alias_lock(alias): + self._require_douyin_generation(alias, input) + try: + result = self.browser.message_history( + alias, expected_uid, target_uid, limit + ) + self._require_douyin_generation(alias, input) + except DouyinError as exc: + LOG.warning( + "Douyin message history failed alias=%s reason=%s", + alias, + str(exc), + ) + raise RequestError("Douyin message history failed") from exc + return result + def douyin_action(self, alias: str, input: dict) -> dict: expected_uid = input.get("expected_uid", "") action = input.get("action", "") @@ -903,6 +956,12 @@ class Gateway: self.browser.identity(alias, expected_uid) return self.subscriptions.start(alias, expected_uid) except DouyinError as exc: + LOG.warning( + "Douyin event listener start failed alias=%s uid=%s reason=%s", + alias, + expected_uid, + exc, + ) raise RequestError("Douyin event listener could not start") from exc def poll_douyin_events(self, alias: str, input: dict, query: dict) -> list[dict]: @@ -925,6 +984,11 @@ class Gateway: self.subscriptions.ack(alias, delivery_ids) return self.subscriptions.poll(alias, limit, wait) except DouyinError as exc: + LOG.warning( + "Douyin event listener poll failed alias=%s reason=%s", + alias, + exc, + ) raise RequestError("Douyin event listener is unavailable") from exc def stop_douyin_events(self, alias: str, input: dict) -> None: @@ -949,6 +1013,19 @@ class Gateway: return container_id, True def _require_douyin_generation(self, alias: str, input: dict) -> None: + if self.external_cdp: + if ( + alias != self.external_cdp["alias"] + or any( + input.get(key) != self.external_cdp[key] + for key in ("binding_version", "runtime_id", "network_id") + ) + or input.get("network_exit_id", "") + ): + raise RequestError( + "external browser generation does not match request", 409 + ) + return container_id, labels = self.docker.managed_container(alias) if ( container_id != input["runtime_id"] @@ -1109,6 +1186,10 @@ class Gateway: ) def _alias_lock(self, alias: str): + if self.external_cdp: + if alias != self.external_cdp["alias"]: + raise GenerationConflict("external browser alias does not match") + return nullcontext() return _AliasLock(self.reservations, alias) @@ -1289,7 +1370,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|events)", + r"/v1/browsers/([a-z0-9][a-z0-9-]{0,31})/douyin/(get|media|identity|action|messages|events)", path, ) if match: @@ -1302,6 +1383,8 @@ class GatewayHandler(BaseHTTPRequestHandler): return gateway.douyin_identity(alias, body) if action == "action" and method == "POST": return gateway.douyin_action(alias, body) + if action == "messages" and method == "POST": + return gateway.douyin_message_history(alias, body) if action == "events": if method == "POST": return gateway.start_douyin_events(alias, body) @@ -1752,23 +1835,39 @@ def valid_douyin_url(raw: object) -> bool: query = parse_qs(parsed.query, keep_blank_values=True) if parsed.path == DOUYIN_IDENTITY_PATH: return query == {"aid": ["6383"], "device_platform": ["webapp"]} + if parsed.path == DOUYIN_PROFILE_OTHER_PATH: + return valid_douyin_profile_query(query) if parsed.path == DOUYIN_WORKS_PATH: - return ( - len(query) == 3 - and valid_account_key_query(query, "sec_user_id") - and query.get("count") == ["20"] - and numeric_cursor(query.get("max_cursor")) - ) + return valid_douyin_api_query(query, "sec_user_id", "max_cursor") if parsed.path == DOUYIN_COMMENTS_PATH: - return ( - len(query) == 3 - and valid_account_key_query(query, "aweme_id") - and query.get("count") == ["20"] - and numeric_cursor(query.get("cursor")) - ) + return valid_douyin_api_query(query, "aweme_id", "cursor") return False +def valid_douyin_profile_query(query: dict[str, list[str]]) -> bool: + account_fields = [field for field in ("user_id", "sec_user_id") if field in query] + return ( + len(query) == 3 + and query.get("aid") == ["6383"] + and query.get("device_platform") == ["webapp"] + and len(account_fields) == 1 + and valid_account_key_query(query, account_fields[0]) + ) + + +def valid_douyin_api_query( + query: dict[str, list[str]], account_field: str, cursor_field: str +) -> bool: + return ( + len(query) == 5 + and query.get("aid") == ["6383"] + and query.get("device_platform") == ["webapp"] + and valid_account_key_query(query, account_field) + and query.get("count") == ["20"] + and numeric_cursor(query.get(cursor_field)) + ) + + def valid_account_key_query(query: dict[str, list[str]], key: str) -> bool: return len(query.get(key, [])) == 1 and bool( DOUYIN_ACCOUNT_KEY_RE.fullmatch(query[key][0]) @@ -1802,6 +1901,8 @@ def load_config(env: Mapping[str, str] | None = None) -> dict: socket_path = env.get("DOCKER_SOCKET", "/var/run/docker.sock").strip() network = env.get("BROWSER_NETWORK", "creatorhub_browser").strip() token = env.get("GATEWAY_TOKEN", "").strip() + cdp_url = env.get("BROWSER_CDP_URL", "").strip() + cdp_target_id = env.get("BROWSER_CDP_TARGET_ID", "").strip() host, port = split_listen_address(listen) if ( not socket_path @@ -1812,14 +1913,58 @@ def load_config(env: Mapping[str, str] | None = None) -> dict: raise ValueError("invalid gateway configuration") if not 1 <= port <= 65535: raise ValueError("LISTEN_ADDR port must be 1..65535") + external_cdp = None + if cdp_url: + if not valid_cdp_url(cdp_url) or ( + cdp_target_id and not re.fullmatch(r"^[A-Za-z0-9_-]{1,128}$", cdp_target_id) + ): + raise ValueError("BROWSER_CDP_URL or BROWSER_CDP_TARGET_ID is invalid") + try: + binding_version = int(env.get("BROWSER_CDP_BINDING_VERSION", "1")) + except (TypeError, ValueError) as exc: + raise ValueError("BROWSER_CDP_BINDING_VERSION must be an integer") from exc + external_cdp = { + "url": cdp_url, + "target_id": cdp_target_id, + "alias": env.get("BROWSER_CDP_ALIAS", "local-cdp").strip(), + "runtime_id": env.get("BROWSER_CDP_RUNTIME_ID", "0" * 64).strip(), + "network_id": env.get("BROWSER_CDP_NETWORK_ID", "local-cdp").strip(), + "binding_version": binding_version, + } + if ( + not RUNTIME_ID_RE.fullmatch(external_cdp["alias"]) + or not CONTAINER_ID_RE.fullmatch(external_cdp["runtime_id"]) + or not EXIT_ID_RE.fullmatch(external_cdp["network_id"]) + or external_cdp["binding_version"] < 1 + ): + raise ValueError("BROWSER_CDP generation is invalid") return { "listen": (host, port), "docker_socket": socket_path, "network": network, "token": token, + "external_cdp": external_cdp, } +def valid_cdp_url(raw: str) -> bool: + try: + parsed = urlsplit(raw) + port = parsed.port + except ValueError: + return False + return ( + parsed.scheme == "http" + and bool(parsed.hostname) + and port is not None + and parsed.path in ("", "/") + and not parsed.username + and not parsed.password + and not parsed.query + and not parsed.fragment + ) + + def split_listen_address(value: str) -> tuple[str, int]: if value.startswith(":"): host, port_text = "", value[1:] @@ -1843,7 +1988,29 @@ def run() -> None: config = load_config() logging.basicConfig(level=logging.INFO, format="%(message)s") docker = DockerClient(config["docker_socket"]) - gateway = Gateway(docker, config["network"], config["token"], socket.gethostname()) + external_cdp = config["external_cdp"] + browser = None + xiaohongshu_browser = None + if external_cdp: + + def endpoint(_alias: str) -> str: + return cast(str, external_cdp["url"]) + + browser = DouyinBrowser( + endpoint, target_id=cast(str, external_cdp["target_id"]) + ) + xiaohongshu_browser = XiaohongshuBrowser( + endpoint, target_id=cast(str, external_cdp["target_id"]) + ) + gateway = Gateway( + docker, + config["network"], + config["token"], + socket.gethostname(), + browser=browser, + xiaohongshu_browser=xiaohongshu_browser, + external_cdp=external_cdp, + ) server = GatewayHTTPServer(config["listen"], gateway) LOG.info( json.dumps( diff --git a/cmd/docker_gateway/test_gateway.py b/cmd/docker_gateway/test_gateway.py index ed3ff94..5f3e8bb 100644 --- a/cmd/docker_gateway/test_gateway.py +++ b/cmd/docker_gateway/test_gateway.py @@ -189,6 +189,25 @@ class GatewayValidationTests(unittest.TestCase): } ) self.assertEqual(config["listen"], ("", 8081)) + external = load_config( + { + "LISTEN_ADDR": ":8081", + "DOCKER_SOCKET": "/var/run/docker.sock", + "BROWSER_NETWORK": "creatorhub_browser", + "GATEWAY_TOKEN": "0123456789abcdef", + "BROWSER_CDP_URL": "http://127.0.0.1:9222", + "BROWSER_CDP_TARGET_ID": "target_1", + } + ) + self.assertEqual(external["external_cdp"]["url"], "http://127.0.0.1:9222") + self.assertEqual(external["external_cdp"]["target_id"], "target_1") + with self.assertRaises(ValueError): + load_config( + { + "GATEWAY_TOKEN": "0123456789abcdef", + "BROWSER_CDP_URL": "http://user:pass@127.0.0.1:9222", + } + ) tmp_mount = next(path for path in browser_tmpfs() if path.endswith("tmp")) self.assertTrue(browser_tmpfs()[tmp_mount].startswith("rw,")) self.assertTrue( @@ -196,16 +215,60 @@ class GatewayValidationTests(unittest.TestCase): "https://www.douyin.com/aweme/v1/web/user/profile/self/?aid=6383&device_platform=webapp" ) ) + self.assertTrue( + valid_douyin_url( + "https://www.douyin.com/aweme/v1/web/user/profile/other/?aid=6383&device_platform=webapp&user_id=2328120603967913" + ) + ) + self.assertTrue( + valid_douyin_url( + "https://www.douyin.com/aweme/v1/web/user/profile/other/?aid=6383&device_platform=webapp&sec_user_id=MS4wLjABAAAA9f_a7k0bzVizLYXlpC7R61EIaqJ8Ordug7yp7AB8fGKuuF8Fzqk5_DM-eutXnPIK" + ) + ) + self.assertFalse( + valid_douyin_url( + "https://www.douyin.com/aweme/v1/web/user/profile/other/?aid=6383&device_platform=webapp&user_id=1&count=20" + ) + ) self.assertFalse(valid_douyin_url("https://www.douyin.com.evil/")) self.assertTrue(is_douyin_url("https://www.douyin.com/video/123")) self.assertFalse(is_douyin_url("https://www.douyin.com.evil/video/123")) + def test_external_cdp_is_virtual_and_generation_bound(self) -> None: + external = { + "url": "http://127.0.0.1:9222", + "target_id": "target_1", + "alias": "local-cdp", + "runtime_id": "0" * 64, + "network_id": "local-cdp", + "binding_version": 1, + } + gateway = Gateway( + cast(DockerClient, FakeDocker([])), + "creatorhub_browser", + "0123456789abcdef", + "gateway", + external_cdp=external, + ) + self.assertEqual(gateway.list_browsers()[0]["state"], "external") + generation = { + "binding_version": 1, + "runtime_id": "0" * 64, + "network_id": "local-cdp", + } + gateway._require_douyin_generation("local-cdp", generation) + with self.assertRaises(RequestError): + gateway._require_douyin_generation( + "local-cdp", {**generation, "binding_version": 2} + ) + def test_http_routes_and_body_validation(self) -> None: handler = gateway_module.GatewayHandler.__new__(gateway_module.GatewayHandler) gateway = Mock() gateway.list_browsers.return_value = [] gateway.douyin_identity.return_value = {"uid": "123"} gateway.douyin_action.return_value = {"status": "succeeded"} + gateway.douyin_message_history.return_value = {"status": "succeeded"} gateway.poll_douyin_events.return_value = [] server = Mock() server.gateway = gateway @@ -233,6 +296,11 @@ class GatewayValidationTests(unittest.TestCase): handler._route("POST", "/v1/browsers/safe/douyin/get", {}, {}) handler._route("POST", "/v1/browsers/safe/douyin/identity", {}, {}) handler._route("POST", "/v1/browsers/safe/douyin/action", {}, {}) + self.assertEqual( + handler._route("POST", "/v1/browsers/safe/douyin/messages", {}, {}), + {"status": "succeeded"}, + ) + gateway.douyin_message_history.assert_called_once_with("safe", {}) self.assertEqual( handler._route("GET", "/v1/browsers/safe/douyin/events", {}, {}), [] ) @@ -570,6 +638,7 @@ class BrowserCDP: self.values = list(values) self.commands: list[tuple[str, dict | None]] = [] self.events: list[str] = [] + self.expressions: list[str] = [] self.closed = False def command(self, method: str, params: dict | None = None) -> dict: @@ -584,7 +653,7 @@ class BrowserCDP: return {"method": method} def evaluate(self, expression: str) -> object: - del expression + self.expressions.append(expression) if not self.values: raise DouyinError("fake CDP value exhausted") return self.values.pop(0) @@ -650,11 +719,47 @@ class BrowserTests(unittest.TestCase): self.assertEqual(response.status, 200) self.assertNotIn("Network.setCookies", [method for method, _ in cdp.commands]) + 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) + ] + targets.extend( + [ + { + "id": "other", + "type": "page", + "url": "https://www.douyin.com/video/1", + "webSocketDebuggerUrl": "ws://127.0.0.1:9222/devtools/page/1", + }, + { + "id": "selected", + "type": "page", + "url": "https://www.douyin.com/user/self", + "webSocketDebuggerUrl": "ws://127.0.0.1:9222/devtools/page/2", + }, + ] + ) + http = FakeHTTPConnection(FakeHTTPResponse(200, json.dumps(targets).encode())) + with ( + patch.object( + douyin_module.http.client, "HTTPConnection", return_value=http + ), + patch.object( + douyin_module.websocket, + "create_connection", + return_value=FakeSocket([]), + ), + ): + connection = DouyinBrowser( + lambda alias: "http://127.0.0.1:9222", target_id="selected" + )._connect("safe") + self.assertIsInstance(connection, CDPConnection) + def test_media_download_is_browser_mediated_and_bounded(self) -> None: payload = base64.b64encode(b"video-bytes").decode("ascii") cdp = BrowserCDP( [ - "complete", + {"url": "https://www.douyin.com/video/123", "readyState": "complete"}, {"status": 200, "content_type": "video/mp4", "body": payload}, ] ) @@ -665,6 +770,8 @@ class BrowserTests(unittest.TestCase): self.assertEqual(response.content_type, "video/mp4") self.assertEqual(base64.b64decode(response.body_base64), b"video-bytes") self.assertIn("Page.navigate", [method for method, _ in cdp.commands]) + self.assertIn("querySelectorAll", cdp.expressions[-1]) + self.assertIn("uuu_265.mp4", cdp.expressions[-1]) def test_connect_identity_and_actions(self) -> None: target = [ @@ -741,6 +848,28 @@ class BrowserTests(unittest.TestCase): ack = ack_expression("__creatorhub_notice_sub_alpha", ["browser-1"]) self.assertIn("browser-1", ack) self.assertIn("__creatorhub_notice_sub_alpha", ack) + self.assertIn( + "crypto.randomUUID", douyin_module.install_expression("alpha", "123") + ) + + def test_listener_start_retries_until_runtime_ready(self) -> None: + subscription = DouyinSubscription.__new__(DouyinSubscription) + subscription.alias = "safe" + subscription.uid = "123" + subscription.stopped = threading.Event() + subscription.stopped.wait = Mock(return_value=False) + subscription._open_listener = Mock( + side_effect=[ + DouyinError("SDK_NOT_READY"), + (Mock(), "2026-09-14T18:00:00+00:00"), + ] + ) + with patch.object(douyin_module.time, "monotonic", side_effect=[0.0, 1.0]): + connection, boundary = subscription._open_listener_until_ready() + self.assertEqual(boundary, "2026-09-14T18:00:00+00:00") + self.assertIsNotNone(connection) + self.assertEqual(subscription._open_listener.call_count, 2) + subscription.stopped.wait.assert_called_once_with(0.5) def test_subscription_receipts_are_replayed_until_ack(self) -> None: subscription = DouyinSubscription.__new__(DouyinSubscription) @@ -1726,6 +1855,17 @@ class AdditionalGatewayCoverageTests(unittest.TestCase): 200, json.dumps({"status_code": 0, "user": {"uid": "1", "sec_uid": "sec"}}) ) self.assertEqual(browser.identity("safe")["uid"], "1") + cast(Any, browser).get = lambda alias, target: BrowserResponse( + 200, + json.dumps( + { + "status_code": 0, + "extra": {"now": 1789401629000}, + "user": {"uid": "1", "sec_uid": "sec"}, + } + ), + ) + self.assertTrue(browser.identity("safe")["platform_now"].startswith("2026-")) cast(Any, browser).get = lambda alias, target: BrowserResponse( 403, json.dumps({"status_code": 0, "user": {"uid": "1", "sec_uid": "sec"}}) ) @@ -2065,6 +2205,232 @@ class AdditionalGatewayCoverageTests(unittest.TestCase): ) self.assertEqual(result.id, "observed") + def test_configured_target_and_media_wait_edge_cases(self) -> None: + targets = [ + { + "type": "page", + "id": "other", + "url": "https://www.douyin.com/1", + "webSocketDebuggerUrl": "ws://browser:9222/devtools/page/1", + } + ] + http = FakeHTTPConnection(FakeHTTPResponse(200, json.dumps(targets).encode())) + with ( + patch.object( + douyin_module.http.client, "HTTPConnection", return_value=http + ), + self.assertRaisesRegex(DouyinError, "configured browser page target"), + ): + DouyinBrowser( + lambda _: "http://browser:9222", target_id="missing" + )._connect("safe") + + cdp = Mock() + cdp.command.return_value = {"frameId": "frame-1"} + cdp.evaluate.side_effect = [ + DouyinError("page is still loading"), + {"url": "https://www.douyin.com/video/123", "readyState": "complete"}, + {"error": "media_source_unavailable"}, + ] + browser = DouyinBrowser() + BrowserTests()._with_connection(browser, cast(BrowserCDP, cdp)) + with ( + patch.object(douyin_module.time, "monotonic", side_effect=[0.0, 1.0, 2.0]), + patch.object(douyin_module.time, "sleep"), + self.assertRaisesRegex(DouyinError, "media_source_unavailable"), + ): + browser.get_media("safe", "https://www.douyin.com/video/123") + + def test_identity_and_message_history_reject_invalid_values(self) -> None: + browser = DouyinBrowser() + for server_now in (0, float("nan")): + cast(Any, browser).get = lambda alias, target, now=server_now: BrowserResponse( + 200, + json.dumps( + { + "status_code": 0, + "extra": {"now": now}, + "user": {"uid": "123", "sec_uid": "sec"}, + } + ), + ) + with self.subTest(server_now=server_now), self.assertRaisesRegex( + DouyinError, "platform clock" + ): + browser.identity("safe") + + identity = Mock(return_value={"uid": "123"}) + evaluate = Mock(return_value={"status": "succeeded", "messages": []}) + cast(Any, browser).identity = identity + cast(Any, browser)._evaluate = evaluate + result = browser.message_history("safe", "123", "456", 20) + self.assertEqual(result["status"], "succeeded") + identity.assert_called_once_with("safe", "123") + self.assertIn("456", evaluate.call_args.args[1]) + for expected_uid, target_uid, limit in ( + ("bad", "456", 20), + ("123", "bad", 20), + ("123", "456", 0), + ("123", "456", 201), + ): + with self.subTest( + expected_uid=expected_uid, target_uid=target_uid, limit=limit + ), self.assertRaises(DouyinError): + browser.message_history("safe", expected_uid, target_uid, limit) + evaluate.return_value = "bad" + with self.assertRaisesRegex(DouyinError, "message history response"): + browser.message_history("safe", "123", "456") + + def test_listener_open_and_recovery_boundaries(self) -> None: + browser = Mock() + browser.identity.return_value = { + "uid": "123", + "platform_now": "2026-09-14T00:00:00+00:00", + } + connection = BrowserCDP([json.dumps({"connected": True})]) + browser._connect.return_value = connection + subscription = DouyinSubscription.__new__(DouyinSubscription) + subscription.browser = browser + subscription.alias = "safe" + subscription.uid = "123" + subscription.key = "__creatorhub_notice_sub_safe" + opened, boundary_at = subscription._open_listener() + self.assertIs(opened, connection) + self.assertEqual(boundary_at, "2026-09-14T00:00:00+00:00") + + constructor_connection = BrowserCDP([json.dumps({"connected": True})]) + browser._connect.return_value = constructor_connection + with patch.object(threading.Thread, "start"): + constructed = DouyinSubscription(browser, "safe", "123") + self.assertEqual(constructed.queue[0]["kind"], "baseline") + constructed.stopped.set() + constructed._detail_pool.shutdown(wait=True, cancel_futures=True) + + broken = BrowserCDP([json.dumps({"connected": True}), "{}"]) + browser._connect.return_value = broken + browser.identity.return_value = {"uid": "123"} + with self.assertRaisesRegex(DouyinError, "event boundary"): + subscription._open_listener() + self.assertTrue(broken.closed) + + old = Mock() + old.evaluate.side_effect = DouyinError("stale listener") + new = Mock() + subscription.connection = old + subscription._connection_lock = threading.RLock() + subscription._epoch = 0 + subscription._initial_boundary_pending = False + subscription.stopped = threading.Event() + subscription.stopped.wait = Mock(return_value=False) + subscription._open_listener = Mock( + return_value=(new, "2026-09-14T00:00:01+00:00") + ) + subscription._put = Mock() + subscription._recover() + old.close.assert_called_once_with() + self.assertIs(subscription.connection, new) + self.assertEqual(subscription._epoch, 1) + self.assertEqual( + [call.args[0]["kind"] for call in subscription._put.call_args_list], + ["reconnected", "baseline"], + ) + + def test_external_message_history_and_generation_fences(self) -> None: + external = { + "url": "http://127.0.0.1:9222", + "target_id": "target_1", + "alias": "local-cdp", + "runtime_id": "0" * 64, + "network_id": "local-cdp", + "binding_version": 1, + } + gateway = Gateway( + cast(DockerClient, FakeDocker([])), + "creatorhub_browser", + "0123456789abcdef", + "gateway", + external_cdp=external, + ) + self.assertEqual(gateway._browser_endpoint("local-cdp"), external["url"]) + with self.assertRaises(GenerationConflict): + gateway._browser_endpoint("other") + with gateway._alias_lock("local-cdp"): + pass + + browser = Mock() + browser.message_history.return_value = { + "status": "succeeded", + "history_source": "im_sdk_pull", + "messages": [], + } + gateway.browser = browser + generation = { + "binding_version": 1, + "runtime_id": "0" * 64, + "network_id": "local-cdp", + "expected_uid": "123", + "target_uid": "456", + "limit": 20, + } + self.assertEqual( + gateway.douyin_message_history("local-cdp", generation)["status"], + "succeeded", + ) + browser.message_history.assert_called_once_with("local-cdp", "123", "456", 20) + with self.assertRaises(RequestError): + gateway.douyin_message_history( + "local-cdp", {**generation, "target_uid": "123"} + ) + browser.message_history.side_effect = DouyinError("offline") + with self.assertRaises(RequestError): + gateway.douyin_message_history("local-cdp", generation) + + def test_douyin_api_urls_and_external_config_edges(self) -> None: + self.assertTrue( + valid_douyin_url( + "https://www.douyin.com" + f"{gateway_module.DOUYIN_WORKS_PATH}" + "?aid=6383&device_platform=webapp&sec_user_id=sec&count=20&max_cursor=0" + ) + ) + self.assertTrue( + valid_douyin_url( + "https://www.douyin.com" + f"{gateway_module.DOUYIN_COMMENTS_PATH}" + "?aid=6383&device_platform=webapp&aweme_id=123&count=20&cursor=0" + ) + ) + self.assertFalse( + valid_douyin_url( + "https://www.douyin.com" + f"{gateway_module.DOUYIN_WORKS_PATH}" + "?aid=6383&device_platform=webapp&sec_user_id=sec&count=10&max_cursor=0" + ) + ) + self.assertTrue(gateway_module.valid_cdp_url("http://127.0.0.1:9222/")) + for raw in ( + "https://127.0.0.1:9222", + "http://127.0.0.1:9222/devtools", + "http://127.0.0.1:9222?x=1", + "http://user:pass@127.0.0.1:9222", + "http://127.0.0.1:not-a-port", + ): + with self.subTest(raw=raw): + self.assertFalse(gateway_module.valid_cdp_url(raw)) + base = { + "LISTEN_ADDR": ":8081", + "DOCKER_SOCKET": "/var/run/docker.sock", + "BROWSER_NETWORK": "creatorhub_browser", + "GATEWAY_TOKEN": "0123456789abcdef", + "BROWSER_CDP_URL": "http://127.0.0.1:9222", + } + with self.assertRaises(ValueError): + load_config({**base, "BROWSER_CDP_BINDING_VERSION": "not-an-int"}) + with self.assertRaises(ValueError): + load_config({**base, "BROWSER_CDP_BINDING_VERSION": "0"}) + with self.assertRaises(ValueError): + load_config({**base, "BROWSER_CDP_ALIAS": "bad alias"}) + if __name__ == "__main__": unittest.main() diff --git a/cmd/docker_gateway/test_xiaohongshu.py b/cmd/docker_gateway/test_xiaohongshu.py index e01a20b..0822560 100644 --- a/cmd/docker_gateway/test_xiaohongshu.py +++ b/cmd/docker_gateway/test_xiaohongshu.py @@ -5,6 +5,7 @@ from typing import Any, cast from unittest.mock import Mock from . import gateway as gateway_module +from . import xiaohongshu as xiaohongshu_module Gateway = gateway_module.Gateway RequestError = gateway_module.RequestError @@ -78,6 +79,12 @@ class XiaohongshuValidationTests(unittest.TestCase): ) ) + def test_facade_exports_shared_browser(self) -> None: + self.assertIs( + xiaohongshu_module.XiaohongshuBrowser, + gateway_module.XiaohongshuBrowser, + ) + class XiaohongshuRouteTests(unittest.TestCase): def test_read_only_routes_dispatch_without_action_or_event_routes(self) -> None: diff --git a/compose.dev.yaml b/compose.dev.yaml index d1ae8b7..f8f0e36 100644 --- a/compose.dev.yaml +++ b/compose.dev.yaml @@ -1,5 +1,7 @@ +--- # 本地热加载开发用 override：只跑 postgres + docker-gateway，宿主机直接跑 control-plane 与前端。 -# 用法：docker compose -f compose.yaml -f compose.dev.yaml up -d postgres docker-gateway +# 用法：docker compose -f compose.yaml -f compose.dev.yaml up -d \ +# postgres docker-gateway services: creator-hub: # 本地开发不跑容器版 control-plane；profile 化后不启动。 @@ -12,3 +14,10 @@ services: docker-gateway: ports: - "8081:8081" + extra_hosts: + - "host.docker.internal:host-gateway" + environment: + BROWSER_CDP_URL: ${BROWSER_CDP_URL:-} + BROWSER_CDP_TARGET_ID: ${BROWSER_CDP_TARGET_ID:-} + BROWSER_CDP_ALIAS: ${BROWSER_CDP_ALIAS:-local-cdp} + BROWSER_CDP_NETWORK_ID: ${BROWSER_CDP_NETWORK_ID:-local-cdp} diff --git a/compose.yaml b/compose.yaml index 2c6ba3d..b4ddcb4 100644 --- a/compose.yaml +++ b/compose.yaml @@ -27,6 +27,18 @@ services: postgres: condition: service_healthy networks: [control] + healthcheck: + test: + - CMD + - python3 + - -c + - >- + import urllib.request; urllib.request.urlopen( + 'http://127.0.0.1:8080/readyz', timeout=2 + ) + interval: 2s + timeout: 3s + retries: 15 restart: unless-stopped postgres: diff --git a/docs/deployment.md b/docs/deployment.md index df3e315..d8e3fdc 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -75,6 +75,10 @@ curl --fail --silent --show-error \ --retry 30 --retry-delay 2 --retry-connrefused \ --output /dev/null "http://127.0.0.1:${CREATORHUB_PORT}/healthz" +curl --fail --silent --show-error \ + --retry 30 --retry-delay 2 --retry-connrefused \ + --output /dev/null "http://127.0.0.1:${CREATORHUB_PORT}/readyz" + curl --fail --silent --show-error \ --retry 30 --retry-delay 2 --retry-connrefused \ --user "${CONTROL_PLANE_USERNAME}:${CONTROL_PLANE_PASSWORD}" \ @@ -287,10 +291,28 @@ Compose 部署时通常只需设置以下宿主机变量： | `docker-gateway` | `DOCKER_SOCKET` | 默认值和 Compose 挂载均固定为 `/var/run/docker.sock`；不能只覆盖环境变量 | | `docker-gateway` | `BROWSER_NETWORK` | `creatorhub_browser` | | `docker-gateway` | `GATEWAY_TOKEN` | 与平台注册值一致，长度 ≥16；`/v1` 全部接口校验 Bearer 令牌 | +| `docker-gateway` | `BROWSER_CDP_URL` | 空；仅本地联调时连接外部 CDP，例如 `http://host.docker.internal:9222` | +| `docker-gateway` | `BROWSER_CDP_TARGET_ID` | 空；外部 CDP 多页面时必须指定抖音页面 ID | | `docker-gateway` | `LOG_LEVEL` | 默认 `info` | 不要把凭据写入仓库或 Compose 文件。 +### 本地 9222 CDP 联调 + +网关支持显式连接已经登录的本地 CDP，不创建或回收 Docker 浏览器。仅用于开发和真实平台链路测试，不应在生产 Compose 中设置。先从 `http://127.0.0.1:9222/json/list` 选择唯一的抖音页面 `id`，再在宿主机启动网关： + +```bash +export LISTEN_ADDR=127.0.0.1:18081 +export GATEWAY_TOKEN=0123456789abcdef +export BROWSER_CDP_URL=http://127.0.0.1:9222 +export BROWSER_CDP_TARGET_ID=<已登录抖音页面的 target id> +export BROWSER_CDP_ALIAS=local-cdp +export BROWSER_CDP_NETWORK_ID=local-cdp +python3 -m cmd.docker_gateway.gateway +``` + +该模式的 generation 固定为 `binding_version=1`、`runtime_id=64 个 0`、`network_id=local-cdp`。生命周期接口明确不可用于此浏览器；身份、受限读取、事件订阅和动作接口仍要求完整 generation，并继续执行 UID 核对。`BROWSER_CDP_TARGET_ID` 必须固定到抖音页面，不能把包含多个网站的 CDP 页面列表交给自动猜测。 + ### P0-lite 停机通知 当前只使用 `creator-hub` 的专用 Logrus JSON logger 作为通知渠道；它固定输出警告，不继承业务 `LOG_LEVEL`。选择它是因为 Compose 已可靠收集服务日志，不需要新增外部账号、凭据、网络重试或通知依赖。仅 `policy_hold` 和 `needs_confirmation` 会产生 `operator attention required`，字段限定为 `event_type`、`reason_code`、账号/任务 ID；不会包含请求头、Secret 引用或凭据值。运维可用下列命令接入现有日志采集或人工查看： diff --git a/docs/e2e-test-plan.md b/docs/e2e-test-plan.md index 3427b6d..f877457 100644 --- a/docs/e2e-test-plan.md +++ b/docs/e2e-test-plan.md @@ -1556,13 +1556,15 @@ G-douyin-base（cookies/get/identity/action/events均在此对象上加各自字 | POST `/v1/browsers/{alias}/douyin/cookies` | G-douyin-base + `"cookies":[{"name":"sessionid","value":"仅在私密请求文件填入","domain":".douyin.com","path":"/"}]` | GATE-04 | | POST `/v1/browsers/{alias}/douyin/get` | G-douyin-base + `"url":"https://www.douyin.com/aweme/v1/web/user/profile/self/?aid=6383&device_platform=webapp"` | GATE-04 | | POST `/v1/browsers/{alias}/douyin/identity` | G-douyin-base + `"expected_account_key":"实际UID或已核实账号标识"` | AC-06、GATE-04 | +| POST `/v1/browsers/{alias}/douyin/messages` | G-douyin-base + `"expected_uid":"真实数字UID","target_uid":"真实对端UID","limit":200`；只读平台 IM 历史，必须返回 `history_source` 与 `history_has_more` | GATE-04、A6 | | POST `/v1/browsers/{alias}/douyin/action` | G-action；confirm默认false，不得直接把预览改true绕开授权 | GATE-05、OP-07 | | POST `/v1/browsers/{alias}/douyin/events` | G-douyin-base + `"expected_uid":"真实数字UID"` | GATE-06 | | GET `/v1/browsers/{alias}/douyin/events` | body=G-douyin-base；query `?limit=100&wait=25`；确认 `?ack=实际delivery_id&limit=1&wait=0` | GATE-06 | | DELETE `/v1/browsers/{alias}/douyin/events` | body=G-douyin-base | GATE-06 | -受限作品/评论读取在 G-douyin-base 中增加 url： +受限作品/评论/目标账号读取在 G-douyin-base 中增加 url： +- 目标账号：`https://www.douyin.com/aweme/v1/web/user/profile/other/?aid=6383&device_platform=webapp&user_id=实际UID`，或将 `user_id` 换成 `sec_user_id`；只允许这两个身份参数之一。 - 作品：`https://www.douyin.com/aweme/v1/web/aweme/post/?sec_user_id=实际sec_user_id&count=20&max_cursor=0`；下一页仅使用真实返回cursor。 - 一级评论：`https://www.douyin.com/aweme/v1/web/comment/list/?aweme_id=实际平台work_key&count=20&cursor=0`；下一页同理。 - 不允许其他host、额外参数、任意CDP或脚本。源码当前允许非负数字分页，不采用旧文档“只能max_cursor=0”的描述。 diff --git a/docs/evidence/douyin-live-20260914.md b/docs/evidence/douyin-live-20260914.md new file mode 100644 index 0000000..9743d9a --- /dev/null +++ b/docs/evidence/douyin-live-20260914.md @@ -0,0 +1,25 @@ +# 抖音真实链路证据（2026-09-14） + +## 运行范围 + +- 分支：`fix/douyin-production-readiness` +- 浏览器：本机已登录 Douyin CDP，外部 gateway 显式绑定 `/jingxuan` 页面 target；未创建 Docker 浏览器。 +- 接收账号 UID：`99491952055` +- 实测私信对象 UID：`2328120603967913` + +## 已验证 + +1. 外部 CDP gateway 健康检查成功。 +2. 身份读取成功：UID `99491952055`，昵称“抖盲求真”，返回 SecUID。 +3. 目标账号读取成功：`/user/profile/other/` 用 `user_id=2328120603967913` 返回目标账号资料及 SecUID。 +4. 私信历史读取成功：`history_source=im_sdk_pull`，会话 `0:1:99491952055:2328120603967913`，返回 6 条消息，`history_has_more=false`；其中包含文本、卡片和评价表单等非文本消息。 +5. 人工私信动作先预览后确认发送成功。实测文本：`CREATORHUB-DM-LIVE-20260914-A`；平台返回成功状态及服务端消息 ID `7685455260048311865`。 +6. 作品媒体下载成功：作品 `7675372156041187705` 返回 `video/mp4`，Base64 解码前长度 `1795724`。 +7. 页面 reload 后立即启动监听会经历 IM SDK 未就绪重试，最终 POST 返回 HTTP 200；随后取得 `boundary_source=douyin_identity_extra_now` 的 baseline。 + +## 未宣称通过 + +- 本次发送为出站私信，未产生可用于验收“平台→后台”入站通知的事件；没有把它当作入站事件证据。 +- 尚未取得可验证、跨进程重启的 Douyin durable event cursor；监听恢复只能建立新的平台时间边界，断连期间事件仍应显示为 gap，不得自动补发。 +- 评论、点赞、关注、转发的真实写操作以及自动响应策略尚未逐项验收；`repost` 仍由 gateway 明确返回不可用。 +- 生产部署版本、真实小号执行、AI/转写供应商和 PostgreSQL 集成尚未用本分支复验。 diff --git a/internal/creator/content.go b/internal/creator/content.go index 8c573c2..838090e 100644 --- a/internal/creator/content.go +++ b/internal/creator/content.go @@ -363,6 +363,21 @@ func (s *Store) GetWork(ctx context.Context, id string) (Work, error) { return result, nil } +func (s *Store) GetWorkByKey(ctx context.Context, platform, workKey string) (Work, error) { + platform, workKey = strings.TrimSpace(platform), strings.TrimSpace(workKey) + if !ValidatePlatform(platform) || workKey == "" { + return Work{}, ErrInvalid + } + result, err := scanWork(s.db.QueryRowContext(ctx, workSelect+` WHERE platform = $1 AND work_key = $2`, platform, workKey)) + if err != nil { + return Work{}, rowError(err) + } + if err := s.loadWorkSources(ctx, &result); err != nil { + return Work{}, err + } + return result, nil +} + func (s *Store) ListWorks(ctx context.Context, filter WorkFilter) ([]Work, error) { query, args := workSelect, make([]any, 0, 8) where := make([]string, 0, 7) @@ -687,6 +702,15 @@ func (s *Store) GetComment(ctx context.Context, id string) (Comment, error) { return result, rowError(err) } +func (s *Store) GetCommentByKey(ctx context.Context, platform, commentKey string) (Comment, error) { + platform, commentKey = strings.TrimSpace(platform), strings.TrimSpace(commentKey) + if !ValidatePlatform(platform) || commentKey == "" { + return Comment{}, ErrInvalid + } + result, err := scanComment(s.db.QueryRowContext(ctx, commentSelect+` WHERE platform = $1 AND comment_key = $2`, platform, commentKey)) + return result, rowError(err) +} + func pageBounds(page, pageSize int) (int, int, error) { if page < 1 || pageSize < 1 || pageSize > 100 { return 0, 0, ErrInvalid diff --git a/internal/creator/store.go b/internal/creator/store.go index 5938db5..fe8f26a 100644 --- a/internal/creator/store.go +++ b/internal/creator/store.go @@ -101,6 +101,8 @@ func Open(ctx context.Context, databaseURL string) (*Store, error) { func (s *Store) Close() error { return s.db.Close() } +func (s *Store) Ping(ctx context.Context) error { return s.db.PingContext(ctx) } + func (s *Store) SetSecretBridge(bridge SecretBridge) { s.secrets = bridge } func (s *Store) automaticExecutionLock(accountID string) *sync.Mutex { diff --git a/internal/douyin/connector.go b/internal/douyin/connector.go index 50fbd9f..232b2f5 100644 --- a/internal/douyin/connector.go +++ b/internal/douyin/connector.go @@ -32,6 +32,10 @@ const ( worksEndpoint = "https://www.douyin.com/aweme/v1/web/aweme/post/" ) +func douyinAPIQuery() url.Values { + return url.Values{"aid": {"6383"}, "device_platform": {"webapp"}} +} + var keyPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,127}$`) var ErrInvalid = errors.New("invalid douyin connector input") @@ -156,7 +160,10 @@ func (connector Connector) Sync(ctx context.Context, request Request) (Result, e evidence := Evidence{Phase: "works", IdentityVerified: true} cursor := int64(0) for page := 0; page < 100; page++ { - query := url.Values{"sec_user_id": {identity.User.SecUID}, "count": {"20"}, "max_cursor": {strconv.FormatInt(cursor, 10)}} + query := douyinAPIQuery() + query.Set("sec_user_id", identity.User.SecUID) + query.Set("count", "20") + query.Set("max_cursor", strconv.FormatInt(cursor, 10)) worksResponse, err := connector.Browser.Get(ctx, worksEndpoint+"?"+query.Encode()) if err != nil { return connector.stop(ctx, request.AccountID, StateNeedsConfirmation, ReasonUnknown, evidence) diff --git a/internal/douyin/creator_collector.go b/internal/douyin/creator_collector.go index 6c30f92..6c72349 100644 --- a/internal/douyin/creator_collector.go +++ b/internal/douyin/creator_collector.go @@ -4,7 +4,6 @@ import ( "context" "encoding/json" "fmt" - "net/url" "regexp" "strconv" "strings" @@ -13,7 +12,10 @@ import ( "git.ipao.vip/rogee/creator-hub/internal/creator" ) -const commentsEndpoint = "https://www.douyin.com/aweme/v1/web/comment/list/" +const ( + commentsEndpoint = "https://www.douyin.com/aweme/v1/web/comment/list/" + profileOtherEndpoint = "https://www.douyin.com/aweme/v1/web/user/profile/other/" +) type CreatorCollector struct { Browser Browser @@ -30,14 +32,27 @@ func (c CreatorCollector) CanonicalSecUID(ctx context.Context, expectedKey strin if err != nil { return "", err } - if err := creatorResponseError(response, "identity"); err != nil { + return canonicalSecUID(response, expectedKey) +} + +func (c CreatorCollector) CanonicalTargetSecUID(ctx context.Context, expectedKey string) (string, error) { + if c.Browser == nil || !keyPattern.MatchString(expectedKey) { + return "", fmt.Errorf("%w: invalid target identity request", ErrInvalid) + } + field := "user_id" + if _, err := strconv.ParseUint(expectedKey, 10, 64); err != nil { + if !strings.HasPrefix(expectedKey, "MS4") { + return "", fmt.Errorf("%w: target requires a Douyin UID or sec UID", ErrInvalid) + } + field = "sec_user_id" + } + query := douyinAPIQuery() + query.Set(field, expectedKey) + response, err := c.Browser.Get(ctx, profileOtherEndpoint+"?"+query.Encode()) + if err != nil { return "", err } - identity, ok := parseIdentity(response.Body) - if !ok || expectedKey != identity.User.UID && expectedKey != identity.User.SecUID && expectedKey != identity.User.UniqueID { - return "", fmt.Errorf("%w: douyin identity mismatch", ErrInvalid) - } - return identity.User.SecUID, nil + return canonicalSecUID(response, expectedKey) } func (c CreatorCollector) VerifyIdentity(ctx context.Context, expectedKey string) error { @@ -48,6 +63,17 @@ func (c CreatorCollector) VerifyIdentity(ctx context.Context, expectedKey string return err } +func canonicalSecUID(response Response, expectedKey string) (string, error) { + if err := creatorResponseError(response, "identity"); err != nil { + return "", err + } + identity, ok := parseIdentity(response.Body) + if !ok || expectedKey != identity.User.UID && expectedKey != identity.User.SecUID && expectedKey != identity.User.UniqueID { + return "", fmt.Errorf("%w: douyin identity mismatch", ErrInvalid) + } + return identity.User.SecUID, nil +} + func (c CreatorCollector) ListWorks(ctx context.Context, accountKey, cursor string) (creator.WorkPage, error) { if c.AccountKey != "" { accountKey = c.AccountKey @@ -59,7 +85,10 @@ func (c CreatorCollector) ListWorks(ctx context.Context, accountKey, cursor stri if cursor != "" { maxCursor = cursor } - query := url.Values{"sec_user_id": {accountKey}, "count": {"20"}, "max_cursor": {maxCursor}} + query := douyinAPIQuery() + query.Set("sec_user_id", accountKey) + query.Set("count", "20") + query.Set("max_cursor", maxCursor) response, err := c.Browser.Get(ctx, worksEndpoint+"?"+query.Encode()) if err != nil { return creator.WorkPage{}, err @@ -110,7 +139,10 @@ func (c CreatorCollector) ListTopLevelComments(ctx context.Context, workKey, cur if cursor != "" { cursorValue = cursor } - query := url.Values{"aweme_id": {workKey}, "count": {"20"}, "cursor": {cursorValue}} + query := douyinAPIQuery() + query.Set("aweme_id", workKey) + query.Set("count", "20") + query.Set("cursor", cursorValue) response, err := c.Browser.Get(ctx, commentsEndpoint+"?"+query.Encode()) if err != nil { return creator.CommentPage{}, err @@ -118,8 +150,12 @@ func (c CreatorCollector) ListTopLevelComments(ctx context.Context, workKey, cur if err := creatorResponseError(response, "comments"); err != nil { return creator.CommentPage{}, err } + return parseCreatorCommentsPage(response.Body) +} + +func parseCreatorCommentsPage(body []byte) (creator.CommentPage, error) { var envelope commentEnvelope - if len(response.Body) > 4<<20 || json.Unmarshal(response.Body, &envelope) != nil || envelope.StatusCode == nil || *envelope.StatusCode != 0 || envelope.HasMore == nil || envelope.Comments == nil { + if len(body) > 4<<20 || json.Unmarshal(body, &envelope) != nil || envelope.StatusCode == nil || *envelope.StatusCode != 0 || envelope.HasMore == nil { return creator.CommentPage{}, fmt.Errorf("%w: invalid douyin comments response", ErrInvalid) } items := make([]creator.CommentInput, 0, len(envelope.Comments)) diff --git a/internal/douyin/creator_collector_test.go b/internal/douyin/creator_collector_test.go index bb37fa9..afffd17 100644 --- a/internal/douyin/creator_collector_test.go +++ b/internal/douyin/creator_collector_test.go @@ -1,6 +1,51 @@ package douyin -import "testing" +import ( + "context" + "net/url" + "testing" +) + +type collectorBrowser struct { + response Response + url string +} + +func (b *collectorBrowser) Get(_ context.Context, target string) (Response, error) { + b.url = target + return b.response, nil +} + +func TestCanonicalTargetSecUIDUsesTargetProfileEndpoint(t *testing.T) { + browser := &collectorBrowser{response: Response{Status: 200, Body: []byte(`{"status_code":0,"user":{"uid":"2328120603967913","sec_uid":"MS4wLjABAAAA9f_a7k0bzVizLYXlpC7R61EIaqJ8Ordug7yp7AB8fGKuuF8Fzqk5_DM-eutXnPIK","unique_id":"96332518739"}}`)}} + collector := CreatorCollector{Browser: browser} + secUID, err := collector.CanonicalTargetSecUID(context.Background(), "2328120603967913") + if err != nil || secUID != "MS4wLjABAAAA9f_a7k0bzVizLYXlpC7R61EIaqJ8Ordug7yp7AB8fGKuuF8Fzqk5_DM-eutXnPIK" { + t.Fatalf("canonical target identity: sec_uid=%q err=%v", secUID, err) + } + parsed, err := url.Parse(browser.url) + if err != nil || parsed.Path != "/aweme/v1/web/user/profile/other/" || parsed.Query().Get("user_id") != "2328120603967913" { + t.Fatalf("unexpected target identity URL: %s", browser.url) + } +} + +func TestCanonicalTargetSecUIDRejectsUniqueIDLookup(t *testing.T) { + browser := &collectorBrowser{} + _, err := (CreatorCollector{Browser: browser}).CanonicalTargetSecUID(context.Background(), "creator_handle") + if err == nil { + t.Fatal("expected a numeric target key to be treated as a UID") + } +} + +func TestParseCreatorCommentsPageAllowsEmptyComments(t *testing.T) { + page, err := parseCreatorCommentsPage([]byte(`{"status_code":0,"has_more":false,"cursor":20,"comments":null}`)) + if err != nil { + t.Fatal(err) + } + if page.HasMore || page.NextCursor != "20" || len(page.Items) != 0 { + t.Fatalf("unexpected empty comments page: %+v", page) + } +} func TestParseCreatorWorksPageKeepsPartialMetadata(t *testing.T) { body := []byte(`{"status_code":0,"has_more":false,"aweme_list":[{"aweme_id":"123","desc":"partial","statistics":{"digg_count":7}}]}`) diff --git a/internal/hub/store.go b/internal/hub/store.go index cabade0..0ea78b4 100644 --- a/internal/hub/store.go +++ b/internal/hub/store.go @@ -144,6 +144,8 @@ func Open(ctx context.Context, databaseURL string) (*Store, error) { func (s *Store) Close() error { return s.db.Close() } +func (s *Store) Ping(ctx context.Context) error { return s.db.PingContext(ctx) } + func (s *Store) SetTaskNotifier(notify taskstate.Notifier) { s.notify = notify } func (s *Store) notifyTransitions(transitions []taskstate.Transition) { diff --git a/internal/phasea/store.go b/internal/phasea/store.go index 77db8c6..7bedecc 100644 --- a/internal/phasea/store.go +++ b/internal/phasea/store.go @@ -229,6 +229,8 @@ func Open(ctx context.Context, databaseURL string) (*Store, error) { func (s *Store) Close() error { return s.db.Close() } +func (s *Store) Ping(ctx context.Context) error { return s.db.PingContext(ctx) } + func (s *Store) SetTaskNotifier(notify taskstate.Notifier) { s.notify = notify } func (s *Store) notifyTransitions(transitions []taskstate.Transition) { diff --git a/web/src/CreatorPages.test.jsx b/web/src/CreatorPages.test.jsx index 7d8ac59..4d7dd9f 100644 --- a/web/src/CreatorPages.test.jsx +++ b/web/src/CreatorPages.test.jsx @@ -317,6 +317,17 @@ describe("creator pages", () => { ); }); + it("subscribes to realtime updates on non-DM operational tabs", async () => { + const creatorSubscribe = vi.fn(() => new Promise(() => {})); + const dataProvider = provider({ creatorSubscribe }); + renderPage(, dataProvider); + await waitFor(() => expect(creatorSubscribe).toHaveBeenCalledTimes(1)); + fireEvent.click(await screen.findByRole("tab", { name: "事件监听" })); + await waitFor(() => expect(creatorSubscribe).toHaveBeenCalledTimes(2)); + fireEvent.click(screen.getByRole("tab", { name: "操作记录" })); + await waitFor(() => expect(creatorSubscribe).toHaveBeenCalledTimes(3)); + }); + it("isolates private messages by account and confirms one durable send", async () => { const conversation = { id: "conversation-a", diff --git a/web/src/CreatorWorkbenchPage.jsx b/web/src/CreatorWorkbenchPage.jsx index ba8c1d9..dced68e 100644 --- a/web/src/CreatorWorkbenchPage.jsx +++ b/web/src/CreatorWorkbenchPage.jsx @@ -123,6 +123,7 @@ export function CreatorWorkbenchPage() { const [messagePending, setMessagePending] = useState(false); const [messageError, setMessageError] = useState(null); const [messageRetry, setMessageRetry] = useState(0); + const [messageSyncing, setMessageSyncing] = useState(false); const [updateStatus, setUpdateStatus] = useState("disconnected"); const [conversationID, setConversationID] = useState(""); const [dmAccountID, setDmAccountID] = useState(""); @@ -300,13 +301,42 @@ export function CreatorWorkbenchPage() { if (sequence === messageSequence.current) setMessagePending(false); }; }, [conversationID, tab, dataProvider, messageRetry, messagePage]); + const syncMessages = async () => { + if (!conversationID || messageSyncing) return; + setMessageSyncing(true); + setMessageError(null); + try { + const result = await dataProvider.creatorSyncConversation(conversationID); + setNotice({ + variant: result?.history_has_more ? "warning" : "success", + text: result?.history_has_more + ? `已同步 ${result?.messages ?? 0} 条平台私信历史，仍有更多内容待同步。` + : `已同步 ${result?.messages ?? 0} 条平台私信历史。`, + }); + setMessageRetry((value) => value + 1); + } catch (syncError) { + setMessageError(syncError); + } finally { + setMessageSyncing(false); + } + }; useEffect(() => { if (reply.target_comment_id) { writeDraft(replyDraftKey(reply.target_comment_id), reply); } }, [reply]); useEffect(() => { - if (tab !== "dms" || typeof dataProvider.creatorSubscribe !== "function") { + const realtimeTabs = new Set([ + "comments", + "leads", + "events", + "dms", + "operations", + ]); + if ( + !realtimeTabs.has(tab) || + typeof dataProvider.creatorSubscribe !== "function" + ) { setUpdateStatus("disconnected"); return undefined; } @@ -321,7 +351,9 @@ export function CreatorWorkbenchPage() { "/creator/updates", () => { load(); - setMessageRetry((value) => value + 1); + if (tab === "dms") { + setMessageRetry((value) => value + 1); + } }, controller.signal, (status) => setUpdateStatus(status), @@ -343,7 +375,7 @@ export function CreatorWorkbenchPage() { window.clearTimeout(retryTimer); setUpdateStatus("disconnected"); }; - }, [tab, dmAccountID, dataProvider]); + }, [tab, dataProvider]); useEffect(() => { if (!dmAccountID || !conversationID) return; const draft = readDraft(dmDraftKey(dmAccountID, conversationID)); @@ -1135,14 +1167,25 @@ export function CreatorWorkbenchPage() { } /> - +
+ + +