Blablablab/audio-classification

1"""2Config file validator CLI.3 4Usage:5    python -m potato.validate_cli <config.yaml>6    python -m potato.validate_cli <config.yaml> --strict7    python -m potato.validate_cli <config.yaml> --json8 9Checks performed:10    1. YAML parses cleanly11    2. All required top-level keys present (item_properties, data_files,12       task_dir, output_annotation_dir, annotation_task_name)13    3. Deep structural validation (annotation_schemes, phases, auth, etc.)14       via config_module.validate_yaml_structure()15    4. Unrecognized keys at any nesting level via validate_unknown_keys()16 17Exit codes:18    0 — all checks passed19    1 — fatal errors found (invalid YAML, missing required fields,20        structural errors)21    2 — warnings only (unrecognized keys) and --strict was passed22 23In default mode unknown keys are reported but do not fail the exit24code. Use --strict to treat them as fatal (useful for CI).25"""26 27from __future__ import annotations28 29import argparse30import json31import logging32import os33import sys34from dataclasses import dataclass, field35from typing import List, Optional36 37import yaml38 39from potato.server_utils.config_module import (40    ConfigValidationError,41    ConfigSecurityError,42    validate_yaml_structure,43)44 45 46@dataclass47class ValidationReport:48    config_file: str49    ok: bool50    errors: List[str] = field(default_factory=list)51    unknown_keys: List[str] = field(default_factory=list)52    other_warnings: List[str] = field(default_factory=list)53 54    def to_dict(self) -> dict:55        return {56            "config_file": self.config_file,57            "ok": self.ok,58            "errors": self.errors,59            "unknown_keys": self.unknown_keys,60            "other_warnings": self.other_warnings,61        }62 63 64class _WarningCollector(logging.Handler):65    """Captures WARNING-level records from config_module.66 67    `validate_unknown_keys()` logs unrecognized keys as WARNINGs on the68    `potato.server_utils.config_module` logger rather than raising. We69    install this handler around the validation call to collect them.70    """71 72    def __init__(self):73        super().__init__(level=logging.WARNING)74        self.unknown_keys: List[str] = []75        self.other_warnings: List[str] = []76 77    def emit(self, record: logging.LogRecord) -> None:78        msg = record.getMessage()79        # validate_unknown_keys emits this exact prefix80        if "Unrecognized config key" in msg:81            self.unknown_keys.append(msg)82        else:83            self.other_warnings.append(msg)84 85 86def validate_config_file(config_file: str) -> ValidationReport:87    """Validate a single config file and return a structured report."""88    report = ValidationReport(config_file=config_file, ok=True)89 90    # Existence and readability91    if not os.path.isfile(config_file):92        report.ok = False93        report.errors.append(f"Config file not found: {config_file}")94        return report95 96    # YAML parse97    try:98        with open(config_file, "r", encoding="utf-8") as f:99            config_data = yaml.safe_load(f)100    except yaml.YAMLError as e:101        report.ok = False102        report.errors.append(f"YAML parse error: {e}")103        return report104    except OSError as e:105        report.ok = False106        report.errors.append(f"Could not read file: {e}")107        return report108 109    if config_data is None:110        report.ok = False111        report.errors.append("Config file is empty")112        return report113 114    if not isinstance(config_data, dict):115        report.ok = False116        report.errors.append(117            f"Config root must be a YAML mapping (got {type(config_data).__name__})"118        )119        return report120 121    # Capture unknown-key warnings while running the full validator122    logger = logging.getLogger("potato.server_utils.config_module")123    collector = _WarningCollector()124    # Save and override propagation so the handler actually sees records125    prior_level = logger.level126    prior_propagate = logger.propagate127    logger.addHandler(collector)128    logger.setLevel(logging.WARNING)129    logger.propagate = False130 131    config_file_dir = os.path.dirname(os.path.abspath(config_file))132    try:133        validate_yaml_structure(134            config_data,135            project_dir=config_file_dir,136            config_file_dir=config_file_dir,137        )138    except (ConfigValidationError, ConfigSecurityError) as e:139        report.ok = False140        report.errors.append(str(e))141    except Exception as e:142        # Surface unexpected errors but do not crash the CLI143        report.ok = False144        report.errors.append(f"Unexpected validation error ({type(e).__name__}): {e}")145    finally:146        logger.removeHandler(collector)147        logger.setLevel(prior_level)148        logger.propagate = prior_propagate149 150    report.unknown_keys = collector.unknown_keys151    report.other_warnings = collector.other_warnings152    return report153 154 155def _format_human(report: ValidationReport) -> str:156    lines = []157    lines.append(f"Config: {report.config_file}")158    if report.errors:159        lines.append("")160        lines.append("ERRORS:")161        for e in report.errors:162            lines.append(f"  - {e}")163    if report.unknown_keys:164        lines.append("")165        lines.append("UNKNOWN KEYS:")166        for w in report.unknown_keys:167            lines.append(f"  - {w}")168    if report.other_warnings:169        lines.append("")170        lines.append("OTHER WARNINGS:")171        for w in report.other_warnings:172            lines.append(f"  - {w}")173    lines.append("")174    if report.ok and not report.unknown_keys and not report.other_warnings:175        lines.append("OK — no issues found.")176    elif report.ok and report.unknown_keys:177        lines.append(178            f"OK with {len(report.unknown_keys)} unknown-key warning(s). "179            "Re-run with --strict to fail on unknown keys."180        )181    else:182        lines.append(f"FAILED — {len(report.errors)} error(s).")183    return "\n".join(lines)184 185 186def main(argv: Optional[List[str]] = None) -> int:187    parser = argparse.ArgumentParser(188        prog="potato.validate_cli",189        description=(190            "Validate a Potato YAML config file: checks required keys, "191            "deep structural constraints, and unrecognized keys."192        ),193        formatter_class=argparse.RawDescriptionHelpFormatter,194        epilog=(195            "Exit codes:\n"196            "  0 — all checks passed\n"197            "  1 — fatal errors (invalid YAML, missing required fields,\n"198            "      structural errors, or unknown keys with --strict)\n"199        ),200    )201    parser.add_argument("config_file", help="Path to YAML config file")202    parser.add_argument(203        "--strict",204        action="store_true",205        help="Treat unknown keys as fatal (exit 1 instead of just warning)",206    )207    parser.add_argument(208        "--json",209        action="store_true",210        dest="emit_json",211        help="Emit JSON report on stdout instead of human-readable text",212    )213    parser.add_argument(214        "--quiet",215        action="store_true",216        help="Only emit output if there are errors or (in --strict) warnings",217    )218    args = parser.parse_args(argv)219 220    report = validate_config_file(args.config_file)221 222    fatal = not report.ok or (args.strict and bool(report.unknown_keys))223 224    if args.emit_json:225        print(json.dumps(report.to_dict(), indent=2))226    elif not args.quiet or fatal or report.unknown_keys or report.other_warnings:227        print(_format_human(report))228 229    return 1 if fatal else 0230 231 232if __name__ == "__main__":233    sys.exit(main())234