"""Personal data still readable in an exported video: every frame (or 4 per second) read by OCR.

A blur in the project is a layer on top: a frame without it (the first frame, one frame before the blur comes
in, a menu that opens for 0.15 s, a line that moved) shows the data, and a contact sheet with one frame per
second does not catch it. Seen on 29/09/2026: 4 leaks in a draft that the contact sheet did not show.

This helper reads the video frame by frame and looks for:

  termo        the words given with --termos / --termos-arquivo (names of people, of chats, of projects, a
               company, an address), accents and case ignored, tolerant to one OCR slip in long words;
  e-mail       anything@anything.tld
  cpf / cnpj   formatted, or 11 digits with valid check digits
  telefone     (11) 91234-5678, 11 91234 5678, +55 ...
  cartao       4 groups of 4 digits that pass the card check (Luhn)
  perfil       @handle (Instagram, TikTok)
  nome?        a common Brazilian first name followed by a capitalized word ("Maria Souza"): a guess, confira

Every frame is read by default up to 3 minutes of video (a leak can last one frame); longer videos are
sampled at 4 frames per second unless --todos is given. It says, for each finding, the text read, the time and the frame (from and to) and where it is in the frame
(x, y, width, height in pixels of the video), so the blur can be placed or its window widened.

Reader: Apple Vision on macOS (the same compiled reader as texto_na_tela.py); Tesseract on other systems when
it is installed (`tesseract` in the PATH, with the Portuguese language). Without a reader it stops and says
how to check by eye.

Usage:
    uv run --directory '<SKILL>' python helpers/conferir_dados.py '<VIDEO>' [--termos 'Maria Souza, Loja X']
        [--termos-arquivo termos.txt] [--ignorar 'André, @andre'] [--todos | --fps 4] [--de S --ate S]
        [--sem-padroes] [--json relatorio.json]
Exit code 0 when nothing was found, 1 when something was found, 2 on error.
A clean result is not proof: OCR misses tiny, blurred or rotated text. Look at what it cannot read.
"""

from __future__ import annotations

import argparse
import difflib
import json
import math
import re
import shutil
import subprocess
import sys
import tempfile
import time
import unicodedata
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from common import fmt_time, probe_video, require_binary  # noqa: E402
import texto_na_tela as tnt  # noqa: E402

FPS_PADRAO = 4.0         # frames per second read when the video is long (or with --fps)
TODOS_ATE_S = 180.0      # up to this length every frame is read by default (a leak can last 1 frame)
LOTE = 200               # frames read per batch (progress and temp disk)
LADO_MAX = 1920          # OCR on the video size, capped on the long side
CONF_MIN = 0.30
FUZZY = 0.86             # similarity for a term of 6+ letters read with a slip

# common Brazilian first names (IBGE 2010 top names and frequent ones); a name + a capitalized word is flagged
NOMES = """
ana maria jose joao antonio francisco carlos paulo pedro lucas luiz marcos luis gabriel rafael daniel marcelo
bruno eduardo felipe raimundo rodrigo manoel mateus andre fernando fabio leonardo gustavo guilherme leandro
tiago thiago anderson ricardo jorge alexandre roberto edson diego vitor victor sergio claudio matheus thiago
julio juliana adriana marcia fernanda patricia aline sandra camila amanda bruna jessica leticia julia luciana
vanessa mariana gabriela vera vitoria larissa claudia beatriz rita luana sonia renata eliane josefa simone
natalia cristiane carla debora rosangela jaqueline rosa daniela aparecida marlene terezinha raimunda andreia
fabiana lucia raquel angela rafaela joana luzia isabel cleide tatiane priscila carolina vivian viviane helena
alice laura manuela valentina sophia sofia heloisa isadora lorena livia cecilia elisa yasmin bianca giovanna
enzo arthur davi heitor bernardo samuel murilo miguel henrique caio igor otavio vinicius kaique joaquim renan
""".split()
NOMES_SET = set(NOMES)

