brasiliano/pipeline/baixador.py

#!/usr/bin/env python3
"""
Baixa os arquivos da Brasiliana e deposita no S3, um por vez, sem acumular.

Roda no servidor (Finlandia): mede 4,7 MB/s no bitstream e o bucket e Hetzner
tambem, entao a transferencia nao sai do datacenter. NUNCA guarda mais de um
arquivo em disco: baixa -> sha256 -> sobe -> apaga. Pico de uso = 1 arquivo.

Os bitstreams do DSpace passam pelo Cloudflare com curl simples (as paginas
HTML e que sao desafiadas -- essas vem do coletor.js, no navegador).
A copia legada do objdigital NAO e alcancavel da Finlandia: rode --somente-legado
na maquina local para essas.

    python3 baixador.py --catalogo catalogo.jsonl
    python3 baixador.py --catalogo catalogo.jsonl --somente-legado   # na sua maquina
    python3 baixador.py --status
"""
import argparse, hashlib, json, os, shutil, sqlite3, subprocess, sys, time
from pathlib import Path
from urllib.parse import unquote, urlparse

UA = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
      "(KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36")
CONTACT = os.environ.get("ARCHIVE_CONTACT", "")
S3_ENDPOINT = os.environ.get("S3_ENDPOINT", "https://hel1.your-objectstorage.com")
BUCKET = os.environ.get("ARCHIVE_BUCKET", "brasiliana")

ROOT = Path(__file__).resolve().parent
DB = ROOT / "baixador.sqlite"
TMP = ROOT / ".tmp"
PAUSA = 2.0
MIN_LIVRE = 2 * 1024**3          # 2 GB de folga; abaixo disso, para


def db():
    con = sqlite3.connect(DB)
    con.execute("PRAGMA journal_mode=WAL")
    con.executescript("""
    CREATE TABLE IF NOT EXISTS arquivos (
      handle TEXT, tipo TEXT, url TEXT, nome TEXT,
      bytes INTEGER, sha256 TEXT, status TEXT, erro TEXT,
      PRIMARY KEY (handle, tipo));
    """)
    return con


def livre():
    return shutil.disk_usage(ROOT).free


def baixar(url, destino):
    """curl -L para arquivo. Retorna (http, bytes) sem manter nada em memoria."""
    r = subprocess.run(
        ["curl", "-sL", "-o", str(destino), "-w", "%{http_code} %{size_download}",
         "-A", UA, *(["-H", f"From: {CONTACT}"] if CONTACT else []),
         "--max-time", "600", "--retry", "2", "--retry-delay", "5", url],
        capture_output=True, text=True)
    try:
        http, size = r.stdout.strip().split()
        return int(http), int(size)
    except Exception:
        return 0, 0


def s3_put(local, key):
    return subprocess.run(
        ["aws", "--endpoint-url", S3_ENDPOINT, "s3", "cp", str(local),
         f"s3://{BUCKET}/{key}", "--only-show-errors"],
        capture_output=True, text=True).returncode == 0


def nome_de(url):
    n = unquote(urlparse(url).path.rsplit("/", 1)[-1]) or "arquivo.bin"
    return n.replace("/", "_")[:150]


def processa(con, rec, tipo, url, args):
    handle = rec["handle"]
    ja = con.execute("SELECT status FROM arquivos WHERE handle=? AND tipo=?",
                     (handle, tipo)).fetchone()
    if ja and ja[0] == "ok":
        return "pulado"
    if livre() < MIN_LIVRE:
        sys.exit(f"\nESPACO BAIXO ({livre()/1024**3:.1f} GB livres). Parando por seguranca.")

    TMP.mkdir(exist_ok=True)
    nome = nome_de(url)
    tmp = TMP / nome
    http, size = baixar(url, tmp)
    if http != 200 or size <= 0:
        tmp.unlink(missing_ok=True)
        con.execute("INSERT OR REPLACE INTO arquivos VALUES (?,?,?,?,NULL,NULL,'erro',?)",
                    (handle, tipo, url, nome, f"HTTP {http}"))
        con.commit()
        return f"erro HTTP {http}"

    sha = hashlib.sha256()
    with tmp.open("rb") as fh:
        for bloco in iter(lambda: fh.read(1 << 20), b""):
            sha.update(bloco)
    dig = sha.hexdigest()

    pasta = handle.replace("/", "_")
    sub = "preservacao" if tipo == "preservacao" else "legado"
    ok = s3_put(tmp, f"itens/{pasta}/{sub}/{nome}")
    tmp.unlink(missing_ok=True)          # apaga SEMPRE, deu certo ou nao

    if not ok:
        con.execute("INSERT OR REPLACE INTO arquivos VALUES (?,?,?,?,?,?,'erro','falha no upload')",
                    (handle, tipo, url, nome, size, dig))
        con.commit()
        return "erro no S3"

    con.execute("INSERT OR REPLACE INTO arquivos VALUES (?,?,?,?,?,?,'ok',NULL)",
                (handle, tipo, url, nome, size, dig))
    con.commit()
    return f"{size/1048576:.1f} MB"


