diff --git a/database.py b/database.py index 85d30c0..3b045cc 100644 --- a/database.py +++ b/database.py @@ -1172,8 +1172,15 @@ class Database: result = await session.execute(stmt) return [row[0] for row in result.fetchall()] - async def get_job_media_timeline(self, job_id: int) -> List[Dict[str, Any]]: - """Get all media for a job ordered by discovery time for replay. + async def get_job_media_timeline( + self, job_id: int, limit: int = None, offset: int = 0 + ) -> List[Dict[str, Any]]: + """Get media for a job ordered by discovery time for replay. + + Args: + job_id: The crawl job ID + limit: Max items to return (None = all) + offset: Number of items to skip Returns list of {md5_hash, media_type, discovered_at, page_uri, page_title} sorted by discovered_at ascending. @@ -1187,6 +1194,9 @@ class Database: .join(MediaSource, Media.md5_hash == MediaSource.md5_hash) .where(MediaSource.crawl_job_id == job_id) .order_by(MediaSource.discovered_at.asc()) + .offset(offset) ) + if limit: + stmt = stmt.limit(limit) result = await session.execute(stmt) return [dict(row._mapping) for row in result.fetchall()] diff --git a/serp.py b/serp.py index 9280bab..e19c960 100644 --- a/serp.py +++ b/serp.py @@ -2734,46 +2734,55 @@ async def replay_crawl_job(job_id: int, speed: float = Query(1.0, ge=0.1, le=100 if not job: raise HTTPException(status_code=404, detail="Job not found") - # Get media timeline for this job - timeline = await db.get_job_media_timeline(job_id) - if not timeline: - raise HTTPException(status_code=404, detail="No media found for this job") - async def replay_generator(): prev_time = None - for item in timeline: - # Parse discovery time - discovered_at = item.get('discovered_at') - if discovered_at: - try: - curr_time = datetime.fromisoformat(discovered_at.replace('Z', '+00:00')) + total_count = 0 + batch_size = 100 + offset = 0 - # Calculate delay from previous item - if prev_time: - delta = (curr_time - prev_time).total_seconds() - # Apply speed multiplier and cap delay at 5 seconds max - delay = min(delta / speed, 5.0) - if delay > 0: - await asyncio.sleep(delay) + while True: + # Fetch next batch + batch = await db.get_job_media_timeline(job_id, limit=batch_size, offset=offset) + if not batch: + break - prev_time = curr_time - except (ValueError, TypeError): - pass + for item in batch: + total_count += 1 - # Format media item for SSE (same format as live stream) - media_event = { - 'md5_hash': item['md5_hash'], - 'media_type': item.get('media_type', 'image'), - 'mime_type': item.get('mime_type', ''), - 'page_uri': item.get('page_uri', ''), - 'page_title': item.get('page_title', ''), - 'replay': True, # Flag to indicate this is a replay - 'job_id': job_id, - } - yield f"data: {json.dumps(media_event)}\n\n" + # Parse discovery time + discovered_at = item.get('discovered_at') + if discovered_at: + try: + curr_time = datetime.fromisoformat(discovered_at.replace('Z', '+00:00')) + + # Calculate delay from previous item + if prev_time: + delta = (curr_time - prev_time).total_seconds() + # Apply speed multiplier and cap delay at 5 seconds max + delay = min(delta / speed, 5.0) + if delay > 0: + await asyncio.sleep(delay) + + prev_time = curr_time + except (ValueError, TypeError): + pass + + # Format media item for SSE (same format as live stream) + media_event = { + 'md5_hash': item['md5_hash'], + 'media_type': item.get('media_type', 'image'), + 'mime_type': item.get('mime_type', ''), + 'page_uri': item.get('page_uri', ''), + 'page_title': item.get('page_title', ''), + 'replay': True, + 'job_id': job_id, + } + yield f"data: {json.dumps(media_event)}\n\n" + + offset += batch_size # Send end-of-replay marker - yield f"data: {json.dumps({'type': 'replay_end', 'job_id': job_id, 'total': len(timeline)})}\n\n" + yield f"data: {json.dumps({'type': 'replay_end', 'job_id': job_id, 'total': total_count})}\n\n" return StreamingResponse( replay_generator(), diff --git a/templates/live.html.j2 b/templates/live.html.j2 index 554e878..16a6bd7 100644 --- a/templates/live.html.j2 +++ b/templates/live.html.j2 @@ -261,7 +261,11 @@ function getScoreBadge(score, mediaType) { return `${label}`; } -poll(); +// In replay mode: start with empty feed, only stream replay data +// In normal mode: poll for existing media and connect to live stream +if (!replayMode) { + poll(); +} connectSSE(); {% endblock %}