# "a@b.com" or, as the OCR sometimes spaces it, "a @ b.com" (never "segue a @perfil"); a real ending (.com, .br)
RE_EMAIL = re.compile(r"[\w.+-]{2,}(?:@| @ )[\w-]+(?:\.[\w-]+)*\.[a-zA-Z]{2,6}(?![\w])")
RE_CPF_FMT = re.compile(r"(?<!\d)\d{3}\.\s?\d{3}\.\s?\d{3}\s?-\s?\d{2}(?!\d)")
RE_ONZE = re.compile(r"(?<![\d.])\d{11}(?![\d.])")
RE_CNPJ = re.compile(r"(?<!\d)\d{2}\.?\d{3}\.?\d{3}\s?/\s?\d{4}\s?-?\s?\d{2}(?!\d)")
RE_FONE = re.compile(r"(?<![\d/.])(?:\+?\s?55\s?)?(?:\(\s?\d{2}\s?\)|\d{2})\s?9?\s?\d{4}\s?[-.\s]\s?\d{4}(?![\d/])")
RE_CARTAO = re.compile(r"(?<!\d)(?:\d{4}[ -]){3}\d{4}(?!\d)")
RE_PERFIL = re.compile(r"(?<![\w@.])@[A-Za-z0-9_.]{3,30}")
RE_NOME = re.compile(r"\b([A-ZÁÉÍÓÚÂÊÔÃÕÇ][a-záéíóúâêôãõç]+)\s+(?:(?:d[aeo]s?)\s+)?([A-ZÁÉÍÓÚÂÊÔÃÕÇ][a-záéíóúâêôãõç]+)")


def norm(texto: str) -> str:
    t = unicodedata.normalize("NFKD", (texto or "").lower())
    t = "".join(c for c in t if not unicodedata.combining(c))
    return re.sub(r"\s+", " ", re.sub(r"[^\w@.\- ]+", " ", t)).strip()


def cpf_valido(d: str) -> bool:
    if len(d) != 11 or d == d[0] * 11:
        return False
    for n in (9, 10):
        s = sum(int(d[i]) * (n + 1 - i) for i in range(n))
        if (s * 10 % 11) % 10 != int(d[n]):
            return False
    return True


def luhn(d: str) -> bool:
    tot, dobra = 0, False
    for c in reversed(d):
        v = int(c)
        if dobra:
            v = v * 2 - 9 if v * 2 > 9 else v * 2
        tot += v
        dobra = not dobra
    return tot % 10 == 0


def termos_da_lista(texto: str) -> list[str]:
    partes = re.split(r"[,;\n]", texto or "")
    return [p.strip() for p in partes if p.strip()]


def achar_termo(linha_norm: str, termo_norm: str) -> bool:
    if not termo_norm:
        return False
    if re.search(r"(?<![a-z0-9])" + re.escape(termo_norm) + r"(?![a-z0-9])", linha_norm):
        return True
    if len(termo_norm.replace(" ", "")) < 6:
        return False
    toks, n = linha_norm.split(), len(termo_norm.split())
    for i in range(0, max(1, len(toks) - n + 1)):
        jan = " ".join(toks[i:i + n])
        if difflib.SequenceMatcher(a=jan, b=termo_norm).ratio() >= FUZZY:
            return True
    return False


