"""Catalog module: JSON-based wardrobe storage with search. The catalog is a simple JSON file that stores garment entries. Each garment has a unique ID and structured attributes extracted by the vision pipeline. No vector DB, no embeddings. In-memory filtering over a list of dicts. A wardrobe will never exceed a few hundred items. """ import json import logging from pathlib import Path from .storage import save_image, get_image_path, delete_image logger = logging.getLogger(__name__) CATALOG_PATH = Path(__file__).resolve().parent.parent / "data" / "catalog.json" def load_catalog() -> list[dict]: """Load the catalog from disk. Returns empty list if file doesn't exist.""" if not CATALOG_PATH.exists(): logger.debug("Catalog file not found: %s", CATALOG_PATH) return [] try: with open(CATALOG_PATH, "r", encoding="utf-8") as f: catalog = json.load(f) logger.debug("Loaded catalog: %d garments from %s", len(catalog), CATALOG_PATH) return catalog except (json.JSONDecodeError, IOError) as e: logger.error("Failed to load catalog: %s", e) return [] def save_catalog(garments: list[dict]) -> None: """Write the full catalog to disk.""" CATALOG_PATH.parent.mkdir(parents=True, exist_ok=True) with open(CATALOG_PATH, "w", encoding="utf-8") as f: json.dump(garments, f, indent=2, ensure_ascii=False) logger.info("Catalog saved: %d garments", len(garments)) def _next_id(catalog: list[dict]) -> int: """Get the next available garment ID number.""" if not catalog: return 1 existing_ids = [] for g in catalog: gid = g.get("id", "") if isinstance(gid, str) and gid.startswith("garment_"): try: existing_ids.append(int(gid.split("_")[1])) except (ValueError, IndexError): pass return max(existing_ids, default=0) + 1 def add_garments( garments_with_images: list[tuple[dict, bytes | None]], ) -> list[dict]: """Add new garments to the catalog. Assigns IDs and persists images. Accepts a list of (garment_dict, image_bytes) tuples. Each garment gets its own individual image saved. Returns the garments with their assigned IDs. """ catalog = load_catalog() next_num = _next_id(catalog) added = [] for i, (garment, img_bytes) in enumerate(garments_with_images): garment_id = f"garment_{next_num + i:03d}" garment["id"] = garment_id if img_bytes: filename = save_image(garment_id, img_bytes) garment["image_ref"] = filename catalog.append(garment) added.append(garment) save_catalog(catalog) return added def search( garment_type: str | None = None, color: str | None = None, season: str | None = None, formality: str | None = None, ) -> list[dict]: """Filter catalog by attributes. All filters are AND-combined.""" catalog = load_catalog() results = catalog if garment_type: garment_type = garment_type.lower().strip() results = [g for g in results if garment_type in g.get("type", "")] if color: color = color.lower().strip() results = [g for g in results if color in g.get("color", "")] if season: season = season.lower().strip() results = [g for g in results if g.get("season", "") in (season, "all")] if formality: formality = formality.lower().strip() results = [g for g in results if formality in g.get("formality", "")] return results def get_catalog_summary() -> str: """Format the catalog as a text block for LLM context injection.""" catalog = load_catalog() if not catalog: return "The wardrobe is empty. No garments have been cataloged yet." lines = [f"Wardrobe catalog ({len(catalog)} garments):\n"] for g in catalog: gid = g.get("id", "?") gtype = g.get("type", "?") color = g.get("color", "?") material = g.get("material", "?") pattern = g.get("pattern", "?") season = g.get("season", "?") formality = g.get("formality", "?") description = g.get("description", "") line = ( f"- [{gid}] {color} {gtype} ({material}, {pattern}) " f"| season: {season} | style: {formality}" ) if description: line += f"\n → {description}" lines.append(line) return "\n".join(lines) def get_catalog_stats() -> dict: """Return aggregate statistics about the catalog.""" catalog = load_catalog() if not catalog: return {"total": 0} types: dict[str, int] = {} colors: dict[str, int] = {} seasons: dict[str, int] = {} for g in catalog: t = g.get("type", "unknown") types[t] = types.get(t, 0) + 1 c = g.get("color", "unknown") colors[c] = colors.get(c, 0) + 1 s = g.get("season", "unknown") seasons[s] = seasons.get(s, 0) + 1 return { "total": len(catalog), "by_type": dict(sorted(types.items(), key=lambda x: -x[1])), "by_color": dict(sorted(colors.items(), key=lambda x: -x[1])), "by_season": dict(sorted(seasons.items(), key=lambda x: -x[1])), } def get_garment_image_path(garment_id: str) -> str | None: """Resolve the local filesystem path for a garment's image.""" return get_image_path(garment_id) def clear_catalog() -> None: """Remove all garments from the catalog and their images.""" catalog = load_catalog() for g in catalog: gid = g.get("id", "") if gid: delete_image(gid) save_catalog([]) logger.info("Catalog cleared")