shawn
17b870e4cc
- Auto-save photo_path to matched assets when photo uploaded via /api/ocr - Auto-extract and save serial_number from OCR text to blank-serial matched assets - Add _extract_serial_from_text helper for serial number pattern matching - Create scripts/backfill_vision_photos.py for batch-processing unlinked photos - Add docs/OCR_PIPELINE.md documenting the full OCR/vision pipeline - Fix test DB schema: add connect_id, equipment_id, barcode, customer_name - Fix OCR test assertions to match actual endpoint behavior
343 lines
12 KiB
Python
343 lines
12 KiB
Python
#!/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()
|