def achados_na_linha(texto: str, termos: list[str], padroes: bool = True,
                     ignorar: list[str] | None = None) -> list[tuple[str, str]]:
    """[(categoria, texto achado)] in one line read by the OCR."""
    out: list[tuple[str, str]] = []
    ln = norm(texto)
    for t in termos:
        if achar_termo(ln, norm(t)):
            out.append(("termo", t))
    if padroes:
        for m in RE_EMAIL.finditer(texto):
            out.append(("e-mail", re.sub(r"\s", "", m.group(0))))
        for m in RE_CPF_FMT.finditer(texto):
            out.append(("cpf", m.group(0)))
        for m in RE_ONZE.finditer(texto):
            if cpf_valido(m.group(0)):
                out.append(("cpf", m.group(0)))
        for m in RE_CNPJ.finditer(texto):
            if "/" in m.group(0) or "." in m.group(0):
                out.append(("cnpj", m.group(0)))
        for m in RE_CARTAO.finditer(texto):
            if luhn(re.sub(r"\D", "", m.group(0))):
                out.append(("cartao", m.group(0)))
        for m in RE_FONE.finditer(texto):
            dig = re.sub(r"\D", "", m.group(0))
            if 10 <= len(dig) <= 13 and not any(c == "cartao" and dig in re.sub(r"\D", "", x) for c, x in out):
                out.append(("telefone", m.group(0).strip()))
        emails = " ".join(x for c, x in out if c == "e-mail")
        for m in RE_PERFIL.finditer(texto):
            if m.group(0)[1:] not in emails:
                out.append(("perfil", m.group(0)))
        for m in RE_NOME.finditer(texto):
            if norm(m.group(1)) in NOMES_SET:
                out.append(("nome?", f"{m.group(1)} {m.group(2)}"))
    if ignorar:
        ig = [norm(x) for x in ignorar if x.strip()]
        out = [(c, x) for c, x in out if not any(i and (i in norm(x) or norm(x) in i) for i in ig)]
    vistos, uniq = set(), []
    for c, x in out:
        if (c, norm(x)) not in vistos:
            vistos.add((c, norm(x)))
            uniq.append((c, x))
    return uniq


# ------------------------------------------------------------------ reading the frames

def leitor() -> tuple[str, object] | None:
    """("vision", binary) on macOS, ("tesseract", path) elsewhere, None without any reader."""
    b = tnt.compilar_vision()
    if b:
        return "vision", b
    t = shutil.which("tesseract")
    if t:
        return "tesseract", t
    return None


def parse_tesseract_tsv(tsv: str, largura: int, altura: int) -> list[dict]:
    """Lines of a `tesseract ... tsv` output, in the same shape as the Vision reader (fractions, y from the bottom)."""
    linhas: dict[tuple, dict] = {}
    for row in tsv.splitlines()[1:]:
        c = row.split("\t")
        if len(c) < 12 or c[0] != "5" or not c[11].strip():
            continue
        key = (c[2], c[3], c[4])
        x, y, w, h, conf = int(c[6]), int(c[7]), int(c[8]), int(c[9]), float(c[10])
        ln = linhas.setdefault(key, {"t": [], "c": [], "x0": x, "y0": y, "x1": x + w, "y1": y + h})
        ln["t"].append(c[11])
        ln["c"].append(conf)
        ln["x0"], ln["y0"] = min(ln["x0"], x), min(ln["y0"], y)
        ln["x1"], ln["y1"] = max(ln["x1"], x + w), max(ln["y1"], y + h)
    out = []
    for ln in linhas.values():
        out.append({"t": " ".join(ln["t"]), "c": max(0.0, min(ln["c"])) / 100.0,
                    "x": ln["x0"] / largura, "w": (ln["x1"] - ln["x0"]) / largura,
                    "y": 1 - ln["y1"] / altura, "h": (ln["y1"] - ln["y0"]) / altura})
    return out


def ler_pasta(motor: tuple[str, object], pasta: Path) -> dict[str, list[dict]]:
    if motor[0] == "vision":
        saida = pasta / "ocr.json"
        subprocess.run([str(motor[1]), str(pasta), str(saida)], capture_output=True, text=True, check=True)
        return json.loads(saida.read_text())
    out: dict[str, list[dict]] = {}
    from PIL import Image  # noqa: PLC0415
    for img in sorted(pasta.glob("f_*.jpg")):
        w, h = Image.open(img).size
        tsv = ""
        for lang in ("por", "eng"):
            r = subprocess.run([str(motor[1]), str(img), "-", "-l", lang, "tsv"], capture_output=True, text=True)
            if r.returncode == 0:
                tsv = r.stdout
                break
        out[img.name] = parse_tesseract_tsv(tsv, w, h)
    return out


