diff --git a/backend/alembic/versions/20261101_f60123456789_backfill.py b/backend/alembic/versions/20261101_f60123456789_backfill.py new file mode 100644 index 0000000..09e1ccd --- /dev/null +++ b/backend/alembic/versions/20261101_f60123456789_backfill.py @@ -0,0 +1,32 @@ +"""Normalize historical items and track the batch's downstream targets.""" +from alembic import op +import sqlalchemy as sa +revision = 'f60123456789' +down_revision = 'e5f701234567' +branch_labels = depends_on = None + + +def upgrade(): + op.create_table('news_ai_reprocess_items', + sa.Column('id', sa.BigInteger(), primary_key=True), + sa.Column('job_id', sa.BigInteger(), sa.ForeignKey('news_ai_reprocess_jobs.id'), nullable=False, index=True), + sa.Column('news_id', sa.BigInteger(), sa.ForeignKey('news_items.id'), nullable=False, index=True), + sa.Column('content_hash', sa.String(64), nullable=False), + sa.Column('mode', sa.String(16), nullable=False), + sa.Column('state', sa.String(16), nullable=False, index=True), sa.Column('error', sa.Text()), + sa.Column('started_at', sa.DateTime(timezone=True)), sa.Column('finished_at', sa.DateTime(timezone=True)), + sa.UniqueConstraint('job_id', 'news_id')) + op.create_table('news_ai_reprocess_targets', + sa.Column('id', sa.BigInteger(), primary_key=True), + sa.Column('job_id', sa.BigInteger(), sa.ForeignKey('news_ai_reprocess_jobs.id'), nullable=False, index=True), + sa.Column('kind', sa.String(16), nullable=False), sa.Column('object_id', sa.BigInteger(), nullable=False), + sa.Column('revision', sa.Integer(), nullable=False), sa.UniqueConstraint('job_id', 'kind', 'object_id')) + op.execute("""INSERT INTO news_ai_reprocess_items(job_id,news_id,content_hash,mode,state) + SELECT j.id,(i->>'id')::bigint,i->>'hash','legacy',i->>'state' + FROM news_ai_reprocess_jobs j, jsonb_array_elements(j.items) i + JOIN news_items n ON n.id=(i->>'id')::bigint ON CONFLICT DO NOTHING""") + + +def downgrade(): + op.drop_table('news_ai_reprocess_targets') + op.drop_table('news_ai_reprocess_items') diff --git a/backend/src/alphapulse/api/risk.py b/backend/src/alphapulse/api/risk.py index 6eae324..034e49c 100644 --- a/backend/src/alphapulse/api/risk.py +++ b/backend/src/alphapulse/api/risk.py @@ -1,7 +1,7 @@ """Authenticated news risk research endpoints.""" from datetime import UTC, datetime, timedelta from enum import IntEnum -from typing import Annotated +from typing import Annotated, Literal from fastapi import APIRouter, Body, Depends, HTTPException, Query from pydantic import BaseModel, Field, model_validator from sqlalchemy import select, func, text @@ -195,7 +195,9 @@ async def overview(session: DB, hours: RiskWindow = RiskWindow.DAY): if runtime: concurrency.update(json.loads(runtime)) from alphapulse.models.risk import ReprocessJob jobs = (await session.scalars(select(ReprocessJob).where(ReprocessJob.state.in_(['pending', 'running'])))).all() + from alphapulse.models.risk import ReprocessItem historical_pending = sum(i['state'] == 'pending' for job in jobs for i in job.items) + historical_pending += await session.scalar(select(func.count()).select_from(ReprocessItem).where(ReprocessItem.job_id.in_([j.id for j in jobs if j.settings.get('request')]), ReprocessItem.state=='pending')) or 0 coverage_data = await coverage(session) concurrency['queued'] = historical_pending + sum(stage2.get(k, 0) for k in ('pending',)) + event_states.get('pending', 0) + coverage_data['pending'] return success({'concurrency': concurrency, 'event_states': event_states, 'stage2': stage2, 'stage3': stage3, 'as_of': now, 'counts': counts, 'coverage': coverage_data, 'scenario': scenario, 'briefing': snapshot.result if snapshot else None, @@ -372,11 +374,13 @@ async def event_timeline(identifier: int, session: DB): class ReprocessIn(BaseModel): + modes: list[Literal['missing','review']] = Field(min_length=1,max_length=2) start: datetime end: datetime @model_validator(mode='after') def range_valid(self): + if len(set(self.modes)) != len(self.modes): raise ValueError('回补选项不能重复') if self.start.tzinfo is None or self.end.tzinfo is None or self.start >= self.end: raise ValueError('请提供带时区的有效起止时间') return self @@ -384,15 +388,19 @@ class ReprocessIn(BaseModel): @router.post('/reprocess/preview') async def reprocess_preview(body: ReprocessIn, session: DB): - from alphapulse.news_risk.reprocess import preview - counts, _ = await preview(session, body.start, body.end) - return success(counts) + from alphapulse.news_risk.backfill import preview + try: return success(await preview(session,body.start,body.end,body.modes)) + except ValueError as error: raise HTTPException(409,str(error)) from error + + +class ReprocessStart(BaseModel): + preview_id: str = Field(min_length=20,max_length=80,pattern='^[A-Za-z0-9_-]+$') @router.post('/reprocess', status_code=202) -async def reprocess_start(body: ReprocessIn, session: DB): - from alphapulse.news_risk.reprocess import start_job - try: result = await start_job(session, body.start, body.end) +async def reprocess_start(body: ReprocessStart, session: DB): + from alphapulse.news_risk.backfill import start + try: result = await start(session, body.preview_id) except ValueError as error: raise HTTPException(409, str(error)) from error return success(result) @@ -401,8 +409,8 @@ async def reprocess_start(body: ReprocessIn, session: DB): async def reprocess_list(session: DB): from alphapulse.models.risk import ReprocessJob rows = (await session.scalars(select(ReprocessJob).order_by(ReprocessJob.id.desc()).limit(20))).all() - return success([{'id': r.id, 'state': r.state, 'total': len(r.items), 'completed': r.position, 'failures': r.failures, - 'created_at': r.created_at} for r in rows]) + from alphapulse.news_risk.queue import job_out + return success([await job_out(session,r) for r in rows]) @router.post('/reprocess/{identifier}/cancel') @@ -423,6 +431,10 @@ async def reprocess_retry(identifier: int, session: DB): if job is None or job.state != 'failed': raise HTTPException(409, '仅可重试失败任务') if await session.scalar(select(ReprocessJob.id).where(ReprocessJob.state.in_(['pending', 'running'])).limit(1)): raise HTTPException(409, '已有重整理任务') + if job.settings.get('request'): + from alphapulse.models.risk import ReprocessItem + from sqlalchemy import update + await session.execute(update(ReprocessItem).where(ReprocessItem.job_id==job.id,ReprocessItem.state=='failed').values(state='pending',error=None)) outlook_ids = {f['id'] for f in job.failures if f['stage'] == 'outlook'} snapshots = (await session.scalars(select(RiskSnapshot).where(RiskSnapshot.id.in_(outlook_ids), RiskSnapshot.state == 'failed', RiskSnapshot.quotes['job_id'].as_integer() == job.id))).all() @@ -442,3 +454,24 @@ async def reprocess_retry(identifier: int, session: DB): job.position = len(job.items) job.failures = []; job.state = 'running' await session.commit(); return success(None) + + +@router.get('/queue') +async def queue(session: DB, state: Literal['running','queued','waiting','failed','done'] | None = None, + stage: Literal['grouping','analysis','events','outlook','backfill'] | None = None, + origin: Literal['history','realtime'] | None = None, job_id: int | None = Query(None,gt=0), + page: int = Query(1,ge=1), limit: int = Query(20,ge=1,le=100)): + from alphapulse.news_risk.queue import listing + return success(await listing(session,state,stage,origin,job_id,page,limit)) + + +@router.get('/reprocess/{identifier}') +async def reprocess_detail(identifier: int, session: DB, page: int = Query(1,ge=1), limit: int = Query(20,ge=1,le=100)): + from alphapulse.models.risk import ReprocessJob, ReprocessItem + from alphapulse.news_risk.queue import job_out + job=await session.get(ReprocessJob,identifier) + if job is None: raise HTTPException(404,'回补批次不存在') + rows=(await session.execute(select(ReprocessItem,NewsItem.title,NewsItem.url).join(NewsItem).where( + ReprocessItem.job_id==identifier).order_by(ReprocessItem.id).offset((page-1)*limit).limit(limit))).all() + return success({**await job_out(session,job),'items':[{'id':i.id,'news_id':i.news_id,'title':title,'url':url, + 'mode':i.mode,'state':i.state,'error':i.error} for i,title,url in rows]}) diff --git a/backend/src/alphapulse/models/risk.py b/backend/src/alphapulse/models/risk.py index 0a39327..a498062 100644 --- a/backend/src/alphapulse/models/risk.py +++ b/backend/src/alphapulse/models/risk.py @@ -224,3 +224,27 @@ class EventLink(Base): event_id: Mapped[int] = mapped_column(ForeignKey('news_ai_events.id'), primary_key=True) reason: Mapped[str] = mapped_column(Text) package_revision: Mapped[int] = mapped_column(Integer) + + +class ReprocessItem(Base): + __tablename__ = 'news_ai_reprocess_items' + __table_args__ = (UniqueConstraint('job_id', 'news_id'),) + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + job_id: Mapped[int] = mapped_column(ForeignKey('news_ai_reprocess_jobs.id'), index=True) + news_id: Mapped[int] = mapped_column(ForeignKey('news_items.id'), index=True) + content_hash: Mapped[str] = mapped_column(String(64)) + mode: Mapped[str] = mapped_column(String(16)) + state: Mapped[str] = mapped_column(String(16), default='pending', index=True) + error: Mapped[str | None] = mapped_column(Text) + started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + + +class ReprocessTarget(Base): + __tablename__ = 'news_ai_reprocess_targets' + __table_args__ = (UniqueConstraint('job_id', 'kind', 'object_id'),) + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + job_id: Mapped[int] = mapped_column(ForeignKey('news_ai_reprocess_jobs.id'), index=True) + kind: Mapped[str] = mapped_column(String(16)) + object_id: Mapped[int] = mapped_column(BigInteger) + revision: Mapped[int] = mapped_column(Integer) diff --git a/backend/src/alphapulse/news_risk/backfill.py b/backend/src/alphapulse/news_risk/backfill.py new file mode 100644 index 0000000..1869cf4 --- /dev/null +++ b/backend/src/alphapulse/news_risk/backfill.py @@ -0,0 +1,171 @@ +"""Snapshot-selected historical completion and explicit full rechecks.""" +import json +import secrets +from datetime import UTC, datetime +from sqlalchemy import select, func, text, exists, or_, any_, literal, BigInteger +from sqlalchemy.dialects.postgresql import insert, ARRAY +from alphapulse.core.db import SessionLocal +from alphapulse.core.redis import redis +from alphapulse.models import NewsItem +from alphapulse.models.risk import (RiskSettings, ReprocessJob, ReprocessItem, ReprocessTarget, + ArticleAnalysis, SummaryArticle, NewsSummary, NewsEvent, EventMember) +from alphapulse.news_risk.reprocess import capture + + +async def classification(session, news): + article = await session.scalar(select(ArticleAnalysis).where(ArticleAnalysis.news_id == news.id, ArticleAnalysis.content_hash == news.content_hash)) + packages = [] if article is None else (await session.scalars(select(NewsSummary).join(SummaryArticle).where( + SummaryArticle.analysis_id == article.id, SummaryArticle.active.is_(True), NewsSummary.replacement_ids == [], NewsSummary.state != 'withdrawn'))).all() + complete = bool(article and article.result and packages and all(p.result and p.published_revision == p.revision for p in packages)) + return complete, article, packages + + +async def preview(session, start, end, modes): + linked = select(NewsSummary.id).join(SummaryArticle).where( + SummaryArticle.analysis_id == ArticleAnalysis.id, SummaryArticle.active.is_(True), + NewsSummary.replacement_ids == [], NewsSummary.state != 'withdrawn') + incomplete = linked.where(or_(NewsSummary.result.is_(None), func.jsonb_typeof(NewsSummary.result) == 'null', + NewsSummary.published_revision != NewsSummary.revision)) + complete = exists(select(ArticleAnalysis.id).where(ArticleAnalysis.news_id == NewsItem.id, + ArticleAnalysis.content_hash == NewsItem.content_hash, ArticleAnalysis.result.is_not(None), + func.jsonb_typeof(ArticleAnalysis.result) != 'null', exists(linked), ~exists(incomplete))) + rows = (await session.execute(select(NewsItem.id, NewsItem.content_hash, complete.label('complete')).where( + NewsItem.published_at >= start, NewsItem.published_at < end, NewsItem.published_at <= datetime.now(UTC)) + .order_by(NewsItem.published_at, NewsItem.id))).all() + counts = {'total':len(rows), 'missing':0, 'review':0} + items = [] + for row in rows: + mode = 'review' if row.complete else 'missing' + counts[mode] += 1 + if mode in modes: items.append({'id':row.id, 'hash':row.content_hash, 'mode':mode}) + packages = select(NewsSummary.id).join(SummaryArticle).join(ArticleAnalysis).join(NewsItem).where( + NewsItem.id == any_(literal([i['id'] for i in items],type_=ARRAY(BigInteger))), ArticleAnalysis.content_hash == NewsItem.content_hash, + SummaryArticle.active.is_(True), NewsSummary.replacement_ids == [], NewsSummary.state != 'withdrawn').distinct() + counts['articles'] = len(items) + counts['packages'] = await session.scalar(select(func.count()).select_from(packages.subquery())) + counts['events'] = await session.scalar(select(func.count(func.distinct(EventMember.event_id))).where(EventMember.package_id.in_(packages))) + settings = await session.get(RiskSettings, 1) + if settings is None: raise ValueError('请先配置新闻AI') + config = await capture(session, settings) + identifier = secrets.token_urlsafe(24) + value = {'items':items, 'config':config, 'request':{'start':start.isoformat(),'end':end.isoformat(),'modes':modes}} + await redis.set('ap:risk:preview:'+identifier, json.dumps(value, default=str), ex=900) + return {**counts, 'preview_id':identifier, 'models':{k:{field:v['provider'][field] for field in ('model','reasoning_effort')} for k,v in config['configurations'].items()}} + + +async def start(session, identifier): + raw = await redis.get('ap:risk:preview:'+identifier) + if not raw: raise ValueError('预览已过期，请重新预览') + value = json.loads(raw) + await session.execute(text('SELECT pg_advisory_xact_lock(20261101, 2)')) + if await session.scalar(select(ReprocessJob.id).where(ReprocessJob.state.in_(['pending','running']))): + raise ValueError('已有回补任务') + settings = await session.get(RiskSettings, 1) + if settings is None or await capture(session, settings) != value['config']: + raise ValueError('模型或Skill配置已变化，请重新预览') + if not value['items']: raise ValueError('没有符合选择条件的新闻') + job = ReprocessJob(state='pending', items=[], position=0, failures=[], settings=value['config']) + session.add(job); await session.flush() + job.settings = {**job.settings, 'job_id':job.id, 'request':value['request']} + session.add_all([ReprocessItem(job_id=job.id, news_id=i['id'], content_hash=i['hash'], mode=i['mode'], state='pending') for i in value['items']]) + await session.commit() + await redis.delete('ap:risk:preview:'+identifier) + return {'id':job.id} + + +async def target(session, job_id, kind, obj): + await session.execute(insert(ReprocessTarget).values(job_id=job_id, kind=kind, object_id=obj.id, revision=obj.revision) + .on_conflict_do_update(index_elements=['job_id','kind','object_id'], set_={'revision':obj.revision})) + + +async def process(identifier): + from alphapulse.news_risk import worker + async with SessionLocal() as session: + item = await session.get(ReprocessItem, identifier, with_for_update=True) + if item is None or item.state != 'pending': return + job = await session.get(ReprocessJob, item.job_id, with_for_update=True) + settings = await session.get(RiskSettings, 1) + if job.state not in ('pending','running') or not settings.enabled: return + job_id, news_id, mode, content_hash = item.job_id, item.news_id, item.mode, item.content_hash + item.state = 'processing'; item.error = None; item.finished_at = None; item.started_at = datetime.now(UTC); job.state = 'running' + await session.commit() + try: + news = await session.get(NewsItem, item.news_id) + if news is None or news.content_hash != item.content_hash: raise ValueError('原文版本变化，请重新预览范围') + complete, article, packages = await classification(session, news) + if mode == 'missing' and complete: + item.state = 'skipped'; item.finished_at = datetime.now(UTC); await session.commit(); return + needs_grouping = mode == 'review' or not article or not article.result or not packages + if article is None: + article = ArticleAnalysis(news_id=news.id, content_hash=news.content_hash, state='reprocess', input={ + 'title':news.title,'summary':news.summary,'content':news.content_text,'url':news.url,'source':news.source_code, + 'source_tags':news.tags,'revision_no':news.revision_no,'published_at':news.published_at.isoformat(), + 'progress_at':max(news.published_at,news.source_updated_at or news.published_at).isoformat()}) + session.add(article); await session.flush() + article_id = article.id + await session.commit() + if needs_grouping: + outcome = await worker.analyze(article_id, job.settings) + if outcome == 'deferred': + item.state = 'pending'; await session.commit(); return + if outcome != 'done': + row = await session.get(ArticleAnalysis, article_id, populate_existing=True) + raise ValueError(row.error or '原文已变化，分包未完成') + await session.rollback() + job = await session.get(ReprocessJob, job_id, populate_existing=True) + news = await session.get(NewsItem,news_id,populate_existing=True) + if news.content_hash != content_hash: raise ValueError('原文版本变化，请重新预览范围') + _, _, packages = await classification(session,news) + for package in packages: + if package.result and package.published_revision == package.revision and mode == 'missing': continue + previous = await session.scalar(select(ReprocessTarget).where(ReprocessTarget.job_id==job.id, + ReprocessTarget.kind=='analysis',ReprocessTarget.object_id==package.id)) + if previous is None or package.revision > previous.revision: + if package.state not in ('pending','processing'): + worker.schedule(package,datetime.now(UTC),True) + package.review_config = job.settings + await target(session,job.id,'analysis',package) + item = await session.get(ReprocessItem, identifier, populate_existing=True) + item.state = 'done'; item.finished_at = datetime.now(UTC) + await session.commit() + except Exception as error: + await session.rollback() + item = await session.get(ReprocessItem,identifier) + item.state='failed'; item.error=str(error)[:1500]; item.finished_at=datetime.now(UTC) + await session.commit() + + +async def reconcile(session, job): + job = await session.get(ReprocessJob,job.id,with_for_update={"key_share":True},populate_existing=True) + if job.state not in ('pending','running'): return + if await session.scalar(select(ReprocessItem.id).where(ReprocessItem.job_id==job.id,ReprocessItem.state.in_(['pending','processing'])).limit(1)): return + items = (await session.scalars(select(ReprocessItem).where(ReprocessItem.job_id==job.id,ReprocessItem.state=='failed'))).all() + targets = (await session.scalars(select(ReprocessTarget).where(ReprocessTarget.job_id==job.id))).all() + failures = [{'stage':'grouping','id':i.news_id,'error':i.error} for i in items if i.state=='failed'] + waiting = False + for row in targets: + obj = await session.get(NewsSummary if row.kind=='analysis' else NewsEvent,row.object_id) + if obj is None: continue + if row.kind=='analysis' and (obj.replacement_ids or obj.state=='withdrawn'): + for replacement in obj.replacement_ids: + linked=await session.get(NewsSummary,replacement) + await target(session,job.id,'analysis',linked) + if obj.replacement_ids: + await session.delete(row); waiting=True + continue + if obj.state=='retired': continue + if obj.state=='failed': failures.append({'stage':row.kind,'id':obj.id,'error':obj.error}) + elif row.kind == 'events' and await session.scalar(select(NewsSummary.id).join(EventMember).where( + EventMember.event_id == obj.id, NewsSummary.state == 'failed', NewsSummary.replacement_ids == []).limit(1)): + failures.append({'stage':'events','id':obj.id,'error':'组成包研判失败，请先重试失败包'}) + elif obj.state not in ('done','retired') or obj.published_revision < row.revision: waiting=True + if row.kind=='analysis': + membership=await session.get(EventMember,obj.id) + if membership: + event=await session.get(NewsEvent,membership.event_id) + await target(session,job.id,'events',event) + if not any(t.kind=='events' and t.object_id==event.id for t in targets): waiting=True + if event.state=='failed': failures.append({'stage':'events','id':event.id,'error':event.error}) + if waiting: await session.commit(); return + job.failures=list({(f['stage'],f['id']):f for f in failures}.values()); job.state='failed' if failures else 'done' + await session.commit() diff --git a/backend/src/alphapulse/news_risk/corrections.py b/backend/src/alphapulse/news_risk/corrections.py index 19b2966..480ccc4 100644 --- a/backend/src/alphapulse/news_risk/corrections.py +++ b/backend/src/alphapulse/news_risk/corrections.py @@ -67,6 +67,9 @@ async def apply_correction(session, proposal: Correction, current_id, researched link = SummaryArticle(summary_id=package.id, analysis_id=assignment.analysis_id); session.add(link) link.active = True; link.unit = {'fact':replacement.fact,'evidence':assignment.evidence,'correction':True} schedule(package, now, True); outputs.append(package.id) + if package.review_config and package.review_config.get('request'): + from alphapulse.news_risk.backfill import target + await target(session, package.review_config['job_id'], 'analysis', package) for package in packages: if package.id not in outputs: package.replacement_ids = outputs; package.state = 'superseded' diff --git a/backend/src/alphapulse/news_risk/events.py b/backend/src/alphapulse/news_risk/events.py index 8d862d7..c1d144e 100644 --- a/backend/src/alphapulse/news_risk/events.py +++ b/backend/src/alphapulse/news_risk/events.py @@ -7,6 +7,7 @@ from alphapulse.core.db import SessionLocal from alphapulse.news_risk.pipeline import configuration from alphapulse.news_risk.contracts import PackageTimePoint from alphapulse.news_risk.service import complete +from alphapulse.news_risk.progress import step class TimelinePoint(PackageTimePoint): @@ -54,6 +55,9 @@ async def dirty_event(session, package_id, review_config=None): event.review_config = review_config event.state = 'pending' event.due_at = datetime.now(UTC)+timedelta(seconds=60) + if review_config and review_config.get('request'): + from alphapulse.news_risk.backfill import target + await target(session, review_config['job_id'], 'events', event) async def associate(session, package, result, researched_ids, researched_events=None): @@ -89,6 +93,9 @@ async def associate(session, package, result, researched_ids, researched_events= event.review_config = package.review_config event.state = 'pending'; event.due_at = datetime.now(UTC)+timedelta(seconds=60) event.updated_at = max(p.updated_at for p in members) + if package.review_config and package.review_config.get('request'): + from alphapulse.news_risk.backfill import target + await target(session, package.review_config['job_id'], 'events', event) async def organize(target_id=None): @@ -122,7 +129,9 @@ async def organize(target_id=None): version_id, event_id = version.id, event.id try: if config_error: raise ValueError(config_error) + step('梳理事件时间线') result, usage = await complete(provider, system, {'task': 'event_timeline', 'previous': event.result, 'packages': packages}, EventResult) + step('核对时间线与证据') validate_timeline(result, packages) version.result = result.model_dump(mode='json'); version.usage = usage; version.state = 'done' from alphapulse.news_risk.worker import publication_lock diff --git a/backend/src/alphapulse/news_risk/progress.py b/backend/src/alphapulse/news_risk/progress.py new file mode 100644 index 0000000..67aa6ec --- /dev/null +++ b/backend/src/alphapulse/news_risk/progress.py @@ -0,0 +1,27 @@ +"""Best-effort actual execution steps, scoped to background news tasks.""" +from contextvars import ContextVar +from datetime import UTC, datetime +import logging + +current = ContextVar('news_ai_progress', default=None) +active = {} +logger = logging.getLogger(__name__) + + +def step(phase, **details): + record = current.get() + if record is not None: + record.update(phase=phase, updated_at=datetime.now(UTC).isoformat(), **details) + + +async def execute(key, action): + identity = str(key) + record = {'key': identity, 'stage': key[0], 'object_id': key[1], 'phase': '准备输入', + 'started_at': datetime.now(UTC).isoformat(), 'updated_at': datetime.now(UTC).isoformat()} + active[identity] = record + token = current.set(record) + try: + return await action() + finally: + current.reset(token) + active.pop(identity, None) diff --git a/backend/src/alphapulse/news_risk/queue.py b/backend/src/alphapulse/news_risk/queue.py new file mode 100644 index 0000000..d8eaa82 --- /dev/null +++ b/backend/src/alphapulse/news_risk/queue.py @@ -0,0 +1,113 @@ +"""Queue presentation combines durable work with short-lived execution steps.""" +import json +from datetime import UTC, datetime +from sqlalchemy import select, func, or_, exists +from alphapulse.models import NewsItem +from alphapulse.models.risk import (ReprocessItem, ReprocessTarget, ArticleAnalysis, NewsSummary, NewsEvent, RiskSnapshot, RiskSettings, ReprocessJob) +from alphapulse.core.redis import redis + + +async def job_out(session, job): + counts = dict((await session.execute(select(ReprocessItem.state,func.count()).where(ReprocessItem.job_id==job.id).group_by(ReprocessItem.state))).all()) + if not job.settings.get('request'): + counts = {state:sum(i.get('state','pending')==state for i in job.items) for state in ('pending','processing','done','failed')} + return {'id':job.id,'state':job.state,'total':len(job.items),'completed':job.position,'failures':job.failures,'request':None,'counts':counts} + stages = {} + for kind, model in (('analysis',NewsSummary),('events',NewsEvent)): + stages[kind] = dict((await session.execute(select(model.state,func.count()).join( + ReprocessTarget,ReprocessTarget.object_id==model.id).where(ReprocessTarget.job_id==job.id, + ReprocessTarget.kind==kind).group_by(model.state))).all()) + return {'id':job.id,'state':job.state,'request':job.settings['request'],'counts':counts,'stages':stages, + 'total':sum(counts.values()),'completed':sum(counts.get(s,0) for s in ('done','failed','skipped')),'failures':job.failures} + + +async def listing(session, state, stage, origin, job_id, page, limit): + now = datetime.now(UTC) + runtime = await redis.get('ap:risk:concurrency') + runtime = json.loads(runtime) if runtime else None + raw = await redis.get('ap:risk:active') + active = json.loads(raw) if raw else None + settings = await session.get(RiskSettings,1) + entries=[] + total=0 + offset=(page-1)*limit + for kind,model in [('grouping',ArticleAnalysis),('analysis',NewsSummary),('events',NewsEvent),('outlook',RiskSnapshot),('backfill',ReprocessItem)]: + if stage and stage!=kind:continue + query=select(model) + busy = [a['object_id'] for a in (active or []) if a['stage']==kind] + if kind=='grouping': + held = exists(select(ReprocessItem.id).join(ReprocessJob).where(ReprocessItem.news_id==model.news_id, + ReprocessItem.content_hash==model.content_hash, ReprocessItem.state.in_(['pending','processing']), + ReprocessJob.state.in_(['pending','running']))) + running = model.id.in_(busy) if active is not None else ( + (model.state=='processing') & ~held.where(ReprocessItem.state=='processing')) + query=query.join(NewsItem).where(model.content_hash==NewsItem.content_hash,or_(~held,running)) + pending = model.state.in_(['pending','processing','reprocess']) + due = getattr(model,'due_at',None) + waiting = due > now if due is not None else False + if kind=='events': + from alphapulse.models.risk import EventMember + blocked=select(EventMember.event_id).join(NewsSummary,NewsSummary.id==EventMember.package_id).where( + NewsSummary.replacement_ids==[],NewsSummary.state!='withdrawn', + or_(NewsSummary.state!='done',NewsSummary.result==None)) + waiting=or_(waiting,model.id.in_(blocked)) + if state=='running': query=query.where(model.id.in_(busy) if active is not None else model.state=='processing') + elif state=='waiting': query=query.where(pending,model.id.not_in(busy),waiting) + elif state=='queued': query=query.where(pending,model.state!='processing' if active is None else True,model.id.not_in(busy),~waiting if due is not None else True) + elif state in ('failed','done'):query=query.where(model.state==state) + else:query=query.where(pending) + if kind=='backfill': + query=query.join(ReprocessJob,ReprocessJob.id==model.job_id).where(or_(ReprocessJob.state.in_(['pending','running']),model.state=='processing') if state not in ('failed','done') else True) + if origin=='realtime':continue + if job_id:query=query.where(model.job_id==job_id) + elif kind in ('analysis','events'): + batch=model.review_config['job_id'].as_integer() + if origin:query=query.where(batch.is_not(None) if origin=='history' else batch.is_(None)) + if job_id: + tracked=exists(select(ReprocessTarget.id).where(ReprocessTarget.job_id==job_id, + ReprocessTarget.kind==kind,ReprocessTarget.object_id==model.id)) + query=query.where(or_(tracked,batch==job_id)) + elif kind=='outlook': + if job_id:query=query.where(model.quotes['job_id'].as_integer()==job_id) + if origin:query=query.where(model.quotes['job_id'].as_integer().is_not(None) if origin=='history' else model.quotes['job_id'].as_integer().is_(None)) + elif job_id or origin=='history':continue + count = await session.scalar(select(func.count()).select_from(query.subquery())) + total += count + if offset >= count: + offset -= count; continue + if len(entries) == limit: continue + rows=(await session.scalars(query.order_by(model.id.desc()).offset(offset).limit(limit-len(entries)))).all() + offset=0 + for obj in rows: + url = None + if kind=='backfill': + news=await session.get(NewsItem,obj.news_id); title=news.title if news else f'新闻 #{obj.news_id}' + batch=obj.job_id + url=news.url if news else None + elif kind=='grouping':title=obj.input.get('title','');batch=None;url=obj.input.get('url') + elif kind=='outlook':title='72小时阅读与趋势';batch=obj.quotes.get('job_id') + else:title=obj.title;batch=(obj.review_config or {}).get('job_id') + source='history' if batch else 'realtime' + config_stage = 'grouping' if kind=='backfill' else 'analysis' if kind=='events' else kind + frozen = (await session.get(ReprocessJob,batch)).settings if batch else None + config = frozen['configurations'][config_stage]['provider'] if frozen else {} + if not config and settings: + prefix = {'grouping':'','analysis':'second_','outlook':'third_'}[config_stage] + from alphapulse.models import AiProvider + provider = await session.get(AiProvider,getattr(settings,prefix+'provider_id')) if getattr(settings,prefix+'provider_id') else None + config = {'model':getattr(settings,prefix+'model') or (provider.model if provider else None), + 'reasoning_effort':getattr(settings,prefix+'reasoning_effort') if getattr(settings,prefix+'reasoning_effort') is not None else provider.reasoning_effort if provider else None} + actual=next((a for a in (active or []) if a['stage']==kind and a['object_id']==obj.id),None) + waiting = getattr(obj,'due_at',now)>now + status='running' if actual else 'waiting' if waiting else 'queued' if obj.state in ('pending','processing','reprocess') else obj.state + if state: status=state + if obj.state=='processing' and active is None: status='unknown' + entries.append({'key':f'{kind}:{obj.id}','stage':kind,'object_id':obj.id,'title':title,'job_id':batch, + 'url':url,'finished_at':getattr(obj,'finished_at',None),'origin':source,'state':status,'phase':actual.get('phase') if actual else '已完成' if status=='done' else '处理失败' if status=='failed' else '等待条件或聚合时间' if status=='waiting' else 'Worker心跳过期，执行状态未知' if status=='unknown' else 'AI已暂停' if settings and not settings.enabled else '等待调度', + 'started_at':actual.get('started_at') if actual else getattr(obj,'started_at',None), + 'updated_at':actual.get('updated_at') if actual else None,'round':actual.get('round') if actual else None, + 'tool':actual.get('tool') if actual else None,'model':(actual or {}).get('model') or config.get('model') or getattr(obj,'model',None), + 'reasoning_effort':actual.get('reasoning_effort',config.get('reasoning_effort')) if actual else config.get('reasoning_effort'),'error':getattr(obj,'error',None)}) + order = {'grouping':0,'analysis':1,'events':2,'outlook':3,'backfill':4} + entries.sort(key=lambda r:(order[r['stage']],-r['object_id'])) + return {'items':entries,'total':total,'as_of':now,'worker_alive':active is not None,'concurrency':runtime,'enabled':bool(settings and settings.enabled),'limit':settings.max_concurrency if settings else 5} diff --git a/backend/src/alphapulse/news_risk/reprocess.py b/backend/src/alphapulse/news_risk/reprocess.py index 4ecea13..c5b7e22 100644 --- a/backend/src/alphapulse/news_risk/reprocess.py +++ b/backend/src/alphapulse/news_risk/reprocess.py @@ -1,8 +1,8 @@ """Explicit bounded historical jobs, sharing the normal news pipeline.""" -from sqlalchemy import select, text +from sqlalchemy import select, text, update from alphapulse.core.db import SessionLocal from alphapulse.models import NewsItem -from alphapulse.models.risk import RiskSettings, ReprocessJob, ArticleAnalysis, NewsSummary, EventMember +from alphapulse.models.risk import RiskSettings, ReprocessJob, ReprocessItem, ArticleAnalysis, NewsSummary, EventMember from alphapulse.news_risk.pipeline import configuration @@ -21,6 +21,7 @@ async def capture(session, settings): from alphapulse.api.risk import config_out value = config_out(settings) value.pop('start_at') + value.pop('enabled', None) value.pop('max_concurrency', None) configurations = {} for stage in ('grouping', 'analysis', 'outlook'): @@ -80,6 +81,11 @@ async def process_job(item_index=None): if item.get('state') not in (None, 'pending'): return job.items = [{**entry, 'state': 'processing'} if n == index else entry for n, entry in enumerate(job.items)] job_id = job.id + durable = await session.scalar(select(ReprocessItem).where(ReprocessItem.job_id==job.id,ReprocessItem.news_id==item['id'])) + if durable: + durable.state='processing' + from alphapulse.news_risk.progress import step + step('历史新闻分包',stage='backfill',object_id=durable.id) try: for stage in ('grouping', 'analysis', 'outlook'): await frozen_configuration(session, job.settings, stage) @@ -98,6 +104,7 @@ async def process_job(item_index=None): if outcome == 'deferred': job = await session.get(ReprocessJob, job_id, populate_existing=True, with_for_update=True) job.items = [{**entry, 'state': 'pending'} if n == index else entry for n, entry in enumerate(job.items)] + await session.execute(update(ReprocessItem).where(ReprocessItem.job_id==job_id,ReprocessItem.news_id==item['id']).values(state='pending')) await session.commit(); return row = await session.get(ArticleAnalysis, row_id, populate_existing=True) if outcome != 'done' or row.state != 'done' or row.error: raise ValueError(row.error or '原文已变化') @@ -111,6 +118,7 @@ async def process_job(item_index=None): job = await session.get(ReprocessJob, job_id, populate_existing=True, with_for_update=True) job.items = [{**entry, 'state': terminal} if n == index else entry for n, entry in enumerate(job.items)] job.position = sum(entry.get('state') in ('done', 'failed') for entry in job.items) + await session.execute(update(ReprocessItem).where(ReprocessItem.job_id==job_id,ReprocessItem.news_id==item['id']).values(state=terminal)) await session.commit() diff --git a/backend/src/alphapulse/news_risk/research.py b/backend/src/alphapulse/news_risk/research.py index f39267f..f7efb02 100644 --- a/backend/src/alphapulse/news_risk/research.py +++ b/backend/src/alphapulse/news_risk/research.py @@ -124,6 +124,8 @@ class Research: async def tool(self, name, arguments): if name not in TOOLS: raise ValueError('未知研究工具') + from alphapulse.news_risk.progress import step + step('查询资料', tool=name) args = TOOLS[name][0].model_validate(arguments) key = name+json.dumps(arguments, sort_keys=True) if key in self.cache: return self.cache[key] @@ -196,7 +198,9 @@ class Research: submit = llm.ToolSpec('submit', '提交完整研判，引用实际读取来源', schema.model_json_schema()) messages = [{'role': 'user', 'text': json.dumps(payload, ensure_ascii=False)}] usages = [] + from alphapulse.news_risk.progress import step, current for round_no in range(9): + if current.get() is not None: step('模型研判', round=round_no+1, model=provider.model, reasoning_effort=provider.reasoning_effort) if not self.fits_messages(provider, system, messages, schema): raise ValueError('研究上下文不足，请增大上下文配置；证据未截断') completion = None diff --git a/backend/src/alphapulse/news_risk/scheduler.py b/backend/src/alphapulse/news_risk/scheduler.py index ddbe41c..7e3c25c 100644 --- a/backend/src/alphapulse/news_risk/scheduler.py +++ b/backend/src/alphapulse/news_risk/scheduler.py @@ -7,7 +7,7 @@ from datetime import UTC, datetime from sqlalchemy import select, text, func from alphapulse.core.db import SessionLocal from alphapulse.core.redis import redis -from alphapulse.models.risk import RiskSettings, NewsSummary, NewsEvent, ArticleAnalysis, ReprocessJob +from alphapulse.models.risk import RiskSettings, NewsSummary, NewsEvent, ArticleAnalysis, ReprocessJob, ReprocessItem logger = logging.getLogger(__name__) GATE = 20261102 @@ -18,7 +18,8 @@ class Pool: self.tasks = {} def start(self, key, action): - self.tasks[asyncio.create_task(action())] = key + from alphapulse.news_risk.progress import execute + self.tasks[asyncio.create_task(execute(key, action))] = key async def reap(self, timeout=1): if not self.tasks: @@ -50,6 +51,9 @@ async def candidate(active, cursor, allow_outlook=True): await worker.enqueue(session, settings) history_news = [] for kind, identity in active: + if kind == 'backfill': + item = await session.get(ReprocessItem, identity) + history_news.append(item.news_id) if kind == 'history' and identity[1] >= 0: job = await session.get(ReprocessJob, identity[0]) history_news.append(job.items[identity[1]]['id']) @@ -61,7 +65,12 @@ async def candidate(active, cursor, allow_outlook=True): busy = [key[1] for key in active if key[0] == stage] if stage == 'outlook': if not busy and allow_outlook: - return (stage, 0), worker.outlook, (cursor + offset + 1) % 4 + try: + ready = await worker.outlook(prepare_only=True) + except ValueError: + logger.exception('阅读编排配置检查失败') + ready = False + if ready: return (stage, 0), worker.outlook, (cursor + offset + 1) % 4 continue model = {'grouping': ArticleAnalysis, 'analysis': NewsSummary, 'events': NewsEvent}[stage] query = select(model.id).where(model.state == 'pending', model.id.not_in(busy)) @@ -81,17 +90,29 @@ async def candidate(active, cursor, allow_outlook=True): action = {'grouping': worker.analyze, 'analysis': worker.summarize, 'events': events.organize}[stage] return (stage, identifier), lambda action=action, identifier=identifier: action(identifier), (cursor + offset + 1) % 4 job = await session.scalar(select(ReprocessJob).where(ReprocessJob.state.in_(['pending', 'running'])).order_by(ReprocessJob.id).limit(1)) + if job and job.settings.get('request'): + from alphapulse.news_risk.backfill import process + busy = [k[1] for k in active if k[0]=='backfill'] + candidates = (await session.scalars(select(ReprocessItem).where(ReprocessItem.job_id==job.id, + ReprocessItem.state=='pending',ReprocessItem.id.not_in(busy),ReprocessItem.news_id.not_in(history_news)) + .order_by(ReprocessItem.id).limit(len(active)+1))).all() + for item in candidates: + article_id = await session.scalar(select(ArticleAnalysis.id).where(ArticleAnalysis.news_id==item.news_id, + ArticleAnalysis.content_hash==item.content_hash)) + if ('grouping',article_id) not in active: + return ('backfill',item.id), lambda identifier=item.id: process(identifier), cursor + return None if job: for index, item in enumerate(job.items): key = ('history', (job.id, index)) - if item.get('state', 'pending') == 'pending' and key not in active: + if item.get('state', 'pending') == 'pending' and key not in active and item['id'] not in history_news: # Real-time and historical grouping share the article identity. row_id = await session.scalar(select(ArticleAnalysis.id).where(ArticleAnalysis.news_id == item['id'], ArticleAnalysis.content_hash == item['hash'])) if row_id is not None and ('grouping', row_id) in active: continue return key, lambda index=index: reprocess.process_job(index), cursor if not any(k[0] == 'history' for k in active): - return ('history', (job.id, -1)), lambda: reprocess.process_job(-1), cursor + await reprocess.process_job(-1) return None @@ -103,6 +124,11 @@ async def run(lock): # Interrupted historical items are reclaimed independently of completion order. async with SessionLocal() as session: jobs = (await session.scalars(select(ReprocessJob).where(ReprocessJob.state.in_(['pending', 'running'])))).all() + from sqlalchemy import update + interrupted = (await session.scalars(select(ReprocessItem).where(ReprocessItem.state=='processing'))).all() + for item in interrupted: + await session.execute(update(ArticleAnalysis).where(ArticleAnalysis.news_id==item.news_id, ArticleAnalysis.content_hash==item.content_hash, ArticleAnalysis.state=='pending').values(state='reprocess')) + await session.execute(update(ReprocessItem).where(ReprocessItem.state=='processing').values(state='pending')) for job in jobs: for item in job.items: if item['state'] == 'processing': @@ -136,9 +162,16 @@ async def run(lock): async with SessionLocal() as session: settings = await session.get(RiskSettings, 1) limit, enabled = settings.max_concurrency, settings.enabled + async with SessionLocal() as session: + from alphapulse.news_risk.backfill import reconcile + jobs = (await session.scalars(select(ReprocessJob).where(ReprocessJob.state.in_(['pending','running'])))).all() + for job in jobs: + if job.settings.get('request'): await reconcile(session,job) now = datetime.now(UTC).isoformat() - counts = {stage: sum(k[0] == stage for k in pool.tasks.values()) for stage in ('grouping', 'analysis', 'events', 'outlook', 'history')} + counts = {stage: sum(k[0] == stage for k in pool.tasks.values()) for stage in ('grouping', 'analysis', 'events', 'outlook', 'history', 'backfill')} try: + from alphapulse.news_risk.progress import active + await redis.set('ap:risk:active', json.dumps(list(active.values())), ex=15) await redis.set('ap:risk:concurrency', json.dumps({'running': len(pool.tasks), 'stages': counts, 'updated_at': now}), ex=15) for stage in ('', 'analysis:', 'events:', 'outlook:', 'reprocess:'): await redis.set('ap:risk:' + stage + 'heartbeat', now, ex=15) diff --git a/backend/src/alphapulse/news_risk/service.py b/backend/src/alphapulse/news_risk/service.py index c61a153..00a69f8 100644 --- a/backend/src/alphapulse/news_risk/service.py +++ b/backend/src/alphapulse/news_risk/service.py @@ -76,6 +76,8 @@ class OutputTruncated(ValueError): async def complete(provider: AiProvider, system: str, payload: dict, schema: type[BaseModel]): + from alphapulse.news_risk.progress import step, current + if current.get() is not None: step('等待模型响应', model=provider.model, reasoning_effort=provider.reasoning_effort) text = json.dumps(payload, ensure_ascii=False) if not fits_context(provider, system, payload, schema): raise ValueError('新闻输入超出模型上下文；未截断或发送，请调整模型配置') diff --git a/backend/src/alphapulse/news_risk/worker.py b/backend/src/alphapulse/news_risk/worker.py index ea38b6b..3e11b8a 100644 --- a/backend/src/alphapulse/news_risk/worker.py +++ b/backend/src/alphapulse/news_risk/worker.py @@ -11,6 +11,7 @@ from alphapulse.news_risk.service import complete from alphapulse.news_risk.pipeline import configuration, Grouping, summarize_articles, validate_evidence, Evidence, grouping_payload, pack_candidates from alphapulse.news_risk.research import Research from alphapulse.news_risk import briefing +from alphapulse.news_risk.progress import step logger = logging.getLogger(__name__) publication_lock = asyncio.Lock() @@ -73,6 +74,7 @@ async def _analyze(target_id=None, frozen=None): row_id = row.id row_input = row.input try: + step('提取与归并新闻') payload, usages = await grouping_payload(provider, system, {'id': row.id, **row.input}, candidate_inputs) for attempt in range(2): try: @@ -87,6 +89,7 @@ async def _analyze(target_id=None, frozen=None): if attempt: raise payload = pack_candidates(provider, system, {**payload, 'correction': str(error)}, payload['candidates']) + step('校验并保存分包结果') async with publication_lock: # Check the article again before changing event membership. current = await session.get(NewsItem, row.news_id, populate_existing=True) @@ -133,6 +136,9 @@ async def _analyze(target_id=None, frozen=None): if event.id in changed: event.review_config = frozen schedule(event, now, any(m.urgent for m in result.packages)) + if frozen and frozen.get('request'): + from alphapulse.news_risk.backfill import target + await target(session, frozen['job_id'], 'analysis', event) session.add(GroupingVersion(analysis_id=row.id, input=row.input, result=result.model_dump(mode='json'), rules=rules, state='done')) row.result = result.model_dump(mode='json'); row.usage = {'attempts': usages}; row.state = 'done' row.error = None; row.finished_at = now @@ -201,6 +207,7 @@ async def summarize(target_id=None): try: if config_error: raise ValueError(config_error) result, usage = await summarize_articles(provider, system, articles, package, research) + step('校验并保存包研判') async with publication_lock: event = await session.get(NewsSummary, event_id, populate_existing=True, with_for_update=True) if event.replacement_ids or event.state == 'withdrawn': @@ -240,7 +247,7 @@ async def summarize(target_id=None): await session.commit(); logger.exception('事件总结失败') -async def outlook(): +async def outlook(prepare_only=False): async with SessionLocal() as session: settings = await session.get(RiskSettings, 1) if not settings or not settings.enabled: return @@ -252,6 +259,7 @@ async def outlook(): inputs = await reading_inputs(session, now-timedelta(hours=72)) from alphapulse.models.risk import ReprocessJob job = await session.scalar(select(ReprocessJob).where(ReprocessJob.state == 'running').limit(1)) + if job and job.settings.get('request'): job = None if job and job.position < len(job.items): if requested and requested.quotes.get('job_id') == job.id: return job = None @@ -266,7 +274,17 @@ async def outlook(): semantic = briefing.semantic_packages(inputs) key = briefing.fingerprint(semantic, rules) important = {p.get('kind', 'package')+':'+str(p['id']): briefing.fingerprint(p, {}) for p in semantic if p['importance_stars'] == 5} - if requested is None and latest and latest.quotes.get('fingerprint') == key and (job is None or latest.quotes.get('job_id') == job.id): return + unchanged = requested is None and latest and latest.quotes.get('fingerprint') == key and (job is None or latest.quotes.get('job_id') == job.id) + if unchanged and (inputs or latest.state == 'done'): return + if not inputs: + # Empty reading windows are durable housekeeping, without an AI slot. + session.add(RiskSnapshot(state='done', model=provider.model, + quotes={'fingerprint':key,'important':{},**({'job_id':job.id} if job else {})}, + result={'reading':[],'synthesis':None,'usage':[],'expansion':[],'expansion_usage':[], + 'inputs':[],'rules':rules,'as_of':now.isoformat()})) + if requested: requested.quotes = {k:v for k,v in requested.quotes.items() if k != 'retry_requested'} + await session.commit() + return if latest and requested is None and job is None: urgent_change = any(latest.quotes.get('important', {}).get(identifier) != value for identifier, value in important.items()) if urgent_change: @@ -277,8 +295,11 @@ async def outlook(): latest.quotes = {**latest.quotes, 'dirty_since': now.isoformat()} await session.commit(); return if (now-datetime.fromisoformat(dirty)).total_seconds() < 900: return + if prepare_only: return True + step('整理72小时阅读与趋势') snapshot = RiskSnapshot(state='processing', model=provider.model, quotes={'fingerprint': key, 'important': important, **({'job_id': job.id} if job else {})}) session.add(snapshot); await session.commit() + step('生成阅读与趋势', object_id=snapshot.id) try: main = [p for p in inputs if p['importance_stars'] >= 3] low = [p for p in inputs if p['importance_stars'] < 3] @@ -294,6 +315,7 @@ async def outlook(): if len(extra) == 3: break snapshot.result = {**result, 'expansion': extra, 'inputs': inputs, 'rules': rules, 'expansion_usage': expansion.get('usage', []), 'as_of': now.isoformat()} + step('保存阅读与趋势') snapshot.state = 'done' except Exception as error: snapshot.state = 'failed'; snapshot.error = str(error)[:2000] diff --git a/backend/tests/test_news_backfill.py b/backend/tests/test_news_backfill.py new file mode 100644 index 0000000..c1e628b --- /dev/null +++ b/backend/tests/test_news_backfill.py @@ -0,0 +1,63 @@ +"""Backfill selection and task progress use isolated model and storage boundaries.""" +import asyncio +from types import SimpleNamespace as N +from unittest.mock import AsyncMock, MagicMock +import pytest + + +def test_progress_is_scoped_and_removed_on_failure(): + from alphapulse.news_risk.progress import execute, step, active, current + async def check(): + async def task(): + step('查询资料',tool='read_news') + assert len(active)==1 and current.get()['tool']=='read_news' + raise ValueError('Synthetic') + with pytest.raises(ValueError): await execute(('analysis',7),task) + assert active=={} and current.get() is None + asyncio.run(check()) + + +@pytest.mark.parametrize('published,revisions,expected', [(True,(2,2),True),(True,(1,2),False),(False,(2,2),False)]) +def test_all_current_packages_require_published_results(trading,published,revisions,expected): + from alphapulse.news_risk.backfill import classification + package=N(result={'summary':'ok'} if published else None,published_revision=revisions[0],revision=revisions[1]) + session=N(scalar=AsyncMock(return_value=N(id=3,result={'packages':[]})),scalars=AsyncMock(return_value=N(all=lambda:[package]))) + actual,_,_=asyncio.run(classification(session,N(id=1,content_hash='current'))) + assert actual is expected + + +def test_empty_active_memberships_are_incomplete(trading): + from alphapulse.news_risk.backfill import classification + session=N(scalar=AsyncMock(return_value=N(id=3,result={'old':True})),scalars=AsyncMock(return_value=N(all=lambda:[]))) + assert asyncio.run(classification(session,N(id=1,content_hash='current')))[0] is False + + +def test_expired_preview_does_not_create_work(trading,monkeypatch): + from alphapulse.news_risk import backfill + monkeypatch.setattr(backfill.redis,'get',AsyncMock(return_value=None)) + session=MagicMock() + with pytest.raises(ValueError,match='预览已过期'):asyncio.run(backfill.start(session,'expired')) + session.add.assert_not_called() + + +@pytest.mark.parametrize('modes',[[],['missing','missing'],['review','other'],['other']]) +def test_range_rejects_empty_duplicate_or_unknown_modes(trading,modes): + from pydantic import ValidationError + from alphapulse.api.risk import ReprocessIn + with pytest.raises(ValidationError): + ReprocessIn(start='2026-10-01T00:00:00Z',end='2026-11-01T00:00:00Z',modes=modes) + + +@pytest.mark.parametrize('failure',[False,True]) +def test_empty_or_invalid_outlook_poll_does_not_admit_a_task(trading,monkeypatch,failure): + from alphapulse.news_risk import scheduler, worker + session=MagicMock() + session.__aenter__=AsyncMock(return_value=session); session.__aexit__=AsyncMock() + session.get=AsyncMock(return_value=N(enabled=True,start_at='configured')) + session.scalar=AsyncMock(return_value=None) + monkeypatch.setattr(scheduler,'SessionLocal',lambda:session) + monkeypatch.setattr(worker,'enqueue',AsyncMock()) + check=AsyncMock(side_effect=ValueError('Synthetic configuration failure') if failure else None,return_value=False) + monkeypatch.setattr(worker,'outlook',check) + assert asyncio.run(scheduler.candidate(set(),0)) is None + check.assert_awaited_once_with(prepare_only=True) diff --git a/backend/tests/test_news_concurrency.py b/backend/tests/test_news_concurrency.py index 7f41108..eabdd02 100644 --- a/backend/tests/test_news_concurrency.py +++ b/backend/tests/test_news_concurrency.py @@ -12,6 +12,7 @@ def fake_session(settings): session.get = AsyncMock(return_value=settings) session.scalars = AsyncMock(return_value=N(all=lambda: [])) session.commit = AsyncMock() + session.execute = AsyncMock() return session @@ -117,7 +118,7 @@ def test_history_completion_is_counted_independently_of_item_order(trading, monk items=[{'id':i,'hash':'current','state':'pending'} for i in (1,2,3)]) article = N(id=30, state='done', error=None) session = fake_session(None) - session.scalar = AsyncMock(side_effect=[job, article]) + session.scalar = AsyncMock(side_effect=[job, None, article]) session.get = AsyncMock(side_effect=lambda model,*a,**kw: { RiskSettings:N(enabled=True), ReprocessJob:job, ArticleAnalysis:article, NewsItem:N(id=3, content_hash='current')}[model]) diff --git a/backend/tests/test_news_reprocess.py b/backend/tests/test_news_reprocess.py index a3198bd..4f8e198 100644 --- a/backend/tests/test_news_reprocess.py +++ b/backend/tests/test_news_reprocess.py @@ -21,7 +21,7 @@ def test_pause_after_claim_does_not_count_old_success(trading, monkeypatch): job = N(id=20, state='pending', position=0, items=[{'id':21,'hash':'abc'}], failures=[], settings={}) old = N(id=25, state='done', error=None, result={'old_success':True}) session = session_mock() - session.scalar.side_effect = [None, None, job, old] + session.scalar.side_effect = [None, None, job, None, old] session.get.side_effect = lambda model, *a, **kw: { RiskSettings:N(enabled=True), NewsItem:N(id=21,content_hash='abc'), ArticleAnalysis:old, ReprocessJob:job}[model] paused = session_mock() @@ -59,7 +59,7 @@ def test_downstream_failure_does_not_mark_unrelated_news_failed(trading, monkeyp @pytest.mark.parametrize('stage', ['grouping', 'analysis', 'events', 'outlook']) def test_retry_resumes_only_failed_stage(trading, stage): from alphapulse.api.risk import reprocess_retry - job = N(id=10,state='failed',position=2,items=[{'id':11},{'id':12}], + job = N(id=10,state='failed',position=2,settings={},items=[{'id':11},{'id':12}], failures=[{'stage':stage,'id':11,'error':'Synthetic failure'}]) failed = N(id=11,state='failed',quotes={}) session = session_mock() @@ -81,7 +81,7 @@ def test_job_outlook_retry_waits_for_its_grouping_retry(trading, monkeypatch): from alphapulse.news_risk import reprocess, worker, events session = session_mock() session.get.return_value = N(enabled=True) - session.scalar.side_effect = [None, N(quotes={'job_id':10}), N(id=10,position=0,items=[{'id':11}])] + session.scalar.side_effect = [None, N(quotes={'job_id':10}), N(id=10,position=0,settings={},items=[{'id':11}])] monkeypatch.setattr(worker,'SessionLocal',lambda:session) monkeypatch.setattr(events,'reading_inputs',AsyncMock(return_value=[])) current = AsyncMock(side_effect=AssertionError('Must not use current settings for a job retry')) diff --git a/deploy/README.md b/deploy/README.md index cd7b800..a532ac4 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -122,3 +122,45 @@ individual states, so completion order does not change progress or skip unfinish Migration `e5f701234567` adds the limit and converts legacy historical checkpoints to item states. Deployment and execution of this migration require separate authorization. + +### Historical news AI backfill and queue + +The home page's **AI 历史回补** selects publication times in `[start, end)` using +stored current news revisions. `missing` completes unfinished grouping/package analysis +and reuses published successes; `review` explicitly repeats grouping/TAG and package +verification, preserving previous successful versions. Selecting both includes all +news in the range. Related event timelines are refreshed; a batch waits only for its +own package/event targets, while the home reading window stays at 72 hours. +Corrections register every resulting package in the same transaction, including new +branches when a split retains the original ID. An empty reading window saves a cleared +snapshot without a model call or concurrency slot. + +`POST /api/risk/reprocess/preview` accepts `start`, `end`, and `modes`. Its opaque +`preview_id` expires after 15 minutes and fixes news IDs, content hashes, models, +reasoning efforts, and published Skills. `POST /api/risk/reprocess` accepts that ID; +changed configuration requires a new preview, and changed source content fails its +individual item explicitly. A missing item completed before execution is skipped. +Only one batch runs at a time. Batches use spare slots in the shared concurrency +pool, respect global AI pause, support cancellation of further news admissions and +explicit retry of failed items, and reclaim interrupted news items after restart. +Already scheduled downstream package/event updates continue after cancellation. +An in-flight news item also keeps its identity reserved after cancellation, so a new +batch waits for that item to finish while unrelated news can still use spare slots. + +`GET /api/risk/reprocess/{id}` returns paginated per-news states and downstream +stage counts. `GET /api/risk/queue` filters by `state`, `stage`, `origin`, `job_id` +and page. The home queue refreshes every three seconds while visible. Actual model +waits, research rounds, tool calls and save steps are transient telemetry under +`ap:risk:active`; PostgreSQL remains the durable queue. Heartbeat expiry means unknown +execution state. Empty outlook polls and batch completion checks do not consume AI +slots or appear as running model tasks. +Batch filters include durable package/event targets even after real-time updates +change their execution configuration; displayed model/source metadata describes the +current execution. Open news details refresh with the batch list, retaining their page +and rejecting responses from an older selection or session. + +Migration `f60123456789` adds normalized batch news items and package/event targets, +retaining legacy job JSON and execution semantics. Isolated acceptance includes +`scripts/smoke_backfill.py`; it mocks all model/Redis I/O and checks mode behavior, +frozen previews, source changes, target completion and queue queries. Live migration, +deployment and real backfill execution require separate authorization. diff --git a/scripts/ci.py b/scripts/ci.py index ea003f1..6ce55f9 100644 --- a/scripts/ci.py +++ b/scripts/ci.py @@ -183,6 +183,8 @@ def smoke(build, label): input=(ROOT / 'scripts/smoke_risk.py').read_text()), flush=True) print(run('docker', 'exec', '-i', names['api'], 'python', '-', input=(ROOT / 'scripts/smoke_events.py').read_text()), flush=True) + print(run('docker', 'exec', '-i', names['api'], 'python', '-', + input=(ROOT / 'scripts/smoke_backfill.py').read_text()), flush=True) print(run('docker', 'exec', '-i', names['api'], 'python', '-', input=(ROOT / 'scripts/smoke_calendar.py').read_text()), flush=True) output = run('docker', 'exec', '-i', '-e', 'SMOKE_WEB=' + names['web'], names['api'], 'python', '-', diff --git a/scripts/smoke_backfill.py b/scripts/smoke_backfill.py new file mode 100644 index 0000000..6172ed1 --- /dev/null +++ b/scripts/smoke_backfill.py @@ -0,0 +1,340 @@ +"""Validate backfill modes and queue queries only in a disposable CI database.""" +import asyncio +import os +import json +from datetime import UTC, datetime, timedelta +from unittest.mock import AsyncMock, patch +from types import SimpleNamespace +from urllib.parse import urlparse +from sqlalchemy import select, delete, update, func +from alphapulse.core.db import SessionLocal, engine +from alphapulse.models import AiProvider, NewsItem, SkillVersion, SkillNode +from alphapulse.models.risk import (RiskSettings, ArticleAnalysis, SummaryArticle, NewsSummary, + GroupingVersion, ReprocessJob, ReprocessItem, ReprocessTarget, NewsEvent, EventMember, PackageCorrection, RiskSnapshot) +from alphapulse.ai.skills import create_node, add_version +from alphapulse.news_risk import backfill, pipeline, worker, queue, scheduler, events, briefing + + +async def main(): + url = urlparse(os.environ['DATABASE_URL']) + assert url.username == 'ci' and url.password == 'ci-only' and url.path == '/ci' + now = datetime.now(UTC) - timedelta(hours=3) + storage = {} + async def save(key, value, **_): storage[key] = value + async def remove(key): storage.pop(key, None) + fake_redis = type('Redis', (), {'set': AsyncMock(side_effect=save), 'get': AsyncMock(side_effect=lambda k: storage.get(k)), 'delete': AsyncMock(side_effect=remove)})() + async with SessionLocal() as session: + assert not await session.scalar(select(func.count()).select_from(RiskSettings)) + provider = AiProvider(name='backfill-ci', protocol='openai', base_url='https://example.invalid', + api_key_encrypted=b'fake', api_key_hint='fake', model='shared', max_tokens=8192, context_window=200000) + session.add(provider); await session.flush() + skills = {} + for stage, (name, content) in pipeline.TEMPLATES.items(): + node = await create_node(session, None, 'skill', name, scope='news') + version = await add_version(session, node, content, 'user', 'CI') + node.published_version_id = version.id; skills[stage+'_skill_id'] = node.id + session.add(RiskSettings(id=1, enabled=True, start_at=now+timedelta(days=1), provider_id=provider.id, + second_provider_id=provider.id, third_provider_id=provider.id, model='luna', second_model='reviewer', + third_model='senior', reasoning_effort='low', second_reasoning_effort='high', **skills)) + news_ids, package_ids = [], [] + for i in range(3): + news = NewsItem(source_code='wscn_news', external_id='ci-backfill-'+str(i), + url='https://example.invalid/'+str(i), title='Evidence '+str(i), summary='', + content_text='Evidence '+str(i), content_hash='ci-backfill-'+str(i), published_at=now+timedelta(minutes=i)) + session.add(news); await session.flush(); news_ids.append(news.id) + if i == 2: continue + article = ArticleAnalysis(news_id=news.id, content_hash=news.content_hash, state='done', result={'old':True}, + input={'title':news.title,'summary':'','content':news.content_text,'url':news.url,'source':'wscn_news', + 'revision_no':1,'published_at':news.published_at.isoformat(),'progress_at':news.published_at.isoformat()}) + session.add(article); await session.flush() + package = NewsSummary(title=news.title, fact=news.title, category='science', entities=[], tags=[], + published_at=now, updated_at=now, changed_at=now, due_at=now, revision=1, published_revision=1 if i==0 else 0, + state='done' if i==0 else 'failed', result={'summary':'Original'} if i==0 else None) + session.add(package); await session.flush(); package_ids.append(package.id) + session.add(SummaryArticle(summary_id=package.id, analysis_id=article.id, active=True)) + await session.commit() + grouping_calls = [] + async def group(provider, system, payload, schema): + grouping_calls.append(provider.model) + article = payload['article'] + return pipeline.Grouping(reason='Same concrete fact', packages=[dict(package_id=package_ids[1], group_key=None, + title='Evidence', fact='Evidence', evidence=article['title'], category='science', entities=[], tags=['science'], + urgent=False, change_kind='repost', content_kind='report')]), {} + async def prepare(start, end, modes): + async with SessionLocal() as session: return await backfill.preview(session, start, end, modes) + async def begin(value): + async with SessionLocal() as session: return (await backfill.start(session, value['preview_id']))['id'] + async def process_all(job_id): + async with SessionLocal() as session: + ids = (await session.scalars(select(ReprocessItem.id).where(ReprocessItem.job_id==job_id))).all() + for identifier in ids: await backfill.process(identifier) + with patch.object(backfill, 'redis', fake_redis), patch.object(queue, 'redis', fake_redis), patch.object(worker, 'complete', group): + window = (now-timedelta(seconds=1), now+timedelta(minutes=4)) + preview = await prepare(*window, ['missing']) + assert (preview['total'],preview['missing'],preview['review'],preview['articles']) == (3,2,1,2), preview + job_id = await begin(preview) + await process_all(job_id) + assert grouping_calls == ['luna'], grouping_calls + async with SessionLocal() as session: + items = (await session.scalars(select(ReprocessItem).where(ReprocessItem.job_id==job_id))).all() + assert all(i.state=='done' for i in items), [(i.state,i.error) for i in items] + targets = (await session.scalars(select(ReprocessTarget).where(ReprocessTarget.job_id==job_id))).all() + assert len(targets)==1 and targets[0].object_id==package_ids[1] + package = await session.get(NewsSummary, package_ids[1]); revision = package.revision + package.state='done'; package.result={'summary':'Verified'}; package.published_revision=revision + # An unrelated pending package must not hold this batch open. + other = NewsSummary(title='Unrelated',fact='Other',category='science',entities=[],published_at=now, + updated_at=now,changed_at=now,due_at=now,state='pending') + session.add(other); await session.commit() + job = await session.get(ReprocessJob,job_id); await backfill.reconcile(session,job) + assert job.state=='done', (job.state,job.failures) + assert (await queue.job_out(session,job))['completed']==2 + for state in ('running','queued','waiting','failed','done',None): + result = await queue.listing(session,state,None,None,None,1,2) + assert result['total']>=len(result['items']) + result = await queue.listing(session,'done','backfill','history',job_id,1,1) + next_page = await queue.listing(session,'done','backfill','history',job_id,2,1) + assert result['total']==2 and result['items'][0]['key']!=next_page['items'][0]['key'] + # Recheck must rerun grouping, preserve grouping history and refresh old/new targets. + preview = await prepare(now-timedelta(seconds=1),now+timedelta(seconds=1),['review']) + review_id = await begin(preview) + await process_all(review_id) + assert grouping_calls==['luna','luna'] + async with SessionLocal() as session: + item = await session.scalar(select(ReprocessItem).where(ReprocessItem.job_id==review_id)) + assert item.state=='done', item.error + assert await session.scalar(select(func.count()).select_from(GroupingVersion))==2 + targets = (await session.scalars(select(ReprocessTarget).where(ReprocessTarget.job_id==review_id))).all() + assert {t.object_id for t in targets}==set(package_ids) + for target in targets: + package=await session.get(NewsSummary,target.object_id) + package.state='done'; package.result={'summary':'Verified'}; package.published_revision=package.revision + event=NewsEvent(title='Related event',updated_at=now,due_at=now,state='pending',revision=1) + session.add(event); await session.flush() + session.add(EventMember(package_id=package_ids[1],event_id=event.id,reason='Synthetic')) + await session.commit() + job=await session.get(ReprocessJob,review_id); await backfill.reconcile(session,job) + assert job.state=='running' + event.state='done'; event.published_revision=1; await session.commit() + await backfill.reconcile(session,job); assert job.state=='done' + # The fixed preview refuses changed settings, and a changed source fails visibly. + preview = await prepare(now+timedelta(minutes=2),now+timedelta(minutes=3),['review']) + async with SessionLocal() as session: + settings=await session.get(RiskSettings,1); settings.model='changed'; await session.commit() + try: await backfill.start(session,preview['preview_id']) + except ValueError as error: assert '配置已变化' in str(error) + else: raise AssertionError('Changed configuration accepted') + settings.model='luna'; await session.commit() + changed_id = await begin(preview) + async with SessionLocal() as session: + news=await session.get(NewsItem,news_ids[2]); news.content_hash='changed'; await session.commit() + await process_all(changed_id) + async with SessionLocal() as session: + job=await session.get(ReprocessJob,changed_id); await backfill.reconcile(session,job) + item=await session.scalar(select(ReprocessItem).where(ReprocessItem.job_id==changed_id)) + assert job.state=='failed' and '版本变化' in item.error + assert grouping_calls==['luna','luna'] + # Missing items completed after preview are skipped, and pause/cancel stop admissions. + async with SessionLocal() as session: + package=await session.get(NewsSummary,package_ids[1]); package.published_revision=0; package.state='pending' + await session.commit() + preview=await prepare(now-timedelta(seconds=1),now+timedelta(seconds=1),['missing']) + skip_id=await begin(preview) + async with SessionLocal() as session: + settings=await session.get(RiskSettings,1); settings.enabled=False + package=await session.get(NewsSummary,package_ids[1]); package.published_revision=package.revision; package.state='done' + await session.commit() + await process_all(skip_id) + async with SessionLocal() as session: + item=await session.scalar(select(ReprocessItem).where(ReprocessItem.job_id==skip_id)); assert item.state=='pending' + settings=await session.get(RiskSettings,1); settings.enabled=True; await session.commit() + await process_all(skip_id) + async with SessionLocal() as session: + item=await session.scalar(select(ReprocessItem).where(ReprocessItem.job_id==skip_id)); assert item.state=='skipped' + job=await session.get(ReprocessJob,skip_id); await backfill.reconcile(session,job); assert job.state=='done' + # A failed member must fail its batch without waiting forever for the blocked event. + job=await session.get(ReprocessJob,review_id); job.state='running' + package=await session.get(NewsSummary,package_ids[1]); package.state='failed'; package.error='Synthetic failure' + event=await session.scalar(select(NewsEvent)); event.state='pending'; event.revision+=1 + await session.commit() + await backfill.reconcile(session,job) + assert job.state=='failed' and {f['stage'] for f in job.failures}=={'analysis','events'}, job.failures + from alphapulse.api.risk import reprocess_retry + await reprocess_retry(job.id,session) + assert (await session.scalar(select(ReprocessItem).where(ReprocessItem.job_id==job.id))).state=='done' + assert (await session.get(NewsSummary,package_ids[1])).state=='pending' + job.state='cancelled'; await session.commit() + preview=await prepare(now-timedelta(seconds=1),now+timedelta(seconds=1),['review']) + cancel_id=await begin(preview) + async with SessionLocal() as session: + item=await session.scalar(select(ReprocessItem).where(ReprocessItem.job_id==cancel_id)); item.state='processing' + article=await session.scalar(select(ArticleAnalysis).where(ArticleAnalysis.news_id==item.news_id)); article.state='pending' + await session.commit() + lock=SimpleNamespace(scalar=AsyncMock(return_value=True),execute=AsyncMock(),commit=AsyncMock()) + with patch.object(scheduler,'candidate',AsyncMock(side_effect=asyncio.CancelledError)): + try: await scheduler.run(lock) + except asyncio.CancelledError: pass + else: raise AssertionError('Restart test did not reach admission') + async with SessionLocal() as session: + item=await session.scalar(select(ReprocessItem).where(ReprocessItem.job_id==cancel_id)); assert item.state=='pending' + article=await session.scalar(select(ArticleAnalysis).where(ArticleAnalysis.news_id==item.news_id)); assert article.state=='reprocess' + job=await session.get(ReprocessJob,cancel_id); job.state='cancelled'; await session.commit() + await process_all(cancel_id) + async with SessionLocal() as session: + job=await session.get(ReprocessJob,cancel_id); await backfill.reconcile(session,job) + assert job.state=='cancelled' + assert grouping_calls==['luna','luna'] + # Cancelled calls drain before another batch can review the same news. + from alphapulse.api.risk import reprocess_cancel + async with SessionLocal() as session: + config=json.loads(json.dumps((await session.get(ReprocessJob,cancel_id)).settings)) + config['configurations']['grouping']['provider']['model']='luna-old' + old=ReprocessJob(state='pending',items=[],position=0,failures=[],settings=config) + session.add(old); await session.flush(); old.settings={**old.settings,'job_id':old.id} + news=await session.get(NewsItem,news_ids[0]) + old_item=ReprocessItem(job_id=old.id,news_id=news.id,content_hash=news.content_hash,mode='review',state='pending') + session.add(old_item); await session.commit() + old_id,old_item_id=old.id,old_item.id + entered=asyncio.Event(); release=asyncio.Event() + async def controlled(provider,system,payload,schema): + if provider.model=='luna-old': + entered.set(); await release.wait() + result,usage=await group(provider,system,payload,schema) + result.reason=provider.model + return result,usage + with patch.object(worker,'complete',controlled),patch.object(worker,'outlook',AsyncMock(return_value=False)): + old_task=asyncio.create_task(backfill.process(old_item_id)) + try: + async with asyncio.timeout(5): await entered.wait() + async with SessionLocal() as session: + await reprocess_cancel(old_id,session) + config=json.loads(json.dumps((await session.get(ReprocessJob,old_id)).settings)) + config['configurations']['grouping']['provider']['model']='luna-new' + new=ReprocessJob(state='pending',items=[],position=0,failures=[],settings=config) + session.add(new); await session.flush(); new.settings={**new.settings,'job_id':new.id} + retried=ReprocessItem(job_id=new.id,news_id=news_ids[0],content_hash='ci-backfill-0',mode='review',state='pending') + other_item=ReprocessItem(job_id=new.id,news_id=news_ids[1],content_hash='ci-backfill-1',mode='review',state='pending') + session.add_all([retried,other_item]); await session.commit() + new_id,retried_id,other_item_id=new.id,retried.id,other_item.id + active={('analysis',i) for i in (await session.scalars(select(NewsSummary.id))).all()} + active.update(('events',i) for i in (await session.scalars(select(NewsEvent.id))).all()) + picked=await scheduler.candidate(active|{('backfill',old_item_id)},0) + assert picked and picked[0]==('backfill',other_item_id), picked + assert 'luna-new' not in grouping_calls + finally: + release.set(); await old_task + picked=await scheduler.candidate(active,0) + assert picked and picked[0]==('backfill',retried_id), picked + await picked[1]() + async with SessionLocal() as session: + article=await session.scalar(select(ArticleAnalysis).where(ArticleAnalysis.news_id==news_ids[0])) + assert article.result['reason']=='luna-new' and article.model=='luna-new' + new=await session.get(ReprocessJob,new_id); new.state='cancelled'; await session.commit() + # A pending backfill must not hide real-time grouping that won admission. + async with SessionLocal() as session: + job=ReprocessJob(state='running',items=[],position=0,failures=[],settings=(await session.get(ReprocessJob,cancel_id)).settings) + session.add(job); await session.flush() + job.settings={**job.settings,'job_id':job.id} + news=await session.get(NewsItem,news_ids[0]) + item=ReprocessItem(job_id=job.id,news_id=news.id,content_hash=news.content_hash,mode='review',state='pending') + session.add(item) + article=await session.scalar(select(ArticleAnalysis).where(ArticleAnalysis.news_id==news.id)); article.state='processing' + await session.commit() + task={'stage':'grouping','object_id':article.id,'phase':'等待模型响应', + 'started_at':now.isoformat(),'updated_at':now.isoformat()} + storage['ap:risk:active']=json.dumps([task]) + result=await queue.listing(session,'running','grouping',None,None,1,20) + assert [r['object_id'] for r in result['items']]==[article.id] and result['total']==1 + storage.pop('ap:risk:active') + result=await queue.listing(session,'running','grouping',None,None,1,20) + assert len(result['items'])==1 and result['items'][0]['state']=='unknown' + item.state='processing'; await session.commit() + storage['ap:risk:active']=json.dumps([{**task,'stage':'backfill','object_id':item.id}]) + assert (await queue.listing(session,'running','grouping',None,None,1,20))['total']==0 + assert (await queue.listing(session,'running','backfill',None,job.id,1,20))['total']==1 + storage.pop('ap:risk:active') + # Batch membership follows durable targets; execution metadata remains real-time. + shared=await session.get(NewsSummary,package_ids[1]); shared.review_config=None; shared.state='pending'; shared.due_at=now + await backfill.target(session,job.id,'analysis',shared) + await session.commit() + result=await queue.listing(session,'queued','analysis',None,job.id,1,20) + assert result['total']==1 and result['items'][0]['object_id']==shared.id + assert result['items'][0]['origin']=='realtime' and result['items'][0]['model']=='reviewer' + assert (await queue.listing(session,'queued','analysis','realtime',job.id,1,20))['total']==1 + assert (await queue.listing(session,'queued','analysis','history',job.id,1,20))['total']==0 + event=await session.scalar(select(NewsEvent)); event.review_config=None; event.state='pending'; event.due_at=now + shared.state='done'; shared.published_revision=shared.revision + await backfill.target(session,job.id,'events',event) + await session.commit() + result=await queue.listing(session,'queued','events',None,job.id,1,20) + assert result['total']==1 and result['items'][0]['object_id']==event.id + await session.execute(delete(ReprocessTarget).where(ReprocessTarget.job_id==job.id)) + await session.commit() + # Retaining the original package ID during a split must track both branches. + article.state='done'; item.state='done' + package=NewsSummary(title='Two facts',fact='Two facts',category='science',entities=[],tags=['science'], + published_at=now,updated_at=now,changed_at=now,due_at=now,state='done',revision=1,published_revision=1, + result={'summary':'Original'},review_config=job.settings) + session.add(package); await session.flush() + session.add(SummaryArticle(summary_id=package.id,analysis_id=article.id,active=True)) + await session.flush() + await backfill.target(session,job.id,'analysis',package) + from alphapulse.news_risk.contracts import Correction + from alphapulse.news_risk.corrections import apply_correction + proposal=Correction(packages=[{'id':package.id,'revision':package.revision}],replacements=[ + {'keep_id':package.id,'title':'First fact','fact':'First fact','category':'science','tags':['first'], + 'content_kind':'report','assignments':[{'analysis_id':article.id,'evidence':article.input['content']}]}, + {'keep_id':None,'title':'Second fact','fact':'Second fact','category':'science','tags':['second'], + 'content_kind':'report','assignments':[{'analysis_id':article.id,'evidence':article.input['content']}]}], + reason='Two distinct facts in one article') + outputs=await apply_correction(session,proposal,package.id,set(),{'backfill-split':True}) + package.state='done'; package.published_revision=package.revision + await session.commit() + await backfill.reconcile(session,job) + assert job.state=='running' + assert set((await session.scalars(select(ReprocessTarget.object_id).where( + ReprocessTarget.job_id==job.id,ReprocessTarget.kind=='analysis'))).all())==set(outputs) + created=await session.get(NewsSummary,outputs[1]); created.state='failed'; created.error='Synthetic split failure' + await session.commit(); await backfill.reconcile(session,job) + assert job.state=='failed' and job.failures==[{'stage':'analysis','id':created.id,'error':created.error}] + await reprocess_retry(job.id,session) + assert (await session.get(NewsSummary,created.id)).state=='pending' + created.state='done'; created.published_revision=created.revision; created.result={'summary':'Verified split'} + await session.commit(); await backfill.reconcile(session,job); assert job.state=='done' + # An empty 72h window clears old reading immediately, without models or a pool slot. + async with SessionLocal() as session: + previous=RiskSnapshot(state='done',model='senior',quotes={'fingerprint':'obsolete','important':{}}, + result={'reading':[{'id':package.id}],'inputs':[{'id':package.id}],'synthesis':{'summary':'Obsolete'}}) + session.add(previous); await session.commit() + count=await session.scalar(select(func.count()).select_from(RiskSnapshot)) + no_model=AsyncMock(side_effect=AssertionError('Empty windows must not call a model')) + with patch.object(events,'reading_inputs',AsyncMock(return_value=[])),patch.object(briefing,'generate',no_model): + assert not await worker.outlook(prepare_only=True) + assert not await worker.outlook(prepare_only=True) + async with SessionLocal() as session: + latest=await session.scalar(select(RiskSnapshot).order_by(RiskSnapshot.id.desc()).limit(1)) + assert latest.state=='done' and latest.result['inputs']==[] and latest.result['reading']==[] + assert latest.result['expansion']==[] and latest.result['synthesis'] is None + assert await session.scalar(select(func.count()).select_from(RiskSnapshot))==count+1 + latest.state='failed'; latest.quotes={**latest.quotes,'retry_requested':True}; await session.commit() + failed_id=latest.id + assert not await worker.outlook(prepare_only=True) + async with SessionLocal() as session: + previous=await session.get(RiskSnapshot,failed_id); assert 'retry_requested' not in previous.quotes + assert await session.scalar(select(func.count()).select_from(RiskSnapshot))==count+2 + assert not await worker.outlook(prepare_only=True) + no_model.assert_not_awaited() + async with SessionLocal() as session: + for model in (ReprocessTarget,ReprocessItem,ReprocessJob,EventMember,NewsEvent,PackageCorrection,RiskSnapshot,GroupingVersion,SummaryArticle,NewsSummary,ArticleAnalysis): + await session.execute(delete(model)) + await session.execute(delete(NewsItem).where(NewsItem.id.in_(news_ids))) + await session.execute(delete(RiskSettings)) + await session.execute(update(SkillNode).where(SkillNode.id.in_(skills.values())).values(published_version_id=None)) + await session.execute(delete(SkillVersion).where(SkillVersion.skill_id.in_(skills.values()))) + await session.execute(delete(SkillNode).where(SkillNode.id.in_(skills.values()))) + await session.execute(delete(AiProvider).where(AiProvider.id==provider.id)); await session.commit() + await engine.dispose() + print('Historical backfill: selection, frozen preview, reuse, recheck, targets, source revisions and queue passed') + + +asyncio.run(main()) diff --git a/web/src/common/apis/risk/index.ts b/web/src/common/apis/risk/index.ts index 5b5a8a7..a8bfafb 100644 --- a/web/src/common/apis/risk/index.ts +++ b/web/src/common/apis/risk/index.ts @@ -15,9 +15,12 @@ export const createNewsSkillTemplatesApi = (refreshExisting = false) => request< export const listNewsEventsApi = (page = 1) => request<{ items: Risk.NewsEvent[], total: number }>({ url: "/risk/events", method: "get", params: { page } }) export const getNewsEventApi = (id: number) => request({ url: `/risk/events/${id}`, method: "get" }) export const previewReprocessApi = (data: Risk.ReprocessRange) => request({ url: "/risk/reprocess/preview", method: "post", data }) -export const startReprocessApi = (data: Risk.ReprocessRange) => request<{ id: number }>({ url: "/risk/reprocess", method: "post", data }) +export const startReprocessApi = (data: { preview_id: string }) => request<{ id: number }>({ url: "/risk/reprocess", method: "post", data }) export const listReprocessApi = () => request({ url: "/risk/reprocess", method: "get" }) export const cancelReprocessApi = (id: number) => request({ url: `/risk/reprocess/${id}/cancel`, method: "post" }) export const retryReprocessApi = (id: number) => request({ url: `/risk/reprocess/${id}/retry`, method: "post" }) export const saveRiskConcurrencyApi = (max_concurrency: number) => request<{ max_concurrency: number }>({ url: "/risk/settings/concurrency", method: "patch", data: { max_concurrency } }) + +export const listRiskQueueApi = (params: { state?: string, stage?: string, origin?: string, job_id?: number, page?: number }) => request({ url: "/risk/queue", method: "get", params }) +export const getReprocessDetailApi = (id: number, page = 1) => request({ url: `/risk/reprocess/${id}`, method: "get", params: { page } }) diff --git a/web/src/common/apis/risk/type.ts b/web/src/common/apis/risk/type.ts index c15d930..9f085f3 100644 --- a/web/src/common/apis/risk/type.ts +++ b/web/src/common/apis/risk/type.ts @@ -118,6 +118,17 @@ export interface EventDetail extends NewsEvent { packages: RiskEvent[] versions: { id: number, revision: number, state: string, error: string | null, result: NewsEvent["result"] }[] } -export interface ReprocessRange { start: string, end: string } -export interface ReprocessPreview { articles: number, packages: number, events: number } -export interface ReprocessJob { id: number, state: string, total: number, completed: number, failures: { stage: "grouping" | "analysis" | "events" | "outlook", id: number, error: string }[] } +export interface ReprocessRange { start: string, end: string, modes: ("missing" | "review")[] } +export interface ReprocessPreview { articles: number, packages: number, events: number, total: number, missing: number, review: number, preview_id: string, models: Record } +export interface ReprocessJob { request?: { start: string, end: string, modes: string[] }, counts?: Record, stages?: Record>, id: number, state: string, total: number, completed: number, failures: { stage: "grouping" | "analysis" | "events" | "outlook", id: number, error: string }[] } + +export interface QueueItem { key: string, stage: string, object_id: number, title: string, job_id: number | null, origin: string, state: string, phase: string, model: string | null, reasoning_effort: string | null, started_at: string | null, updated_at: string | null, finished_at: string | null, url: string | null, round: number | null, tool: string | null, error: string | null } +export interface Queue { + items: QueueItem[] + total: number + as_of: string + worker_alive: boolean + enabled: boolean + limit: number + concurrency: { running: number, stages: Record } | null +} diff --git a/web/src/pages/home/components/AiQueue.vue b/web/src/pages/home/components/AiQueue.vue new file mode 100644 index 0000000..9dc7c65 --- /dev/null +++ b/web/src/pages/home/components/AiQueue.vue @@ -0,0 +1,119 @@ + + + diff --git a/web/src/pages/home/components/EventResearch.vue b/web/src/pages/home/components/EventResearch.vue index 366d4d2..4300d42 100644 --- a/web/src/pages/home/components/EventResearch.vue +++ b/web/src/pages/home/components/EventResearch.vue @@ -14,8 +14,33 @@ const opened = ref(false) const detailError = ref(false) const detailId = ref(0) const range = ref<[Date, Date] | null>(null) +const modes = ref<("missing" | "review")[]>(["missing"]) const preview = ref(null) const jobs = ref([]) +const batch = ref> | null>(null) +const batchId = ref(0) +const batchPage = ref(1) +const batchLoading = ref(false) +const batchError = ref(false) +let batchSequence = 0 +async function openBatch(id: number, page = 1) { + if (batchId.value !== id) batch.value = null + batchId.value = id + batchPage.value = page + batchLoading.value = true + const version = user.sessionVersion + const seq = ++batchSequence + try { + const data = await api.getReprocessDetailApi(id, page) + if (!active || seq !== batchSequence || version !== user.sessionVersion) return + batch.value = data + batchError.value = false + } catch { + if (active && seq === batchSequence && version === user.sessionVersion) batchError.value = true + } finally { + if (active && seq === batchSequence && version === user.sessionVersion) batchLoading.value = false + } +} const busy = ref(false) const error = ref(false) let sequence = 0 @@ -32,6 +57,7 @@ async function load() { total.value = events.total jobs.value = runs error.value = false + if (batchId.value && !batchLoading.value) await openBatch(batchId.value, batchPage.value) } catch { if (active && seq === sequence && version === user.sessionVersion) error.value = true } @@ -53,12 +79,12 @@ async function open(id: number) { async function inspect() { if (!range.value || busy.value) return const version = user.sessionVersion - const data = { start: range.value[0].toISOString(), end: range.value[1].toISOString() } + const data = { start: range.value[0].toISOString(), end: range.value[1].toISOString(), modes: [...modes.value] } const op = ++operation busy.value = true try { const result = await api.previewReprocessApi(data) - if (active && version === user.sessionVersion && range.value && data.start === range.value[0].toISOString() && data.end === range.value[1].toISOString()) preview.value = result + if (active && version === user.sessionVersion && range.value && data.start === range.value[0].toISOString() && data.end === range.value[1].toISOString() && data.modes.join() === modes.value.join()) preview.value = result } finally { if (active && op === operation && version === user.sessionVersion) busy.value = false } @@ -66,13 +92,15 @@ async function inspect() { async function start() { if (!range.value || !preview.value || busy.value) return const version = user.sessionVersion - const data = { start: range.value[0].toISOString(), end: range.value[1].toISOString() } - await ElMessageBox.confirm(`重整理 ${preview.value.articles} 篇新闻将产生模型费用，关联包和事件也会更新。是否启动？`, "重整理") + const data = { start: range.value[0].toISOString(), end: range.value[1].toISOString(), modes: [...modes.value] } + const previewId = preview.value.preview_id + await ElMessageBox.confirm(`回补 ${preview.value.articles} 篇新闻将产生模型费用，关联包和事件也会更新。是否启动？`, "AI 历史回补") if (!active || version !== user.sessionVersion || busy.value) return const op = ++operation busy.value = true try { - await api.startReprocessApi(data) + if (!preview.value || preview.value.preview_id !== previewId || data.modes.join() !== modes.value.join() || !range.value || data.start !== range.value[0].toISOString() || data.end !== range.value[1].toISOString()) return + await api.startReprocessApi({ preview_id: previewId }) if (active && version === user.sessionVersion) { preview.value = null await load() @@ -86,11 +114,12 @@ async function action(id: number, retry: boolean) { await (retry ? api.retryReprocessApi(id) : api.cancelReprocessApi(id)) if (active && version === user.sessionVersion) await load() } -watch(range, () => { +watch([range, modes], () => { preview.value = null }) watch(page, load) watch(() => user.sessionVersion, () => { + batchSequence++ sequence++ detailSequence++ operation++ @@ -98,6 +127,10 @@ watch(() => user.sessionVersion, () => { detailError.value = false items.value = [] jobs.value = [] + batch.value = null + batchId.value = 0 + batchLoading.value = false + batchError.value = false detail.value = null preview.value = null opened.value = false @@ -114,6 +147,7 @@ onMounted(() => { }) onBeforeUnmount(() => { active = false + batchSequence++ sequence++ detailSequence++ clearInterval(timer) @@ -144,20 +178,43 @@ defineExpose({ open })
- 按时间范围重整理新闻 + AI 历史回补

按新闻发布时间选择范围，使用现存最新修订。不会还原历史时点；共享包和事件可能同时更新。AI 暂停时任务等待。

+ + + 补齐未完成研判 + + + 已研判重新核对（重新分包及核验） + + - + 预览范围
- #{{ job.id }} · {{ job.state }} · {{ job.completed }}/{{ job.total }} + #{{ job.id }} · {{ job.state }} · 新闻处理 {{ job.completed }}/{{ job.total }} +

+ {{ formatTime(job.request.start) }} 至 {{ formatTime(job.request.end) }} · {{ job.request.modes.map(m => m === 'missing' ? '补齐未完成' : '重新核对').join('、') }} +

+

+ 成功 {{ job.counts.done ?? 0 }} · 处理中 {{ job.counts.processing ?? 0 }} · 待处理 {{ job.counts.pending ?? 0 }} · 失败 {{ job.counts.failed ?? 0 }} · 跳过 {{ job.counts.skipped ?? 0 }} +

+

+ {{ stage === 'analysis' ? '包研判' : '事件时间线' }}：待处理 {{ counts.pending ?? 0 }} · 运行 {{ counts.processing ?? 0 }} · 完成 {{ counts.done ?? 0 }} · 失败 {{ counts.failed ?? 0 }} +

+ + 查看新闻明细 + 取消 @@ -168,6 +225,17 @@ defineExpose({ open }) {{ ({ grouping: "新闻初筛", analysis: "包研判", events: "事件时间线", outlook: "阅读编排" })[failure.stage] }} #{{ failure.id }}：{{ failure.error }}

+
+

回补 #{{ batchId }} 新闻明细

+ + + 刷新明细 + +

+ {{ item.title }}（新闻 #{{ item.news_id }}）· {{ item.mode === 'missing' ? '补齐' : item.mode === 'review' ? '重新核对' : '旧批次' }} · {{ item.state }} {{ item.error }} +

+ +
diff --git a/web/src/pages/home/index.vue b/web/src/pages/home/index.vue index 64d6e6c..d214e71 100644 --- a/web/src/pages/home/index.vue +++ b/web/src/pages/home/index.vue @@ -9,6 +9,7 @@ import * as api from "@/common/apis/risk" import { listProviderModelsApi, listProvidersApi } from "@/common/apis/settings" import { useUserStore } from "@/pinia/stores/user" +import AiQueue from "./components/AiQueue.vue" import EventResearch from "./components/EventResearch.vue" const eventResearch = ref | null>(null) @@ -257,6 +258,7 @@ onBeforeUnmount(() => {