fix: enhance monitoring metrics with latest sample aggregation and derived metrics calculations
Backend CI / Backend (push) Successful in 21m2s
Backend CI / Backend (push) Successful in 21m2s
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
@@ -9,6 +10,8 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestDecideHeartbeatPrimaryLease(t *testing.T) {
|
||||
@@ -111,6 +114,42 @@ func TestMonitoringHeartbeatPrimaryLeaseMetrics(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadMonitoringBrandDailyAggregateUsesLatestQuestionPlatformSamples(t *testing.T) {
|
||||
tx := &captureDailyMonitorTaskTx{}
|
||||
|
||||
_, err := (&MonitoringCallbackService{}).loadMonitoringBrandDailyAggregate(
|
||||
context.Background(),
|
||||
tx,
|
||||
3,
|
||||
3,
|
||||
time.Date(2026, 6, 29, 0, 0, 0, 0, monitoringBusinessLocation()),
|
||||
)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Len(t, tx.queryRowSQLs, 5)
|
||||
combined := normalizeSQLWhitespace(strings.Join(tx.queryRowSQLs, " "))
|
||||
assert.Contains(t, combined, "SELECT DISTINCT ON (business_date, question_id, ai_platform_id)")
|
||||
assert.Contains(t, combined, "p.brand_mentioned IS TRUE AND p.sentiment_label = 'positive'")
|
||||
}
|
||||
|
||||
func TestLoadMonitoringPlatformDailyAggregatesUsesLatestQuestionPlatformSamples(t *testing.T) {
|
||||
tx := &captureDailyMonitorTaskTx{}
|
||||
|
||||
_, err := (&MonitoringCallbackService{}).loadMonitoringPlatformDailyAggregates(
|
||||
context.Background(),
|
||||
tx,
|
||||
3,
|
||||
3,
|
||||
time.Date(2026, 6, 29, 0, 0, 0, 0, monitoringBusinessLocation()),
|
||||
[]monitoringPlatformMetadata{{ID: "doubao", Name: "豆包"}},
|
||||
)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.GreaterOrEqual(t, len(tx.querySQLs), 2)
|
||||
combined := normalizeSQLWhitespace(strings.Join(tx.querySQLs, " "))
|
||||
assert.Contains(t, combined, "SELECT DISTINCT ON (business_date, question_id, ai_platform_id)")
|
||||
}
|
||||
|
||||
func TestDecideMonitoringHeartbeatSnapshot(t *testing.T) {
|
||||
now := time.Date(2026, 4, 24, 3, 0, 0, 0, time.UTC)
|
||||
base := monitoringPlatformAccessSnapshotState{
|
||||
|
||||
@@ -2,6 +2,7 @@ package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"runtime"
|
||||
@@ -745,7 +746,12 @@ type captureDailyMonitorTaskTx struct {
|
||||
queryCalled bool
|
||||
querySQL string
|
||||
queryArgs []any
|
||||
commandTag pgconn.CommandTag
|
||||
querySQLs []string
|
||||
queryArgses []any
|
||||
|
||||
queryRowSQLs []string
|
||||
queryRowArgses [][]any
|
||||
commandTag pgconn.CommandTag
|
||||
}
|
||||
|
||||
func (tx *captureDailyMonitorTaskTx) Begin(context.Context) (pgx.Tx, error) { return tx, nil }
|
||||
@@ -773,9 +779,13 @@ func (tx *captureDailyMonitorTaskTx) Query(_ context.Context, sql string, argume
|
||||
tx.queryCalled = true
|
||||
tx.querySQL = sql
|
||||
tx.queryArgs = arguments
|
||||
tx.querySQLs = append(tx.querySQLs, sql)
|
||||
tx.queryArgses = append(tx.queryArgses, arguments...)
|
||||
return emptyDailyMonitorRows{}, nil
|
||||
}
|
||||
func (tx *captureDailyMonitorTaskTx) QueryRow(context.Context, string, ...any) pgx.Row {
|
||||
func (tx *captureDailyMonitorTaskTx) QueryRow(_ context.Context, sql string, arguments ...any) pgx.Row {
|
||||
tx.queryRowSQLs = append(tx.queryRowSQLs, sql)
|
||||
tx.queryRowArgses = append(tx.queryRowArgses, arguments)
|
||||
return boolDailyMonitorRow(true)
|
||||
}
|
||||
func (tx *captureDailyMonitorTaskTx) Conn() *pgx.Conn { return nil }
|
||||
@@ -785,9 +795,18 @@ var _ pgx.Tx = (*captureDailyMonitorTaskTx)(nil)
|
||||
type boolDailyMonitorRow bool
|
||||
|
||||
func (r boolDailyMonitorRow) Scan(dest ...any) error {
|
||||
if len(dest) == 1 {
|
||||
if value, ok := dest[0].(*bool); ok {
|
||||
for _, target := range dest {
|
||||
switch value := target.(type) {
|
||||
case *bool:
|
||||
*value = bool(r)
|
||||
case *int64:
|
||||
*value = 0
|
||||
case *string:
|
||||
*value = ""
|
||||
case *sql.NullTime:
|
||||
*value = sql.NullTime{}
|
||||
case *sql.NullString:
|
||||
*value = sql.NullString{}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
|
||||
@@ -173,12 +173,20 @@ func (s *MonitoringCallbackService) loadMonitoringBrandDailyAggregate(ctx contex
|
||||
|
||||
if err := tx.QueryRow(ctx, `
|
||||
SELECT COUNT(*), MAX(completed_at)
|
||||
FROM question_monitor_runs
|
||||
WHERE tenant_id = $1
|
||||
AND brand_id = $2
|
||||
AND collector_type = $3
|
||||
AND business_date = $4::date
|
||||
AND status = 'succeeded'
|
||||
FROM (
|
||||
SELECT DISTINCT ON (business_date, question_id, ai_platform_id)
|
||||
completed_at,
|
||||
updated_at,
|
||||
created_at,
|
||||
id
|
||||
FROM question_monitor_runs
|
||||
WHERE tenant_id = $1
|
||||
AND brand_id = $2
|
||||
AND collector_type = $3
|
||||
AND business_date = $4::date
|
||||
AND status = 'succeeded'
|
||||
ORDER BY business_date, question_id, ai_platform_id, COALESCE(completed_at, updated_at, created_at) DESC NULLS LAST, id DESC
|
||||
) latest_runs
|
||||
`, tenantID, brandID, monitoringCollectorType, dateText).Scan(&agg.ActualSampleCount, &agg.SnapshotUpdatedAt); err != nil {
|
||||
return agg, monitoringInternalError(50041, "run_projection_lookup_failed", "failed to load monitoring run counts", err)
|
||||
}
|
||||
@@ -188,15 +196,27 @@ func (s *MonitoringCallbackService) loadMonitoringBrandDailyAggregate(ctx contex
|
||||
COUNT(*) FILTER (WHERE p.brand_mentioned IS TRUE)::bigint,
|
||||
COUNT(*) FILTER (WHERE p.brand_mention_position = 'top1')::bigint,
|
||||
COUNT(*) FILTER (WHERE p.first_recommended IS TRUE)::bigint,
|
||||
COUNT(*) FILTER (WHERE p.sentiment_label = 'positive')::bigint
|
||||
FROM question_monitor_runs r
|
||||
COUNT(*) FILTER (WHERE p.brand_mentioned IS TRUE AND p.sentiment_label = 'positive')::bigint
|
||||
FROM (
|
||||
SELECT DISTINCT ON (business_date, question_id, ai_platform_id)
|
||||
id,
|
||||
tenant_id,
|
||||
business_date,
|
||||
question_id,
|
||||
ai_platform_id,
|
||||
completed_at,
|
||||
updated_at,
|
||||
created_at
|
||||
FROM question_monitor_runs
|
||||
WHERE tenant_id = $1
|
||||
AND brand_id = $2
|
||||
AND collector_type = $3
|
||||
AND business_date = $4::date
|
||||
AND status = 'succeeded'
|
||||
ORDER BY business_date, question_id, ai_platform_id, COALESCE(completed_at, updated_at, created_at) DESC NULLS LAST, id DESC
|
||||
) r
|
||||
LEFT JOIN question_monitor_parse_results p
|
||||
ON p.tenant_id = r.tenant_id AND p.run_id = r.id
|
||||
WHERE r.tenant_id = $1
|
||||
AND r.brand_id = $2
|
||||
AND r.collector_type = $3
|
||||
AND r.business_date = $4::date
|
||||
AND r.status = 'succeeded'
|
||||
`, tenantID, brandID, monitoringCollectorType, dateText).Scan(
|
||||
&agg.MentionedCount,
|
||||
&agg.Top1MentionedCount,
|
||||
@@ -280,14 +300,24 @@ func (s *MonitoringCallbackService) loadMonitoringPlatformDailyAggregates(ctx co
|
||||
COUNT(*) FILTER (WHERE p.brand_mentioned IS TRUE)::bigint AS mentioned_count,
|
||||
COUNT(*) FILTER (WHERE p.brand_mention_position = 'top1')::bigint AS top1_mentioned_count,
|
||||
MAX(r.completed_at) AS last_sampled_at
|
||||
FROM question_monitor_runs r
|
||||
FROM (
|
||||
SELECT DISTINCT ON (business_date, question_id, ai_platform_id)
|
||||
id,
|
||||
tenant_id,
|
||||
ai_platform_id,
|
||||
completed_at,
|
||||
updated_at,
|
||||
created_at
|
||||
FROM question_monitor_runs
|
||||
WHERE tenant_id = $1
|
||||
AND brand_id = $2
|
||||
AND collector_type = $3
|
||||
AND business_date = $4::date
|
||||
AND status = 'succeeded'
|
||||
ORDER BY business_date, question_id, ai_platform_id, COALESCE(completed_at, updated_at, created_at) DESC NULLS LAST, id DESC
|
||||
) r
|
||||
LEFT JOIN question_monitor_parse_results p
|
||||
ON p.tenant_id = r.tenant_id AND p.run_id = r.id
|
||||
WHERE r.tenant_id = $1
|
||||
AND r.brand_id = $2
|
||||
AND r.collector_type = $3
|
||||
AND r.business_date = $4::date
|
||||
AND r.status = 'succeeded'
|
||||
GROUP BY r.ai_platform_id
|
||||
`, tenantID, brandID, monitoringCollectorType, dateText)
|
||||
if err != nil {
|
||||
|
||||
@@ -349,6 +349,21 @@ type monitoringDerivedMetrics struct {
|
||||
ByQuestion map[int64]monitoringDerivedRateStats
|
||||
}
|
||||
|
||||
type monitoringDerivedSampleKey struct {
|
||||
Date string
|
||||
PlatformID string
|
||||
QuestionID int64
|
||||
}
|
||||
|
||||
type monitoringDerivedParseSample struct {
|
||||
BusinessDate time.Time
|
||||
PlatformID string
|
||||
QuestionID int64
|
||||
EventAt sql.NullTime
|
||||
RunID int64
|
||||
Summary monitoringAnswerParseSummary
|
||||
}
|
||||
|
||||
type monitoringPlatformMetadata struct {
|
||||
ID string
|
||||
Name string
|
||||
@@ -2239,7 +2254,7 @@ func (stats *monitoringDerivedRateStats) add(summary monitoringAnswerParseSummar
|
||||
if summary.FirstRecommended {
|
||||
stats.FirstRecommend++
|
||||
}
|
||||
if summary.SentimentLabel == "positive" {
|
||||
if summary.BrandMentioned && summary.SentimentLabel == "positive" {
|
||||
stats.PositiveMentions++
|
||||
}
|
||||
}
|
||||
@@ -2264,6 +2279,71 @@ func (stats monitoringDerivedRateStats) positiveMentionRate() *float64 {
|
||||
return divideAsPointer(stats.PositiveMentions, stats.Total)
|
||||
}
|
||||
|
||||
func addMonitoringDerivedParseSample(metrics *monitoringDerivedMetrics, sample monitoringDerivedParseSample) {
|
||||
if metrics == nil {
|
||||
return
|
||||
}
|
||||
|
||||
dateKey := sample.BusinessDate.Format("2006-01-02")
|
||||
dateStats := metrics.ByDate[dateKey]
|
||||
dateStats.add(sample.Summary)
|
||||
metrics.ByDate[dateKey] = dateStats
|
||||
|
||||
platformStatsByDate, ok := metrics.ByDatePlatform[dateKey]
|
||||
if !ok {
|
||||
platformStatsByDate = make(map[string]monitoringDerivedRateStats)
|
||||
metrics.ByDatePlatform[dateKey] = platformStatsByDate
|
||||
}
|
||||
platformStats := platformStatsByDate[sample.PlatformID]
|
||||
platformStats.add(sample.Summary)
|
||||
platformStatsByDate[sample.PlatformID] = platformStats
|
||||
|
||||
questionStatsByDate, ok := metrics.ByDateQuestion[dateKey]
|
||||
if !ok {
|
||||
questionStatsByDate = make(map[int64]monitoringDerivedRateStats)
|
||||
metrics.ByDateQuestion[dateKey] = questionStatsByDate
|
||||
}
|
||||
questionStatsForDate := questionStatsByDate[sample.QuestionID]
|
||||
questionStatsForDate.add(sample.Summary)
|
||||
questionStatsByDate[sample.QuestionID] = questionStatsForDate
|
||||
|
||||
questionStats := metrics.ByQuestion[sample.QuestionID]
|
||||
questionStats.add(sample.Summary)
|
||||
metrics.ByQuestion[sample.QuestionID] = questionStats
|
||||
}
|
||||
|
||||
func monitoringDerivedParseSampleIsNewer(candidate, existing monitoringDerivedParseSample) bool {
|
||||
if candidate.EventAt.Valid && existing.EventAt.Valid && !candidate.EventAt.Time.Equal(existing.EventAt.Time) {
|
||||
return candidate.EventAt.Time.After(existing.EventAt.Time)
|
||||
}
|
||||
if candidate.EventAt.Valid != existing.EventAt.Valid {
|
||||
return candidate.EventAt.Valid
|
||||
}
|
||||
return candidate.RunID > existing.RunID
|
||||
}
|
||||
|
||||
func upsertLatestMonitoringDerivedParseSample(samples map[monitoringDerivedSampleKey]monitoringDerivedParseSample, sample monitoringDerivedParseSample) {
|
||||
if samples == nil {
|
||||
return
|
||||
}
|
||||
key := monitoringDerivedSampleKey{
|
||||
Date: sample.BusinessDate.Format("2006-01-02"),
|
||||
PlatformID: sample.PlatformID,
|
||||
QuestionID: sample.QuestionID,
|
||||
}
|
||||
if existing, ok := samples[key]; !ok || monitoringDerivedParseSampleIsNewer(sample, existing) {
|
||||
samples[key] = sample
|
||||
}
|
||||
}
|
||||
|
||||
func buildMonitoringDerivedMetricsFromLatestSamples(samples map[monitoringDerivedSampleKey]monitoringDerivedParseSample) monitoringDerivedMetrics {
|
||||
metrics := newMonitoringDerivedMetrics()
|
||||
for _, sample := range samples {
|
||||
addMonitoringDerivedParseSample(&metrics, sample)
|
||||
}
|
||||
return metrics
|
||||
}
|
||||
|
||||
func (s *MonitoringService) loadDerivedParseMetrics(
|
||||
ctx context.Context,
|
||||
tenantID, brandID int64,
|
||||
@@ -2280,35 +2360,51 @@ func (s *MonitoringService) loadDerivedParseMetrics(
|
||||
|
||||
rows, err := s.monitoringPool.Query(ctx, `
|
||||
SELECT
|
||||
r.id,
|
||||
r.business_date,
|
||||
r.ai_platform_id,
|
||||
r.question_id,
|
||||
r.event_at,
|
||||
r.raw_answer_text,
|
||||
p.brand_mentioned,
|
||||
p.brand_mention_position,
|
||||
p.first_recommended,
|
||||
p.sentiment_label,
|
||||
p.matched_brand_terms
|
||||
FROM question_monitor_runs r
|
||||
FROM (
|
||||
SELECT DISTINCT ON (business_date, question_id, ai_platform_id)
|
||||
id,
|
||||
tenant_id,
|
||||
business_date,
|
||||
ai_platform_id,
|
||||
question_id,
|
||||
raw_answer_text,
|
||||
COALESCE(completed_at, updated_at, created_at) AS event_at
|
||||
FROM question_monitor_runs
|
||||
WHERE tenant_id = $1
|
||||
AND brand_id = $2
|
||||
AND collector_type = $3
|
||||
AND status = 'succeeded'
|
||||
AND business_date BETWEEN $4::date AND $5::date
|
||||
AND ($6::bigint[] IS NULL OR question_id = ANY($6))
|
||||
AND ($7::text[] IS NULL OR ai_platform_id = ANY($7))
|
||||
ORDER BY business_date, question_id, ai_platform_id, COALESCE(completed_at, updated_at, created_at) DESC NULLS LAST, id DESC
|
||||
) r
|
||||
LEFT JOIN question_monitor_parse_results p
|
||||
ON p.tenant_id = r.tenant_id AND p.run_id = r.id
|
||||
WHERE r.tenant_id = $1
|
||||
AND 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))
|
||||
`, tenantID, brandID, monitoringCollectorType, startDate.Format("2006-01-02"), endDate.Format("2006-01-02"), nullableInt64Array(questionIDs), nullableStringArray(platformQueryIDs))
|
||||
if err != nil {
|
||||
return metrics, response.ErrInternal(50041, "query_failed", "failed to load monitoring derived parse metrics")
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
samples := make(map[monitoringDerivedSampleKey]monitoringDerivedParseSample)
|
||||
for rows.Next() {
|
||||
var runID int64
|
||||
var businessDate time.Time
|
||||
var platformID string
|
||||
var questionID int64
|
||||
var eventAt sql.NullTime
|
||||
var answerText sql.NullString
|
||||
var brandMentioned sql.NullBool
|
||||
var brandMentionPosition sql.NullString
|
||||
@@ -2316,9 +2412,11 @@ func (s *MonitoringService) loadDerivedParseMetrics(
|
||||
var sentimentLabel sql.NullString
|
||||
var matchedBrandTermsJSON []byte
|
||||
if scanErr := rows.Scan(
|
||||
&runID,
|
||||
&businessDate,
|
||||
&platformID,
|
||||
&questionID,
|
||||
&eventAt,
|
||||
&answerText,
|
||||
&brandMentioned,
|
||||
&brandMentionPosition,
|
||||
@@ -2345,39 +2443,21 @@ func (s *MonitoringService) loadDerivedParseMetrics(
|
||||
},
|
||||
)
|
||||
|
||||
dateKey := businessDate.Format("2006-01-02")
|
||||
dateStats := metrics.ByDate[dateKey]
|
||||
dateStats.add(summary)
|
||||
metrics.ByDate[dateKey] = dateStats
|
||||
|
||||
platformStatsByDate, ok := metrics.ByDatePlatform[dateKey]
|
||||
if !ok {
|
||||
platformStatsByDate = make(map[string]monitoringDerivedRateStats)
|
||||
metrics.ByDatePlatform[dateKey] = platformStatsByDate
|
||||
}
|
||||
platformStats := platformStatsByDate[platformID]
|
||||
platformStats.add(summary)
|
||||
platformStatsByDate[platformID] = platformStats
|
||||
|
||||
questionStatsByDate, ok := metrics.ByDateQuestion[dateKey]
|
||||
if !ok {
|
||||
questionStatsByDate = make(map[int64]monitoringDerivedRateStats)
|
||||
metrics.ByDateQuestion[dateKey] = questionStatsByDate
|
||||
}
|
||||
questionStatsForDate := questionStatsByDate[questionID]
|
||||
questionStatsForDate.add(summary)
|
||||
questionStatsByDate[questionID] = questionStatsForDate
|
||||
|
||||
questionStats := metrics.ByQuestion[questionID]
|
||||
questionStats.add(summary)
|
||||
metrics.ByQuestion[questionID] = questionStats
|
||||
upsertLatestMonitoringDerivedParseSample(samples, monitoringDerivedParseSample{
|
||||
BusinessDate: businessDate,
|
||||
PlatformID: platformID,
|
||||
QuestionID: questionID,
|
||||
EventAt: eventAt,
|
||||
RunID: runID,
|
||||
Summary: summary,
|
||||
})
|
||||
}
|
||||
|
||||
if err := rows.Err(); err != nil {
|
||||
return metrics, response.ErrInternal(50041, "scan_failed", "failed to iterate monitoring derived metrics")
|
||||
}
|
||||
|
||||
return metrics, nil
|
||||
return buildMonitoringDerivedMetricsFromLatestSamples(samples), nil
|
||||
}
|
||||
|
||||
func decodeMonitoringMatchedBrandTerms(raw []byte) []string {
|
||||
@@ -2608,18 +2688,30 @@ func (s *MonitoringService) loadOverviewRunAggregate(
|
||||
COUNT(*) FILTER (WHERE p.brand_mentioned IS TRUE)::bigint AS mentioned_count,
|
||||
COUNT(*) FILTER (WHERE p.brand_mention_position = 'top1')::bigint AS top1_mentioned_count,
|
||||
COUNT(*) FILTER (WHERE p.first_recommended IS TRUE)::bigint AS first_recommended_count,
|
||||
COUNT(*) FILTER (WHERE p.sentiment_label = 'positive')::bigint AS positive_mentioned_count,
|
||||
COUNT(*) FILTER (WHERE p.brand_mentioned IS TRUE AND p.sentiment_label = 'positive')::bigint AS positive_mentioned_count,
|
||||
MAX(r.completed_at) AS last_sampled_at
|
||||
FROM question_monitor_runs r
|
||||
FROM (
|
||||
SELECT DISTINCT ON (business_date, question_id, ai_platform_id)
|
||||
id,
|
||||
tenant_id,
|
||||
business_date,
|
||||
question_id,
|
||||
ai_platform_id,
|
||||
completed_at,
|
||||
updated_at,
|
||||
created_at
|
||||
FROM question_monitor_runs
|
||||
WHERE tenant_id = $1
|
||||
AND brand_id = $2
|
||||
AND collector_type = $3
|
||||
AND status = 'succeeded'
|
||||
AND business_date BETWEEN $4::date AND $5::date
|
||||
AND question_id = ANY($6)
|
||||
AND ($7::text[] IS NULL OR ai_platform_id = ANY($7))
|
||||
ORDER BY business_date, question_id, ai_platform_id, COALESCE(completed_at, updated_at, created_at) DESC NULLS LAST, id DESC
|
||||
) r
|
||||
LEFT JOIN question_monitor_parse_results p
|
||||
ON p.tenant_id = r.tenant_id AND p.run_id = r.id
|
||||
WHERE r.tenant_id = $1
|
||||
AND r.brand_id = $2
|
||||
AND r.collector_type = $3
|
||||
AND r.status = 'succeeded'
|
||||
AND r.business_date BETWEEN $4::date AND $5::date
|
||||
AND r.question_id = ANY($6)
|
||||
AND ($7::text[] IS NULL OR r.ai_platform_id = ANY($7))
|
||||
GROUP BY r.business_date
|
||||
ORDER BY r.business_date DESC
|
||||
`, tenantID, brandID, monitoringCollectorType, startDate.Format("2006-01-02"), endDate.Format("2006-01-02"), questionIDs, nullableStringArray(platformIDs))
|
||||
@@ -3136,15 +3228,25 @@ func (s *MonitoringService) loadPlatformBreakdownFromRuns(ctx context.Context, t
|
||||
COUNT(*) FILTER (WHERE p.brand_mentioned IS TRUE)::bigint AS mentioned_count,
|
||||
COUNT(*) FILTER (WHERE p.brand_mention_position = 'top1')::bigint AS top1_mentioned_count,
|
||||
MAX(r.completed_at) AS last_sampled_at
|
||||
FROM question_monitor_runs r
|
||||
FROM (
|
||||
SELECT DISTINCT ON (business_date, question_id, ai_platform_id)
|
||||
id,
|
||||
tenant_id,
|
||||
ai_platform_id,
|
||||
completed_at,
|
||||
updated_at,
|
||||
created_at
|
||||
FROM question_monitor_runs
|
||||
WHERE tenant_id = $1
|
||||
AND brand_id = $2
|
||||
AND collector_type = $3
|
||||
AND business_date = $4::date
|
||||
AND status = 'succeeded'
|
||||
AND question_id = ANY($5)
|
||||
ORDER BY business_date, question_id, ai_platform_id, COALESCE(completed_at, updated_at, created_at) DESC NULLS LAST, id DESC
|
||||
) r
|
||||
LEFT JOIN question_monitor_parse_results p
|
||||
ON p.tenant_id = r.tenant_id AND p.run_id = r.id
|
||||
WHERE r.tenant_id = $1
|
||||
AND r.brand_id = $2
|
||||
AND r.collector_type = $3
|
||||
AND r.business_date = $4::date
|
||||
AND r.status = 'succeeded'
|
||||
AND r.question_id = ANY($5)
|
||||
GROUP BY r.ai_platform_id
|
||||
`, tenantID, brandID, monitoringCollectorType, dateKey, questionIDs)
|
||||
if err != nil {
|
||||
@@ -3192,6 +3294,7 @@ func (s *MonitoringService) loadPlatformBreakdownFromRuns(ctx context.Context, t
|
||||
if platformStats, ok := derived.ByDatePlatform[dateKey][platform.ID]; ok && platformStats.hasSamples() {
|
||||
item.MentionRate = pointerFloat64ToNull(platformStats.mentionRate())
|
||||
item.Top1MentionRate = pointerFloat64ToNull(platformStats.top1MentionRate())
|
||||
item.ActualSampleCount = platformStats.Total
|
||||
}
|
||||
breakdown = append(breakdown, MonitoringPlatformDaily{
|
||||
AIPlatformID: platform.ID,
|
||||
@@ -3236,11 +3339,15 @@ func (s *MonitoringService) loadHotQuestions(
|
||||
platformQueryIDs := monitoringPlatformQueryIDs(aiPlatformID)
|
||||
|
||||
rows, err := s.monitoringPool.Query(ctx, `
|
||||
WITH stats AS (
|
||||
SELECT
|
||||
WITH latest_runs AS (
|
||||
SELECT DISTINCT ON (r.question_id, r.ai_platform_id)
|
||||
r.question_id,
|
||||
AVG(CASE WHEN p.brand_mentioned IS TRUE THEN 1 ELSE 0 END)::double precision AS mention_rate,
|
||||
MAX(r.completed_at) AS last_sampled_at
|
||||
r.ai_platform_id,
|
||||
r.question_hash,
|
||||
r.completed_at,
|
||||
r.updated_at,
|
||||
r.created_at,
|
||||
p.brand_mentioned
|
||||
FROM question_monitor_runs r
|
||||
LEFT JOIN question_monitor_parse_results p
|
||||
ON p.tenant_id = r.tenant_id AND p.run_id = r.id
|
||||
@@ -3251,21 +3358,22 @@ func (s *MonitoringService) loadHotQuestions(
|
||||
AND r.business_date = $4::date
|
||||
AND r.question_id = ANY($5)
|
||||
AND ($6::text[] IS NULL OR r.ai_platform_id = ANY($6))
|
||||
GROUP BY r.question_id
|
||||
ORDER BY r.question_id, r.ai_platform_id, COALESCE(r.completed_at, r.updated_at, r.created_at) DESC NULLS LAST, r.id DESC
|
||||
),
|
||||
stats AS (
|
||||
SELECT
|
||||
question_id,
|
||||
AVG(CASE WHEN brand_mentioned IS TRUE THEN 1 ELSE 0 END)::double precision AS mention_rate,
|
||||
MAX(completed_at) AS last_sampled_at
|
||||
FROM latest_runs
|
||||
GROUP BY question_id
|
||||
),
|
||||
latest_hash AS (
|
||||
SELECT DISTINCT ON (r.question_id)
|
||||
r.question_id,
|
||||
r.question_hash
|
||||
FROM question_monitor_runs r
|
||||
WHERE r.tenant_id = $1
|
||||
AND r.brand_id = $2
|
||||
AND r.collector_type = $3
|
||||
AND r.status = 'succeeded'
|
||||
AND r.business_date = $4::date
|
||||
AND r.question_id = ANY($5)
|
||||
AND ($6::text[] IS NULL OR r.ai_platform_id = ANY($6))
|
||||
ORDER BY r.question_id, r.business_date DESC, COALESCE(r.completed_at, r.updated_at, r.created_at) DESC NULLS LAST, r.id DESC
|
||||
FROM latest_runs r
|
||||
ORDER BY r.question_id, COALESCE(r.completed_at, r.updated_at, r.created_at) DESC NULLS LAST
|
||||
)
|
||||
SELECT
|
||||
s.question_id,
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -234,3 +235,73 @@ func TestApplyDerivedRatesToOverviewSnapshotUsesAllSamplesForMentionRate(t *test
|
||||
assert.InDelta(t, 0.25, snapshot.FirstRecommendRate.Float64, 0.0001)
|
||||
assert.InDelta(t, 0.25, snapshot.PositiveMentionRate.Float64, 0.0001)
|
||||
}
|
||||
|
||||
func TestBuildMonitoringDerivedMetricsUsesLatestSamplePerQuestionPlatform(t *testing.T) {
|
||||
day := time.Date(2026, 6, 29, 0, 0, 0, 0, monitoringBusinessLocation())
|
||||
questionID := int64(177)
|
||||
samples := map[monitoringDerivedSampleKey]monitoringDerivedParseSample{}
|
||||
|
||||
upsertLatestMonitoringDerivedParseSample(samples, monitoringDerivedParseSample{
|
||||
BusinessDate: day,
|
||||
PlatformID: "doubao",
|
||||
QuestionID: questionID,
|
||||
EventAt: sql.NullTime{Time: day.Add(10 * time.Hour), Valid: true},
|
||||
RunID: 2507,
|
||||
Summary: monitoringAnswerParseSummary{
|
||||
BrandMentioned: true,
|
||||
BrandMentionPosition: "mentioned",
|
||||
},
|
||||
})
|
||||
upsertLatestMonitoringDerivedParseSample(samples, monitoringDerivedParseSample{
|
||||
BusinessDate: day,
|
||||
PlatformID: "doubao",
|
||||
QuestionID: questionID,
|
||||
EventAt: sql.NullTime{Time: day.Add(11 * time.Hour), Valid: true},
|
||||
RunID: 2628,
|
||||
Summary: monitoringAnswerParseSummary{
|
||||
BrandMentioned: false,
|
||||
BrandMentionPosition: "not_mentioned",
|
||||
},
|
||||
})
|
||||
upsertLatestMonitoringDerivedParseSample(samples, monitoringDerivedParseSample{
|
||||
BusinessDate: day,
|
||||
PlatformID: "qwen",
|
||||
QuestionID: questionID,
|
||||
EventAt: sql.NullTime{Time: day.Add(10*time.Hour + 30*time.Minute), Valid: true},
|
||||
RunID: 2510,
|
||||
Summary: monitoringAnswerParseSummary{
|
||||
BrandMentioned: true,
|
||||
BrandMentionPosition: "mentioned",
|
||||
},
|
||||
})
|
||||
|
||||
metrics := buildMonitoringDerivedMetricsFromLatestSamples(samples)
|
||||
stats := metrics.ByDateQuestion["2026-06-29"][questionID]
|
||||
|
||||
require.True(t, stats.hasSamples())
|
||||
assert.Equal(t, int64(2), stats.Total)
|
||||
assert.Equal(t, int64(1), stats.MentionedCount)
|
||||
require.NotNil(t, stats.mentionRate())
|
||||
assert.InDelta(t, 0.5, *stats.mentionRate(), 0.0001)
|
||||
}
|
||||
|
||||
func TestMonitoringDerivedPositiveMentionRequiresBrandMention(t *testing.T) {
|
||||
var stats monitoringDerivedRateStats
|
||||
|
||||
stats.add(monitoringAnswerParseSummary{
|
||||
BrandMentioned: false,
|
||||
BrandMentionPosition: "not_mentioned",
|
||||
SentimentLabel: "positive",
|
||||
})
|
||||
stats.add(monitoringAnswerParseSummary{
|
||||
BrandMentioned: true,
|
||||
BrandMentionPosition: "mentioned",
|
||||
SentimentLabel: "positive",
|
||||
})
|
||||
|
||||
assert.Equal(t, int64(2), stats.Total)
|
||||
assert.Equal(t, int64(1), stats.MentionedCount)
|
||||
assert.Equal(t, int64(1), stats.PositiveMentions)
|
||||
require.NotNil(t, stats.positiveMentionRate())
|
||||
assert.InDelta(t, 0.5, *stats.positiveMentionRate(), 0.0001)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user