From be3cf45362942edfe9517ecca1950e30046e265a Mon Sep 17 00:00:00 2001 From: eezd Date: Mon, 2 Nov 2026 21:20:44 +0800 Subject: [PATCH] fix(news): exclude null AI results from homepage reading inputs --- backend/src/alphapulse/api/risk.py | 6 ++-- backend/src/alphapulse/news_risk/events.py | 6 ++-- scripts/smoke_events.py | 36 +++++++++++++++++++++- 3 files changed, 41 insertions(+), 7 deletions(-) diff --git a/backend/src/alphapulse/api/risk.py b/backend/src/alphapulse/api/risk.py index 3b67103..c24599b 100644 --- a/backend/src/alphapulse/api/risk.py +++ b/backend/src/alphapulse/api/risk.py @@ -175,8 +175,8 @@ async def coverage(session): async def overview(session: DB, hours: RiskWindow = RiskWindow.DAY): now = datetime.now(UTC); cutoff = now - timedelta(hours=hours) counts = dict((await session.execute(select(NewsSummary.importance_stars, func.count()).where( - NewsSummary.updated_at >= cutoff, NewsSummary.result.is_not(None), NewsSummary.replacement_ids == [], NewsSummary.state != 'withdrawn').group_by(NewsSummary.importance_stars))).all()) - snapshot = await session.scalar(select(RiskSnapshot).where(RiskSnapshot.state == 'done', RiskSnapshot.result['inputs'].is_not(None)).order_by(RiskSnapshot.id.desc()).limit(1)) + NewsSummary.updated_at >= cutoff, func.jsonb_typeof(NewsSummary.result) == 'object', NewsSummary.replacement_ids == [], NewsSummary.state != 'withdrawn').group_by(NewsSummary.importance_stars))).all()) + snapshot = await session.scalar(select(RiskSnapshot).where(RiskSnapshot.state == 'done', func.jsonb_typeof(RiskSnapshot.result['inputs']) == 'array').order_by(RiskSnapshot.id.desc()).limit(1)) last = await session.scalar(select(RiskSnapshot).order_by(RiskSnapshot.id.desc()).limit(1)) scenario = None stage2 = dict((await session.execute(select(NewsSummary.state, func.count()).group_by(NewsSummary.state))).all()) @@ -211,7 +211,7 @@ async def overview(session: DB, hours: RiskWindow = RiskWindow.DAY): async def events(session: DB, hours: RiskWindow | None = None, stars: int = Query(0, ge=0, le=5), tag: str = '', importance: int = Query(0, ge=0, le=5), category: str = '', asset: str = '', q: str = '', start: datetime | None = None, end: datetime | None = None, page: int = Query(1, ge=1), limit: int = Query(20, ge=1, le=100)): - conditions = [NewsSummary.replacement_ids == [], NewsSummary.state != 'withdrawn', NewsSummary.stars >= stars, NewsSummary.result.is_not(None)] + conditions = [NewsSummary.replacement_ids == [], NewsSummary.state != 'withdrawn', NewsSummary.stars >= stars, func.jsonb_typeof(NewsSummary.result) == 'object'] if importance: conditions.append(NewsSummary.importance_stars >= importance) if tag: conditions.append(NewsSummary.tags.contains([tag.strip().casefold()])) if hours: conditions.append(NewsSummary.updated_at >= datetime.now(UTC)-timedelta(hours=hours)) diff --git a/backend/src/alphapulse/news_risk/events.py b/backend/src/alphapulse/news_risk/events.py index e66e74a..3eaa79a 100644 --- a/backend/src/alphapulse/news_risk/events.py +++ b/backend/src/alphapulse/news_risk/events.py @@ -1,7 +1,7 @@ """Version-bound package relationships and evidence-backed event timelines.""" from datetime import UTC, datetime, timedelta from pydantic import BaseModel, Field -from sqlalchemy import select +from sqlalchemy import select, func from alphapulse.models.risk import NewsSummary, NewsEvent, EventMember, EventVersion, RiskSettings, EventLink from alphapulse.core.db import SessionLocal from alphapulse.news_risk.pipeline import configuration @@ -168,9 +168,9 @@ async def organize(target_id=None): async def reading_inputs(session, cutoff): - packages = (await session.scalars(select(NewsSummary).where(NewsSummary.result.is_not(None), NewsSummary.replacement_ids == [], NewsSummary.state != 'withdrawn'))).all() + packages = (await session.scalars(select(NewsSummary).where(func.jsonb_typeof(NewsSummary.result) == 'object', NewsSummary.replacement_ids == [], NewsSummary.state != 'withdrawn'))).all() memberships = {m.package_id: m.event_id for m in (await session.scalars(select(EventMember))).all()} - events = {e.id: e for e in (await session.scalars(select(NewsEvent).where(NewsEvent.result.is_not(None), NewsEvent.state != 'retired'))).all()} + events = {e.id: e for e in (await session.scalars(select(NewsEvent).where(func.jsonb_typeof(NewsEvent.result) == 'object', NewsEvent.state != 'retired'))).all()} inputs, included = [], set() for package in packages: if package.updated_at < cutoff: continue diff --git a/scripts/smoke_events.py b/scripts/smoke_events.py index c4a7124..ab32c6b 100644 --- a/scripts/smoke_events.py +++ b/scripts/smoke_events.py @@ -5,7 +5,9 @@ import os from types import SimpleNamespace from unittest.mock import patch from urllib.parse import urlparse -from sqlalchemy import select, func, delete +import httpx +from fastapi import FastAPI +from sqlalchemy import select, func, delete, null from alphapulse.core.db import SessionLocal, engine from alphapulse.models import AiProvider, NewsItem, SkillNode, SkillVersion from alphapulse.models.risk import NewsSummary, ArticleAnalysis, SummaryArticle, RiskSettings, NewsEvent, EventMember, EventVersion, ReprocessJob, PackageCorrection, GroupingVersion, SummaryVersion, PackageEvidence, RiskSnapshot, EventLink @@ -50,6 +52,38 @@ async def main(): session.add(withdrawn); await session.flush() session.add(EventMember(package_id=withdrawn.id,event_id=event_id,reason='Previous member')) await session.commit() + async with SessionLocal() as session: + nested = await session.begin_nested() + await session.execute(NewsEvent.__table__.update().where(NewsEvent.id == event_id).values(result=None)) + await session.execute(NewsSummary.__table__.update().where(NewsSummary.id.in_(package_ids)).values(state='pending')) + for empty in (None, null()): + session.add(NewsSummary(title='Unreviewed',fact='Pending',category='security',entities=[], + published_at=now,updated_at=now,changed_at=now,due_at=now,state='pending',result=empty)) + session.add(NewsEvent(title='Unreviewed',updated_at=now,due_at=now,state='pending',result=empty)) + for empty in (None, null(), {'inputs':None}): + session.add(RiskSnapshot(state='done',result=empty,quotes={},model='synthetic')) + await session.flush() + inputs = await events.reading_inputs(session,now-timedelta(hours=72)) + assert len(briefing.semantic_packages(inputs)) == 2 + assert {(p['kind'],p['id']) for p in inputs} == {('package',i) for i in package_ids} + from alphapulse.api import risk as api + app = FastAPI() + app.include_router(api.router) + app.dependency_overrides[api.require_admin] = lambda: None + app.dependency_overrides[api.get_session] = lambda: session + async with httpx.AsyncClient(transport=httpx.ASGITransport(app),base_url='http://ci') as client: + for hours in (24,48,72,168): + response = await client.get('/api/risk/overview',params={'hours':hours}) + assert response.status_code == 200, response.text + data = response.json()['data'] + assert data['briefing'] is None and data['stale'] + assert sum(data['counts'].values()) == 2 + response = await client.get('/api/risk/summaries') + assert response.status_code == 200, response.text + data = response.json()['data'] + assert data['total'] == 2 and {p['id'] for p in data['items']} == set(package_ids) + await nested.rollback() + print('Reading API: SQL NULL and JSON null results excluded; pending events preserve independent packages') async def complete(provider,system,payload,schema): points=[] for p in payload['packages']: