feat(creator): 账号画像补互关指标并让监控列表携带最新快照与评论聚合
- migration 1044:creator_account_metric 加 friend_count(self profile 接口 user.friend_count,即互关数) - douyin connector 解析 friend_count(负值校验同其他计数),快照落库透传 - monitor-views 视图新增:每账号最新非空画像计数(粉丝/关注/获赞/作品/互关,按采集时间回填缺失维度) - 评论口径:自有作品最近指标 comments_count 求和(comment_total),无账号级评论采集源 - 详情页指标时序 AccountMetricPoint 带回 friend_count
This commit is contained in:
@@ -1663,7 +1663,7 @@ func recordCreatorAccountMetricSnapshot(ctx context.Context, store *creator.Stor
|
||||
return store.RecordAccountMetric(ctx, creator.AccountMetricInput{
|
||||
AccountID: accountID, CollectedAt: time.Now().UTC(),
|
||||
FollowerCount: profile.FollowerCount, FollowingCount: profile.FollowingCount,
|
||||
TotalFavorited: profile.TotalFavorited, AwemeCount: profile.AwemeCount,
|
||||
TotalFavorited: profile.TotalFavorited, AwemeCount: profile.AwemeCount, FriendCount: profile.FriendCount,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -105,11 +105,17 @@ func (s *Store) ListAccountProfiles(ctx context.Context) ([]AccountProfile, erro
|
||||
return profiles, rows.Err()
|
||||
}
|
||||
|
||||
// AccountMonitorView 自有账号监控列表视图:账号画像 + 作品聚合统计(作品数/最近发布)。
|
||||
// AccountMonitorView 自有账号监控列表视图:账号画像 + 作品聚合统计(作品数/最近发布)+ 最新画像指标快照 + 自有作品评论累计。
|
||||
type AccountMonitorView struct {
|
||||
AccountProfile
|
||||
WorkCount int64 `json:"work_count"`
|
||||
LatestPublishedAt *time.Time `json:"latest_published_at,omitempty"`
|
||||
FollowerCount *int64 `json:"follower_count,omitempty"`
|
||||
FollowingCount *int64 `json:"following_count,omitempty"`
|
||||
TotalFavorited *int64 `json:"total_favorited,omitempty"`
|
||||
AwemeCount *int64 `json:"aweme_count,omitempty"`
|
||||
FriendCount *int64 `json:"friend_count,omitempty"`
|
||||
CommentTotal int64 `json:"comment_total"`
|
||||
}
|
||||
|
||||
// ListAccountMonitorViews 自有账号监控列表:画像 + owned 作品聚合,形态对齐竞品的 ListCompetitorsWithProfile。
|
||||
@@ -122,11 +128,19 @@ func (s *Store) ListAccountMonitorViews(ctx context.Context) ([]AccountMonitorVi
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
latest, err := s.latestAccountMetrics(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
views := make([]AccountMonitorView, 0, len(profiles))
|
||||
for _, profile := range profiles {
|
||||
view := AccountMonitorView{AccountProfile: profile}
|
||||
if stat, exists := stats[profile.ID]; exists {
|
||||
view.WorkCount, view.LatestPublishedAt = stat.WorkCount, stat.LatestPublishedAt
|
||||
view.WorkCount, view.LatestPublishedAt, view.CommentTotal = stat.WorkCount, stat.LatestPublishedAt, stat.CommentTotal
|
||||
}
|
||||
if metric, exists := latest[profile.ID]; exists {
|
||||
view.FollowerCount, view.FollowingCount, view.TotalFavorited, view.AwemeCount, view.FriendCount =
|
||||
metric.FollowerCount, metric.FollowingCount, metric.TotalFavorited, metric.AwemeCount, metric.FriendCount
|
||||
}
|
||||
views = append(views, view)
|
||||
}
|
||||
@@ -136,12 +150,67 @@ func (s *Store) ListAccountMonitorViews(ctx context.Context) ([]AccountMonitorVi
|
||||
type ownedWorkStat struct {
|
||||
WorkCount int64
|
||||
LatestPublishedAt *time.Time
|
||||
CommentTotal int64
|
||||
}
|
||||
|
||||
// ownedWorkStats 按 source_id 聚合 owned 作品数与最近发布时间。
|
||||
// latestAccountMetricRow 最新快照内一行非空计数(无对应快照时不返回行)。
|
||||
type latestAccountMetricRow struct {
|
||||
FollowerCount *int64
|
||||
FollowingCount *int64
|
||||
TotalFavorited *int64
|
||||
AwemeCount *int64
|
||||
FriendCount *int64
|
||||
}
|
||||
|
||||
// latestAccountMetrics 每账号取最新非空画像计数:按采集时间倒序回填,缺失维度保留上一轮的值(详情页同口径)。
|
||||
func (s *Store) latestAccountMetrics(ctx context.Context) (map[string]latestAccountMetricRow, error) {
|
||||
rows, err := s.db.QueryContext(ctx, `
|
||||
SELECT account.account_id, metric.follower_count, metric.following_count, metric.total_favorited, metric.aweme_count, metric.friend_count
|
||||
FROM creator_account_metric metric
|
||||
JOIN social_account account ON account.id = metric.account_id
|
||||
ORDER BY account.account_id, metric.collected_at`)
|
||||
if err != nil {
|
||||
return nil, databaseError(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
latest := make(map[string]latestAccountMetricRow)
|
||||
for rows.Next() {
|
||||
var accountID string
|
||||
var follower, following, favorited, aweme, friend sql.NullInt64
|
||||
if err := rows.Scan(&accountID, &follower, &following, &favorited, &aweme, &friend); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
current := latest[accountID]
|
||||
if follower.Valid {
|
||||
value := follower.Int64
|
||||
current.FollowerCount = &value
|
||||
}
|
||||
if following.Valid {
|
||||
value := following.Int64
|
||||
current.FollowingCount = &value
|
||||
}
|
||||
if favorited.Valid {
|
||||
value := favorited.Int64
|
||||
current.TotalFavorited = &value
|
||||
}
|
||||
if aweme.Valid {
|
||||
value := aweme.Int64
|
||||
current.AwemeCount = &value
|
||||
}
|
||||
if friend.Valid {
|
||||
value := friend.Int64
|
||||
current.FriendCount = &value
|
||||
}
|
||||
latest[accountID] = current
|
||||
}
|
||||
return latest, rows.Err()
|
||||
}
|
||||
|
||||
// AccountCollectionStatus 自有账号采集状态(checkpoint 形态,对齐竞品的 sync_status 展示语义)。
|
||||
func (s *Store) ownedWorkStats(ctx context.Context) (map[string]ownedWorkStat, error) {
|
||||
rows, err := s.db.QueryContext(ctx, `
|
||||
SELECT w.source_id, COUNT(*) AS work_count, MAX(w.published_at) AS latest_published_at
|
||||
SELECT w.source_id, COUNT(*) AS work_count, MAX(w.published_at) AS latest_published_at,
|
||||
COALESCE(SUM(w.comments_count), 0) AS comment_total
|
||||
FROM creator_work w
|
||||
WHERE w.source_type = 'owned'
|
||||
GROUP BY w.source_id`)
|
||||
@@ -154,7 +223,7 @@ func (s *Store) ownedWorkStats(ctx context.Context) (map[string]ownedWorkStat, e
|
||||
var stat ownedWorkStat
|
||||
var sourceID string
|
||||
var latest sql.NullTime
|
||||
if err := rows.Scan(&sourceID, &stat.WorkCount, &latest); err != nil {
|
||||
if err := rows.Scan(&sourceID, &stat.WorkCount, &latest, &stat.CommentTotal); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stat.LatestPublishedAt = nullableTime(latest)
|
||||
|
||||
@@ -623,16 +623,18 @@ func nullableArg(value time.Time) any {
|
||||
// RecordAccountMetric 落库账号画像快照。同一 (account_id, collected_at) 重复采集覆盖更新。
|
||||
func (s *Store) RecordAccountMetric(ctx context.Context, input AccountMetricInput) error {
|
||||
if input.AccountID == "" || input.CollectedAt.IsZero() || input.FollowerCount != nil && *input.FollowerCount < 0 ||
|
||||
input.FollowingCount != nil && *input.FollowingCount < 0 || input.TotalFavorited != nil && *input.TotalFavorited < 0 || input.AwemeCount != nil && *input.AwemeCount < 0 {
|
||||
input.FollowingCount != nil && *input.FollowingCount < 0 || input.TotalFavorited != nil && *input.TotalFavorited < 0 ||
|
||||
input.AwemeCount != nil && *input.AwemeCount < 0 || input.FriendCount != nil && *input.FriendCount < 0 {
|
||||
return ErrInvalid
|
||||
}
|
||||
result, err := s.db.ExecContext(ctx, `
|
||||
INSERT INTO creator_account_metric (account_id, collected_at, follower_count, following_count, total_favorited, aweme_count)
|
||||
SELECT account.id, $2, $3, $4, $5, $6 FROM social_account account WHERE account.account_id = $1
|
||||
INSERT INTO creator_account_metric (account_id, collected_at, follower_count, following_count, total_favorited, aweme_count, friend_count)
|
||||
SELECT account.id, $2, $3, $4, $5, $6, $7 FROM social_account account WHERE account.account_id = $1
|
||||
ON CONFLICT (account_id, collected_at) DO UPDATE SET
|
||||
follower_count = EXCLUDED.follower_count, following_count = EXCLUDED.following_count,
|
||||
total_favorited = EXCLUDED.total_favorited, aweme_count = EXCLUDED.aweme_count`,
|
||||
input.AccountID, input.CollectedAt.UTC(), input.FollowerCount, input.FollowingCount, input.TotalFavorited, input.AwemeCount)
|
||||
total_favorited = EXCLUDED.total_favorited, aweme_count = EXCLUDED.aweme_count,
|
||||
friend_count = EXCLUDED.friend_count`,
|
||||
input.AccountID, input.CollectedAt.UTC(), input.FollowerCount, input.FollowingCount, input.TotalFavorited, input.AwemeCount, input.FriendCount)
|
||||
if err != nil {
|
||||
return databaseError(err)
|
||||
}
|
||||
@@ -651,7 +653,7 @@ func (s *Store) ListAccountMetrics(ctx context.Context, accountID string) ([]Acc
|
||||
return nil, ErrInvalid
|
||||
}
|
||||
rows, err := s.db.QueryContext(ctx, `
|
||||
SELECT collected_at, follower_count, following_count, total_favorited, aweme_count
|
||||
SELECT collected_at, follower_count, following_count, total_favorited, aweme_count, friend_count
|
||||
FROM creator_account_metric metric JOIN social_account account ON account.id = metric.account_id WHERE account.account_id = $1 ORDER BY metric.collected_at`, accountID)
|
||||
if err != nil {
|
||||
return nil, databaseError(err)
|
||||
@@ -660,13 +662,14 @@ func (s *Store) ListAccountMetrics(ctx context.Context, accountID string) ([]Acc
|
||||
result := make([]AccountMetricPoint, 0)
|
||||
for rows.Next() {
|
||||
var point AccountMetricPoint
|
||||
var follower, following, favorited, aweme sql.NullInt64
|
||||
if err := rows.Scan(&point.CollectedAt, &follower, &following, &favorited, &aweme); err != nil {
|
||||
var follower, following, favorited, aweme, friend sql.NullInt64
|
||||
if err := rows.Scan(&point.CollectedAt, &follower, &following, &favorited, &aweme, &friend); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
point.CollectedAt = point.CollectedAt.UTC()
|
||||
point.FollowerCount, point.FollowingCount = nullableInt64(follower), nullableInt64(following)
|
||||
point.TotalFavorited, point.AwemeCount = nullableInt64(favorited), nullableInt64(aweme)
|
||||
point.FriendCount = nullableInt64(friend)
|
||||
result = append(result, point)
|
||||
}
|
||||
return result, rows.Err()
|
||||
|
||||
@@ -552,6 +552,13 @@ func TestCreatorPostgresAccountMonitorViewAndCollectionStatus(t *testing.T) {
|
||||
if view == nil || view.WorkCount != 0 || view.LatestPublishedAt != nil {
|
||||
t.Fatalf("monitor view without works: %+v", view)
|
||||
}
|
||||
// 无指标快照:画像指标为空(不伪造成 0)。
|
||||
if view.FollowerCount != nil || view.FollowingCount != nil || view.TotalFavorited != nil || view.AwemeCount != nil {
|
||||
t.Fatalf("monitor view without metrics must leave profile counters nil: %+v", view)
|
||||
}
|
||||
if view.CommentTotal != 0 {
|
||||
t.Fatalf("monitor view without works must have zero comments: %+v", view)
|
||||
}
|
||||
|
||||
// 写入 checkpoint(succeeded + 窗口)与 owned 作品后:状态带下次窗口,视图聚合作品数。
|
||||
lease, err := store.beginCheckpoint(ctx, SourceOwned, accountID, "works", now.Add(-24*time.Hour), now)
|
||||
@@ -585,6 +592,43 @@ func TestCreatorPostgresAccountMonitorViewAndCollectionStatus(t *testing.T) {
|
||||
t.Fatalf("monitor view aggregates: %+v", view)
|
||||
}
|
||||
}
|
||||
|
||||
// 画像指标快照:写入两轮后视图取最新一轮;评论聚合来自 owned 作品最近指标之和。
|
||||
first, second := int64(100), int64(200)
|
||||
following := int64(30)
|
||||
favorited := int64(5000)
|
||||
aweme := int64(7)
|
||||
mutual := int64(9)
|
||||
if err := store.RecordAccountMetric(ctx, AccountMetricInput{AccountID: accountID, CollectedAt: now.Add(-2 * time.Hour), FollowerCount: &first, FollowingCount: &following, TotalFavorited: &favorited, AwemeCount: &aweme, FriendCount: &mutual}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.RecordAccountMetric(ctx, AccountMetricInput{AccountID: accountID, CollectedAt: now.Add(-time.Hour), FollowerCount: &second}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
monitorWork, _, err := store.UpsertWork(ctx, WorkInput{Platform: PlatformDouyin, WorkKey: "monitor-view-" + stamp, SourceType: SourceOwned, SourceID: accountID, Title: "MonitorView", PublishedAt: &published, PublishedAtStatus: "verified"}, now)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := store.RecordMetric(ctx, MetricInput{WorkID: monitorWork.ID, CollectedAt: now, CommentsCount: &favorited}, Settings{LookbackDays: 30, NewWorkIntervalSeconds: 1800, MetricInitialIntervalSeconds: 3600, MetricMultiplier: 2, MetricMaxIntervalSeconds: 86400, MetricAgeSeconds: 30 * 24 * 60 * 60}, now); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
views, err = store.ListAccountMonitorViews(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
view = viewFor(accountID)
|
||||
if view == nil || view.FollowerCount == nil || *view.FollowerCount != second {
|
||||
t.Fatalf("monitor view must carry the latest metric snapshot: %+v", view)
|
||||
}
|
||||
if view.FollowingCount == nil || *view.FollowingCount != following || view.TotalFavorited == nil || *view.TotalFavorited != favorited || view.AwemeCount == nil || *view.AwemeCount != aweme {
|
||||
t.Fatalf("monitor view must fall back to the latest non-null counters: %+v", view)
|
||||
}
|
||||
if view.FriendCount == nil || *view.FriendCount != mutual {
|
||||
t.Fatalf("monitor view must carry the latest mutual friend count: %+v", view)
|
||||
}
|
||||
if view.CommentTotal != favorited {
|
||||
t.Fatalf("monitor view must aggregate owned work comments: got %d want %d", view.CommentTotal, favorited)
|
||||
}
|
||||
// failed checkpoint 带错误信息(普通错误落在 failed,不是 blocked)。
|
||||
lease, err = store.beginCheckpoint(ctx, SourceOwned, accountID, "comments", now.Add(-24*time.Hour), now)
|
||||
if err != nil {
|
||||
|
||||
@@ -259,6 +259,7 @@ type AccountMetricInput struct {
|
||||
FollowingCount *int64
|
||||
TotalFavorited *int64
|
||||
AwemeCount *int64
|
||||
FriendCount *int64
|
||||
}
|
||||
|
||||
type AccountMetricPoint struct {
|
||||
@@ -267,6 +268,7 @@ type AccountMetricPoint struct {
|
||||
FollowingCount *int64 `json:"following_count"`
|
||||
TotalFavorited *int64 `json:"total_favorited"`
|
||||
AwemeCount *int64 `json:"aweme_count"`
|
||||
FriendCount *int64 `json:"friend_count"`
|
||||
}
|
||||
|
||||
type MaterialJob struct {
|
||||
|
||||
@@ -70,6 +70,6 @@ func TestMigration043ConsolidatedSchemaShape(t *testing.T) {
|
||||
// audit_event 任务列已删
|
||||
assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.columns WHERE table_schema = current_schema()
|
||||
AND table_name = 'audit_event' AND column_name IN ('confirmation_id','confirmation_version','attempt_id','task_id','runtime_instance_id')`, 0)
|
||||
// 统一登记表:1-38(除 36)、1017-1043、43 全部登记
|
||||
assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration`, 50)
|
||||
// 统一登记表:1-38(除 36)、1017-1044、43 全部登记
|
||||
assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration`, 51)
|
||||
}
|
||||
|
||||
@@ -30,7 +30,7 @@ func TestUnifiedAccountMigration(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration`, 50)
|
||||
assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration`, 51)
|
||||
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')`, 3)
|
||||
assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.tables WHERE table_schema = current_schema() AND table_name = 'browser_image'`, 0)
|
||||
@@ -42,7 +42,7 @@ func TestUnifiedAccountMigration(t *testing.T) {
|
||||
|
||||
store = openFullyMigratedHub(t, ctx, testURL)
|
||||
store.Close()
|
||||
assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration`, 50)
|
||||
assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration`, 51)
|
||||
})
|
||||
|
||||
t.Run("legacy migration 013 without account secrets is repaired forward", func(t *testing.T) {
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
-- 自有账号画像快照补互关数:self profile 接口 user.friend_count(实测字段存在),与粉丝/关注同节奏落库。
|
||||
ALTER TABLE creator_account_metric ADD COLUMN IF NOT EXISTS friend_count bigint;
|
||||
@@ -170,6 +170,9 @@ var migration1042 string
|
||||
//go:embed migrations/1043_gateway_health.sql
|
||||
var migration1043 string
|
||||
|
||||
//go:embed migrations/1044_creator_account_metric_friend_count.sql
|
||||
var migration1044 string
|
||||
|
||||
var (
|
||||
ErrConflict = errors.New("resource conflicts with existing state")
|
||||
ErrInvalid = errors.New("invalid environment input")
|
||||
@@ -309,7 +312,7 @@ func (s *Store) migrate(ctx context.Context) error {
|
||||
{1023, migration1023}, {1024, migration1024}, {1025, migration1025}, {1026, migration1026}, {1027, migration1027}, {1028, migration1028},
|
||||
{1029, migration1029}, {1030, migration1030}, {1031, migration1031}, {1032, migration1032}, {1033, migration1033}, {1034, migration1034},
|
||||
{1035, migration1035}, {1036, migration1036}, {1037, migration1037}, {1038, migration1038}, {1039, migration1039}, {1040, migration1040},
|
||||
{1041, migration1041}, {1042, migration1042}, {1043, migration1043},
|
||||
{1041, migration1041}, {1042, migration1042}, {1043, migration1043}, {1044, migration1044},
|
||||
{43, migration043}, {44, migration044}} {
|
||||
var applied bool
|
||||
if err := tx.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM schema_migration WHERE version = $1)`, migration.version).Scan(&applied); err != nil {
|
||||
|
||||
@@ -319,6 +319,7 @@ type identityEnvelope struct {
|
||||
FollowerCount *int64 `json:"follower_count"`
|
||||
FollowingCount *int64 `json:"following_count"`
|
||||
AwemeCount *int64 `json:"aweme_count"`
|
||||
FriendCount *int64 `json:"friend_count"`
|
||||
TotalFavorited *flexibleInt `json:"total_favorited"`
|
||||
} `json:"user"`
|
||||
}
|
||||
@@ -356,7 +357,7 @@ func parseIdentity(body []byte) (identityEnvelope, bool) {
|
||||
(identity.User.UniqueID != "" && !keyPattern.MatchString(identity.User.UniqueID)) {
|
||||
return identityEnvelope{}, false
|
||||
}
|
||||
for _, value := range []*int64{identity.User.FollowerCount, identity.User.FollowingCount, identity.User.AwemeCount} {
|
||||
for _, value := range []*int64{identity.User.FollowerCount, identity.User.FollowingCount, identity.User.AwemeCount, identity.User.FriendCount} {
|
||||
if value != nil && *value < 0 {
|
||||
return identityEnvelope{}, false
|
||||
}
|
||||
@@ -375,6 +376,7 @@ type SelfProfile struct {
|
||||
FollowingCount *int64
|
||||
TotalFavorited *int64
|
||||
AwemeCount *int64
|
||||
FriendCount *int64
|
||||
}
|
||||
|
||||
// FetchSelfProfile 经网关受限 fetch 拉取登录账号自己的画像(复用 self profile 身份接口,返回体自带粉丝/关注/获赞/作品数)。
|
||||
@@ -413,6 +415,7 @@ func parseSelfProfile(body []byte) (SelfProfile, bool) {
|
||||
return SelfProfile{
|
||||
UID: identity.User.UID, SecUID: identity.User.SecUID, UniqueID: identity.User.UniqueID, Nickname: identity.User.Nickname, AvatarURL: avatarURL,
|
||||
FollowerCount: identity.User.FollowerCount, FollowingCount: identity.User.FollowingCount, TotalFavorited: totalFavorited, AwemeCount: identity.User.AwemeCount,
|
||||
FriendCount: identity.User.FriendCount,
|
||||
}, true
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user