"""Rule-based PO parser — fallback when the 7B model is unavailable or returns bad JSON.""" import re from typing import Optional from data import SAMPLE_CATALOG # Common Indonesian / informal aliases → canonical catalog names PRODUCT_ALIASES: dict[str, str] = { "kucing hologram": "Holographic Cat Sticker", "kucing holo": "Holographic Cat Sticker", "holographic cat": "Holographic Cat Sticker", "stiker kucing hologram": "Holographic Cat Sticker", "sakura die-cut": "Sakura Die-Cut Sticker", "stiker sakura": "Sakura Die-Cut Sticker", "sakura": "Sakura Die-Cut Sticker", "cherry blossom": "Cherry Blossom Sticker", "rainbow unicorn": "Rainbow Unicorn Sticker", "unicorn": "Rainbow Unicorn Sticker", "boba tea": "Kawaii Boba Tea Sticker", "boba": "Kawaii Boba Tea Sticker", "sushi kawaii": "Kawaii Sushi Sticker", "sushi": "Kawaii Sushi Sticker", "ramen kawaii": "Kawaii Ramen Sticker", "ramen": "Kawaii Ramen Sticker", "floral wreath": "Floral Wreath Sticker", "butterfly garden": "Butterfly Garden Sticker", "vintage bicycle": "Vintage Bicycle Sticker", "coffee cup": "Coffee Cup Sticker", "galaxy star": "Galaxy Star Sticker", "monstera leaf": "Monstera Leaf Sticker", "monstera": "Monstera Leaf Sticker", "batik pattern": "Batik Pattern Sticker", "batik": "Batik Pattern Sticker", "koi fish": "Koi Fish Sticker", "koi": "Koi Fish Sticker", "borobudur": "Borobudur Temple Sticker", "wayang": "Wayang Shadow Sticker", "gamelan": "Gamelan Sticker", "penari jawa": "Javanese Dancer Sticker", "javanese dancer": "Javanese Dancer Sticker", "mushroom forest": "Mushroom Forest Sticker", "space astronaut": "Space Astronaut Sticker", "mountain landscape": "Mountain Landscape Sticker", "ocean wave": "Ocean Wave Sticker", "cactus succulent": "Cactus Succulent Sticker", "cute dog": "Cute Dog Sticker", "cute rabbit": "Cute Rabbit Sticker", "panda bear": "Panda Bear Sticker", "retro cassette": "Retro Cassette Sticker", "tropical bird": "Tropical Bird Sticker", "komodo dragon": "Komodo Dragon Sticker", } _CATALOG_LOWER = {p.lower(): p for p in SAMPLE_CATALOG} def _normalize_product(raw: str) -> str: cleaned = re.sub(r"^[\-\*\d\.\)\s]+", "", raw.strip()) cleaned = re.sub(r"\s*(pcs|pc|lembar|sheet|sheets|buah|unit)\.?\s*$", "", cleaned, flags=re.I) cleaned = cleaned.strip(" ,;:") if not cleaned: return "" key = cleaned.lower() if key in _CATALOG_LOWER: return _CATALOG_LOWER[key] for alias, canonical in sorted(PRODUCT_ALIASES.items(), key=lambda x: -len(x[0])): if alias in key or key in alias: return canonical for cat_lower, canonical in _CATALOG_LOWER.items(): if cat_lower in key or key in cat_lower: return canonical return cleaned.title() def _extract_qty(text: str) -> int: m = re.search(r"(\d+)\s*(?:pcs|pc|lembar|sheet|sheets|buah|unit)?\.?\s*$", text, re.I) if m: return int(m.group(1)) m = re.search(r"x\s*(\d+)\s*$", text, re.I) if m: return int(m.group(1)) m = re.search(r"\b(\d+)\b", text) return int(m.group(1)) if m else 0 def _parse_line(line: str) -> Optional[dict]: line = line.strip() if not line or len(line) < 3: return None if re.match(r"^(item|product|qty|quantity|no\.?|#)\b", line, re.I): return None if re.match(r"^(pesanan|po\s*#|monthly|tolong|please|note|hai|makasih|thanks)", line, re.I): return None # "Product x25" or "Product | 25" m = re.match(r"^(.+?)\s*[x×]\s*(\d+)\s*$", line, re.I) if m: product = _normalize_product(m.group(1)) return {"product": product, "quantity": int(m.group(2)), "notes": ""} if product else None m = re.match(r"^(.+?)\s*\|\s*(\d+)\s*$", line) if m: product = _normalize_product(m.group(1)) return {"product": product, "quantity": int(m.group(2)), "notes": ""} if product else None # "name 20 pcs" or "name, 20" m = re.match(r"^(.+?)[,\s]+(\d+)\s*(?:pcs|pc)?\.?\s*$", line, re.I) if m: product = _normalize_product(m.group(1)) return {"product": product, "quantity": int(m.group(2)), "notes": ""} if product else None # Comma-separated inline: "borobudur 20, wayang 15" if "," not in line and re.search(r"\d", line): qty = _extract_qty(line) if qty > 0: product_part = re.sub(r"\d+.*$", "", line).strip(" ,-") product = _normalize_product(product_part) if product: return {"product": product, "quantity": qty, "notes": ""} return None def _parse_comma_list(text: str) -> list[dict]: items: list[dict] = [] for chunk in re.split(r"[,;\n]+", text): chunk = chunk.strip() if not chunk: continue m = re.match(r"^(.+?)\s+(\d+)\s*$", chunk) if m: product = _normalize_product(m.group(1)) if product: items.append({"product": product, "quantity": int(m.group(2)), "notes": ""}) return items def parse_po_fallback(po_text: str) -> dict: """Extract items from messy PO text without an LLM.""" items: list[dict] = [] seen: set[tuple[str, int]] = set() store_notes: list[str] = [] for line in po_text.splitlines(): parsed = _parse_line(line) if parsed and parsed["quantity"] > 0: key = (parsed["product"], parsed["quantity"]) if key not in seen: seen.add(key) items.append(parsed) elif line.strip() and re.match(r"^(note|catatan|tolong|please|ship|kirim)", line.strip(), re.I): store_notes.append(line.strip()) if len(items) < 2: items.extend(_parse_comma_list(po_text)) # Deduplicate by product, sum quantities merged: dict[str, dict] = {} for it in items: name = it["product"] if name in merged: merged[name]["quantity"] += it["quantity"] else: merged[name] = dict(it) return { "items": list(merged.values()), "store_notes": " ".join(store_notes)[:200], }