python-basic-pitch/report_audio_review_queue.py

from __future__ import annotations

import argparse
import csv
import sys
from dataclasses import dataclass
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

import main as pitch_main
from harmonic_pipeline import analyze_harmonic_audio


PROJECT_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_AUDIO_DIR = PROJECT_ROOT / "audios-revisao"
DEFAULT_OUTPUT_MD = DEFAULT_AUDIO_DIR / "relatorio-revisao.md"
DEFAULT_OUTPUT_CSV = DEFAULT_AUDIO_DIR / "relatorio-revisao.csv"
SUPPORTED_EXTENSIONS = {".mp3", ".wav", ".ogg", ".mp4", ".m4a", ".aac", ".flac"}
FAIXAS = {
    "violao": (80.0, 1200.0),
    "teclado": (27.5, 4200.0),
    "sax_alto": (130.0, 900.0),
}


@dataclass(frozen=True)
class ReviewItem:
    arquivo: str
    instrumento: str
    tipo_analise: str
    saida_detectada: str
    tom: str
    modo: str
    observacoes: str
    status: str = "pendente"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Gera fila de revisao para audios reais sem ground truth.")
    parser.add_argument("--audio-dir", type=Path, default=DEFAULT_AUDIO_DIR, help="Pasta com os audios reais para revisao.")
    parser.add_argument("--output-md", type=Path, default=DEFAULT_OUTPUT_MD, help="Arquivo Markdown de saida.")
    parser.add_argument("--output-csv", type=Path, default=DEFAULT_OUTPUT_CSV, help="Arquivo CSV de saida.")
    return parser.parse_args()


def infer_instrument(path: Path) -> str:
    lowered = path.stem.casefold()
    if "sax" in lowered:
        return "sax_alto"
    if "teclado" in lowered or "piano" in lowered or "keys" in lowered:
        return "teclado"
    return "violao"


def collect_audio_files(folder: Path) -> list[Path]:
    if not folder.exists():
        return []
    return sorted(
        [
            path
            for path in folder.iterdir()
            if path.is_file() and path.suffix.casefold() in SUPPORTED_EXTENSIONS
        ],
        key=lambda item: item.name.casefold(),
    )


def build_review_item(path: Path) -> ReviewItem:
    instrumento = infer_instrument(path)
    if instrumento == "sax_alto":
        result = pitch_main.analisar_melodico(path, instrumento)
        detected = " ".join(result.get("notas_resumo", []))
        total = result.get("total_eventos_pitch", 0)
        observacoes = f"eventos={total}"
        tipo = "melodico"
    else:
        result = analyze_harmonic_audio(str(path), instrumento, FAIXAS[instrumento])
        detected = result.get("cifra_palco", "").strip()
        aux = result.get("base_harmonica_auxiliar", "").strip()
        observacoes = f"aux={aux}" if aux else ""
        tipo = "harmonico"

    return ReviewItem(
        arquivo=path.name,
        instrumento=instrumento,
        tipo_analise=tipo,
        saida_detectada=detected,
        tom=str(result.get("tom", "")),
        modo=str(result.get("modo", "")),
        observacoes=observacoes,
    )


def write_csv(path: Path, items: list[ReviewItem]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.writer(handle)
        writer.writerow(
            [
                "arquivo",
                "instrumento",
                "tipo_analise",
                "saida_detectada",
                "tom",
                "modo",
                "status",
                "cifra_correta_ou_notas",
                "observacoes",
            ]
        )
        for item in items:
            writer.writerow(
                [
                    item.arquivo,
                    item.instrumento,
                    item.tipo_analise,
                    item.saida_detectada,
                    item.tom,
                    item.modo,
                    item.status,
                    "",
                    item.observacoes,
                ]
            )


def write_markdown(path: Path, items: list[ReviewItem]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    lines = [
        "# Fila de Revisao de Audios Reais",
        "",
        "Preencha a coluna `cifra_correta_ou_notas` e troque o `status` para `confirmado` ou `corrigir` conforme a revisao musical.",
        "",
        "| arquivo | instrumento | tipo | saida_detectada | tom | modo | status | cifra_correta_ou_notas | observacoes |",
        "|---|---|---|---|---|---|---|---|---|",
    ]
    for item in items:
        lines.append(
            f"| {escape_md(item.arquivo)} | {item.instrumento} | {item.tipo_analise} | "
            f"{escape_md(item.saida_detectada)} | {item.tom} | {item.modo} | {item.status} |  | {escape_md(item.observacoes)} |"
        )
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def escape_md(value: str) -> str:
    return (value or "").replace("|", "\\|").replace("\n", " ").strip()


def main() -> int:
    args = parse_args()
    files = collect_audio_files(args.audio_dir)
    items = [build_review_item(path) for path in files]
    write_csv(args.output_csv, items)
    write_markdown(args.output_md, items)
    print(f"Audios analisados: {len(items)}")
    print(f"Markdown: {args.output_md}")
    print(f"CSV: {args.output_csv}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())