""" Cliente HTTP ligero (con cache) para la Danbooru API. Solo metadata: NUNCA descarga imagenes. Respeta la politica de uso de Danbooru (User-Agent identificable, delay educado entre peticiones, respeta 429/Retry-After) y guarda la respuesta JSON cruda en disco para no repetir inferencia de red. cache/.json = {"url", "fetched", "data"} Uso: from danbooru_client import DanbooruClient c = DanbooruClient() c.posts("beach rating:general", min_score=20, limit=50) """ import hashlib import json import os import sys import time from datetime import datetime, timezone from pathlib import Path try: sys.stdout.reconfigure(encoding="utf-8", errors="backslashreplace") except Exception: pass import requests # noqa: E402 BASE_DIR = Path(__file__).resolve().parent.parent BASE_URL = "https://danbooru.donmai.us" DEFAULT_USER_AGENT = "plumloom-danbooru/1.0 (+contact: set DANBOORU_USER_AGENT)" DEFAULT_DELAY = 1.0 DEFAULT_TIMEOUT = 30 MAX_ATTEMPTS = 3 class DanbooruError(Exception): """Error de red, HTTP o cache del cliente de Danbooru.""" def default_user_agent(): """User-Agent identificable; configurable con DANBOORU_USER_AGENT.""" return os.environ.get("DANBOORU_USER_AGENT") or DEFAULT_USER_AGENT def _env_float(name, default): raw = os.environ.get(name) if raw is None or raw == "": return default try: return float(raw) except ValueError: return default def _env_int_or_none(name): raw = os.environ.get(name) if raw is None or raw == "": return None try: return int(raw) except ValueError: return None class DanbooruClient: def __init__(self, user_agent=None, cache_dir=None, delay=None, ttl=None, base_url=BASE_URL, offline=False, session=None): self.user_agent = user_agent or default_user_agent() self.base_url = (base_url or BASE_URL).rstrip("/") self.delay = _env_float("DANBOORU_DELAY", DEFAULT_DELAY) if delay is None else float(delay) self.ttl = _env_int_or_none("DANBOORU_CACHE_TTL") if ttl is None else int(ttl) self.offline = bool(offline) self.timeout = DEFAULT_TIMEOUT if cache_dir is None: cache_dir = BASE_DIR / "Archivos" / "danbooru" / "cache" self.cache_dir = Path(cache_dir) self.cache_dir.mkdir(parents=True, exist_ok=True) if session is None: session = requests.Session() session.headers["User-Agent"] = self.user_agent self.session = session self._last_request = None # ------------------------------------------------------------------ API def posts(self, tags, rating=None, min_score=20, limit=200, order="score"): """Posts que casan con `tags`. Devuelve la lista cruda de dicts.""" parts = [tags] if tags else [] if rating: parts.append(f"rating:{rating}") if min_score is not None: parts.append(f"score:>={int(min_score)}") if order: parts.append(f"order:{order}") params = {"tags": " ".join(p for p in parts if p), "limit": int(limit)} data = self._request("/posts.json", params) if data is None: return [] if not isinstance(data, list): raise DanbooruError(f"Respuesta inesperada de /posts.json: {type(data).__name__}") return data def tag(self, name): """Metadata de un tag concreto, o None si no existe.""" data = self._request("/tags.json", {"search[name_matches]": name, "limit": 1}) if not isinstance(data, list) or not data: return None for item in data: if isinstance(item, dict) and item.get("name") == name: return item return None # -------------------------------------------------------------- cache def _cache_path(self, path, params): src = path + json.dumps(sorted(params.items()), sort_keys=True, ensure_ascii=False) digest = hashlib.sha1(src.encode("utf-8")).hexdigest() return self.cache_dir / f"{digest}.json" def _read_cache(self, cache_path): try: with cache_path.open(encoding="utf-8") as f: entry = json.load(f) except (OSError, ValueError): return None if not isinstance(entry, dict) or "data" not in entry: return None return entry def _write_cache(self, cache_path, url, data): entry = { "url": url, "fetched": datetime.now(timezone.utc).isoformat(), "data": data, } self.cache_dir.mkdir(parents=True, exist_ok=True) tmp = cache_path.parent / (cache_path.name + ".tmp") with tmp.open("w", encoding="utf-8", newline="\n") as f: json.dump(entry, f, ensure_ascii=False) tmp.replace(cache_path) @staticmethod def _is_fresh(entry, ttl): if ttl is None: return True try: fetched = datetime.fromisoformat(entry["fetched"]) if fetched.tzinfo is None: fetched = fetched.replace(tzinfo=timezone.utc) except (KeyError, ValueError, TypeError): return False age = (datetime.now(timezone.utc) - fetched).total_seconds() return age < ttl # ------------------------------------------------------------- request def _request(self, path, params): """Cache -> red -> cache. Devuelve el JSON (list/dict) ya parseado.""" params = params or {} cache_path = self._cache_path(path, params) cached = self._read_cache(cache_path) if cached is not None: if self.offline: return cached["data"] if self._is_fresh(cached, self.ttl): return cached["data"] if self.offline: raise DanbooruError( f"offline=True y no hay cache para {path} {sorted(params.items())}" ) url = self.base_url + path data = self._fetch_with_retry(url, params, path) self._write_cache(cache_path, url, data) return data @staticmethod def _retry_after(resp, fallback): raw = None try: raw = resp.headers.get("Retry-After") except AttributeError: raw = None if raw: try: value = float(raw) return max(0.0, value) except (TypeError, ValueError): pass return fallback def _throttle(self): if self._last_request is None or self.delay <= 0: return elapsed = time.monotonic() - self._last_request if elapsed < self.delay: time.sleep(self.delay - elapsed) def _fetch_with_retry(self, url, params, path): last_error = None for attempt in range(MAX_ATTEMPTS): if attempt == 0: self._throttle() try: self._last_request = time.monotonic() resp = self.session.get(url, params=params, timeout=self.timeout) except requests.RequestException as exc: last_error = f"error de red: {exc}" if attempt < MAX_ATTEMPTS - 1: time.sleep(self.delay * (2 ** attempt)) continue raise DanbooruError(f"{path}: {last_error}") from exc status = resp.status_code if status == 429: last_error = "HTTP 429 (rate limited)" if attempt < MAX_ATTEMPTS - 1: wait = self._retry_after(resp, self.delay) time.sleep(wait) continue raise DanbooruError(f"{path}: {last_error} tras {MAX_ATTEMPTS} intentos") if 500 <= status < 600: last_error = f"HTTP {status}" if attempt < MAX_ATTEMPTS - 1: time.sleep(self.delay * (2 ** attempt)) continue raise DanbooruError(f"{path}: {last_error} tras {MAX_ATTEMPTS} intentos") if status >= 400: raise DanbooruError(f"{path}: HTTP {status}") try: return resp.json() except ValueError as exc: raise DanbooruError(f"{path}: JSON invalido") from exc raise DanbooruError(f"{path}: {last_error or 'fallo desconocido'}")