def extrair(video: Path, pasta: Path, primeiro: int, n: int, rate: float, fps_video: float, amostra: bool,
            escala: str) -> int:
    """n frames as f_00001.jpg..., starting at index `primeiro` of the grid of `rate` per second (every frame of the
    video when not `amostra`). Counted in frames, not in seconds: a batch cut by time repeated or skipped the frame
    on its border and the frame numbers came out one off."""
    t0 = max(0.0, primeiro / rate - 0.5 / fps_video)      # half a frame early: the frame on the grid is the first
    vf = ([f"fps={rate}"] if amostra else []) + ([escala] if escala else [])
    cmd = ["ffmpeg", "-v", "error", "-y", "-ss", f"{t0:.4f}", "-i", str(video), "-frames:v", str(n), "-an"]
    if vf:
        cmd += ["-vf", ",".join(vf)]
    if not amostra:
        cmd += ["-fps_mode", "passthrough"]
    cmd += ["-q:v", "2", str(pasta / "f_%05d.jpg")]
    subprocess.run(cmd, capture_output=True, text=True, check=True)
    return len(list(pasta.glob("f_*.jpg")))


def conferir(video: Path, termos: list[str], *, fps: float | None = FPS_PADRAO, de: float = 0.0,
             ate: float | None = None, padroes: bool = True, ignorar: list[str] | None = None,
             motor: tuple[str, object] | None = None, progresso=print) -> dict:
    require_binary("ffmpeg")
    info = probe_video(video)
    W, H = int(info.get("display_width") or info["width"]), int(info.get("display_height") or info["height"])
    fps_video = float(info.get("fps") or 30.0)
    dur_total = float(info["duration"])
    fim = min(dur_total, ate) if ate is not None else dur_total
    motor = motor or leitor()
    if motor is None:
        raise RuntimeError("sem leitor de texto neste computador")
    rate = fps or fps_video
    escala = f"scale='if(gt(iw,ih),min(iw,{LADO_MAX}),-2)':'if(gt(iw,ih),-2,min(ih,{LADO_MAX}))'" \
        if max(W, H) > LADO_MAX else ""
    ocorr: list[dict] = []
    idx = int(math.ceil(max(0.0, de) * rate - 1e-6))
    ultimo = int(math.floor(fim * rate - 1e-6))          # the last frame starts before the end
    total = max(0, ultimo - idx + 1)
    lidos = 0
    t_ini = time.monotonic()
    while idx <= ultimo:
        n_pedido = min(LOTE, ultimo - idx + 1)
        with tempfile.TemporaryDirectory(prefix="conferir-dados-") as tmp:
            pasta = Path(tmp)
            n = extrair(video, pasta, idx, n_pedido, rate, fps_video, bool(fps), escala)
            bruto = ler_pasta(motor, pasta) if n else {}
        for k in range(n):
            tq = (idx + k) / rate
            for it in bruto.get(f"f_{k + 1:05d}.jpg", []):
                if float(it.get("c", 1)) < CONF_MIN:
                    continue
                for cat, achado in achados_na_linha(str(it.get("t", "")), termos, padroes, ignorar):
                    x, y, w, h = (float(it[c]) for c in ("x", "y", "w", "h"))
                    ocorr.append({"categoria": cat, "texto": achado, "linha": it.get("t", ""), "t": round(tq, 3),
                                  "quadro": int(round(tq * fps_video)),
                                  "caixa": [round(x * W), round((1 - y - h) * H), round(w * W), round(h * H)]})
        lidos += n
        progresso(f"  lidos {lidos} de {total} quadros ({time.monotonic() - t_ini:.0f} s)")
        idx += n_pedido
        if n == 0:
            break
    return {"video": str(video), "largura": W, "altura": H, "fps_video": fps_video,
            "amostragem": "todos os quadros" if not fps else f"{fps:g} por segundo", "leitor": motor[0],
            "de": round(max(0.0, de), 3), "ate": round(fim, 3), "quadros_lidos": lidos,
            "termos": termos, "padroes": padroes, "achados": juntar(ocorr, 1.0 / rate)}


