#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ SINCRONIZAR LAS LICITACIONES DE SONDEVA CON SU BASE DE DATOS -- ejemplo, version 1 --. export SONDEVA_LLAVE=sdl_... # su llave; nunca en la linea de ordenes python3 sincronizar.py --sqlite licitaciones.db # en un fichero SQLite (sin instalar nada) python3 sincronizar.py --postgres "postgresql://usuario@servidor/base" # (pip install "psycopg[binary]") LO QUE HACE, y puede lanzarse cada hora o cada noche: retoma solo donde lo dejo. 1. LA PRIMERA VEZ, la foto completa por ficheros (`/v1/ficheros`): cada parte se baja y se comprueba con su huella SHA-256 antes de cargarla, y despues los ficheros de cambios que van detras. Con una llave de prueba (sin ficheros) empieza por la API desde `--desde` (por omision, hace 7 dias). 2. DESPUES, los cambios por la API (`/v1/licitaciones`), pagina a pagina con el cursor. 3. CADA LICITACION SE GUARDA POR SU `id`. Si ya estaba, gana la de `cambiada_en` mas reciente: da igual el orden en que lleguen o que alguna llegue dos veces. 4. Donde se quedo (el `cambiada_en` mas reciente visto) se guarda en la tabla `sondeva_estado`. Respeta los limites: ante un 429 espera lo que diga `Retry-After` y sigue; si se acaba el cupo del dia, para y guarda (la vez siguiente sigue). Ante un 503, reintenta unas veces. Solo usa la biblioteca estandar de Python 3.9+ (y `psycopg` para Postgres). Documentacion de la API: https://sondeva.es/api-de-licitaciones/ · descripcion OpenAPI: https://api.sondeva.es/v1/openapi.json """ import argparse import datetime as dt import gzip import hashlib import json import os import sqlite3 import sys import time import urllib.error import urllib.parse import urllib.request API = "https://api.sondeva.es" POR_PAGINA = 500 #: Los ficheros de cambios se solapan 6 horas; la API se retoma con el mismo margen. SOLAPE = dt.timedelta(hours=6) REINTENTOS = 5 #: Una espera de mas de esto (el cupo del dia) no se hace: se para y se sigue la proxima vez. ESPERA_MAXIMA = 120 class ErrorDeLaApi(RuntimeError): def __init__(self, estado, mensaje): super().__init__("%s: %s" % (estado, mensaje)) self.estado = estado class CupoGastado(RuntimeError): pass def orden(cambiada_en): """`cambiada_en` en una forma que ordena bien como texto: UTC, con microsegundos y `Z`.""" if not cambiada_en: return "" t = dt.datetime.fromisoformat(str(cambiada_en).replace("Z", "+00:00")) if t.tzinfo is None: t = t.replace(tzinfo=dt.timezone.utc) return t.astimezone(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ") # ── LA API ─────────────────────────────────────────────────────────────────── class Api: def __init__(self, base, llave, dormir=time.sleep, abrir=urllib.request.urlopen): self.base, self.llave, self.dormir, self.abrir = base.rstrip("/"), llave, dormir, abrir def get(self, ruta, parametros=None): url = self.base + ruta + ("?" + urllib.parse.urlencode(parametros) if parametros else "") for intento in range(REINTENTOS): pedido = urllib.request.Request(url, headers={"Authorization": "Bearer " + self.llave, "Accept-Encoding": "gzip", "User-Agent": "sondeva-sincronizar/1"}) try: with self.abrir(pedido, timeout=120) as r: crudo = r.read() if r.headers.get("Content-Encoding") == "gzip": crudo = gzip.decompress(crudo) return json.loads(crudo) except urllib.error.HTTPError as e: cuerpo = e.read() if e.headers.get("Content-Encoding") == "gzip": cuerpo = gzip.decompress(cuerpo) try: mensaje = json.loads(cuerpo).get("error", "") except ValueError: mensaje = cuerpo[:200].decode("utf-8", "replace") espera = int(e.headers.get("Retry-After") or 30) if e.code in (429, 503) and intento < REINTENTOS - 1: if espera > ESPERA_MAXIMA: raise CupoGastado(mensaje) self.dormir(espera) continue raise ErrorDeLaApi(e.code, mensaje) except urllib.error.URLError: if intento == REINTENTOS - 1: raise self.dormir(5 * (intento + 1)) raise ErrorDeLaApi(503, "sin respuesta") def bajar(url, abrir=urllib.request.urlopen): with abrir(urllib.request.Request(url, headers={"User-Agent": "sondeva-sincronizar/1"}), timeout=600) as r: return r.read() # ── DONDE SE GUARDA ────────────────────────────────────────────────────────── class EnSqlite: def __init__(self, ruta): self.c = sqlite3.connect(ruta) self.c.execute("create table if not exists licitaciones (id text primary key, cambiada_en text, " "registro text not null)") self.c.execute("create table if not exists sondeva_estado (clave text primary key, valor text)") self.c.commit() def guardar(self, registros): filas = [(r["id"], orden(r.get("cambiada_en")), json.dumps(r, ensure_ascii=False)) for r in registros if r.get("id")] self.c.executemany( "insert into licitaciones (id, cambiada_en, registro) values (?, ?, ?) " "on conflict (id) do update set cambiada_en = excluded.cambiada_en, registro = excluded.registro " "where excluded.cambiada_en >= licitaciones.cambiada_en", filas) self.c.commit() return len(filas) def leer(self, clave): fila = self.c.execute("select valor from sondeva_estado where clave = ?", (clave,)).fetchone() return fila[0] if fila else None def escribir(self, clave, valor): self.c.execute("insert into sondeva_estado (clave, valor) values (?, ?) " "on conflict (clave) do update set valor = excluded.valor", (clave, valor)) self.c.commit() def cuantas(self): return self.c.execute("select count(*) from licitaciones").fetchone()[0] class EnPostgres: #: El mismo esquema, con `jsonb` para consultar dentro del registro. CREAR = ("create table if not exists licitaciones (id text primary key, cambiada_en timestamptz, " "registro jsonb not null)", "create table if not exists sondeva_estado (clave text primary key, valor text)") GUARDAR = ("insert into licitaciones (id, cambiada_en, registro) values (%s, %s::timestamptz, %s::jsonb) " "on conflict (id) do update set cambiada_en = excluded.cambiada_en, registro = excluded.registro " "where excluded.cambiada_en >= licitaciones.cambiada_en") def __init__(self, dsn, conectar=None): if conectar is None: import psycopg # pip install "psycopg[binary]" conectar = psycopg.connect self.c = conectar(dsn) with self.c.cursor() as cur: for sql in self.CREAR: cur.execute(sql) self.c.commit() def guardar(self, registros): filas = [(r["id"], r.get("cambiada_en"), json.dumps(r, ensure_ascii=False)) for r in registros if r.get("id")] with self.c.cursor() as cur: cur.executemany(self.GUARDAR, filas) self.c.commit() return len(filas) def leer(self, clave): with self.c.cursor() as cur: cur.execute("select valor from sondeva_estado where clave = %s", (clave,)) fila = cur.fetchone() return fila[0] if fila else None def escribir(self, clave, valor): with self.c.cursor() as cur: cur.execute("insert into sondeva_estado (clave, valor) values (%s, %s) " "on conflict (clave) do update set valor = excluded.valor", (clave, valor)) self.c.commit() def cuantas(self): with self.c.cursor() as cur: cur.execute("select count(*) from licitaciones") return cur.fetchone()[0] # ── LA SINCRONIZACION ──────────────────────────────────────────────────────── def cargar_fichero(manifiesto, destino, bajar_=bajar, decir=print): """Las partes de un manifiesto, cada una comprobada con su huella antes de cargarla.""" n = 0 for parte in manifiesto["partes"]: crudo = bajar_(parte["url"]) if hashlib.sha256(crudo).hexdigest() != parte["sha256"]: raise RuntimeError("la parte %s no coincide con su huella: no se carga" % parte["nombre"]) lineas = gzip.decompress(crudo).decode("utf-8").splitlines() n += destino.guardar([json.loads(x) for x in lineas if x.strip()]) decir(" %s: %d" % (parte["nombre"], len(lineas))) return n def sincronizar(api, destino, desde_si_no_hay_ficheros, bajar_=bajar, decir=print): """`{"de_ficheros", "de_la_api", "parado_por_cupo"}`.""" hecho = {"de_ficheros": 0, "de_la_api": 0, "parado_por_cupo": False} desde = destino.leer("desde") if desde is None: try: f = api.get("/v1/ficheros") except ErrorDeLaApi as e: if e.estado != 403: raise f = None # una llave de prueba: sin ficheros if f: decir("Primera carga: la foto completa (%d filas)." % f["completo"]["filas"]) hecho["de_ficheros"] += cargar_fichero(f["completo"], destino, bajar_, decir) ultimo = f["completo"]["hasta"] for c in f.get("cambios") or []: hecho["de_ficheros"] += cargar_fichero(c, destino, bajar_, decir) ultimo = c["hasta"] t = dt.datetime.fromisoformat(ultimo.replace("Z", "+00:00")) - SOLAPE desde = t.astimezone(dt.timezone.utc).isoformat() else: desde = desde_si_no_hay_ficheros destino.escribir("desde", desde) parametros = {"desde": desde, "limite": POR_PAGINA} mas_reciente = orden(desde) if "T" in desde else "" try: while True: d = api.get("/v1/licitaciones", parametros) hecho["de_la_api"] += destino.guardar(d["datos"]) for r in d["datos"]: o = orden(r.get("cambiada_en")) if o > mas_reciente: mas_reciente, desde = o, r["cambiada_en"] #: SE GUARDA DESPUES DE CADA PAGINA: si se corta, la vez siguiente sigue desde aqui. destino.escribir("desde", desde) if not d.get("siguiente"): break parametros = {"cursor": d["siguiente"], "limite": POR_PAGINA} except CupoGastado as e: hecho["parado_por_cupo"] = True decir("Cupo de hoy gastado (%s). Se sigue la proxima vez." % e) return hecho def main(argv=None): p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) donde = p.add_mutually_exclusive_group(required=True) donde.add_argument("--sqlite", metavar="FICHERO") donde.add_argument("--postgres", metavar="DSN") p.add_argument("--api", default=API) p.add_argument("--llave-en", metavar="FICHERO", help="si no, la variable SONDEVA_LLAVE") p.add_argument("--desde", default=(dt.date.today() - dt.timedelta(days=7)).isoformat(), help="sin ficheros (llave de prueba): desde que dia empezar") a = p.parse_args(argv) if a.llave_en: with open(a.llave_en, encoding="utf-8") as f: llave = f.read().strip() else: llave = os.environ.get("SONDEVA_LLAVE", "").strip() if not llave.startswith("sdl_"): print("Falta la llave: SONDEVA_LLAVE=sdl_... o --llave-en FICHERO.", file=sys.stderr) return 2 destino = EnSqlite(a.sqlite) if a.sqlite else EnPostgres(a.postgres) try: hecho = sincronizar(Api(a.api, llave), destino, a.desde) except ErrorDeLaApi as e: print("La API dice %s" % e, file=sys.stderr) return 1 print("Hecho: %d de ficheros, %d de la API; %d licitaciones en su base." % ( hecho["de_ficheros"], hecho["de_la_api"], destino.cuantas())) return 0 if __name__ == "__main__": sys.exit(main())