""" Analysis Runner — programmatic (non-SSE) analysis for the Worker. Extracts the core full-analysis logic from analyze.py without SSE wrapping. Used by the pre-cache service to run analyses in the background. """ import asyncio import contextlib import json import logging import time from datetime import datetime, timezone from typing import Any, AsyncGenerator from app.core.config import settings from app.core.freshness import FreshnessStatus from app.core.sampling import create_sample_plan from app.core.ttl_tiers import get_ttl_hours from app.core.worker_logging import AsyncTimingContext, get_structured_logger, log_structured from app.db.mongodb import mongodb from app.models.schemas import ( AnalysisProgress, AnalysisResult, GameInfo, Highlight, TopicHighlights, TopicSentiment, ) from app.services.highlights_service import HighlightsCollector from app.services.analysis_utils import ( aggregate_topics, calculate_prediction, compute_preferred_context, datetime_from_timestamp, filter_topics_by_min_mentions, normalize_legacy_results, scale_topics, serialize_datetime, ) from app.services.nlp_service import NLPService from app.services.steam_service import SteamService logger = logging.getLogger(__name__) async def iter_incremental_analysis_events( game: GameInfo, stale_doc: dict[str, Any], steam_svc: SteamService, nlp_svc: NLPService, patch_timestamp: int | None = None, *, source: str = "live", ) -> AsyncGenerator[dict[str, str], None]: """Yield incremental-analysis progress and final result events.""" ttl_hours = await get_ttl_hours(game.app_id) old_results = normalize_legacy_results(stale_doc.get("results", {})) old_review_ids: list[str] = stale_doc.get("analyzed_review_ids", []) old_review_ids_set = set(old_review_ids) nlp_cumulative_s: float = 0.0 old_general = [TopicSentiment(**topic) for topic in old_results.get("general_topics", [])] old_recent = ( [TopicSentiment(**topic) for topic in old_results.get("recent_topics", [])] if old_results.get("recent_topics") else [] ) old_current_patch = ( [TopicSentiment(**topic) for topic in old_results.get("current_patch_topics", [])] if old_results.get("current_patch_topics") else [] ) old_last_patch = ( [TopicSentiment(**topic) for topic in old_results.get("last_patch_topics", [])] if old_results.get("last_patch_topics") else None ) old_last_patch_count = old_results.get("last_patch_reviews_count", 0) old_patch_ts = old_results.get("current_patch_timestamp") new_items = await steam_svc.fetch_recent_reviews( game.app_id, exclude_ids=old_review_ids_set, ) if not new_items: refreshed_at = datetime.now(timezone.utc) refreshed_results = { **old_results, "cached_at": refreshed_at, "analysis_date": refreshed_at, "current_patch_date": datetime_from_timestamp( patch_timestamp if patch_timestamp is not None else old_results.get("current_patch_timestamp") ), "freshness_status": FreshnessStatus.FRESH.value, "staleness_reason": None, "is_refreshing": False, } await mongodb.save_analysis( game.app_id, refreshed_results, analyzed_review_ids=old_review_ids, latest_review_timestamp=stale_doc.get("latest_review_timestamp", 0), ttl_hours=ttl_hours, analyzed_at=refreshed_at, ) yield { "event": "complete", "data": json.dumps(refreshed_results, default=serialize_datetime), } return new_texts = [item.text for item in new_items] new_review_ids = [item.recommendation_id for item in new_items] latest_timestamp = max( (item.timestamp_created for item in new_items), default=stale_doc.get("latest_review_timestamp", 0), ) batch_size = settings.review_batch_size delta_topics: list[TopicSentiment] = [] delta_current_patch_topics: list[TopicSentiment] = [] delta_current_patch_count = 0 highlights_collector = HighlightsCollector() processed = 0 total_skipped = 0 for i in range(0, len(new_texts), batch_size): chunk_texts = new_texts[i:i + batch_size] chunk_items = new_items[i:i + batch_size] batch_skipped = 0 if patch_timestamp: for review_item, text in zip(chunk_items, chunk_texts): categories = ["recent"] if review_item.timestamp_created >= patch_timestamp: categories.append("current_patch") nlp_start = time.monotonic() result_topics, skipped = await nlp_svc.analyze_batch( [text], highlights_collector=highlights_collector, categories=categories, ) nlp_cumulative_s += time.monotonic() - nlp_start batch_skipped += skipped if result_topics: delta_topics = aggregate_topics(delta_topics, result_topics) if review_item.timestamp_created >= patch_timestamp: delta_current_patch_topics = aggregate_topics( delta_current_patch_topics, result_topics, ) delta_current_patch_count += 1 total_skipped += batch_skipped else: nlp_start = time.monotonic() batch_results, batch_skipped = await nlp_svc.analyze_batch( chunk_texts, highlights_collector=highlights_collector, categories=["recent"], ) nlp_cumulative_s += time.monotonic() - nlp_start if batch_results: delta_topics = aggregate_topics(delta_topics, batch_results) total_skipped += batch_skipped processed += len(chunk_texts) progress = AnalysisProgress( processed=processed, total=len(new_texts), current_topics=delta_topics, skipped_count=total_skipped, ) yield {"event": "progress", "data": progress.model_dump_json()} new_general = aggregate_topics(old_general, delta_topics) old_recent_count = old_results.get("recent_reviews_count", 0) new_count = len(new_texts) if ( old_recent_count + new_count > settings.recent_sample_limit and old_recent and old_recent_count > 0 ): overflow = old_recent_count + new_count - settings.recent_sample_limit retain_ratio = max(0.2, 1.0 - overflow / old_recent_count) scaled_old = scale_topics(old_recent, retain_ratio) new_recent = aggregate_topics(scaled_old, delta_topics) recent_count = int(old_recent_count * retain_ratio) + new_count else: new_recent = aggregate_topics(old_recent, delta_topics) if old_recent else delta_topics recent_count = old_recent_count + new_count last_patch_topics = old_last_patch last_patch_count = old_last_patch_count if patch_timestamp and old_patch_ts and patch_timestamp != old_patch_ts: last_patch_topics = old_current_patch if old_current_patch else None last_patch_count = old_results.get("current_patch_reviews_count", 0) old_current_patch = [] new_current_patch = ( aggregate_topics(old_current_patch, delta_current_patch_topics) if old_current_patch else (delta_current_patch_topics if delta_current_patch_topics else []) ) base_current_patch_count = ( 0 if (patch_timestamp and old_patch_ts and patch_timestamp != old_patch_ts) else old_results.get("current_patch_reviews_count", 0) ) new_current_patch_count = base_current_patch_count + delta_current_patch_count has_current_patch = patch_timestamp is not None and ( new_current_patch_count > 0 or bool(old_current_patch) ) # Apply min-mentions filter on final aggregates (not per-review — see nlp_service.py). new_general = filter_topics_by_min_mentions(new_general) new_recent = filter_topics_by_min_mentions(new_recent) new_current_patch = filter_topics_by_min_mentions(new_current_patch) prediction = calculate_prediction(new_general) highlights_data = highlights_collector.compute_highlights() general_highlights = highlights_data["general"] recent_highlights = highlights_data["recent"] current_patch_highlights = highlights_data["current_patch"] topic_highlights_dict = highlights_data["topics"] # Restrict topic highlights to topics that survived the min-mentions filter, # so the topic_highlights set is always consistent with general_topics. _surviving_topics = {t.topic for t in new_general} topic_highlights_list = [ TopicHighlights( topic=topic, highlights=[Highlight(**highlight) for highlight in highlights], ) for topic, highlights in topic_highlights_dict.items() if topic in _surviving_topics ] merged_review_ids = old_review_ids + new_review_ids analysis_generated_at = datetime.now(timezone.utc) result = AnalysisResult( game=game, general_topics=new_general, recent_topics=new_recent, recent_reviews_count=recent_count, current_patch_topics=new_current_patch if has_current_patch else None, current_patch_reviews_count=new_current_patch_count if has_current_patch else 0, last_patch_topics=last_patch_topics, last_patch_reviews_count=last_patch_count, current_patch_timestamp=patch_timestamp, analysis_date=analysis_generated_at, current_patch_date=datetime_from_timestamp(patch_timestamp), prediction=prediction, analyzed_reviews=old_results.get("analyzed_reviews", 0) + processed, skipped_count=old_results.get("skipped_count", 0) + total_skipped, general_highlights=[Highlight(**highlight) for highlight in general_highlights], recent_highlights=[Highlight(**highlight) for highlight in recent_highlights] if recent_highlights else None, current_patch_highlights=[Highlight(**highlight) for highlight in current_patch_highlights] if current_patch_highlights else None, topic_highlights=topic_highlights_list, cached_at=analysis_generated_at, preferred_context=compute_preferred_context(patch_timestamp), freshness_status=FreshnessStatus.FRESH.value, is_refreshing=False, ) await mongodb.save_analysis( game.app_id, result.model_dump(), analyzed_review_ids=merged_review_ids, latest_review_timestamp=latest_timestamp, ttl_hours=ttl_hours, analyzed_at=analysis_generated_at, ) if get_structured_logger(): log_structured( "incremental_analysis_complete", app_id=game.app_id, game_name=game.name if hasattr(game, "name") else str(game.app_id), source=source, reviews_processed=processed, topics_found=len(new_general), detail={"nlp_cumulative_s": round(nlp_cumulative_s, 3)}, ) yield {"event": "complete", "data": result.model_dump_json()} async def run_incremental_analysis( app_id: str, game_name: str, steam_svc: SteamService, nlp_svc: NLPService, ) -> dict[str, Any] | None: """Run a non-SSE incremental analysis for worker jobs.""" slog = get_structured_logger() try: stale_doc = await mongodb.get_analysis(app_id) if not stale_doc or not stale_doc.get("results") or not stale_doc.get("analyzed_review_ids"): return await run_full_analysis(app_id, game_name, steam_svc, nlp_svc, stale_doc=stale_doc) # Long gap guard: if the most recent review we have is too old, Steam's cursor-based # API may not reliably surface all reviews since then. Fall back to full analysis. latest_ts = stale_doc.get("latest_review_timestamp", 0) if latest_ts > 0: gap_days = (time.time() - latest_ts) / 86400 if gap_days > settings.incremental_max_gap_days: logger.info( f"Incremental gap {gap_days:.0f}d > {settings.incremental_max_gap_days}d " f"for {app_id} ({game_name}) — falling back to full analysis" ) return await run_full_analysis(app_id, game_name, steam_svc, nlp_svc, stale_doc=stale_doc) game = await steam_svc.get_game_info(app_id) if not game: cached_game = stale_doc.get("results", {}).get("game") if isinstance(cached_game, dict): game = GameInfo(**cached_game) else: game = GameInfo(app_id=app_id, name=game_name) patch_date = await mongodb.get_game_patch_date(app_id) patch_timestamp = int(patch_date.timestamp()) if patch_date else None if patch_timestamp: game = game.model_copy(update={"last_game_update_at": patch_timestamp}) final_payload: dict[str, Any] | None = None async for event in iter_incremental_analysis_events( game, stale_doc, steam_svc, nlp_svc, patch_timestamp=patch_timestamp, source="worker", ): if event.get("event") == "complete": final_payload = json.loads(event["data"]) return final_payload except Exception as e: logger.error(f"Incremental analysis runner error for {app_id} ({game_name}): {e}", exc_info=True) if slog: log_structured( "analysis_error", level=logging.ERROR, app_id=app_id, game_name=game_name, source="worker", error=str(e), ) return None async def run_full_analysis( app_id: str, game_name: str, steam_svc: SteamService, nlp_svc: NLPService, stale_doc: dict[str, Any] | None = None, ) -> dict[str, Any] | None: """ Run a full analysis for a game (no SSE, no streaming). Returns: Analysis result dict, or None on error. """ slog = get_structured_logger() try: # Phase 1: Setup — game info + review stats + sample plan async with AsyncTimingContext() as t_setup: # 1. Get game info game = await steam_svc.get_game_info(app_id) if not game: logger.warning(f"Analysis runner: game info not found for {app_id}") return None # 2. Get review stats stats = await steam_svc.get_review_stats(app_id) if stats.total == 0: logger.warning(f"Analysis runner: no reviews for {app_id}") return None # 3. Create sample plan sample_plan = create_sample_plan(stats.total, stats.positive, stats.negative) ttl_hours = await get_ttl_hours(app_id) # 3b. Fetch game patch date for current_patch splitting patch_date = await mongodb.get_game_patch_date(app_id) patch_timestamp = int(patch_date.timestamp()) if patch_date else None if patch_timestamp and isinstance(game, GameInfo): game = game.model_copy(update={"last_game_update_at": patch_timestamp}) # Phase 2: Fetch + Analyze — producer-consumer loop nlp_cumulative_s: float = 0.0 async with AsyncTimingContext() as t_fetch_analyze: # 4. Producer-consumer fetch + analyze queue: asyncio.Queue = asyncio.Queue(maxsize=5) async def fetch_worker(): try: async for batch in steam_svc.fetch_reviews_stratified(app_id, sample_plan): await queue.put(batch) except Exception as e: await queue.put(e) finally: await queue.put(None) fetch_task = asyncio.create_task(fetch_worker()) processed = 0 total_skipped = 0 aggregated_topics: list[TopicSentiment] = [] recent_processed = 0 recent_limit = settings.recent_sample_limit all_review_ids: list[str] = [] latest_timestamp = 0 highlights_collector = HighlightsCollector() current_patch_topics: list[TopicSentiment] = [] current_patch_count = 0 review_topic_results: list[tuple[int, list[TopicSentiment]]] = [] try: while True: item = await queue.get() if item is None: break if isinstance(item, Exception): raise item batch = item if not batch.reviews: continue for ri in batch.review_items: all_review_ids.append(ri.recommendation_id) if ri.timestamp_created > latest_timestamp: latest_timestamp = ri.timestamp_created batch_skipped = 0 if patch_timestamp and batch.review_items: for ri, text in zip(batch.review_items, batch.reviews): is_recent = recent_processed < recent_limit cat = [] if is_recent: cat.append("recent") if ri.timestamp_created >= patch_timestamp: cat.append("current_patch") nlp_start = time.monotonic() res, skipped = await nlp_svc.analyze_batch( [text], highlights_collector=highlights_collector, categories=cat ) nlp_cumulative_s += time.monotonic() - nlp_start batch_skipped += skipped if res: aggregated_topics = aggregate_topics(aggregated_topics, res) current_patch_topics = aggregate_topics(current_patch_topics, res) review_topic_results.append((ri.timestamp_created, res)) current_patch_count += 1 else: nlp_start = time.monotonic() res, skipped = await nlp_svc.analyze_batch( [text], highlights_collector=highlights_collector, categories=cat ) nlp_cumulative_s += time.monotonic() - nlp_start batch_skipped += skipped if res: aggregated_topics = aggregate_topics(aggregated_topics, res) review_topic_results.append((ri.timestamp_created, res)) recent_processed += 1 else: for ri, text in zip(batch.review_items, batch.reviews) if batch.review_items else enumerate(batch.reviews): is_recent = recent_processed < recent_limit cat = ["recent"] if is_recent else [] nlp_start = time.monotonic() res, skipped = await nlp_svc.analyze_batch( [text], highlights_collector=highlights_collector, categories=cat ) nlp_cumulative_s += time.monotonic() - nlp_start batch_skipped += skipped ts = ri.timestamp_created if batch.review_items else 0 if res: aggregated_topics = aggregate_topics(aggregated_topics, res) review_topic_results.append((ts, res)) recent_processed += 1 total_skipped += batch_skipped processed += len(batch.reviews) await fetch_task except BaseException: fetch_task.cancel() with contextlib.suppress(asyncio.CancelledError): await fetch_task raise # Phase 3: Save — highlights + MongoDB save async with AsyncTimingContext() as t_save: # 5. Compute prediction + highlights # Build recent_topics from highest-timestamp reviews review_topic_results.sort(key=lambda x: x[0], reverse=True) recent_entries = review_topic_results[:recent_limit] recent_topics: list[TopicSentiment] = [] for _, topics_batch in recent_entries: for ts in topics_batch: recent_topics = aggregate_topics(recent_topics, [ts]) recent_reviews_count = len(recent_entries) # Apply min-mentions filter on final aggregates (not per-review — see nlp_service.py). aggregated_topics = filter_topics_by_min_mentions(aggregated_topics) recent_topics = filter_topics_by_min_mentions(recent_topics) current_patch_topics = filter_topics_by_min_mentions(current_patch_topics) prediction = calculate_prediction(aggregated_topics) highlights_data = highlights_collector.compute_highlights() general_highlights = highlights_data["general"] recent_highlights = highlights_data["recent"] current_patch_highlights = highlights_data["current_patch"] topic_highlights_dict = highlights_data["topics"] # Restrict topic highlights to topics that survived the min-mentions filter, # so the topic_highlights set is always consistent with general_topics. _surviving_topics = {t.topic for t in aggregated_topics} topic_highlights_list = [ TopicHighlights( topic=topic, highlights=[Highlight(**h) for h in highlights], ) for topic, highlights in topic_highlights_dict.items() if topic in _surviving_topics ] has_recent_split = processed > recent_limit has_current_patch = patch_timestamp is not None and current_patch_count > 0 analysis_generated_at = datetime.now(timezone.utc) current_patch_date = ( datetime.fromtimestamp(patch_timestamp, tz=timezone.utc) if patch_timestamp is not None else None ) # Archive last_patch_topics when full analysis replaces a doc with a different patch. last_patch_topics: list[TopicSentiment] | None = None last_patch_reviews_count = 0 if stale_doc: old_r = normalize_legacy_results(stale_doc.get("results", {})) old_patch_ts = old_r.get("current_patch_timestamp") if patch_timestamp and old_patch_ts and patch_timestamp != old_patch_ts: raw_cp = old_r.get("current_patch_topics") last_patch_topics = [TopicSentiment(**t) for t in raw_cp] if raw_cp else None last_patch_reviews_count = old_r.get("current_patch_reviews_count", 0) else: raw_lp = old_r.get("last_patch_topics") last_patch_topics = [TopicSentiment(**t) for t in raw_lp] if raw_lp else None last_patch_reviews_count = old_r.get("last_patch_reviews_count", 0) result = AnalysisResult( game=game, general_topics=aggregated_topics, recent_topics=recent_topics if has_recent_split else None, recent_reviews_count=recent_reviews_count if has_recent_split else 0, current_patch_topics=current_patch_topics if has_current_patch else None, current_patch_reviews_count=current_patch_count if has_current_patch else 0, last_patch_topics=last_patch_topics, last_patch_reviews_count=last_patch_reviews_count, current_patch_timestamp=patch_timestamp, analysis_date=analysis_generated_at, current_patch_date=current_patch_date, prediction=prediction, analyzed_reviews=processed, skipped_count=total_skipped, general_highlights=[Highlight(**h) for h in general_highlights], recent_highlights=[Highlight(**h) for h in recent_highlights] if recent_highlights else None, current_patch_highlights=[Highlight(**h) for h in current_patch_highlights] if current_patch_highlights else None, topic_highlights=topic_highlights_list, cached_at=analysis_generated_at, preferred_context=compute_preferred_context(patch_timestamp), freshness_status=FreshnessStatus.FRESH.value, is_refreshing=False, ) # 6. Save to cache await mongodb.save_analysis( game.app_id, result.model_dump(), analyzed_review_ids=all_review_ids, latest_review_timestamp=latest_timestamp, ttl_hours=ttl_hours, analyzed_at=analysis_generated_at, ) total_elapsed = t_setup.elapsed_s + t_fetch_analyze.elapsed_s + t_save.elapsed_s logger.info( f"Analysis runner: completed {app_id} ({game_name}) — " f"{processed} reviews, {len(aggregated_topics)} topics" ) if slog: log_structured( "analysis_complete", app_id=app_id, game_name=game_name, elapsed_s=round(total_elapsed, 3), source="worker", breakdown={ "setup_s": t_setup.elapsed_s, "fetch_analyze_s": t_fetch_analyze.elapsed_s, "nlp_cumulative_s": round(nlp_cumulative_s, 3), "save_s": t_save.elapsed_s, }, reviews_processed=processed, topics_found=len(aggregated_topics), ) return result.model_dump() except Exception as e: logger.error(f"Analysis runner error for {app_id} ({game_name}): {e}", exc_info=True) if slog: log_structured( "analysis_error", level=logging.ERROR, app_id=app_id, game_name=game_name, source="worker", error=str(e), ) return None