def juntar(ocorr: list[dict], passo: float) -> list[dict]:
    """Occurrences of the same text in consecutive frames become one finding with from/to."""
    grupos: dict[tuple, list[dict]] = {}
    for o in sorted(ocorr, key=lambda o: o["t"]):
        key = (o["categoria"], norm(o["texto"]))
        if key not in grupos:     # the same text read with a slip on one frame ("(17)" for "(11)") joins it
            key = next((k for k in grupos if k[0] == key[0]
                        and difflib.SequenceMatcher(a=k[1], b=key[1]).ratio() >= FUZZY), key)
        grupos.setdefault(key, []).append(o)
    out = []
    for (cat, _), oc in grupos.items():
        janelas: list[list[dict]] = []
        for o in oc:
            if janelas and o["t"] - janelas[-1][-1]["t"] <= passo * 1.6:
                janelas[-1].append(o)
            else:
                janelas.append([o])
        for j in janelas:
            xs = [o["caixa"] for o in j]
            x0, y0 = min(c[0] for c in xs), min(c[1] for c in xs)
            x1, y1 = max(c[0] + c[2] for c in xs), max(c[1] + c[3] for c in xs)
            out.append({"categoria": cat, "texto": j[0]["texto"], "linha": j[0]["linha"],
                        "de": j[0]["t"], "ate": j[-1]["t"], "quadro_de": j[0]["quadro"], "quadro_ate": j[-1]["quadro"],
                        "vezes": len(j), "caixa": [x0, y0, x1 - x0, y1 - y0]})
    return sorted(out, key=lambda a: (a["de"], a["categoria"]))


NOME_CAT = {"termo": "termo", "e-mail": "e-mail", "cpf": "CPF", "cnpj": "CNPJ", "telefone": "telefone",
            "cartao": "cartão", "perfil": "perfil (@)", "nome?": "possível nome"}


def relatorio(rep: dict) -> list[str]:
    ach = rep["achados"]
    linhas = [f"conferir_dados: {rep['quadros_lidos']} quadros lidos ({rep['amostragem']}) de "
              f"{Path(rep['video']).name}, {fmt_time(rep['de'])} a {fmt_time(rep['ate'])}, leitor {rep['leitor']} · "
              f"{len(ach)} achado(s)"]
    for a in ach:
        quando = (f"{fmt_time(a['de'])} (quadro {a['quadro_de']})" if a["quadro_de"] == a["quadro_ate"] else
                  f"{fmt_time(a['de'])} a {fmt_time(a['ate'])} (quadros {a['quadro_de']} a {a['quadro_ate']})")
        x, y, w, h = a["caixa"]
        extra = "  confira se pode aparecer" if a["categoria"] == "nome?" else ""
        linhas.append(f"  {NOME_CAT.get(a['categoria'], a['categoria'])} \"{a['texto']}\": {quando}, "
                      f"caixa x {x} y {y} largura {w} altura {h}{extra}")
        if norm(a["linha"]) != norm(a["texto"]):
            linhas.append(f"      linha lida: \"{a['linha'][:90]}\"")
    if ach and rep["amostragem"] != "todos os quadros":
        linhas.append("  Achou com amostra: rode com --todos em volta de cada achado (--de/--ate) para achar o "
                      "primeiro e o último quadro antes de alargar o borrão.")
    linhas.append("  Nada achado não prova que não tem dado: o leitor erra letra pequena, borrada, girada ou em "
                  "movimento. Olhe também o primeiro e o último quadro e cada troca de tela.")
    return linhas