def sobe_metadados(con, rec):
    """metadata.json + manifest-sha256.txt do item, a partir do catalogo.

    Nome de arquivo temporario inclui o PID: baixador.py e subidor.py
    moram na mesma pasta (mesmo TMP) e podem rodar ao mesmo tempo -- um
    nome fixo faz um apagar o arquivo que o outro acabou de criar
    (achado 2026-09-10, derrubou o baixador.py com FileNotFoundError)."""
    handle = rec["handle"]; pasta = handle.replace("/", "_")
    TMP.mkdir(exist_ok=True)
    mj = TMP / f"metadata.{os.getpid()}.json"
    mj.write_text(json.dumps(rec, ensure_ascii=False, indent=2), encoding="utf-8")
    s3_put(mj, f"itens/{pasta}/metadata.json")
    linhas = [f"{sha}  {'preservacao' if t=='preservacao' else 'legado'}/{n}"
              for t, n, sha in con.execute(
                  "SELECT tipo,nome,sha256 FROM arquivos WHERE handle=? AND status='ok'",
                  (handle,))]
    linhas.append(f"{hashlib.sha256(mj.read_bytes()).hexdigest()}  metadata.json")
    mf = TMP / f"manifest-sha256.{os.getpid()}.txt"
    mf.write_text("\n".join(linhas) + "\n", encoding="utf-8")
    s3_put(mf, f"itens/{pasta}/manifest-sha256.txt")
    mj.unlink(missing_ok=True); mf.unlink(missing_ok=True)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--catalogo")
    ap.add_argument("--limit", type=int, default=0)
    ap.add_argument("--somente-legado", action="store_true")
    ap.add_argument("--pular-legado", action="store_true")
    ap.add_argument("--status", action="store_true")
    args = ap.parse_args()
    con = db()

    if args.status:
        for tipo, st, n, b in con.execute(
                "SELECT tipo,status,COUNT(*),COALESCE(SUM(bytes),0) FROM arquivos GROUP BY 1,2"):
            print(f"  {tipo:<12} {st:<6} {n:>6}  {b/1024**3:.2f} GB")
        print(f"  disco livre: {livre()/1024**3:.1f} GB")
        return

    if not args.catalogo:
        sys.exit("informe --catalogo catalogo.jsonl")
    if not os.environ.get("AWS_ACCESS_KEY_ID"):
        sys.exit("credenciais S3 ausentes no ambiente")

    recs = [json.loads(l) for l in open(args.catalogo, encoding="utf-8") if l.strip()]
    recs = [r for r in recs if not r.get("erro")]
    if args.limit:
        recs = recs[: args.limit]
    print(f"{len(recs)} itens | destino s3://{BUCKET}/ | livre {livre()/1024**3:.1f} GB\n")

    for i, rec in enumerate(recs, 1):
        partes = []
        if not args.somente_legado and rec.get("preservation_url"):
            partes.append(("preservacao", rec["preservation_url"]))
        if not args.pular_legado and rec.get("legacy_url"):
            # objdigital.bn.br reseta conexao na porta 80 (http simples) --
            # so responde em https. O DC.identifier na fonte traz http://.
            legado_url = rec["legacy_url"].replace("http://", "https://", 1)
            partes.append(("legado", legado_url))
        if not partes:
            continue
        res = []
        for tipo, url in partes:
            res.append(f"{tipo}:{processa(con, rec, tipo, url, args)}")
            time.sleep(PAUSA)
        sobe_metadados(con, rec)
        print(f"[{i}/{len(recs)}] {rec['handle'].split('/')[-1]:<6} {' '.join(res)}", flush=True)

    print("\nconcluido")


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\nInterrompido. Estado salvo -- rode de novo para continuar.")