download_lattes_xml.py

#!/usr/bin/env python3
"""Baixa XMLs do CV Lattes a partir de uma lista de CPFs ou IDs Lattes.

Uso:
    python download_lattes_xml.py entrada.txt
    python download_lattes_xml.py entrada.txt --output-dir xmls

Dependencia:
    pip install zeep
"""

from __future__ import annotations

import argparse
import io
import re
import sys
import time
import zipfile
from pathlib import Path

import zeep


WSDL_URL = "https://cnpqwsproxy.ufscar.br:7443/srvcurriculo/WSCurriculo?wsdl"


def only_digits(value: str) -> str:
    return re.sub(r"\D", "", value)


def call_with_retry(func, *args, tries: int = 3, delay: float = 1.0, **kwargs):
    last_error = None
    for attempt in range(tries):
        try:
            return func(*args, **kwargs)
        except Exception as exc:  # noqa: BLE001 - script CLI deve reportar e seguir.
            last_error = exc
            if attempt + 1 < tries:
                time.sleep(delay)
    raise last_error


def id_lattes_from_cpf(client: zeep.Client, cpf: str) -> str | None:
    result = call_with_retry(
        client.service.getIdentificadorCNPq,
        cpf=cpf,
        nomeCompleto="",
        dataNascimento="",
    )
    if result is None:
        return None
    result = only_digits(str(result))
    return result or None


def xml_from_id_lattes(client: zeep.Client, id_lattes: str) -> bytes | None:
    zipped = call_with_retry(client.service.getCurriculoCompactado, id=id_lattes)
    if zipped is None:
        return None


    with zipfile.ZipFile(io.BytesIO(zipped)) as archive:
        names = archive.namelist()
        if not names:
            return None
        return archive.read(names[0])


def iter_entries(path: Path):
    with path.open(encoding="utf-8") as file:
        for line_number, line in enumerate(file, start=1):
            raw = line.strip()
            if not raw or raw.startswith("#"):
                continue
            yield line_number, raw, only_digits(raw)


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Baixa XMLs do CV Lattes a partir de CPFs ou IDs Lattes."
    )
    parser.add_argument("input_file", type=Path, help="arquivo com um CPF ou ID Lattes por linha")
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path("."),
        help="diretorio onde salvar {id_lattes}.xml (padrao: diretorio atual)",
    )
    parser.add_argument("--wsdl", default=WSDL_URL, help=argparse.SUPPRESS)
    args = parser.parse_args()

    args.output_dir.mkdir(parents=True, exist_ok=True)
    client = zeep.Client(args.wsdl)

    failures = 0
    for line_number, raw, digits in iter_entries(args.input_file):
        try:
            if len(digits) == 11:
                id_lattes = id_lattes_from_cpf(client, digits)
                if not id_lattes:
                    raise RuntimeError("CPF nao retornou ID Lattes")
            elif len(digits) == 16:
                id_lattes = digits
            else:
                raise ValueError("linha nao parece CPF (11 digitos) nem ID Lattes (16 digitos)")

            xml = xml_from_id_lattes(client, id_lattes)
            if xml is None:
                raise RuntimeError("servico nao retornou XML")

            output_path = args.output_dir / f"{id_lattes}.xml"
            output_path.write_bytes(xml)
            print(f"OK linha {line_number}: {raw} -> {output_path}")
        except Exception as exc:  # noqa: BLE001 - continua processando as demais linhas.
            failures += 1
            print(f"ERRO linha {line_number}: {raw}: {exc}", file=sys.stderr)

    return 1 if failures else 0


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