def main(argv: list[str] | None = None) -> int:
    tnt.utf8_console()
    ap = argparse.ArgumentParser(description="Procura dado pessoal (nomes, e-mail, CPF, telefone e termos dados) "
                                             "em todos os quadros de um vídeo exportado.")
    ap.add_argument("video", type=Path)
    ap.add_argument("--termos", default="", help="nomes e palavras que não podem aparecer, separados por vírgula")
    ap.add_argument("--termos-arquivo", type=Path, help="arquivo com um termo por linha")
    ap.add_argument("--ignorar", default="", help="termos que podem aparecer (o nome da própria pessoa, a marca)")
    ap.add_argument("--fps", type=float, default=None,
                    help=f"lê só N quadros por segundo (padrão: todos até {TODOS_ATE_S / 60:.0f} min de vídeo, "
                         f"{FPS_PADRAO:g} por segundo nos mais longos)")
    ap.add_argument("--todos", action="store_true", help="lê todos os quadros mesmo num vídeo longo")
    ap.add_argument("--de", type=float, default=0.0, help="começa neste segundo")
    ap.add_argument("--ate", type=float, help="termina neste segundo")
    ap.add_argument("--sem-padroes", action="store_true", help="só os termos dados (sem e-mail, CPF, telefone, nomes...)")
    ap.add_argument("--json", type=Path, help="grava o relatório completo neste arquivo")
    a = ap.parse_args(argv)
    video = a.video.expanduser().absolute()
    if not video.is_file():
        print(f"ERRO: não achei {video}", file=sys.stderr)
        return 2
    termos = termos_da_lista(a.termos)
    if a.termos_arquivo:
        if not a.termos_arquivo.is_file():
            print(f"ERRO: não achei {a.termos_arquivo}", file=sys.stderr)
            return 2
        termos += termos_da_lista(a.termos_arquivo.read_text(encoding="utf-8"))
    if a.sem_padroes and not termos:
        print("ERRO: com --sem-padroes, passe os --termos a procurar", file=sys.stderr)
        return 2
    motor = leitor()
    if motor is None:
        print("ERRO: sem leitor de texto neste computador. No Mac, instale as ferramentas da Apple "
              "(xcode-select --install, traz o swiftc) e rode de novo. No Windows ou Linux, instale o Tesseract com o "
              "idioma português (tesseract no PATH). Sem leitor, confira olhando: tira de quadros a 4 por segundo com "
              "helpers/tira_de_quadros.py e cada troca de tela quadro a quadro.", file=sys.stderr)
        return 2
    fps = a.fps
    if fps is None and not a.todos:
        trecho = (a.ate if a.ate is not None else float(probe_video(video)["duration"])) - a.de
        if trecho > TODOS_ATE_S:
            fps = FPS_PADRAO
            print(f"vídeo de {trecho / 60:.1f} min: lendo {FPS_PADRAO:g} quadros por segundo (--todos lê todos)")
    try:
        rep = conferir(video, termos, fps=fps, de=a.de, ate=a.ate, padroes=not a.sem_padroes,
                       ignorar=termos_da_lista(a.ignorar), motor=motor, progresso=lambda m: print(m, flush=True))
    except (subprocess.CalledProcessError, RuntimeError, KeyError, ValueError) as exc:
        print(f"ERRO: não consegui ler {video.name}: {exc}", file=sys.stderr)
        return 2
    print("\n".join(relatorio(rep)))
    if a.json:
        a.json.write_text(json.dumps(rep, ensure_ascii=False, indent=1), encoding="utf-8")
        print(f"relatório: {a.json}")
    return 1 if rep["achados"] else 0


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