#!/usr/bin/env python3 """ Backfill: OCR all unlinked photos and update matching assets. Scans uploads/photos/ for image files whose photo_path is not set on any asset in the database. For each unlinked photo, runs Tesseract OCR (and Ollama vision if available) to extract identifiers, cross-references against the assets DB, and updates matching assets with: - photo_path pointing to the photo file - serial_number if extracted from OCR and currently blank Usage: python3 scripts/backfill_vision_photos.py # dry-run (report only) python3 scripts/backfill_vision_photos.py --apply # write changes python3 scripts/backfill_vision_photos.py --apply --force # overwrite existing photo_path """ import argparse import os import re import sqlite3 import sys from pathlib import Path # Add project root to path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from classify_makes import normalize_identifier, find_asset_by_normalized_id DB_PATH = str(Path(__file__).resolve().parent.parent / "assets.db") UPLOADS_DIR = Path(__file__).resolve().parent.parent / "uploads" / "photos" PHOTO_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".bmp", ".dng", ".tiff", ".tif"} # ── OCR helpers (imported from server but keeping script self-contained) ──── def _has_ollama() -> bool: """Check if Ollama vision model is available.""" try: import json, urllib.request req = urllib.request.Request( "http://127.0.0.1:11434/api/generate", data=json.dumps({"model": "qwen2.5vl:3b", "prompt": "ping", "stream": False}).encode(), headers={"Content-Type": "application/json"}, ) urllib.request.urlopen(req, timeout=5) return True except Exception: return False def _has_tesseract() -> bool: """Check if Tesseract is available.""" try: import pytesseract pytesseract.get_tesseract_version() return True except Exception: return False def ocr_via_ollama(image_path: Path) -> str: """Run Ollama vision model on image, return extracted text.""" import json, urllib.request, base64 from PIL import Image as PILImage img = PILImage.open(image_path) img.thumbnail((640, 480)) data_b64 = base64.b64encode(img.tobytes()).decode() with open(image_path, "rb") as f: b64_data = base64.b64encode(f.read()).decode() payload = json.dumps({ "model": "qwen2.5vl:3b", "prompt": "Extract all text, numbers, serial numbers, and IDs visible " "on this sticker or label. Return ONLY the raw text content, " "one item per line. Do not describe the image.", "images": [b64_data], "stream": False, }).encode() req = urllib.request.Request( "http://127.0.0.1:11434/api/generate", data=payload, headers={"Content-Type": "application/json"}, ) resp = urllib.request.urlopen(req, timeout=120) result = json.loads(resp.read().decode()) return result.get("response", "").strip() def ocr_via_tesseract(image_path: Path) -> str: """Run Tesseract OCR on image, return extracted text.""" import pytesseract from PIL import Image as PILImage img = PILImage.open(image_path) img_gray = img.convert("L") text = pytesseract.image_to_string(img_gray, config="--psm 6") return text.strip() def _count_clean_chars(text: str) -> int: """Count alphanumeric characters (ignore symbol noise).""" return sum(1 for c in text if c.isalnum() or c in ' \n/-.') def _extract_serial_from_text(text: str) -> str | None: """Try to extract a plausible serial number from OCR text.""" if not text: return None patterns = [ r'(?:S/N|SN|SERIAL\s*NO|SERIAL|SERIAL\s*#)\s*[:=#]?\s*([A-Za-z0-9.\-/]{6,})', r'(?:EQUIPMENT\s*ID|EQ\s*ID|ASSET\s*ID)\s*[:=]?\s*([A-Za-z0-9.\-/]{6,})', r'(?:MODEL\s*NO|MODEL\s*#|PART\s*NO|P/N)\s*[:=]?\s*([A-Za-z0-9.\-/]{6,})', ] for pat in patterns: m = re.search(pat, text, re.IGNORECASE) if m: val = m.group(1).strip().rstrip('.') if re.match(r'^\d{4,}$', val.replace('-', '').replace('.', '')): continue # Probably a Connect ID, not a serial clean = sum(1 for c in val if c.isalnum()) if clean >= 6: return val return None def _find_identifiers(text: str) -> list: """Extract all plausible identifiers from OCR text.""" identifiers = [] lines = text.split('\n') for line in lines: line = line.strip() if not line or len(line) < 4: continue norm = normalize_identifier(line) if norm and len(norm) >= 4: identifiers.append({"raw": line, "normalized": norm, "type": "full_line"}) tokens = re.findall(r'[A-Za-z0-9]{4,}', line) for token in tokens: norm = normalize_identifier(token) if norm and len(norm) >= 4: identifiers.append({"raw": token, "normalized": norm, "type": "token"}) return identifiers def _match_identifiers(identifiers: list, db_path: str) -> list: """Match identifiers against DB, return deduplicated asset matches.""" results = [] seen_ids = set() for ident in identifiers: assets = find_asset_by_normalized_id(db_path, ident["normalized"]) for a in assets: if a["id"] not in seen_ids: seen_ids.add(a["id"]) results.append({ "asset": a, "matched_on": ident["normalized"], "source_text": ident["raw"], }) return results def _get_unlinked_photos(db_path: str) -> list: """Get list of photos in uploads/photos/ not linked to any asset.""" # Get all photo_paths already in DB conn = sqlite3.connect(db_path) linked = {row[0] for row in conn.execute( "SELECT photo_path FROM assets WHERE photo_path IS NOT NULL AND photo_path != ''" ).fetchall()} conn.close() unlinked = [] for f in sorted(UPLOADS_DIR.iterdir()): if f.suffix.lower() not in PHOTO_EXTS: continue rel_path = f"/uploads/photos/{f.name}" if rel_path not in linked: unlinked.append({"path": f, "rel": rel_path}) return unlinked def main(): parser = argparse.ArgumentParser( description="Backfill: OCR all unlinked photos and update matching assets" ) parser.add_argument("--apply", action="store_true", help="Write changes to DB (default: dry-run)") parser.add_argument("--force", action="store_true", help="Overwrite existing photo_path on assets") parser.add_argument("--db", default=DB_PATH, help=f"Database path (default: {DB_PATH})") parser.add_argument("--photos-dir", default=str(UPLOADS_DIR), help=f"Photos directory (default: {UPLOADS_DIR})") args = parser.parse_args() db_path = args.db photos_dir = Path(args.photos_dir) print(f"🔍 Scanning {photos_dir} for unlinked photos...") unlinked = _get_unlinked_photos(db_path) print(f" Found {len(unlinked)} unlinked photo(s) out of " f"{len(list(photos_dir.glob('*')))} total files in directory.\n") if not unlinked: print("✅ All photos are already linked to assets. Nothing to do.") return has_ollama = _has_ollama() has_tess = _has_tesseract() ollama_status = "✓ connected" if has_ollama else "✗ not available" tess_status = "✓ installed" if has_tess else "✗ not available" print(f" Ollama vision: {ollama_status}") print(f" Tesseract: {tess_status}") if not has_ollama and not has_tess: print("⚠️ No OCR service available. Install Tesseract or start Ollama.") return total_photos = len(unlinked) matched_count = 0 updated_photo_count = 0 updated_serial_count = 0 for i, photo in enumerate(unlinked, 1): img_path = photo["path"] rel_path = photo["rel"] print(f"\n{'='*60}") print(f"[{i}/{total_photos}] {img_path.name}") print(f"{'='*60}") # Step 1: OCR text = "" ocr_source = "none" if has_ollama: print(" [vision] Running Ollama...") try: text = ocr_via_ollama(img_path) if text and _count_clean_chars(text) >= 10: ocr_source = "ollama" print(f" ✓ Ollama extracted {len(text)} chars") except Exception as e: print(f" ⚠ Ollama error: {e}") if ocr_source == "none" and has_tess: print(" [ocr] Running Tesseract...") try: text = ocr_via_tesseract(img_path) if text and _count_clean_chars(text) >= 10: ocr_source = "tesseract" print(f" ✓ Tesseract extracted {len(text)} chars") except Exception as e: print(f" ⚠ Tesseract error: {e}") if not text or _count_clean_chars(text) < 10: print(" ⚠ Could not extract sufficient text from this image.") continue print(f" Text preview: {text[:150]}") # Step 2: Extract identifiers identifiers = _find_identifiers(text) if not identifiers: print(" ⚠ No identifiers found in OCR text.") continue # Step 3: Match against DB matches = _match_identifiers(identifiers, db_path) if not matches: print(" ⚠ No DB matches found for extracted identifiers.") continue matched_count += 1 print(f" ✓ Matched {len(matches)} asset(s):") for m in matches: a = m["asset"] print(f" ├─ Asset #{a['id']}: {a['name'][:50]}") print(f" ├─ Machine ID: {a['machine_id']}") print(f" ├─ Serial: {a['serial_number'][:40] or '(blank)'}") print(f" └─ Matched on: {m['matched_on']}") if not args.apply: print(" ℹ️ Dry-run — use --apply to write changes.") continue # Step 4: Apply changes conn = sqlite3.connect(db_path) try: for m in matches: a = m["asset"] # Update photo_path if blank (or forced) needs_photo = args.force or not a.get("photo_path") if needs_photo: conn.execute( "UPDATE assets SET photo_path = ?, updated_at = datetime('now') WHERE id = ?", (rel_path, a["id"]), ) updated_photo_count += 1 print(f" ✓ Set photo_path → {rel_path}") # Update serial_number if blank if not a.get("serial_number"): serial = _extract_serial_from_text(text) if serial: conn.execute( "UPDATE assets SET serial_number = ?, updated_at = datetime('now') WHERE id = ?", (serial, a["id"]), ) updated_serial_count += 1 print(f" ✓ Set serial_number → {serial}") conn.commit() except Exception as e: print(f" ✗ DB error: {e}") conn.rollback() finally: conn.close() # Summary print(f"\n{'='*60}") print(f"SUMMARY") print(f"{'='*60}") print(f" Total unlinked photos: {total_photos}") print(f" Photos with DB matches: {matched_count}") if args.apply: print(f" Photo paths updated: {updated_photo_count}") print(f" Serial numbers updated: {updated_serial_count}") else: print(f" (dry-run — no changes written)") print(f" Re-run with --apply to write changes.") print() if __name__ == "__main__": main()