From cbbec8da856f627bdcc5d21e975341eca76309a5 Mon Sep 17 00:00:00 2001 From: liangxu Date: Thu, 25 Jun 2026 09:31:18 +0800 Subject: [PATCH] fix monitoring manual mark attribution --- .../src/views/TrackingQuestionDetailView.vue | 64 +++-- .../tenant/app/monitoring_callback_service.go | 15 +- .../app/monitoring_callback_service_test.go | 24 +- .../app/monitoring_marked_article_service.go | 249 ++++++++++++++++-- .../tenant/app/monitoring_projection.go | 1 - .../internal/tenant/app/monitoring_service.go | 13 +- ...0_speed_manual_mark_citation_sync.down.sql | 2 + ...000_speed_manual_mark_citation_sync.up.sql | 21 ++ ...nual_mark_citation_slash_variants.down.sql | 53 ++++ ...manual_mark_citation_slash_variants.up.sql | 60 +++++ 10 files changed, 436 insertions(+), 66 deletions(-) create mode 100644 server/migrations_monitoring/20260625102000_speed_manual_mark_citation_sync.down.sql create mode 100644 server/migrations_monitoring/20260625102000_speed_manual_mark_citation_sync.up.sql create mode 100644 server/migrations_monitoring/20260625103500_backfill_manual_mark_citation_slash_variants.down.sql create mode 100644 server/migrations_monitoring/20260625103500_backfill_manual_mark_citation_slash_variants.up.sql diff --git a/apps/admin-web/src/views/TrackingQuestionDetailView.vue b/apps/admin-web/src/views/TrackingQuestionDetailView.vue index 9ad14aa..d780adc 100644 --- a/apps/admin-web/src/views/TrackingQuestionDetailView.vue +++ b/apps/admin-web/src/views/TrackingQuestionDetailView.vue @@ -2,6 +2,7 @@ import { ArrowLeftOutlined, BookOutlined, CopyOutlined } from '@ant-design/icons-vue' import type { MonitoringQuestionCitationAnalysisItem, + MonitoringQuestionContentCitation, MonitoringQuestionDetailCitation, MonitoringQuestionDetailPlatform, MonitoringSourceItem, @@ -198,7 +199,10 @@ const activeDeepSeekCitationReferenceLabels = computed(() => { }) const activeInlineContentCitations = computed(() => { - return analyzeInlineContentCitations(activePlatform.value) + return analyzeContentCitations( + activePlatform.value, + detailQuery.data.value?.content_citations ?? [], + ) }) const shouldShowInlineContentCitations = computed(() => { @@ -266,7 +270,7 @@ function openImitationCreate( } function markCitationAsExternal(citation: MonitoringQuestionDetailCitation): void { - if (!canMarkCitationAsExternal(citation, activePlatform.value?.ai_platform_id)) { + if (!canMarkCitationAsExternal(citation)) { return } markCitationMutation.mutate(citation) @@ -351,9 +355,8 @@ function isManualMarkedCitation( function canMarkCitationAsExternal( citation: MonitoringQuestionDetailCitation | null | undefined, - platformId: string | null | undefined, ): boolean { - if (!citation || !supportsContentCitationDisplay(platformId)) { + if (!citation) { return false } if (!String(citation.cited_url ?? '').trim()) { @@ -367,7 +370,7 @@ function canMarkCitationAsExternal( } function isLinkedInlineCitationPlatform(platformId: string | null | undefined): boolean { - return platformId === 'yuanbao' + return platformId === 'yuanbao' || platformId === 'qwen' } function isKimiInlineCitationPlatform(platformId: string | null | undefined): boolean { @@ -378,10 +381,6 @@ function isDeepSeekInlineCitationPlatform(platformId: string | null | undefined) return platformId === 'deepseek' } -function supportsContentCitationDisplay(platformId: string | null | undefined): boolean { - return platformId !== 'qwen' -} - function createQwenSourceGroupPattern(): RegExp { return /\[\[source_group_web_(\d+)\]\]/g } @@ -1140,6 +1139,36 @@ function collectDeepSeekInlineCitationOccurrences( })) } +function analyzeContentCitations( + platform: MonitoringQuestionDetailPlatform | null, + contentCitations: MonitoringQuestionContentCitation[], +): TrackingAnalyzedContentCitation[] { + const platformId = normalizeMonitoringPlatformId(platform?.ai_platform_id) + const serverItems = contentCitations + .filter((item) => normalizeMonitoringPlatformId(item.ai_platform_id) === platformId) + .sort((left, right) => { + if (right.citation_count !== left.citation_count) { + return right.citation_count - left.citation_count + } + return left.article_title.localeCompare(right.article_title, 'zh-CN') + }) + .map((item) => ({ + key: `server-${item.ai_platform_id}-${item.source_type}-${item.source_id}`, + contentName: item.article_title || item.source_url || '--', + channelName: item.publish_platform || deriveCitationChannelName(item.source_url), + citationCount: item.citation_count, + citationRate: item.citation_rate, + citedURL: item.source_url || null, + sourceType: item.source_type, + })) + + if (serverItems.length > 0) { + return serverItems + } + + return analyzeInlineContentCitations(platform) +} + function analyzeInlineContentCitations( platform: MonitoringQuestionDetailPlatform | null, ): TrackingAnalyzedContentCitation[] { @@ -1563,7 +1592,7 @@ const contentCitationColumns = computed(() => [
{{ t('tracking.contentCitationBadge') }} @@ -1654,10 +1677,7 @@ const contentCitationColumns = computed(() => [ {{ record.site_name }} {{ diff --git a/server/internal/tenant/app/monitoring_callback_service.go b/server/internal/tenant/app/monitoring_callback_service.go index 95cab00..5ae60dc 100644 --- a/server/internal/tenant/app/monitoring_callback_service.go +++ b/server/internal/tenant/app/monitoring_callback_service.go @@ -3213,11 +3213,11 @@ func (s *MonitoringCallbackService) loadExistingRunSummary(ctx context.Context, if err := s.monitoringPool.QueryRow(ctx, ` SELECT COUNT(*)::int AS citation_source_count, - COUNT(*) FILTER (WHERE content_source_type IS NOT NULL AND content_source_id IS NOT NULL AND $3::text <> 'qwen')::int AS content_citation_count + COUNT(*) FILTER (WHERE content_source_type IS NOT NULL AND content_source_id IS NOT NULL)::int AS content_citation_count FROM monitoring_citation_facts - WHERE tenant_id = $1 - AND run_id = $2 - `, task.TenantID, runID.Int64, normalizeMonitoringPlatformID(task.AIPlatformID)).Scan(&citationSourceCount, &contentCitationCount); err != nil { + WHERE tenant_id = $1 + AND run_id = $2 + `, task.TenantID, runID.Int64).Scan(&citationSourceCount, &contentCitationCount); err != nil { return nil, 0, 0, response.ErrInternal(50041, "citation_lookup_failed", "failed to inspect existing citation facts") } @@ -3340,12 +3340,7 @@ func monitoringPlatformUsesDedicatedCitationPanel(platformID string) bool { } func monitoringPlatformResolvesContentCitations(platformID string) bool { - switch normalizeMonitoringPlatformID(platformID) { - case "qwen": - return false - default: - return true - } + return normalizeMonitoringPlatformID(platformID) != "" } func newMonitoringTaskResultEnvelope(task *monitoringCollectTask, installationID string, receivedAt time.Time, req MonitoringTaskResultRequest) monitoringTaskResultEnvelope { diff --git a/server/internal/tenant/app/monitoring_callback_service_test.go b/server/internal/tenant/app/monitoring_callback_service_test.go index 733dcdc..302f813 100644 --- a/server/internal/tenant/app/monitoring_callback_service_test.go +++ b/server/internal/tenant/app/monitoring_callback_service_test.go @@ -535,7 +535,17 @@ func TestBuildMonitoringRawPayloadQwenIncludesSearchResultsInCitationInputs(t *t } } -func TestBuildMonitoringCitationFactCanDisableContentResolution(t *testing.T) { +func TestMonitoringPlatformResolvesContentCitationsForAllKnownModels(t *testing.T) { + for _, platformID := range []string{"doubao", "qwen", "kimi", "deepseek", "wenxin", "yuanbao"} { + t.Run(platformID, func(t *testing.T) { + if !monitoringPlatformResolvesContentCitations(platformID) { + t.Fatalf("monitoringPlatformResolvesContentCitations(%q) = false, want true", platformID) + } + }) + } +} + +func TestBuildMonitoringCitationFactResolvesContentForQwen(t *testing.T) { aliasArticleID := int64(42) aliasPublishRecordID := int64(84) normalizedURL := "https://example.com/article" @@ -561,15 +571,15 @@ func TestBuildMonitoringCitationFactCanDisableContentResolution(t *testing.T) { t.Fatalf("resolved PublishRecordID = %v, want %d", resolvedFact.PublishRecordID, aliasPublishRecordID) } - sourceOnlyFact, ok := buildMonitoringCitationFact(input, aliasMap, false) + sourceOnlyFact, ok := buildMonitoringCitationFact(input, aliasMap, monitoringPlatformResolvesContentCitations("qwen")) if !ok { - t.Fatal("buildMonitoringCitationFact() without content resolution returned !ok") + t.Fatal("buildMonitoringCitationFact() for qwen returned !ok") } - if sourceOnlyFact.ArticleID != nil { - t.Fatalf("source-only ArticleID = %v, want nil", sourceOnlyFact.ArticleID) + if sourceOnlyFact.ArticleID == nil || *sourceOnlyFact.ArticleID != aliasArticleID { + t.Fatalf("qwen ArticleID = %v, want %d", sourceOnlyFact.ArticleID, aliasArticleID) } - if sourceOnlyFact.PublishRecordID != nil { - t.Fatalf("source-only PublishRecordID = %v, want nil", sourceOnlyFact.PublishRecordID) + if sourceOnlyFact.PublishRecordID == nil || *sourceOnlyFact.PublishRecordID != aliasPublishRecordID { + t.Fatalf("qwen PublishRecordID = %v, want %d", sourceOnlyFact.PublishRecordID, aliasPublishRecordID) } } diff --git a/server/internal/tenant/app/monitoring_marked_article_service.go b/server/internal/tenant/app/monitoring_marked_article_service.go index 079a88d..795ed74 100644 --- a/server/internal/tenant/app/monitoring_marked_article_service.go +++ b/server/internal/tenant/app/monitoring_marked_article_service.go @@ -5,11 +5,13 @@ import ( "database/sql" "fmt" "net/url" + "sort" "strconv" "strings" "time" "github.com/jackc/pgx/v5" + "go.uber.org/zap" "github.com/geo-platform/tenant-api/internal/shared/auth" "github.com/geo-platform/tenant-api/internal/shared/response" @@ -61,6 +63,11 @@ type monitoringMarkedArticleURLMeta struct { LastPathSegment *string } +type monitoringProjectionRefreshScope struct { + BrandID int64 + BusinessDate time.Time +} + func (s *MonitoringService) ListMarkedArticles(ctx context.Context, params MonitoringMarkedArticleListParams) (*MonitoringMarkedArticleListResponse, error) { actor := auth.MustActor(ctx) pageReq := normalizeMonitoringPageRequest(MonitoringPageRequest{ @@ -208,9 +215,11 @@ func (s *MonitoringService) CreateMarkedArticle(ctx context.Context, req Monitor return nil, response.ErrInternal(50041, "marked_article_create_failed", "failed to save marked article") } - if err := s.syncMarkedArticleCitationFacts(ctx, actor.TenantID, id, meta.NormalizedURL); err != nil { + scopes, err := s.syncMarkedArticleCitationFacts(ctx, actor.TenantID, id, meta) + if err != nil { return nil, err } + s.refreshMarkedArticleProjectionScopes(ctx, actor.TenantID, scopes) return s.GetMarkedArticle(ctx, id) } @@ -260,14 +269,20 @@ func (s *MonitoringService) UpdateMarkedArticle(ctx context.Context, id int64, r if tag.RowsAffected() == 0 { return nil, response.ErrNotFound(40441, "marked_article_not_found", "marked article not found") } + scopes := make([]monitoringProjectionRefreshScope, 0) if oldNormalizedURL != "" && oldNormalizedURL != meta.NormalizedURL { - if err := s.clearMarkedArticleCitationFacts(ctx, actor.TenantID, id, oldNormalizedURL); err != nil { + clearedScopes, err := s.clearMarkedArticleCitationFacts(ctx, actor.TenantID, id, "") + if err != nil { return nil, err } + scopes = append(scopes, clearedScopes...) } - if err := s.syncMarkedArticleCitationFacts(ctx, actor.TenantID, id, meta.NormalizedURL); err != nil { + syncedScopes, err := s.syncMarkedArticleCitationFacts(ctx, actor.TenantID, id, meta) + if err != nil { return nil, err } + scopes = append(scopes, syncedScopes...) + s.refreshMarkedArticleProjectionScopes(ctx, actor.TenantID, scopes) return s.GetMarkedArticle(ctx, id) } @@ -297,9 +312,11 @@ func (s *MonitoringService) DeleteMarkedArticle(ctx context.Context, id int64) e if tag.RowsAffected() == 0 { return response.ErrNotFound(40441, "marked_article_not_found", "marked article not found") } - if err := s.clearMarkedArticleCitationFacts(ctx, actor.TenantID, id, normalizedURL); err != nil { + scopes, err := s.clearMarkedArticleCitationFacts(ctx, actor.TenantID, id, "") + if err != nil { return err } + s.refreshMarkedArticleProjectionScopes(ctx, actor.TenantID, scopes) return nil } @@ -431,27 +448,95 @@ func normalizeMarkedArticleDomain(value string) string { return trimmed } -func (s *MonitoringService) syncMarkedArticleCitationFacts(ctx context.Context, tenantID, markedArticleID int64, normalizedURL string) error { - if tenantID <= 0 || markedArticleID <= 0 || strings.TrimSpace(normalizedURL) == "" { - return nil +func (s *MonitoringService) syncMarkedArticleCitationFacts(ctx context.Context, tenantID, markedArticleID int64, meta monitoringMarkedArticleURLMeta) ([]monitoringProjectionRefreshScope, error) { + if tenantID <= 0 || markedArticleID <= 0 || strings.TrimSpace(meta.NormalizedURL) == "" { + return nil, nil + } + + matches, scopes, err := s.findMarkedArticleCitationFactMatches(ctx, tenantID, markedArticleID, meta) + if err != nil { + return nil, err + } + if len(matches) == 0 { + return scopes, nil } if _, err := s.monitoringPool.Exec(ctx, ` UPDATE monitoring_citation_facts SET content_source_type = 'manual_mark', content_source_id = $3 WHERE tenant_id = $1 - AND normalized_url = $2 + AND id = ANY($2::bigint[]) AND content_source_type IS NULL AND content_source_id IS NULL - `, tenantID, strings.TrimSpace(normalizedURL), markedArticleID); err != nil { - return response.ErrInternal(50041, "marked_article_citation_sync_failed", "failed to sync marked article citations") + `, tenantID, matches, markedArticleID); err != nil { + return nil, response.ErrInternal(50041, "marked_article_citation_sync_failed", "failed to sync marked article citations") } - return nil + return scopes, nil } -func (s *MonitoringService) clearMarkedArticleCitationFacts(ctx context.Context, tenantID, markedArticleID int64, normalizedURL string) error { +func (s *MonitoringService) findMarkedArticleCitationFactMatches(ctx context.Context, tenantID, markedArticleID int64, meta monitoringMarkedArticleURLMeta) ([]int64, []monitoringProjectionRefreshScope, error) { + targetKeys := monitoringCitationCandidateKeySet(meta.OriginalURL, meta.NormalizedURL) + if len(targetKeys) == 0 { + return nil, nil, nil + } + + rows, err := s.monitoringPool.Query(ctx, ` + SELECT + id, + cited_url, + normalized_url, + brand_id, + business_date + FROM monitoring_citation_facts + WHERE tenant_id = $1 + AND business_date >= (CURRENT_DATE - INTERVAL '6 months')::date + AND ( + (content_source_type IS NULL AND content_source_id IS NULL) + OR (content_source_type = 'manual_mark' AND content_source_id = $2) + ) + AND ( + normalized_url = $3 + OR LOWER(registrable_domain) = ANY($4::text[]) + OR LOWER(site_key) = ANY($4::text[]) + OR LOWER(host) = ANY($4::text[]) + ) + `, tenantID, markedArticleID, strings.TrimSpace(meta.NormalizedURL), markedArticleDomainKeys(meta)) + if err != nil { + return nil, nil, response.ErrInternal(50041, "marked_article_citation_lookup_failed", "failed to inspect marked article citations") + } + defer rows.Close() + + matchedIDs := make([]int64, 0) + scopeSet := make(map[string]monitoringProjectionRefreshScope) + for rows.Next() { + var factID int64 + var citedURL string + var normalizedURL string + var brandID int64 + var businessDate time.Time + if scanErr := rows.Scan(&factID, &citedURL, &normalizedURL, &brandID, &businessDate); scanErr != nil { + return nil, nil, response.ErrInternal(50041, "marked_article_citation_scan_failed", "failed to parse marked article citations") + } + if !monitoringCitationMatchesAnyKey(citedURL, normalizedURL, targetKeys) { + continue + } + matchedIDs = append(matchedIDs, factID) + key := monitoringProjectionScopeKey(brandID, businessDate) + scopeSet[key] = monitoringProjectionRefreshScope{ + BrandID: brandID, + BusinessDate: monitoringCanonicalBusinessDay(businessDate), + } + } + if err := rows.Err(); err != nil { + return nil, nil, response.ErrInternal(50041, "marked_article_citation_scan_failed", "failed to iterate marked article citations") + } + + return matchedIDs, sortedMonitoringProjectionRefreshScopes(scopeSet), nil +} + +func (s *MonitoringService) clearMarkedArticleCitationFacts(ctx context.Context, tenantID, markedArticleID int64, normalizedURL string) ([]monitoringProjectionRefreshScope, error) { if tenantID <= 0 || markedArticleID <= 0 { - return nil + return nil, nil } args := []interface{}{tenantID, markedArticleID} whereURL := "" @@ -459,17 +544,147 @@ func (s *MonitoringService) clearMarkedArticleCitationFacts(ctx context.Context, args = append(args, trimmed) whereURL = " AND normalized_url = $3" } - if _, err := s.monitoringPool.Exec(ctx, ` + rows, err := s.monitoringPool.Query(ctx, ` UPDATE monitoring_citation_facts SET content_source_type = NULL, content_source_id = NULL WHERE tenant_id = $1 AND content_source_type = 'manual_mark' AND content_source_id = $2 - `+whereURL, args...); err != nil { - return response.ErrInternal(50041, "marked_article_citation_clear_failed", "failed to clear marked article citations") + `+whereURL+` + RETURNING brand_id, business_date + `, args...) + if err != nil { + return nil, response.ErrInternal(50041, "marked_article_citation_clear_failed", "failed to clear marked article citations") } - return nil + defer rows.Close() + + scopeSet := make(map[string]monitoringProjectionRefreshScope) + for rows.Next() { + var brandID int64 + var businessDate time.Time + if scanErr := rows.Scan(&brandID, &businessDate); scanErr != nil { + return nil, response.ErrInternal(50041, "marked_article_citation_clear_scan_failed", "failed to parse cleared marked article citations") + } + key := monitoringProjectionScopeKey(brandID, businessDate) + scopeSet[key] = monitoringProjectionRefreshScope{ + BrandID: brandID, + BusinessDate: monitoringCanonicalBusinessDay(businessDate), + } + } + if err := rows.Err(); err != nil { + return nil, response.ErrInternal(50041, "marked_article_citation_clear_scan_failed", "failed to iterate cleared marked article citations") + } + + return sortedMonitoringProjectionRefreshScopes(scopeSet), nil +} + +func (s *MonitoringService) refreshMarkedArticleProjectionScopes(ctx context.Context, tenantID int64, scopes []monitoringProjectionRefreshScope) { + if s == nil || s.monitoringPool == nil || tenantID <= 0 || len(scopes) == 0 { + return + } + + grouped := make(map[string][]int64) + for _, scope := range scopes { + if scope.BrandID <= 0 || scope.BusinessDate.IsZero() { + continue + } + dateText := monitoringBusinessDateText(scope.BusinessDate) + grouped[dateText] = append(grouped[dateText], scope.BrandID) + } + if len(grouped) == 0 { + return + } + + rebuilder := &MonitoringCallbackService{ + monitoringPool: s.monitoringPool, + logger: s.logger, + } + for dateText, brandIDs := range grouped { + businessDate, err := parseMonitoringBusinessDateInLocation(dateText, monitoringBusinessLocation()) + if err != nil { + continue + } + rebuildCtx, cancel := context.WithTimeout(ctx, monitoringProjectionFallbackTimeout) + err = rebuilder.rebuildBrandDailyProjectionBatch(rebuildCtx, tenantID, businessDate, brandIDs) + cancel() + if err != nil && s.logger != nil { + s.logger.Warn("marked article projection refresh failed", + zap.Int64("tenant_id", tenantID), + zap.String("business_date", dateText), + zap.Int("brand_count", len(uniqueSortedMonitoringProjectionBrandIDs(brandIDs))), + zap.Error(err), + ) + } + } +} + +func monitoringCitationCandidateKeySet(rawURL, normalizedURL string) map[string]struct{} { + keys := monitoringCitationCandidateKeys(rawURL, normalizedURL) + result := make(map[string]struct{}, len(keys)) + for _, key := range keys { + key = strings.TrimSpace(key) + if key != "" { + result[key] = struct{}{} + } + } + return result +} + +func monitoringCitationMatchesAnyKey(rawURL, normalizedURL string, targetKeys map[string]struct{}) bool { + if len(targetKeys) == 0 { + return false + } + for _, key := range monitoringCitationCandidateKeys(rawURL, normalizedURL) { + if _, ok := targetKeys[key]; ok { + return true + } + } + return false +} + +func markedArticleDomainKeys(meta monitoringMarkedArticleURLMeta) []string { + values := []string{meta.RegistrableDomain, meta.SiteKey, meta.Host} + result := make([]string, 0, len(values)) + seen := make(map[string]struct{}, len(values)) + for _, value := range values { + key := strings.ToLower(strings.TrimSpace(value)) + if key == "" { + continue + } + if _, ok := seen[key]; ok { + continue + } + seen[key] = struct{}{} + result = append(result, key) + } + return result +} + +func monitoringProjectionScopeKey(brandID int64, businessDate time.Time) string { + return strconv.FormatInt(brandID, 10) + ":" + monitoringBusinessDateText(businessDate) +} + +func sortedMonitoringProjectionRefreshScopes(scopeSet map[string]monitoringProjectionRefreshScope) []monitoringProjectionRefreshScope { + if len(scopeSet) == 0 { + return nil + } + items := make([]monitoringProjectionRefreshScope, 0, len(scopeSet)) + for _, item := range scopeSet { + if item.BrandID <= 0 || item.BusinessDate.IsZero() { + continue + } + items = append(items, item) + } + sort.Slice(items, func(left, right int) bool { + leftDate := monitoringBusinessDateText(items[left].BusinessDate) + rightDate := monitoringBusinessDateText(items[right].BusinessDate) + if leftDate != rightDate { + return leftDate < rightDate + } + return items[left].BrandID < items[right].BrandID + }) + return items } func nullableTimePtr(value *time.Time) interface{} { diff --git a/server/internal/tenant/app/monitoring_projection.go b/server/internal/tenant/app/monitoring_projection.go index 7ae441f..3801517 100644 --- a/server/internal/tenant/app/monitoring_projection.go +++ b/server/internal/tenant/app/monitoring_projection.go @@ -216,7 +216,6 @@ func (s *MonitoringCallbackService) loadMonitoringBrandDailyAggregate(ctx contex AND cf.business_date = $4::date AND cf.content_source_type IS NOT NULL AND cf.content_source_id IS NOT NULL - AND r.ai_platform_id <> 'qwen' AND r.collector_type = $3 AND r.business_date = $4::date AND r.status = 'succeeded' diff --git a/server/internal/tenant/app/monitoring_service.go b/server/internal/tenant/app/monitoring_service.go index 6576987..8b97f09 100644 --- a/server/internal/tenant/app/monitoring_service.go +++ b/server/internal/tenant/app/monitoring_service.go @@ -3454,18 +3454,15 @@ func (s *MonitoringService) loadCitationRanking( fr.ai_platform_id, COUNT(cf.id)::bigint AS citation_source_count, COUNT(cf.id) FILTER ( - WHERE fr.ai_platform_id <> 'qwen' - AND cf.content_source_type IS NOT NULL + WHERE cf.content_source_type IS NOT NULL AND cf.content_source_id IS NOT NULL )::bigint AS saas_source_count, COUNT(DISTINCT cf.run_id) FILTER ( - WHERE fr.ai_platform_id <> 'qwen' - AND cf.content_source_type IS NOT NULL + WHERE cf.content_source_type IS NOT NULL AND cf.content_source_id IS NOT NULL )::bigint AS cited_answer_count, COUNT(DISTINCT cf.content_source_type || ':' || cf.content_source_id::text) FILTER ( - WHERE fr.ai_platform_id <> 'qwen' - AND cf.content_source_type IS NOT NULL + WHERE cf.content_source_type IS NOT NULL AND cf.content_source_id IS NOT NULL )::bigint AS cited_article_count FROM filtered_runs fr @@ -3658,7 +3655,6 @@ func (s *MonitoringService) loadCitedArticleSourceRows( AND r.id = cf.run_id WHERE cf.tenant_id = $1 AND ($2::bigint = 0 OR cf.brand_id = $2) - AND cf.ai_platform_id <> 'qwen' AND cf.business_date BETWEEN $4::date AND $5::date AND cf.content_source_type IS NOT NULL AND cf.content_source_id IS NOT NULL @@ -4278,7 +4274,7 @@ func (s *MonitoringService) loadQuestionCitationAnalysis(ctx context.Context, te } } item.CitationCount++ - if platformID != "qwen" && sourceType.Valid && strings.TrimSpace(sourceType.String) != "" && sourceID.Valid { + if sourceType.Valid && strings.TrimSpace(sourceType.String) != "" && sourceID.Valid { item.ContentCitationCount++ } aggregates[key] = item @@ -4324,7 +4320,6 @@ func (s *MonitoringService) loadQuestionContentCitations(ctx context.Context, te ON cf.tenant_id = r.tenant_id AND cf.run_id = r.id WHERE r.tenant_id = $1 AND r.status = 'succeeded' - AND r.ai_platform_id <> 'qwen' AND r.id = ANY($2) `, tenantID, runIDs) if err != nil { diff --git a/server/migrations_monitoring/20260625102000_speed_manual_mark_citation_sync.down.sql b/server/migrations_monitoring/20260625102000_speed_manual_mark_citation_sync.down.sql new file mode 100644 index 0000000..4287a8a --- /dev/null +++ b/server/migrations_monitoring/20260625102000_speed_manual_mark_citation_sync.down.sql @@ -0,0 +1,2 @@ +DROP INDEX IF EXISTS idx_citation_facts_manual_mark_refresh; +DROP INDEX IF EXISTS idx_citation_facts_manual_mark_sync_candidates; diff --git a/server/migrations_monitoring/20260625102000_speed_manual_mark_citation_sync.up.sql b/server/migrations_monitoring/20260625102000_speed_manual_mark_citation_sync.up.sql new file mode 100644 index 0000000..4dda13c --- /dev/null +++ b/server/migrations_monitoring/20260625102000_speed_manual_mark_citation_sync.up.sql @@ -0,0 +1,21 @@ +CREATE INDEX IF NOT EXISTS idx_citation_facts_manual_mark_sync_candidates + ON monitoring_citation_facts ( + tenant_id, + business_date, + registrable_domain, + site_key, + host + ) + WHERE content_source_type IS NULL + AND content_source_id IS NULL; + +CREATE INDEX IF NOT EXISTS idx_citation_facts_manual_mark_refresh + ON monitoring_citation_facts ( + tenant_id, + content_source_type, + content_source_id, + brand_id, + business_date + ) + WHERE content_source_type = 'manual_mark' + AND content_source_id IS NOT NULL; diff --git a/server/migrations_monitoring/20260625103500_backfill_manual_mark_citation_slash_variants.down.sql b/server/migrations_monitoring/20260625103500_backfill_manual_mark_citation_slash_variants.down.sql new file mode 100644 index 0000000..7e90d4b --- /dev/null +++ b/server/migrations_monitoring/20260625103500_backfill_manual_mark_citation_slash_variants.down.sql @@ -0,0 +1,53 @@ +CREATE TEMP TABLE tmp_manual_mark_backfill_affected ON COMMIT DROP AS +WITH cleared_facts AS ( + UPDATE monitoring_citation_facts cf + SET content_source_type = NULL, + content_source_id = NULL + FROM monitoring_user_marked_articles marked + WHERE marked.tenant_id = cf.tenant_id + AND marked.id = cf.content_source_id + AND cf.content_source_type = 'manual_mark' + AND btrim(cf.normalized_url, '/') = btrim(marked.normalized_url, '/') + RETURNING cf.tenant_id, cf.brand_id, cf.business_date +) +SELECT DISTINCT tenant_id, brand_id, business_date +FROM cleared_facts; + +WITH citation_counts AS ( + SELECT + affected.tenant_id, + affected.brand_id, + affected.business_date, + runs.collector_type, + COUNT(DISTINCT cf.run_id)::int AS cited_answer_count + FROM tmp_manual_mark_backfill_affected affected + JOIN monitoring_citation_facts cf + ON cf.tenant_id = affected.tenant_id + AND cf.brand_id = affected.brand_id + AND cf.business_date = affected.business_date + AND cf.content_source_type IS NOT NULL + AND cf.content_source_id IS NOT NULL + JOIN question_monitor_runs runs + ON runs.tenant_id = cf.tenant_id + AND runs.id = cf.run_id + AND runs.business_date = cf.business_date + AND runs.status = 'succeeded' + GROUP BY affected.tenant_id, affected.brand_id, affected.business_date, runs.collector_type +) +UPDATE monitoring_brand_daily daily +SET cited_answer_count = COALESCE(counts.cited_answer_count, 0), + citation_rate = CASE + WHEN daily.actual_sample_count > 0 + THEN COALESCE(counts.cited_answer_count, 0)::numeric / daily.actual_sample_count::numeric + ELSE NULL + END, + updated_at = NOW() +FROM tmp_manual_mark_backfill_affected affected +LEFT JOIN citation_counts counts + ON counts.tenant_id = affected.tenant_id + AND counts.brand_id = affected.brand_id + AND counts.business_date = affected.business_date +WHERE daily.tenant_id = affected.tenant_id + AND daily.brand_id = affected.brand_id + AND daily.business_date = affected.business_date + AND (counts.collector_type IS NULL OR daily.collector_type = counts.collector_type); diff --git a/server/migrations_monitoring/20260625103500_backfill_manual_mark_citation_slash_variants.up.sql b/server/migrations_monitoring/20260625103500_backfill_manual_mark_citation_slash_variants.up.sql new file mode 100644 index 0000000..07e72d0 --- /dev/null +++ b/server/migrations_monitoring/20260625103500_backfill_manual_mark_citation_slash_variants.up.sql @@ -0,0 +1,60 @@ +CREATE TEMP TABLE tmp_manual_mark_backfill_affected ON COMMIT DROP AS +WITH matched_facts AS ( + SELECT DISTINCT ON (cf.id) + cf.id AS citation_fact_id, + marked.id AS marked_article_id + FROM monitoring_citation_facts cf + JOIN monitoring_user_marked_articles marked + ON marked.tenant_id = cf.tenant_id + AND marked.expires_at > NOW() + WHERE cf.content_source_type IS NULL + AND cf.content_source_id IS NULL + AND cf.business_date >= (CURRENT_DATE - INTERVAL '6 months')::date + AND btrim(cf.normalized_url, '/') = btrim(marked.normalized_url, '/') + ORDER BY cf.id, marked.updated_at DESC, marked.id DESC +), +updated_facts AS ( + UPDATE monitoring_citation_facts cf + SET content_source_type = 'manual_mark', + content_source_id = matched.marked_article_id + FROM matched_facts matched + WHERE cf.id = matched.citation_fact_id + RETURNING cf.tenant_id, cf.brand_id, cf.business_date +) +SELECT DISTINCT tenant_id, brand_id, business_date +FROM updated_facts; + +WITH citation_counts AS ( + SELECT + affected.tenant_id, + affected.brand_id, + affected.business_date, + runs.collector_type, + COUNT(DISTINCT cf.run_id)::int AS cited_answer_count + FROM tmp_manual_mark_backfill_affected affected + JOIN monitoring_citation_facts cf + ON cf.tenant_id = affected.tenant_id + AND cf.brand_id = affected.brand_id + AND cf.business_date = affected.business_date + AND cf.content_source_type IS NOT NULL + AND cf.content_source_id IS NOT NULL + JOIN question_monitor_runs runs + ON runs.tenant_id = cf.tenant_id + AND runs.id = cf.run_id + AND runs.business_date = cf.business_date + AND runs.status = 'succeeded' + GROUP BY affected.tenant_id, affected.brand_id, affected.business_date, runs.collector_type +) +UPDATE monitoring_brand_daily daily +SET cited_answer_count = counts.cited_answer_count, + citation_rate = CASE + WHEN daily.actual_sample_count > 0 + THEN counts.cited_answer_count::numeric / daily.actual_sample_count::numeric + ELSE NULL + END, + updated_at = NOW() +FROM citation_counts counts +WHERE daily.tenant_id = counts.tenant_id + AND daily.brand_id = counts.brand_id + AND daily.business_date = counts.business_date + AND daily.collector_type = counts.collector_type;