""" danbooru_harvest: cosecha de metadatos de la API de Danbooru. Contraparte de wd14_harvest.py: en vez de pasar imagenes por WD14, reutiliza el `tag_string` que Danbooru ya trae (tags hechos por la comunidad + rating) y su coocurrencia para expandir los pools por nivel y construir la biblioteca de temas. NUNCA descarga imagenes. Este fichero crece por tareas (plan danbooru-metadata-expansion): T2 load_dictionary / load_seeds / aggregate / bucketize / strip_scene_tags T3 builders + writers (pools, snapshot, validacion) T4 temas (themes.json) T5 CLI (--refresh / --dry-run / --posts-file) """ import sys sys.stdout.reconfigure(encoding="utf-8", errors="backslashreplace") import argparse # noqa: E402 import json # noqa: E402 import os # noqa: E402 import re # noqa: E402 from collections import Counter # noqa: E402 from datetime import datetime, timezone # noqa: E402 from pathlib import Path # noqa: E402 BASE_DIR = Path(__file__).resolve().parent.parent sys.path.insert(0, str(Path(__file__).resolve().parent)) # Reuso del motor de clasificacion ya curado (NO reimplementar). `normalize_key` # vive en pose_bank.py, no en wd14_harvest.py; esta tarea no lo necesita. from wd14_harvest import BUCKETS, classify, load_danbooru # noqa: E402 # Reuso del ensamblado de bloques (NO reimplementar pesos ni severidad). from wd14_blocks import apply_weight, severity, strip_weights_of # noqa: E402 # Reuso del cliente HTTP (Task 1). `DanbooruError` se re-exporta aqui a proposito: # `from danbooru_harvest import refresh, DanbooruError` debe funcionar. from danbooru_client import DanbooruClient, DanbooruError # noqa: E402 LEVELS = ("sugerente", "spicy", "explicito") RATING_MAP = {"sugerente": "g+s", "spicy": "s+q", "explicito": "q+e"} OUT_DIR = BASE_DIR / "Archivos" / "danbooru" # --- Task 3: constantes de pools/bloques ------------------------------------- # Orden de categorias DENTRO de una linea (la pose va primero, como en el motor). BLOCK_ORDER = ("pose", "scene", "state", "expression", "camera") # Coherencia de bloque: topes por categoria y grupos mutuamente excluyentes. # La agregacion por seed premia tags omnipresentes; sin esto las lineas salen # con "standing, sitting, lying" u "open_mouth, closed_mouth" (contradicciones). BLOCK_CAPS = {"pose": 2, "scene": 2, "state": 2, "expression": 1, "camera": 1} BLOCK_CONFLICT_GROUPS = ( frozenset({"standing", "sitting", "lying", "kneeling", "crouching", "squatting", "all_fours", "on_all_fours", "walking", "running", "on_back", "on_stomach", "on_side", "crawling", "reclining", "straddling"}), frozenset({"open_mouth", "closed_mouth", "parted_lips"}), frozenset({"indoors", "outdoors"}), frozenset({"day", "night"}), frozenset({"nude", "completely_nude", "topless", "bottomless", "clothed"}), ) BLOCK_STOPLIST = re.compile( r"^(holding|holding_weapon|full_body|letterboxed|multiple_views|" r"cumulonimbus_cloud)$" ) # nivel de seeds -> pool de severidad (bloques_{sfw,spicy,explicit}.txt) LEVEL_TO_SEVERITY = {"sugerente": "sfw", "spicy": "spicy", "explicito": "explicit"} SEVERITIES = ("sfw", "spicy", "explicit") # Mismo suelo que `wd14_blocks --min-tokens`: menos de esto no es un bloque util. MIN_POOL_TOKENS = 4 def load_dictionary(): """Diccionario Danbooru -> (tags de personaje, tags de copyright, aliases).""" return load_danbooru() def load_seeds(path): """Lee un fichero de seeds por nivel -> {nivel: [tags]}. Formato: una seccion `[sugerente]`/`[spicy]`/`[explicito]` seguida de un tag por linea. Se saltan comentarios (`#`) y lineas en blanco. Una cabecera de seccion desconocida se ignora con un aviso (y no aporta tags). """ seeds = {level: [] for level in LEVELS} current = None text = Path(path).read_text(encoding="utf-8") for raw in text.splitlines(): line = raw.split("#", 1)[0].strip() if not line: continue if line.startswith("[") and line.endswith("]"): name = line[1:-1].strip().lower() if name in seeds: current = name else: print(f"[!] seccion desconocida ignorada: [{line[1:-1].strip()}]") current = None continue if current is None: print(f"[!] tag sin seccion ignorado: {line}") continue seeds[current].append(line) return seeds def aggregate(posts): """Coocurrencia de tags, ponderada por max(score, 1) de cada post. `posts` es una lista de dicts de Danbooru; solo se lee `score` y `tag_string` (separado por espacios). Campos desconocidos se ignoran. """ counter = Counter() for post in posts: try: score = int(post.get("score") or 0) except (TypeError, ValueError): score = 0 weight = max(score, 1) for tag in (post.get("tag_string") or "").split(): counter[tag] += weight return counter def bucketize(counter, chars, copies): """Reparte un Counter tag->peso en los buckets de classify(). -> (buckets, dropped). `buckets` tiene las BUCKETS de wd14_harvest (incluye `otros`); `dropped` es [(tag, motivo)] para lo que classify() rechaza. """ buckets = {name: Counter() for name in BUCKETS} dropped = [] for tag, weight in counter.items(): bucket, reason = classify(tag, chars, copies) if bucket is None: dropped.append((tag, reason)) else: buckets[bucket][tag] += weight return buckets, dropped def strip_scene_tags(block, chars=(), copies=()): """Quita de `block` los tags cuyo bucket classify() es `scene`. Conserva el resto y el orden original. `chars`/`copies` por defecto vacios (Ruling 1): las pool lines ya llegan sin tags de personaje/copyright, asi que Task 6 puede llamarla sin cargar el diccionario. """ if not block: return "" kept = [] for token in block.split(","): tag = token.strip() if not tag: continue bucket, _ = classify(tag, chars, copies) if bucket == "scene": continue kept.append(tag) return ", ".join(kept) # ============================================================================= # Task 3: builders + writers (pools, snapshot, validacion) # ============================================================================= def _tokens(line): """Separa una linea de bloque en tags (por comas), sin vacios.""" return [t.strip() for t in (line or "").split(",") if t.strip()] def _top_pairs(bucket): """[(tag, peso)] de mayor a menor. Acepta Counter o dict plano.""" if bucket is None: return [] if hasattr(bucket, "most_common"): return bucket.most_common() return sorted(bucket.items(), key=lambda kv: (-kv[1], kv[0])) def _category_of(tag, chars, copies): """Categoria Danbooru si la conocemos por el diccionario, si no None. El snapshot guarda `category` desde `DanbooruClient.tag()`, pero este builder no recibe cliente (no debe tocar la red): usa el diccionario offline como aproximacion (4=personaje, 3=copyright) y `null` para el resto. """ t = (tag or "").lower() if t in chars: return 4 if t in copies: return 3 return None def build_block(seed, buckets, max_tokens=28): """Una linea de prompt a partir de la coocurrencia de UN seed. Orden: `1girl` + `seed` (garantizado) + top `pose` + top `scene` + top `state` + top `expression` + top `camera`. Cada tag pasa por `apply_weight` (NO se reimplementa el peso). Se recorta a `max_tokens` tags separados por comas. `buckets` es `{bucket: Counter|dict}`. Coherencia (`BLOCK_CAPS`/`BLOCK_CONFLICT_GROUPS`/`BLOCK_STOPLIST`): tope por categoria, un solo tag por grupo excluyente (posture/mouth/place/time/ nudity) y fuera los omnipresentes (`holding`, `full_body`...). Sin esto la agregacion mezcla opuestos ("standing, sitting", "open_mouth, closed_mouth"). """ line = ["1girl"] seen = {"1girl"} seed_norm = (seed or "").strip() if seed_norm and seed_norm not in seen: line.append(seed_norm) seen.add(seed_norm) used_groups = { idx for idx, group in enumerate(BLOCK_CONFLICT_GROUPS) if seed_norm.lower() in group } for cat in BLOCK_ORDER: cap = BLOCK_CAPS.get(cat, 2) taken = 0 for tag, _ in _top_pairs(buckets.get(cat)): if taken >= cap or len(line) >= max_tokens: break tag = (tag or "").strip() if not tag or tag in seen or BLOCK_STOPLIST.match(tag.lower()): continue groups = [ idx for idx, group in enumerate(BLOCK_CONFLICT_GROUPS) if tag.lower() in group ] if groups and groups[0] in used_groups: continue if groups: used_groups.add(groups[0]) seen.add(tag) line.append(tag) taken += 1 return ", ".join(apply_weight(t) for t in line[:max_tokens]) def build_pools(seeds, per_level_buckets, chars, copies): """`{nivel: {seed: {bucket: Counter}}}` -> `{severity: [lineas]}`. El nivel del seed manda sobre la severidad de la linea (mapeo fijo `LEVEL_TO_SEVERITY`). Se descartan los seeds que rinden menos de `MIN_POOL_TOKENS` tags utiles (mismo suelo que `wd14_blocks --min-tokens`); "utiles" = los que vienen de los buckets, SIN contar el `1girl` de cabecera ni el `seed` (ambos garantizados). Tambien se descartan lineas repetidas. `chars`/`copies` se aceptan para simetria con el resto del motor; `bucketize()` ya los aplico, aqui no re-clasifica. """ pools = {sev: [] for sev in SEVERITIES} for level in LEVELS: target = LEVEL_TO_SEVERITY[level] seed_buckets = per_level_buckets.get(level) or {} # tolera dos formas: {seed: {bucket: Counter}} (una por seed) o # {bucket: Counter} (una por nivel, compartida por sus seeds) shared = bool(set(seed_buckets) & set(BUCKETS)) seen_lines = set(pools[target]) for seed in seeds.get(level, []): buckets = seed_buckets if shared else seed_buckets.get(seed) if not buckets: continue line = build_block(seed, buckets) # tags utiles = solo los que vienen de los buckets: fuera el `1girl` # de cabecera y el `seed` (ambos garantizados, no cuentan al suelo) seed_norm = (seed or "").strip() useful = sum( 1 for t in _tokens(strip_weights_of(line)) if t.lower() != "1girl" and t != seed_norm ) if useful < MIN_POOL_TOKENS: continue if line in seen_lines: continue seen_lines.add(line) pools[target].append(line) return pools def _write_text_atomic(path, text): """Escribe `text` en `path` via `.tmp` + `os.replace`. Deja intacto el fichero previo si algo falla.""" path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) tmp = path.parent / (path.name + ".tmp") try: with tmp.open("w", encoding="utf-8", newline="\n") as f: f.write(text) os.replace(tmp, path) finally: if tmp.exists(): try: os.unlink(tmp) except OSError: pass def write_json_atomic(path, obj): """Volcado atomico de `obj` a JSON (`tmp` + `os.replace`).""" path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) tmp = path.parent / (path.name + ".tmp") try: with tmp.open("w", encoding="utf-8", newline="\n") as f: json.dump(obj, f, ensure_ascii=False, indent=2) f.write("\n") os.replace(tmp, path) finally: if tmp.exists(): try: os.unlink(tmp) except OSError: pass def write_pools(pools, out_dir=OUT_DIR, keep_on_empty=True): """Escribe `bloques_{sfw,spicy,explicit}.txt` -> `{severity: n_lineas}`. Un pool vacio **no** pisa el fichero previo (keep_on_empty=True) y avisa; con keep_on_empty=False escribe solo la cabecera. Escritura atomica. """ out_dir = Path(out_dir) out_dir.mkdir(parents=True, exist_ok=True) result = {} for sev in SEVERITIES: lines = [l.strip() for l in (pools.get(sev) or []) if l and l.strip()] path = out_dir / f"bloques_{sev}.txt" if not lines: result[sev] = 0 if keep_on_empty: print(f"[!] pool {sev} vacio: se conserva {path.name} sin tocar") continue # severidad sobre tags SIN peso: `(spread_ass:0.6)` no casa con los sets dist = Counter( severity(_tokens(strip_weights_of(l))) for l in lines ) header = ( f"# danbooru bloques {sev.upper()}: {len(lines)} lineas\n" "# contenido por severidad: " + ", ".join(f"{k}={dist.get(k, 0)}" for k in SEVERITIES) + "\n\n" ) _write_text_atomic(path, header + "\n".join(lines) + "\n") result[sev] = len(lines) return result def build_snapshot(seeds, posts_by_level, chars, copies, params): """Snapshot local: `{fecha, params, tags, cooc}` (ver plan Task 3). `posts_by_level` es `{nivel: {seed: [posts]}}`. `tags` agrega TODOS los posts (`count` = apariciones crudas, `score` = peso sumado, `category` del diccionario o null). `cooc` es `{seed: {tag: peso}}` por seed. """ tag_count = Counter() tag_score = Counter() cooc = {} for level in LEVELS: for seed, posts in (posts_by_level.get(level) or {}).items(): posts = posts or [] counter = aggregate(posts) merged = cooc.setdefault(seed, {}) for tag, weight in counter.items(): merged[tag] = merged.get(tag, 0) + int(weight) for post in posts: for tag in (post.get("tag_string") or "").split(): tag_count[tag] += 1 tag_score.update(counter) tags = { tag: { "count": int(count), "category": _category_of(tag, chars, copies), "score": int(tag_score.get(tag, 0)), } for tag, count in tag_count.items() } return { "fecha": datetime.now(timezone.utc).isoformat(), "params": params, "tags": tags, "cooc": cooc, } def write_validation_report(path, accepted, dropped, canonicalized, revisar): """Escribe `validacion.txt` (atomico) y anade los tags `revisar` a `revisar/tags.txt` (uno por linea, deduplicado). Nunca descarta nada. """ path = Path(path) accepted = accepted or {} dropped = dropped or [] canonicalized = canonicalized or {} revisar = [t.strip() for t in (revisar or []) if t and t.strip()] out = [ "# validacion danbooru", f"# fecha: {datetime.now(timezone.utc).isoformat()}", f"# aceptados: {len(accepted)} | rechazados: {len(dropped)} " f"| sinonimos: {len(canonicalized)} | a revisar: {len(revisar)}", "", "[ACEPTADOS]", ] if isinstance(accepted, dict): for tag in sorted(accepted, key=str): out.append(f" {tag}\t{accepted[tag]}") else: for tag in sorted(accepted, key=str): out.append(f" {tag}") out.append("[RECHAZADOS]") for item in dropped: if isinstance(item, (tuple, list)) and len(item) >= 2: out.append(f" {item[0]}\t{item[1]}") else: out.append(f" {item}") out.append("[CANONICALIZADOS]") if isinstance(canonicalized, dict): for raw in sorted(canonicalized, key=str): out.append(f" {raw}\t->\t{canonicalized[raw]}") else: for raw in canonicalized: out.append(f" {raw}") out.append("[A REVISAR (otros)]") for tag in revisar: out.append(f" {tag}") _write_text_atomic(path, "\n".join(out) + "\n") if revisar: rpath = path.parent / "revisar" / "tags.txt" existing = [] seen = set() if rpath.exists(): for line in rpath.read_text(encoding="utf-8").splitlines(): tag = line.strip() if tag and tag not in seen: seen.add(tag) existing.append(tag) for tag in revisar: if tag not in seen: seen.add(tag) existing.append(tag) # escritura ATOMICA (tmp + os.replace): un fallo deja el previo intacto _write_text_atomic(rpath, "\n".join(existing) + "\n") # ============================================================================= # Task 4: temas (themes.json) # ============================================================================= # Ejes de un tema (spec §6). `outfit` cuelga un nivel LEVELS; el resto, variantes # planas. La validacion recorre estos cuatro. THEME_AXES = ("outfit", "scene", "lighting", "props") # `classify()` no tiene bucket propio de luz (la luz vive dentro de `scene`), asi # que para poblar el eje `lighting` de los temas auto se usa este regex. NO # sustituye a `classify`: solo separa, dentro de los tags de escena, los de luz. LIGHTING = re.compile( r"sunset|sunrise|golden_hour|backlight|backlighting|rim_light|" r"cinematic_lighting|lighting|^light$|^lights$|candle|firelight|" r"moonlight|^neon$|neon_lights|window_light|dappled_sunlight|sunlight|" r"^sun$|soft_light|hard_light|^glow$|glowing|^dark$|shadow|silhouette|" r"spotlight|^night$|^day$|^dusk$|^dawn$|^bright$", re.I, ) def _sorted_pairs(pairs): """[(tag, peso)] de mayor a menor; tolera pesos no numericos (-> 0).""" def key(item): try: weight = float(item[1]) except (TypeError, ValueError): weight = 0.0 return (-weight, str(item[0])) return sorted((pairs or {}).items(), key=key) def _total_weight(pairs): """Peso total de un `{tag: peso}` (tolerante a valores no numericos).""" total = 0.0 for value in (pairs or {}).values(): try: total += float(value) except (TypeError, ValueError): continue return total def _empty_theme(source): """Tema vacio con el schema de §6 (todos los ejes presentes).""" return { "source": source, "seed_tags": [], "outfit": {}, "scene": {}, "lighting": {}, "props": {}, } def _coherent_tags(tags, limit=None): """Filtra una lista de tags con los grupos excluyentes y la stoplist. Mantiene el orden de entrada (mayor peso primero) y se queda con el primer tag de cada grupo (`BLOCK_CONFLICT_GROUPS`). `limit` recorta el resultado. """ out = [] used = set() for tag in tags: low = (tag or "").strip().lower() if not low or BLOCK_STOPLIST.match(low): continue groups = [ idx for idx, group in enumerate(BLOCK_CONFLICT_GROUPS) if low in group ] if groups and groups[0] in used: continue if groups: used.add(groups[0]) out.append(tag) if limit and len(out) >= limit: break return out def discover_themes(cooc, chars, copies, top=8): """Temas auto: uno por cada top scene-seed de la coocurrencia. `cooc` es `{seed: {tag: peso}}` (como el `cooc` del snapshot). Se consideran "seed de escena" los que `classify()` mete en el bucket `scene`; se ordenan por peso total y se toman `top`. Para cada uno: seed_tags = [seed] scene = {seed: [seed, *top tags de escena]} lighting = {top_tag_de_luz: [*tags de luz]} si los hay outfit = {"sugerente": {"theme": [*top tags de ropa]}} si los hay props = {} El resultado es `{seed: tema}`. """ chars = chars or () copies = copies or () scene_seeds = [] for seed, pairs in (cooc or {}).items(): bucket, _ = classify(seed, chars, copies) if bucket != "scene": continue scene_seeds.append((_total_weight(pairs), str(seed), seed)) # mayor peso primero; string como desempate estable scene_seeds.sort(key=lambda item: (-item[0], item[1])) if top: scene_seeds = scene_seeds[:top] themes = {} for _total, _key, seed in scene_seeds: pairs = cooc.get(seed) or {} scene_tags, outfit_tags, light_tags = [], [], [] for tag, _weight in _sorted_pairs(pairs): if not tag or tag == seed or tag.lower() in ("1girl", "solo"): continue bucket, _ = classify(tag, chars, copies) if bucket == "scene": scene_tags.append(tag) if LIGHTING.search(tag): light_tags.append(tag) elif bucket == "outfit": outfit_tags.append(tag) if top: scene_tags = _coherent_tags(scene_tags, top) outfit_tags = outfit_tags[:top] light_tags = _coherent_tags(light_tags, top) theme = _empty_theme("auto") theme["seed_tags"] = [seed] theme["scene"] = {seed: [seed] + [t for t in scene_tags if t != seed]} if outfit_tags: theme["outfit"] = {"sugerente": {"theme": outfit_tags}} if light_tags: theme["lighting"] = {light_tags[0]: light_tags} themes[seed] = theme return themes def _apply_curated_line(theme, key, tags): """Vuelca una linea `clave = tags` sobre un tema curado (in-place).""" if key == "seed": for tag in tags: if tag not in theme["seed_tags"]: theme["seed_tags"].append(tag) return parts = key.split(".") axis = parts[0] if axis == "outfit": if len(parts) < 3: print(f"[!] linea de outfit incompleta ignorada: {key}") return level = parts[1] variant = ".".join(parts[2:]) theme["outfit"].setdefault(level, {})[variant] = tags elif axis in ("scene", "lighting", "props"): if len(parts) < 2: print(f"[!] linea incompleta ignorada: {key}") return variant = ".".join(parts[1:]) theme[axis][variant] = tags else: print(f"[!] clave curada desconocida ignorada: {key}") def load_curated(path): """Parsea `temas_curados.txt` -> `{nombre: tema}`; `{}` si no existe. Formato (spec Task 4): [tema:playa] seed = beach outfit.sugerente.bikini = bikini, swimsuit scene.shore = beach, ocean, horizon lighting.sunset = sunset, backlight, golden hour props.towel = beach towel, beach umbrella Se saltan comentarios (`#`) y lineas en blanco. Los valores conservan su forma exacta (espacios incluidos); solo se recortan extremos. """ path = Path(path) if not path.exists(): return {} themes = {} current = None for raw in path.read_text(encoding="utf-8").splitlines(): line = raw.split("#", 1)[0].strip() if not line: continue if line.startswith("[") and line.endswith("]"): header = line[1:-1].strip() if header.lower().startswith("tema:"): name = header.split(":", 1)[1].strip() if not name: print("[!] cabecera de tema vacia ignorada") current = None continue current = name themes.setdefault(name, _empty_theme("curado")) else: print(f"[!] seccion desconocida ignorada: [{header}]") current = None continue if current is None: print(f"[!] linea sin tema ignorada: {line}") continue if "=" not in line: print(f"[!] linea sin '=' ignorada: {line}") continue key, value = line.split("=", 1) key = key.strip().lower() tags = [t.strip() for t in value.split(",") if t.strip()] _apply_curated_line(themes[current], key, tags) return themes def _check_tags(name, path, tags, errors): """Anade a `errors` si `tags` no es lista de strings no vacios.""" if not isinstance(tags, list): errors.append(f"tema '{name}': '{path}' debe ser una lista de tags") return for tag in tags: if not isinstance(tag, str) or not tag.strip(): errors.append( f"tema '{name}': '{path}' contiene un tag invalido: {tag!r}" ) break def validate_theme(name, theme): """Valida un tema contra el schema de §6 -> lista de errores ([] == valido). Reglas: `seed_tags` lista no vacia; al menos un eje de `THEME_AXES` con variantes; toda clave de `outfit` en `LEVELS`; todo valor de variante, lista de strings no vacios. """ errors = [] if not isinstance(theme, dict): return [f"tema '{name}': no es un objeto (dict)"] seeds = theme.get("seed_tags") if not isinstance(seeds, list) or not seeds: errors.append(f"tema '{name}': 'seed_tags' debe ser una lista no vacia") non_empty_axis = False for axis in THEME_AXES: val = theme.get(axis) if val is None or val == {}: continue if not isinstance(val, dict): errors.append(f"tema '{name}': eje '{axis}' debe ser un objeto") continue non_empty_axis = non_empty_axis or bool(val) if axis == "outfit": for level, variants in val.items(): if level not in LEVELS: errors.append( f"tema '{name}': nivel de outfit invalido '{level}' " f"(validos: {', '.join(LEVELS)})" ) if not isinstance(variants, dict): errors.append( f"tema '{name}': 'outfit.{level}' debe ser un objeto " "de variantes" ) continue for variant, tags in variants.items(): _check_tags(name, f"outfit.{level}.{variant}", tags, errors) else: for variant, tags in val.items(): _check_tags(name, f"{axis}.{variant}", tags, errors) if not non_empty_axis: errors.append( f"tema '{name}': al menos uno de {THEME_AXES} debe tener variantes" ) return errors def build_themes(auto, curated, chars, copies): """Fusiona auto + curados (el curado pisa al auto por nombre) y valida. Los temas **curados son SIEMPRE del usuario**: se conservan aunque sean invalidos (marcados `source="curado"`), de modo que el unico muro duro, `write_themes`, frene la escritura con un error claro que hay que arreglar; un bloque roto de `temas_curados.txt` NO desaparece en silencio. Los temas **auto** que no validan si se descartan (ruido de maquina) con aviso. `chars`/`copies` se aceptan por simetria con el resto del motor (aqui no se reclasifica: `discover_themes` ya lo hizo). """ auto = auto or {} curated = curated or {} merged = {} for name, theme in auto.items(): merged[name] = theme for name, theme in curated.items(): merged[name] = theme curated_names = set(curated) result = {} for name, theme in merged.items(): if name in curated_names: if isinstance(theme, dict): theme["source"] = "curado" result[name] = theme continue errs = validate_theme(name, theme) if errs: for err in errs: print(f"[!] tema auto '{name}' descartado: {err}") continue result[name] = theme return result def write_themes(themes, path): """Valida TODOS los temas y escribe `themes.json` (atomico) si son validos. Si algun tema falla, imprime cada error y lanza `ValueError` **sin escribir** (el fichero destino queda intacto o ausente). Si todos son validos, delega en `write_json_atomic`. """ path = Path(path) themes = themes or {} all_errors = [] for name in sorted(themes, key=str): all_errors.extend(validate_theme(name, themes[name])) if all_errors: for err in all_errors: print(f"[!] {err}") raise ValueError( f"themes invalidos ({len(all_errors)} error/es); no se escribio {path}" ) write_json_atomic(path, themes) # ============================================================================= # Task 5: refresco (red) + CLI # ============================================================================= SEEDS_FILE = BASE_DIR / "Archivos" / "danbooru_seeds.txt" CURATED_FILE = BASE_DIR / "Archivos" / "temas_curados.txt" # Umbrales de partida del plan (ajustables tras el primer snapshot real). DANBOORU_MIN_SCORE = 20 DANBOORU_DEFAULT_LIMIT = 200 # `order:score` da HTTP 500 (QueryCanceled) en tags ultra-comunes; se reintenta # con ordenes baratos (`id` y el orden por defecto) antes de descartar el seed. FALLBACK_ORDERS = ("id", None) def _read_posts_file(path): """Lee un fixture `{nivel: {seed: [posts]}}` para `--dry-run`/`--posts-file`. Normaliza a `{nivel: {seed: [posts]}}` para TODOS los niveles (los que falten quedan como dicts vacios). No toca la red. """ path = Path(path) if not path.exists(): raise FileNotFoundError(f"posts-file no existe: {path}") # `utf-8-sig`: tolera el BOM que PowerShell 5.1 `Set-Content -Encoding utf8` # deja al crear el fixture (y funciona igual sin BOM). data = json.loads(path.read_text(encoding="utf-8-sig")) if not isinstance(data, dict): raise ValueError( "posts-file debe ser un objeto JSON {nivel: {seed: [posts, ...]}}" ) posts_by_level = {level: {} for level in LEVELS} for level, by_seed in data.items(): if level not in posts_by_level or not isinstance(by_seed, dict): continue for seed, posts in by_seed.items(): posts_by_level[level][seed] = list(posts or []) return posts_by_level def _fetch_posts(client, seeds, limit): """Descarga posts por seed -> `(posts_by_level, skipped, fetched)`. `client.posts(seed, rating=RATING_MAP[level], min_score=20, limit=limit or 200)`. **Fallback de orden:** si `order:score` falla (HTTP 500 `QueryCanceled` en tags ultra-comunes), se reintenta con `order:id` antes de descartar el seed. **Aislamiento por seed (T7):** un `DanbooruError` de UN seed que falla en ambos ordenes NO aborta la corrida: se anota `"{nivel}:{seed}: {err}"` en `skipped`, se avisa y se continua con el resto. `fetched` = n de llamadas con exito (0 == todos los seeds fallaron). """ limit = int(limit) if limit else DANBOORU_DEFAULT_LIMIT posts_by_level = {level: {} for level in LEVELS} skipped = [] fetched = 0 for level in LEVELS: for seed in seeds.get(level, []): try: posts = client.posts( seed, rating=RATING_MAP[level], min_score=DANBOORU_MIN_SCORE, limit=limit, ) except DanbooruError as exc: posts = None for fallback in FALLBACK_ORDERS: try: posts = client.posts( seed, rating=RATING_MAP[level], min_score=DANBOORU_MIN_SCORE, limit=limit, order=fallback, ) except DanbooruError: continue print( f"[!] {level}:{seed}: {exc}; recuperado con " f"order:{fallback or 'default'}" ) break if posts is None: note = f"{level}:{seed}: {exc}" skipped.append(note) print(f"[!] {note} (se continua con el resto de seeds)") continue posts_by_level[level][seed] = list(posts or []) fetched += 1 return posts_by_level, skipped, fetched def _bucketize_levels(posts_by_level, chars, copies): """`{nivel: {seed: [posts]}}` -> `(per_level_buckets, accepted, dropped, revisar)`. `per_level_buckets` = `{nivel: {seed: {bucket: Counter}}}` (una linea por seed). `accepted` = `{tag: bucket}` (solo los clasificados); `dropped` = `{tag: motivo}`; `revisar` = tags que `classify()` manda a `otros` (sin categoria clara). """ per_level_buckets = {level: {} for level in LEVELS} accepted = {} dropped = {} revisar = [] for level in LEVELS: for seed, posts in (posts_by_level.get(level) or {}).items(): buckets, drop = bucketize(aggregate(posts), chars, copies) per_level_buckets[level][seed] = buckets for tag, reason in drop: dropped.setdefault(tag, reason) for name, counter in buckets.items(): for tag in counter: if name == "otros": revisar.append(tag) else: accepted[tag] = name return per_level_buckets, accepted, dropped, revisar def refresh(limit=None, dry_run=False, posts_file=None, out_dir=OUT_DIR, seeds_path=SEEDS_FILE, curated_path=CURATED_FILE, client=None): """Refresco manual: red -> pools/themes/snapshot/validacion (todo atomico). Devuelve `{"blocks": {severity: n_lineas}, "themes": n, "skipped": [...]}`; `skipped` viene vacio en un refresco correcto. `dry_run`/`posts_file` leen los posts de un fixture (`posts_file`) en vez del cliente y escriben en `/_preview/` **sin tocar la red**. **Aislamiento por seed (T7):** si un seed falla (`DanbooruError`), se anota en `"skipped"` y se continua con el resto; basta UN seed con exito para construir y escribir con los datos disponibles (los niveles vacios conservan su `bloques_*.txt`). Si **todos** los seeds fallan (nada descargado) se mantiene el contrato previo: aviso `[!] Danbooru no disponible; ...` y **no se toca** ningun artefacto. Sin resultados (pero sin errores) tampoco se escribe nada. """ try: seeds = load_seeds(seeds_path) chars, copies, aliases = load_dictionary() fixture_mode = posts_file is not None or dry_run skipped = [] if posts_file is not None: posts_by_level = _read_posts_file(posts_file) elif dry_run: print( "[!] --dry-run sin --posts-file: sin datos de entrada " "(no se toca la red)" ) posts_by_level = {level: {} for level in LEVELS} else: if client is None: client = DanbooruClient() limit_used = int(limit) if limit else DANBOORU_DEFAULT_LIMIT print(f"[*] descargando metadatos ({limit_used} posts/seed)...") posts_by_level, skipped, fetched = _fetch_posts(client, seeds, limit) if fetched == 0 and skipped: # TODOS los seeds fallaron: contrato previo, no tocar artefactos. print( "[!] Danbooru no disponible; se conservan los artefactos " "previos" ) return {"blocks": {}, "themes": 0, "skipped": skipped} if skipped: print( f"[!] {len(skipped)} seed(s) fallaron; se continua con " f"{fetched} seed(s) correctos" ) total_posts = sum( len(posts) for by_seed in posts_by_level.values() for posts in by_seed.values() ) if total_posts == 0: print("[!] sin resultados; se conservan los artefactos previos") return { "blocks": {}, "themes": 0, "skipped": skipped or ["sin resultados; artefactos previos intactos"], } target = (Path(out_dir) / "_preview") if fixture_mode else Path(out_dir) per_level_buckets, accepted, dropped, revisar = _bucketize_levels( posts_by_level, chars, copies ) pools = build_pools(seeds, per_level_buckets, chars, copies) params = { "min_score": DANBOORU_MIN_SCORE, "limit": int(limit) if limit else DANBOORU_DEFAULT_LIMIT, } snapshot = build_snapshot(seeds, posts_by_level, chars, copies, params) auto = discover_themes(snapshot.get("cooc") or {}, chars, copies) curated = load_curated(curated_path) themes = build_themes(auto, curated, chars, copies) # `write_themes` valida TODOS los temas antes de escribir: si un tema # curado esta roto, lanza ValueError y no se escribe ningun pool. write_themes(themes, target / "themes.json") blocks = write_pools(pools, out_dir=target) write_json_atomic(target / "snapshot.json", snapshot) canonicalized = { tag: aliases[tag] for tag in set(accepted) | set(dropped) if tag in aliases and aliases[tag] != tag } write_validation_report( target / "validacion.txt", accepted, list(dropped.items()), canonicalized, revisar, ) if not fixture_mode: print(f"[ok] artefactos en {target}") return {"blocks": blocks, "themes": len(themes), "skipped": skipped} except DanbooruError as exc: # Red de seguridad: con el aislamiento por seed, `_fetch_posts` ya captura # los fallos; esto cubre un `DanbooruError` inesperado fuera de ese bucle. print("[!] Danbooru no disponible; se conservan los artefactos previos") print(f" ({exc})") return {"blocks": {}, "themes": 0, "skipped": [str(exc) or "DanbooruError"]} def _build_parser(): parser = argparse.ArgumentParser( prog="danbooru_harvest", description="Cosecha de metadatos de Danbooru -> pools y temas (manual).", ) parser.add_argument( "--refresh", action="store_true", help="descarga metadatos y regenera pools/temas/snapshot/validacion", ) parser.add_argument( "--limit", type=int, default=None, help="posts por seed (default 200)", ) parser.add_argument( "--dry-run", action="store_true", help="no toca la red; escribe en /_preview/", ) parser.add_argument( "--posts-file", default=None, help="fixture JSON {nivel: {seed: [posts]}} en vez del cliente", ) parser.add_argument("--out", default=None, help="directorio de salida") parser.add_argument("--seeds", default=None, help="fichero de seeds por nivel") parser.add_argument("--curated", default=None, help="fichero de temas curados") return parser def main(argv=None): parser = _build_parser() args = parser.parse_args(argv) if not (args.refresh or args.dry_run or args.posts_file): parser.print_help() return 0 result = refresh( limit=args.limit, dry_run=args.dry_run, posts_file=args.posts_file, out_dir=Path(args.out) if args.out else OUT_DIR, seeds_path=Path(args.seeds) if args.seeds else SEEDS_FILE, curated_path=Path(args.curated) if args.curated else CURATED_FILE, ) blocks = result.get("blocks") or {} if blocks: print( "[ok] bloques por nivel: " + ", ".join(f"{sev}={blocks.get(sev, 0)}" for sev in SEVERITIES) ) print(f"[ok] temas: {result.get('themes', 0)}") if result.get("skipped"): print("[!] saltado: " + "; ".join(result["skipped"])) return 0 if __name__ == "__main__": raise SystemExit(main())