267 lines
8.5 KiB
Python
267 lines
8.5 KiB
Python
import csv
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import pdfplumber
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
BASE_NAME = "relatorio_ciclo_inspecoes_c105_2805_2026-06-15"
|
|
PDF_PATH = ROOT / "raw" / f"{BASE_NAME}.pdf"
|
|
TXT_PATH = ROOT / "pre_process" / f"{BASE_NAME}_texto.txt"
|
|
JSON_PATH = ROOT / "pre_process" / f"{BASE_NAME}_inspecoes.json"
|
|
CSV_PATH = ROOT / "pre_process" / f"{BASE_NAME}_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 interval_number(intervalo):
|
|
if not intervalo:
|
|
return ""
|
|
first_token = intervalo.split()[0]
|
|
if ":" in first_token:
|
|
return int(first_token.split(":", 1)[0])
|
|
if first_token.isdigit():
|
|
return int(first_token)
|
|
return ""
|
|
|
|
|
|
def parse_control_fields(tokens):
|
|
nivel_descricao = {
|
|
"B": "Base",
|
|
"P": "Parque",
|
|
"O": "Orgânico",
|
|
}
|
|
fields = {
|
|
"zera_tso": "",
|
|
"letra": "",
|
|
"nivel": "",
|
|
"nivel_descricao": "",
|
|
"var_media": "",
|
|
"duracao": "",
|
|
"duracao_valor": "",
|
|
"duracao_unidade": "",
|
|
}
|
|
|
|
if tokens:
|
|
fields["zera_tso"] = tokens[0]
|
|
tokens = tokens[1:]
|
|
|
|
if len(tokens) >= 2 and tokens[-1] in {"D", "H"} and tokens[-2].isdigit():
|
|
duracao_valor = tokens[-2]
|
|
duracao_codigo = tokens[-1]
|
|
tokens = tokens[:-2]
|
|
fields["duracao"] = f"{duracao_valor} {duracao_codigo}"
|
|
fields["duracao_valor"] = int(duracao_valor)
|
|
fields["duracao_unidade"] = "dias" if duracao_codigo == "D" else "horas"
|
|
|
|
if len(tokens) == 1:
|
|
if tokens[0] in nivel_descricao:
|
|
fields["nivel"] = tokens[0]
|
|
else:
|
|
fields["letra"] = tokens[0]
|
|
elif len(tokens) >= 2:
|
|
if tokens[1] in nivel_descricao:
|
|
fields["letra"] = tokens[0]
|
|
fields["nivel"] = tokens[1]
|
|
fields["var_media"] = " ".join(tokens[2:])
|
|
elif tokens[0] in nivel_descricao:
|
|
fields["nivel"] = tokens[0]
|
|
fields["var_media"] = " ".join(tokens[1:])
|
|
else:
|
|
fields["letra"] = tokens[0]
|
|
fields["nivel"] = tokens[1]
|
|
fields["var_media"] = " ".join(tokens[2:])
|
|
|
|
fields["nivel_descricao"] = nivel_descricao.get(fields["nivel"], "")
|
|
|
|
return fields
|
|
|
|
|
|
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()
|
|
|
|
controle_tokens = tokens
|
|
controle = " ".join(controle_tokens)
|
|
control_fields = parse_control_fields(controle_tokens)
|
|
|
|
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_tso": control_fields["zera_tso"],
|
|
"letra": control_fields["letra"],
|
|
"nivel": control_fields["nivel"],
|
|
"nivel_descricao": control_fields["nivel_descricao"],
|
|
"var_media": control_fields["var_media"],
|
|
"duracao": control_fields["duracao"],
|
|
"duracao_valor": control_fields["duracao_valor"],
|
|
"duracao_unidade": control_fields["duracao_unidade"],
|
|
"controle_original": controle,
|
|
"intervalo_horas_voo": intervalo_horas_voo,
|
|
"intervalo_horas_voo_valor": interval_number(intervalo_horas_voo),
|
|
"intervalo_meses_continuos": intervalo_meses_continuos,
|
|
"intervalo_meses_continuos_valor": interval_number(intervalo_meses_continuos),
|
|
"intervalo_pousos": intervalo_pousos,
|
|
"intervalo_pousos_valor": interval_number(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_tso",
|
|
"letra",
|
|
"nivel",
|
|
"nivel_descricao",
|
|
"var_media",
|
|
"duracao",
|
|
"duracao_valor",
|
|
"duracao_unidade",
|
|
"controle_original",
|
|
"intervalo_horas_voo",
|
|
"intervalo_horas_voo_valor",
|
|
"intervalo_meses_continuos",
|
|
"intervalo_meses_continuos_valor",
|
|
"intervalo_pousos",
|
|
"intervalo_pousos_valor",
|
|
"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()
|