#!/usr/bin/env python3 """ dogv_anuncios.py ================ Cliente mínimo del Diari Oficial de la Generalitat Valenciana (DOGV). Lista, en castellano, las disposiciones y anuncios publicados un día o en un intervalo de días dentro de una subsección (o sección) del sumario. Subsecciones más consultadas: 28 II.A Ofertas de empleo público, oposiciones y concursos 62 V.B Licitación y adjudicación de contratos 44 III.B Subvenciones y becas La orden «--subsecciones» muestra la lista completa con sus números. Fuente oficial -------------- Generalitat Valenciana, Diari Oficial de la Generalitat Valenciana. Portal: https://dogv.gva.es/ Punto de acceso: https://dogv.gva.es/dogv-portal (API REST del propio portal) La API es pública y no necesita claves, pero no tiene documentación oficial publicada: son las rutas que usa la web del DOGV para mostrar el sumario de cada día. Puede cambiar sin aviso; por eso el script comprueba la forma de las respuestas y, si deja de ser la esperada, da un error en lugar de devolver una lista vacía que parecería un día sin anuncios. Cómo funciona ------------- 1. /dogv/calendar?startDate=...&endDate=... días con diario publicado. 2. /dogv?date=AAAA-MM-DD&lang=es sumario ordinario de cada día. 3. /dogv/{número}/versionesSumario y /dogv/{número}?ordenSumario=N sumarios adicionales («bis») de los días en que el calendario los marca. Siempre se pide lang=es: sin ese parámetro la API contesta con el error 440. Las peticiones van de una en una, con pausa entre ellas, y el intervalo máximo es de 31 días. Uso --- python dogv_anuncios.py # último DOGV, subsección 28 python dogv_anuncios.py --fecha 2026-10-30 python dogv_anuncios.py --fecha 30/10/2026 --subseccion 62 python dogv_anuncios.py --desde 2026-10-26 --hasta 2026-10-30 --buscar "auxiliar administrativo" python dogv_anuncios.py --fecha 2026-10-29 --seccion 12 # I. Disposiciones generales python dogv_anuncios.py --subsecciones python dogv_anuncios.py --fecha 2026-10-30 --json Requisitos: Python 3.10 o superior. Solo biblioteca estándar, sin claves. Códigos de salida: 0 consulta correcta (también si no hay anuncios); 1 error de red, del servicio o respuesta con un formato inesperado; 2 argumentos no válidos. Aviso: la referencia oficial es el DOGV publicado; compruebe siempre el texto en la disposición enlazada. Licencia -------- SPDX-License-Identifier: MIT Copyright (c) 2026 LloretIA Por la presente se concede permiso, libre de cargos, a cualquier persona que obtenga una copia de este software y de los archivos de documentación asociados (el «Software»), a utilizar el Software sin restricción, incluidos sin limitación los derechos a usar, copiar, modificar, fusionar, publicar, distribuir, sublicenciar y/o vender copias del Software, y a permitir a las personas a las que se les proporcione el Software a hacer lo mismo, con sujeción a las siguientes condiciones: El aviso de copyright anterior y este aviso de permiso se incluirán en todas las copias o partes sustanciales del Software. EL SOFTWARE SE PROPORCIONA «TAL CUAL», SIN GARANTÍA DE NINGÚN TIPO, EXPRESA O IMPLÍCITA, INCLUIDAS, ENTRE OTRAS, LAS GARANTÍAS DE COMERCIABILIDAD, IDONEIDAD PARA UN FIN DETERMINADO Y NO INFRACCIÓN. EN NINGÚN CASO LOS AUTORES O TITULARES DEL COPYRIGHT SERÁN RESPONSABLES DE NINGUNA RECLAMACIÓN, DAÑO U OTRA RESPONSABILIDAD, YA SEA EN UNA ACCIÓN CONTRACTUAL, EXTRACONTRACTUAL O DE OTRO TIPO, DERIVADA DEL SOFTWARE, DE SU USO O DE OTRAS OPERACIONES CON ÉL. Texto de referencia de la licencia (en inglés): https://opensource.org/license/mit """ import argparse import datetime as dt import http.client import json import socket import ssl import sys import time import unicodedata import urllib.error import urllib.parse import urllib.request API = "https://dogv.gva.es/dogv-portal" WEB = "https://dogv.gva.es" IDIOMA = "es" # castellano; la API también admite «va» USER_AGENT = "lloretia-datos-abiertos/1.0 (+https://lloretia.com/codigo)" TIEMPO_ESPERA = 30 # segundos máximos por petición PAUSA_ENTRE_PETICIONES = 1.0 # segundos de cortesía entre peticiones seguidas MAX_DIAS = 31 # intervalo máximo admitido SUBSECCION_POR_DEFECTO = 28 # II.A Ofertas de empleo público, oposiciones y concursos # -------------------------------------------------------------------------- # Ayuda en español. argparse escribe en inglés sus textos internos («usage:», # «options», mensajes de error) a través de gettext; se sustituye su función # de traducción por esta tabla para que toda la ayuda salga en castellano. # -------------------------------------------------------------------------- _TEXTOS_ARGPARSE = { "usage: ": "uso: ", "options": "opciones", "optional arguments": "opciones", "positional arguments": "argumentos", "show this help message and exit": "muestra esta ayuda y termina", "%(prog)s: error: %(message)s\n": "%(prog)s: error: %(message)s\n", "the following arguments are required: %s": "faltan argumentos obligatorios: %s", "unrecognized arguments: %s": "argumentos no reconocidos: %s", "argument %(argument_name)s: %(message)s": "argumento %(argument_name)s: %(message)s", "invalid %(type)s value: %(value)r": "valor no válido: %(value)r", "invalid choice: %(value)r (choose from %(choices)s)": "opción no válida: %(value)r (opciones posibles: %(choices)s)", "expected one argument": "falta el valor del argumento", "ambiguous option: %(option)s could match %(matches)s": "opción ambigua: %(option)s puede ser %(matches)s", "not allowed with argument %s": "no se puede usar junto con %s", } argparse._ = lambda texto: _TEXTOS_ARGPARSE.get(texto, texto) argparse.ngettext = lambda singular, plural, n: _TEXTOS_ARGPARSE.get( singular if n == 1 else plural, singular if n == 1 else plural) # -------------------------------------------------------------------------- # Acceso HTTP # -------------------------------------------------------------------------- class ErrorConsulta(Exception): """Error al consultar la API, con el código de salida que le corresponde.""" def __init__(self, mensaje: str, codigo_salida: int = 1) -> None: super().__init__(mensaje) self.codigo_salida = codigo_salida _ultima_peticion = 0.0 def _motivo_del_servidor(cuerpo: bytes) -> str: """Extrae el texto explicativo de una respuesta de error de la API.""" try: datos = json.loads(cuerpo.decode("utf-8", errors="replace")) except ValueError: return "" if isinstance(datos, dict): return str(datos.get("mensajeTecnico") or datos.get("usuarioMensaje") or "").strip() return "" def describir_fallo_de_red(motivo) -> str: """Explica en castellano por qué no se pudo completar la conexión.""" servidor = urllib.parse.urlsplit(API).hostname if isinstance(motivo, TimeoutError): return f"{servidor} no respondió en {TIEMPO_ESPERA} segundos (tiempo de espera agotado)." if isinstance(motivo, socket.gaierror): return f"no se encontró el servidor {servidor}; compruebe la conexión a internet." if isinstance(motivo, ssl.SSLError): return f"falló la conexión segura (TLS) con {servidor}." if isinstance(motivo, ConnectionError): return f"la conexión con {servidor} se cortó o fue rechazada." return f"no se pudo conectar con {servidor} (detalle técnico: {motivo})." def pedir_json(ruta: str, parametros: dict | None = None): """Hace una petición GET a la API y devuelve el JSON ya interpretado. Entre dos peticiones seguidas se respeta una pausa de cortesía: nunca se lanzan en paralelo. """ global _ultima_peticion espera = PAUSA_ENTRE_PETICIONES - (time.monotonic() - _ultima_peticion) if _ultima_peticion and espera > 0: time.sleep(espera) url = API + ruta if parametros: url += "?" + urllib.parse.urlencode(parametros) peticion = urllib.request.Request( url, headers={"User-Agent": USER_AGENT, "Accept": "application/json"}) try: with urllib.request.urlopen(peticion, timeout=TIEMPO_ESPERA) as respuesta: cuerpo = respuesta.read() except urllib.error.HTTPError as error: try: motivo = _motivo_del_servidor(error.read()) except (OSError, http.client.HTTPException): motivo = "" detalle = f": {motivo}" if motivo else "" consejo = " Inténtelo de nuevo dentro de unos minutos." if error.code in (429, 502, 503, 504) else "" raise ErrorConsulta(f"el DOGV respondió con el error HTTP {error.code}{detalle}.{consejo}") from None except urllib.error.URLError as error: raise ErrorConsulta(describir_fallo_de_red(error.reason)) from None except (OSError, http.client.HTTPException) as error: raise ErrorConsulta(describir_fallo_de_red(error)) from None finally: _ultima_peticion = time.monotonic() try: return json.loads(cuerpo.decode("utf-8")) except (UnicodeDecodeError, ValueError): raise ErrorConsulta("el DOGV devolvió una respuesta que no es JSON válido.") from None def formato_inesperado(que: str) -> ErrorConsulta: return ErrorConsulta(f"la respuesta del DOGV ({que}) no tiene el formato esperado; " "es posible que la API haya cambiado.") # -------------------------------------------------------------------------- # Lectura del DOGV # -------------------------------------------------------------------------- def dias_publicados(desde: dt.date, hasta: dt.date) -> list[tuple[dt.date, bool]]: """Días con DOGV entre dos fechas y si ese día hubo además sumario «bis».""" datos = pedir_json("/dogv/calendar", {"startDate": desde.isoformat(), "endDate": hasta.isoformat()}) if not isinstance(datos, list): raise formato_inesperado("calendario") dias = [] for dia in datos: try: fecha = dt.date.fromisoformat(str(dia["fecha"])[:10]) except (TypeError, KeyError, ValueError): raise formato_inesperado("calendario") from None if desde <= fecha <= hasta: dias.append((fecha, bool(dia.get("esBis")))) return sorted(set(dias)) def comprobar_sumario(sumario) -> bool: """Devuelve True si hay sumario publicado; da error si su forma no es la esperada.""" if not isinstance(sumario, dict): raise formato_inesperado("sumario") if sumario.get("cabecera") is None: # la API responde así a los días sin diario return False disposiciones = sumario.get("disposiciones") if not isinstance(disposiciones, list) or not isinstance(sumario["cabecera"], dict): raise formato_inesperado("sumario") # Trampa conocida: «seccion» y «subseccion» son objetos {"id", "descripcion"}, # no números. Si ninguna disposición trae su sección reconocible, se avisa # en lugar de devolver cero resultados en silencio. if disposiciones and not any(id_de(d.get("seccion")) is not None for d in disposiciones): raise formato_inesperado("sección de las disposiciones") return True def id_de(campo) -> int | None: """Número de una sección o subsección, venga como objeto o como número.""" if isinstance(campo, dict): campo = campo.get("id") try: return int(campo) if campo is not None else None except (TypeError, ValueError): return None def descripcion_de(campo) -> str: return " ".join(str(campo.get("descripcion") or "").split()) if isinstance(campo, dict) else "" def normalizar(texto: str) -> str: """Minúsculas y sin tildes, para filtrar sin distinguirlas.""" texto = unicodedata.normalize("NFKD", texto) return "".join(c for c in texto if not unicodedata.combining(c)).lower() def ficha_disposicion(d: dict, numero: int, fecha: dt.date, bis: bool) -> dict: """Datos de una disposición, con los enlaces a su ficha y a su PDF.""" codigo = str(d.get("codigoInsercion") or "").strip() ruta_pdf = str(d.get("urlPdf") or "").strip() return { "numero_dogv": numero, "fecha": fecha.isoformat(), "sumario_bis": bis, "seccion": descripcion_de(d.get("seccion")), "subseccion": descripcion_de(d.get("subseccion")), "seccion_id": id_de(d.get("seccion")), "subseccion_id": id_de(d.get("subseccion")), "organismo": " ".join(str(d.get("organismo") or "").split()), "titulo": " ".join(str(d.get("titulo") or "").split()), "codigo_insercion": codigo, "url_ficha": f"{WEB}/{IDIOMA}/resultat-dogv?signatura={codigo}" if codigo else "", "url_pdf": f"{WEB}/datos{ruta_pdf}" if ruta_pdf.startswith("/") else "", } def recopilar(desde: dt.date, hasta: dt.date, avisar_progreso: bool) -> tuple[list[dict], list[dict]]: """Lee todos los sumarios del intervalo. Devuelve (diarios, disposiciones).""" diarios, disposiciones = [], [] for fecha, hay_bis in dias_publicados(desde, hasta): if avisar_progreso: print(f"Leyendo el DOGV del {fecha:%d/%m/%Y}...", file=sys.stderr) sumario = pedir_json("/dogv", {"date": fecha.isoformat(), "lang": IDIOMA}) if not comprobar_sumario(sumario): continue numero = sumario["cabecera"].get("numeroDogv") sumarios = [(1, sumario)] if hay_bis and numero is not None: versiones = pedir_json(f"/dogv/{numero}/versionesSumario") if not isinstance(versiones, list): raise formato_inesperado("versiones del sumario") for orden in sorted(v for v in versiones if isinstance(v, int) and v > 1): extra = pedir_json(f"/dogv/{numero}", {"lang": IDIOMA, "ordenSumario": orden}) if comprobar_sumario(extra): sumarios.append((orden, extra)) diarios.append({"numero": numero, "fecha": fecha.isoformat(), "sumarios": [o for o, _ in sumarios]}) for orden, contenido in sumarios: for d in contenido["disposiciones"]: if isinstance(d, dict): disposiciones.append(ficha_disposicion(d, numero, fecha, orden > 1)) return diarios, disposiciones def fecha_del_ultimo_dogv() -> dt.date: """Fecha del último DOGV publicado.""" sumario = pedir_json("/dogv/latest", {"lang": IDIOMA}) if not comprobar_sumario(sumario): raise formato_inesperado("último sumario") candidatos = [str(sumario.get("fechaSumario") or "")] candidatos += [str(d.get("fechaPublicacion") or "") for d in sumario["disposiciones"][:1]] for texto in candidatos: try: return dt.datetime.strptime(texto.replace("/", "-"), "%d-%m-%Y").date() except ValueError: continue raise formato_inesperado("fecha del último sumario") # -------------------------------------------------------------------------- # Órdenes # -------------------------------------------------------------------------- def listar_subsecciones(como_json: bool) -> int: datos = pedir_json("/busquedaAvanzada/seccionesDogv", {"lang": IDIOMA}) if not isinstance(datos, list): raise formato_inesperado("lista de secciones") secciones = [] for s in datos: if not isinstance(s, dict): continue secciones.append({ "id": id_de(s.get("id")), "titulo": " ".join(str(s.get("titulo") or "").split()), "subsecciones": [{"id": id_de(x.get("id")), "descripcion": descripcion_de(x)} for x in s.get("subsecciones") or [] if isinstance(x, dict)], }) if como_json: print(json.dumps(secciones, ensure_ascii=False, indent=2)) return 0 print("Secciones y subsecciones del DOGV (número entre corchetes):") for s in secciones: print(f"[{s['id']}] {s['titulo']}") for x in s["subsecciones"]: print(f" [{x['id']}] {x['descripcion']}") print("\nUse --subseccion N (o --seccion N si la sección no tiene subsecciones).") return 0 def listar_anuncios(args: argparse.Namespace) -> int: if args.fecha: desde = hasta = args.fecha elif args.desde: desde, hasta = args.desde, args.hasta or dt.date.today() else: desde = hasta = fecha_del_ultimo_dogv() if hasta < desde: raise ErrorConsulta("la fecha final es anterior a la inicial.", 2) if (hasta - desde).days + 1 > MAX_DIAS: raise ErrorConsulta(f"el intervalo máximo es de {MAX_DIAS} días; acórtelo.", 2) seccion = args.seccion subseccion = args.subseccion if seccion is None and subseccion is None: subseccion = SUBSECCION_POR_DEFECTO diarios, todas = recopilar(desde, hasta, avisar_progreso=desde != hasta) elegidas = [d for d in todas if (seccion is None or d["seccion_id"] == seccion) and (subseccion is None or d["subseccion_id"] == subseccion)] if args.buscar: aguja = normalizar(args.buscar) elegidas = [d for d in elegidas if aguja in normalizar(d["titulo"] + " " + d["organismo"])] periodo = f"el {desde:%d/%m/%Y}" if desde == hasta else f"del {desde:%d/%m/%Y} al {hasta:%d/%m/%Y}" if args.json: print(json.dumps({ "desde": desde.isoformat(), "hasta": hasta.isoformat(), "filtro": {"seccion": seccion, "subseccion": subseccion, "texto": args.buscar}, "diarios": diarios, "total": len(elegidas), "disposiciones": elegidas, }, ensure_ascii=False, indent=2)) return 0 if not diarios: print(f"No se publicó ningún DOGV {periodo}.") return 0 if subseccion is not None: filtro = f"la subsección {subseccion}" else: filtro = f"la sección {seccion}" if elegidas: nombre = elegidas[0]["subseccion"] if subseccion is not None else elegidas[0]["seccion"] filtro += f" ({nombre})" if nombre else "" if args.buscar: filtro += f" que contienen «{args.buscar}»" actual = None for i, d in enumerate(elegidas, 1): cabecera = (d["numero_dogv"], d["sumario_bis"]) if cabecera != actual: actual = cabecera bis = " (sumario bis)" if d["sumario_bis"] else "" fecha = dt.date.fromisoformat(d["fecha"]) print(f"\nDOGV núm. {d['numero_dogv']}{bis}, de {fecha:%d/%m/%Y}") print(f"{i:>3}. {d['titulo']}") print(f" Organismo: {d['organismo'] or '(sin indicar)'}") print(f" Ficha: {d['url_ficha']}") print(f" PDF: {d['url_pdf']}") print() numeros = ", ".join(str(x["numero"]) for x in diarios) print(f"{len(elegidas)} disposición(es) de {filtro}, {periodo} (DOGV núm. {numeros}).") if not elegidas: print("Use --subsecciones para ver los números de sección y subsección.") return 0 # -------------------------------------------------------------------------- # Línea de órdenes # -------------------------------------------------------------------------- def fecha_argumento(valor: str) -> dt.date: """Admite AAAA-MM-DD o dd/mm/aaaa.""" for formato in ("%Y-%m-%d", "%d/%m/%Y"): try: return dt.datetime.strptime(valor.strip(), formato).date() except ValueError: continue raise argparse.ArgumentTypeError(f"«{valor}» no es una fecha válida (use AAAA-MM-DD o dd/mm/aaaa)") def crear_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="dogv_anuncios.py", description="Lista, en castellano, los anuncios del DOGV de un día o de un intervalo " "para una subsección del sumario (por defecto, la 28: ofertas de empleo " "público, oposiciones y concursos). Sin fechas, usa el último DOGV publicado.", epilog="Ejemplo: python dogv_anuncios.py --fecha 2026-10-30 --subseccion 62", ) cuando = parser.add_mutually_exclusive_group() cuando.add_argument("--fecha", type=fecha_argumento, help="un día concreto (AAAA-MM-DD o dd/mm/aaaa)") cuando.add_argument("--desde", type=fecha_argumento, help=f"primer día del intervalo (máximo {MAX_DIAS} días)") parser.add_argument("--hasta", type=fecha_argumento, help="último día del intervalo (por defecto, hoy); requiere --desde") parser.add_argument("--subseccion", type=int, metavar="N", help=f"número de subsección (por defecto, {SUBSECCION_POR_DEFECTO}; 62 = contratos)") parser.add_argument("--seccion", type=int, metavar="N", help="número de sección, para las que no tienen subsecciones (12 = I, 17 = IV)") parser.add_argument("--buscar", metavar="TEXTO", help="filtra por texto en el título o el organismo (sin distinguir tildes)") parser.add_argument("--subsecciones", action="store_true", help="muestra las secciones y subsecciones con su número y termina") parser.add_argument("--json", action="store_true", help="salida en JSON en lugar de texto") return parser def _consola_utf8() -> None: """Escribe en UTF-8 para que tildes y eñes no fallen en consolas de Windows.""" for flujo in (sys.stdout, sys.stderr): try: flujo.reconfigure(encoding="utf-8", errors="replace") except (AttributeError, ValueError, OSError): pass def main(argv: list[str] | None = None) -> int: _consola_utf8() parser = crear_parser() args = parser.parse_args(argv) if args.hasta and not args.desde: parser.error("--hasta requiere --desde") try: if args.subsecciones: return listar_subsecciones(args.json) return listar_anuncios(args) except ErrorConsulta as error: print(f"Error: {error}", file=sys.stderr) return error.codigo_salida except KeyboardInterrupt: print("Consulta interrumpida.", file=sys.stderr) return 130 if __name__ == "__main__": sys.exit(main())