feat: wire vision/OCR results back into asset records
- 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
This commit is contained in:
@@ -0,0 +1,342 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user