""" regen_batch_items: regenera imagenes sueltas de un batch de `batch_poses_chars`. Reconstruye el job EXACTO (mismo tema, mismas variantes de outfit/escena/luz/ props, misma camara y mismo dress) leyendo `plan.json` + `manifest_all.json`, y solo cambia la semilla (aleatoria). Sirve para corregir una imagen sin tocar el resto del lote: lo viejo se mueve a `/_reemplazadas/`. python Engine/regen_batch_items.py --run "Output/pilots/.../BATCH_X" \ --items "clave:pose_id,clave:pose_id" --dry-run python Engine/regen_batch_items.py --run "..." --items "..." [--neg-extra "..."] Nota: el prompt nuevo sale de la linea ACTUAL del banco y del outfit ACTUAL del tema, asi que primero hay que corregir el banco/themes.json. """ import argparse import json import random import shutil import sys from pathlib import Path sys.stdout.reconfigure(encoding="utf-8", errors="backslashreplace") BASE_DIR = Path(__file__).resolve().parent.parent sys.path.insert(0, str(BASE_DIR / "Engine")) sys.path.insert(0, str(BASE_DIR / "Archivos")) import pose_bank # noqa: E402 import pack_builder as pb # noqa: E402 from prompt_builder import PromptBuilder # noqa: E402 LEVELS = ("sugerente", "spicy", "explicito") def parse_items(text): """`clave:pose_id[:stage]` -> [(clave, pose_id, stage|None)]""" out = [] for chunk in text.split(","): chunk = chunk.strip() if not chunk: continue parts = [p.strip() for p in chunk.split(":")] if len(parts) < 2 or len(parts) > 3: raise SystemExit(f"[!] item sin 'clave:pose_id[:stage]': '{chunk}'") char, pose = parts[0], parts[1] stage = parts[2] if len(parts) == 3 else None out.append((char, pose, stage)) return out def main(): p = argparse.ArgumentParser(description="Regenera imagenes sueltas de un batch") p.add_argument("--run", required=True, help="carpeta del batch (BATCH_...)") p.add_argument("--items", required=True, help="clave:pose_id separados por coma " "(ej: 'ChamBronya...:pb0025,ASueStorm_character:pb0003')") p.add_argument("--neg-extra", default=None, help="tags extra para el NEGATIVE de estas imagenes " "(p.ej. 'backless')") p.add_argument("--seed", type=int, default=None, help="semilla fija (default: aleatoria por imagen)") p.add_argument("--budget", type=int, default=4) p.add_argument("--dry-run", action="store_true") args = p.parse_args() run_dir = Path(args.run) plan_path = run_dir / "plan.json" all_path = run_dir / "manifest_all.json" if not plan_path.exists() or not all_path.exists(): raise SystemExit(f"[!] {run_dir} no parece un batch (falta plan.json " "o manifest_all.json)") plan_doc = json.loads(plan_path.read_text(encoding="utf-8")) all_records = json.loads(all_path.read_text(encoding="utf-8")) bank = pose_bank.load_bank() avoid = pose_bank.load_avoid() cameras = pb.load_cameras() pools = {lv: pb.load_pool_lines(lv) for lv in LEVELS} B = PromptBuilder(str(BASE_DIR)) neg_base = B.negative_prompt if args.neg_extra: neg_base = f"{neg_base}, {args.neg_extra}" if neg_base else args.neg_extra items = parse_items(args.items) rng = random.Random(0) if not args.dry_run and not pb.forge_available(): raise SystemExit(f"[!] Forge NO responde en {pb.FORGE}. " "Arranca webui-user.bat --api") for char_key, pose_id, stage in items: item = next((x for x in plan_doc["plan"] if x["char"] == char_key), None) if not item: raise SystemExit(f"[!] '{char_key}' no esta en el plan") rec = next((r for r in all_records if (r.get("char") or r["char_dir"]) == char_key and r["pose_id"] == pose_id and (stage is None or r.get("stage") == stage)), None) if not rec: raise SystemExit(f"[!] '{char_key}' no tiene la pose '{pose_id}'" + (f" en el stage '{stage}'" if stage else "")) # la carpeta puede no ser la clave (Reze lleva `\(`): el manifiesto y el # plan guardan la carpeta real en `char_dir` folder = rec.get("char_dir") or char_key char = B.get_character(char_key) if not char: raise SystemExit(f"[!] '{char_key}' no esta en personajes.txt") nivel = rec["nivel"] quotas = {lv: (1 if lv == nivel else 0) for lv in LEVELS} variants = {k: v for k, v in (rec.get("variants") or {}).items() if v} theme = rec["theme"] # Lote de 5 stages: hay que reconstruir CON el stage; sin el, # `compose_theme_jobs` cae al path legacy y el outfit (y por tanto el # prompt) no es el del lote. extra = {} if rec.get("stage"): extra["stage_quotas"] = {rec["stage"]: 1} extra["allow_legacy_outfit"] = True jobs = pb.compose_theme_jobs( [theme], 1, quotas, "fixed", variants, rec["dress"], 0.0, bank, avoid, pools, cameras, rng, max_variants=1, bank_ids=[pose_id], **extra) if len(jobs) != 1: raise SystemExit(f"[!] {char_key}/{pose_id}: salieron {len(jobs)} jobs") job = jobs[0] # la camara original manda: con 1 solo job el indice cambia y el pool # asignaria otro angulo job.camera = rec["camera"] job.seed = args.seed prompt = pb.build_run_prompt(B.master_prompt, char["base"], job) print("=" * 90) print(f"{char_key} | {nivel} | {pose_id} | tema {theme} | cam {job.camera}") print(f" pose nueva: {job.pose_line}") print(f" outfit: {job.outfit_key} -> {job.outfit_line or '-'}") print(f" variants: {job.variant_keys}") if args.dry_run: print(" prompt:") print(" ", prompt.replace("\n", " | ")) continue char_dir = run_dir / folder img_dir = char_dir / "img" tmp_dir = char_dir / "_regen_tmp" if tmp_dir.exists(): shutil.rmtree(tmp_dir) records, _ = pb.generate([job], char["base"], B.master_prompt, neg_base, tmp_dir, args.seed, args.budget, char=char_key) if not records: print(f" [!] FALLO: no se genero nada para {char_key}/{pose_id}") shutil.rmtree(tmp_dir, ignore_errors=True) continue new_rec = records[0] new_name = Path(new_rec["file"]).name # conservar el prefijo de orden original (01/02/03) para no romper el # orden del lote old_name = Path(rec["file"]).name prefix = old_name.split("_")[0] new_name = f"{prefix}_{new_name.split('_', 1)[1]}" dest = img_dir / new_name if dest.exists(): raise SystemExit(f"[!] ya existe {dest}; borra o mueve antes") shutil.move(str(tmp_dir / Path(new_rec["file"]).name), str(dest)) shutil.rmtree(tmp_dir, ignore_errors=True) old_png = img_dir / old_name if old_png.exists(): hist = char_dir / "_reemplazadas" hist.mkdir(parents=True, exist_ok=True) shutil.move(str(old_png), str(hist / old_name)) new_rec = dict(new_rec) new_rec["file"] = f"img/{new_name}" # --- manifests --- char_man = char_dir / "pack_manifest.json" char_recs = json.loads(char_man.read_text(encoding="utf-8")) char_recs = [new_rec if (r.get("pose_id") == pose_id and r.get("stage") == rec.get("stage")) else r for r in char_recs] char_man.write_text(json.dumps(char_recs, ensure_ascii=False, indent=1), encoding="utf-8") all_records = [ ({**new_rec, "char_dir": folder, "file": f"{folder}/{new_rec['file']}"} if (r["char_dir"] == folder and r["pose_id"] == pose_id and r.get("stage") == rec.get("stage")) else r) for r in all_records] all_path.write_text(json.dumps(all_records, ensure_ascii=False, indent=1), encoding="utf-8") for j in item["jobs"]: if (j["pose_id"] == pose_id and j.get("stage") == rec.get("stage")): j.update({"nivel": new_rec["nivel"], "camera": new_rec["camera"], "dress": new_rec["dress"], "seed": new_rec["seed"], "file": new_rec["file"]}) plan_doc["plan"] = plan_doc["plan"] plan_path.write_text(json.dumps(plan_doc, ensure_ascii=False, indent=1), encoding="utf-8") print(f" [OK] {old_name} -> {new_name} (seed {new_rec['seed']})") if not args.dry_run: print(f"\n[OK] {len(items)} imagen(es) regeneradas en {run_dir}") if __name__ == "__main__": main()