arara_oarmp/pre_process/preprocess_pdf.py

import csv
import json
import re
from pathlib import Path

import pdfplumber


ROOT = Path(__file__).resolve().parents[1]
PDF_PATH = ROOT / "raw" / "documento joaomarcos.pdf"
TXT_PATH = ROOT / "pre_process" / "documento_joaomarcos_texto.txt"
JSON_PATH = ROOT / "pre_process" / "documento_joaomarcos_inspecoes.json"
CSV_PATH = ROOT / "pre_process" / "documento_joaomarcos_inspecoes.csv"

ROW_START_RE = re.compile(r"^(?P<seq>\d+)\s+")
INTERVAL_RE = re.compile(
    r"(?P<intervalo>\d+(?::\d+)?\s*(?:HORAS DE V[ÔO]O|MESES CONT[ÍI]NUOS|POUSOS))",
    re.IGNORECASE,
)


def extract_text(pdf_path):
    pages = []
    with pdfplumber.open(pdf_path) as pdf:
        for page in pdf.pages:
            pages.append(page.extract_text(x_tolerance=1, y_tolerance=3) or "")
    return "\n\n".join(pages).strip()


def normalize_interval(value):
    return re.sub(r"\s+", " ", value.replace("VÔO", "VOO")).strip()


def split_records(text):
    records = []
    current = []

    for raw_line in text.splitlines():
        line = re.sub(r"\s+", " ", raw_line).strip()
        if not line:
            continue
        if ROW_START_RE.match(line) and " Término da Anterior " in line:
            if current:
                records.append(" ".join(current))
            current = [line]
        elif current:
            current.append(line)

    if current:
        records.append(" ".join(current))

    return records


def parse_header(text):
    header = {}

    patterns = {
        "data_relatorio": r"Data:\s*(\d{2}/\d{2}/\d{4})",
        "hora_relatorio": r"Hora:\s*(\d{2}:\d{2}:\d{2})",
        "pn": r"PN:\s*(.*?)\s+CFF:",
        "cff": r"CFF:\s*(.*?)\s+Nomenclatura:",
        "nomenclatura": r"Nomenclatura:\s*(.*?)\s+SN:",
        "sn": r"SN:\s*(.*?)\s+Matrícula:",
        "matricula": r"Matrícula:\s*(.*?)\s+Modelo:",
        "modelo": r"Modelo:\s*(.*?)\s+Ciclo Atual:",
        "ciclo_atual": r"Ciclo Atual:\s*(\S+)",
    }

    compact_text = re.sub(r"\s+", " ", text)
    for key, pattern in patterns.items():
        match = re.search(pattern, compact_text)
        if match:
            header[key] = match.group(1).strip()

    return header


def parse_record(record):
    record = record.replace("- Tipo D)", " - Tipo D)")
    has_tipo_d = " - Tipo D)" in record
    left, right = record.split(" Término da Anterior ", 1)

    seq_match = ROW_START_RE.match(left)
    seq = int(seq_match.group("seq"))
    left_body = left[seq_match.end() :]

    intervalos = [normalize_interval(m.group("intervalo")) for m in INTERVAL_RE.finditer(right)]
    right_before_interval = INTERVAL_RE.split(right, maxsplit=1)[0].strip()
    tokens = right_before_interval.split()

    zera = tokens[0] if tokens else ""
    controle = " ".join(tokens[1:])

    especial_marker = " Especial (a contar dela mesma"
    if especial_marker in left_body:
        before_ref, referencia = left_body.split(especial_marker, 1)
        referencia = "Especial (a contar dela mesma"
        if has_tipo_d:
            referencia += " - Tipo D)"
    else:
        ref_match = re.search(r"(Última .*)$", left_body)
        referencia = ref_match.group(1).strip() if ref_match else ""
        before_ref = left_body[: ref_match.start()].strip() if ref_match else left_body

    upper_before_ref = before_ref.upper()
    desc_idx = upper_before_ref.find("INSPEÇÃO")
    if desc_idx == -1:
        desc_idx = upper_before_ref.find("INPEÇÃO")
    if desc_idx == -1:
        desc_idx = upper_before_ref.find("CHECK OPERACIONAL")

    if desc_idx != -1:
        sigla = before_ref[:desc_idx].strip()
        descricao = before_ref[desc_idx:].strip()
    else:
        parts = before_ref.split(maxsplit=1)
        sigla = parts[0] if parts else ""
        descricao = parts[1] if len(parts) > 1 else ""

    intervalo_horas_voo = next((item for item in intervalos if "HORAS DE VOO" in item), "")
    intervalo_meses_continuos = next((item for item in intervalos if "MESES CONTÍNUOS" in item), "")
    intervalo_pousos = next((item for item in intervalos if "POUSOS" in item), "")

    return {
        "seq": seq,
        "sigla_mnt": sigla,
        "descricao_mnt": descricao,
        "referencia": referencia,
        "tipo_vencimento": "Término da Anterior",
        "zera": zera,
        "controle_original": controle,
        "intervalo_horas_voo": intervalo_horas_voo,
        "intervalo_meses_continuos": intervalo_meses_continuos,
        "intervalo_pousos": intervalo_pousos,
        "intervalos": intervalos,
        "linha_original": record,
    }


def main():
    text = extract_text(PDF_PATH)
    TXT_PATH.write_text(text + "\n", encoding="utf-8")

    payload = {
        "fonte": str(PDF_PATH.relative_to(ROOT)),
        "metadados": parse_header(text),
        "inspecoes": [parse_record(record) for record in split_records(text)],
    }

    JSON_PATH.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")

    fields = [
        "seq",
        "sigla_mnt",
        "descricao_mnt",
        "referencia",
        "tipo_vencimento",
        "zera",
        "controle_original",
        "intervalo_horas_voo",
        "intervalo_meses_continuos",
        "intervalo_pousos",
        "intervalos",
    ]
    with CSV_PATH.open("w", newline="", encoding="utf-8-sig") as csv_file:
        writer = csv.DictWriter(csv_file, fieldnames=fields, delimiter=";")
        writer.writeheader()
        for row in payload["inspecoes"]:
            csv_row = {key: row[key] for key in fields}
            csv_row["intervalos"] = " | ".join(row["intervalos"])
            writer.writerow(csv_row)

    print(f"Gerado: {TXT_PATH.relative_to(ROOT)}")
    print(f"Gerado: {JSON_PATH.relative_to(ROOT)}")
    print(f"Gerado: {CSV_PATH.relative_to(ROOT)}")
    print(f"Inspeções extraídas: {len(payload['inspecoes'])}")


if __name__ == "__main__":
    main()