backend/health_checker.py

Rama master · réplica de lectura del repositorio original.

"""Subtitle health-check engine: orchestration, scoring, and auto-fix.

The detection rules live in health_checks.py, the repair routines in
health_fixes.py. This module wires them together:

- run_health_checks(path) loads the subtitle, runs every check and
  returns a scored report.
- apply_fixes(path, names) backs the file up, runs the requested fixers
  and re-scores.

The check/fix functions and registries are re-exported below so existing
imports (`from health_checker import check_duplicate_lines`, etc.) keep
working unchanged.
"""

import logging
import os
import shutil
from datetime import UTC, datetime

from health_checks import (
    ALL_CHECKS,
    _calculate_score,
    check_duplicate_lines,
    check_empty_events,
    check_encoding_issues,
    check_excessive_duration,
    check_line_too_long,
    check_missing_newlines,
    check_missing_styles,
    check_negative_timing,
    check_timing_overlaps,
    check_zero_duration,
)
from health_fixes import (
    FIX_MAP,
    fix_duplicate_lines,
    fix_empty_events,
    fix_missing_styles,
    fix_negative_timing,
    fix_timing_overlaps,
    fix_zero_duration,
)
from subtitle_encoding import load_subtitle_file

logger = logging.getLogger(__name__)

# Known auto-fixable check names
FIXABLE_CHECKS = {
    "duplicate_lines",
    "timing_overlaps",
    "missing_styles",
    "empty_events",
    "negative_timing",
    "zero_duration",
}

__all__ = [
    "ALL_CHECKS",
    "FIXABLE_CHECKS",
    "FIX_MAP",
    "_calculate_score",
    "apply_fixes",
    "check_duplicate_lines",
    "check_empty_events",
    "check_encoding_issues",
    "check_excessive_duration",
    "check_line_too_long",
    "check_missing_newlines",
    "check_missing_styles",
    "check_negative_timing",
    "check_timing_overlaps",
    "check_zero_duration",
    "fix_duplicate_lines",
    "fix_empty_events",
    "fix_missing_styles",
    "fix_negative_timing",
    "fix_timing_overlaps",
    "fix_zero_duration",
    "run_health_checks",
]


def run_health_checks(file_path: str) -> dict:
    """Run all health checks on a subtitle file.

    Args:
        file_path: Absolute path to a subtitle file (.ass, .srt, .ssa).

    Returns:
        Dict with file_path, checks_run, issues, score, checked_at.
    """

    all_issues = []
    checks_run = 0

    # Read raw bytes for encoding check
    raw_bytes = None
    try:
        with open(file_path, "rb") as f:
            raw_bytes = f.read()
    except OSError as e:
        logger.error("Cannot read file %s: %s", file_path, e)
        return {
            "file_path": file_path,
            "checks_run": 0,
            "issues": [],
            "score": 0,
            "checked_at": datetime.now(UTC),
        }

    # Parse subtitle
    try:
        subs = load_subtitle_file(file_path)
    except Exception as e:
        logger.error("Cannot parse subtitle %s: %s", file_path, e)
        return {
            "file_path": file_path,
            "checks_run": 0,
            "issues": [
                {
                    "check": "parse_error",
                    "severity": "error",
                    "message": f"Failed to parse subtitle file: {e}",
                    "line": None,
                    "auto_fixable": False,
                    "fix": None,
                }
            ],
            "score": 0,
            "checked_at": datetime.now(UTC),
        }

    # Run encoding check (needs raw_bytes)
    encoding_issues = check_encoding_issues(subs, raw_bytes)
    all_issues.extend(encoding_issues)
    checks_run += 1

    # Run all other checks
    for check_fn in ALL_CHECKS:
        try:
            issues = check_fn(subs)
            all_issues.extend(issues)
        except Exception as e:
            logger.warning("Check %s failed for %s: %s", check_fn.__name__, file_path, e)
        checks_run += 1

    score = _calculate_score(all_issues)

    return {
        "file_path": file_path,
        "checks_run": checks_run,
        "issues": all_issues,
        "score": score,
        "checked_at": datetime.now(UTC),
    }


def apply_fixes(file_path: str, fix_names: list) -> dict:
    """Apply requested fixes to a subtitle file.

    Creates a .bak backup before saving. Re-runs health check after fixes
    to return the updated score.

    Args:
        file_path: Absolute path to subtitle file.
        fix_names: List of check names to fix (e.g., ["duplicate_lines"]).

    Returns:
        Dict with fixes_applied, counts, new_score.
    """

    # Create backup using same .bak pattern as tools.py
    base, ext = os.path.splitext(file_path)
    bak_path = f"{base}.bak{ext}"
    shutil.copy2(file_path, bak_path)

    subs = load_subtitle_file(file_path)

    fixes_applied = []
    counts = {}

    for name in fix_names:
        fix_fn = FIX_MAP.get(name)
        if fix_fn is None:
            logger.warning("Unknown fix: %s", name)
            continue
        count = fix_fn(subs)
        if count > 0:
            fixes_applied.append(name)
            counts[name] = count

    # Save modified subtitle
    subs.save(file_path)

    # Re-run health check for updated score
    result = run_health_checks(file_path)

    return {
        "fixes_applied": fixes_applied,
        "counts": counts,
        "new_score": result["score"],
    }