root/geo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(monitoring): attribute SaaS citations by registrable domain

Browse Source 
Persist host/registrable_domain/site_key on monitoring article URL
aliases, populate them on desktop-publish completion, and use them in
the citation pipeline to count how often SaaS-published content
appears as a citation source. Expose a /citation-summary endpoint with
a 7/30-day window switch and surface the new source-share metrics in
the tracking dashboard.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
liangxu
2026-04-30 17:12:09 +08:00
parent 6e6e19cccb
commit fb166fa9fe
15 changed files with 836 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/tenant/app/monitoring_service.go
+215 -86
View File
@@ -74,12 +74,25 @@ func (s *MonitoringService) WithRedis(redis *goredis.Client) *MonitoringService
type MonitoringDashboardCompositeResponse struct {
Overview MonitoringOverview `json:"overview"`
Runtime MonitoringDashboardRuntime `json:"runtime"`
CitationWindow MonitoringDashboardWindow `json:"citation_window"`
PlatformBreakdown []MonitoringPlatformDaily `json:"platform_breakdown"`
HotQuestions []MonitoringHotQuestion `json:"hot_questions"`
CitationRanking []MonitoringCitationRanking `json:"citation_ranking"`
CitedArticles []MonitoringCitedArticle `json:"cited_articles"`
}
type MonitoringCitationSummaryResponse struct {
CitationWindow MonitoringDashboardWindow `json:"citation_window"`
CitationRanking []MonitoringCitationRanking `json:"citation_ranking"`
CitedArticles []MonitoringCitedArticle `json:"cited_articles"`
}
type MonitoringDashboardWindow struct {
Days int `json:"days"`
From string `json:"from"`
To string `json:"to"`
}
type MonitoringDashboardRuntime struct {
CurrentUserClientOnline bool `json:"current_user_client_online"`
PlatformAuthorizationStatus string `json:"platform_authorization_status"`
@@ -126,12 +139,15 @@ type MonitoringHotQuestion struct {
}
type MonitoringCitationRanking struct {
AIPlatformID string `json:"ai_platform_id"`
PlatformName string `json:"platform_name"`
SampleStatus string `json:"sample_status"`
CitedAnswerCount int64 `json:"cited_answer_count"`
CitedArticleCount int64 `json:"cited_article_count"`
CitationRate *float64 `json:"citation_rate"`
AIPlatformID string `json:"ai_platform_id"`
PlatformName string `json:"platform_name"`
SampleStatus string `json:"sample_status"`
CitedAnswerCount int64 `json:"cited_answer_count"`
CitedArticleCount int64 `json:"cited_article_count"`
CitationSourceCount int64 `json:"citation_source_count"`
SaaSSourceCount int64 `json:"saas_source_count"`
SaaSSourceRate *float64 `json:"saas_source_rate"`
CitationRate *float64 `json:"citation_rate"`
}
type MonitoringCitedArticle struct {
@@ -140,6 +156,7 @@ type MonitoringCitedArticle struct {
PublishPlatform string `json:"publish_platform"`
CitationCount int64 `json:"citation_count"`
CitationRate *float64 `json:"citation_rate"`
SourceShare *float64 `json:"source_share"`
}
type MonitoringQuestionDetailResponse struct {
@@ -347,7 +364,6 @@ func (s *MonitoringService) DashboardComposite(
if err != nil {
return nil, err
}
questionIDs := configuredQuestionIDs(configuredQuestions)
platforms := defaultMonitoringPlatformMetadata()
selectedPlatformID := normalizedOptionalMonitoringPlatformPointer(aiPlatformID)
@@ -380,23 +396,59 @@ func (s *MonitoringService) DashboardComposite(
return nil, err
}
citationRanking, err := s.loadCitationRanking(ctx, actor.TenantID, brand.ID, questionIDs, startDate, endDate, accessStates, selectedPlatformID)
if err != nil {
return nil, err
}
citedArticles, err := s.loadCitedArticles(ctx, actor.TenantID, brand.ID, questionIDs, startDate, endDate, selectedPlatformID)
if err != nil {
return nil, err
}
return &MonitoringDashboardCompositeResponse{
Overview: overview,
Runtime: runtime,
Overview: overview,
Runtime: runtime,
CitationWindow: MonitoringDashboardWindow{
Days: defaultTrackingDays,
From: endDate.AddDate(0, 0, -(defaultTrackingDays - 1)).Format("2006-01-02"),
To: endDate.Format("2006-01-02"),
},
PlatformBreakdown: platformBreakdown,
HotQuestions: hotQuestions,
CitationRanking: citationRanking,
CitedArticles: citedArticles,
CitationRanking: []MonitoringCitationRanking{},
CitedArticles: []MonitoringCitedArticle{},
}, nil
}
func (s *MonitoringService) CitationSummary(
ctx context.Context,
days int,
) (*MonitoringCitationSummaryResponse, error) {
actor := auth.MustActor(ctx)
workspaceID := auth.CurrentWorkspaceID(ctx)
quota, err := s.loadQuota(ctx, actor, workspaceID)
if err != nil {
return nil, err
}
citationWindowDays := normalizeCitationWindowDays(days)
startDate, endDate := trackingDateWindow(citationWindowDays)
accessStates, err := s.loadAccessStates(ctx, actor.TenantID, workspaceID, quota.PrimaryClientID, endDate)
if err != nil {
return nil, err
}
citationRanking, err := s.loadCitationRanking(ctx, actor.TenantID, 0, nil, startDate, endDate, accessStates, nil)
if err != nil {
return nil, err
}
citedArticles, err := s.loadCitedArticles(ctx, actor.TenantID, 0, nil, startDate, endDate, nil)
if err != nil {
return nil, err
}
return &MonitoringCitationSummaryResponse{
CitationWindow: MonitoringDashboardWindow{
Days: citationWindowDays,
From: startDate.Format("2006-01-02"),
To: endDate.Format("2006-01-02"),
},
CitationRanking: citationRanking,
CitedArticles: citedArticles,
}, nil
}
@@ -509,12 +561,12 @@ func (s *MonitoringService) QuestionDetail(ctx context.Context, brandID, questio
})
}
citationAnalysis, err := s.loadQuestionCitationAnalysis(ctx, actor.TenantID, brand.ID, questionID, hashBytes, fromDate, toDate, selectedPlatformID)
citationAnalysis, err := s.loadQuestionCitationAnalysis(ctx, actor.TenantID, runIDs)
if err != nil {
return nil, err
}
contentCitations, err := s.loadQuestionContentCitations(ctx, actor.TenantID, brand.ID, questionID, hashBytes, fromDate, toDate, selectedPlatformID)
contentCitations, err := s.loadQuestionContentCitations(ctx, actor.TenantID, runIDs)
if err != nil {
return nil, err
}
@@ -2430,13 +2482,26 @@ func (s *MonitoringService) loadCitationRanking(
platformQueryIDs := monitoringPlatformQueryIDs(aiPlatformID)
rows, err := s.monitoringPool.Query(ctx, `
WITH run_counts AS (
WITH published_domains AS (
SELECT DISTINCT domain_key
FROM (
SELECT NULLIF(LOWER(TRIM(host)), '') AS domain_key
FROM monitoring_article_url_aliases
WHERE tenant_id = $1
UNION
SELECT NULLIF(LOWER(TRIM(site_key)), '') AS domain_key
FROM monitoring_article_url_aliases
WHERE tenant_id = $1
) domains
WHERE domain_key IS NOT NULL
),
run_counts AS (
SELECT
r.ai_platform_id,
COUNT(*) AS sample_count
FROM question_monitor_runs r
WHERE r.tenant_id = $1
AND r.brand_id = $2
AND ($2::bigint = 0 OR r.brand_id = $2)
AND r.collector_type = $3
AND r.status = 'succeeded'
AND r.business_date BETWEEN $4::date AND $5::date
@@ -2444,36 +2509,65 @@ func (s *MonitoringService) loadCitationRanking(
AND ($7::text[] IS NULL OR r.ai_platform_id = ANY($7))
GROUP BY r.ai_platform_id
),
citation_counts AS (
citation_sources AS (
SELECT
r.ai_platform_id,
COUNT(DISTINCT cf.run_id) FILTER (WHERE cf.article_id IS NOT NULL AND r.ai_platform_id <> 'qwen') AS cited_answer_count,
COUNT(DISTINCT cf.article_id) FILTER (WHERE cf.article_id IS NOT NULL AND r.ai_platform_id <> 'qwen') AS cited_article_count
cf.id,
cf.run_id,
cf.article_id,
(
cf.article_id IS NOT NULL
OR EXISTS (
SELECT 1
FROM published_domains pd
WHERE pd.domain_key = ANY(ARRAY[
LOWER(NULLIF(cf.registrable_domain, '')),
LOWER(NULLIF(cf.site_key, '')),
LOWER(NULLIF(cf.host, ''))
])
)
) AS matched_saas_source
FROM question_monitor_runs r
JOIN monitoring_citation_facts cf
ON cf.tenant_id = r.tenant_id AND cf.run_id = r.id
WHERE r.tenant_id = $1
AND r.brand_id = $2
AND ($2::bigint = 0 OR r.brand_id = $2)
AND r.collector_type = $3
AND r.status = 'succeeded'
AND r.business_date BETWEEN $4::date AND $5::date
AND ($6::bigint[] IS NULL OR r.question_id = ANY($6))
AND ($7::text[] IS NULL OR r.ai_platform_id = ANY($7))
GROUP BY r.ai_platform_id
),
citation_counts AS (
SELECT
ai_platform_id,
COUNT(*) AS citation_source_count,
COUNT(*) FILTER (WHERE matched_saas_source AND ai_platform_id <> 'qwen') AS saas_source_count,
COUNT(DISTINCT run_id) FILTER (WHERE matched_saas_source AND ai_platform_id <> 'qwen') AS cited_answer_count,
COUNT(DISTINCT article_id) FILTER (WHERE article_id IS NOT NULL AND ai_platform_id <> 'qwen') AS cited_article_count
FROM citation_sources
GROUP BY ai_platform_id
)
SELECT
rc.ai_platform_id,
rc.sample_count,
COALESCE(cc.cited_answer_count, 0) AS cited_answer_count,
COALESCE(cc.cited_article_count, 0) AS cited_article_count,
COALESCE(cc.citation_source_count, 0) AS citation_source_count,
COALESCE(cc.saas_source_count, 0) AS saas_source_count,
CASE
WHEN COALESCE(cc.citation_source_count, 0) > 0
THEN COALESCE(cc.saas_source_count, 0)::double precision / cc.citation_source_count::double precision
ELSE NULL
END AS saas_source_rate,
CASE
WHEN rc.sample_count > 0 THEN COALESCE(cc.cited_answer_count, 0)::double precision / rc.sample_count
ELSE NULL
END AS citation_rate
FROM run_counts rc
LEFT JOIN citation_counts cc ON cc.ai_platform_id = rc.ai_platform_id
WHERE COALESCE(cc.cited_answer_count, 0) > 0
ORDER BY cited_answer_count DESC, cited_article_count DESC, rc.ai_platform_id ASC
WHERE COALESCE(cc.saas_source_count, 0) > 0
ORDER BY cited_answer_count DESC, saas_source_count DESC, cited_article_count DESC, rc.ai_platform_id ASC
`, tenantID, brandID, monitoringCollectorType, startDate.Format("2006-01-02"), endDate.Format("2006-01-02"), nullableInt64Array(questionIDs), nullableStringArray(platformQueryIDs))
if err != nil {
return nil, response.ErrInternal(50041, "query_failed", "failed to load citation ranking")
@@ -2481,9 +2575,11 @@ func (s *MonitoringService) loadCitationRanking(
defer rows.Close()
type citationRankingAggregate struct {
CitedAnswerCount int64
CitedArticleCount int64
SampleCount int64
CitedAnswerCount int64
CitedArticleCount int64
CitationSourceCount int64
SaaSSourceCount int64
SampleCount int64
}
aggregates := make(map[string]citationRankingAggregate)
@@ -2492,7 +2588,9 @@ func (s *MonitoringService) loadCitationRanking(
var sampleCount int64
var citedAnswerCount int64
var citedArticleCount int64
if scanErr := rows.Scan(&platformID, &sampleCount, &citedAnswerCount, &citedArticleCount, new(sql.NullFloat64)); scanErr != nil {
var citationSourceCount int64
var saasSourceCount int64
if scanErr := rows.Scan(&platformID, &sampleCount, &citedAnswerCount, &citedArticleCount, &citationSourceCount, &saasSourceCount, new(sql.NullFloat64), new(sql.NullFloat64)); scanErr != nil {
return nil, response.ErrInternal(50041, "scan_failed", "failed to parse citation ranking")
}
platformID = normalizeMonitoringPlatformID(platformID)
@@ -2502,6 +2600,8 @@ func (s *MonitoringService) loadCitationRanking(
item := aggregates[platformID]
item.CitedAnswerCount += citedAnswerCount
item.CitedArticleCount += citedArticleCount
item.CitationSourceCount += citationSourceCount
item.SaaSSourceCount += saasSourceCount
item.SampleCount += sampleCount
aggregates[platformID] = item
}
@@ -2509,18 +2609,24 @@ func (s *MonitoringService) loadCitationRanking(
items := make([]MonitoringCitationRanking, 0, len(aggregates))
for platformID, aggregate := range aggregates {
items = append(items, MonitoringCitationRanking{
AIPlatformID: platformID,
PlatformName: platformDisplayName(platformID),
SampleStatus: derivePlatformSampleStatus("", accessStates[platformID]),
CitedAnswerCount: aggregate.CitedAnswerCount,
CitedArticleCount: aggregate.CitedArticleCount,
CitationRate: divideAsPointer(aggregate.CitedAnswerCount, aggregate.SampleCount),
AIPlatformID: platformID,
PlatformName: platformDisplayName(platformID),
SampleStatus: derivePlatformSampleStatus("", accessStates[platformID]),
CitedAnswerCount: aggregate.CitedAnswerCount,
CitedArticleCount: aggregate.CitedArticleCount,
CitationSourceCount: aggregate.CitationSourceCount,
SaaSSourceCount: aggregate.SaaSSourceCount,
SaaSSourceRate: divideAsPointer(aggregate.SaaSSourceCount, aggregate.CitationSourceCount),
CitationRate: divideAsPointer(aggregate.CitedAnswerCount, aggregate.SampleCount),
})
}
sort.Slice(items, func(left, right int) bool {
if items[left].CitedAnswerCount != items[right].CitedAnswerCount {
return items[left].CitedAnswerCount > items[right].CitedAnswerCount
}
if items[left].SaaSSourceCount != items[right].SaaSSourceCount {
return items[left].SaaSSourceCount > items[right].SaaSSourceCount
}
if items[left].CitedArticleCount != items[right].CitedArticleCount {
return items[left].CitedArticleCount > items[right].CitedArticleCount
}
@@ -2547,7 +2653,7 @@ func (s *MonitoringService) loadCitedArticles(
SELECT COUNT(*)
FROM question_monitor_runs r
WHERE r.tenant_id = $1
AND r.brand_id = $2
AND ($2::bigint = 0 OR r.brand_id = $2)
AND r.collector_type = $3
AND r.status = 'succeeded'
AND r.business_date BETWEEN $4::date AND $5::date
@@ -2558,26 +2664,51 @@ func (s *MonitoringService) loadCitedArticles(
}
rows, err := s.monitoringPool.Query(ctx, `
WITH article_counts AS (
SELECT
cf.article_id,
COALESCE(MAX(alias.article_title_snapshot), '未命名文章') AS article_title,
COALESCE(MAX(alias.publish_platform_name_snapshot), '已发布内容') AS publish_platform,
COUNT(*) AS citation_count
FROM monitoring_citation_facts cf
JOIN question_monitor_runs r
ON r.tenant_id = cf.tenant_id AND r.id = cf.run_id
LEFT JOIN LATERAL (
SELECT
alias.article_title_snapshot,
alias.publish_platform_name_snapshot
FROM monitoring_article_url_aliases alias
WHERE alias.tenant_id = cf.tenant_id
AND alias.article_id = cf.article_id
ORDER BY
(alias.publish_record_id = cf.publish_record_id) DESC NULLS LAST,
alias.updated_at DESC,
alias.id DESC
LIMIT 1
) alias ON TRUE
WHERE cf.tenant_id = $1
AND ($2::bigint = 0 OR cf.brand_id = $2)
AND cf.article_id IS NOT NULL
AND r.ai_platform_id <> 'qwen'
AND r.collector_type = $3
AND r.status = 'succeeded'
AND r.business_date BETWEEN $4::date AND $5::date
AND ($6::bigint[] IS NULL OR r.question_id = ANY($6))
AND ($7::text[] IS NULL OR r.ai_platform_id = ANY($7))
GROUP BY cf.article_id
),
totals AS (
SELECT COALESCE(SUM(citation_count), 0) AS total_article_citation_count
FROM article_counts
)
SELECT
cf.article_id,
COALESCE(MAX(alias.article_title_snapshot), '未命名文章') AS article_title,
COALESCE(MAX(alias.publish_platform_name_snapshot), '已发布内容') AS publish_platform,
COUNT(*) AS citation_count
FROM monitoring_citation_facts cf
JOIN question_monitor_runs r
ON r.tenant_id = cf.tenant_id AND r.id = cf.run_id
LEFT JOIN monitoring_article_url_aliases alias
ON alias.tenant_id = cf.tenant_id AND alias.article_id = cf.article_id
WHERE cf.tenant_id = $1
AND cf.brand_id = $2
AND cf.article_id IS NOT NULL
AND r.ai_platform_id <> 'qwen'
AND r.collector_type = $3
AND r.status = 'succeeded'
AND r.business_date BETWEEN $4::date AND $5::date
AND ($6::bigint[] IS NULL OR r.question_id = ANY($6))
AND ($7::text[] IS NULL OR r.ai_platform_id = ANY($7))
GROUP BY cf.article_id
cf.article_title,
cf.publish_platform,
cf.citation_count,
t.total_article_citation_count
FROM article_counts cf
CROSS JOIN totals t
ORDER BY citation_count DESC, cf.article_id ASC
LIMIT 10
`, tenantID, brandID, monitoringCollectorType, startDate.Format("2006-01-02"), endDate.Format("2006-01-02"), nullableInt64Array(questionIDs), nullableStringArray(platformQueryIDs))
@@ -2592,7 +2723,8 @@ func (s *MonitoringService) loadCitedArticles(
var title string
var publishPlatform string
var citationCount int64
if scanErr := rows.Scan(&articleID, &title, &publishPlatform, &citationCount); scanErr != nil {
var totalArticleCitationCount int64
if scanErr := rows.Scan(&articleID, &title, &publishPlatform, &citationCount, &totalArticleCitationCount); scanErr != nil {
return nil, response.ErrInternal(50041, "scan_failed", "failed to parse cited articles")
}
items = append(items, MonitoringCitedArticle{
@@ -2601,6 +2733,7 @@ func (s *MonitoringService) loadCitedArticles(
PublishPlatform: publishPlatform,
CitationCount: citationCount,
CitationRate: divideAsPointer(citationCount, totalSampleCount),
SourceShare: divideAsPointer(citationCount, totalArticleCitationCount),
})
}
@@ -2902,8 +3035,10 @@ func (s *MonitoringService) loadCitationsForRuns(ctx context.Context, tenantID i
return result, nil
}
func (s *MonitoringService) loadQuestionCitationAnalysis(ctx context.Context, tenantID, brandID, questionID int64, hashBytes []byte, fromDate, toDate time.Time, aiPlatformID *string) ([]MonitoringQuestionCitationStats, error) {
platformQueryIDs := monitoringPlatformQueryIDs(aiPlatformID)
func (s *MonitoringService) loadQuestionCitationAnalysis(ctx context.Context, tenantID int64, runIDs []int64) ([]MonitoringQuestionCitationStats, error) {
if len(runIDs) == 0 {
return []MonitoringQuestionCitationStats{}, nil
}
rows, err := s.monitoringPool.Query(ctx, `
WITH filtered_facts AS (
@@ -2918,13 +3053,8 @@ func (s *MonitoringService) loadQuestionCitationAnalysis(ctx context.Context, te
JOIN monitoring_citation_facts cf
ON cf.tenant_id = r.tenant_id AND cf.run_id = r.id
WHERE r.tenant_id = $1
AND r.brand_id = $2
AND r.question_id = $3
AND r.question_hash = $4
AND r.collector_type = $5
AND r.status = 'succeeded'
AND r.business_date BETWEEN $6::date AND $7::date
AND ($8::text[] IS NULL OR r.ai_platform_id = ANY($8))
AND r.id = ANY($2)
),
domain_keys AS (
SELECT DISTINCT host, registrable_domain, site_key
@@ -3003,7 +3133,7 @@ func (s *MonitoringService) loadQuestionCitationAnalysis(ctx context.Context, te
JOIN totals t
ON t.ai_platform_id = g.ai_platform_id
ORDER BY g.ai_platform_id ASC, g.citation_count DESC, g.site_name ASC
`, tenantID, brandID, questionID, hashBytes, monitoringCollectorType, fromDate.Format("2006-01-02"), toDate.Format("2006-01-02"), nullableStringArray(platformQueryIDs))
`, tenantID, runIDs)
if err != nil {
return nil, response.ErrInternal(50041, "query_failed", "failed to load question citation analysis")
}
@@ -3071,8 +3201,10 @@ func (s *MonitoringService) loadQuestionCitationAnalysis(ctx context.Context, te
return items, nil
}
func (s *MonitoringService) loadQuestionContentCitations(ctx context.Context, tenantID, brandID, questionID int64, hashBytes []byte, fromDate, toDate time.Time, aiPlatformID *string) ([]MonitoringQuestionContentCitation, error) {
platformQueryIDs := monitoringPlatformQueryIDs(aiPlatformID)
func (s *MonitoringService) loadQuestionContentCitations(ctx context.Context, tenantID int64, runIDs []int64) ([]MonitoringQuestionContentCitation, error) {
if len(runIDs) == 0 {
return []MonitoringQuestionContentCitation{}, nil
}
rows, err := s.monitoringPool.Query(ctx, `
WITH platform_totals AS (
@@ -3083,13 +3215,8 @@ func (s *MonitoringService) loadQuestionContentCitations(ctx context.Context, te
JOIN monitoring_citation_facts cf
ON cf.tenant_id = r.tenant_id AND cf.run_id = r.id
WHERE r.tenant_id = $1
AND r.brand_id = $2
AND r.question_id = $3
AND r.question_hash = $4
AND r.collector_type = $5
AND r.status = 'succeeded'
AND r.business_date BETWEEN $6::date AND $7::date
AND ($8::text[] IS NULL OR r.ai_platform_id = ANY($8))
AND r.id = ANY($2)
GROUP BY r.ai_platform_id
)
SELECT
@@ -3107,18 +3234,13 @@ func (s *MonitoringService) loadQuestionContentCitations(ctx context.Context, te
LEFT JOIN monitoring_article_url_aliases alias
ON alias.tenant_id = cf.tenant_id AND alias.article_id = cf.article_id
WHERE r.tenant_id = $1
AND r.brand_id = $2
AND r.question_id = $3
AND r.question_hash = $4
AND r.collector_type = $5
AND r.status = 'succeeded'
AND r.business_date BETWEEN $6::date AND $7::date
AND cf.article_id IS NOT NULL
AND r.ai_platform_id <> 'qwen'
AND ($8::text[] IS NULL OR r.ai_platform_id = ANY($8))
AND r.id = ANY($2)
GROUP BY cf.article_id, r.ai_platform_id
ORDER BY citation_count DESC, cf.article_id ASC
`, tenantID, brandID, questionID, hashBytes, monitoringCollectorType, fromDate.Format("2006-01-02"), toDate.Format("2006-01-02"), nullableStringArray(platformQueryIDs))
`, tenantID, runIDs)
if err != nil {
return nil, response.ErrInternal(50041, "query_failed", "failed to load content citations")
}
@@ -3326,6 +3448,13 @@ func normalizeTrackingDays(days int) int {
return days
}
func normalizeCitationWindowDays(days int) int {
if days == maxTrackingDays {
return maxTrackingDays
}
return defaultTrackingDays
}
func resolveMonitoringPlatforms(enabled []string) []monitoringPlatformMetadata {
if len(enabled) == 0 {
return nil