"""Durable news AI stages dispatched through the shared concurrency pool.""" import asyncio import logging from datetime import UTC, datetime, timedelta from sqlalchemy import BigInteger, select, update, func, exists, case from sqlalchemy.dialects.postgresql import insert from alphapulse.core.db import SessionLocal from alphapulse.models import NewsItem from alphapulse.models.risk import RiskSettings, RiskSnapshot, ArticleAnalysis, NewsSummary, SummaryArticle, SummaryVersion, PackageEvidence, GroupingVersion, NewsEvent, EventMember from alphapulse.news_risk.service import complete, accumulate_usage 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() def eligible(start): return (func.greatest(NewsItem.published_at, NewsItem.source_updated_at) >= start) & (NewsItem.published_at <= datetime.now(UTC)) def schedule(event, now, urgent): event.revision += 1 event.changed_at = now event.dirty_since = event.dirty_since or now event.due_at = now if urgent or event.result is None else min(now+timedelta(seconds=60), event.dirty_since+timedelta(seconds=180)) event.state = 'pending' event.error = None async def enqueue(session, settings): rows = (await session.scalars(select(NewsItem).where(eligible(settings.start_at), ~exists(select(ArticleAnalysis.id).where(ArticleAnalysis.news_id == NewsItem.id, ArticleAnalysis.content_hash == NewsItem.content_hash))).order_by(NewsItem.published_at.desc()).limit(200))).all() for news in rows: await session.execute(insert(ArticleAnalysis).values(news_id=news.id, content_hash=news.content_hash, state='pending', input={'title': news.title, 'summary': news.summary, 'content': news.content_text, 'published_at': news.published_at.isoformat(), 'progress_at': max(news.published_at, news.source_updated_at or news.published_at).isoformat(), 'source_tags': news.tags, 'url': news.url, 'source': news.source_code, 'revision_no': news.revision_no}).on_conflict_do_nothing()) await session.commit() async def analyze(target_id=None, frozen=None): return await _analyze(target_id, frozen) async def _analyze(target_id=None, frozen=None): async with SessionLocal() as session: settings = await session.get(RiskSettings, 1) if not settings or not settings.enabled or not settings.start_at: return 'deferred' if target_id is None: await enqueue(session, settings) row = await session.scalar(select(ArticleAnalysis).join(NewsItem).where( ArticleAnalysis.id == target_id if target_id else ArticleAnalysis.state == 'pending', ArticleAnalysis.content_hash == NewsItem.content_hash ).order_by(NewsItem.published_at.desc()).limit(1)) if row is None: return 'missing' if frozen: from alphapulse.news_risk.reprocess import frozen_configuration provider, system, rules = await frozen_configuration(session, frozen, 'grouping') else: provider, system, rules, _ = await configuration(session, settings, 'grouping') catalog = await session.scalar(select(func.max(NewsSummary.id))) tag_score = sum((case((NewsSummary.tags.contains([tag.strip().casefold()]), 1), else_=0) for tag in row.input.get('source_tags', [])[:8] if isinstance(tag, str)), 0) from alphapulse.news_risk.retrieval import terms_for, matching # Untitled news still has source text; bound retrieval without changing model inputs. query = (row.input['title'].strip() or (row.input.get('summary') or '').strip() or (row.input.get('content') or '').strip())[:256] _, keyword_score = matching(NewsSummary, terms_for(query, row.input.get('source_tags', [])[:6])) candidates = (await session.scalars(select(NewsSummary).where(NewsSummary.replacement_ids == [], NewsSummary.state != 'withdrawn').order_by( (func.greatest(func.similarity(NewsSummary.title, query), func.similarity(NewsSummary.fact, query)) + tag_score + keyword_score).desc(), NewsSummary.updated_at.desc()).limit(80))).all() memberships = dict((await session.execute(select(EventMember.package_id, EventMember.event_id).where(EventMember.package_id.in_([p.id for p in candidates])))).all()) event_catalog = await session.scalar(select(func.max(NewsEvent.id))) event_candidates = list(await session.scalars(select(NewsEvent).where(NewsEvent.state != 'retired').order_by( func.similarity(NewsEvent.fact, query).desc(), NewsEvent.updated_at.desc()).limit(60))) event_inputs = [{'id':e.id,'title':e.title,'fact':e.fact,'tags':e.tags,'revision':e.revision} for e in event_candidates] candidate_inputs = [{'main_event_id': memberships.get(e.id), 'id': e.id, 'title': e.title, 'published_at': e.published_at.isoformat(), 'entities': e.entities, 'tags': e.tags, 'fact': e.fact, 'content_kind': e.content_kind} for e in candidates] had_success = row.result is not None previous_usage = row.usage row.state = 'processing'; row.started_at = datetime.now(UTC); row.rules = rules; row.model = provider.model await session.commit() row_id = row.id row_input = row.input usages = [] try: step('提取与归并新闻') payload, usages = await grouping_payload(provider, system, {'id': row.id, **row.input}, candidate_inputs, event_inputs) for attempt in range(2): try: result, usage = await complete(provider, system, payload, Grouping) usages.append(usage) for match in result.packages: if match.package_id is not None and match.package_id not in {e['id'] for e in payload['candidates']}: raise ValueError('模型引用未知候选包') if match.event.event_id is not None and match.event.event_id not in {e['id'] for e in payload['events']} | {e.get('main_event_id') for e in payload['candidates']}: raise ValueError('模型引用未知候选事件') validate_evidence([Evidence(article_id=row.id, quote=match.evidence)], {row.id: row.input}) break except ValueError as error: if attempt or getattr(error, "correction_attempted", False) or any( isinstance(u, dict) and len(u.get("rounds", [])) > 1 for u in usages): raise if getattr(error,'usage',None): usages.append(error.usage) 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) if current.content_hash != row.content_hash: row.state = 'superseded'; row.usage=accumulate_usage(previous_usage,{'attempts':usages}); await session.commit(); return 'superseded' if (any(m.package_id is None for m in result.packages) and catalog != await session.scalar(select(func.max(NewsSummary.id)))) or (any(m.event.event_id is None for m in result.packages) and event_catalog != await session.scalar(select(func.max(NewsEvent.id)))): row.state = 'reprocess' if frozen else 'pending'; row.usage=accumulate_usage(previous_usage,{'attempts':usages}); await session.commit(); return 'deferred' now = datetime.now(UTC) old_ids = select(SummaryArticle.summary_id).join(ArticleAnalysis).where(ArticleAnalysis.news_id == row.news_id) old_events = (await session.scalars(select(NewsSummary).where(NewsSummary.id.in_(old_ids), NewsSummary.replacement_ids==[], NewsSummary.state!="withdrawn") .order_by(NewsSummary.id).with_for_update().execution_options(populate_existing=True))).all() touched = {e.id: e for e in old_events} previous_ids = set(touched) await session.execute(update(SummaryArticle).where(SummaryArticle.analysis_id == row.id).values(active=False)) groups = {} event_groups = {} changed = set() assigned = set() for match in result.packages: event = await session.get(NewsSummary, match.package_id, with_for_update=True, populate_existing=True) if match.package_id else groups.get(match.group_key) if event is not None and (event.replacement_ids or event.state == 'withdrawn'): raise ValueError('候选包已被纠错或撤回，请重试初筛') progress = datetime.fromisoformat(row.input['progress_at']) published = datetime.fromisoformat(row.input['published_at']) if event is None: event = NewsSummary(title=match.title, category=match.category, entities=match.entities, published_at=published, updated_at=progress, changed_at=now, due_at=now, revision=0, published_revision=0, state='pending') session.add(event); await session.flush(); groups[match.group_key] = event event.tags = sorted(set(event.tags or []) | {t.strip().casefold() for t in match.tags if t.strip()}) if match.change_kind != 'repost' or not event.fact: event.content_kind = match.content_kind; event.fact = match.fact event.published_at = min(event.published_at, published) if match.change_kind != 'repost' or row.input['revision_no'] > 1: event.updated_at = max(event.updated_at, progress) link = await session.scalar(select(SummaryArticle).where(SummaryArticle.summary_id == event.id, SummaryArticle.analysis_id == row.id)) if link is None: link = SummaryArticle(summary_id=event.id, analysis_id=row.id); session.add(link) link.active = True; link.unit = match.model_dump() if match.change_kind != 'repost' or event.revision == 0 or row.input['revision_no'] > 1: changed.add(event.id) touched[event.id] = event assigned.add(event.id) from alphapulse.news_risk.event_research import attach await attach(session, event, match.event, event_groups, event.id in changed, match.urgent, frozen) await session.flush() changed.update(previous_ids - assigned) if row.input['revision_no'] > 1: changed.update(previous_ids) for event in touched.values(): if event.id in changed: event.revision += 1 event.changed_at = now event.state = 'done' await session.flush() active = await session.scalar(select(SummaryArticle.analysis_id).join(ArticleAnalysis).join(NewsItem).where( SummaryArticle.summary_id==event.id, SummaryArticle.active.is_(True), ArticleAnalysis.content_hash==NewsItem.content_hash).limit(1)) if active is None: event.state='withdrawn' # Removed/revised originals also invalidate the owning event. if event.id in previous_ids - assigned: from alphapulse.news_risk.events import dirty_event await dirty_event(session, event.id, frozen) 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 = accumulate_usage(previous_usage,{'attempts':usages}); row.state = 'done' row.error = None; row.finished_at = now await session.commit() return 'done' except Exception as error: await session.rollback() session.add(GroupingVersion(analysis_id=row_id, input=row_input, result=None, rules=rules, state='failed', error=str(error)[:2000])) await session.execute(update(ArticleAnalysis).where(ArticleAnalysis.id == row_id).values( state='done' if had_success else 'failed', error=str(error)[:2000], usage=accumulate_usage(previous_usage,{'attempts':usages,'failed':getattr(error,'usage',None)}), finished_at=datetime.now(UTC))) await session.commit() logger.exception('新闻初筛失败') return 'failed' async def summarize(target_id=None): async with SessionLocal() as session: settings = await session.get(RiskSettings, 1) if not settings or not settings.enabled: return event = await session.scalar(select(NewsSummary).where(NewsSummary.state == 'pending', NewsSummary.due_at <= datetime.now(UTC), NewsSummary.id == target_id if target_id else True).order_by(NewsSummary.due_at, NewsSummary.id).limit(1).with_for_update()) if event is None: return config_error = None if event.review_config: from alphapulse.news_risk.reprocess import frozen_configuration rules = event.review_config['configurations']['analysis']['rules'] fingerprint = briefing.fingerprint({}, rules) try: provider, system, rules = await frozen_configuration(session, event.review_config, 'analysis') except ValueError as error: provider = None; config_error = str(error) else: provider, system, rules, fingerprint = await configuration(session, settings, 'analysis') rows = (await session.scalars(select(ArticleAnalysis).join(SummaryArticle).join(NewsItem).where( SummaryArticle.summary_id == event.id, SummaryArticle.active.is_(True), ArticleAnalysis.content_hash == NewsItem.content_hash, ArticleAnalysis.state == 'done').order_by(ArticleAnalysis.id))).all() articles = [{'id': r.id, 'news_id': r.news_id, 'content_hash': r.content_hash, **r.input} for r in rows] package = {key: getattr(event, key) for key in ('id', 'title', 'fact', 'category', 'tags', 'content_kind', 'revision')} package['previous'] = event.result package['sources'] = [{'analysis_id': r.id, 'news_id': r.news_id, 'content_hash': r.content_hash} for r in rows] if not articles: from alphapulse.news_risk.events import dirty_event await dirty_event(session, event.id, event.review_config) event.result = None; event.state = 'withdrawn'; await session.commit(); return version = await session.scalar(select(SummaryVersion).where(SummaryVersion.summary_id == event.id, SummaryVersion.revision == event.revision, SummaryVersion.fingerprint == fingerprint)) if version and version.state == 'done': event.state = 'done'; event.error = None; event.dirty_since = None await session.commit() return if version is None: version = SummaryVersion(summary_id=event.id, revision=event.revision, fingerprint=fingerprint, state='processing', input={'package': package, 'articles': articles}, rules=rules, model=provider.model if provider else rules['model_config']['model'], started_at=datetime.now(UTC)) session.add(version) else: version.state = 'processing'; version.error = None event.state = 'processing' await session.commit() identifier, event_id, revision = version.id, event.id, version.revision research = Research(package) evidence_rows = (await session.scalars(select(PackageEvidence).where(PackageEvidence.version_id == identifier).order_by(PackageEvidence.id))).all() research.restore([r.data for r in evidence_rows]) await session.commit() usage = 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': version.state = 'superseded'; await session.commit(); return if result.correction is not None and event.revision != revision: version.state = 'superseded'; await session.commit(); return if result.correction is not None: from alphapulse.news_risk.corrections import apply_correction from alphapulse.news_risk.contracts import Correction await apply_correction(session, Correction.model_validate(result.correction), event_id, research.read_package_ids, rules) version.result = result.model_dump(mode='json'); version.usage = usage; version.state = 'corrected'; version.finished_at = datetime.now(UTC) for record in research.records: session.add(PackageEvidence(summary_id=event_id, version_id=identifier, kind=record['tool'], reference=str(record['arguments']), data=record)) await session.commit(); return version.result = result.model_dump(mode='json'); version.usage = usage; version.state = 'done'; version.finished_at = datetime.now(UTC) if event.revision == revision and revision > event.published_revision: event.result = result.model_dump(mode='json') event.stars = result.risk_stars; event.peak_stars = max(event.peak_stars, result.risk_stars) event.importance_stars = result.importance_stars event.published_revision = revision; event.analyzed_at = version.finished_at; event.error = None event.state = 'done' if event.revision == revision else 'pending' if event.state == 'done': event.dirty_since = None for record in research.records: session.add(PackageEvidence(summary_id=event_id, version_id=identifier, kind=record['tool'], reference=str(record['arguments']), data=record)) from alphapulse.news_risk.events import associate if event.revision == revision: await associate(session, event, result, research.read_package_ids, research.read_event_ids) await session.commit() except Exception as error: await session.rollback() for record in research.records: session.add(PackageEvidence(summary_id=event_id, version_id=identifier, kind=record['tool'], reference=str(record['arguments']), data=record)) await session.execute(update(SummaryVersion).where(SummaryVersion.id == identifier).values( state='failed', error=str(error)[:2000], usage=getattr(error, 'usage', None) or usage or {'rounds': getattr(research, 'usages', []), 'tools': research.records}, finished_at=datetime.now(UTC))) await session.execute(update(NewsSummary).where(NewsSummary.id == event_id, NewsSummary.replacement_ids == [], NewsSummary.state != 'withdrawn').values( state=case((NewsSummary.revision > revision, 'pending'), else_='failed'), error=str(error)[:2000])) await session.commit(); logger.exception('事件总结失败') 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 now = datetime.now(UTC) latest = await session.scalar(select(RiskSnapshot).order_by(RiskSnapshot.id.desc()).limit(1)) requested = await session.scalar(select(RiskSnapshot).where(RiskSnapshot.state == 'failed', RiskSnapshot.quotes['retry_requested'].as_boolean().is_(True)).order_by(RiskSnapshot.id).limit(1).with_for_update()) from alphapulse.news_risk.events import reading_inputs 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', {}).get('run_mode') != 'history_only' and job.settings.get('request'): job = None if job and job.settings.get('request', {}).get('run_mode') == 'history_only': from alphapulse.models.risk import ReprocessItem if await session.scalar(select(ReprocessItem.id).where(ReprocessItem.job_id == job.id, ReprocessItem.state.in_(['pending','processing','failed'])).limit(1)): return if job and job.position < len(job.items): if requested and requested.quotes.get('job_id') == job.id: return job = None if not inputs and latest is not None and job is None: waiting = await session.scalar(select(NewsEvent.id).where(NewsEvent.updated_at >= now-timedelta(hours=72), NewsEvent.state.in_(['pending','processing','failed']), NewsEvent.published_revision != NewsEvent.revision).limit(1)) if waiting is not None: return if job: if await session.scalar(select(NewsEvent.id).where(NewsEvent.state.in_(['pending', 'processing', 'failed']), NewsEvent.review_config['job_id'].astext.cast(BigInteger) == job.id).limit(1)): return from alphapulse.news_risk.reprocess import frozen_configuration provider, system, rules = await frozen_configuration(session, job.settings, 'outlook') else: provider, system, rules, _ = await configuration(session, settings, '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} 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: if (now-latest.created_at).total_seconds() < 60: return else: dirty = latest.quotes.get('dirty_since') if dirty is None: 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) ranked, expansion = {}, {} try: main = [p for p in inputs if p['importance_stars'] >= 3] low = [p for p in inputs if p['importance_stars'] < 3] ranked = await briefing.generate(provider, system, main, ranking_only=True) if main else {'reading': [], 'usage': []} candidates = briefing.shortlist(low) expansion = await briefing.generate(provider, system, candidates, ranking_only=True) if candidates else {'reading': [], 'usage': []} domains, extra = set(), [] by_id = {(p.get('kind', 'package'), p['id']): p for p in inputs} for row in expansion['reading']: category = by_id[(row.get('kind', 'package'), row['id'])]['category'] if category not in domains: domains.add(category); extra.append(row) if len(extra) == 3: break chosen = main + [by_id[(r.get('kind', 'package'), r['id'])] for r in extra] result = await briefing.generate(provider, system, chosen, rankings=ranked['reading']+extra) snapshot.result = {**result, 'reading': ranked['reading'], 'expansion': extra, 'inputs': inputs, 'rules': rules, 'usage': ranked['usage']+result['usage'], 'expansion_usage': expansion['usage'], 'as_of': now.isoformat()} step('保存阅读与趋势') async with publication_lock: fresh = await reading_inputs(session,datetime.now(UTC)-timedelta(hours=72)) snapshot.state = 'done' if briefing.semantic_packages(fresh)==briefing.semantic_packages(inputs) else 'superseded' except Exception as error: snapshot.state = 'failed'; snapshot.error = str(error)[:2000] snapshot.quotes = {**snapshot.quotes, 'failed_usage': getattr(error, 'usage', None), 'completed_usage': ranked.get('usage', []) + expansion.get('usage', [])} logger.exception('阅读编排失败') if requested: requested.quotes = {k: v for k, v in requested.quotes.items() if k != 'retry_requested'} await session.commit() async def main(): from alphapulse.news_research.worker import run await run() if __name__ == '__main__': logging.basicConfig(level=logging.INFO) asyncio.run(main())