#!/usr/bin/env python3 """ Cross-reference extracted photo IDs against assets DB. Processes all photos in uploads/photos/, runs vision/OCR text extraction, matches identifiers against assets.machine_id / serial_number / connect_id, updates matched assets with photo_path, and reports unmatched IDs. Usage: python3 scripts/crossref_photos.py # process all photos python3 scripts/crossref_photos.py --photo # single photo python3 scripts/crossref_photos.py --text "S/N: 1234" # from text only """ import argparse import json import os import re import sqlite3 import sys from datetime import datetime from pathlib import 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") PHOTOS_DIR = str(Path(__file__).resolve().parent.parent / "uploads" / "photos") def process_photo(photo_path: str, db_path: str = DB_PATH) -> dict: """Process a single photo: extract text, match DB, update photo_path.""" fname = os.path.basename(photo_path) result = { "photo": fname, "photo_path": f"/uploads/photos/{fname}", "size_bytes": os.path.getsize(photo_path), "vision_text": "", "extracted_ids": [], "matches": [], "updated": False, } # Try Tesseract OCR try: import pytesseract from PIL import Image as PILImage img = PILImage.open(photo_path) img_gray = img.convert("L") text = pytesseract.image_to_string(img_gray, config="--psm 6") text = text.strip() clean_chars = sum(1 for c in text if c.isalnum() or c in " \n/-.") if text and clean_chars >= 4: result["vision_text"] = text except ImportError: pass if not result["vision_text"]: result["vision_source"] = "none" return result result["vision_source"] = "tesseract" # Extract identifiers from text identifiers = [] lines = result["vision_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(norm) tokens = re.findall(r"[A-Za-z0-9]{4,}", line) for token in tokens: norm = normalize_identifier(token) if norm and len(norm) >= 4 and norm not in identifiers: identifiers.append(norm) result["extracted_ids"] = identifiers # Match against DB conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row seen_ids = set() for norm_id in identifiers: assets = find_asset_by_normalized_id(db_path, norm_id) for a in assets: if a["id"] not in seen_ids: seen_ids.add(a["id"]) result["matches"].append(a) # Update photo_path if empty existing = conn.execute( "SELECT photo_path FROM assets WHERE id = ?", (a["id"],) ).fetchone() if existing and not existing["photo_path"]: conn.execute( "UPDATE assets SET photo_path = ?, updated_at = datetime('now') WHERE id = ?", (result["photo_path"], a["id"]), ) result["updated"] = True conn.commit() conn.close() return result def main(): parser = argparse.ArgumentParser( description="Cross-reference photo IDs against assets DB" ) parser.add_argument( "--photo", "-p", help="Process a single photo by filename" ) parser.add_argument( "--text", "-t", help="Process raw text directly (skip OCR)" ) parser.add_argument( "--db", default=DB_PATH, help=f"Database path (default: {DB_PATH})", ) args = parser.parse_args() results = [] if args.text: # Process from text directly print(f"\n{'=' * 70}") print(f"PROCESSING: supplied text ({len(args.text)} chars)") print(f"{'=' * 70}") print(f" Text: {args.text[:500]}") identifiers = [] norm = normalize_identifier(args.text) if norm and len(norm) >= 4: identifiers.append(norm) tokens = re.findall(r"[A-Za-z0-9]{4,}", args.text) for token in tokens: n = normalize_identifier(token) if n and len(n) >= 4 and n not in identifiers: identifiers.append(n) print(f"\n Normalized IDs: {identifiers}") conn = sqlite3.connect(args.db) conn.row_factory = sqlite3.Row for norm_id in identifiers: assets = find_asset_by_normalized_id(args.db, norm_id) if assets: for a in assets: print(f"\n ✓ MATCH → Asset #{a['id']}") print(f" Machine ID: {a['machine_id']}") print(f" Name: {a['name'][:60]}") else: print(f"\n ✗ No match for '{norm_id}'") conn.close() return if args.photo: photo_path = os.path.join(PHOTOS_DIR, args.photo) if not os.path.exists(photo_path): print(f"ERROR: {photo_path} not found") sys.exit(1) results.append(process_photo(photo_path, args.db)) else: # Process all photos for fname in sorted(os.listdir(PHOTOS_DIR)): photo_path = os.path.join(PHOTOS_DIR, fname) if os.path.isfile(photo_path): results.append(process_photo(photo_path, args.db)) # Print summary matched = [r for r in results if r["matches"]] unmatched = [r for r in results if not r["matches"] and r["vision_text"]] no_text = [r for r in results if not r["vision_text"]] print(f"\n{'=' * 70}") print(f"CROSS-REFERENCE SUMMARY ({datetime.now().isoformat()})") print(f"{'=' * 70}") print(f" Photos scanned: {len(results)}") print(f" Photos with text: {len(results) - len(no_text)}") print(f" Matches found: {len(matched)}") print(f" Unmatched: {len(unmatched)}") print(f" No text (small/thumbs): {len(no_text)}") for r in matched: for a in r["matches"]: status = "UPDATED" if r["updated"] else "already set" print( f"\n ✓ {r['photo']} → Asset #{a['id']} " f"(machine_id={a['machine_id']}) [{status}]" ) for r in unmatched: print(f"\n ✗ {r['photo']} — no DB match") print(f" IDs: {r['extracted_ids'][:5]}") print(f" Text preview: {r['vision_text'][:100]}") print() if __name__ == "__main__": main()