canteen-asset-tracker/scripts/crossref_photos.py

#!/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 <name>  # 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()