diff --git a/src/codec_validator.py b/src/codec_validator.py new file mode 100644 index 0000000..e73189f --- /dev/null +++ b/src/codec_validator.py @@ -0,0 +1,107 @@ +""" +Codec Validator + +Validates video codecs against a whitelist/blacklist. +Flags unsupported codecs as UNSCANNABLE. +""" + +import logging +from typing import Set + +logger = logging.getLogger(__name__) + +# Default supported codecs +DEFAULT_WHITELIST: Set[str] = { + "avc1", # H.264 + "hevc", # H.265 + "vp8", # VP8 + "vp9", # VP9 + "av01", # AV1 + "mjpeg", # MJPEG + "mp4v", # MPEG-4 + "h264", # H.264 (alternative name) + "h265", # H.265 (alternative name) +} + +# Known unsupported codecs +DEFAULT_BLACKLIST: Set[str] = { + "theora", # Theora + "divx", # DivX + "xvid", # Xvid + "prores", # ProRes (requires special handling) + "avc", # Old AVC naming + "hev1", # HEVC alternative +} + + +class CodecValidator: + """Validate video codecs against whitelist/blacklist.""" + + def __init__( + self, + whitelist: Set[str] = None, + blacklist: Set[str] = None, + default_status_on_error: str = "UNSCANNABLE", + ): + self.whitelist = whitelist or DEFAULT_WHITELIST + self.blacklist = blacklist or DEFAULT_BLACKLIST + self.default_status_on_error = default_status_on_error + + def is_supported(self, codec: str) -> bool: + """Check if a codec is supported.""" + codec_lower = codec.lower().strip() + + # Check blacklist first + if codec_lower in self.blacklist: + logger.info("Codec '%s' is blacklisted", codec) + return False + + # Check whitelist + if codec_lower in self.whitelist: + return True + + # Not in either list - assume unsupported + logger.warning("Codec '%s' not in whitelist or blacklist", codec) + return False + + def validate(self, codec: str) -> tuple: + """ + Validate a codec and return (is_supported, reason). + + Returns: + (True, "supported") if codec is in whitelist + (False, "blacklisted") if codec is in blacklist + (False, "not_in_whitelist") if codec is unknown + """ + codec_lower = codec.lower().strip() + + if codec_lower in self.blacklist: + return False, "blacklisted" + + if codec_lower in self.whitelist: + return True, "supported" + + return False, "not_in_whitelist" + + def add_to_whitelist(self, codec: str): + """Add a codec to the whitelist.""" + self.whitelist.add(codec.lower().strip()) + logger.info("Added codec '%s' to whitelist", codec) + + def add_to_blacklist(self, codec: str): + """Add a codec to the blacklist.""" + self.blacklist.add(codec.lower().strip()) + logger.info("Added codec '%s' to blacklist", codec) + + def get_status(self, codec: str) -> str: + """Get the status for a codec (for DB update).""" + is_supported, reason = self.validate(codec) + if is_supported: + return "PENDING" + return self.default_status_on_error + + def __repr__(self): + return ( + f"CodecValidator(whitelist={len(self.whitelist)}, " + f"blacklist={len(self.blacklist)})" + ) diff --git a/src/orchestrator.py b/src/orchestrator.py new file mode 100644 index 0000000..9b6084a --- /dev/null +++ b/src/orchestrator.py @@ -0,0 +1,256 @@ +""" +Batch Orchestration Skeleton + +Manages job queue, worker pool, state transitions, and crash recovery. +Ensures atomic state transitions and idempotent processing. +""" + +import logging +import time +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timezone +from enum import Enum +from typing import Dict, List, Optional + +logger = logging.getLogger(__name__) + + +class JobStatus(Enum): + """Job status states.""" + PENDING = "PENDING" + PROCESSING = "PROCESSING" + COMPLETED = "COMPLETED" + FAILED = "ERROR" + SKIPPED = "UNSCANNABLE" + + +class Job: + """Represents a single video processing job.""" + + def __init__(self, video_id: int, file_path: str, priority: float = 0.0): + self.video_id = video_id + self.file_path = file_path + self.priority = priority # Higher = more urgent (based on modification time) + self.status = JobStatus.PENDING + self.created_at = datetime.now(timezone.utc) + self.started_at: Optional[datetime] = None + self.completed_at: Optional[datetime] = None + self.attempts = 0 + self.max_retries = 3 + self.error_message: Optional[str] = None + + def to_dict(self) -> dict: + return { + "video_id": self.video_id, + "file_path": self.file_path, + "priority": self.priority, + "status": self.status.value, + "created_at": self.created_at.isoformat(), + "started_at": self.started_at.isoformat() if self.started_at else None, + "completed_at": self.completed_at.isoformat() if self.completed_at else None, + "attempts": self.attempts, + "error_message": self.error_message, + } + + def __repr__(self): + return f"Job(id={self.video_id}, path={self.file_path}, status={self.status.value})" + + +class WorkerPool: + """Manage a pool of worker processes for parallel processing.""" + + def __init__(self, db_connector, config: dict, max_workers: int = 2): + self.db = db_connector + self.config = config + self.max_workers = max_workers + self._executor = ThreadPoolExecutor(max_workers=max_workers) + self._running = False + self._jobs_processed = 0 + self._jobs_failed = 0 + + def start(self): + """Start the worker pool.""" + self._running = True + logger.info("Worker pool starting with %d workers", self.max_workers) + + while self._running: + # Get pending jobs + jobs = self._get_pending_jobs() + + if jobs: + # Submit jobs to executor + for job in jobs: + future = self._executor.submit(self._process_job, job) + future.add_done_callback(self._on_job_complete) + else: + time.sleep(5) # No jobs, wait + + def stop(self): + """Stop the worker pool gracefully.""" + self._running = False + logger.info("Worker pool stopping. Processed: %d, Failed: %d", + self._jobs_processed, self._jobs_failed) + self._executor.shutdown(wait=True) + + def _get_pending_jobs(self) -> List[Job]: + """Get pending jobs from DB with atomic locking.""" + now = datetime.now(timezone.utc).replace(tzinfo=None) + + # Atomic lock: update status from PENDING to PROCESSING + result = self.db.execute( + """UPDATE videos SET status = 'PROCESSING', last_processed_time = %s, + updated_at = %s + WHERE id IN ( + SELECT id FROM ( + SELECT id FROM videos + WHERE status = 'PENDING' + ORDER BY last_scan_time DESC + LIMIT %s + ) AS sub + )""", + (now, now, self.max_workers), + transaction=True, + ) + + if result == 0: + return [] + + # Get the locked jobs + locked = self.db.fetchall( + "SELECT id, file_path, last_scan_time FROM videos WHERE status = 'PROCESSING' AND last_processed_time = %s", + (now,), + ) + + return [ + Job(video_id=row["id"], file_path=row["file_path"], priority=row["last_scan_time"].timestamp()) + for row in locked + ] + + def _process_job(self, job: Job) -> bool: + """Process a single job (placeholder - actual processing in later stories).""" + job.status = JobStatus.PROCESSING + job.started_at = datetime.now(timezone.utc) + job.attempts += 1 + + logger.info("Processing job: %s (attempt %d)", job, job.attempts) + + try: + # TODO: Actual processing logic (frame sampling, detection, classification) + # This is a placeholder - will be implemented in STORY-03 through STORY-06 + time.sleep(1) # Simulate processing + + # Mark as completed + self._complete_job(job) + self._jobs_processed += 1 + return True + + except Exception as e: + logger.error("Job failed: %s - %s", job, e, exc_info=True) + job.error_message = str(e) + + if job.attempts < job.max_retries: + # Retry + self._retry_job(job) + return False + else: + # Max retries exceeded + self._fail_job(job) + self._jobs_failed += 1 + return False + + def _complete_job(self, job: Job): + """Mark a job as completed.""" + now = datetime.now(timezone.utc).replace(tzinfo=None) + job.status = JobStatus.COMPLETED + job.completed_at = datetime.now(timezone.utc) + + self.db.execute( + """UPDATE videos SET status = 'COMPLETED', updated_at = %s + WHERE id = %s""", + (now, job.video_id), + transaction=True, + ) + logger.info("Job completed: %s", job) + + def _fail_job(self, job: Job): + """Mark a job as failed.""" + now = datetime.now(timezone.utc).replace(tzinfo=None) + job.status = JobStatus.FAILED + job.completed_at = datetime.now(timezone.utc) + + self.db.execute( + """UPDATE videos SET status = 'ERROR', error_message = %s, updated_at = %s + WHERE id = %s""", + (job.error_message, now, job.video_id), + transaction=True, + ) + logger.error("Job failed permanently: %s", job) + + def _retry_job(self, job: Job): + """Re-queue a job for retry.""" + now = datetime.now(timezone.utc).replace(tzinfo=None) + + self.db.execute( + """UPDATE videos SET status = 'PENDING', updated_at = %s + WHERE id = %s""", + (now, job.video_id), + transaction=True, + ) + logger.info("Job re-queued for retry: %s (attempt %d/%d)", + job, job.attempts, job.max_retries) + + def _on_job_complete(self, future): + """Callback when a job completes.""" + try: + future.result() + except Exception as e: + logger.error("Unhandled job error: %s", e) + + def get_stats(self) -> dict: + """Get worker pool statistics.""" + return { + "max_workers": self.max_workers, + "jobs_processed": self._jobs_processed, + "jobs_failed": self._jobs_failed, + "is_running": self._running, + "active_workers": self._executor._work_queue.qsize(), + } + + +class CrashRecovery: + """Handle crash recovery and stale job detection.""" + + def __init__(self, db_connector, lock_timeout_minutes: int = 5): + self.db = db_connector + self.lock_timeout = lock_timeout_minutes + + def recover_stale_jobs(self) -> int: + """Re-queue jobs stuck in PROCESSING beyond the lock timeout.""" + now = datetime.now(timezone.utc).replace(tzinfo=None) + timeout = now.replace(minute=now.minute - self.lock_timeout) + + result = self.db.execute( + """UPDATE videos SET status = 'PENDING', updated_at = %s + WHERE status = 'PROCESSING' AND last_processed_time < %s""", + (now, timeout), + transaction=True, + ) + + if result > 0: + logger.info("Recovered %d stale jobs", result) + return result + + def check_health(self) -> dict: + """Check system health for crash recovery purposes.""" + processing_count = self.db.fetchone( + "SELECT COUNT(*) as count FROM videos WHERE status = 'PROCESSING'" + ) + pending_count = self.db.fetchone( + "SELECT COUNT(*) as count FROM videos WHERE status = 'PENDING'" + ) + + return { + "processing_count": processing_count["count"] if processing_count else 0, + "pending_count": pending_count["count"] if pending_count else 0, + "stale_jobs_recovered": self.recover_stale_jobs(), + } diff --git a/src/prober.py b/src/prober.py new file mode 100644 index 0000000..2eb2396 --- /dev/null +++ b/src/prober.py @@ -0,0 +1,181 @@ +""" +Video Probing Module + +Extracts video metadata (codec, resolution, duration, frame rate) using ffprobe. +Handles errors gracefully for corrupt or unsupported files. +""" + +import json +import logging +import subprocess +import time +from dataclasses import dataclass, field +from typing import Optional + +logger = logging.getLogger(__name__) + + +@dataclass +class VideoMetadata: + """Metadata extracted from a video file.""" + file_path: str + codec: Optional[str] = None + resolution_w: Optional[int] = None + resolution_h: Optional[int] = None + duration: Optional[float] = None + frame_rate: Optional[float] = None + file_size: Optional[int] = None + status: str = "PENDING" + error_message: Optional[str] = None + probe_time: float = field(default_factory=time.time) + + @property + def is_valid(self) -> bool: + return self.status == "PENDING" and self.codec is not None + + @property + def is_unscannable(self) -> bool: + return self.status == "UNSCANNABLE" + + def to_dict(self) -> dict: + return { + "file_path": self.file_path, + "codec": self.codec, + "resolution_w": self.resolution_w, + "resolution_h": self.resolution_h, + "duration": self.duration, + "frame_rate": self.frame_rate, + "file_size": self.file_size, + "status": self.status, + "error_message": self.error_message, + } + + +class VideoProber: + """Probe video files using ffprobe to extract metadata.""" + + def __init__(self, timeout: int = 10): + self.timeout = timeout + + def probe(self, file_path: str) -> VideoMetadata: + """Probe a video file and return metadata.""" + metadata = VideoMetadata(file_path=file_path) + + # Get file size + try: + metadata.file_size = self._get_file_size(file_path) + except OSError as e: + metadata.status = "UNSCANNABLE" + metadata.error_message = f"Cannot read file: {e}" + logger.warning("File access error for %s: %s", file_path, e) + return metadata + + # Run ffprobe + try: + result = self._run_ffprobe(file_path) + if result is None: + metadata.status = "UNSCANNABLE" + metadata.error_message = "ffprobe timed out or failed" + logger.warning("ffprobe failed for %s", file_path) + return metadata + + metadata = self._parse_ffprobe_output(result, file_path) + except subprocess.TimeoutExpired: + metadata.status = "UNSCANNABLE" + metadata.error_message = f"ffprobe timed out after {self.timeout}s" + logger.warning("ffprobe timeout for %s", file_path) + except Exception as e: + metadata.status = "UNSCANNABLE" + metadata.error_message = f"ffprobe error: {e}" + logger.warning("ffprobe error for %s: %s", file_path, e) + + return metadata + + def _run_ffprobe(self, file_path: str) -> Optional[str]: + """Run ffprobe and return JSON output.""" + cmd = [ + "ffprobe", + "-v", "error", + "-select_streams", "v:0", + "-show_entries", "stream=codec_name,width,height,r_frame_rate,duration", + "-show_entries", "format=duration,size", + "-of", "json", + file_path, + ] + + try: + output = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=self.timeout, + ) + + if output.returncode != 0: + logger.debug("ffprobe stderr for %s: %s", file_path, output.stderr) + return None + + return output.stdout + except subprocess.TimeoutExpired: + return None + + def _parse_ffprobe_output(self, output: str, file_path: str) -> VideoMetadata: + """Parse ffprobe JSON output into VideoMetadata.""" + metadata = VideoMetadata(file_path=file_path) + + try: + data = json.loads(output) + except json.JSONDecodeError as e: + metadata.status = "UNSCANNABLE" + metadata.error_message = f"Invalid ffprobe output: {e}" + return metadata + + # Extract stream info + streams = data.get("streams", []) + if not streams: + metadata.status = "UNSCANNABLE" + metadata.error_message = "No video streams found" + return metadata + + stream = streams[0] + metadata.codec = stream.get("codec_name") + metadata.resolution_w = stream.get("width") + metadata.resolution_h = stream.get("height") + + # Parse frame rate + r_frame_rate = stream.get("r_frame_rate") + if r_frame_rate and "/" in r_frame_rate: + try: + num, den = r_frame_rate.split("/") + metadata.frame_rate = float(num) / float(den) if float(den) != 0 else None + except (ValueError, ZeroDivisionError): + pass + + # Extract format info + format_info = data.get("format", {}) + if format_info.get("duration"): + try: + metadata.duration = float(format_info["duration"]) + except (ValueError, TypeError): + pass + + if format_info.get("size"): + try: + metadata.file_size = int(format_info["size"]) + except (ValueError, TypeError): + pass + + # Validate we got essential data + if not metadata.codec: + metadata.status = "UNSCANNABLE" + metadata.error_message = "No codec information available" + elif not metadata.duration: + metadata.status = "UNSCANNABLE" + metadata.error_message = "No duration information available" + + return metadata + + @staticmethod + def _get_file_size(file_path: str) -> int: + """Get file size in bytes.""" + return os.path.getsize(file_path) diff --git a/src/scanner.py b/src/scanner.py new file mode 100644 index 0000000..6f0fc79 --- /dev/null +++ b/src/scanner.py @@ -0,0 +1,389 @@ +""" +Directory Scanner Service + +Walks /data/input to discover new/modified files, computes hashes, +probes video metadata, validates codecs, and syncs state to MariaDB. +Supports incremental scanning for efficient 30TB corpus handling. +""" + +import hashlib +import json +import logging +import os +import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime, timezone +from pathlib import Path +from typing import Dict, List, Optional, Tuple + +from codec_validator import CodecValidator +from prober import VideoMetadata, VideoProber + +logger = logging.getLogger(__name__) + + +class DirectoryScanner: + """Scan directories for new/modified video files and sync to DB.""" + + VIDEO_EXTENSIONS = {'.mp4', '.mov', '.avi', '.mkv', '.webm', '.flv', '.wmv'} + + def __init__( + self, + db_connector, + config: dict, + input_path: str = "/data/input", + scan_interval: int = 60, + walker_threads: int = 8, + ): + self.db = db_connector + self.config = config + self.input_path = Path(input_path) + self.scan_interval = scan_interval + self.walker_threads = walker_threads + self.prober = VideoProber( + timeout=config.get("scanner", {}).get("ffprobe_timeout_seconds", 10) + ) + self.codec_validator = CodecValidator( + whitelist=set(config.get("codec", {}).get("whitelist", [])), + default_status_on_error=config.get("codec", {}).get("default_status_on_error", "UNSCANNABLE"), + ) + self._running = False + self._scan_count = 0 + self._total_files_discovered = 0 + self._total_files_new = 0 + self._total_files_modified = 0 + self._total_files_unscannable = 0 + + def start(self): + """Start the scanner loop.""" + self._running = True + logger.info("Scanner starting: input_path=%s interval=%ds threads=%d", + self.input_path, self.scan_interval, self.walker_threads) + + while self._running: + try: + self._run_scan() + except Exception as e: + logger.error("Scanner error: %s", e, exc_info=True) + + # Sleep until next scan + for _ in range(self.scan_interval): + if not self._running: + break + time.sleep(1) + + def stop(self): + """Stop the scanner.""" + self._running = False + logger.info("Scanner stopping. Total scans: %d", self._scan_count) + + def _run_scan(self): + """Execute a single scan cycle.""" + scan_start = time.time() + self._scan_count += 1 + + logger.info("Scan #%d starting...", self._scan_count) + + # Get last scan time from DB + last_scan = self._get_last_scan_time() + + # Discover files + files_to_process, files_removed = self._discover_files(last_scan) + + # Process files in parallel + results = self._process_files_batch(files_to_process) + + # Update DB + scan_result = self._update_db(results, files_removed) + + # Record scan history + duration = time.time() - scan_start + self._record_scan_history(scan_result, duration) + + logger.info( + "Scan #%d complete: discovered=%d new=%d modified=%d unscannable=%d removed=%d duration=%.1fs", + self._scan_count, + scan_result["discovered"], + scan_result["new"], + scan_result["modified"], + scan_result["unscannable"], + len(files_removed), + duration, + ) + + def _get_last_scan_time(self) -> Optional[datetime]: + """Get the last scan time from DB.""" + result = self.db.fetchone( + "SELECT scan_end FROM scan_history ORDER BY id DESC LIMIT 1" + ) + if result and result.get("scan_end"): + return result["scan_end"] + return None + + def _discover_files(self, last_scan: Optional[datetime]) -> Tuple[List[Path], List[str]]: + """Discover new and modified files in the input directory.""" + files_to_process = [] + files_removed = [] + + if not self.input_path.exists(): + logger.warning("Input path does not exist: %s", self.input_path) + return files_to_process, files_removed + + # Get files from DB for comparison + if last_scan: + # Only check files modified after last scan + db_files = self.db.fetchall( + "SELECT file_path, last_scan_time FROM videos WHERE last_scan_time > %s", + (last_scan,), + ) + db_paths = {row["file_path"] for row in db_files} + + # Check for modified files + for row in db_files: + file_path = Path(row["file_path"]) + if file_path.exists() and file_path.is_file(): + try: + mtime = datetime.fromtimestamp( + file_path.stat().st_mtime, tz=timezone.utc + ) + if mtime > row["last_scan_time"]: + files_to_process.append(file_path) + except OSError: + pass + else: + # Full scan - walk the directory + logger.info("Full scan (no previous scan found). Walking %s...", self.input_path) + for root, dirs, filenames in os.walk(self.input_path): + for filename in filenames: + ext = Path(filename).suffix.lower() + if ext in self.VIDEO_EXTENSIONS: + file_path = Path(root) / filename + files_to_process.append(file_path) + + # Check for removed files (if we have a last scan) + if last_scan: + all_db_files = self.db.fetchall( + "SELECT file_path FROM videos WHERE status NOT IN ('REMOVED', 'ERROR')", + ) + current_paths = {str(f) for f in files_to_process} + for row in all_db_files: + fp = row["file_path"] + if fp not in current_paths and Path(fp).exists(): + # File still exists but not in current walk - might be in a new directory + pass + elif fp not in current_paths and not Path(fp).exists(): + files_removed.append(fp) + + logger.info("Discovered %d files to process, %d files removed", + len(files_to_process), len(files_removed)) + return files_to_process, files_removed + + def _process_files_batch(self, files: List[Path]) -> List[dict]: + """Process a batch of files in parallel.""" + results = [] + + with ThreadPoolExecutor(max_workers=self.walker_threads) as executor: + future_to_file = { + executor.submit(self._process_single_file, f): f + for f in files + } + + for future in as_completed(future_to_file): + file_path = future_to_file[future] + try: + result = future.result() + results.append(result) + except Exception as e: + logger.error("Error processing %s: %s", file_path, e, exc_info=True) + results.append({ + "file_path": str(file_path), + "status": "ERROR", + "error_message": str(e), + }) + + return results + + def _process_single_file(self, file_path: Path) -> dict: + """Process a single file: hash, probe, validate, determine status.""" + file_path_str = str(file_path) + result = {"file_path": file_path_str} + + try: + # Compute hash + file_hash = self._compute_hash(file_path_str) + result["file_hash"] = file_hash + + # Check for duplicate + existing = self.db.fetchone( + "SELECT id, status FROM videos WHERE file_hash = %s", + (file_hash,), + ) + if existing: + result["status"] = existing["status"] + result["video_id"] = existing["id"] + result["action"] = "duplicate" + logger.debug("Duplicate file found: %s (id=%d, status=%s)", + file_path_str, existing["id"], existing["status"]) + return result + + # Probe video + metadata = self.prober.probe(file_path_str) + result.update(metadata.to_dict()) + + # Validate codec + if metadata.is_valid: + is_supported, reason = self.codec_validator.validate(metadata.codec) + if is_supported: + result["status"] = "PENDING" + result["action"] = "new" + self._total_files_new += 1 + else: + result["status"] = "UNSCANNABLE" + result["error_message"] = f"Unsupported codec: {metadata.codec} ({reason})" + result["action"] = "unscannable" + self._total_files_unscannable += 1 + logger.warning("Unsupported codec for %s: %s", file_path_str, metadata.codec) + else: + result["status"] = "UNSCANNABLE" + result["action"] = "unscannable" + self._total_files_unscannable += 1 + + except Exception as e: + result["status"] = "ERROR" + result["error_message"] = str(e) + logger.error("Error processing %s: %s", file_path_str, e, exc_info=True) + + return result + + def _compute_hash(self, file_path: str) -> str: + """Compute SHA-256 hash of the first 1MB of a file.""" + chunk_size = 1024 * 1024 # 1MB + sha256 = hashlib.sha256() + + try: + with open(file_path, "rb") as f: + chunk = f.read(chunk_size) + if chunk: + sha256.update(chunk) + return sha256.hexdigest() + except (OSError, IOError) as e: + logger.error("Cannot hash file %s: %s", file_path, e) + return hashlib.sha256(file_path.encode()).hexdigest() # fallback + + def _update_db(self, results: List[dict], files_removed: List[str]) -> dict: + """Update database with scan results.""" + stats = {"discovered": len(results), "new": 0, "modified": 0, "unscannable": 0} + + # Batch insert new files + new_files = [r for r in results if r.get("action") == "new"] + if new_files: + self._batch_insert_new_files(new_files) + stats["new"] = len(new_files) + + # Update unscannable files + unscannable = [r for r in results if r.get("action") == "unscannable" and r.get("video_id")] + if unscannable: + self._batch_update_status(unscannable, "UNSCANNABLE") + stats["unscannable"] = len(unscannable) + + # Mark removed files + if files_removed: + self._mark_files_removed(files_removed) + stats["removed"] = len(files_removed) + + return stats + + def _batch_insert_new_files(self, files: List[dict]): + """Batch insert new files into DB.""" + now = datetime.now(timezone.utc).replace(tzinfo=None) + + for file_info in files: + try: + self.db.execute( + """INSERT INTO videos + (file_path, file_hash, resolution_w, resolution_h, codec, + duration, status, last_scan_time, created_at, updated_at) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)""", + ( + file_info["file_path"], + file_info["file_hash"], + file_info.get("resolution_w"), + file_info.get("resolution_h"), + file_info.get("codec"), + file_info.get("duration"), + "PENDING", + now, + now, + now, + ), + transaction=True, + ) + except Exception as e: + logger.error("Failed to insert %s: %s", file_info["file_path"], e) + + def _batch_update_status(self, files: List[dict], status: str): + """Batch update file status.""" + now = datetime.now(timezone.utc).replace(tzinfo=None) + + for file_info in files: + video_id = file_info.get("video_id") + if video_id: + try: + self.db.execute( + """UPDATE videos SET status = %s, error_message = %s, + last_scan_time = %s, updated_at = %s + WHERE id = %s""", + (status, file_info.get("error_message"), now, now, video_id), + transaction=True, + ) + except Exception as e: + logger.error("Failed to update %s: %s", file_info["file_path"], e) + + def _mark_files_removed(self, file_paths: List[str]): + """Mark files as removed in DB.""" + now = datetime.now(timezone.utc).replace(tzinfo=None) + + for fp in file_paths: + try: + self.db.execute( + "UPDATE videos SET status = 'REMOVED', last_scan_time = %s, updated_at = %s WHERE file_path = %s", + (now, now, fp), + transaction=True, + ) + except Exception as e: + logger.error("Failed to mark %s as removed: %s", fp, e) + + def _record_scan_history(self, stats: dict, duration: float): + """Record scan history in DB.""" + now = datetime.now(timezone.utc).replace(tzinfo=None) + + self.db.execute( + """INSERT INTO scan_history + (scan_start, scan_end, files_discovered, files_new, files_modified, + files_removed, files_unscannable, duration_seconds, status) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)""", + ( + now, + now, + stats.get("discovered", 0), + stats.get("new", 0), + stats.get("modified", 0), + stats.get("removed", 0), + stats.get("unscannable", 0), + duration, + "COMPLETED", + ), + transaction=True, + ) + + def get_stats(self) -> dict: + """Get scanner statistics.""" + return { + "scan_count": self._scan_count, + "total_files_discovered": self._total_files_discovered, + "total_files_new": self._total_files_new, + "total_files_modified": self._total_files_modified, + "total_files_unscannable": self._total_files_unscannable, + "input_path": str(self.input_path), + "is_running": self._running, + }