feat(creator): 竞品作品同步改用匿名浏览器并限制并发

- 竞品同步不再依赖自有采集账号：每次同步按名额（并发上限 2）创建匿名浏览器，导航至抖音首页等待访客 cookie 稳定后经网关受限 fetch 拉取作品 - 调度器对到期竞品并发同步，名额耗尽时排队等待 - 平台风控语义（实测，与公开爬虫项目一致）：作品列表匿名仅能取第一页（16-20 条），后续页一律空 envelope；空 envelope 视为正常分页终止 - 评论解析适配真实响应：reply_id="0" 为顶层评论、作者取 user 对象、纯图片评论（text 空但带 image_list）记为「[图片]」 - fetch 签名与白名单：注入 byted_acrawler.frontierSign 的 a_bogus，网关放宽响应上限至 8MB 并允许签名参数
2026-09-27 15:43:03 +08:00
parent dbb675c658
commit 221ed65e32
10 changed files with 296 additions and 106 deletions
@@ -332,8 +332,8 @@ func TestCreatorFixtureRoutesPostgres(t *testing.T) {
 			t.Fatalf("competitor %s: %d %s", action, response.Code, response.Body.String())
 		}
 	}
-	if _, err := syncCreatorCompetitorWithClaim(ctx, creatorStore, phaseAStore, hubStore, competitorID, "fixture-account", true); err == nil {
-		t.Fatal("competitor sync without browser unexpectedly succeeded")
+	if _, err := syncCreatorCompetitorWithClaim(ctx, creatorStore, hubStore, competitorID, true); err == nil {
+		t.Fatal("competitor sync without anonymous browser unexpectedly succeeded")
 	}
 	if _, err := previewDouyinCompetitor(ctx, creatorStore, phaseAStore, hubStore, "fixture-account", creator.CompetitorInput{Platform: creator.PlatformDouyin, PlatformAccountKey: "preview-key", Nickname: "Preview", HomepageURL: "https://www.douyin.com/user/preview-key"}); err == nil {
 		t.Fatal("competitor preview without browser unexpectedly succeeded")
@@ -16,6 +16,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync"
 	"time"

 	accountdomain "git.ipao.vip/rogee/creator-hub/internal/account"
@@ -347,13 +348,8 @@ func registerCreatorWithServices(app *fiber.App, store *creator.Store, phaseASto
 		return c.JSON(item)
 	})
 	app.Post("/api/creator/competitors/:id/sync", func(c fiber.Ctx) error {
-		var input struct {
-			AccountID string `json:"account_id"`
-		}
-		if err := decodeCreator(c, &input); err != nil {
-			return creatorError(c, err)
-		}
-		report, err := syncCreatorCompetitor(c.Context(), store, phaseAStore, hubStore, c.Params("id"), input.AccountID)
+		// 强制同步走临时匿名浏览器，不需要自有账号身份。
+		report, err := syncCreatorCompetitor(c.Context(), store, hubStore, c.Params("id"))
 		if err != nil {
 			return creatorError(c, err)
 		}
@@ -1371,7 +1367,8 @@ func verifyCreatorAccount(ctx context.Context, store *creator.Store, phaseAStore
 func (browser creatorGatewayBrowser) Get(ctx context.Context, target string) (douyin.Response, error) {
 	payload := gatewayGenerationPayload(browser.environment)
 	payload["url"] = target
-	status, body, err := gatewayCall(ctx, browser.gateway, http.MethodPost, "/v1/browsers/"+url.PathEscape(browser.environment.Alias)+"/douyin/get", payload, 30*time.Second)
+	// 网关側 socket 读超时 5s，1MB+ 响应体的全量传输需要分多次读：客户端超时须远大于服务端单次读超时。
+	status, body, err := gatewayCallWithLimit(ctx, browser.gateway, http.MethodPost, "/v1/browsers/"+url.PathEscape(browser.environment.Alias)+"/douyin/get", payload, 120*time.Second, largeGatewayResponseLimit)
 	if err != nil {
 		return douyin.Response{}, err
 	}
@@ -1528,9 +1525,16 @@ func previewCompetitorShare(ctx context.Context, store *creator.Store, phaseASto
 type anonymousBrowserLease struct {
 	gateway     hub.Gateway
 	environment hub.EnvironmentContext
+	// slot 占用全局匿名浏览器并发名额；非 nil 时 close 必须释放，保证任何路径都不会泄漏名额。
+	slot chan struct{}
 }

 func (lease anonymousBrowserLease) close() error {
+	defer func() {
+		if lease.slot != nil {
+			<-lease.slot
+		}
+	}()
 	cleanupContext, cancel := context.WithTimeout(context.Background(), 90*time.Second)
 	defer cancel()
 	payload := gatewayCleanupGenerationPayload(lease.environment)
@@ -1575,6 +1579,11 @@ func randomGateway(gateways []hub.Gateway) hub.Gateway {
 	return gateways[rand.Intn(len(gateways))]
 }

+// 匿名浏览器全局并发上限：每个实例都是一个完整的指纹浏览器，同时拉起过多会耗尽机器资源。
+// 所有路径（竞品同步、share 链接解析、调度器并发）共享同一份名额。
+// ponytail: 固定值 2 够本地单机使用；需要按机器规格调整时再提升为配置项。
+var anonymousBrowserSlots = make(chan struct{}, 2)
+
 func newAnonymousBrowser(ctx context.Context, store *hub.Store) (anonymousBrowserLease, error) {
 	if store == nil {
 		return anonymousBrowserLease{}, creator.ErrUnavailable
@@ -1591,6 +1600,13 @@ func newAnonymousBrowser(ctx context.Context, store *hub.Store) (anonymousBrowse
 		return anonymousBrowserLease{}, fmt.Errorf("%w: anonymous browser gateway is not configured", creator.ErrUnavailable)
 	}

+	// 拿不到并发名额就在此等待（ctx 取消可退出），保证浏览器实例数永不起过上限。
+	select {
+	case anonymousBrowserSlots <- struct{}{}:
+	case <-ctx.Done():
+		return anonymousBrowserLease{}, ctx.Err()
+	}
+
 	gateway := randomGateway(gateways)
 	var template hub.Env
 	for _, candidate := range envs {
@@ -1619,7 +1635,7 @@ func newAnonymousBrowser(ctx context.Context, store *hub.Store) (anonymousBrowse
 		var created runtimeStatus
 		if json.Unmarshal(body, &created) == nil && created.Alias == environment.Alias && created.BindingVersion == environment.BindingVersion && created.State == "running" && validCreatedRuntime(created, environment, true) {
 			environment.RuntimeID, environment.RuntimeNetworkID = created.ID, created.NetworkID
-			return anonymousBrowserLease{gateway: gateway, environment: environment}, nil
+			return anonymousBrowserLease{gateway: gateway, environment: environment, slot: anonymousBrowserSlots}, nil
 		}
 		callErr = errors.New("gateway returned an invalid anonymous runtime generation")
 	} else if callErr == nil {
@@ -1628,6 +1644,7 @@ func newAnonymousBrowser(ctx context.Context, store *hub.Store) (anonymousBrowse
 		callErr = gatewayUnreachable(callErr)
 	}
 	cleanupErr := cleanupAnonymousRuntimeAfterCreateFailure(ctx, gateway, environment)
+	<-anonymousBrowserSlots
 	return anonymousBrowserLease{}, errors.Join(callErr, cleanupErr)
 }

@@ -1750,16 +1767,16 @@ func previewDouyinCompetitor(ctx context.Context, store *creator.Store, phaseASt
 	}, nil
 }

-func syncCreatorCompetitor(ctx context.Context, store *creator.Store, phaseAStore *accountdomain.Store, hubStore *hub.Store, competitorID, accountID string) (creator.CollectionReport, error) {
-	return syncCreatorCompetitorWithClaim(ctx, store, phaseAStore, hubStore, competitorID, accountID, true)
+func syncCreatorCompetitor(ctx context.Context, store *creator.Store, hubStore *hub.Store, competitorID string) (creator.CollectionReport, error) {
+	return syncCreatorCompetitorWithClaim(ctx, store, hubStore, competitorID, true)
 }

-func syncCreatorCompetitorDue(ctx context.Context, store *creator.Store, phaseAStore *accountdomain.Store, hubStore *hub.Store, competitorID, accountID string) (creator.CollectionReport, error) {
-	return syncCreatorCompetitorWithClaim(ctx, store, phaseAStore, hubStore, competitorID, accountID, false)
+func syncCreatorCompetitorDue(ctx context.Context, store *creator.Store, hubStore *hub.Store, competitorID string) (creator.CollectionReport, error) {
+	return syncCreatorCompetitorWithClaim(ctx, store, hubStore, competitorID, false)
 }

-func syncCreatorCompetitorWithClaim(ctx context.Context, store *creator.Store, phaseAStore *accountdomain.Store, hubStore *hub.Store, competitorID, accountID string, force bool) (creator.CollectionReport, error) {
-	if store == nil || phaseAStore == nil || hubStore == nil || accountID == "" {
+func syncCreatorCompetitorWithClaim(ctx context.Context, store *creator.Store, hubStore *hub.Store, competitorID string, force bool) (creator.CollectionReport, error) {
+	if store == nil || hubStore == nil || competitorID == "" {
 		return creator.CollectionReport{}, creator.ErrUnavailable
 	}
 	competitor, err := store.GetCompetitor(ctx, competitorID)
@@ -1778,7 +1795,7 @@ func syncCreatorCompetitorWithClaim(ctx context.Context, store *creator.Store, p
 	if !claimed {
 		return creator.CollectionReport{}, creator.ErrConflict
 	}
-	// blocked（环境暂不可用）也必须排下一次重试：否则 next_sync_at 置空后定时获取对该账号永久失效。
+	// blocked（网关不可用等环境暂不可用）也必须排下一次重试：否则 next_sync_at 置空后定时获取对该账号永久失效。
 	nextRetry := func() *time.Time {
 		nextBase := now
 		if competitor.NextSyncAt != nil {
@@ -1797,46 +1814,23 @@ func syncCreatorCompetitorWithClaim(ctx context.Context, store *creator.Store, p
 	if competitor.Platform != creator.PlatformDouyin {
 		return blocked(fmt.Errorf("%w: unsupported creator platform %s", creator.ErrUnavailable, competitor.Platform))
 	}
-	account, err := phaseAStore.GetAccount(ctx, accountID)
+	// 竞品作品是公开数据：用临时匿名浏览器采集，不依赖任何已登录自有账号；
+	// 并发名额由 newAnonymousBrowser 全局信号量控制（含调度器并发与手动同步）。
+	lease, err := newAnonymousBrowser(ctx, hubStore)
 	if err != nil {
-		return blocked(err)
-	}
-	if account.Platform != competitor.Platform || account.AuthorizationStatus != "authorized" {
-		return blocked(creator.ErrConflict)
-	}
-	profile, err := store.GetAccountProfile(ctx, accountID)
-	if err != nil {
-		return blocked(err)
-	}
-	if (profile.BusinessStatus != "normal" && profile.BusinessStatus != "muted") || profile.LoginStatus != "logged_in" {
-		return blocked(creator.ErrConflict)
-	}
-	environment, err := hubStore.GetEnvironmentContextForAccount(ctx, accountID)
-	if err != nil {
-		return blocked(fmt.Errorf("%w: account environment unavailable: %v", creator.ErrUnavailable, err))
-	}
-	if environment.RuntimeID == "" || environment.RuntimeNetworkID == "" || environment.BindingVersion <= 0 {
-		return blocked(fmt.Errorf("%w: account runtime is not running", creator.ErrUnavailable))
-	}
-	gateway, err := hubStore.GetGateway(ctx, environment.Gateway)
-	if err != nil {
-		return blocked(fmt.Errorf("%w: gateway unavailable: %v", creator.ErrUnavailable, err))
-	}
-	useCtx, runtimeUse, err := beginRuntimeUseForEnvironment(ctx, hubStore, environment, "task", "creator-sync-"+competitor.ID)
-	if err != nil {
-		return blocked(fmt.Errorf("%w: runtime use unavailable: %v", creator.ErrUnavailable, err))
-	}
-	collector, _, err := newCreatorCollector(useCtx, competitor.Platform, gateway, environment, account.PlatformAccountKey, competitor.PlatformAccountKey, competitor.HomepageURL, creator.SourceCompetitor, competitor.ID)
-	if err != nil {
-		return blocked(errors.Join(fmt.Errorf("%w: account identity verification failed: %v", creator.ErrConflict, err), runtimeUse.Close()))
+		return blocked(fmt.Errorf("%w: anonymous browser unavailable: %v", creator.ErrUnavailable, err))
 	}
+	collector := douyinCollector(
+		creatorGatewayBrowser{gateway: lease.gateway, environment: lease.environment},
+		competitor.PlatformAccountKey, creator.SourceCompetitor, competitor.ID,
+	)
 	collectionNow := now
 	if competitor.NextSyncAt != nil && !competitor.NextSyncAt.After(now) {
 		collectionNow = competitor.NextSyncAt.UTC()
 	}
-	report, collectErr := store.CollectSource(useCtx, competitor.Platform, creator.SourceCompetitor, competitor.ID, collector, collectionNow)
-	if releaseErr := runtimeUse.Close(); releaseErr != nil {
-		collectErr = errors.Join(collectErr, releaseErr)
+	report, collectErr := store.CollectSource(ctx, competitor.Platform, creator.SourceCompetitor, competitor.ID, collector, collectionNow)
+	if closeErr := lease.close(); closeErr != nil {
+		collectErr = errors.Join(collectErr, closeErr)
 	}
 	if collectErr != nil {
 		nextBase := now
@@ -1931,36 +1925,18 @@ func runCreatorScheduleOnce(ctx context.Context, store *creator.Store, phaseASto
 	if err != nil {
 		return err
 	}
+	var wg sync.WaitGroup
 	for _, competitor := range competitors {
-		accountID, err := creatorCollectionAccount(ctx, store, phaseAStore, hubStore, competitor.Platform)
-		if err != nil {
-			leaseToken, claimed, claimErr := store.ClaimCompetitorSync(ctx, competitor.ID, false, now)
-			if claimErr != nil {
-				logrus.WithError(claimErr).WithField("competitor_id", competitor.ID).Warn("creator competitor sync claim failed")
-				continue
+		// 多个到期账号并发同步；匿名浏览器名额由全局信号量限制（上限 2），不会同时拉起过多实例。
+		wg.Add(1)
+		go func(competitor creator.Competitor) {
+			defer wg.Done()
+			if _, err := syncCreatorCompetitorDue(ctx, store, hubStore, competitor.ID); err != nil {
+				logrus.WithError(err).WithField("competitor_id", competitor.ID).Warn("creator competitor scheduled sync failed")
 			}
-			if claimed {
-				// blocked 也排下一次重试（按新作品间隔），避免无采集账号时永久卡死。
-				nextBase := now
-				if competitor.NextSyncAt != nil {
-					nextBase = competitor.NextSyncAt.UTC()
-				}
-				next := creator.NextFixedRun(nextBase, now, time.Duration(settings.NewWorkIntervalSeconds)*time.Second)
-				var nextAt *time.Time
-				if !next.IsZero() {
-					nextAt = &next
-				}
-				if markErr := store.MarkCompetitorSync(ctx, competitor.ID, leaseToken, "blocked", "", err.Error(), nextAt); markErr != nil {
-					logrus.WithError(markErr).WithField("competitor_id", competitor.ID).Warn("creator competitor sync block update failed")
-				}
-			}
-			logrus.WithError(err).WithField("competitor_id", competitor.ID).Warn("creator competitor sync blocked")
-			continue
-		}
-		if _, err := syncCreatorCompetitorDue(ctx, store, phaseAStore, hubStore, competitor.ID, accountID); err != nil {
-			logrus.WithError(err).WithField("competitor_id", competitor.ID).Warn("creator competitor scheduled sync failed")
-		}
+		}(competitor)
 	}
+	wg.Wait()
 	ownedAccounts, err := store.ListDueOwnedAccounts(ctx, now, settings.NewWorkIntervalSeconds)
 	if err != nil {
 		return err
@@ -144,9 +144,9 @@ func TestCreatorSchedulerAndPreviewGuards(t *testing.T) {
 			_, err := previewDouyinCompetitor(ctx, nil, nil, nil, "account", creator.CompetitorInput{})
 			return err
 		}},
-		{"sync competitor", func() error { _, err := syncCreatorCompetitor(ctx, nil, nil, nil, "competitor", "account"); return err }},
+		{"sync competitor", func() error { _, err := syncCreatorCompetitor(ctx, nil, nil, "competitor"); return err }},
 		{"sync due competitor", func() error {
-			_, err := syncCreatorCompetitorDue(ctx, nil, nil, nil, "competitor", "account")
+			_, err := syncCreatorCompetitorDue(ctx, nil, nil, "competitor")
 			return err
 		}},
 		{"sync owned", func() error { return syncCreatorOwned(ctx, nil, nil, nil, "account", time.Now()) }},
@@ -54,7 +54,7 @@ func TestCreatorWriteRoutesRejectMalformedInputBeforeStoreAccess(t *testing.T) {
 		{method: http.MethodPut, path: "/api/creator/strategies/strategy-1"},
 		{method: http.MethodPost, path: "/api/creator/competitor-share-jobs"},
 		{method: http.MethodPut, path: "/api/creator/competitors/competitor-1"},
-		{method: http.MethodPost, path: "/api/creator/competitors/competitor-1/sync"},
+		// sync 路由不解析请求体（强制同步走匿名浏览器，无输入字段），不适用畸形 body 校验。
 		{method: http.MethodPost, path: "/api/creator/test/works"},
 		{method: http.MethodPost, path: "/api/creator/works/work-1/metrics"},
 		{method: http.MethodPost, path: "/api/creator/works/work-1/material/rewrite/confirm"},
@@ -92,6 +92,40 @@ func TestAnonymousBrowserLeasePurgesRuntimeAndProfile(t *testing.T) {
 	}
 }

+func TestAnonymousBrowserLeaseReleasesConcurrencySlot(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodDelete || r.URL.Path != "/v1/browsers/anon-test" {
+			t.Fatalf("unexpected cleanup request: %s %s", r.Method, r.URL.Path)
+		}
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	defer server.Close()
+
+	slots := make(chan struct{}, 1)
+	slots <- struct{}{}
+	lease := anonymousBrowserLease{gateway: hub.Gateway{Endpoint: server.URL, Token: "test-token"}, slot: slots, environment: hub.EnvironmentContext{Env: hub.Env{Alias: "anon-test"}, BindingVersion: 1}}
+	if err := lease.close(); err != nil {
+		t.Fatal(err)
+	}
+	// 释放 = 名额回到池里（能再次非阻塞占用）。
+	select {
+	case slots <- struct{}{}:
+	default:
+		t.Fatal("close must release the anonymous browser concurrency slot")
+	}
+
+	// 创建失败路径也必须释放名额，否则失败几次后永久占满。
+	failed := anonymousBrowserLease{gateway: hub.Gateway{Endpoint: "http://127.0.0.1:1", Token: "test-token"}, slot: slots}
+	if err := failed.close(); err == nil {
+		t.Fatal("cleanup against unreachable gateway must fail")
+	}
+	select {
+	case slots <- struct{}{}:
+	default:
+		t.Fatal("failed close must still release the anonymous browser concurrency slot")
+	}
+}
+
 func TestDouyinWorkKeyFromURL(t *testing.T) {
 	for _, test := range []struct {
 		url  string