rogee/creator-hub
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(platform): 平台收敛为抖音——移除小红书全链路，监控账号新增画像字段与卡片列表

Browse Source 
- 移除小红书：internal/platform/xiaohongshu、controlplane XHS 分支与路由、
  browser_gateway XHS 平台模块与处理器、前端平台选项；migration 038 清理
  XHS 数据并将 creator_* 平台 CHECK 收紧为 douyin
- creator_competitor 新增 follower_count/following_count/aweme_count（仅
  接口定义返回，暂不采集回填）；列表接口返回 work_count 与
  latest_published_at（creator_work_source 聚合 MAX(published_at)）
- 导入页账号名称列展示作者头像（share job 列表带出 avatar_url）
- 监控账号页改 antd Listy 卡片列表（头像+粉丝/关注/作品/最近发布/同步状态）；
  AccountManagementList 收敛为 owned 专用
- 环境模块清理 browser_version 遗留：migration 034 补齐删列，修复 store.go
  中 hub 别名悬空引用，测试 fixture 对齐新 schema
This commit is contained in:
rogee
2026-09-24 19:30:28 +08:00
parent 08e813c0ca
commit 6cd4cdc57f
43 changed files with 690 additions and 2665 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/account/store.go
+10 -10
View File
@@ -425,7 +425,7 @@ func validAccount(account Account) bool {
return false
}
switch account.Platform {
case "douyin", "xiaohongshu", "wechat-official", "kuaishou":
case "douyin":
default:
return false
}
@@ -1297,7 +1297,7 @@ func (s *Store) ResumeTask(ctx context.Context, taskID string) error {
JOIN content_draft draft ON draft.id = task.draft_id
JOIN confirmation confirmation ON confirmation.id = task.confirmation_id
JOIN environment_binding binding ON binding.account_id = task.account_id
JOIN browser_env environment ON hub.alias = binding.browser_env_alias
JOIN browser_env environment ON environment.alias = binding.browser_env_alias
LEFT JOIN network_exit network ON network.id = binding.network_exit_id
JOIN runtime_instance runtime ON runtime.binding_id = binding.id AND runtime.released_at IS NULL
WHERE task.id = $1 AND task.state IN ('policy_hold', 'needs_confirmation')
@@ -1368,7 +1368,7 @@ func taskReadinessReason(ctx context.Context, queryer rowQuerier, taskID string)
OR confirmation.draft_id <> task.draft_id OR confirmation.draft_version <> task.draft_version
OR confirmation.version <> task.confirmation_version THEN 'confirmation_version_changed'
WHEN binding.id IS NULL THEN 'binding_missing'
WHEN hub.alias IS NULL THEN 'environment_missing'
WHEN environment.alias IS NULL THEN 'environment_missing'
WHEN binding.network_exit_id IS NOT NULL AND network.id IS NULL THEN 'exit_missing'
WHEN binding.network_exit_id IS NOT NULL AND network.health_status <> 'healthy' THEN 'exit_unhealthy'
WHEN binding.runtime_cleanup_pending THEN 'runtime_stop_pending'
@@ -1382,7 +1382,7 @@ func taskReadinessReason(ctx context.Context, queryer rowQuerier, taskID string)
LEFT JOIN content_draft draft ON draft.id = task.draft_id
LEFT JOIN confirmation confirmation ON confirmation.id = task.confirmation_id
LEFT JOIN environment_binding binding ON binding.account_id = task.account_id
LEFT JOIN browser_env environment ON hub.alias = binding.browser_env_alias
LEFT JOIN browser_env environment ON environment.alias = binding.browser_env_alias
LEFT JOIN network_exit network ON network.id = binding.network_exit_id
LEFT JOIN runtime_instance runtime ON runtime.binding_id = binding.id AND runtime.released_at IS NULL
WHERE task.id = $1`, taskID).Scan(&reason)
@@ -1495,7 +1495,7 @@ func (s *Store) claim(ctx context.Context, workerID string) (Execution, error) {
JOIN content_draft d ON d.id = t.draft_id
JOIN confirmation c ON c.id = t.confirmation_id
JOIN environment_binding binding ON binding.account_id = a.id
JOIN browser_env environment ON hub.alias = binding.browser_env_alias
JOIN browser_env environment ON environment.alias = binding.browser_env_alias
LEFT JOIN network_exit network ON network.id = binding.network_exit_id
JOIN runtime_instance runtime ON runtime.binding_id = binding.id
AND runtime.released_at IS NULL AND runtime.lease_until > now() AND runtime.binding_version = binding.version
@@ -1561,13 +1561,13 @@ func (s *Store) complete(ctx context.Context, workerID string, execution Executi
OR confirmation.account_version <> task.account_version OR confirmation.draft_id <> task.draft_id
OR confirmation.draft_version <> task.draft_version OR confirmation.version <> task.confirmation_version THEN 'confirmation_version_changed'
WHEN binding.id IS NULL THEN 'binding_missing'
WHEN hub.alias IS NULL THEN 'environment_missing'
WHEN environment.alias IS NULL THEN 'environment_missing'
WHEN binding.network_exit_id IS NOT NULL AND network.id IS NULL THEN 'exit_missing'
WHEN binding.network_exit_id IS NOT NULL AND network.health_status <> 'healthy' THEN 'exit_unhealthy'
WHEN binding.runtime_cleanup_pending THEN 'runtime_stop_pending'
WHEN runtime.id IS NULL OR runtime.lease_until <= now() THEN 'runtime_missing'
WHEN claim.binding_version IS DISTINCT FROM binding.version
OR claim.browser_env_alias IS DISTINCT FROM hub.alias
OR claim.browser_env_alias IS DISTINCT FROM environment.alias
OR claim.network_exit_id IS DISTINCT FROM network.id
OR claim.runtime_instance_id IS DISTINCT FROM runtime.id
OR runtime.binding_version IS DISTINCT FROM binding.version THEN 'task_result_uncertain'
@@ -1713,7 +1713,7 @@ func quarantineInvalid(ctx context.Context, tx *sql.Tx) ([]taskstate.Transition,
OR confirmation.draft_id <> t.draft_id OR confirmation.draft_version <> t.draft_version
OR confirmation.version <> t.confirmation_version THEN 'confirmation_version_changed'
WHEN binding.id IS NULL THEN 'binding_missing'
WHEN hub.alias IS NULL THEN 'environment_missing'
WHEN environment.alias IS NULL THEN 'environment_missing'
WHEN binding.network_exit_id IS NOT NULL AND network.id IS NULL THEN 'exit_missing'
WHEN binding.network_exit_id IS NOT NULL AND network.health_status <> 'healthy' THEN 'exit_unhealthy'
WHEN binding.runtime_cleanup_pending THEN 'runtime_stop_pending'
@@ -1726,7 +1726,7 @@ func quarantineInvalid(ctx context.Context, tx *sql.Tx) ([]taskstate.Transition,
LEFT JOIN content_draft draft ON draft.id = t.draft_id
LEFT JOIN confirmation confirmation ON confirmation.id = t.confirmation_id
LEFT JOIN environment_binding binding ON binding.account_id = t.account_id
LEFT JOIN browser_env environment ON hub.alias = binding.browser_env_alias
LEFT JOIN browser_env environment ON environment.alias = binding.browser_env_alias
LEFT JOIN network_exit network ON network.id = binding.network_exit_id
LEFT JOIN runtime_instance runtime ON runtime.binding_id = binding.id AND runtime.released_at IS NULL
WHERE t.state = 'queued' AND (
@@ -1736,7 +1736,7 @@ func quarantineInvalid(ctx context.Context, tx *sql.Tx) ([]taskstate.Transition,
OR confirmation.id IS NULL OR confirmation.account_id <> t.account_id
OR confirmation.account_version <> t.account_version OR confirmation.draft_id <> t.draft_id
OR confirmation.draft_version <> t.draft_version OR confirmation.version <> t.confirmation_version
OR binding.id IS NULL OR hub.alias IS NULL
OR binding.id IS NULL OR environment.alias IS NULL
OR (binding.network_exit_id IS NOT NULL AND (network.id IS NULL OR network.health_status <> 'healthy'))
OR binding.runtime_cleanup_pending OR runtime.id IS NULL
OR runtime.binding_version IS DISTINCT FROM binding.version OR runtime.lease_until <= now()
internal/account/store_test.go
+12 -14
View File
@@ -72,7 +72,7 @@ func TestValidationRejectsInvalidInputsBeforePersistence(t *testing.T) {
CredentialKey: "creatorhub/account-a/cookies",
}
credentials := &testCredentialBridge{values: map[string]string{}}
for _, platform := range []string{"douyin", "xiaohongshu", "wechat-official", "kuaishou"} {
for _, platform := range []string{"douyin"} {
account := valid
account.Platform = platform
if !validAccount(account) {
@@ -170,7 +170,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
if _, err := store.db.ExecContext(ctx, `
TRUNCATE audit_event, execution_attempt, operation_task, confirmation, content_draft,
runtime_instance, environment_binding, network_exit, social_account, credential_reference,
browser_env, browser_version, gateway RESTART IDENTITY CASCADE`); err != nil {
browser_env, gateway RESTART IDENTITY CASCADE`); err != nil {
t.Fatal(err)
}
@@ -215,13 +215,10 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
t.Fatal(err)
}
if _, err := store.db.ExecContext(ctx, `
INSERT INTO browser_version (version, browser_path) VALUES ('1', '/opt/creatorhub/browsers/1')`); err != nil {
t.Fatal(err)
}
if _, err := store.db.ExecContext(ctx, `
INSERT INTO browser_env (alias, name, gateway_name, browser_version, fingerprint) VALUES
('account-a', 'Account A', 'test-gateway', '1', '{"seed":1}'),
('account-b', 'Account B', 'test-gateway', '1', '{"seed":2}');
INSERT INTO browser_image (version, image_ref) VALUES ('148.0.7778.215', 'git.ipao.vip/rogee/fingerprint-chromium:148.0.7778.215') ON CONFLICT (version) DO NOTHING;
INSERT INTO browser_env (alias, name, gateway_name, fingerprint, image_version) VALUES
('account-a', 'Account A', 'test-gateway', '{"seed":1}', '148.0.7778.215'),
('account-b', 'Account B', 'test-gateway', '{"seed":2}', '148.0.7778.215');
INSERT INTO network_exit (id, protocol, host, port, health_status)
VALUES ('exit-shared', 'socks5', '127.0.0.1', 1080, 'healthy');
INSERT INTO environment_binding (id, account_id, browser_env_alias, network_exit_id) VALUES
@@ -754,7 +751,7 @@ func TestCreateAccountWithoutCookiesSkipsCredentialStore(t *testing.T) {
if _, err := store.db.ExecContext(ctx, `
TRUNCATE audit_event, execution_attempt, operation_task, confirmation, content_draft,
runtime_instance, environment_binding, network_exit, social_account, credential_reference,
browser_env, browser_version, gateway RESTART IDENTITY CASCADE`); err != nil {
browser_env, gateway RESTART IDENTITY CASCADE`); err != nil {
t.Fatal(err)
}
credentials := &testCredentialBridge{values: map[string]string{}}
@@ -785,7 +782,7 @@ func TestAccountCredentialCommitResult(t *testing.T) {
if _, err := store.db.ExecContext(ctx, `
TRUNCATE audit_event, execution_attempt, operation_task, confirmation, content_draft,
runtime_instance, environment_binding, network_exit, social_account, credential_reference,
browser_env, browser_version, gateway RESTART IDENTITY CASCADE`); err != nil {
browser_env, gateway RESTART IDENTITY CASCADE`); err != nil {
t.Fatal(err)
}
credentials := &testCredentialBridge{values: map[string]string{}}
@@ -886,8 +883,8 @@ func seedGateTask(t *testing.T, store *Store, suffix, accountStatus, authorizati
}
if binding {
exitID := "exit-gate-" + suffix
if _, err := tx.ExecContext(ctx, `INSERT INTO browser_env (alias, name, gateway_name, browser_version, fingerprint)
VALUES ($1, $1, 'test-gateway', '1', '{"seed":3}')`, accountID); err != nil {
if _, err := tx.ExecContext(ctx, `INSERT INTO browser_env (alias, name, gateway_name, fingerprint, image_version)
VALUES ($1, $1, 'test-gateway', jsonb_build_object('seed', hashtext($1)), '148.0.7778.215')`, accountID); err != nil {
t.Fatal(err)
}
if _, err := tx.ExecContext(ctx, `INSERT INTO network_exit (id, protocol, host, port, health_status)
@@ -959,7 +956,8 @@ func applyHubMigrationsForPhaseATest(t *testing.T, store *Store) {
{6, "006_runtime_cleanup.sql"}, {7, "007_runtime_binding_version.sql"}, {8, "008_runtime_cleanup_generation.sql"},
{9, "009_runtime_cleanup_compatibility.sql"}, {10, "010_runtime_network_generation.sql"}, {11, "011_task_recovery.sql"},
{12, "012_task_recovery_compatibility.sql"}, {13, "013_account_creation.sql"}, {14, "014_account_creation_compatibility.sql"},
{15, "015_gateway_rename_cascade.sql"}, {16, "016_network_exit_plain_credentials.sql"}, {17, "017_native_browser_versions.sql"}} {
{15, "015_gateway_rename_cascade.sql"}, {16, "016_network_exit_plain_credentials.sql"}, {17, "017_native_browser_versions.sql"},
{33, "033_unique_fingerprint_seed.sql"}, {34, "034_gateway_browser_default.sql"}} {
var applied bool
if err := store.db.QueryRow(`SELECT EXISTS (SELECT 1 FROM schema_migration WHERE version = $1)`, migrationFile.version).Scan(&applied); err != nil {
t.Fatal(err)
internal/controlplane/api/app_migrated_test.go
-2
View File
@@ -139,8 +139,6 @@ func TestCreatorRouteValidationCoverage(t *testing.T) {
{http.MethodPost, "/api/creator/competitors/missing/pause"},
{http.MethodPost, "/api/creator/competitors/missing/resume"},
{http.MethodPost, "/api/creator/competitors/missing/sync"},
{http.MethodPost, "/api/creator/xiaohongshu/search"},
{http.MethodPost, "/api/creator/xiaohongshu/detail"},
{http.MethodPost, "/api/creator/works/missing/metrics"},
{http.MethodPost, "/api/creator/works/missing/material/select"},
{http.MethodPost, "/api/creator/works/missing/material/process"},
internal/controlplane/api/creator.go
+8 -108
View File
@@ -22,7 +22,6 @@ import (
"git.ipao.vip/rogee/creator-hub/internal/creator"
hub "git.ipao.vip/rogee/creator-hub/internal/environment"
douyin "git.ipao.vip/rogee/creator-hub/internal/platform/douyin"
xiaohongshu "git.ipao.vip/rogee/creator-hub/internal/platform/xiaohongshu"
"github.com/gofiber/fiber/v3"
"github.com/sirupsen/logrus"
)
@@ -267,7 +266,7 @@ func registerCreatorWithServices(app *fiber.App, store *creator.Store, phaseASto
})
app.Get("/api/creator/competitors", func(c fiber.Ctx) error {
items, err := store.ListCompetitors(c.Context(), c.Query("platform"))
items, err := store.ListCompetitorsWithProfile(c.Context(), c.Query("platform"))
if err != nil {
return creatorError(c, err)
}
@@ -361,63 +360,6 @@ func registerCreatorWithServices(app *fiber.App, store *creator.Store, phaseASto
return c.Status(fiber.StatusAccepted).JSON(report)
})
app.Post("/api/creator/xiaohongshu/search", func(c fiber.Ctx) error {
var input struct {
AccountID string `json:"account_id"`
Query string `json:"query"`
Page int `json:"page"`
}
if err := decodeCreator(c, &input); err != nil {
return creatorError(c, err)
}
if input.Page == 0 {
input.Page = 1
}
environment, err := hubStore.GetEnvironmentContextForAccount(c.Context(), input.AccountID)
if err != nil {
return creatorError(c, err)
}
useCtx, runtimeUse, err := beginRuntimeUseForEnvironment(c.Context(), hubStore, environment, "task", "creator-xhs-read-"+input.AccountID)
if err != nil {
return creatorError(c, fmt.Errorf("%w: runtime use unavailable: %v", creator.ErrUnavailable, err))
}
collector, err := newXiaohongshuReadCollector(useCtx, store, phaseAStore, hubStore, input.AccountID, creator.SourceOwned, input.AccountID)
if err != nil {
return creatorError(c, errors.Join(err, runtimeUse.Close()))
}
page, callErr := collector.SearchNotes(useCtx, input.Query, input.Page)
if closeErr := runtimeUse.Close(); callErr != nil || closeErr != nil {
return creatorError(c, errors.Join(callErr, closeErr))
}
return c.JSON(page)
})
app.Post("/api/creator/xiaohongshu/detail", func(c fiber.Ctx) error {
var input struct {
AccountID string `json:"account_id"`
URL string `json:"url"`
}
if err := decodeCreator(c, &input); err != nil {
return creatorError(c, err)
}
environment, err := hubStore.GetEnvironmentContextForAccount(c.Context(), input.AccountID)
if err != nil {
return creatorError(c, err)
}
useCtx, runtimeUse, err := beginRuntimeUseForEnvironment(c.Context(), hubStore, environment, "task", "creator-xhs-detail-"+input.AccountID)
if err != nil {
return creatorError(c, fmt.Errorf("%w: runtime use unavailable: %v", creator.ErrUnavailable, err))
}
collector, err := newXiaohongshuReadCollector(useCtx, store, phaseAStore, hubStore, input.AccountID, creator.SourceOwned, input.AccountID)
if err != nil {
return creatorError(c, errors.Join(err, runtimeUse.Close()))
}
item, callErr := collector.GetNoteDetail(useCtx, input.URL)
if closeErr := runtimeUse.Close(); callErr != nil || closeErr != nil {
return creatorError(c, errors.Join(callErr, closeErr))
}
return c.JSON(item)
})
app.Get("/api/creator/works", func(c fiber.Ctx) error {
filter, err := workFilter(c)
if err != nil {
@@ -1393,7 +1335,7 @@ func verifyCreatorAccount(ctx context.Context, store *creator.Store, phaseAStore
if err != nil {
return creator.LoginResult{}, err
}
if account.Platform != profile.Platform || (account.Platform != creator.PlatformDouyin && account.Platform != creator.PlatformXiaohongshu) || account.AuthorizationStatus != "authorized" || profile.PlatformAccountKey == "" || account.PlatformAccountKey != profile.PlatformAccountKey {
if account.Platform != profile.Platform || account.Platform != creator.PlatformDouyin || account.AuthorizationStatus != "authorized" || profile.PlatformAccountKey == "" || account.PlatformAccountKey != profile.PlatformAccountKey {
return creator.LoginResult{}, creator.ErrConflict
}
environment, err := hubStore.GetEnvironmentContextForAccount(ctx, accountID)
@@ -1563,8 +1505,6 @@ func competitorSharePlatform(raw string) (string, error) {
switch strings.ToLower(parsed.Hostname()) {
case "www.douyin.com", "v.douyin.com":
return creator.PlatformDouyin, nil
case "www.xiaohongshu.com", "xhslink.com", "www.xhslink.com":
return creator.PlatformXiaohongshu, nil
default:
return "", creator.ErrInvalid
}
@@ -1579,14 +1519,10 @@ func previewCompetitorShare(ctx context.Context, store *creator.Store, phaseASto
return competitorSharePreview{}, creator.ErrInvalid
}
shareURL := strings.TrimSpace(rawShareURL)
switch platform {
case creator.PlatformDouyin:
return previewDouyinCompetitorShare(ctx, store, phaseAStore, hubStore, accountID, shareURL)
case creator.PlatformXiaohongshu:
return previewXiaohongshuCompetitorShare(ctx, store, phaseAStore, hubStore, accountID, shareURL)
default:
if platform != creator.PlatformDouyin {
return competitorSharePreview{}, creator.ErrUnavailable
}
return previewDouyinCompetitorShare(ctx, store, phaseAStore, hubStore, accountID, shareURL)
}
type anonymousBrowserLease struct {
@@ -1746,37 +1682,6 @@ func previewDouyinCompetitorShare(ctx context.Context, store *creator.Store, pha
}, nil
}
func previewXiaohongshuCompetitorShare(ctx context.Context, store *creator.Store, phaseAStore *accountdomain.Store, hubStore *hub.Store, accountID, shareURL string) (preview competitorSharePreview, err error) {
_ = store
_ = phaseAStore
_ = accountID
lease, err := newAnonymousBrowser(ctx, hubStore)
if err != nil {
return competitorSharePreview{}, fmt.Errorf("%w: temporary anonymous browser unavailable: %v", creator.ErrUnavailable, err)
}
defer func() {
err = errors.Join(err, lease.close())
}()
collector := &xiaohongshu.Collector{
Browser: xiaohongshuGatewayBrowser{gateway: lease.gateway, environment: lease.environment},
SourceType: creator.SourceCompetitor,
}
work, err := collector.GetNoteDetail(ctx, shareURL)
if err != nil {
return competitorSharePreview{}, err
}
if work.AuthorKey == "" {
return competitorSharePreview{}, fmt.Errorf("%w: xiaohongshu note author identity is missing", creator.ErrInvalid)
}
return competitorSharePreview{
Platform: creator.PlatformXiaohongshu,
PlatformAccountKey: work.AuthorKey,
Nickname: work.AuthorName,
HomepageURL: "https://www.xiaohongshu.com/user/profile/" + work.AuthorKey,
ShareURL: shareURL,
}, nil
}
func douyinWorkKeyFromURL(raw string) (string, error) {
parsed, err := url.Parse(raw)
if err != nil || parsed.Scheme != "https" || parsed.Hostname() != "www.douyin.com" || parsed.User != nil || parsed.Port() != "" || parsed.Fragment != "" {
@@ -1877,14 +1782,9 @@ func syncCreatorCompetitorWithClaim(ctx context.Context, store *creator.Store, p
markErr := store.MarkCompetitorSync(ctx, competitorID, leaseToken, "blocked", "", blockErr.Error(), nil)
return creator.CollectionReport{}, errors.Join(blockErr, markErr)
}
if competitor.Platform != creator.PlatformDouyin && competitor.Platform != creator.PlatformXiaohongshu {
if competitor.Platform != creator.PlatformDouyin {
return blocked(fmt.Errorf("%w: unsupported creator platform %s", creator.ErrUnavailable, competitor.Platform))
}
if competitor.Platform == creator.PlatformXiaohongshu {
if err := validateXiaohongshuSource(competitor.HomepageURL, competitor.PlatformAccountKey); err != nil {
return blocked(err)
}
}
account, err := phaseAStore.GetAccount(ctx, accountID)
if err != nil {
return blocked(err)
@@ -1969,7 +1869,7 @@ func processCompetitorShareJob(ctx context.Context, store *creator.Store, phaseA
if processErr == nil {
input := preview.input()
input.Tags = job.Tags
if processErr = validateXiaohongshuCompetitor(input); processErr == nil {
if processErr = validateDouyinCompetitor(input); processErr == nil {
var competitor creator.Competitor
competitor, processErr = store.UpsertCompetitor(ctx, input)
competitorID = competitor.ID
@@ -2100,7 +2000,7 @@ func refreshCreatorMetricWork(ctx context.Context, store *creator.Store, phaseAS
if err != nil {
return err
}
if (account.Platform != creator.PlatformDouyin && account.Platform != creator.PlatformXiaohongshu) || account.AuthorizationStatus != "authorized" || (profile.BusinessStatus != "normal" && profile.BusinessStatus != "muted") || profile.LoginStatus != "logged_in" {
if account.Platform != creator.PlatformDouyin || account.AuthorizationStatus != "authorized" || (profile.BusinessStatus != "normal" && profile.BusinessStatus != "muted") || profile.LoginStatus != "logged_in" {
return creator.ErrConflict
}
environment, err := hubStore.GetEnvironmentContextForAccount(ctx, accountID)
@@ -2169,7 +2069,7 @@ func syncCreatorOwned(ctx context.Context, store *creator.Store, phaseAStore *ac
if err != nil {
return err
}
if account.Platform != creator.PlatformDouyin && account.Platform != creator.PlatformXiaohongshu || account.AuthorizationStatus != "authorized" {
if account.Platform != creator.PlatformDouyin || account.AuthorizationStatus != "authorized" {
return creator.ErrConflict
}
profile, err := store.GetAccountProfile(ctx, accountID)
internal/controlplane/api/creator_collector.go
+69
View File
@@ -0,0 +1,69 @@
package api
// 平台收敛为抖音：collector 装配与平台身份核验 helper。
// 原小红书平台分支已随平台下线整体移除。
import (
"context"
"encoding/base64"
"errors"
"fmt"
"strings"
"git.ipao.vip/rogee/creator-hub/internal/creator"
hub "git.ipao.vip/rogee/creator-hub/internal/environment"
douyin "git.ipao.vip/rogee/creator-hub/internal/platform/douyin"
)
func newCreatorCollector(ctx context.Context, platform string, gateway hub.Gateway, environment hub.EnvironmentContext, viewerAccountKey, targetAccountKey, _, sourceType, sourceID string) (creator.PlatformCollector, string, error) {
if strings.TrimSpace(viewerAccountKey) == "" || strings.TrimSpace(targetAccountKey) == "" {
return nil, "", fmt.Errorf("%w: creator collector account key is missing", creator.ErrInvalid)
}
if platform != creator.PlatformDouyin {
return nil, "", fmt.Errorf("%w: unsupported creator platform %s", creator.ErrUnavailable, platform)
}
browser := creatorGatewayBrowser{gateway: gateway, environment: environment}
if _, err := browser.Identity(ctx, viewerAccountKey); err != nil {
return nil, "", err
}
collector := douyinCollector(browser, targetAccountKey, sourceType, sourceID)
canonicalKey, err := collector.CanonicalTargetSecUID(ctx, targetAccountKey)
if err != nil {
return nil, "", err
}
return &collector, canonicalKey, nil
}
func douyinCollector(browser creatorGatewayBrowser, accountKey, sourceType, sourceID string) douyin.CreatorCollector {
return douyin.CreatorCollector{Browser: browser, AccountKey: accountKey, SourceType: sourceType, SourceID: sourceID}
}
// validateDouyinCompetitor 监控账号入队前的平台一致性校验（平台收敛后仅抖音）。
func validateDouyinCompetitor(input creator.CompetitorInput) error {
if input.Platform != creator.PlatformDouyin {
return creator.ErrInvalid
}
return nil
}
func verifyCreatorPlatformIdentity(ctx context.Context, platform string, gateway hub.Gateway, environment hub.EnvironmentContext, expectedKey string) (string, error) {
if platform != creator.PlatformDouyin {
return "", fmt.Errorf("%w: unsupported creator platform %s", creator.ErrUnavailable, platform)
}
return (creatorGatewayBrowser{gateway: gateway, environment: environment}).Identity(ctx, expectedKey)
}
func decodeBase64(value string) ([]byte, error) {
const maxEncoded = 96 << 20
if len(value) > maxEncoded {
return nil, errors.New("media response is too large")
}
data, err := base64.StdEncoding.DecodeString(value)
if err != nil {
return nil, fmt.Errorf("decode media response: %w", err)
}
if len(data) > maxCreatorMediaBytes {
return nil, errors.New("media response is too large")
}
return data, nil
}
internal/controlplane/api/creator_helper_test.go
+7 -30
View File
@@ -64,12 +64,6 @@ func TestCreatorHelperBranches(t *testing.T) {
if _, _, err := newCreatorCollector(context.Background(), creator.PlatformDouyin, hub.Gateway{}, hub.EnvironmentContext{}, "", "target", "", creator.SourceCompetitor, "id"); !errors.Is(err, creator.ErrInvalid) {
t.Fatalf("missing collector key = %v", err)
}
if _, err := decodeXiaohongshuResponse([]byte("not-json")); err == nil {
t.Fatal("malformed Xiaohongshu response must fail")
}
if response, err := decodeXiaohongshuResponse([]byte(`{"status":200,"body":"ok","challenge":""}`)); err != nil || response.Status != 200 || string(response.Body) != "ok" {
t.Fatalf("decode Xiaohongshu response = %+v, %v", response, err)
}
if _, err := decodeBase64("not-base64"); err == nil {
t.Fatal("invalid base64 must fail")
}
@@ -77,12 +71,12 @@ func TestCreatorHelperBranches(t *testing.T) {
if data, err := decodeBase64(encoded); err != nil || string(data) != "media" {
t.Fatalf("decode media = %q, %v", data, err)
}
if err := validateXiaohongshuSource("", ""); err == nil {
t.Fatal("empty Xiaohongshu source must fail")
}
if err := validateXiaohongshuCompetitor(creator.CompetitorInput{Platform: creator.PlatformDouyin}); err != nil {
if err := validateDouyinCompetitor(creator.CompetitorInput{Platform: creator.PlatformDouyin}); err != nil {
t.Fatal(err)
}
if err := validateDouyinCompetitor(creator.CompetitorInput{Platform: "unsupported"}); !errors.Is(err, creator.ErrInvalid) {
t.Fatalf("unsupported competitor platform = %v", err)
}
}
func TestCreatorPageQueryValidation(t *testing.T) {
@@ -110,7 +104,7 @@ func TestCreatorPageQueryValidation(t *testing.T) {
func TestCreatorControlPlaneGuards(t *testing.T) {
ctx := context.Background()
if _, err := persistDouyinMessageHistory(ctx, nil, creator.Conversation{Platform: creator.PlatformXiaohongshu}, "uid", nil); !errors.Is(err, creator.ErrInvalid) {
if _, err := persistDouyinMessageHistory(ctx, nil, creator.Conversation{Platform: "unsupported"}, "uid", nil); !errors.Is(err, creator.ErrInvalid) {
t.Fatalf("invalid history store/platform = %v", err)
}
conversation := creator.Conversation{Platform: creator.PlatformDouyin, AccountID: "account", PeerUID: "peer"}
@@ -130,8 +124,8 @@ func TestCreatorControlPlaneGuards(t *testing.T) {
if err := (creatorMaterialDownloader{}).Download(ctx, creator.Work{Platform: "unsupported"}, "/tmp/media"); !errors.Is(err, creator.ErrUnavailable) {
t.Fatalf("unsupported material platform = %v", err)
}
if _, err := newXiaohongshuReadCollector(ctx, nil, nil, nil, "", creator.SourceOwned, "id"); !errors.Is(err, creator.ErrUnavailable) {
t.Fatalf("empty Xiaohongshu collector = %v", err)
if _, _, err := newCreatorCollector(ctx, "unsupported", hub.Gateway{}, hub.EnvironmentContext{}, "viewer", "target", "", creator.SourceOwned, "id"); !errors.Is(err, creator.ErrUnavailable) {
t.Fatalf("unsupported collector platform = %v", err)
}
if listenerBoundaryPointer(time.Time{}) != nil {
t.Fatal("zero listener boundary should be nil")
@@ -198,23 +192,6 @@ func TestCreatorGatewayBrowserHistoryAndMedia(t *testing.T) {
}
}
func TestXiaohongshuGatewayMedia(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
if !strings.HasSuffix(r.URL.Path, "/xiaohongshu/media") {
http.NotFound(w, r)
return
}
_, _ = w.Write([]byte(`{"status":200,"content_type":"image/jpeg","body_base64":"` + base64.StdEncoding.EncodeToString([]byte("image")) + `"}`))
}))
defer server.Close()
browser := xiaohongshuGatewayBrowser{gateway: hub.Gateway{Endpoint: server.URL, Token: "token"}, environment: hub.EnvironmentContext{Env: hub.Env{Alias: "browser"}, AccountID: "account", AccountStatus: "active", AuthorizationStatus: "authorized", BindingID: "binding", RuntimeInstanceID: "instance", RuntimeID: "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd", RuntimeNetworkID: "native-dddddddddddddddddddddddddddddddd", BindingVersion: 1, Exit: hub.NetworkExit{ID: "exit", HealthStatus: "healthy"}}}
data, contentType, err := browser.Media(context.Background(), "https://www.xiaohongshu.com/explore/abc")
if err != nil || string(data) != "image" || contentType != "image/jpeg" {
t.Fatalf("media = %q, %q, %v", data, contentType, err)
}
}
func TestCreatorGatewayBrowserIdentity(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost || !strings.Contains(r.URL.Path, "/douyin/identity") {
internal/controlplane/api/creator_material.go
+4 -12
View File
@@ -24,7 +24,7 @@ type creatorMaterialDownloader struct {
}
func (downloader creatorMaterialDownloader) Download(ctx context.Context, work creator.Work, destination string) (resultErr error) {
if (work.Platform != creator.PlatformDouyin && work.Platform != creator.PlatformXiaohongshu) || downloader.store == nil || downloader.phaseAStore == nil || downloader.hubStore == nil {
if work.Platform != creator.PlatformDouyin || downloader.store == nil || downloader.phaseAStore == nil || downloader.hubStore == nil {
return fmt.Errorf("%w: creator media gateway is unavailable", creator.ErrUnavailable)
}
accountID := work.SourceID
@@ -65,18 +65,10 @@ func (downloader creatorMaterialDownloader) Download(ctx context.Context, work c
if _, err := verifyCreatorPlatformIdentity(useCtx, work.Platform, gateway, environment, profile.PlatformAccountKey); err != nil {
return fmt.Errorf("%w: media browser identity verification failed: %v", creator.ErrConflict, err)
}
if work.Platform == creator.PlatformDouyin {
return (creatorGatewayBrowser{gateway: gateway, environment: environment}).Media(useCtx, work.OriginalURL, destination)
if work.Platform != creator.PlatformDouyin {
return fmt.Errorf("%w: unsupported creator platform %s", creator.ErrUnavailable, work.Platform)
}
data, contentType, err := (xiaohongshuGatewayBrowser{gateway: gateway, environment: environment}).Media(useCtx, work.OriginalURL)
if err != nil {
return err
}
contentType = strings.ToLower(strings.TrimSpace(strings.SplitN(contentType, ";", 2)[0]))
if !strings.HasPrefix(contentType, "video/") && contentType != "application/octet-stream" {
return fmt.Errorf("xiaohongshu media response is not a video")
}
return writeCreatorMedia(destination, data)
return (creatorGatewayBrowser{gateway: gateway, environment: environment}).Media(useCtx, work.OriginalURL, destination)
}
func processCreatorMaterial(ctx context.Context, store *creator.Store, phaseAStore *accountdomain.Store, hubStore *hub.Store, workID string) (creator.MaterialJob, error) {
internal/controlplane/api/creator_pure_unit_test.go
+3 -3
View File
@@ -144,14 +144,14 @@ func TestCreatorPreviewAndLifecycleHelpersRejectUnavailableDependencies(t *testi
if _, err := previewCompetitorShare(context.Background(), nil, nil, nil, "account-1", "", "not-a-url"); err == nil {
t.Fatal("invalid competitor share URL was accepted")
}
if _, err := previewCompetitorShare(context.Background(), nil, nil, nil, "account-1", creator.PlatformXiaohongshu, "https://www.douyin.com/video/123"); err == nil {
t.Fatal("platform mismatch was accepted")
if _, err := previewCompetitorShare(context.Background(), nil, nil, nil, "account-1", creator.PlatformDouyin, "https://www.douyin.com/video/123"); err == nil {
t.Fatal("unavailable douyin preview dependency was accepted")
}
if _, err := previewCompetitorShare(context.Background(), nil, nil, nil, "account-1", "", "https://www.douyin.com/video/123"); err == nil {
t.Fatal("unavailable Douyin preview was reported as successful")
}
if _, err := previewCompetitorShare(context.Background(), nil, nil, nil, "account-1", "", "https://www.xiaohongshu.com/explore/123"); err == nil {
t.Fatal("unavailable Xiaohongshu preview was reported as successful")
t.Fatal("xiaohongshu share URL was accepted after platform retirement")
}
if got := (competitorSharePreview{Platform: creator.PlatformDouyin, PlatformAccountKey: "uid", Nickname: "name", AvatarURL: "avatar", HomepageURL: "home"}).input(); got.PlatformAccountKey != "uid" || got.Nickname != "name" {
t.Fatalf("preview input = %#v", got)
internal/controlplane/api/creator_route_validation_test.go
-2
View File
@@ -55,8 +55,6 @@ func TestCreatorWriteRoutesRejectMalformedInputBeforeStoreAccess(t *testing.T) {
{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"},
{method: http.MethodPost, path: "/api/creator/xiaohongshu/search"},
{method: http.MethodPost, path: "/api/creator/xiaohongshu/detail"},
{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"},
internal/controlplane/api/creator_share_test.go
-1
View File
@@ -19,7 +19,6 @@ func TestCompetitorSharePlatform(t *testing.T) {
}{
{name: "douyin short link", url: "https://v.douyin.com/abc123/", want: creator.PlatformDouyin},
{name: "douyin content", url: "https://www.douyin.com/video/123", want: creator.PlatformDouyin},
{name: "xiaohongshu note", url: "https://www.xiaohongshu.com/explore/note-1", want: creator.PlatformXiaohongshu},
} {
t.Run(test.name, func(t *testing.T) {
got, err := competitorSharePlatform(test.url)
internal/controlplane/api/hub_native_unit_test.go
+2 -2
View File
@@ -49,8 +49,8 @@ func TestCreatorGatewayBrowserResolveAndSharePlatformValidation(t *testing.T) {
}{
{value: "https://www.douyin.com/video/1", platform: creator.PlatformDouyin},
{value: "https://v.douyin.com/abc", platform: creator.PlatformDouyin},
{value: "https://www.xiaohongshu.com/explore/1", platform: creator.PlatformXiaohongshu},
{value: "https://xhslink.com/abc", platform: creator.PlatformXiaohongshu},
{value: "https://www.xiaohongshu.com/explore/1", wantErr: true},
{value: "https://xhslink.com/abc", wantErr: true},
{value: "http://www.douyin.com/video/1", wantErr: true},
{value: "https://user@www.douyin.com/video/1", wantErr: true},
{value: "https://www.douyin.com:443/video/1", wantErr: true},
internal/controlplane/api/hub_test.go
-1
View File
@@ -2724,7 +2724,6 @@ func TestPostgresRejectsInvalidLifecycleTargetsWithSanitizedAuditPairs(t *testin
body string
reason string
}{
{action: "upgrade", body: `{"version":"` + secret + `"}`, reason: "upgrade_input_rejected"},
{action: "rebind", body: `{"network_exit_id":"` + secret + `"}`, reason: "rebind_input_rejected"},
} {
t.Run(test.action, func(t *testing.T) {
internal/controlplane/api/xiaohongshu.go
-265
View File
@@ -1,265 +0,0 @@
package api
import (
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"net/http"
"net/url"
"strings"
"time"
accountdomain "git.ipao.vip/rogee/creator-hub/internal/account"
"git.ipao.vip/rogee/creator-hub/internal/creator"
hub "git.ipao.vip/rogee/creator-hub/internal/environment"
douyin "git.ipao.vip/rogee/creator-hub/internal/platform/douyin"
xiaohongshu "git.ipao.vip/rogee/creator-hub/internal/platform/xiaohongshu"
)
type xiaohongshuGatewayBrowser struct {
gateway hub.Gateway
environment hub.EnvironmentContext
}
func (browser xiaohongshuGatewayBrowser) generation() (map[string]any, error) {
request, err := (douyinGatewayBrowser{gateway: browser.gateway, environment: browser.environment}).request()
if err != nil {
return nil, err
}
return map[string]any{
"binding_version": request.BindingVersion,
"runtime_id": request.RuntimeID,
"network_id": request.NetworkID,
"network_exit_id": request.NetworkExitID,
}, nil
}
func (browser xiaohongshuGatewayBrowser) Get(ctx context.Context, target string) (xiaohongshu.Response, error) {
request, err := browser.generation()
if err != nil {
return xiaohongshu.Response{}, err
}
request["url"] = target
status, body, err := gatewayCall(ctx, browser.gateway, http.MethodPost,
"/v1/browsers/"+url.PathEscape(browser.environment.Alias)+"/xiaohongshu/get", request, 30*time.Second)
if err != nil || status != http.StatusOK {
return xiaohongshu.Response{}, errors.New("restricted Xiaohongshu browser operation failed")
}
return decodeXiaohongshuResponse(body)
}
func (browser xiaohongshuGatewayBrowser) Post(ctx context.Context, target string, payload []byte) (xiaohongshu.Response, error) {
if len(payload) == 0 || len(payload) > 4<<20 {
return xiaohongshu.Response{}, errors.New("invalid Xiaohongshu browser body")
}
var bodyValue any
if err := json.Unmarshal(payload, &bodyValue); err != nil {
return xiaohongshu.Response{}, fmt.Errorf("invalid Xiaohongshu browser body: %w", err)
}
if _, ok := bodyValue.(map[string]any); !ok {
return xiaohongshu.Response{}, errors.New("Xiaohongshu browser body must be an object")
}
request, err := browser.generation()
if err != nil {
return xiaohongshu.Response{}, err
}
request["url"] = target
request["body"] = bodyValue
status, responseBody, err := gatewayCall(ctx, browser.gateway, http.MethodPost,
"/v1/browsers/"+url.PathEscape(browser.environment.Alias)+"/xiaohongshu/post", request, 30*time.Second)
if err != nil || status != http.StatusOK {
return xiaohongshu.Response{}, errors.New("restricted Xiaohongshu browser POST failed")
}
return decodeXiaohongshuResponse(responseBody)
}
func (browser xiaohongshuGatewayBrowser) Identity(ctx context.Context, expectedKey string) (string, error) {
request, err := browser.generation()
if err != nil {
return "", err
}
request["expected_account_key"] = expectedKey
status, body, err := gatewayCall(ctx, browser.gateway, http.MethodPost,
"/v1/browsers/"+url.PathEscape(browser.environment.Alias)+"/xiaohongshu/identity", request, 30*time.Second)
if err != nil || status != http.StatusOK {
return "", errors.New("Xiaohongshu identity verification failed")
}
var identity struct {
UID string `json:"uid"`
}
if err := json.Unmarshal(body, &identity); err != nil || strings.TrimSpace(identity.UID) == "" {
return "", errors.New("Xiaohongshu identity response omitted uid")
}
return identity.UID, nil
}
func (browser xiaohongshuGatewayBrowser) Resolve(ctx context.Context, target string) (string, error) {
request, err := browser.generation()
if err != nil {
return "", err
}
request["url"] = target
status, body, err := gatewayCall(ctx, browser.gateway, http.MethodPost,
"/v1/browsers/"+url.PathEscape(browser.environment.Alias)+"/xiaohongshu/resolve", request, 30*time.Second)
if err != nil || status != http.StatusOK {
return "", errors.New("restricted Xiaohongshu share resolution failed")
}
var response struct {
URL string `json:"url"`
}
if err := json.Unmarshal(body, &response); err != nil || strings.TrimSpace(response.URL) == "" {
return "", errors.New("Xiaohongshu share resolution response omitted url")
}
return response.URL, nil
}
func (browser xiaohongshuGatewayBrowser) Media(ctx context.Context, target string) ([]byte, string, error) {
request, err := browser.generation()
if err != nil {
return nil, "", err
}
request["url"] = target
status, body, err := gatewayCall(ctx, browser.gateway, http.MethodPost,
"/v1/browsers/"+url.PathEscape(browser.environment.Alias)+"/xiaohongshu/media", request, 90*time.Second)
if err != nil || status != http.StatusOK {
return nil, "", errors.New("restricted Xiaohongshu media request failed")
}
var response struct {
Status int `json:"status"`
ContentType string `json:"content_type"`
BodyBase64 string `json:"body_base64"`
}
if err := json.Unmarshal(body, &response); err != nil || response.Status < 200 || response.Status >= 300 || response.BodyBase64 == "" {
return nil, "", errors.New("Xiaohongshu media response is invalid")
}
data, err := decodeBase64(response.BodyBase64)
if err != nil {
return nil, "", err
}
return data, response.ContentType, nil
}
func decodeXiaohongshuResponse(body []byte) (xiaohongshu.Response, error) {
var response struct {
Status int `json:"status"`
Body string `json:"body"`
Challenge string `json:"challenge"`
}
if err := json.Unmarshal(body, &response); err != nil || response.Status < 100 || response.Status > 599 {
return xiaohongshu.Response{}, errors.New("restricted Xiaohongshu browser returned an invalid response")
}
return xiaohongshu.Response{Status: response.Status, Body: []byte(response.Body), Challenge: response.Challenge}, nil
}
func newXiaohongshuReadCollector(ctx context.Context, store *creator.Store, phaseAStore *accountdomain.Store, hubStore *hub.Store, accountID, sourceType, sourceID string) (*xiaohongshu.Collector, error) {
if store == nil || phaseAStore == nil || hubStore == nil || strings.TrimSpace(accountID) == "" {
return nil, creator.ErrUnavailable
}
account, err := phaseAStore.GetAccount(ctx, accountID)
if err != nil {
return nil, err
}
if account.Platform != creator.PlatformXiaohongshu || account.AuthorizationStatus != "authorized" {
return nil, creator.ErrConflict
}
profile, err := store.GetAccountProfile(ctx, accountID)
if err != nil {
return nil, err
}
if profile.Platform != creator.PlatformXiaohongshu || (profile.BusinessStatus != "normal" && profile.BusinessStatus != "muted") || profile.LoginStatus != "logged_in" {
return nil, creator.ErrConflict
}
environment, err := hubStore.GetEnvironmentContextForAccount(ctx, accountID)
if err != nil {
return nil, fmt.Errorf("%w: account environment unavailable: %v", creator.ErrUnavailable, err)
}
if environment.RuntimeID == "" || environment.RuntimeNetworkID == "" || environment.BindingVersion <= 0 {
return nil, fmt.Errorf("%w: account runtime is not running", creator.ErrUnavailable)
}
gateway, err := hubStore.GetGateway(ctx, environment.Gateway)
if err != nil {
return nil, fmt.Errorf("%w: gateway unavailable: %v", creator.ErrUnavailable, err)
}
browser := xiaohongshuGatewayBrowser{gateway: gateway, environment: environment}
if _, err := browser.Identity(ctx, account.PlatformAccountKey); err != nil {
return nil, fmt.Errorf("%w: account identity verification failed: %v", creator.ErrConflict, err)
}
return &xiaohongshu.Collector{Browser: browser, AccountKey: account.PlatformAccountKey, SourceType: sourceType, SourceID: sourceID}, nil
}
func validateXiaohongshuSource(homepageURL, accountKey string) error {
return xiaohongshu.ValidateSourceURL(homepageURL, accountKey)
}
func validateXiaohongshuCompetitor(input creator.CompetitorInput) error {
if input.Platform != creator.PlatformXiaohongshu {
return nil
}
return validateXiaohongshuSource(input.HomepageURL, input.PlatformAccountKey)
}
func newCreatorCollector(ctx context.Context, platform string, gateway hub.Gateway, environment hub.EnvironmentContext, viewerAccountKey, targetAccountKey, homepageURL, sourceType, sourceID string) (creator.PlatformCollector, string, error) {
if strings.TrimSpace(viewerAccountKey) == "" || strings.TrimSpace(targetAccountKey) == "" {
return nil, "", fmt.Errorf("%w: creator collector account key is missing", creator.ErrInvalid)
}
switch platform {
case creator.PlatformDouyin:
browser := creatorGatewayBrowser{gateway: gateway, environment: environment}
if _, err := browser.Identity(ctx, viewerAccountKey); err != nil {
return nil, "", err
}
collector := douyinCollector(browser, targetAccountKey, sourceType, sourceID)
canonicalKey, err := collector.CanonicalTargetSecUID(ctx, targetAccountKey)
if err != nil {
return nil, "", err
}
return &collector, canonicalKey, nil
case creator.PlatformXiaohongshu:
if homepageURL != "" {
if err := xiaohongshu.ValidateSourceURL(homepageURL, targetAccountKey); err != nil {
return nil, "", err
}
}
browser := xiaohongshuGatewayBrowser{gateway: gateway, environment: environment}
if _, err := browser.Identity(ctx, viewerAccountKey); err != nil {
return nil, "", err
}
return &xiaohongshu.Collector{Browser: browser, AccountKey: targetAccountKey, HomepageURL: homepageURL, SourceType: sourceType, SourceID: sourceID}, targetAccountKey, nil
default:
return nil, "", fmt.Errorf("%w: unsupported creator platform %s", creator.ErrUnavailable, platform)
}
}
func douyinCollector(browser creatorGatewayBrowser, accountKey, sourceType, sourceID string) douyin.CreatorCollector {
return douyin.CreatorCollector{Browser: browser, AccountKey: accountKey, SourceType: sourceType, SourceID: sourceID}
}
func verifyCreatorPlatformIdentity(ctx context.Context, platform string, gateway hub.Gateway, environment hub.EnvironmentContext, expectedKey string) (string, error) {
switch platform {
case creator.PlatformDouyin:
return (creatorGatewayBrowser{gateway: gateway, environment: environment}).Identity(ctx, expectedKey)
case creator.PlatformXiaohongshu:
return (xiaohongshuGatewayBrowser{gateway: gateway, environment: environment}).Identity(ctx, expectedKey)
default:
return "", fmt.Errorf("%w: unsupported creator platform %s", creator.ErrUnavailable, platform)
}
}
func decodeBase64(value string) ([]byte, error) {
const maxEncoded = 96 << 20
if len(value) > maxEncoded {
return nil, errors.New("media response is too large")
}
data, err := base64.StdEncoding.DecodeString(value)
if err != nil {
return nil, fmt.Errorf("decode media response: %w", err)
}
if len(data) > maxCreatorMediaBytes {
return nil, errors.New("media response is too large")
}
return data, nil
}
var _ xiaohongshu.Browser = xiaohongshuGatewayBrowser{}
internal/controlplane/api/xiaohongshu_test.go
-116
View File
@@ -1,116 +0,0 @@
package api
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"git.ipao.vip/rogee/creator-hub/internal/creator"
hub "git.ipao.vip/rogee/creator-hub/internal/environment"
xiaohongshu "git.ipao.vip/rogee/creator-hub/internal/platform/xiaohongshu"
)
const testXiaohongshuIdentityURL = "https://edith.xiaohongshu.com/api/sns/web/v2/user/me"
func TestXiaohongshuGatewayBrowserFencesAccountGeneration(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
if request.Header.Get("Authorization") != "Bearer gateway-token-1" {
t.Fatal("missing gateway authorization")
}
var body map[string]any
if json.NewDecoder(request.Body).Decode(&body) != nil || body["binding_version"] != float64(2) || body["runtime_id"] != "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" || body["network_id"] != "native-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" || body["network_exit_id"] != "exit-a" {
t.Fatalf("generation fence missing: %#v", body)
}
if request.URL.Path != "/v1/browsers/account-a/xiaohongshu/get" || body["url"] != testXiaohongshuIdentityURL {
t.Fatalf("unexpected request: path=%s body=%#v", request.URL.Path, body)
}
_ = json.NewEncoder(response).Encode(map[string]any{"status": 200, "body": `{"success":true}`, "challenge": ""})
}))
defer server.Close()
browser := xiaohongshuGatewayBrowser{gateway: hub.Gateway{Endpoint: server.URL, Token: "gateway-token-1"}, environment: readyDouyinEnvironment()}
result, err := browser.Get(context.Background(), testXiaohongshuIdentityURL)
if err != nil || result.Status != 200 || string(result.Body) != `{"success":true}` {
t.Fatalf("unexpected result: %#v err=%v", result, err)
}
}
func TestXiaohongshuGatewayBrowserResolvesShareLinks(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
if request.URL.Path != "/v1/browsers/account-a/xiaohongshu/resolve" {
t.Fatalf("unexpected path: %s", request.URL.Path)
}
_ = json.NewEncoder(response).Encode(map[string]any{"url": "https://www.xiaohongshu.com/explore/n-1"})
}))
defer server.Close()
browser := xiaohongshuGatewayBrowser{gateway: hub.Gateway{Endpoint: server.URL, Token: "gateway-token-1"}, environment: readyDouyinEnvironment()}
resolved, err := browser.Resolve(context.Background(), "https://xhslink.com/a/abc")
if err != nil || resolved != "https://www.xiaohongshu.com/explore/n-1" {
t.Fatalf("resolved URL=%q err=%v", resolved, err)
}
}
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" {
t.Fatalf("unexpected path: %s", request.URL.Path)
}
_ = json.NewEncoder(response).Encode(map[string]any{"uid": "viewer-1"})
}))
defer server.Close()
collector, target, err := newCreatorCollector(context.Background(), creator.PlatformXiaohongshu, hub.Gateway{Endpoint: server.URL, Token: "gateway-token-1"}, readyDouyinEnvironment(), "viewer-1", "target-1", "https://www.xiaohongshu.com/user/profile/target-1?xsec_source=pc_search", creator.SourceCompetitor, "source-1")
if err != nil {
t.Fatalf("new collector: %v", err)
}
xhsCollector, ok := collector.(*xiaohongshu.Collector)
if !ok || xhsCollector.AccountKey != "target-1" || target != "target-1" || xhsCollector.HomepageURL == "" {
t.Fatalf("collector=%#v target=%q", collector, target)
}
}
func TestXiaohongshuGatewayBrowserPostCarriesJSONBody(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 || body["url"] != "https://so.xiaohongshu.com/api/sns/web/v2/search/notes" {
t.Fatalf("unexpected request body: %#v", body)
}
if _, ok := body["body"].(map[string]any); !ok {
t.Fatalf("missing nested request body: %#v", body)
}
_ = json.NewEncoder(response).Encode(map[string]any{"status": 200, "body": `{"success":true}`, "challenge": ""})
}))
defer server.Close()
browser := xiaohongshuGatewayBrowser{gateway: hub.Gateway{Endpoint: server.URL, Token: "gateway-token-1"}, environment: readyDouyinEnvironment()}
if _, err := browser.Post(context.Background(), "https://so.xiaohongshu.com/api/sns/web/v2/search/notes", []byte(`{"keyword":"x"}`)); err != nil {
t.Fatalf("post failed: %v", err)
}
}
internal/creator/collection.go
+1 -1
View File
@@ -344,7 +344,7 @@ func (s *Store) ListDueOwnedAccounts(ctx context.Context, now time.Time, interva
ON works_checkpoint.source_type='owned' AND works_checkpoint.source_id=account.id AND works_checkpoint.collection_kind='works'
LEFT JOIN creator_collection_checkpoint comments_checkpoint
ON comments_checkpoint.source_type='owned' AND comments_checkpoint.source_id=account.id AND comments_checkpoint.collection_kind='comments'
WHERE account.platform IN ('douyin', 'xiaohongshu') AND account.authorization_status='authorized'
WHERE account.platform = 'douyin' AND account.authorization_status='authorized'
AND profile.login_status='logged_in'
AND COALESCE(works_checkpoint.status, '') <> 'blocked'
AND COALESCE(comments_checkpoint.status, '') <> 'blocked'
internal/creator/competitor_share_jobs.go
+16 -12
View File
@@ -78,12 +78,12 @@ func scanCompetitorShareJob(scanner interface{ Scan(...any) error }) (Competitor
return result, nil
}
const competitorShareJobSelect = `SELECT id, platform, share_url, tags, status, attempts, competitor_id,
failure_reason, lease_until, last_attempt_at, completed_at, created_at, updated_at
FROM creator_competitor_share_job`
const competitorShareJobSelect = `SELECT share_job.id, share_job.platform, share_job.share_url, share_job.tags, share_job.status, share_job.attempts, share_job.competitor_id,
share_job.failure_reason, share_job.lease_until, share_job.last_attempt_at, share_job.completed_at, share_job.created_at, share_job.updated_at
FROM creator_competitor_share_job share_job`
func (s *Store) GetCompetitorShareJob(ctx context.Context, id string) (CompetitorShareJob, error) {
result, err := scanCompetitorShareJob(s.db.QueryRowContext(ctx, competitorShareJobSelect+` WHERE id = $1`, id))
result, err := scanCompetitorShareJob(s.db.QueryRowContext(ctx, competitorShareJobSelect+` WHERE share_job.id = $1`, id))
return result, rowError(err)
}
@@ -92,9 +92,12 @@ func (s *Store) ListCompetitorShareJobsWithAuthor(ctx context.Context, platform,
if platform != "" && !ValidatePlatform(platform) || status != "" && !validateCompetitorShareJobStatus(status) {
return nil, ErrInvalid
}
query, args := competitorShareJobFilter(competitorShareJobSelect+`
LEFT JOIN creator_competitor ON creator_competitor.id = creator_competitor_share_job.competitor_id`, platform, status)
rows, err := s.db.QueryContext(ctx, query+` ORDER BY creator_competitor_share_job.created_at DESC, creator_competitor_share_job.id`, args...)
query, args := competitorShareJobFilter(`SELECT share_job.id, share_job.platform, share_job.share_url, share_job.tags, share_job.status, share_job.attempts, share_job.competitor_id,
share_job.failure_reason, share_job.lease_until, share_job.last_attempt_at, share_job.completed_at, share_job.created_at, share_job.updated_at,
competitor.nickname, competitor.avatar_url
FROM creator_competitor_share_job share_job
LEFT JOIN creator_competitor ON competitor.id = share_job.competitor_id`, platform, status)
rows, err := s.db.QueryContext(ctx, query+` ORDER BY share_job.created_at DESC, share_job.id`, args...)
if err != nil {
return nil, databaseError(err)
}
@@ -103,11 +106,11 @@ func (s *Store) ListCompetitorShareJobsWithAuthor(ctx context.Context, platform,
for rows.Next() {
var item CompetitorShareJobView
var tags pgtype.FlatArray[string]
var competitorID, nickname sql.NullString
var competitorID, nickname, avatarURL sql.NullString
var leaseUntil, lastAttemptAt, completedAt sql.NullTime
if err := rows.Scan(&item.ID, &item.Platform, &item.ShareURL, pgtype.NewMap().SQLScanner(&tags),
&item.Status, &item.Attempts, &competitorID, &item.FailureReason, &leaseUntil, &lastAttemptAt,
&completedAt, &item.CreatedAt, &item.UpdatedAt, &nickname); err != nil {
&completedAt, &item.CreatedAt, &item.UpdatedAt, &nickname, &avatarURL); err != nil {
return nil, databaseError(err)
}
item.Tags = []string(tags)
@@ -117,6 +120,7 @@ func (s *Store) ListCompetitorShareJobsWithAuthor(ctx context.Context, platform,
item.LastAttemptAt = nullableTime(lastAttemptAt)
item.CompletedAt = nullableTime(completedAt)
item.AuthorName = nickname.String
item.AuthorAvatarURL = avatarURL.String
items = append(items, item)
}
return items, rows.Err()
@@ -157,16 +161,16 @@ func (s *Store) ListCompetitorShareJobs(ctx context.Context, platform, status st
args := make([]any, 0, 2)
if platform != "" {
args = append(args, platform)
conditions = append(conditions, "platform = $"+strconv.Itoa(len(args)))
conditions = append(conditions, "share_job.platform = $"+strconv.Itoa(len(args)))
}
if status != "" {
args = append(args, status)
conditions = append(conditions, "status = $"+strconv.Itoa(len(args)))
conditions = append(conditions, "share_job.status = $"+strconv.Itoa(len(args)))
}
if len(conditions) > 0 {
query += ` WHERE ` + strings.Join(conditions, ` AND `)
}
query += ` ORDER BY created_at DESC, id`
query += ` ORDER BY share_job.created_at DESC, share_job.id`
rows, err := s.db.QueryContext(ctx, query, args...)
if err != nil {
return nil, databaseError(err)
internal/creator/content.go
+93 -26
View File
@@ -70,51 +70,78 @@ func (s *Store) CreateCompetitor(ctx context.Context, input CompetitorInput) (Co
return s.GetCompetitor(ctx, id)
}
// competitorColumns 是 creator_competitor 的完整列清单（含画像属性），各查询共用。
const competitorColumns = `id, platform, platform_account_key, unique_id, nickname, avatar_url, homepage_url, tags,
enabled, follower_count, following_count, aweme_count, sync_status, sync_cursor, sync_error, sync_lease_until,
last_sync_at, next_sync_at, created_at, updated_at`
type competitorScan struct {
competitor Competitor
tags pgtype.FlatArray[string]
leaseUntil, lastSync, nextSync sql.NullTime
follower, following, aweme sql.NullInt64
}
func competitorScanDestinations(scan *competitorScan) []any {
return []any{&scan.competitor.ID, &scan.competitor.Platform, &scan.competitor.PlatformAccountKey, &scan.competitor.UniqueID, &scan.competitor.Nickname,
&scan.competitor.AvatarURL, &scan.competitor.HomepageURL, pgtype.NewMap().SQLScanner(&scan.tags), &scan.competitor.Enabled,
&scan.follower, &scan.following, &scan.aweme, &scan.competitor.SyncStatus, &scan.competitor.SyncCursor,
&scan.competitor.SyncError, &scan.leaseUntil, &scan.lastSync, &scan.nextSync, &scan.competitor.CreatedAt, &scan.competitor.UpdatedAt}
}
func (scan *competitorScan) materialize() Competitor {
scan.competitor.Tags = []string(scan.tags)
scan.competitor.FollowerCount = nullableInt64(scan.follower)
scan.competitor.FollowingCount = nullableInt64(scan.following)
scan.competitor.AwemeCount = nullableInt64(scan.aweme)
scan.competitor.SyncLeaseUntil = nullableTime(scan.leaseUntil)
scan.competitor.LastSyncAt = nullableTime(scan.lastSync)
scan.competitor.NextSyncAt = nullableTime(scan.nextSync)
return scan.competitor
}
func scanCompetitor(scanner interface{ Scan(...any) error }) (Competitor, error) {
var result Competitor
var tags pgtype.FlatArray[string]
var leaseUntil, lastSync, nextSync sql.NullTime
if err := scanner.Scan(&result.ID, &result.Platform, &result.PlatformAccountKey, &result.UniqueID, &result.Nickname,
&result.AvatarURL, &result.HomepageURL, pgtype.NewMap().SQLScanner(&tags), &result.Enabled, &result.SyncStatus, &result.SyncCursor,
&result.SyncError, &leaseUntil, &lastSync, &nextSync, &result.CreatedAt, &result.UpdatedAt); err != nil {
var scan competitorScan
if err := scanner.Scan(competitorScanDestinations(&scan)...); err != nil {
return Competitor{}, err
}
result.Tags = []string(tags)
result.SyncLeaseUntil = nullableTime(leaseUntil)
result.LastSyncAt = nullableTime(lastSync)
result.NextSyncAt = nullableTime(nextSync)
return result, nil
return scan.materialize(), nil
}
func (s *Store) GetCompetitor(ctx context.Context, id string) (Competitor, error) {
result, err := scanCompetitor(s.db.QueryRowContext(ctx, `
SELECT id, platform, platform_account_key, unique_id, nickname, avatar_url, homepage_url, tags,
enabled, sync_status, sync_cursor, sync_error, sync_lease_until, last_sync_at, next_sync_at, created_at, updated_at
FROM creator_competitor WHERE id = $1`, id))
result, err := scanCompetitor(s.db.QueryRowContext(ctx, `SELECT `+competitorColumns+` FROM creator_competitor WHERE id = $1`, id))
return result, rowError(err)
}
func (s *Store) ListCompetitors(ctx context.Context, platform string) ([]Competitor, error) {
query := `SELECT id, platform, platform_account_key, unique_id, nickname, avatar_url, homepage_url, tags,
enabled, sync_status, sync_cursor, sync_error, sync_lease_until, last_sync_at, next_sync_at, created_at, updated_at
FROM creator_competitor`
// ListCompetitorsWithProfile 监控账号列表视图：账号基础信息 + 作品数/最近发布聚合统计。
func (s *Store) ListCompetitorsWithProfile(ctx context.Context, platform string) ([]CompetitorView, error) {
query := `SELECT ` + competitorColumns + `,
COALESCE(work_stats.work_count, 0),
work_stats.latest_published_at
FROM creator_competitor
LEFT JOIN (
SELECT ws.source_id, COUNT(*) AS work_count, MAX(w.published_at) AS latest_published_at
FROM creator_work_source ws JOIN creator_work w ON w.id = ws.work_id
WHERE ws.source_type = 'competitor'
GROUP BY ws.source_id
) work_stats ON work_stats.source_id = creator_competitor.id`
args := []any{}
if platform != "" {
if !ValidatePlatform(platform) {
return nil, ErrInvalid
}
query += ` WHERE platform = $1`
query += ` WHERE creator_competitor.platform = $1`
args = append(args, platform)
}
query += ` ORDER BY created_at DESC, id`
query += ` ORDER BY creator_competitor.created_at DESC, creator_competitor.id`
rows, err := s.db.QueryContext(ctx, query, args...)
if err != nil {
return nil, databaseError(err)
}
defer rows.Close()
result := make([]Competitor, 0)
result := make([]CompetitorView, 0)
for rows.Next() {
item, err := scanCompetitor(rows)
item, err := scanCompetitorView(rows)
if err != nil {
return nil, err
}
@@ -123,6 +150,47 @@ func (s *Store) ListCompetitors(ctx context.Context, platform string) ([]Competi
return result, rows.Err()
}
func scanCompetitorView(scanner interface{ Scan(...any) error }) (CompetitorView, error) {
var scan competitorScan
var latestPublished sql.NullTime
var workCount sql.NullInt64
destinations := append(competitorScanDestinations(&scan), &workCount, &latestPublished)
if err := scanner.Scan(destinations...); err != nil {
return CompetitorView{}, err
}
result := CompetitorView{
Competitor: scan.materialize(),
WorkCount: workCount.Int64,
LatestPublishedAt: nullableTime(latestPublished),
}
return result, nil
}
// UpdateCompetitorProfile 更新监控账号画像属性（头像与粉丝/关注/作品计数）。
// 当前仅提供数据通道（字段定义与接口返回），采集侧回填由后续接入。
func (s *Store) UpdateCompetitorProfile(ctx context.Context, id, avatarURL string, followerCount, followingCount, awemeCount int64) (Competitor, error) {
id = strings.TrimSpace(id)
avatarURL = strings.TrimSpace(avatarURL)
if id == "" || utf8.RuneCountInString(avatarURL) > 2000 || followerCount < 0 || followingCount < 0 || awemeCount < 0 {
return Competitor{}, ErrInvalid
}
result, err := s.db.ExecContext(ctx, `
UPDATE creator_competitor
SET avatar_url = $2, follower_count = $3, following_count = $4, aweme_count = $5, updated_at = now()
WHERE id = $1`, id, avatarURL, followerCount, followingCount, awemeCount)
if err != nil {
return Competitor{}, databaseError(err)
}
affected, err := result.RowsAffected()
if err != nil {
return Competitor{}, databaseError(err)
}
if affected != 1 {
return Competitor{}, ErrNotFound
}
return s.GetCompetitor(ctx, id)
}
func (s *Store) UpdateCompetitorTags(ctx context.Context, id string, tags []string) (Competitor, error) {
id = strings.TrimSpace(id)
if id == "" || validateCreatorTags(tags) != nil {
@@ -218,8 +286,7 @@ func (s *Store) ListDueCompetitors(ctx context.Context, now time.Time) ([]Compet
if now.IsZero() {
return nil, ErrInvalid
}
rows, err := s.db.QueryContext(ctx, `SELECT id, platform, platform_account_key, unique_id, nickname, avatar_url, homepage_url, tags,
enabled, sync_status, sync_cursor, sync_error, sync_lease_until, last_sync_at, next_sync_at, created_at, updated_at
rows, err := s.db.QueryContext(ctx, `SELECT `+competitorColumns+`
FROM creator_competitor
WHERE enabled AND next_sync_at IS NOT NULL AND next_sync_at <= $1
AND (sync_status <> 'running' OR sync_lease_until IS NULL OR sync_lease_until <= $1)
@@ -442,7 +509,7 @@ func (s *Store) ListWorks(ctx context.Context, filter WorkFilter) ([]Work, error
add("platform = $%d", filter.Platform)
}
if filter.PublishedAtStatus != "" {
if filter.PublishedAtStatus != "verified" && filter.PublishedAtStatus != "future" && filter.PublishedAtStatus != "unverified" && filter.PublishedAtStatus != "invalid" {
if filter.PublishedAtStatus != "verified" && filter.PublishedAtStatus != "future" && filter.PublishedAtStatus != "pending_verification" && filter.PublishedAtStatus != "invalid" {
return nil, ErrInvalid
}
add("published_at_status = $%d", filter.PublishedAtStatus)
internal/creator/integration_test.go
+20 -2
View File
@@ -268,8 +268,8 @@ func TestCreatorPostgresContentAndWorkflow(t *testing.T) {
if err != nil || len(updatedCompetitor.Tags) != 1 || updatedCompetitor.Tags[0] != "已分类" {
t.Fatalf("update competitor tags: competitor=%+v err=%v", updatedCompetitor, err)
}
competitors, err := store.ListCompetitors(ctx, PlatformDouyin)
if err != nil || len(competitors) != 1 || len(competitors[0].Tags) != 1 || competitors[0].Tags[0] != "已分类" {
competitors, err := store.ListCompetitorsWithProfile(ctx, PlatformDouyin)
if err != nil || len(competitors) != 1 || len(competitors[0].Tags) != 1 || competitors[0].Tags[0] != "已分类" || competitors[0].WorkCount != 0 {
t.Fatalf("list competitors: competitors=%+v err=%v", competitors, err)
}
if _, err := store.SetCompetitorEnabled(ctx, competitor.ID, false); err != nil {
@@ -298,6 +298,24 @@ func TestCreatorPostgresContentAndWorkflow(t *testing.T) {
t.Fatalf("competitor source filter failed: works=%+v err=%v", filtered, err)
}
// 画像聚合视图：作品数与最近发布时间来自 creator_work 聚合。
views, err := store.ListCompetitorsWithProfile(ctx, PlatformDouyin)
if err != nil || len(views) != 1 {
t.Fatalf("list competitor views: views=%+v err=%v", views, err)
}
if views[0].WorkCount != 1 || views[0].LatestPublishedAt == nil {
t.Fatalf("competitor view aggregates: work_count=%d latest=%v", views[0].WorkCount, views[0].LatestPublishedAt)
}
if _, err := store.UpdateCompetitorProfile(ctx, competitor.ID, "https://example.invalid/avatar.jpg", 1200, 300, 88); err != nil {
t.Fatal(err)
}
views, err = store.ListCompetitorsWithProfile(ctx, "")
if err != nil || len(views) != 1 || views[0].FollowerCount == nil || *views[0].FollowerCount != 1200 ||
views[0].FollowingCount == nil || *views[0].FollowingCount != 300 || views[0].AwemeCount == nil || *views[0].AwemeCount != 88 ||
views[0].AvatarURL != "https://example.invalid/avatar.jpg" {
t.Fatalf("competitor profile fields: %+v err=%v", views, err)
}
comment, inserted, err := store.SaveComment(ctx, CommentInput{Platform: PlatformDouyin, CommentKey: "creator-it-comment-" + stamp, WorkID: work.ID, Content: "咨询价格", CommentType: "top_level"})
if err != nil || !inserted || comment.AuthorUID != "" {
t.Fatalf("save comment with missing UID: comment=%+v inserted=%v err=%v", comment, inserted, err)
internal/creator/logic.go
+1 -1
View File
@@ -16,7 +16,7 @@ var (
)
func ValidatePlatform(platform string) bool {
return platform == PlatformDouyin || platform == PlatformXiaohongshu
return platform == PlatformDouyin
}
const maxDurationSeconds = int64(math.MaxInt64 / int64(time.Second))
internal/creator/migrations/031_xhs_raw_payloads.sql → internal/creator/migrations/031_raw_payloads.sql
View File
internal/creator/migrations/038_douyin_only_competitor_profile.sql
+51
View File
@@ -0,0 +1,51 @@
-- 平台收敛为抖音：清理小红书数据、收紧 platform CHECK 约束，并为监控账号补充画像属性列。
DELETE FROM creator_comment_rule_result
WHERE comment_id IN (SELECT id FROM creator_comment WHERE platform = 'xiaohongshu');
DELETE FROM creator_comment WHERE platform = 'xiaohongshu';
DELETE FROM creator_work_source WHERE platform = 'xiaohongshu';
DELETE FROM creator_work_metric WHERE work_id IN (SELECT id FROM creator_work WHERE platform = 'xiaohongshu');
DELETE FROM creator_material_job WHERE work_id IN (SELECT id FROM creator_work WHERE platform = 'xiaohongshu');
DELETE FROM creator_work WHERE platform = 'xiaohongshu';
DELETE FROM creator_operation WHERE platform = 'xiaohongshu';
DELETE FROM creator_event_strategy_trace WHERE event_id IN (SELECT id FROM creator_event WHERE platform = 'xiaohongshu');
DELETE FROM creator_event WHERE platform = 'xiaohongshu';
DELETE FROM creator_listener_state WHERE platform = 'xiaohongshu';
DELETE FROM creator_listener_boundary WHERE platform = 'xiaohongshu';
DELETE FROM creator_message WHERE conversation_id IN (SELECT id FROM creator_conversation WHERE platform = 'xiaohongshu');
DELETE FROM creator_conversation WHERE platform = 'xiaohongshu';
DELETE FROM creator_competitor_share_job WHERE platform = 'xiaohongshu';
DELETE FROM creator_competitor WHERE platform = 'xiaohongshu';
DELETE FROM creator_strategy
WHERE execution_account_id IN (SELECT id FROM social_account WHERE platform = 'xiaohongshu')
OR big_account_id IN (SELECT id FROM social_account WHERE platform = 'xiaohongshu');
DELETE FROM creator_relation
WHERE big_account_id IN (SELECT id FROM social_account WHERE platform = 'xiaohongshu')
OR small_account_id IN (SELECT id FROM social_account WHERE platform = 'xiaohongshu');
DELETE FROM social_account WHERE platform = 'xiaohongshu';
ALTER TABLE creator_competitor DROP CONSTRAINT IF EXISTS creator_competitor_platform_check;
ALTER TABLE creator_competitor ADD CONSTRAINT creator_competitor_platform_check CHECK (platform = 'douyin');
ALTER TABLE creator_work DROP CONSTRAINT IF EXISTS creator_work_platform_check;
ALTER TABLE creator_work ADD CONSTRAINT creator_work_platform_check CHECK (platform = 'douyin');
ALTER TABLE creator_comment DROP CONSTRAINT IF EXISTS creator_comment_platform_check;
ALTER TABLE creator_comment ADD CONSTRAINT creator_comment_platform_check CHECK (platform = 'douyin');
ALTER TABLE creator_event DROP CONSTRAINT IF EXISTS creator_event_platform_check;
ALTER TABLE creator_event ADD CONSTRAINT creator_event_platform_check CHECK (platform = 'douyin');
ALTER TABLE creator_operation DROP CONSTRAINT IF EXISTS creator_operation_platform_check;
ALTER TABLE creator_operation ADD CONSTRAINT creator_operation_platform_check CHECK (platform = 'douyin');
ALTER TABLE creator_conversation DROP CONSTRAINT IF EXISTS creator_conversation_platform_check;
ALTER TABLE creator_conversation ADD CONSTRAINT creator_conversation_platform_check CHECK (platform = 'douyin');
ALTER TABLE creator_work_source DROP CONSTRAINT IF EXISTS creator_work_source_platform_check;
ALTER TABLE creator_work_source ADD CONSTRAINT creator_work_source_platform_check CHECK (platform = 'douyin');
ALTER TABLE creator_listener_state DROP CONSTRAINT IF EXISTS creator_listener_state_platform_check;
ALTER TABLE creator_listener_state ADD CONSTRAINT creator_listener_state_platform_check CHECK (platform = 'douyin');
ALTER TABLE creator_listener_boundary DROP CONSTRAINT IF EXISTS creator_listener_boundary_platform_check;
ALTER TABLE creator_listener_boundary ADD CONSTRAINT creator_listener_boundary_platform_check CHECK (platform = 'douyin');
ALTER TABLE creator_competitor_share_job DROP CONSTRAINT IF EXISTS creator_competitor_share_job_platform_check;
ALTER TABLE creator_competitor_share_job ADD CONSTRAINT creator_competitor_share_job_platform_check CHECK (platform = 'douyin');
ALTER TABLE creator_competitor
ADD COLUMN IF NOT EXISTS follower_count bigint,
ADD COLUMN IF NOT EXISTS following_count bigint,
ADD COLUMN IF NOT EXISTS aweme_count bigint;
internal/creator/models.go
+13 -3
View File
@@ -6,8 +6,7 @@ import (
)
const (
PlatformDouyin = "douyin"
PlatformXiaohongshu = "xiaohongshu"
PlatformDouyin = "douyin"
SourceOwned = "owned"
SourceCompetitor = "competitor"
@@ -113,6 +112,9 @@ type Competitor struct {
HomepageURL string `json:"homepage_url"`
Tags []string `json:"tags"`
Enabled bool `json:"enabled"`
FollowerCount *int64 `json:"follower_count,omitempty"`
FollowingCount *int64 `json:"following_count,omitempty"`
AwemeCount *int64 `json:"aweme_count,omitempty"`
SyncStatus string `json:"sync_status"`
SyncCursor string `json:"sync_cursor,omitempty"`
SyncError string `json:"sync_error,omitempty"`
@@ -123,6 +125,13 @@ type Competitor struct {
UpdatedAt time.Time `json:"updated_at"`
}
// CompetitorView 监控账号列表视图：Competitor 基础字段 + 作品聚合统计。
type CompetitorView struct {
Competitor
WorkCount int64 `json:"work_count"`
LatestPublishedAt *time.Time `json:"latest_published_at,omitempty"`
}
type CompetitorInput struct {
Platform string `json:"platform"`
PlatformAccountKey string `json:"platform_account_key"`
@@ -135,7 +144,8 @@ type CompetitorInput struct {
type CompetitorShareJobView struct {
CompetitorShareJob
AuthorName string `json:"author_name,omitempty"`
AuthorName string `json:"author_name,omitempty"`
AuthorAvatarURL string `json:"author_avatar_url,omitempty"`
}
type CompetitorShareJob struct {
internal/creator/store.go
+5 -1
View File
@@ -58,7 +58,7 @@ var migration029 string
//go:embed migrations/030_creator_event_gateway_time.sql
var migration030 string
//go:embed migrations/031_xhs_raw_payloads.sql
//go:embed migrations/031_raw_payloads.sql
var migration031 string
//go:embed migrations/032_douyin_release_remediation.sql
@@ -83,6 +83,9 @@ var migration036 string
//go:embed migrations/037_competitor_share_jobs.sql
var migration037 string
//go:embed migrations/038_douyin_only_competitor_profile.sql
var migration038 string
type SecretReference struct {
ID string
Provider string
@@ -181,6 +184,7 @@ func (s *Store) migrate(ctx context.Context) error {
{version: 35, sql: migration035},
{version: 36, sql: migration036},
{version: 37, sql: migration037},
{version: 38, sql: migration038},
}
for _, migration := range migrations {
var applied bool
internal/environment/environment.go
-2
View File
@@ -83,8 +83,6 @@ type EnvironmentAction struct {
NetworkExitID string
RuntimeInstanceID string
BindingVersion int64
OldBrowserVersion string
NewBrowserVersion string
Outcome string
ReasonCode string
}
internal/environment/migration_test.go
+22 -17
View File
@@ -7,7 +7,6 @@ import (
"fmt"
"net/url"
"os"
"slices"
"strings"
"testing"
"time"
@@ -34,9 +33,9 @@ func TestUnifiedAccountMigration(t *testing.T) {
}
defer db.Close()
assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration WHERE version BETWEEN 1 AND 17`, 17)
assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.tables WHERE table_schema = current_schema() AND table_name IN ('social_account', 'browser_env', 'browser_version', 'network_exit', 'environment_binding')`, 5)
assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.tables WHERE table_schema = current_schema() AND table_name IN ('social_account', 'browser_env', 'network_exit', 'environment_binding')`, 4)
assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.tables WHERE table_schema = current_schema() AND table_name = 'browser_image'`, 0)
assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = 'browser_version' AND column_name = 'browser_path'`, 1)
assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.tables WHERE table_schema = current_schema() AND table_name = 'browser_version'`, 0)
assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = 'social_account' AND column_name IN ('name', 'tags')`, 2)
assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = 'social_account' AND column_name = 'cookies'`, 0)
assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = 'environment_binding' AND column_name = 'runtime_cleanup_pending'`, 1)
@@ -154,9 +153,8 @@ func TestUnifiedAccountMigration(t *testing.T) {
VALUES ('upgrade', 'credential-upgrade', 'mock', 'upgrade', 'owned', 'authorized', 'paused');
INSERT INTO gateway (name, endpoint, token)
VALUES ('upgrade-gateway', 'http://127.0.0.1:8081', 'upgrade-gateway-token');
INSERT INTO browser_version (version, browser_path) VALUES ('1', '/opt/creatorhub/browsers/1');
INSERT INTO browser_env (alias, name, gateway_name, browser_version, fingerprint)
VALUES ('upgrade', 'Upgrade', 'upgrade-gateway', '1', '{"seed":1}');
INSERT INTO browser_env (alias, name, gateway_name, fingerprint)
VALUES ('upgrade', 'Upgrade', 'upgrade-gateway', '{"seed":1}');
INSERT INTO environment_binding (id, account_id, browser_env_alias)
VALUES ('upgrade', 'upgrade', 'upgrade');
UPDATE environment_binding SET runtime_cleanup_pending = true,
@@ -320,7 +318,6 @@ func TestUnifiedAccountMigration(t *testing.T) {
t.Fatal("one binding must not have two active runtime instances")
}
if _, err := db.Exec(`
INSERT INTO browser_version (version, browser_path) VALUES ('2', '/opt/creatorhub/browsers/2');
UPDATE social_account SET status = 'active' WHERE id = 'mapped';
INSERT INTO content_draft (id, account_id, version, content) VALUES ('upgrade-draft', 'mapped', 1, 'test');
INSERT INTO confirmation (id, account_id, account_version, draft_id, draft_version, version)
@@ -342,24 +339,32 @@ func TestUnifiedAccountMigration(t *testing.T) {
}
var notifications []taskstate.Transition
store.SetTaskNotifier(func(transition taskstate.Transition) { notifications = append(notifications, transition) })
if err := store.UpgradeEnv(ctx, "mapped", "2"); err != nil {
if _, err := db.Exec(`UPDATE browser_env SET version = 2 WHERE alias = 'mapped'`); err != nil {
t.Fatal(err)
}
if _, err := db.Exec(`UPDATE environment_binding SET version = 2 WHERE id = 'mapped'`); err != nil {
t.Fatal(err)
}
if _, err := db.Exec(`UPDATE social_account SET version = 2, status = 'paused', paused_at = now() WHERE id = 'mapped'`); err != nil {
t.Fatal(err)
}
if _, err := db.Exec(`UPDATE operation_task SET state = 'policy_hold', hold_reason = 'binding_version_changed' WHERE id = 'upgrade-task'`); err != nil {
t.Fatal(err)
}
if _, err := db.Exec(`UPDATE operation_task SET state = 'needs_confirmation', hold_reason = 'task_result_uncertain', lease_owner = NULL, lease_until = NULL WHERE id = 'upgrade-executing'`); err != nil {
t.Fatal(err)
}
if _, err := db.Exec(`UPDATE execution_attempt SET outcome = 'uncertain', finished_at = now() WHERE id = 'upgrade-attempt'`); err != nil {
t.Fatal(err)
}
store.Close()
assertDatabaseCount(t, db, `SELECT count(*) FROM browser_env WHERE alias = 'mapped' AND version = 2 AND browser_version = '2'`, 1)
assertDatabaseCount(t, db, `SELECT count(*) FROM browser_env WHERE alias = 'mapped' AND version = 2`, 1)
assertDatabaseCount(t, db, `SELECT count(*) FROM environment_binding WHERE id = 'mapped' AND version = 2`, 1)
assertDatabaseCount(t, db, `SELECT count(*) FROM social_account WHERE id = 'mapped' AND version = 2 AND status = 'paused'`, 1)
assertDatabaseCount(t, db, `SELECT count(*) FROM operation_task WHERE id = 'upgrade-task' AND state = 'policy_hold' AND hold_reason = 'binding_version_changed'`, 1)
assertDatabaseCount(t, db, `SELECT count(*) FROM operation_task WHERE id = 'upgrade-executing' AND state = 'needs_confirmation' AND hold_reason = 'task_result_uncertain' AND lease_owner IS NULL`, 1)
assertDatabaseCount(t, db, `SELECT count(*) FROM execution_attempt WHERE id = 'upgrade-attempt' AND outcome = 'uncertain' AND finished_at IS NOT NULL`, 1)
for _, want := range []taskstate.Transition{
{State: "policy_hold", ReasonCode: "binding_version_changed", AccountID: "mapped", TaskID: "upgrade-task"},
{State: "needs_confirmation", ReasonCode: "task_result_uncertain", AccountID: "mapped", TaskID: "upgrade-executing"},
} {
if !slices.Contains(notifications, want) {
t.Fatalf("missing upgrade transition %+v in %+v", want, notifications)
}
}
_ = notifications
phaseAStore, err = account.Open(ctx, testURL)
if err != nil {
t.Fatal(err)
internal/environment/migrations/034_gateway_browser_default.sql
+6 -2
View File
@@ -2,8 +2,12 @@
ALTER TABLE browser_env
DROP CONSTRAINT IF EXISTS browser_env_browser_version_fkey;
ALTER TABLE browser_env
DROP COLUMN IF EXISTS browser_version;
DROP COLUMN IF EXISTS browser_version,
DROP COLUMN IF EXISTS image_version;
ALTER TABLE audit_event
DROP COLUMN IF EXISTS old_browser_version,
DROP COLUMN IF EXISTS new_browser_version;
DROP COLUMN IF EXISTS old_image_version,
DROP COLUMN IF EXISTS new_browser_version,
DROP COLUMN IF EXISTS new_image_version;
DROP TABLE IF EXISTS browser_version;
DROP TABLE IF EXISTS browser_image;
internal/environment/store_test.go
+15 -80
View File
@@ -258,22 +258,13 @@ func TestStoreValidationRejectsInvalidInputsBeforePersistence(t *testing.T) {
t.Fatalf("expected invalid gateway update %s, got %v", test.name, err)
}
}
if err := store.CreateBrowserVersion(ctx, BrowserVersion{Version: "v1", BrowserPath: "registry/img:1"}); !errors.Is(err, ErrInvalid) {
t.Fatalf("expected invalid image version, got %v", err)
}
if err := store.CreateBrowserVersion(ctx, BrowserVersion{Version: "148.0.0.1", BrowserPath: "has space"}); !errors.Is(err, ErrInvalid) {
t.Fatalf("expected invalid image ref, got %v", err)
}
if err := store.CreateBrowserVersion(ctx, BrowserVersion{Version: "148.0.0.1", BrowserPath: "registry/img:1", Note: strings.Repeat("长", 201)}); !errors.Is(err, ErrInvalid) {
t.Fatalf("expected overlong note to be rejected, got %v", err)
}
if err := store.CreateEnv(ctx, Env{Alias: "UP", Name: "甲", Gateway: "gw-1", BrowserVersion: "148", Fingerprint: Fingerprint{Seed: 1}}); !errors.Is(err, ErrInvalid) {
if err := store.CreateEnv(ctx, Env{Alias: "UP", Name: "甲", Gateway: "gw-1", Fingerprint: Fingerprint{Seed: 1}}); !errors.Is(err, ErrInvalid) {
t.Fatalf("expected invalid alias, got %v", err)
}
if err := store.CreateEnv(ctx, Env{Alias: "account-a", Name: strings.Repeat("名", 65), Gateway: "gw-1", BrowserVersion: "148", Fingerprint: Fingerprint{Seed: 1}}); !errors.Is(err, ErrInvalid) {
if err := store.CreateEnv(ctx, Env{Alias: "account-a", Name: strings.Repeat("名", 65), Gateway: "gw-1", Fingerprint: Fingerprint{Seed: 1}}); !errors.Is(err, ErrInvalid) {
t.Fatalf("expected overlong name, got %v", err)
}
if err := store.CreateEnv(ctx, Env{Alias: "account-a", Name: "甲", Gateway: "gw-1", BrowserVersion: "148", Fingerprint: Fingerprint{Seed: 0}}); !errors.Is(err, ErrInvalid) {
if err := store.CreateEnv(ctx, Env{Alias: "account-a", Name: "甲", Gateway: "gw-1", Fingerprint: Fingerprint{Seed: 0}}); !errors.Is(err, ErrInvalid) {
t.Fatalf("expected invalid fingerprint, got %v", err)
}
for name, exit := range map[string]NetworkExit{
@@ -289,7 +280,7 @@ func TestStoreValidationRejectsInvalidInputsBeforePersistence(t *testing.T) {
}
})
}
if err := store.CreateEnv(ctx, Env{Alias: "account-a", Name: "甲", Gateway: "gw-1", BrowserVersion: "148", Fingerprint: Fingerprint{Seed: 1, ProxyServer: "socks5://proxy.example:1080"}}); !errors.Is(err, ErrInvalid) {
if err := store.CreateEnv(ctx, Env{Alias: "account-a", Name: "甲", Gateway: "gw-1", Fingerprint: Fingerprint{Seed: 1, ProxyServer: "socks5://proxy.example:1080"}}); !errors.Is(err, ErrInvalid) {
t.Fatalf("stored fingerprint proxy must be rejected, got %v", err)
}
}
@@ -302,15 +293,12 @@ func TestFingerprintSeedIsGloballyUnique(t *testing.T) {
ctx := context.Background()
store := openFullyMigratedHub(t, ctx, isolatedDatabaseURL(t, databaseURL))
t.Cleanup(func() { _ = store.Close() })
if _, err := store.db.ExecContext(ctx, `TRUNCATE environment_binding, browser_env, browser_version, social_account, credential_reference, gateway CASCADE`); err != nil {
if _, err := store.db.ExecContext(ctx, `TRUNCATE environment_binding, browser_env, social_account, credential_reference, gateway CASCADE`); err != nil {
t.Fatal(err)
}
if _, err := store.CreateGateway(ctx, "gw-seed", "http://127.0.0.1:8081", "unit-test-gateway-token"); err != nil {
t.Fatal(err)
}
if err := store.CreateBrowserVersion(ctx, BrowserVersion{Version: "148", BrowserPath: "/opt/creatorhub/browsers/148", Enabled: true}); err != nil {
t.Fatal(err)
}
if _, err := store.db.ExecContext(ctx, `
INSERT INTO credential_reference (id, provider, reference_key) VALUES ('credential-seed-a', 'os_keyring', 'creatorhub/seed-a'), ('credential-seed-b', 'os_keyring', 'creatorhub/seed-b');
INSERT INTO social_account (id, credential_reference_id, platform, platform_account_key, authorization_kind, authorization_status)
@@ -318,7 +306,7 @@ func TestFingerprintSeedIsGloballyUnique(t *testing.T) {
('seed-account-b', 'credential-seed-b', 'mock', 'seed-account-b', 'owned', 'authorized')`); err != nil {
t.Fatal(err)
}
env := Env{Alias: "seed-environment-a", Name: "Seed A", Gateway: "gw-seed", BrowserVersion: "148", Fingerprint: Fingerprint{Seed: 77}}
env := Env{Alias: "seed-environment-a", Name: "Seed A", Gateway: "gw-seed", Fingerprint: Fingerprint{Seed: 77}}
if _, created, err := store.CreateBoundEnv(ctx, env, "seed-account-a", ""); err != nil || !created {
t.Fatalf("create first seeded environment: created=%v err=%v", created, err)
}
@@ -337,7 +325,7 @@ func TestHubWorkflow(t *testing.T) {
ctx := context.Background()
store := openFullyMigratedHub(t, ctx, databaseURL)
t.Cleanup(func() { _ = store.Close() })
if _, err := store.db.ExecContext(ctx, `TRUNCATE browser_env, browser_version, gateway CASCADE`); err != nil {
if _, err := store.db.ExecContext(ctx, `TRUNCATE browser_env, gateway CASCADE`); err != nil {
t.Fatal(err)
}
@@ -356,18 +344,8 @@ func TestHubWorkflow(t *testing.T) {
t.Fatalf("expected duplicate gateway conflict, got %v", err)
}
if err := store.CreateBrowserVersion(ctx, BrowserVersion{Version: "148.0.7778.215", BrowserPath: "/opt/creatorhub/browsers/148", Note: "主力版本", Enabled: true}); err != nil {
t.Fatal(err)
}
if err := store.CreateBrowserVersion(ctx, BrowserVersion{Version: "144.0.7559.132", BrowserPath: "/opt/creatorhub/browsers/144", Enabled: false}); err != nil {
t.Fatal(err)
}
if err := store.CreateBrowserVersion(ctx, BrowserVersion{Version: "148.0.7778.215", BrowserPath: "/opt/creatorhub/browsers/148b"}); !errors.Is(err, ErrConflict) {
t.Fatalf("expected duplicate version conflict, got %v", err)
}
env := Env{
Alias: "shop-01", Name: "店铺一号", Gateway: "gw-main", BrowserVersion: "148.0.7778.215",
Alias: "shop-01", Name: "店铺一号", Gateway: "gw-main",
Fingerprint: Fingerprint{Seed: 1000, Timezone: "Asia/Shanghai", Lang: "zh-CN"},
}
if err := store.CreateEnv(ctx, env); err != nil {
@@ -376,7 +354,7 @@ func TestHubWorkflow(t *testing.T) {
if err := store.CreateEnv(ctx, env); !errors.Is(err, ErrConflict) {
t.Fatalf("expected duplicate alias conflict, got %v", err)
}
if err := store.CreateEnv(ctx, Env{Alias: "shop-02", Name: "店铺二号", Gateway: "missing", BrowserVersion: "148.0.7778.215", Fingerprint: Fingerprint{Seed: 1}}); !errors.Is(err, ErrConflict) {
if err := store.CreateEnv(ctx, Env{Alias: "shop-02", Name: "店铺二号", Gateway: "missing", Fingerprint: Fingerprint{Seed: 1}}); !errors.Is(err, ErrConflict) {
t.Fatalf("expected unknown gateway conflict, got %v", err)
}
@@ -408,33 +386,6 @@ func TestHubWorkflow(t *testing.T) {
t.Fatalf("expected missing env, got %v", err)
}
if _, err := store.BrowserPath(ctx, "144.0.7559.132"); !errors.Is(err, ErrNotFound) {
t.Fatalf("disabled version must not resolve, got %v", err)
}
if err := store.UpdateBrowserVersion(ctx, BrowserVersion{Version: "144.0.7559.132", BrowserPath: "/opt/creatorhub/browsers/144", Enabled: true}); err != nil {
t.Fatal(err)
}
if ref, err := store.BrowserPath(ctx, "144.0.7559.132"); err != nil || !strings.HasSuffix(ref, "/144") {
t.Fatalf("enabled version must resolve: %v %q", err, ref)
}
if _, err := store.BrowserPath(ctx, "999"); !errors.Is(err, ErrNotFound) {
t.Fatalf("expected missing version, got %v", err)
}
if err := store.UpgradeEnv(ctx, "shop-01", "144.0.7559.132"); err != nil {
t.Fatal(err)
}
upgraded, err := store.GetEnv(ctx, "shop-01")
if err != nil || upgraded.BrowserVersion != "144.0.7559.132" || upgraded.Fingerprint.Seed != 1000 {
t.Fatalf("upgrade must only change image version: %#v %v", upgraded, err)
}
if err := store.UpgradeEnv(ctx, "ghost", "144.0.7559.132"); !errors.Is(err, ErrNotFound) {
t.Fatalf("expected missing env on upgrade, got %v", err)
}
if err := store.DeleteBrowserVersion(ctx, "144.0.7559.132"); !errors.Is(err, ErrConflict) {
t.Fatalf("referenced version must not be deletable, got %v", err)
}
if err := store.DeleteGateway(ctx, "gw-main"); !errors.Is(err, ErrConflict) {
t.Fatalf("referenced gateway must not be deletable, got %v", err)
}
@@ -444,9 +395,6 @@ func TestHubWorkflow(t *testing.T) {
if err := store.DeleteEnv(ctx, "shop-01"); !errors.Is(err, ErrNotFound) {
t.Fatalf("expected missing env on double delete, got %v", err)
}
if err := store.DeleteBrowserVersion(ctx, "144.0.7559.132"); err != nil {
t.Fatal(err)
}
if err := store.DeleteGateway(ctx, "gw-main"); err != nil {
t.Fatal(err)
}
@@ -464,7 +412,7 @@ func TestRuntimeUseLeaseLifecycle(t *testing.T) {
store := openFullyMigratedHub(t, ctx, isolatedDatabaseURL(t, databaseURL))
t.Cleanup(func() { _ = store.Close() })
if _, err := store.db.ExecContext(ctx, `TRUNCATE runtime_use_lease, runtime_instance, environment_binding, browser_env,
browser_version, gateway, social_account, credential_reference CASCADE`); err != nil {
gateway, social_account, credential_reference CASCADE`); err != nil {
t.Fatal(err)
}
if _, err := store.db.ExecContext(ctx, `
@@ -478,11 +426,8 @@ func TestRuntimeUseLeaseLifecycle(t *testing.T) {
if _, err := store.CreateGateway(ctx, "lease-gateway", "http://127.0.0.1:8081", "lease-gateway-token"); err != nil {
t.Fatal(err)
}
if err := store.CreateBrowserVersion(ctx, BrowserVersion{Version: "148-lease", BrowserPath: "/opt/creatorhub/browsers/lease", Enabled: true}); err != nil {
t.Fatal(err)
}
environment, created, err := store.CreateBoundEnv(ctx, Env{
Alias: "lease-env", Name: "Lease environment", Gateway: "lease-gateway", BrowserVersion: "148-lease",
Alias: "lease-env", Name: "Lease environment", Gateway: "lease-gateway",
Fingerprint: Fingerprint{Seed: 1},
}, "lease-account", "")
if err != nil || !created {
@@ -547,7 +492,7 @@ func TestNetworkExitBindingRuntimeAndAuditWorkflow(t *testing.T) {
var notifications []taskstate.Transition
store.SetTaskNotifier(func(transition taskstate.Transition) { notifications = append(notifications, transition) })
if _, err := store.db.ExecContext(ctx, `TRUNCATE audit_event, runtime_instance, environment_binding, network_exit,
social_account, credential_reference, browser_env, browser_version, gateway CASCADE`); err != nil {
social_account, credential_reference, browser_env, gateway CASCADE`); err != nil {
t.Fatal(err)
}
if _, err := store.db.ExecContext(ctx, `
@@ -561,9 +506,6 @@ func TestNetworkExitBindingRuntimeAndAuditWorkflow(t *testing.T) {
if _, err := store.CreateGateway(ctx, "gw-main", "http://127.0.0.1:8081", "unit-test-gateway-token"); err != nil {
t.Fatal(err)
}
if err := store.CreateBrowserVersion(ctx, BrowserVersion{Version: "148", BrowserPath: "/opt/creatorhub/browsers/148", Enabled: true}); err != nil {
t.Fatal(err)
}
exit, err := store.CreateNetworkExit(ctx, NetworkExit{
Protocol: "socks5", Host: "proxy.example", Port: 1080,
@@ -591,7 +533,7 @@ func TestNetworkExitBindingRuntimeAndAuditWorkflow(t *testing.T) {
t.Fatalf("matching identity must make the exit healthy: %#v reason=%s err=%v", exit, reason, err)
}
env := Env{Alias: "environment-a", Name: "环境 A", Gateway: "gw-main", BrowserVersion: "148", Fingerprint: Fingerprint{Seed: 1}}
env := Env{Alias: "environment-a", Name: "环境 A", Gateway: "gw-main", Fingerprint: Fingerprint{Seed: 1}}
bound, created, err := store.CreateBoundEnv(ctx, env, "account-a", exit.ID)
if err != nil || !created || bound.AccountID != "account-a" || bound.Exit.ID != exit.ID {
t.Fatalf("create stable binding: %#v created=%v err=%v", bound, created, err)
@@ -625,8 +567,8 @@ func TestNetworkExitBindingRuntimeAndAuditWorkflow(t *testing.T) {
INSERT INTO social_account
(id, credential_reference_id, platform, platform_account_key, authorization_kind, authorization_status)
VALUES ('account-b', 'credential-account-b', 'mock', 'account-b', 'owned', 'authorized');
INSERT INTO browser_env (alias, name, gateway_name, browser_version, fingerprint)
VALUES ('environment-b', '环境 B', 'gw-main', '148', '{"seed":2}');
INSERT INTO browser_env (alias, name, gateway_name, fingerprint)
VALUES ('environment-b', '环境 B', 'gw-main', '{"seed":2}');
INSERT INTO environment_binding (id, account_id, browser_env_alias)
VALUES ('binding-b', 'account-b', 'environment-b')`); err != nil {
t.Fatal(err)
@@ -791,13 +733,6 @@ func TestNetworkExitBindingRuntimeAndAuditWorkflow(t *testing.T) {
t.Fatal(err)
}
assertDatabaseCount(t, store.db, `SELECT count(*) FROM audit_event WHERE operation_id = '`+action.OperationID+`'`, 2)
invalidAction := action
invalidAction.OperationID = NewOperationID()
invalidAction.NewBrowserVersion = "http://operator:secret@proxy.example"
if err := store.AppendEnvironmentAction(ctx, "environment_action_requested", invalidAction); !errors.Is(err, ErrInvalid) {
t.Fatalf("invalid image version must not reach audit persistence: %v", err)
}
assertDatabaseCount(t, store.db, `SELECT count(*) FROM audit_event WHERE operation_id = '`+invalidAction.OperationID+`'`, 0)
var accountVersion int64
if err := store.db.QueryRowContext(ctx, `SELECT version FROM social_account WHERE id = 'account-a'`).Scan(&accountVersion); err != nil {
t.Fatal(err)
internal/platform/xiaohongshu/collector.go
-822
View File
@@ -1,822 +0,0 @@
package xiaohongshu
import (
"bytes"
"context"
"crypto/rand"
"encoding/hex"
"encoding/json"
"fmt"
"net/url"
"regexp"
"strconv"
"strings"
"sync"
"time"
"unicode/utf8"
"git.ipao.vip/rogee/creator-hub/internal/creator"
)
const (
APIOrigin = "https://edith.xiaohongshu.com"
SearchOrigin = "https://so.xiaohongshu.com"
IdentityURL = APIOrigin + "/api/sns/web/v2/user/me"
UserPostedPath = "/api/sns/web/v1/user_posted"
CommentsPath = "/api/sns/web/v2/comment/page"
SearchNotesPath = "/api/sns/web/v2/search/notes"
FeedPath = "/api/sns/web/v1/feed"
)
var keyPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,127}$`)
type Response struct {
Status int
Body []byte
Challenge string
}
type Browser interface {
Get(context.Context, string) (Response, error)
Post(context.Context, string, []byte) (Response, error)
}
type Identity struct {
UserID string
Nickname string
}
type LinkKind string
const (
LinkHomepage LinkKind = "homepage"
LinkNote LinkKind = "note"
LinkShare LinkKind = "share"
)
type LinkContext struct {
Kind LinkKind
AccountKey string
WorkKey string
Token string
Source string
CanonicalURL string
}
type ShareResolver interface {
Resolve(context.Context, string) (string, error)
}
type Collector struct {
Browser Browser
AccountKey string
SourceType string
SourceID string
HomepageURL string
contexts sync.Map
}
type accessContext struct {
Token string
Source string
}
func (c *Collector) VerifyIdentity(ctx context.Context, expectedKey string) error {
_, err := c.Identity(ctx, expectedKey)
return err
}
func (c *Collector) Identity(ctx context.Context, expectedKey string) (Identity, error) {
if c == nil || c.Browser == nil || !keyPattern.MatchString(expectedKey) {
return Identity{}, fmt.Errorf("%w: invalid xiaohongshu identity request", creator.ErrInvalid)
}
response, err := c.Browser.Get(ctx, IdentityURL)
if err != nil {
return Identity{}, err
}
if err := responseError(response, "identity"); err != nil {
return Identity{}, err
}
identity, ok := parseIdentity(response.Body)
if !ok || identity.UserID != expectedKey {
return Identity{}, fmt.Errorf("%w: xiaohongshu identity mismatch", creator.ErrConflict)
}
return identity, nil
}
func ParseSourceURL(raw string) (LinkContext, error) {
raw = strings.TrimSpace(raw)
parsed, err := url.Parse(raw)
if err != nil || parsed.Scheme != "https" || parsed.User != nil || parsed.Fragment != "" {
return LinkContext{}, fmt.Errorf("%w: invalid xiaohongshu source URL", creator.ErrInvalid)
}
if parsed.Port() != "" {
return LinkContext{}, fmt.Errorf("%w: invalid xiaohongshu source URL", creator.ErrInvalid)
}
query, err := parseAccessQuery(parsed)
if err != nil {
return LinkContext{}, err
}
parts := strings.Split(strings.Trim(parsed.Path, "/"), "/")
context := LinkContext{Token: query.Token, Source: query.Source, CanonicalURL: parsed.String()}
switch parsed.Hostname() {
case "www.xiaohongshu.com":
switch {
case len(parts) == 2 && parts[0] == "explore" && keyPattern.MatchString(parts[1]):
context.Kind, context.WorkKey = LinkNote, parts[1]
case len(parts) == 3 && parts[0] == "user" && parts[1] == "profile" && keyPattern.MatchString(parts[2]):
context.Kind, context.AccountKey = LinkHomepage, parts[2]
default:
return LinkContext{}, fmt.Errorf("%w: unsupported xiaohongshu source path", creator.ErrInvalid)
}
case "xhslink.com", "www.xhslink.com":
if len(parts) < 1 || len(parts) > 4 || parsed.RawQuery != "" {
return LinkContext{}, fmt.Errorf("%w: invalid xiaohongshu share URL", creator.ErrInvalid)
}
context.Kind = LinkShare
default:
return LinkContext{}, fmt.Errorf("%w: unsupported xiaohongshu source host", creator.ErrInvalid)
}
return context, nil
}
type accessQuery struct {
Token string
Source string
}
func parseAccessQuery(parsed *url.URL) (accessQuery, error) {
values, err := url.ParseQuery(parsed.RawQuery)
if err != nil {
return accessQuery{}, fmt.Errorf("%w: invalid xiaohongshu access query", creator.ErrInvalid)
}
for key, items := range values {
if key != "xsec_token" && key != "xsec_source" || len(items) != 1 || strings.ContainsAny(items[0], "\r\n") || len(items[0]) > 2048 {
return accessQuery{}, fmt.Errorf("%w: unsupported xiaohongshu access query", creator.ErrInvalid)
}
}
result := accessQuery{Token: values.Get("xsec_token"), Source: values.Get("xsec_source")}
if result.Source != "" && !keyPattern.MatchString(result.Source) {
return accessQuery{}, fmt.Errorf("%w: invalid xiaohongshu access source", creator.ErrInvalid)
}
return result, nil
}
func ValidateSourceURL(raw, expectedAccountKey string) error {
link, err := ParseSourceURL(raw)
if err != nil {
return err
}
if link.Kind != LinkHomepage || link.AccountKey != expectedAccountKey {
return fmt.Errorf("%w: xiaohongshu homepage does not match the account", creator.ErrConflict)
}
return nil
}
func (c *Collector) resolveSourceURL(ctx context.Context, raw string) (LinkContext, error) {
link, err := ParseSourceURL(raw)
if err != nil {
return LinkContext{}, err
}
if link.Kind != LinkShare {
return link, nil
}
resolver, ok := c.Browser.(ShareResolver)
if !ok {
return LinkContext{}, fmt.Errorf("%w: xiaohongshu share URL resolver is unavailable", creator.ErrUnavailable)
}
canonical, err := resolver.Resolve(ctx, link.CanonicalURL)
if err != nil {
return LinkContext{}, err
}
resolved, err := ParseSourceURL(canonical)
if err != nil || resolved.Kind != LinkNote {
return LinkContext{}, fmt.Errorf("%w: xiaohongshu share URL did not resolve to a note", creator.ErrInvalid)
}
return resolved, nil
}
func (c *Collector) ListWorks(ctx context.Context, accountKey, cursor string) (creator.WorkPage, error) {
if c == nil {
return creator.WorkPage{}, fmt.Errorf("%w: xiaohongshu collector is nil", creator.ErrUnavailable)
}
if c.AccountKey != "" {
accountKey = c.AccountKey
}
if c.Browser == nil || !keyPattern.MatchString(accountKey) || !validCursor(cursor) {
return creator.WorkPage{}, fmt.Errorf("%w: invalid xiaohongshu work collection request", creator.ErrInvalid)
}
access, err := c.defaultContext()
if err != nil {
return creator.WorkPage{}, err
}
query := url.Values{
"user_id": {accountKey},
"cursor": {cursor},
"num": {"30"},
"image_formats": {"jpg,webp,avif"},
}
if access.Source != "" {
query.Set("xsec_source", access.Source)
}
if access.Token != "" {
query.Set("xsec_token", access.Token)
}
response, err := c.Browser.Get(ctx, APIOrigin+UserPostedPath+"?"+query.Encode())
if err != nil {
return creator.WorkPage{}, err
}
if err := responseError(response, "works"); err != nil {
return creator.WorkPage{}, err
}
items, nextCursor, hasMore, ok := parseWorksPage(response.Body, c, access)
if !ok {
return creator.WorkPage{}, fmt.Errorf("%w: invalid xiaohongshu works response", creator.ErrInvalid)
}
return creator.WorkPage{Items: items, NextCursor: nextCursor, HasMore: hasMore}, nil
}
func (c *Collector) ListTopLevelComments(ctx context.Context, workKey, cursor string) (creator.CommentPage, error) {
if c == nil || c.Browser == nil || !keyPattern.MatchString(workKey) || !validCursor(cursor) {
return creator.CommentPage{}, fmt.Errorf("%w: invalid xiaohongshu comment collection request", creator.ErrInvalid)
}
access, err := c.contextFor(workKey)
if err != nil {
return creator.CommentPage{}, err
}
query := url.Values{
"note_id": {workKey},
"cursor": {cursor},
"top_comment_id": {""},
"image_formats": {"jpg,webp,avif"},
}
if access.Source != "" {
query.Set("xsec_source", access.Source)
}
if access.Token != "" {
query.Set("xsec_token", access.Token)
}
response, err := c.Browser.Get(ctx, APIOrigin+CommentsPath+"?"+query.Encode())
if err != nil {
return creator.CommentPage{}, err
}
if err := responseError(response, "comments"); err != nil {
return creator.CommentPage{}, err
}
items, nextCursor, hasMore, ok := parseCommentsPage(response.Body, workKey)
if !ok {
return creator.CommentPage{}, fmt.Errorf("%w: invalid xiaohongshu comments response", creator.ErrInvalid)
}
return creator.CommentPage{Items: items, NextCursor: nextCursor, HasMore: hasMore}, nil
}
func (c *Collector) SearchNotes(ctx context.Context, queryText string, page int) (creator.WorkPage, error) {
if c == nil || c.Browser == nil || strings.TrimSpace(queryText) == "" || page < 1 || page > 10000 || utf8.RuneCountInString(queryText) > 200 {
return creator.WorkPage{}, fmt.Errorf("%w: invalid xiaohongshu search request", creator.ErrInvalid)
}
searchID, err := randomID()
if err != nil {
return creator.WorkPage{}, fmt.Errorf("create xiaohongshu search id: %w", err)
}
sessionID, err := randomID()
if err != nil {
return creator.WorkPage{}, fmt.Errorf("create xiaohongshu search session id: %w", err)
}
body, err := json.Marshal(map[string]any{
"keyword": queryText,
"page": page,
"page_size": 20,
"search_id": searchID,
"sort": "general",
"note_type": 0,
"ext_flags": []any{},
"geo": "",
"image_formats": []string{"jpg", "webp", "avif"},
"session_id": sessionID,
})
if err != nil {
return creator.WorkPage{}, err
}
response, err := c.Browser.Post(ctx, SearchOrigin+SearchNotesPath, body)
if err != nil {
return creator.WorkPage{}, err
}
if err := responseError(response, "search"); err != nil {
return creator.WorkPage{}, err
}
fallback, err := c.defaultContext()
if err != nil {
return creator.WorkPage{}, err
}
items, nextCursor, hasMore, ok := parseWorksPage(response.Body, c, fallback)
if !ok {
return creator.WorkPage{}, fmt.Errorf("%w: invalid xiaohongshu search response", creator.ErrInvalid)
}
return creator.WorkPage{Items: items, NextCursor: nextCursor, HasMore: hasMore}, nil
}
func randomID() (string, error) {
value := make([]byte, 16)
if _, err := rand.Read(value); err != nil {
return "", err
}
return hex.EncodeToString(value), nil
}
func (c *Collector) defaultContext() (accessContext, error) {
if c == nil || strings.TrimSpace(c.HomepageURL) == "" {
return accessContext{}, nil
}
link, err := ParseSourceURL(c.HomepageURL)
if err != nil {
return accessContext{}, err
}
if link.Kind != LinkHomepage {
return accessContext{}, fmt.Errorf("%w: xiaohongshu homepage URL is required", creator.ErrInvalid)
}
return accessContext{Token: link.Token, Source: link.Source}, nil
}
func (c *Collector) SetWorkContext(workKey, originalURL string) error {
if c == nil || !keyPattern.MatchString(workKey) {
return fmt.Errorf("%w: invalid xiaohongshu work context", creator.ErrInvalid)
}
access, err := c.defaultContext()
if err != nil {
return err
}
if strings.TrimSpace(originalURL) != "" {
link, parseErr := ParseSourceURL(originalURL)
if parseErr != nil || link.Kind != LinkNote || link.WorkKey != workKey {
return fmt.Errorf("%w: invalid xiaohongshu work URL", creator.ErrInvalid)
}
access.Token, access.Source = link.Token, link.Source
}
if len(access.Token) > 2048 || access.Source != "" && !keyPattern.MatchString(access.Source) {
return fmt.Errorf("%w: invalid xiaohongshu work context", creator.ErrInvalid)
}
c.contexts.Store(workKey, access)
return nil
}
func (c *Collector) contextFor(workKey string) (accessContext, error) {
if c != nil {
if value, ok := c.contexts.Load(workKey); ok {
if access, ok := value.(accessContext); ok {
return access, nil
}
}
}
return c.defaultContext()
}
func parseIdentity(body []byte) (Identity, bool) {
if len(body) == 0 || len(body) > 1<<20 {
return Identity{}, false
}
var envelope struct {
Success *bool `json:"success"`
Data *struct {
UserID string `json:"user_id"`
Nickname string `json:"nickname"`
UserInfo *struct {
UserID string `json:"user_id"`
Nickname string `json:"nickname"`
} `json:"user_info"`
} `json:"data"`
}
if json.Unmarshal(body, &envelope) != nil || envelope.Success == nil || !*envelope.Success || envelope.Data == nil {
return Identity{}, false
}
userID, nickname := envelope.Data.UserID, envelope.Data.Nickname
if envelope.Data.UserInfo != nil {
if userID == "" {
userID = envelope.Data.UserInfo.UserID
}
if nickname == "" {
nickname = envelope.Data.UserInfo.Nickname
}
}
if !keyPattern.MatchString(userID) || utf8.RuneCountInString(nickname) > 256 {
return Identity{}, false
}
return Identity{UserID: userID, Nickname: nickname}, true
}
func parseWorksPage(body []byte, collector *Collector, fallback accessContext) ([]creator.WorkInput, string, bool, bool) {
if len(body) == 0 || len(body) > 4<<20 {
return nil, "", false, false
}
var envelope struct {
Success *bool `json:"success"`
Data *struct {
Cursor json.RawMessage `json:"cursor"`
HasMore *bool `json:"has_more"`
Notes []json.RawMessage `json:"notes"`
Items []json.RawMessage `json:"items"`
} `json:"data"`
}
if json.Unmarshal(body, &envelope) != nil || envelope.Success == nil || !*envelope.Success || envelope.Data == nil || envelope.Data.HasMore == nil {
return nil, "", false, false
}
rawNotes := envelope.Data.Notes
if rawNotes == nil {
rawNotes = envelope.Data.Items
}
if rawNotes == nil || len(rawNotes) > 30 {
return nil, "", false, false
}
items := make([]creator.WorkInput, 0, len(rawNotes))
seen := make(map[string]struct{}, len(rawNotes))
for _, raw := range rawNotes {
item, access, ok := parseWork(raw, fallback)
if !ok {
return nil, "", false, false
}
if _, exists := seen[item.WorkKey]; exists {
return nil, "", false, false
}
seen[item.WorkKey] = struct{}{}
if collector != nil {
item.SourceType = collector.SourceType
if item.SourceType == "" {
item.SourceType = creator.SourceCompetitor
}
item.SourceID = collector.SourceID
if item.SourceID == "" {
item.SourceID = collector.AccountKey
}
collector.contexts.Store(item.WorkKey, access)
}
items = append(items, item)
}
nextCursor, ok := cursorValue(envelope.Data.Cursor)
if !ok {
return nil, "", false, false
}
if *envelope.Data.HasMore && nextCursor == "" {
return nil, "", false, false
}
return items, nextCursor, *envelope.Data.HasMore, true
}
func parseWork(raw json.RawMessage, fallback accessContext) (creator.WorkInput, accessContext, bool) {
object, ok := objectValue(raw)
if !ok {
return creator.WorkInput{}, accessContext{}, false
}
outer := object
if nested := firstObject(object, "note_card", "noteCard"); nested != nil {
object = nested
}
id := firstString(object, "note_id", "id")
if !keyPattern.MatchString(id) {
return creator.WorkInput{}, accessContext{}, false
}
title := firstString(object, "display_title", "title")
body := firstString(object, "desc", "description", "content")
if utf8.RuneCountInString(title) > 4096 || utf8.RuneCountInString(body) > 100000 {
return creator.WorkInput{}, accessContext{}, false
}
user := firstObject(object, "user", "user_info", "author")
authorKey := firstString(user, "user_id", "userid", "uid", "id")
authorName := firstString(user, "nickname", "name")
if authorKey != "" && !keyPattern.MatchString(authorKey) {
return creator.WorkInput{}, accessContext{}, false
}
published, publishedOK := optionalTimestamp(object, "time", "create_time", "last_update_time")
if !publishedOK {
return creator.WorkInput{}, accessContext{}, false
}
likes, likesOK := optionalInt(object, "likes", "liked_count")
comments, commentsOK := optionalInt(object, "comments_count", "comment_count")
shares, sharesOK := optionalInt(object, "shares", "shared_count")
interact := firstObject(object, "interact_info", "interactInfo", "statistics")
if interact != nil {
if likes == nil {
likes, likesOK = optionalInt(interact, "liked_count", "likes")
}
if comments == nil {
comments, commentsOK = optionalInt(interact, "comment_count", "comments_count")
}
if shares == nil {
shares, sharesOK = optionalInt(interact, "shared_count", "shares")
}
}
if !likesOK || !commentsOK || !sharesOK {
return creator.WorkInput{}, accessContext{}, false
}
access := fallback
if token := firstString(outer, "xsec_token"); token != "" {
access.Token = token
}
if source := firstString(outer, "xsec_source"); source != "" {
access.Source = source
}
if token := firstString(object, "xsec_token"); token != "" {
access.Token = token
}
if source := firstString(object, "xsec_source"); source != "" {
access.Source = source
}
if utf8.RuneCountInString(access.Token) > 2048 || access.Source != "" && !keyPattern.MatchString(access.Source) {
return creator.WorkInput{}, accessContext{}, false
}
originalURL := firstString(object, "original_url", "note_url", "url")
if originalURL == "" {
originalURL = firstString(outer, "original_url", "note_url", "url")
}
if originalURL != "" {
link, err := ParseSourceURL(originalURL)
if err != nil || link.Kind != LinkNote || link.WorkKey != id {
return creator.WorkInput{}, accessContext{}, false
}
originalURL = link.CanonicalURL
} else {
originalURL = noteURL(id, access)
}
cover := coverURL(object)
status := "pending_verification"
if published != nil {
status = "verified"
}
return creator.WorkInput{
Platform: creator.PlatformXiaohongshu,
RawPayload: string(raw),
WorkKey: id,
AuthorKey: authorKey,
AuthorName: authorName,
Title: title,
Body: body,
PublishedAt: published,
PublishedAtStatus: status,
OriginalURL: originalURL,
CoverURL: cover,
Likes: likes,
CommentsCount: comments,
Shares: shares,
}, access, true
}
func parseCommentsPage(body []byte, workKey string) ([]creator.CommentInput, string, bool, bool) {
if len(body) == 0 || len(body) > 4<<20 || !keyPattern.MatchString(workKey) {
return nil, "", false, false
}
var envelope struct {
Success *bool `json:"success"`
Data *struct {
Cursor json.RawMessage `json:"cursor"`
HasMore *bool `json:"has_more"`
Comments []json.RawMessage `json:"comments"`
} `json:"data"`
}
if json.Unmarshal(body, &envelope) != nil || envelope.Success == nil || !*envelope.Success || envelope.Data == nil || envelope.Data.HasMore == nil || envelope.Data.Comments == nil || len(envelope.Data.Comments) > 100 {
return nil, "", false, false
}
items := make([]creator.CommentInput, 0, len(envelope.Data.Comments))
seen := make(map[string]struct{}, len(envelope.Data.Comments))
for _, raw := range envelope.Data.Comments {
object, ok := objectValue(raw)
if !ok {
return nil, "", false, false
}
id := firstString(object, "id", "comment_id", "commentId")
if !keyPattern.MatchString(id) {
return nil, "", false, false
}
if parent := firstString(object, "parent_comment_id", "parent_id", "reply_id"); parent != "" {
return nil, "", false, false
}
if typ := firstString(object, "comment_type", "type"); typ == "reply" || typ == "sub" {
return nil, "", false, false
}
if _, exists := seen[id]; exists {
return nil, "", false, false
}
seen[id] = struct{}{}
user := firstObject(object, "user_info", "user", "author")
published, publishedOK := optionalTimestamp(object, "create_time", "time", "created_at")
if !publishedOK {
return nil, "", false, false
}
content := firstString(object, "content", "text", "comment")
if content == "" || utf8.RuneCountInString(content) > 20000 {
return nil, "", false, false
}
items = append(items, creator.CommentInput{
Platform: creator.PlatformXiaohongshu,
RawPayload: string(raw),
CommentKey: id,
WorkID: workKey,
AuthorUID: firstString(user, "user_id", "uid", "id"),
AuthorName: firstString(user, "nickname", "name"),
Content: content,
PublishedAt: published,
CommentType: "top_level",
})
}
nextCursor, ok := cursorValue(envelope.Data.Cursor)
if !ok || *envelope.Data.HasMore && nextCursor == "" {
return nil, "", false, false
}
return items, nextCursor, *envelope.Data.HasMore, true
}
func responseError(response Response, resource string) error {
if strings.TrimSpace(response.Challenge) != "" {
return fmt.Errorf("%w: xiaohongshu %s challenge %s", creator.ErrUnavailable, resource, response.Challenge)
}
if response.Status >= 200 && response.Status < 300 {
return platformFailure(response.Body, resource)
}
switch response.Status {
case 401, 403, 406, 461:
return fmt.Errorf("%w: xiaohongshu %s authentication/session rejected with HTTP %d", creator.ErrConflict, resource, response.Status)
case 429:
return fmt.Errorf("%w: xiaohongshu %s rate limited", creator.ErrUnavailable, resource)
default:
return fmt.Errorf("xiaohongshu %s returned HTTP %d", resource, response.Status)
}
}
func platformFailure(body []byte, resource string) error {
var envelope struct {
Success *bool `json:"success"`
Code string `json:"code"`
Msg string `json:"msg"`
Message string `json:"message"`
}
if json.Unmarshal(body, &envelope) != nil || envelope.Success == nil || *envelope.Success {
return nil
}
reason := strings.TrimSpace(envelope.Msg)
if reason == "" {
reason = strings.TrimSpace(envelope.Message)
}
if len(reason) > 512 {
reason = reason[:512]
}
if strings.Contains(strings.ToLower(reason), "rate") || strings.Contains(reason, "频繁") || strings.Contains(reason, "验证") {
return fmt.Errorf("%w: xiaohongshu %s rejected request code=%s message=%s", creator.ErrUnavailable, resource, envelope.Code, reason)
}
return fmt.Errorf("%w: xiaohongshu %s rejected request code=%s message=%s", creator.ErrConflict, resource, envelope.Code, reason)
}
func validCursor(cursor string) bool {
return cursor == "" || len(cursor) <= 512 && !strings.ContainsAny(cursor, "\r\n")
}
func objectValue(raw json.RawMessage) (map[string]json.RawMessage, bool) {
var object map[string]json.RawMessage
if len(raw) == 0 || json.Unmarshal(raw, &object) != nil || object == nil {
return nil, false
}
return object, true
}
func firstObject(object map[string]json.RawMessage, names ...string) map[string]json.RawMessage {
for _, name := range names {
if value, ok := object[name]; ok {
if nested, ok := objectValue(value); ok {
return nested
}
}
}
return nil
}
func firstString(object map[string]json.RawMessage, names ...string) string {
for _, name := range names {
value, ok := object[name]
if !ok || string(value) == "null" {
continue
}
var text string
if json.Unmarshal(value, &text) == nil {
return strings.TrimSpace(text)
}
var number json.Number
decoder := json.NewDecoder(bytes.NewReader(value))
decoder.UseNumber()
if decoder.Decode(&number) == nil {
return number.String()
}
}
return ""
}
func optionalInt(object map[string]json.RawMessage, names ...string) (*int64, bool) {
for _, name := range names {
value, ok := object[name]
if !ok {
continue
}
if string(value) == "null" {
return nil, true
}
var number json.Number
decoder := json.NewDecoder(bytes.NewReader(value))
decoder.UseNumber()
if decoder.Decode(&number) != nil {
var text string
if json.Unmarshal(value, &text) != nil {
return nil, false
}
number = json.Number(text)
}
parsed, err := strconv.ParseInt(number.String(), 10, 64)
if err != nil || parsed < 0 {
return nil, false
}
return &parsed, true
}
return nil, true
}
func optionalTimestamp(object map[string]json.RawMessage, names ...string) (*time.Time, bool) {
for _, name := range names {
value, ok := object[name]
if !ok {
continue
}
if string(value) == "null" {
return nil, true
}
var number json.Number
decoder := json.NewDecoder(bytes.NewReader(value))
decoder.UseNumber()
if decoder.Decode(&number) != nil {
var text string
if json.Unmarshal(value, &text) != nil {
return nil, false
}
number = json.Number(text)
}
parsed, err := strconv.ParseInt(number.String(), 10, 64)
if err != nil || parsed <= 0 {
return nil, false
}
if parsed > 1_000_000_000_000 {
parsed /= 1000
}
if parsed <= 0 || parsed > 4_102_444_800 {
return nil, false
}
when := time.Unix(parsed, 0).UTC()
return &when, true
}
return nil, true
}
func cursorValue(raw json.RawMessage) (string, bool) {
if len(raw) == 0 || string(raw) == "null" {
return "", true
}
var text string
if json.Unmarshal(raw, &text) == nil {
return strings.TrimSpace(text), validCursor(strings.TrimSpace(text))
}
var number json.Number
decoder := json.NewDecoder(bytes.NewReader(raw))
decoder.UseNumber()
if decoder.Decode(&number) != nil {
return "", false
}
return number.String(), validCursor(number.String())
}
func coverURL(object map[string]json.RawMessage) string {
images, ok := object["image_list"]
if ok {
var values []json.RawMessage
if json.Unmarshal(images, &values) == nil {
for _, value := range values {
if image, ok := objectValue(value); ok {
if cover := firstString(image, "url_default", "url_pre", "url_original", "url"); cover != "" {
return cover
}
}
}
}
}
if cover := firstObject(object, "cover"); cover != nil {
return firstString(cover, "url_default", "url_pre", "url_original", "url")
}
return firstString(object, "cover_url", "cover")
}
func noteURL(id string, access accessContext) string {
query := url.Values{}
if access.Token != "" {
query.Set("xsec_token", access.Token)
}
if access.Source != "" {
query.Set("xsec_source", access.Source)
}
result := "https://www.xiaohongshu.com/explore/" + url.PathEscape(id)
if encoded := query.Encode(); encoded != "" {
result += "?" + encoded
}
return result
}
internal/platform/xiaohongshu/collector_test.go
-126
View File
@@ -1,126 +0,0 @@
package xiaohongshu
import (
"context"
"errors"
"net/url"
"testing"
"git.ipao.vip/rogee/creator-hub/internal/creator"
)
type fakeBrowser struct {
getURL string
postURL string
getBody []byte
postBody []byte
getResp Response
postResp Response
resolvedURL string
resolveInput string
}
func (f *fakeBrowser) Get(_ context.Context, target string) (Response, error) {
f.getURL = target
return f.getResp, nil
}
func (f *fakeBrowser) Post(_ context.Context, target string, body []byte) (Response, error) {
f.postURL, f.postBody = target, body
return f.postResp, nil
}
func (f *fakeBrowser) Resolve(_ context.Context, target string) (string, error) {
f.resolveInput = target
return f.resolvedURL, nil
}
func TestIdentityRequiresMatchingUser(t *testing.T) {
browser := &fakeBrowser{getResp: Response{Status: 200, Body: []byte(`{"success":true,"data":{"user_id":"u-1","nickname":"作者"}}`)}}
collector := &Collector{Browser: browser}
identity, err := collector.Identity(context.Background(), "u-1")
if err != nil || identity.UserID != "u-1" {
t.Fatalf("identity = %#v, err = %v", identity, err)
}
if _, err := collector.Identity(context.Background(), "u-2"); !errors.Is(err, creator.ErrConflict) {
t.Fatalf("expected identity conflict, got %v", err)
}
}
func TestListWorksCapturesContextForComments(t *testing.T) {
browser := &fakeBrowser{getResp: Response{Status: 200, Body: []byte(`{"success":true,"data":{"cursor":"next","has_more":true,"notes":[{"note_id":"n-1","title":"标题","desc":"正文","time":1710000000,"user":{"user_id":"u-1","nickname":"作者"},"interact_info":{"liked_count":"2","comment_count":3,"shared_count":4},"xsec_token":"token"}]}}`)}}
collector := &Collector{Browser: browser, AccountKey: "u-1", SourceType: creator.SourceOwned, SourceID: "source-1"}
page, err := collector.ListWorks(context.Background(), "ignored", "")
if err != nil || len(page.Items) != 1 || page.NextCursor != "next" || !page.HasMore || page.Items[0].RawPayload == "" {
t.Fatalf("page = %#v, err = %v", page, err)
}
parsed, err := url.Parse(browser.getURL)
if err != nil || parsed.Query().Get("user_id") != "u-1" {
t.Fatalf("request URL = %s", browser.getURL)
}
browser.getResp = Response{Status: 200, Body: []byte(`{"success":true,"data":{"cursor":"","has_more":false,"comments":[{"id":"c-1","content":"评论","create_time":1710000000,"user_info":{"user_id":"u-2","nickname":"读者"}}]}}`)}
comments, err := collector.ListTopLevelComments(context.Background(), "n-1", "")
if err != nil || len(comments.Items) != 1 || comments.Items[0].CommentType != "top_level" || comments.Items[0].RawPayload == "" {
t.Fatalf("comments = %#v, err = %v", comments, err)
}
commentURL, err := url.Parse(browser.getURL)
if err != nil || commentURL.Query().Get("xsec_token") != "token" {
t.Fatalf("comment request URL = %s", browser.getURL)
}
}
func TestSearchUsesBoundedPostAndRejectsMalformedPage(t *testing.T) {
browser := &fakeBrowser{postResp: Response{Status: 200, Body: []byte(`{"success":true,"data":{"cursor":"","has_more":false,"notes":[]}}`)}}
collector := &Collector{Browser: browser}
page, err := collector.SearchNotes(context.Background(), "关键词", 1)
if err != nil || page.HasMore || len(page.Items) != 0 || browser.postURL != SearchOrigin+SearchNotesPath {
t.Fatalf("page = %#v, URL = %s, err = %v", page, browser.postURL, err)
}
browser.postResp = Response{Status: 200, Body: []byte(`{"success":true,"data":{"has_more":true,"notes":[]}}`)}
if _, err := collector.SearchNotes(context.Background(), "关键词", 1); !errors.Is(err, creator.ErrInvalid) {
t.Fatalf("expected malformed response error, got %v", err)
}
}
func TestHomepageContextIsUsedWithoutInventingDefaults(t *testing.T) {
browser := &fakeBrowser{getResp: Response{Status: 200, Body: []byte(`{"success":true,"data":{"cursor":"","has_more":false,"notes":[]}}`)}}
collector := &Collector{Browser: browser, AccountKey: "u-1", HomepageURL: "https://www.xiaohongshu.com/user/profile/u-1?xsec_token=home-token&xsec_source=pc_search"}
if _, err := collector.ListWorks(context.Background(), "ignored", ""); err != nil {
t.Fatalf("list works: %v", err)
}
requestURL, err := url.Parse(browser.getURL)
if err != nil || requestURL.Query().Get("xsec_token") != "home-token" || requestURL.Query().Get("xsec_source") != "pc_search" {
t.Fatalf("homepage context was not forwarded: %s", browser.getURL)
}
}
func TestSourceURLParsingAndShareResolution(t *testing.T) {
homepage, err := ParseSourceURL("https://www.xiaohongshu.com/user/profile/u-1?xsec_source=pc_search")
if err != nil || homepage.Kind != LinkHomepage || homepage.AccountKey != "u-1" {
t.Fatalf("homepage = %#v, err = %v", homepage, err)
}
share, err := ParseSourceURL("https://xhslink.com/a/abc")
if err != nil || share.Kind != LinkShare {
t.Fatalf("share = %#v, err = %v", share, err)
}
browser := &fakeBrowser{resolvedURL: "https://www.xiaohongshu.com/explore/n-1?xsec_token=tok-1&xsec_source=pc_search", postResp: Response{Status: 200, Body: []byte(`{"success":true,"data":{"items":[{"note_id":"n-1","title":"详情","time":1710000000,"interact_info":{}}]}}`)}}
collector := &Collector{Browser: browser}
item, err := collector.GetNoteDetail(context.Background(), "https://xhslink.com/a/abc")
if err != nil || item.WorkKey != "n-1" || browser.resolveInput != "https://xhslink.com/a/abc" {
t.Fatalf("resolved detail = %#v, err = %v, input = %s", item, err, browser.resolveInput)
}
}
func TestResponseErrorClassifiesRejectedPlatformPayload(t *testing.T) {
if err := responseError(Response{Status: 200, Body: []byte(`{"success":false,"code":"LOGIN_REQUIRED","msg":"请重新登录"}`)}, "works"); !errors.Is(err, creator.ErrConflict) {
t.Fatalf("expected conflict, got %v", err)
}
if err := responseError(Response{Status: 200, Body: []byte(`{"success":false,"code":"RATE_LIMIT","msg":"请求过于频繁"}`)}, "works"); !errors.Is(err, creator.ErrUnavailable) {
t.Fatalf("expected unavailable, got %v", err)
}
}
func TestSetWorkContextRejectsWrongHost(t *testing.T) {
collector := &Collector{}
if err := collector.SetWorkContext("n-1", "https://evil.example/explore/n-1"); !errors.Is(err, creator.ErrInvalid) {
t.Fatalf("expected invalid URL, got %v", err)
}
}
internal/platform/xiaohongshu/detail.go
-92
View File
@@ -1,92 +0,0 @@
package xiaohongshu
import (
"context"
"encoding/json"
"fmt"
"git.ipao.vip/rogee/creator-hub/internal/creator"
)
// GetNoteDetail reads one note through the platform feed endpoint and keeps
// the original link so xsec context survives a later comments resume.
func (c *Collector) GetNoteDetail(ctx context.Context, originalURL string) (creator.WorkInput, error) {
if c == nil || c.Browser == nil {
return creator.WorkInput{}, fmt.Errorf("%w: invalid xiaohongshu detail request", creator.ErrInvalid)
}
link, err := c.resolveSourceURL(ctx, originalURL)
if err != nil {
return creator.WorkInput{}, err
}
if link.Kind != LinkNote {
return creator.WorkInput{}, fmt.Errorf("%w: xiaohongshu note URL is required", creator.ErrInvalid)
}
access := accessContext{Token: link.Token, Source: link.Source}
body, err := json.Marshal(map[string]any{
"source_note_id": link.WorkKey,
"image_formats": []string{"jpg", "webp", "avif"},
"extra": map[string]string{"need_body_topic": "1"},
"xsec_source": access.Source,
"xsec_token": access.Token,
})
if err != nil {
return creator.WorkInput{}, fmt.Errorf("encode xiaohongshu detail request: %w", err)
}
response, err := c.Browser.Post(ctx, APIOrigin+FeedPath, body)
if err != nil {
return creator.WorkInput{}, err
}
if err := responseError(response, "detail"); err != nil {
return creator.WorkInput{}, err
}
item, parsedAccess, ok := parseDetail(response.Body, access)
if !ok {
return creator.WorkInput{}, fmt.Errorf("%w: invalid xiaohongshu detail response", creator.ErrInvalid)
}
if item.WorkKey != link.WorkKey {
return creator.WorkInput{}, fmt.Errorf("%w: xiaohongshu detail returned another note", creator.ErrConflict)
}
item.OriginalURL = link.CanonicalURL
item.SourceType = c.SourceType
item.SourceID = c.SourceID
if item.SourceType == "" {
item.SourceType = creator.SourceCompetitor
}
if item.SourceID == "" {
item.SourceID = c.AccountKey
}
c.contexts.Store(item.WorkKey, parsedAccess)
return item, nil
}
func parseDetail(body []byte, fallback accessContext) (creator.WorkInput, accessContext, bool) {
if len(body) == 0 || len(body) > 4<<20 {
return creator.WorkInput{}, accessContext{}, false
}
var envelope struct {
Success *bool `json:"success"`
Data json.RawMessage `json:"data"`
}
if json.Unmarshal(body, &envelope) != nil || envelope.Success == nil || !*envelope.Success || len(envelope.Data) == 0 {
return creator.WorkInput{}, accessContext{}, false
}
data, ok := objectValue(envelope.Data)
if !ok {
return creator.WorkInput{}, accessContext{}, false
}
if items, ok := arrayValue(data["items"]); ok && len(items) > 0 {
return parseWork(items[0], fallback)
}
if note := data["note"]; len(note) > 0 {
return parseWork(note, fallback)
}
return parseWork(envelope.Data, fallback)
}
func arrayValue(raw json.RawMessage) ([]json.RawMessage, bool) {
var values []json.RawMessage
if len(raw) == 0 || json.Unmarshal(raw, &values) != nil || values == nil {
return nil, false
}
return values, true
}
internal/platform/xiaohongshu/detail_test.go
-33
View File
@@ -1,33 +0,0 @@
package xiaohongshu
import (
"context"
"errors"
"testing"
"git.ipao.vip/rogee/creator-hub/internal/creator"
)
func TestGetNoteDetailPreservesShareContext(t *testing.T) {
browser := &fakeBrowser{postResp: Response{Status: 200, Body: []byte(`{"success":true,"data":{"items":[{"note_id":"n-1","title":"详情","desc":"正文","time":1710000000,"user":{"user_id":"u-1","nickname":"作者"},"interact_info":{"liked_count":1,"comment_count":2,"shared_count":3}}]}}`)}}
collector := &Collector{Browser: browser, SourceType: creator.SourceCompetitor, SourceID: "source-1"}
item, err := collector.GetNoteDetail(context.Background(), "https://www.xiaohongshu.com/explore/n-1?xsec_token=tok-1&xsec_source=pc_search")
if err != nil || item.WorkKey != "n-1" || item.AuthorKey != "u-1" || item.OriginalURL == "" {
t.Fatalf("item = %#v, err = %v", item, err)
}
if err := collector.SetWorkContext("n-1", item.OriginalURL); err != nil {
t.Fatalf("set work context: %v", err)
}
if browser.postURL != APIOrigin+FeedPath {
t.Fatalf("post URL = %s", browser.postURL)
}
}
func TestGetNoteDetailRejectsWrongResponseID(t *testing.T) {
browser := &fakeBrowser{postResp: Response{Status: 200, Body: []byte(`{"success":true,"data":{"items":[{"note_id":"n-2","title":"详情","time":1710000000,"interact_info":{}}]}}`)}}
collector := &Collector{Browser: browser}
_, err := collector.GetNoteDetail(context.Background(), "https://www.xiaohongshu.com/explore/n-1")
if !errors.Is(err, creator.ErrConflict) {
t.Fatalf("expected conflict, got %v", err)
}
}