From 2af7901e000e3ed884ed4b4a8ee0aae4753c3970 Mon Sep 17 00:00:00 2001 From: xucroyuri Date: Thu, 2 Jul 2026 07:23:18 +0800 Subject: [PATCH] feat: add breaking intelligence v2 calibration --- .../20260702_breaking_v2_calibration.sql | 21 ++ frontend/cloudflare/db/schema.sql | 11 +- frontend/cloudflare/workers/api/bootstrap.ts | 6 +- frontend/cloudflare/workers/api/news.ts | 14 +- frontend/cloudflare/workers/api/webhook.ts | 36 ++- .../workers/lib/breaking-calibration.ts | 239 ++++++++++++++++++ frontend/cloudflare/workers/lib/contracts.ts | 8 + .../workers/lib/public-news-query.ts | 42 ++- .../workers/lib/public-read-snapshots.ts | 6 +- frontend/cloudflare/workers/lib/scheduled.ts | 11 + frontend/public/src/lib/api.test.ts | 8 + frontend/public/src/lib/feed-state.test.ts | 7 +- frontend/public/src/lib/feed-state.ts | 2 +- frontend/public/src/types/public-news.ts | 4 + src/news_sentry/api/schemas.py | 21 ++ src/news_sentry/core/breaking_scoring.py | 85 ++++++- .../core/collector_config_utils.py | 35 ++- src/news_sentry/core/public_translation.py | 3 + tests/unit/test_breaking_scoring.py | 59 +++++ tests/unit/test_cloudflare_native_config.py | 90 +++++-- ...est_cloudflare_refresh_public_snapshots.py | 44 ++++ tools/cloudflare_refresh_public_snapshots.py | 56 +++- 22 files changed, 754 insertions(+), 54 deletions(-) create mode 100644 frontend/cloudflare/db/migrations/20260702_breaking_v2_calibration.sql create mode 100644 frontend/cloudflare/workers/lib/breaking-calibration.ts diff --git a/frontend/cloudflare/db/migrations/20260702_breaking_v2_calibration.sql b/frontend/cloudflare/db/migrations/20260702_breaking_v2_calibration.sql new file mode 100644 index 00000000..648bdb74 --- /dev/null +++ b/frontend/cloudflare/db/migrations/20260702_breaking_v2_calibration.sql @@ -0,0 +1,21 @@ +ALTER TABLE events ADD COLUMN breaking_raw_score REAL; +ALTER TABLE events ADD COLUMN breaking_percentile REAL; +ALTER TABLE events ADD COLUMN breaking_calibrated_score REAL; +ALTER TABLE events ADD COLUMN breaking_adversarial_flags TEXT DEFAULT '{}'; + +UPDATE events +SET + breaking_raw_score = COALESCE(breaking_raw_score, breaking_score), + breaking_percentile = COALESCE(breaking_percentile, breaking_score), + breaking_calibrated_score = COALESCE(breaking_calibrated_score, breaking_score), + breaking_score_version = COALESCE(breaking_score_version, 'breaking-v2.0') +WHERE breaking_score IS NOT NULL; + +DROP INDEX IF EXISTS idx_events_public_breaking; +CREATE INDEX IF NOT EXISTS idx_events_public_breaking +ON events( + pipeline_stage, + breaking_calibrated_score DESC, + breaking_score DESC, + published_at DESC +); diff --git a/frontend/cloudflare/db/schema.sql b/frontend/cloudflare/db/schema.sql index 19988881..f885d460 100644 --- a/frontend/cloudflare/db/schema.sql +++ b/frontend/cloudflare/db/schema.sql @@ -36,11 +36,15 @@ CREATE TABLE IF NOT EXISTS events ( discussion_count INTEGER, classification TEXT DEFAULT '{}', extra TEXT DEFAULT '{}', + breaking_raw_score REAL, + breaking_percentile REAL, + breaking_calibrated_score REAL, breaking_score REAL, breaking_label TEXT, breaking_reason TEXT, breaking_confidence INTEGER, breaking_dimensions TEXT DEFAULT '{}', + breaking_adversarial_flags TEXT DEFAULT '{}', breaking_score_version TEXT, target_timezone TEXT DEFAULT 'UTC', published_at_local TEXT, @@ -56,7 +60,12 @@ CREATE INDEX IF NOT EXISTS idx_events_pipeline_stage ON events(pipeline_stage); CREATE INDEX IF NOT EXISTS idx_events_source_id ON events(source_id); CREATE INDEX IF NOT EXISTS idx_events_value_label ON events(value_label); CREATE INDEX IF NOT EXISTS idx_events_public_featured ON events(pipeline_stage, value_score DESC, published_at DESC); -CREATE INDEX IF NOT EXISTS idx_events_public_breaking ON events(pipeline_stage, breaking_score DESC, published_at DESC); +CREATE INDEX IF NOT EXISTS idx_events_public_breaking ON events( + pipeline_stage, + breaking_calibrated_score DESC, + breaking_score DESC, + published_at DESC +); CREATE TABLE IF NOT EXISTS event_localizations ( event_id TEXT NOT NULL, diff --git a/frontend/cloudflare/workers/api/bootstrap.ts b/frontend/cloudflare/workers/api/bootstrap.ts index 9418729b..14b87ac8 100644 --- a/frontend/cloudflare/workers/api/bootstrap.ts +++ b/frontend/cloudflare/workers/api/bootstrap.ts @@ -20,6 +20,7 @@ import { publicNewsOrderBy, publicNewsLocaleJoin, publicNewsSelectColumnsForLocale, + pollAfterMsForPublicNews, rowToPublicNewsItem, } from "../lib/public-news-query"; import { @@ -177,11 +178,12 @@ export async function handleBootstrap( // 构建 news const newsRows = (newsResult.results || []) as NewsRow[]; + const newsItems = newsRows.map((r) => rowToPublicNewsItem(r)); const newsFeed: PublicNewsFeedResponse = { - items: newsRows.map((r) => rowToPublicNewsItem(r)), + items: newsItems, latestCursor: null, nextCursor: null, - pollAfterMs: featured ? 30000 : 60000, + pollAfterMs: pollAfterMsForPublicNews(featured, newsItems), hasNewer: false, total: newsCountResult?.total ?? newsRows.length, }; diff --git a/frontend/cloudflare/workers/api/news.ts b/frontend/cloudflare/workers/api/news.ts index 38ec2ef3..9e46ce02 100644 --- a/frontend/cloudflare/workers/api/news.ts +++ b/frontend/cloudflare/workers/api/news.ts @@ -18,6 +18,7 @@ import { publicNewsOrderBy, publicNewsLocaleJoin, publicNewsSelectColumnsForLocale, + pollAfterMsForPublicNews, rowToPublicNewsItem, } from "../lib/public-news-query"; import { @@ -46,6 +47,7 @@ interface CursorRow { published_at: string; value_score: number | null; breaking_score: number | null; + breaking_calibrated_score: number | null; } function withLocaleHeaders(response: Response, locale: string): Response { @@ -69,22 +71,22 @@ async function buildCursorFilter( if (!eventId) return { sql: "", bindings: [] }; const cursor = await db - .prepare("SELECT event_id, published_at, value_score, breaking_score FROM events WHERE event_id = ?") + .prepare("SELECT event_id, published_at, value_score, breaking_score, breaking_calibrated_score FROM events WHERE event_id = ?") .bind(eventId) .first(); if (!cursor) return { sql: "", bindings: [] }; if (featured) { - const score = cursor.breaking_score ?? cursor.value_score ?? -1; + const score = cursor.breaking_calibrated_score ?? cursor.breaking_score ?? cursor.value_score ?? -1; const scoreOp = mode === "before" ? "<" : ">"; const timeOp = mode === "before" ? "<" : ">"; const idOp = mode === "before" ? "<" : ">"; return { sql: ` AND ( - COALESCE(breaking_score, value_score, -1) ${scoreOp} ? - OR (COALESCE(breaking_score, value_score, -1) = ? AND events.published_at ${timeOp} ?) - OR (COALESCE(breaking_score, value_score, -1) = ? AND events.published_at = ? AND events.event_id ${idOp} ?) + COALESCE(breaking_calibrated_score, breaking_score, value_score, -1) ${scoreOp} ? + OR (COALESCE(breaking_calibrated_score, breaking_score, value_score, -1) = ? AND events.published_at ${timeOp} ?) + OR (COALESCE(breaking_calibrated_score, breaking_score, value_score, -1) = ? AND events.published_at = ? AND events.event_id ${idOp} ?) ) `, bindings: [score, score, cursor.published_at, score, cursor.published_at, cursor.event_id], @@ -216,7 +218,7 @@ export async function handleNewsFeed( items, latestCursor, nextCursor, - pollAfterMs: featured ? 30000 : 60000, + pollAfterMs: pollAfterMsForPublicNews(featured, items), hasNewer: Boolean(sinceCursor && items.length > 0), total: totalResult?.total ?? rows.length, }; diff --git a/frontend/cloudflare/workers/api/webhook.ts b/frontend/cloudflare/workers/api/webhook.ts index e03143d8..1de6d1cc 100644 --- a/frontend/cloudflare/workers/api/webhook.ts +++ b/frontend/cloudflare/workers/api/webhook.ts @@ -8,6 +8,7 @@ import type { WebhookResponse, ImportResponse, ImportEventItem } from "../lib/contracts"; import { internalError } from "../lib/errors"; +import { refreshBreakingScoreStats } from "../lib/breaking-calibration"; import { refreshPublicReadSnapshots } from "../lib/public-read-snapshots"; type ImportEventWithId = ImportEventItem & { @@ -66,11 +67,15 @@ function hasProjectionUpdate(item: ImportEventWithId): boolean { jsonArrayText(item.image_urls) || jsonArrayText(item.entities) || jsonObjectText(item.classification) || + typeof item.breaking_raw_score === "number" || + typeof item.breaking_percentile === "number" || + typeof item.breaking_calibrated_score === "number" || typeof item.breaking_score === "number" || nonEmptyText(item.breaking_label) || nonEmptyText(item.breaking_reason) || typeof item.breaking_confidence === "number" || jsonObjectText(item.breaking_dimensions) || + jsonObjectText(item.breaking_adversarial_flags) || nonEmptyText(item.target_timezone) || nonEmptyText(item.published_at_local) || (Array.isArray(item.localizations) && item.localizations.length > 0), @@ -204,11 +209,15 @@ export async function importEventsToD1( value_score = COALESCE(?, value_score), china_relevance_label = COALESCE(NULLIF(?, ''), china_relevance_label), classification = COALESCE(?, classification), + breaking_raw_score = COALESCE(?, breaking_raw_score), + breaking_percentile = COALESCE(?, breaking_percentile), + breaking_calibrated_score = COALESCE(?, breaking_calibrated_score), breaking_score = COALESCE(?, breaking_score), breaking_label = COALESCE(NULLIF(?, ''), breaking_label), breaking_reason = COALESCE(NULLIF(?, ''), breaking_reason), breaking_confidence = COALESCE(?, breaking_confidence), breaking_dimensions = COALESCE(?, breaking_dimensions), + breaking_adversarial_flags = COALESCE(?, breaking_adversarial_flags), breaking_score_version = COALESCE(NULLIF(?, ''), breaking_score_version), target_timezone = COALESCE(NULLIF(?, ''), target_timezone), published_at_local = COALESCE(NULLIF(?, ''), published_at_local), @@ -231,11 +240,15 @@ export async function importEventsToD1( typeof item.value_score === "number" ? item.value_score : null, nonEmptyText(item.china_relevance_label), jsonObjectText(item.classification), + typeof item.breaking_raw_score === "number" ? item.breaking_raw_score : null, + typeof item.breaking_percentile === "number" ? item.breaking_percentile : null, + typeof item.breaking_calibrated_score === "number" ? item.breaking_calibrated_score : null, typeof item.breaking_score === "number" ? item.breaking_score : null, nonEmptyText(item.breaking_label), nonEmptyText(item.breaking_reason), typeof item.breaking_confidence === "number" ? item.breaking_confidence : null, jsonObjectText(item.breaking_dimensions), + jsonObjectText(item.breaking_adversarial_flags), nonEmptyText(item.breaking_score_version), nonEmptyText(item.target_timezone), nonEmptyText(item.published_at_local), @@ -260,9 +273,11 @@ export async function importEventsToD1( image_urls, tags, issue_tags, related_tags, region_tags, entities, language, pipeline_stage, value_label, value_score, china_relevance_label, classification - , breaking_score, breaking_label, breaking_reason, breaking_confidence, - breaking_dimensions, breaking_score_version, target_timezone, published_at_local - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + , breaking_raw_score, breaking_percentile, breaking_calibrated_score, + breaking_score, breaking_label, breaking_reason, breaking_confidence, + breaking_dimensions, breaking_adversarial_flags, breaking_score_version, + target_timezone, published_at_local + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(event_id) DO UPDATE SET target_id=excluded.target_id, target_label=excluded.target_label, @@ -297,11 +312,21 @@ export async function importEventsToD1( events.china_relevance_label ), classification=COALESCE(NULLIF(excluded.classification, '{}'), events.classification), + breaking_raw_score=COALESCE(excluded.breaking_raw_score, events.breaking_raw_score), + breaking_percentile=COALESCE(excluded.breaking_percentile, events.breaking_percentile), + breaking_calibrated_score=COALESCE( + excluded.breaking_calibrated_score, + events.breaking_calibrated_score + ), breaking_score=COALESCE(excluded.breaking_score, events.breaking_score), breaking_label=COALESCE(NULLIF(excluded.breaking_label, ''), events.breaking_label), breaking_reason=COALESCE(NULLIF(excluded.breaking_reason, ''), events.breaking_reason), breaking_confidence=COALESCE(excluded.breaking_confidence, events.breaking_confidence), breaking_dimensions=COALESCE(NULLIF(excluded.breaking_dimensions, '{}'), events.breaking_dimensions), + breaking_adversarial_flags=COALESCE( + NULLIF(excluded.breaking_adversarial_flags, '{}'), + events.breaking_adversarial_flags + ), breaking_score_version=COALESCE( NULLIF(excluded.breaking_score_version, ''), events.breaking_score_version @@ -342,11 +367,15 @@ export async function importEventsToD1( item.value_score ?? null, item.china_relevance_label || "未知", jsonText(item.classification, {}), + item.breaking_raw_score ?? item.breaking_score ?? item.value_score ?? null, + item.breaking_percentile ?? item.breaking_score ?? item.value_score ?? null, + item.breaking_calibrated_score ?? item.breaking_score ?? item.value_score ?? null, item.breaking_score ?? item.value_score ?? null, item.breaking_label || null, item.breaking_reason || null, item.breaking_confidence ?? null, jsonText(item.breaking_dimensions, {}), + jsonText(item.breaking_adversarial_flags, {}), item.breaking_score_version || null, item.target_timezone || "UTC", item.published_at_local || null, @@ -407,6 +436,7 @@ export async function handleImport( const body = await importEventsToD1(db, events); if (body.imported > 0 || body.updated > 0) { try { + await refreshBreakingScoreStats(db); await refreshPublicReadSnapshots(db); } catch (error) { console.warn("public snapshot refresh after import failed:", error); diff --git a/frontend/cloudflare/workers/lib/breaking-calibration.ts b/frontend/cloudflare/workers/lib/breaking-calibration.ts new file mode 100644 index 00000000..0467b4a0 --- /dev/null +++ b/frontend/cloudflare/workers/lib/breaking-calibration.ts @@ -0,0 +1,239 @@ +import { BREAKING_SCORE_VERSION } from "./public-news-query"; + +interface BreakingScoreRow { + event_id: string; + target_id: string; + published_at: string; + issue_tags: string | null; + related_tags: string | null; + breaking_raw_score: number | null; + breaking_percentile: number | null; + breaking_calibrated_score: number | null; + breaking_score: number | null; + breaking_label: string | null; + breaking_confidence: number | null; + breaking_score_version: string | null; +} + +interface ScoreStats { + scopeKey: string; + windowDays: number; + values: number[]; + mean: number; + stddev: number; + p50: number; + p75: number; + p90: number; + p95: number; +} + +function clampScore(value: unknown, fallback = 0): number { + const parsed = typeof value === "number" ? value : Number.parseFloat(String(value ?? "")); + if (!Number.isFinite(parsed)) return fallback; + return Math.max(0, Math.min(100, Math.round(parsed))); +} + +function parseJsonArray(raw: string | null): string[] { + if (!raw) return []; + try { + const parsed = JSON.parse(raw); + return Array.isArray(parsed) ? parsed.filter((item): item is string => typeof item === "string") : []; + } catch { + return []; + } +} + +function percentile(values: number[], p: number): number { + if (values.length === 0) return 0; + if (values.length === 1) return values[0] ?? 0; + const sorted = [...values].sort((left, right) => left - right); + const rank = (p / 100) * (sorted.length - 1); + const lower = Math.floor(rank); + const upper = Math.ceil(rank); + if (lower === upper) return sorted[lower] ?? 0; + const lowerValue = sorted[lower] ?? 0; + const upperValue = sorted[upper] ?? lowerValue; + return lowerValue + (upperValue - lowerValue) * (rank - lower); +} + +function buildStats(scopeKey: string, windowDays: number, values: number[]): ScoreStats { + const mean = values.length ? values.reduce((sum, value) => sum + value, 0) / values.length : 0; + const variance = values.length + ? values.reduce((sum, value) => sum + (value - mean) ** 2, 0) / values.length + : 0; + return { + scopeKey, + windowDays, + values, + mean, + stddev: Math.sqrt(variance), + p50: percentile(values, 50), + p75: percentile(values, 75), + p90: percentile(values, 90), + p95: percentile(values, 95), + }; +} + +function percentileForScore(score: number, stats: ScoreStats): number { + if (stats.values.length < 30) return score; + const sorted = [...stats.values].sort((left, right) => left - right); + let count = 0; + for (const value of sorted) { + if (value <= score) count += 1; + } + return Math.max(0, Math.min(100, (count / sorted.length) * 100)); +} + +function labelForScore(score: number, percentileValue: number, confidence: number): string { + if (score >= 85 && percentileValue >= 95 && confidence >= 70) return "flash"; + if (score >= 72 && percentileValue >= 90 && confidence >= 60) return "breaking"; + if (score >= 52 || percentileValue >= 75) return "watch"; + return "timeline"; +} + +function addScore(scopes: Map, scopeKey: string, score: number): void { + const values = scopes.get(scopeKey) ?? []; + values.push(score); + scopes.set(scopeKey, values); +} + +function scopeKeysForRow(row: BreakingScoreRow): string[] { + return [ + "global", + `target:${row.target_id}`, + ...parseJsonArray(row.issue_tags).map((tag) => `issue:${tag}`), + ...parseJsonArray(row.related_tags).map((tag) => `related:${tag}`), + ]; +} + +function chooseStats(row: BreakingScoreRow, stats: Map): ScoreStats | null { + const targetStats = stats.get(`target:${row.target_id}`); + if (targetStats && targetStats.values.length >= 30) return targetStats; + return stats.get("global") ?? null; +} + +function scoreChanged(current: number | null, next: number, precision = 0): boolean { + if (current === null || current === undefined) return true; + const factor = 10 ** precision; + return Math.round(Number(current) * factor) !== Math.round(next * factor); +} + +export async function refreshBreakingScoreStats( + db: D1Database, + options: { windowDays?: number; limit?: number } = {}, +): Promise> { + const windowDays = Math.max(1, Math.trunc(options.windowDays ?? 30)); + const limit = Math.max(100, Math.trunc(options.limit ?? 5000)); + const result = await db + .prepare( + `SELECT event_id, target_id, published_at, issue_tags, related_tags, + breaking_raw_score, breaking_percentile, breaking_calibrated_score, + breaking_score, breaking_label, breaking_confidence, breaking_score_version + FROM events + WHERE pipeline_stage = 'drafts' + AND COALESCE(breaking_raw_score, breaking_score) IS NOT NULL + AND datetime(COALESCE(published_at, collected_at, created_at)) >= datetime('now', ?) + ORDER BY published_at DESC + LIMIT ?`, + ) + .bind(`-${windowDays} days`, limit) + .all(); + const rows = result.results || []; + if (rows.length === 0) { + return { status: "empty", window_days: windowDays, updated_events: 0, stats_scopes: 0 }; + } + + const scopes = new Map(); + for (const row of rows) { + const score = clampScore(row.breaking_raw_score ?? row.breaking_score); + for (const scopeKey of scopeKeysForRow(row)) { + addScore(scopes, scopeKey, score); + } + } + + const stats = new Map(); + for (const [scopeKey, values] of scopes.entries()) { + const next = buildStats(scopeKey, windowDays, values); + stats.set(scopeKey, next); + await db + .prepare( + `INSERT INTO breaking_score_stats ( + scope_key, window_days, mean_score, stddev_score, + p50, p75, p90, p95, sample_count, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now')) + ON CONFLICT(scope_key) DO UPDATE SET + window_days=excluded.window_days, + mean_score=excluded.mean_score, + stddev_score=excluded.stddev_score, + p50=excluded.p50, + p75=excluded.p75, + p90=excluded.p90, + p95=excluded.p95, + sample_count=excluded.sample_count, + updated_at=datetime('now')`, + ) + .bind( + next.scopeKey, + next.windowDays, + next.mean, + next.stddev, + next.p50, + next.p75, + next.p90, + next.p95, + next.values.length, + ) + .run(); + } + + let updatedEvents = 0; + for (const row of rows) { + const rawScore = clampScore(row.breaking_raw_score ?? row.breaking_score); + const selectedStats = chooseStats(row, stats); + const percentileValue = selectedStats ? percentileForScore(rawScore, selectedStats) : rawScore; + const calibratedScore = clampScore(rawScore * 0.7 + percentileValue * 0.3); + const confidence = clampScore(row.breaking_confidence, 60); + const label = labelForScore(calibratedScore, percentileValue, confidence); + const roundedPercentile = Math.round(percentileValue * 100) / 100; + if ( + !scoreChanged(row.breaking_raw_score, rawScore) && + !scoreChanged(row.breaking_percentile, roundedPercentile, 2) && + !scoreChanged(row.breaking_calibrated_score, calibratedScore) && + !scoreChanged(row.breaking_score, calibratedScore) && + row.breaking_label === label && + row.breaking_score_version === BREAKING_SCORE_VERSION + ) { + continue; + } + await db + .prepare( + `UPDATE events + SET breaking_raw_score = ?, + breaking_percentile = ?, + breaking_calibrated_score = ?, + breaking_score = ?, + breaking_label = ?, + breaking_score_version = ? + WHERE event_id = ?`, + ) + .bind( + rawScore, + roundedPercentile, + calibratedScore, + calibratedScore, + label, + BREAKING_SCORE_VERSION, + row.event_id, + ) + .run(); + updatedEvents += 1; + } + + return { + status: "ok", + window_days: windowDays, + scored_events: rows.length, + updated_events: updatedEvents, + stats_scopes: stats.size, + }; +} diff --git a/frontend/cloudflare/workers/lib/contracts.ts b/frontend/cloudflare/workers/lib/contracts.ts index 21e3b196..0ad48019 100644 --- a/frontend/cloudflare/workers/lib/contracts.ts +++ b/frontend/cloudflare/workers/lib/contracts.ts @@ -61,6 +61,10 @@ export interface PublicNewsItem { valueLabel: "精选" | "关注" | "普通" | "待评估"; valueScore?: number | null; breakingScore?: number | null; + breakingRawScore?: number | null; + breakingPercentile?: number | null; + breakingCalibratedScore?: number | null; + breakingVersion?: string | null; breakingLabel?: "flash" | "breaking" | "watch" | "timeline" | null; breakingReason?: string | null; breakingConfidence?: number | null; @@ -161,11 +165,15 @@ export interface ImportEventItem { title?: string; summary?: string; recommendation_reason?: string; + breaking_raw_score?: number; + breaking_percentile?: number; + breaking_calibrated_score?: number; breaking_score?: number; breaking_label?: string; breaking_reason?: string; breaking_confidence?: number; breaking_dimensions?: Record; + breaking_adversarial_flags?: Record; breaking_score_version?: string; target_timezone?: string; published_at_local?: string; diff --git a/frontend/cloudflare/workers/lib/public-news-query.ts b/frontend/cloudflare/workers/lib/public-news-query.ts index 704bfc63..2bff3129 100644 --- a/frontend/cloudflare/workers/lib/public-news-query.ts +++ b/frontend/cloudflare/workers/lib/public-news-query.ts @@ -10,7 +10,9 @@ import type { PublicNewsItem, PublicNewsSource } from "./contracts"; export const PUBLIC_FEATURED_MIN_SCORE = 60; export const PUBLIC_BREAKING_MIN_SCORE = 60; -export const BREAKING_SCORE_VERSION = "breaking-v1.0"; +export const PUBLIC_BREAKING_LEAD_MIN_PERCENTILE = 95; +export const PUBLIC_BREAKING_LEAD_MIN_CONFIDENCE = 70; +export const BREAKING_SCORE_VERSION = "breaking-v2.0"; export const SUPPORTED_PUBLIC_LOCALES = ["zh", "en", "es", "ar", "fr"] as const; export type PublicLocale = (typeof SUPPORTED_PUBLIC_LOCALES)[number]; @@ -39,11 +41,16 @@ export interface NewsRow { discussion_count: number | null; value_label: string; value_score: number | null; + breaking_raw_score: number | null; + breaking_percentile: number | null; + breaking_calibrated_score: number | null; breaking_score: number | null; breaking_label: string | null; breaking_reason: string | null; breaking_confidence: number | null; breaking_dimensions: string | null; + breaking_adversarial_flags: string | null; + breaking_score_version: string | null; target_timezone: string | null; published_at_local: string | null; available_locales: string | null; @@ -58,8 +65,11 @@ export const PUBLIC_NEWS_SELECT_COLUMNS = ` recommendation_reason, full_content, original_url, detail_url, image_urls, tags, issue_tags, related_tags, region_tags, entities, related_count, discussion_count, - value_label, value_score, breaking_score, breaking_label, + value_label, value_score, + breaking_raw_score, breaking_percentile, breaking_calibrated_score, + breaking_score, breaking_label, breaking_reason, breaking_confidence, breaking_dimensions, + breaking_adversarial_flags, breaking_score_version, target_timezone, published_at_local, ( SELECT json_group_array(locale) @@ -85,8 +95,10 @@ export const PUBLIC_NEWS_LOCALE_SELECT_COLUMNS = ` COALESCE(localized.localized_region_tags, events.region_tags) AS region_tags, events.entities, events.related_count, events.discussion_count, events.value_label, events.value_score, + events.breaking_raw_score, events.breaking_percentile, events.breaking_calibrated_score, events.breaking_score, events.breaking_label, events.breaking_reason, events.breaking_confidence, events.breaking_dimensions, + events.breaking_adversarial_flags, events.breaking_score_version, events.target_timezone, events.published_at_local, ( SELECT json_group_array(locale) @@ -116,7 +128,11 @@ export function buildPublicNewsWhere(input: PublicNewsFilterInput): { if (input.featured) { sql += ` AND ( - breaking_score >= ? + ( + breaking_score >= ? + AND breaking_percentile >= ? + AND breaking_confidence >= ? + ) OR (breaking_score IS NULL AND value_score >= ?) ) AND summary IS NOT NULL @@ -137,7 +153,12 @@ export function buildPublicNewsWhere(input: PublicNewsFilterInput): { OR UPPER(TRIM(title)) LIKE 'SUNDAY, %' ) `; - bindings.push(PUBLIC_BREAKING_MIN_SCORE, PUBLIC_FEATURED_MIN_SCORE); + bindings.push( + PUBLIC_BREAKING_MIN_SCORE, + PUBLIC_BREAKING_LEAD_MIN_PERCENTILE, + PUBLIC_BREAKING_LEAD_MIN_CONFIDENCE, + PUBLIC_FEATURED_MIN_SCORE, + ); } if (input.regionId) { @@ -171,7 +192,7 @@ export function buildPublicNewsWhere(input: PublicNewsFilterInput): { export function publicNewsOrderBy(featured: boolean): string { return featured - ? "ORDER BY events.breaking_score DESC, events.published_at DESC, events.event_id DESC" + ? "ORDER BY COALESCE(events.breaking_calibrated_score, events.breaking_score, events.value_score, -1) DESC, events.breaking_percentile DESC, events.published_at DESC, events.event_id DESC" : "ORDER BY events.published_at DESC, events.event_id DESC"; } @@ -267,6 +288,10 @@ export function rowToPublicNewsItem(row: NewsRow): PublicNewsItem { valueLabel: row.value_label as PublicNewsItem["valueLabel"], valueScore: row.value_score, breakingScore: row.breaking_score ?? row.value_score, + breakingRawScore: row.breaking_raw_score, + breakingPercentile: row.breaking_percentile, + breakingCalibratedScore: row.breaking_calibrated_score, + breakingVersion: row.breaking_score_version, breakingLabel: (row.breaking_label as PublicNewsItem["breakingLabel"]) ?? null, breakingReason: row.breaking_reason, breakingConfidence: row.breaking_confidence, @@ -277,3 +302,10 @@ export function rowToPublicNewsItem(row: NewsRow): PublicNewsItem { chinaRelevanceLabel: row.china_relevance_label as PublicNewsItem["chinaRelevanceLabel"], }; } + +export function pollAfterMsForPublicNews(featured: boolean, items: PublicNewsItem[]): number { + if (!featured) return 60000; + if (items.some((item) => item.breakingLabel === "flash")) return 15000; + if (items.some((item) => item.breakingLabel === "breaking")) return 20000; + return 30000; +} diff --git a/frontend/cloudflare/workers/lib/public-read-snapshots.ts b/frontend/cloudflare/workers/lib/public-read-snapshots.ts index 6e83fb43..7d0787e5 100644 --- a/frontend/cloudflare/workers/lib/public-read-snapshots.ts +++ b/frontend/cloudflare/workers/lib/public-read-snapshots.ts @@ -13,6 +13,7 @@ import { publicNewsOrderBy, publicNewsLocaleJoin, publicNewsSelectColumnsForLocale, + pollAfterMsForPublicNews, rowToPublicNewsItem, SUPPORTED_PUBLIC_LOCALES, } from "./public-news-query"; @@ -225,14 +226,15 @@ async function buildNewsFeedSnapshot( ]); const rows = result.results || []; const pageRows = rows.slice(0, PUBLIC_SNAPSHOT_PAGE_SIZE); + const items = pageRows.map((row) => rowToPublicNewsItem(row)); return { - items: pageRows.map((row) => rowToPublicNewsItem(row)), + items, latestCursor: pageRows[0]?.event_id ?? null, nextCursor: rows.length > PUBLIC_SNAPSHOT_PAGE_SIZE ? (pageRows[pageRows.length - 1]?.event_id ?? null) : null, - pollAfterMs: featured ? 30000 : 60000, + pollAfterMs: pollAfterMsForPublicNews(featured, items), hasNewer: false, total: totalResult?.total ?? pageRows.length, }; diff --git a/frontend/cloudflare/workers/lib/scheduled.ts b/frontend/cloudflare/workers/lib/scheduled.ts index e96134d4..07f281c3 100644 --- a/frontend/cloudflare/workers/lib/scheduled.ts +++ b/frontend/cloudflare/workers/lib/scheduled.ts @@ -1,5 +1,6 @@ import { getContainer } from "@cloudflare/containers"; import { importEventsToD1 } from "../api/webhook"; +import { refreshBreakingScoreStats } from "./breaking-calibration"; import type { ImportEventItem } from "./contracts"; import { refreshPublicReadSnapshots } from "./public-read-snapshots"; @@ -553,6 +554,15 @@ export async function runScheduledCloudflareTask( if (task === "collect-cycle" || task === "public-translation-cycle") { importResult = await importContainerEventsToD1(env.DB, details); } + let breakingRefresh: Record; + try { + breakingRefresh = await refreshBreakingScoreStats(env.DB); + } catch (error) { + breakingRefresh = { + status: "error", + message: error instanceof Error ? error.message : String(error), + }; + } let snapshotRefresh: Record; try { snapshotRefresh = await refreshPublicReadSnapshots(env.DB); @@ -569,6 +579,7 @@ export async function runScheduledCloudflareTask( ...(collectBatch === null ? {} : { collect_batch: collectBatchDetails(collectBatch) }), + breaking_stats: breakingRefresh, snapshots: snapshotRefresh, }); const status = diff --git a/frontend/public/src/lib/api.test.ts b/frontend/public/src/lib/api.test.ts index 208f0a90..fbed01d1 100644 --- a/frontend/public/src/lib/api.test.ts +++ b/frontend/public/src/lib/api.test.ts @@ -40,6 +40,11 @@ const item: PublicNewsItem = { discussionCount: 0, valueLabel: "精选", valueScore: 92, + breakingScore: 82, + breakingRawScore: 88, + breakingPercentile: 86.4, + breakingCalibratedScore: 82, + breakingVersion: "breaking-v2.0", chinaRelevanceLabel: "中", } @@ -121,6 +126,9 @@ describe("public news API client", () => { expect(result.notModified).toBe(false) expect(result.etag).toBe('"feed-tag"') expect(result.data?.items[0]?.source.name).toBe("ANSA.it") + expect(result.data?.items[0]?.breakingRawScore).toBe(88) + expect(result.data?.items[0]?.breakingPercentile).toBe(86.4) + expect(result.data?.items[0]?.breakingVersion).toBe("breaking-v2.0") }) it("handles 304 without parsing a body", async () => { diff --git a/frontend/public/src/lib/feed-state.test.ts b/frontend/public/src/lib/feed-state.test.ts index fd9539f1..612c72e3 100644 --- a/frontend/public/src/lib/feed-state.test.ts +++ b/frontend/public/src/lib/feed-state.test.ts @@ -80,10 +80,15 @@ describe("feed state helpers", () => { }) it("clamps normal polling and backs off failures", () => { - expect(nextPollDelayMs({ serverMs: 2_000, failureCount: 0 })).toBe(30_000) + expect(nextPollDelayMs({ serverMs: 2_000, failureCount: 0 })).toBe(15_000) expect(nextPollDelayMs({ serverMs: 60_000, failureCount: 2 })).toBeGreaterThan(60_000) }) + it("allows urgent breaking feeds to poll as fast as fifteen seconds", () => { + expect(nextPollDelayMs({ serverMs: 15_000, failureCount: 0 })).toBe(15_000) + expect(nextPollDelayMs({ serverMs: 15_000, failureCount: 1 })).toBe(30_000) + }) + it("pauses polling when the page is hidden or offline", () => { expect(shouldPausePolling({ visibilityState: "hidden", online: true })).toBe(true) expect(shouldPausePolling({ visibilityState: "visible", online: false })).toBe(true) diff --git a/frontend/public/src/lib/feed-state.ts b/frontend/public/src/lib/feed-state.ts index 8282f3fb..6b672e41 100644 --- a/frontend/public/src/lib/feed-state.ts +++ b/frontend/public/src/lib/feed-state.ts @@ -25,7 +25,7 @@ export interface PollPauseInput { online: boolean } -const MIN_POLL_MS = 30_000 +const MIN_POLL_MS = 15_000 const MAX_POLL_MS = 5 * 60_000 function clean(value: string | undefined) { diff --git a/frontend/public/src/types/public-news.ts b/frontend/public/src/types/public-news.ts index a964a6e6..9bcd66fe 100644 --- a/frontend/public/src/types/public-news.ts +++ b/frontend/public/src/types/public-news.ts @@ -39,6 +39,10 @@ export interface PublicNewsItem { valueLabel: PublicNewsValueLabel valueScore?: number | null breakingScore?: number | null + breakingRawScore?: number | null + breakingPercentile?: number | null + breakingCalibratedScore?: number | null + breakingVersion?: string | null breakingLabel?: "flash" | "breaking" | "watch" | "timeline" | null breakingReason?: string | null breakingConfidence?: number | null diff --git a/src/news_sentry/api/schemas.py b/src/news_sentry/api/schemas.py index c5312660..42859b2b 100644 --- a/src/news_sentry/api/schemas.py +++ b/src/news_sentry/api/schemas.py @@ -929,6 +929,27 @@ class PublicNewsItem(BaseModel): discussion_count: int | None = Field(default=None, alias="discussionCount") value_label: Literal["精选", "关注", "普通", "待评估"] = Field(alias="valueLabel") value_score: int | float | None = Field(default=None, alias="valueScore") + breaking_score: int | float | None = Field(default=None, alias="breakingScore") + breaking_raw_score: int | float | None = Field(default=None, alias="breakingRawScore") + breaking_percentile: int | float | None = Field(default=None, alias="breakingPercentile") + breaking_calibrated_score: int | float | None = Field( + default=None, + alias="breakingCalibratedScore", + ) + breaking_version: str | None = Field(default=None, alias="breakingVersion") + breaking_label: Literal["flash", "breaking", "watch", "timeline"] | None = Field( + default=None, + alias="breakingLabel", + ) + breaking_reason: str | None = Field(default=None, alias="breakingReason") + breaking_confidence: int | float | None = Field(default=None, alias="breakingConfidence") + breaking_dimensions: dict[str, int | float] = Field( + default_factory=dict, + alias="breakingDimensions", + ) + target_timezone: str | None = Field(default=None, alias="targetTimezone") + published_at_local: str | None = Field(default=None, alias="publishedAtLocal") + available_locales: list[str] = Field(default_factory=list, alias="availableLocales") china_relevance_label: Literal["高", "中", "低", "未知"] = Field( default="未知", alias="chinaRelevanceLabel", diff --git a/src/news_sentry/core/breaking_scoring.py b/src/news_sentry/core/breaking_scoring.py index ab765820..b362c32c 100644 --- a/src/news_sentry/core/breaking_scoring.py +++ b/src/news_sentry/core/breaking_scoring.py @@ -5,7 +5,7 @@ from dataclasses import dataclass from typing import Any, Literal -BREAKING_SCORE_VERSION = "breaking-v1.0" +BREAKING_SCORE_VERSION = "breaking-v2.0" BREAKING_DIMENSION_WEIGHTS: dict[str, int] = { "impact_scope": 22, @@ -59,6 +59,22 @@ class BreakingAssessment: dimensions: dict[str, int] penalties: dict[str, int] version: str = BREAKING_SCORE_VERSION + raw_score: int | None = None + percentile: float | None = None + calibrated_score: int | None = None + + +@dataclass(frozen=True) +class BreakingScoreStats: + scope_key: str + window_days: int + mean_score: float = 0.0 + stddev_score: float = 0.0 + p50: float = 0.0 + p75: float = 0.0 + p90: float = 0.0 + p95: float = 0.0 + sample_count: int = 0 def _clamp_score(value: Any, *, default: int = 0) -> int: # noqa: ANN401 @@ -79,6 +95,73 @@ def _label_for_score(score: int, confidence: int = 80) -> BreakingLabel: return "timeline" +def _interpolated_percentile(score: int, stats: BreakingScoreStats) -> float: + if stats.sample_count < 30: + return float(score) + checkpoints = [ + (0.0, 0.0), + (float(stats.p50 or stats.mean_score or 50), 50.0), + (float(stats.p75 or stats.p50 or stats.mean_score or 60), 75.0), + (float(stats.p90 or stats.p75 or stats.mean_score or 72), 90.0), + (float(stats.p95 or stats.p90 or stats.mean_score or 85), 95.0), + (100.0, 100.0), + ] + checkpoints.sort(key=lambda item: item[0]) + previous_score, previous_pct = checkpoints[0] + for next_score, next_pct in checkpoints[1:]: + if score <= next_score: + span = max(1.0, next_score - previous_score) + ratio = (score - previous_score) / span + return max(0.0, min(100.0, previous_pct + (next_pct - previous_pct) * ratio)) + previous_score, previous_pct = next_score, next_pct + return 100.0 + + +def _label_for_calibrated_score( + score: int, + *, + percentile: float, + confidence: int, +) -> BreakingLabel: + if score >= 85 and percentile >= 95 and confidence >= 70: + return "flash" + if score >= 72 and percentile >= 90 and confidence >= 60: + return "breaking" + if score >= 52 or percentile >= 75: + return "watch" + return "timeline" + + +def calibrate_breaking_assessment( + assessment: BreakingAssessment, + stats: BreakingScoreStats | None, +) -> BreakingAssessment: + """Blend raw score with rolling distribution percentile for public ordering.""" + raw_score = _clamp_score(assessment.raw_score or assessment.score) + if stats is None or stats.sample_count < 30: + percentile = float(raw_score) + else: + percentile = _interpolated_percentile(raw_score, stats) + calibrated_score = _clamp_score((raw_score * 0.7) + (percentile * 0.3)) + label = _label_for_calibrated_score( + calibrated_score, + percentile=percentile, + confidence=assessment.confidence, + ) + return BreakingAssessment( + score=calibrated_score, + label=label, + confidence=assessment.confidence, + reason=assessment.reason, + dimensions=assessment.dimensions, + penalties=assessment.penalties, + version=BREAKING_SCORE_VERSION, + raw_score=raw_score, + percentile=round(percentile, 2), + calibrated_score=calibrated_score, + ) + + def _source_reliability(source_type: str, credibility_label: str | None) -> int: label = (credibility_label or "").lower() if label in {"high", "trusted", "official", "authority"}: diff --git a/src/news_sentry/core/collector_config_utils.py b/src/news_sentry/core/collector_config_utils.py index dd037df2..1e2a52a3 100644 --- a/src/news_sentry/core/collector_config_utils.py +++ b/src/news_sentry/core/collector_config_utils.py @@ -1200,6 +1200,21 @@ def _cloudflare_import_payload_from_row(row: sqlite3.Row) -> dict[str, Any] | No issue_tags = [classification_l0] related_tags = _cloudflare_import_tags(publication.get("related_tags")) if ready else topic_tags region_tags = _cloudflare_import_tags(publication.get("region_tags")) if ready else [target_id] + breaking_raw_score = ( + breaking.get("raw_score") + if breaking.get("raw_score") is not None + else breaking.get("score") + ) + breaking_percentile = ( + breaking.get("percentile") + if breaking.get("percentile") is not None + else breaking.get("score") + ) + breaking_calibrated_score = ( + breaking.get("calibrated_score") + if breaking.get("calibrated_score") is not None + else breaking.get("score") + ) title = str(translation.get("title_pre") or title_original).strip() summary = str( @@ -1231,11 +1246,29 @@ def _cloudflare_import_payload_from_row(row: sqlite3.Row) -> dict[str, Any] | No "china_relevance_label": _cloudflare_import_china_relevance_label( row["china_relevance"] ), - "breaking_score": breaking.get("score") or row["news_value_score"], + "breaking_raw_score": ( + breaking_raw_score if breaking_raw_score is not None else row["news_value_score"] + ), + "breaking_percentile": ( + breaking_percentile if breaking_percentile is not None else row["news_value_score"] + ), + "breaking_calibrated_score": ( + breaking_calibrated_score + if breaking_calibrated_score is not None + else row["news_value_score"] + ), + "breaking_score": ( + breaking_calibrated_score + if breaking_calibrated_score is not None + else row["news_value_score"] + ), "breaking_label": breaking.get("label"), "breaking_reason": breaking.get("reason") or recommendation_reason, "breaking_confidence": breaking.get("confidence"), "breaking_dimensions": breaking.get("dimensions") if isinstance(breaking, dict) else {}, + "breaking_adversarial_flags": { + "penalties": breaking.get("penalties", {}), + }, "breaking_score_version": breaking.get("version") if isinstance(breaking, dict) else None, "target_timezone": target_timezone, "published_at_local": published_at_local, diff --git a/src/news_sentry/core/public_translation.py b/src/news_sentry/core/public_translation.py index 96f5b1d3..b8d42f17 100644 --- a/src/news_sentry/core/public_translation.py +++ b/src/news_sentry/core/public_translation.py @@ -1041,6 +1041,9 @@ async def _generate_publication_fields( "localizations": localizations, "breaking": { "score": breaking.score, + "raw_score": breaking.raw_score or breaking.score, + "percentile": breaking.percentile, + "calibrated_score": breaking.calibrated_score or breaking.score, "label": breaking.label, "confidence": breaking.confidence, "reason": breaking.reason, diff --git a/tests/unit/test_breaking_scoring.py b/tests/unit/test_breaking_scoring.py index aad228ee..eb647d09 100644 --- a/tests/unit/test_breaking_scoring.py +++ b/tests/unit/test_breaking_scoring.py @@ -3,12 +3,18 @@ from news_sentry.core.breaking_scoring import ( BREAKING_SCORE_VERSION, BreakingScoreInput, + BreakingScoreStats, BreakingScoreValidationError, + calibrate_breaking_assessment, score_breaking_event, validate_llm_breaking_assessment, ) +def test_breaking_score_contract_is_v2() -> None: + assert BREAKING_SCORE_VERSION == "breaking-v2.0" + + def test_deterministic_breaking_score_separates_flash_from_routine_update() -> None: flash = score_breaking_event( BreakingScoreInput( @@ -50,6 +56,59 @@ def test_deterministic_breaking_score_separates_flash_from_routine_update() -> N assert routine.label == "timeline" +def test_breaking_calibration_uses_percentile_to_avoid_false_flash() -> None: + raw = validate_llm_breaking_assessment( + { + "breaking_score": 88, + "breaking_label": "flash", + "breaking_confidence": 82, + "breaking_reason": "事件有明显跨境影响，但同类高分事件近期较多，需要按分布校准。", + "dimensions": { + "impact_scope": 88, + "urgency": 82, + "novelty": 78, + "source_reliability": 86, + "actionability": 72, + "systemic_or_cross_border": 84, + "human_attention": 70, + "evidence_confidence": 86, + }, + "penalties": { + "duplicate": 0, + "routine": 0, + "sensationalism": 0, + "thin_evidence": 0, + }, + "adversarial_checks": { + "not_routine": True, + "not_opinion": True, + "not_duplicate": True, + "not_single_source_social": True, + "has_trustworthy_timestamp": True, + }, + } + ) + + calibrated = calibrate_breaking_assessment( + raw, + BreakingScoreStats( + scope_key="global", + window_days=30, + mean_score=82, + stddev_score=8, + p75=84, + p90=91, + p95=96, + sample_count=250, + ), + ) + + assert calibrated.raw_score == 88 + assert calibrated.percentile < 90 + assert calibrated.score < raw.score + assert calibrated.label == "watch" + + def test_llm_breaking_assessment_requires_adversarial_checks_and_distribution() -> None: payload = { "breaking_score": 88, diff --git a/tests/unit/test_cloudflare_native_config.py b/tests/unit/test_cloudflare_native_config.py index 7b209594..929fc9b2 100644 --- a/tests/unit/test_cloudflare_native_config.py +++ b/tests/unit/test_cloudflare_native_config.py @@ -86,7 +86,10 @@ def test_cloudflare_public_featured_query_matches_python_quality_gate() -> None: assert 'params.get("featured") !== "false"' in bootstrap_ts assert "PUBLIC_FEATURED_MIN_SCORE = 60" in query_ts assert "PUBLIC_BREAKING_MIN_SCORE = 60" in query_ts + assert "PUBLIC_BREAKING_LEAD_MIN_PERCENTILE = 95" in query_ts assert "breaking_score >= ?" in query_ts + assert "breaking_percentile >= ?" in query_ts + assert "breaking_confidence >= ?" in query_ts assert "value_score >= ?" in query_ts assert "summary IS NOT NULL" in query_ts assert "TRIM(summary) != ''" in query_ts @@ -97,10 +100,11 @@ def test_cloudflare_public_featured_query_matches_python_quality_gate() -> None: assert "NOT IN ('uncategorized', 'other', 'breaking_news')" in query_ts assert "NOT LIKE '%/opinion/todayinhistory/%'" in query_ts assert "UPPER(TRIM(title)) LIKE 'MONDAY, %'" in query_ts - assert ( - "ORDER BY events.breaking_score DESC, events.published_at DESC, events.event_id DESC" - in query_ts + breaking_order = ( + "ORDER BY COALESCE(events.breaking_calibrated_score, events.breaking_score, " + "events.value_score, -1) DESC" ) + assert breaking_order in query_ts assert "publicNewsOrderBy(featured)" in news_ts assert "publicNewsOrderBy(featured)" in bootstrap_ts @@ -109,13 +113,21 @@ def test_cloudflare_breaking_intelligence_contract_is_schema_backed() -> None: schema_sql = _read("db/schema.sql") query_ts = _read("workers/lib/public-news-query.ts") contracts_ts = _read("workers/lib/contracts.ts") + breaking_order = ( + "ORDER BY COALESCE(events.breaking_calibrated_score, events.breaking_score, " + "events.value_score, -1) DESC" + ) for column in ( + "breaking_raw_score REAL", + "breaking_percentile REAL", + "breaking_calibrated_score REAL", "breaking_score REAL", "breaking_label TEXT", "breaking_reason TEXT", "breaking_confidence INTEGER", "breaking_dimensions TEXT DEFAULT '{}'", + "breaking_adversarial_flags TEXT DEFAULT '{}'", "breaking_score_version TEXT", "target_timezone TEXT DEFAULT 'UTC'", "published_at_local TEXT", @@ -126,14 +138,16 @@ def test_cloudflare_breaking_intelligence_contract_is_schema_backed() -> None: assert "CREATE TABLE IF NOT EXISTS breaking_score_stats" in schema_sql assert "idx_events_public_breaking" in schema_sql assert "idx_event_localizations_locale" in schema_sql + assert "refreshBreakingScoreStats" in _read("workers/lib/scheduled.ts") assert "breaking_score >= ?" in query_ts - assert ( - "ORDER BY events.breaking_score DESC, events.published_at DESC, events.event_id DESC" - in query_ts - ) + assert breaking_order in query_ts for field in ( "breakingScore?: number | null", + "breakingRawScore?: number | null", + "breakingPercentile?: number | null", + "breakingCalibratedScore?: number | null", + "breakingVersion?: string | null", "breakingLabel?:", "breakingReason?: string | null", "breakingConfidence?: number | null", @@ -145,6 +159,34 @@ def test_cloudflare_breaking_intelligence_contract_is_schema_backed() -> None: assert field in contracts_ts +def test_cloudflare_public_news_uses_sql_cursor_pagination() -> None: + news_ts = _read("workers/api/news.ts") + query_ts = _read("workers/lib/public-news-query.ts") + + assert "buildCursorFilter" in news_ts + assert 'params.get("before_cursor")' in news_ts + assert 'params.get("since_cursor")' in news_ts + cursor_select = ( + "SELECT event_id, published_at, value_score, breaking_score, " + "breaking_calibrated_score FROM events WHERE event_id = ?" + ) + assert cursor_select in news_ts + assert "COALESCE(breaking_calibrated_score, breaking_score, value_score, -1)" in news_ts + assert "${cursorFilter.sql}" in news_ts + assert "SELECT COUNT(*) AS total FROM events ${filters.sql}" in news_ts + assert "Number.isFinite(requestedPageSize)" in news_ts + assert "Math.min(requestedPageSize, 50)" in news_ts + assert "const pageRows = rows.slice(0, pageSize)" in news_ts + assert "const items = pageRows.map" in news_ts + assert "hasNewer: Boolean(sinceCursor && items.length > 0)" in news_ts + assert ( + "COALESCE(events.breaking_calibrated_score, events.breaking_score, " + "events.value_score, -1)" + ) in query_ts + assert "export function publicNewsOrderBy(featured: boolean)" in query_ts + assert "publicNewsOrderBy(featured)" in news_ts + + def test_cloudflare_public_reads_are_locale_aware_without_breaking_shape() -> None: news_ts = _read("workers/api/news.ts") query_ts = _read("workers/lib/public-news-query.ts") @@ -175,24 +217,24 @@ def test_cloudflare_has_ninety_day_retention_and_cost_guards() -> None: assert "23 * * * *" in wrangler_toml -def test_cloudflare_public_news_uses_sql_cursor_pagination() -> None: - news_ts = _read("workers/api/news.ts") - query_ts = _read("workers/lib/public-news-query.ts") +def test_cloudflare_import_accepts_breaking_v2_projection_fields() -> None: + contracts_ts = _read("workers/lib/contracts.ts") + webhook_ts = _read("workers/api/webhook.ts") - assert "buildCursorFilter" in news_ts - assert 'params.get("before_cursor")' in news_ts - assert 'params.get("since_cursor")' in news_ts - assert ( - "SELECT event_id, published_at, value_score, breaking_score FROM events WHERE event_id = ?" - in news_ts - ) - assert "${cursorFilter.sql}" in news_ts - assert "SELECT COUNT(*) AS total FROM events ${filters.sql}" in news_ts - assert "Number.isFinite(requestedPageSize)" in news_ts - assert "const pageRows = rows.slice(0, pageSize)" in news_ts - assert "const items = pageRows.map" in news_ts - assert "hasNewer: Boolean(sinceCursor && items.length > 0)" in news_ts - assert "ORDER BY events.published_at DESC, events.event_id DESC" in query_ts + for field in ( + "breaking_raw_score?: number", + "breaking_percentile?: number", + "breaking_calibrated_score?: number", + "breaking_adversarial_flags?: Record", + ): + assert field in contracts_ts + for column in ( + "breaking_raw_score", + "breaking_percentile", + "breaking_calibrated_score", + "breaking_adversarial_flags", + ): + assert column in webhook_ts def test_cloudflare_d1_has_public_featured_index() -> None: diff --git a/tests/unit/test_cloudflare_refresh_public_snapshots.py b/tests/unit/test_cloudflare_refresh_public_snapshots.py index da00370e..84cd29c1 100644 --- a/tests/unit/test_cloudflare_refresh_public_snapshots.py +++ b/tests/unit/test_cloudflare_refresh_public_snapshots.py @@ -60,3 +60,47 @@ def test_snapshot_upsert_sql_writes_fixed_keys_to_d1_schema(tmp_path: Path) -> N ("news:featured:v1:page_size=20", 1), ("regions:active:v1", 1), ] + + +def test_snapshot_tool_includes_breaking_v2_public_fields() -> None: + from tools.cloudflare_refresh_public_snapshots import _news_item + + item = _news_item( + { + "event_id": "e1", + "target_id": "france", + "target_label": "法国", + "source_id": "lemonde", + "source_name": "Le Monde", + "source_type": "rss", + "published_at": "2026-07-02T01:00:00Z", + "title": "France announces emergency measure", + "detail_url": "/public-app/news/e1", + "tags": "[]", + "issue_tags": "[]", + "related_tags": "[]", + "region_tags": "[]", + "entities": "[]", + "related_count": 0, + "value_label": "精选", + "value_score": 91, + "breaking_score": 82, + "breaking_raw_score": 88, + "breaking_percentile": 86.4, + "breaking_calibrated_score": 82, + "breaking_label": "watch", + "breaking_reason": "分布校准后仍值得关注，但未达到 flash 阈值。", + "breaking_confidence": 78, + "breaking_dimensions": '{"impact_scope":88}', + "breaking_score_version": "breaking-v2.0", + "target_timezone": "Europe/Paris", + "published_at_local": "2026-07-02T03:00:00+02:00", + "china_relevance_label": "中", + } + ) + + assert item["breakingScore"] == 82 + assert item["breakingRawScore"] == 88 + assert item["breakingPercentile"] == 86.4 + assert item["breakingCalibratedScore"] == 82 + assert item["breakingVersion"] == "breaking-v2.0" diff --git a/tools/cloudflare_refresh_public_snapshots.py b/tools/cloudflare_refresh_public_snapshots.py index 1653263c..7295110b 100644 --- a/tools/cloudflare_refresh_public_snapshots.py +++ b/tools/cloudflare_refresh_public_snapshots.py @@ -25,10 +25,18 @@ recommendation_reason, full_content, original_url, detail_url, image_urls, tags, issue_tags, related_tags, region_tags, entities, related_count, discussion_count, - value_label, value_score, china_relevance_label + value_label, value_score, + breaking_raw_score, breaking_percentile, breaking_calibrated_score, + breaking_score, breaking_label, breaking_reason, breaking_confidence, + breaking_dimensions, breaking_score_version, + target_timezone, published_at_local, + china_relevance_label """ PUBLIC_FEATURED_MIN_SCORE = 60 +PUBLIC_BREAKING_MIN_SCORE = 60 +PUBLIC_BREAKING_LEAD_MIN_PERCENTILE = 95 +PUBLIC_BREAKING_LEAD_MIN_CONFIDENCE = 70 def _sql(value: Any) -> str: @@ -125,7 +133,14 @@ def _public_news_where(featured: bool) -> str: return ( where + f""" - AND value_score >= {PUBLIC_FEATURED_MIN_SCORE} + AND ( + ( + breaking_score >= {PUBLIC_BREAKING_MIN_SCORE} + AND breaking_percentile >= {PUBLIC_BREAKING_LEAD_MIN_PERCENTILE} + AND breaking_confidence >= {PUBLIC_BREAKING_LEAD_MIN_CONFIDENCE} + ) + OR (breaking_score IS NULL AND value_score >= {PUBLIC_FEATURED_MIN_SCORE}) + ) AND summary IS NOT NULL AND TRIM(summary) != '' AND recommendation_reason IS NOT NULL @@ -149,12 +164,23 @@ def _public_news_where(featured: bool) -> str: def _public_news_order_by(featured: bool) -> str: return ( - "ORDER BY value_score DESC, published_at DESC, event_id DESC" + "ORDER BY COALESCE(breaking_calibrated_score, breaking_score, value_score, -1) " + "DESC, breaking_percentile DESC, published_at DESC, event_id DESC" if featured else "ORDER BY published_at DESC, event_id DESC" ) +def _poll_after_ms(featured: bool, items: list[dict[str, Any]]) -> int: + if not featured: + return 60000 + if any(item.get("breakingLabel") == "flash" for item in items): + return 15000 + if any(item.get("breakingLabel") == "breaking" for item in items): + return 20000 + return 30000 + + def _news_item(row: Mapping[str, Any]) -> dict[str, Any]: return { "id": row.get("event_id"), @@ -184,17 +210,33 @@ def _news_item(row: Mapping[str, Any]) -> dict[str, Any]: "discussionCount": row.get("discussion_count"), "valueLabel": row.get("value_label") or "普通", "valueScore": row.get("value_score"), + "breakingScore": ( + row.get("breaking_score") + if row.get("breaking_score") is not None + else row.get("value_score") + ), + "breakingRawScore": row.get("breaking_raw_score"), + "breakingPercentile": row.get("breaking_percentile"), + "breakingCalibratedScore": row.get("breaking_calibrated_score"), + "breakingLabel": row.get("breaking_label"), + "breakingReason": row.get("breaking_reason"), + "breakingConfidence": row.get("breaking_confidence"), + "breakingDimensions": _json_object(row.get("breaking_dimensions")), + "breakingVersion": row.get("breaking_score_version"), + "targetTimezone": row.get("target_timezone"), + "publishedAtLocal": row.get("published_at_local"), "chinaRelevanceLabel": row.get("china_relevance_label") or "未知", } -def _feed_payload(rows: list[dict[str, Any]], total: int) -> dict[str, Any]: +def _feed_payload(rows: list[dict[str, Any]], total: int, *, featured: bool) -> dict[str, Any]: page_rows = rows[:20] + items = [_news_item(row) for row in page_rows] return { - "items": [_news_item(row) for row in page_rows], + "items": items, "latestCursor": page_rows[0].get("event_id") if page_rows else None, "nextCursor": page_rows[-1].get("event_id") if len(rows) > 20 and page_rows else None, - "pollAfterMs": 60000, + "pollAfterMs": _poll_after_ms(featured=featured, items=items), "hasNewer": False, "total": total, } @@ -224,7 +266,7 @@ def _query_news_feed( sql=f"SELECT COUNT(*) AS total FROM events {where}", # noqa: S608 ) total = int((count_rows[0] if count_rows else {}).get("total") or len(rows)) - return _feed_payload(rows, total) + return _feed_payload(rows, total, featured=featured) def _query_facets(*, wrangler_dir: Path, database: str, remote: bool) -> dict[str, Any]: