After story 2

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)})"
+        )

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(),
+        }

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)

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,
+        }