shawn/exif-test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
a8b1e694b0293c0b2879a0061c75712da430155a
exif-test/server.py
T
shawn a8b1e694b0 feat: persistence, rerun OCR, Google Gemini, sticker mode, manual entry 
- SQLite DB (photos.db) — persists all processed photo records
- Dedup by SHA256 hash — same file upload returns duplicate: true
- /api/photos — list previously processed photos
- /api/photos/{id} — get single record
- /api/photos/{id}/file — serve saved image
- /api/photos/{id}/reprocess — re-run OCR with different engine/model
- Google Gemini OCR engine (gemini-2.5-flash, free tier) alongside
  OpenCode Go LLM and Tesseract
- Sticker mode — specialized LLM/Google prompt for green/orange/yellow
  equipment stickers with 2D barcode + machine ID
- Manual machine ID entry — when GPS exists but OCR fails, show text
  input for manual lookup
- Frontend: Previous Photos section, Re-run OCR per photo, duplicate
  badges, engine dropdown, sticker toggle
2026-05-25 17:42:10 -04:00

967 lines
33 KiB
Python
Raw Blame History

"""EXIF + OCR test backend — validate that GPS survives upload pipeline."""
import hashlib, io, json, os, re, uuid, sqlite3, urllib.request
from pathlib import Path
from fastapi import FastAPI, File, Form, HTTPException, Query, UploadFile
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from PIL import Image as PILImage
# Optional HEIC/HEIF support for iPhone photos
try:
from pillow_heif import register_heif_opener
register_heif_opener()
HAS_HEIF = True
except ImportError:
HAS_HEIF = False
try:
import pytesseract
HAS_TESSERACT = True
except ImportError:
HAS_TESSERACT = False
# === LLM OCR via OpenCode Go ===
OPENCODE_GO_KEY = os.environ.get("OPENCODE_GO_API_KEY", "")
OPENCODE_GO_BASE = os.environ.get("OPENCODE_GO_BASE_URL", "https://opencode.ai/zen/go/v1")
LLM_OCR_MODEL = os.environ.get("LLM_OCR_MODEL", "mimo-v2-omni")
# === Google Gemini OCR (free tier) ===
GOOGLE_API_KEY = os.environ.get("GOOGLE_API_KEY", "")
GOOGLE_API_BASE = os.environ.get("GOOGLE_API_BASE_URL", "https://generativelanguage.googleapis.com/v1beta")
GOOGLE_OCR_MODEL = os.environ.get("GOOGLE_OCR_MODEL", "gemini-2.5-flash")
UPLOADS = Path(__file__).parent / "uploads"
UPLOADS.mkdir(exist_ok=True)
PHOTOS_DB = Path(__file__).parent / "photos.db"
CANTEEN_DB = Path(__file__).parent.parent / "canteen-asset-tracker" / "assets.db"
# Max images per batch API call
BATCH_SIZE_LIMIT = 20
# Downscale images to this max dimension before LLM OCR
LLM_IMAGE_MAX_DIM = 1600
# ---------------------------------------------------------------------------
# Sticker-mode prompts
# ---------------------------------------------------------------------------
DEFAULT_OCR_PROMPT = (
"Read ALL text and numbers visible in this photo. "
"Return the exact text shown, nothing else."
)
STICKER_OCR_PROMPT = (
"This photo shows a colored equipment sticker (green, orange, or yellow background) "
"with a 2D barcode and a machine ID number printed below the barcode. "
"Read ONLY the machine ID number that appears below the barcode. "
"It is typically a 5-digit number followed by a dash and 6 more digits "
"(e.g., 12345-678901). Return ONLY the machine ID number, nothing else."
)
DEFAULT_BATCH_PROMPT = (
"I have {n} photos. For EACH photo, read ALL visible text and numbers.\n"
'Respond with a JSON array of objects, one per photo in order:\n'
'[{{"i":0,"text":"all text and digits found","digits":"e.g. 12345-678901 or null if none"}}, ...]\n'
'Return ONLY the JSON array, no markdown, no explanation.'
)
STICKER_BATCH_PROMPT = (
"I have {n} photos of colored equipment stickers (green, orange, or yellow). "
"Each sticker has a 2D barcode and a machine ID printed below it.\n"
'Respond with a JSON array of objects, one per photo in order:\n'
'[{{"i":0,"sticker_color":"green/orange/yellow/unknown","machine_id":"12345-678901 or null if not found"}}, ...]\n'
'Return ONLY the JSON array, no markdown, no explanation.'
)
# ---------------------------------------------------------------------------
# App + DB init
# ---------------------------------------------------------------------------
app = FastAPI(title="EXIF Test")
def _init_photos_db():
"""Create photos table for persistence + dedup."""
conn = sqlite3.connect(str(PHOTOS_DB))
conn.execute("""
CREATE TABLE IF NOT EXISTS photos (
id INTEGER PRIMARY KEY AUTOINCREMENT,
orig_filename TEXT NOT NULL,
file_hash TEXT NOT NULL UNIQUE,
saved_as TEXT NOT NULL,
file_size INTEGER,
exif_json TEXT,
gps_lat REAL,
gps_lng REAL,
ocr_engine TEXT,
ocr_model TEXT,
ocr_raw_text TEXT,
ocr_match_5dash6 TEXT,
ocr_match_5plus TEXT,
machine_id TEXT,
sticker_color TEXT,
has_barcode INTEGER DEFAULT 0,
created_at TEXT DEFAULT (datetime('now'))
)
""")
conn.commit()
conn.close()
_init_photos_db()
# ---------------------------------------------------------------------------
# DB helpers
# ---------------------------------------------------------------------------
def _get_photos_db() -> sqlite3.Connection:
conn = sqlite3.connect(str(PHOTOS_DB))
conn.row_factory = sqlite3.Row
return conn
def _file_hash(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def _save_photo_to_db(
orig_filename: str, file_hash: str, saved_as: str, file_size: int,
exif_result: dict, ocr_result: dict, machine_id: str | None,
) -> int:
"""Insert a photo record. Returns the row id."""
conn = _get_photos_db()
try:
cur = conn.execute(
"""INSERT OR IGNORE INTO photos
(orig_filename, file_hash, saved_as, file_size, exif_json,
gps_lat, gps_lng, ocr_engine, ocr_model, ocr_raw_text,
ocr_match_5dash6, ocr_match_5plus, machine_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
orig_filename,
file_hash,
saved_as,
file_size,
json.dumps(exif_result) if exif_result else None,
exif_result.get("gps", {}).get("lat") if exif_result and exif_result.get("gps") else None,
exif_result.get("gps", {}).get("lng") if exif_result and exif_result.get("gps") else None,
ocr_result.get("engine") if ocr_result else None,
ocr_result.get("llm_model") if ocr_result else None,
ocr_result.get("raw_text") if ocr_result else None,
ocr_result.get("match_5dash6") if ocr_result else None,
ocr_result.get("match_5plus") if ocr_result else None,
machine_id,
),
)
conn.commit()
return cur.lastrowid or 0
except sqlite3.IntegrityError:
return 0
finally:
conn.close()
def _row_to_dict(row: sqlite3.Row) -> dict:
d = dict(row)
d["exif_data"] = json.loads(d.pop("exif_json", "{}") or "{}")
return d
# ---------------------------------------------------------------------------
# Canteen DB helpers
# ---------------------------------------------------------------------------
def _get_canteen_db():
"""Get a read/write connection to the canteen assets database."""
if not CANTEEN_DB.exists():
return None
conn = sqlite3.connect(str(CANTEEN_DB))
conn.row_factory = sqlite3.Row
return conn
def lookup_machine_id(machine_id: str) -> dict | None:
"""Look up an asset by machine_id. Returns asset dict or None."""
conn = _get_canteen_db()
if not conn:
return None
try:
row = conn.execute(
"SELECT id, machine_id, name, category, status, address, building_name, "
"floor, room, latitude, longitude, make, model, description, photo_path "
"FROM assets WHERE machine_id = ?",
(machine_id.strip(),),
).fetchone()
if row:
return {
"id": row["id"],
"machine_id": row["machine_id"],
"name": row["name"],
"category": row["category"],
"status": row["status"],
"address": row["address"],
"building_name": row["building_name"],
"floor": row["floor"],
"room": row["room"],
"latitude": row["latitude"],
"longitude": row["longitude"],
"make": row["make"],
"model": row["model"],
"description": row["description"],
"photo_path": row["photo_path"],
}
finally:
conn.close()
return None
# ---------------------------------------------------------------------------
# EXIF extraction
# ---------------------------------------------------------------------------
def _dms_to_decimal(dms, ref):
"""Convert EXIF DMS tuple to decimal degrees."""
try:
deg, minutes, sec = float(dms[0]), float(dms[1]), float(dms[2])
decimal = deg + minutes / 60.0 + sec / 3600.0
if ref in ("S", "W"):
decimal = -decimal
return round(decimal, 7)
except Exception:
return None
def extract_exif(image_bytes: bytes) -> dict:
"""Pull all useful EXIF fields + GPS from raw image bytes."""
result = {"has_exif": False, "tags": {}, "gps": None}
try:
img = PILImage.open(io.BytesIO(image_bytes))
exif = img.getexif()
if not exif:
return result
for tag_id, value in exif.items():
tag_name = PILImage.ExifTags.TAGS.get(tag_id, f"0x{tag_id:04x}")
if tag_name in ("MakerNote", "UserComment", "PrintImageMatching"):
continue
result["tags"][tag_name] = str(value)[:300]
if result["tags"]:
result["has_exif"] = True
gps_ifd = exif.get_ifd(0x8825)
if gps_ifd:
lat_ref = gps_ifd.get(1, "N")
lat_dms = gps_ifd.get(2)
lng_ref = gps_ifd.get(3, "E")
lng_dms = gps_ifd.get(4)
if lat_dms and lng_dms:
lat = _dms_to_decimal(lat_dms, lat_ref)
lng = _dms_to_decimal(lng_dms, lng_ref)
if lat is not None and lng is not None:
result["gps"] = {
"lat": lat,
"lng": lng,
"lat_ref": str(lat_ref),
"lng_ref": str(lng_ref),
"raw_lat_dms": [float(v) for v in lat_dms],
"raw_lng_dms": [float(v) for v in lng_dms],
}
except Exception as e:
result["error"] = str(e)
return result
# ---------------------------------------------------------------------------
# Tesseract OCR
# ---------------------------------------------------------------------------
def run_ocr(image_bytes: bytes) -> dict:
"""Run Tesseract OCR on the image."""
if not HAS_TESSERACT:
return {"available": False, "text": "", "error": "pytesseract not installed"}
tmp_path = UPLOADS / f"ocr_{uuid.uuid4().hex}.jpg"
tmp_path.write_bytes(image_bytes)
try:
img = PILImage.open(tmp_path)
img_gray = img.convert("L")
text = pytesseract.image_to_string(img_gray, config="--psm 6")
match_5dash = re.search(r"(\d{5})[-\s]*(\d{6,})", text)
match_5plus = re.search(r"(\d{5,})", text)
return {
"available": True,
"raw_text": text.strip()[:500],
"match_5dash6": match_5dash.group(0) if match_5dash else None,
"match_5plus": match_5plus.group(0) if match_5plus else None,
}
finally:
tmp_path.unlink(missing_ok=True)
# ---------------------------------------------------------------------------
# LLM OCR helpers
# ---------------------------------------------------------------------------
def _resize_for_llm(image_bytes: bytes, max_dim: int = LLM_IMAGE_MAX_DIM) -> bytes:
"""Downscale image to max_dim on longest edge to save vision API costs."""
img = PILImage.open(io.BytesIO(image_bytes))
w, h = img.size
if max(w, h) <= max_dim:
return image_bytes # already small enough
ratio = max_dim / max(w, h)
new_size = (int(w * ratio), int(h * ratio))
img_resized = img.resize(new_size, PILImage.LANCZOS)
buf = io.BytesIO()
ext = img.format or "JPEG"
if ext.upper() in ("PNG", "WEBP", "TIFF", "BMP"):
img_resized = img_resized.convert("RGB")
ext = "JPEG"
img_resized.save(buf, format=ext, quality=85)
return buf.getvalue()
def run_ocr_llm(
image_bytes: bytes, model: str | None = None,
sticker_mode: bool = False,
) -> dict:
"""Run OCR via an LLM vision model on OpenCode Go.
Falls back to Tesseract if the API key is missing or the call fails.
Returns the same shape as run_ocr() with an additional 'engine' field.
"""
if not OPENCODE_GO_KEY:
result = run_ocr(image_bytes)
result["engine"] = "tesseract"
result["llm_fallback_reason"] = "no_api_key"
return result
import base64
model = model or LLM_OCR_MODEL
image_bytes = _resize_for_llm(image_bytes)
b64 = base64.b64encode(image_bytes).decode()
prompt = STICKER_OCR_PROMPT if sticker_mode else DEFAULT_OCR_PROMPT
body = json.dumps({
"model": model,
"messages": [{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
]
}],
"max_tokens": 200,
}).encode()
req = urllib.request.Request(
f"{OPENCODE_GO_BASE}/chat/completions",
data=body,
headers={
"Authorization": f"Bearer {OPENCODE_GO_KEY}",
"Content-Type": "application/json",
"User-Agent": "Hermes-Agent/1.0",
},
)
try:
resp = urllib.request.urlopen(req, timeout=60)
result = json.loads(resp.read())
text = result["choices"][0]["message"]["content"].strip()
except Exception as exc:
result = run_ocr(image_bytes)
result["engine"] = "tesseract"
result["llm_fallback_reason"] = str(exc)[:200]
return result
match_5dash = re.search(r"(\d{5})[-\s]*(\d{6,})", text)
match_5plus = re.search(r"(\d{5,})", text)
out = {
"available": True,
"engine": "llm",
"llm_model": model,
"raw_text": text[:500],
"match_5dash6": match_5dash.group(0) if match_5dash else None,
"match_5plus": match_5plus.group(0) if match_5plus else None,
}
if sticker_mode:
# Try to detect sticker color from the response
color_match = re.search(r"\b(green|orange|yellow)\b", text, re.I)
out["sticker_color"] = color_match.group(1).lower() if color_match else "unknown"
return out
def run_ocr_llm_batch(
images: list[tuple[str, bytes]],
model: str | None = None,
max_per_batch: int = BATCH_SIZE_LIMIT,
sticker_mode: bool = False,
) -> list[dict]:
"""Batch OCR multiple images in a single LLM API call."""
import base64
if not OPENCODE_GO_KEY:
return [run_ocr(b) for _, b in images]
model = model or LLM_OCR_MODEL
all_results: list[dict] = []
for batch_start in range(0, len(images), max_per_batch):
batch = images[batch_start:batch_start + max_per_batch]
batch_results = _run_ocr_llm_batch_inner(batch, model, sticker_mode)
all_results.extend(batch_results)
return all_results
def _run_ocr_llm_batch_inner(
batch: list[tuple[str, bytes]], model: str,
sticker_mode: bool = False,
) -> list[dict]:
"""Inner helper: sends one batch of images in a single API call."""
import base64
resized = [(_resize_for_llm(b), n) for n, b in batch]
batch_prompt = (STICKER_BATCH_PROMPT if sticker_mode else DEFAULT_BATCH_PROMPT).format(n=len(batch))
content: list[dict] = [{"type": "text", "text": batch_prompt}]
for idx, (img_bytes, _) in enumerate(resized):
b64 = base64.b64encode(img_bytes).decode()
content.append({
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{b64}"}
})
body = json.dumps({
"model": model,
"messages": [{"role": "user", "content": content}],
"max_tokens": min(300 * len(batch), 8000),
"temperature": 0.1,
}).encode()
req = urllib.request.Request(
f"{OPENCODE_GO_BASE}/chat/completions",
data=body,
headers={
"Authorization": f"Bearer {OPENCODE_GO_KEY}",
"Content-Type": "application/json",
"User-Agent": "Hermes-Agent/1.0",
},
)
try:
resp = urllib.request.urlopen(req, timeout=120)
result = json.loads(resp.read())
raw = result["choices"][0]["message"]["content"].strip()
except Exception:
return [run_ocr_llm(b, model, sticker_mode=sticker_mode) for n, b in batch]
raw_clean = raw
if raw_clean.startswith("```"):
raw_clean = raw_clean.split("\n", 1)[-1]
raw_clean = raw_clean.rsplit("```", 1)[0].strip()
try:
parsed = json.loads(raw_clean)
results: list[dict] = []
for item in parsed:
if sticker_mode:
raw_text = str(item.get("machine_id", "") or "")
else:
raw_text = str(item.get("text", "") or "")
digit_str = str(item.get("digits") or "")
raw_text = raw_text + " " + digit_str
match_5dash = re.search(r"(\d{5})[-\s]*(\d{6,})", raw_text)
match_5plus = re.search(r"(\d{5,})", raw_text)
entry: dict = {
"available": True,
"engine": "llm_batch",
"llm_model": model,
"raw_text": raw_text[:500],
"match_5dash6": match_5dash.group(0) if match_5dash else None,
"match_5plus": match_5plus.group(0) if match_5plus else None,
}
if sticker_mode:
entry["sticker_color"] = str(item.get("sticker_color", "unknown"))
results.append(entry)
return results
except (json.JSONDecodeError, KeyError, TypeError):
return [run_ocr_llm(b, model, sticker_mode=sticker_mode) for n, b in batch]
# ---------------------------------------------------------------------------
# Google Gemini OCR
# ---------------------------------------------------------------------------
def run_ocr_google(image_bytes: bytes, model: str | None = None, sticker_mode: bool = False) -> dict:
"""Run OCR via Google Gemini vision API (free tier)."""
import base64
if not GOOGLE_API_KEY:
result = run_ocr(image_bytes)
result["engine"] = "tesseract"
result["llm_fallback_reason"] = "no_google_api_key"
return result
model = model or GOOGLE_OCR_MODEL
image_bytes = _resize_for_llm(image_bytes)
b64 = base64.b64encode(image_bytes).decode()
prompt = STICKER_OCR_PROMPT if sticker_mode else DEFAULT_OCR_PROMPT
body = json.dumps({
"contents": [{
"parts": [
{"text": prompt},
{"inline_data": {"mime_type": "image/jpeg", "data": b64}}
]
}],
"generationConfig": {"maxOutputTokens": 200, "temperature": 0.1},
}).encode()
req = urllib.request.Request(
f"{GOOGLE_API_BASE}/models/{model}:generateContent?key={GOOGLE_API_KEY}",
data=body,
headers={"Content-Type": "application/json"},
)
try:
resp = urllib.request.urlopen(req, timeout=60)
result = json.loads(resp.read())
text = result["candidates"][0]["content"]["parts"][0]["text"].strip()
except Exception as exc:
result = run_ocr(image_bytes)
result["engine"] = "tesseract"
result["llm_fallback_reason"] = str(exc)[:200]
return result
match_5dash = re.search(r"(\d{5})[-\s]*(\d{6,})", text)
match_5plus = re.search(r"(\d{5,})", text)
out: dict = {
"available": True,
"engine": "google",
"llm_model": model,
"raw_text": text[:500],
"match_5dash6": match_5dash.group(0) if match_5dash else None,
"match_5plus": match_5plus.group(0) if match_5plus else None,
}
if sticker_mode:
cm = re.search(r"\b(green|orange|yellow)\b", text, re.I)
out["sticker_color"] = cm.group(1).lower() if cm else "unknown"
return out
# ---------------------------------------------------------------------------
# Process one photo (shared between analyze + bulk)
# ---------------------------------------------------------------------------
def _process_one(orig_filename: str, contents: bytes, ocr_engine: str,
ocr_model: str | None, sticker_mode: bool) -> dict:
"""Run EXIF + OCR on a single photo. Returns result dict."""
file_size = len(contents)
fhash = _file_hash(contents)
# Check dup
dup_row = None
db_conn = _get_photos_db()
if db_conn:
try:
dup_row = db_conn.execute(
"SELECT id, saved_as FROM photos WHERE file_hash = ?", (fhash,)
).fetchone()
finally:
db_conn.close()
if dup_row:
saved_name = dup_row["saved_as"]
is_dup = True
photo_id = dup_row["id"]
else:
ext = Path(orig_filename or "photo.jpg").suffix.lower()
if ext not in {".jpg", ".jpeg", ".png", ".webp", ".bmp", ".tiff", ".tif", ".dng", ".heic", ".heif"}:
ext = ".jpg"
saved_name = f"{uuid.uuid4().hex}{ext}"
(UPLOADS / saved_name).write_bytes(contents)
is_dup = False
photo_id = None
exif_result = extract_exif(contents)
if ocr_engine == "llm":
ocr_result = run_ocr_llm(contents, ocr_model, sticker_mode=sticker_mode)
elif ocr_engine == "google":
ocr_result = run_ocr_google(contents, ocr_model, sticker_mode=sticker_mode)
elif HAS_TESSERACT:
ocr_result = run_ocr(contents)
ocr_result["engine"] = "tesseract"
else:
ocr_result = {"available": False, "text": "", "engine": "none"}
machine_id = None
asset = None
if ocr_result.get("match_5dash6"):
machine_id = re.sub(r"\D", "", ocr_result["match_5dash6"])[-5:]
asset = lookup_machine_id(machine_id)
# Save to DB if not a duplicate
if not is_dup and not dup_row:
photo_id = _save_photo_to_db(
orig_filename, fhash, saved_name, file_size,
exif_result, ocr_result, machine_id,
)
return {
"filename": orig_filename,
"saved_as": saved_name,
"photo_id": photo_id or (dup_row["id"] if dup_row else None),
"file_size": file_size,
"file_size_kb": round(file_size / 1024, 1),
"duplicate": is_dup,
"exif": exif_result,
"ocr": ocr_result,
"machine_id": machine_id,
"asset": asset,
}
# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
@app.post("/api/analyze")
async def analyze_photo(
file: UploadFile = File(...),
ocr_engine: str = Query(default="tesseract"),
ocr_model: str = Query(default=""),
sticker_mode: bool = Query(default=False),
):
"""Upload a photo, get back EXIF + OCR results.
Query params:
- ocr_engine: 'tesseract' (default) or 'llm'
- ocr_model: model name override (e.g. 'mimo-v2-omni', 'glm-5.1')
- sticker_mode: if true, uses sticker-specific prompt
"""
contents = await file.read()
result = _process_one(
file.filename or "photo.jpg", contents,
ocr_engine, ocr_model or None, sticker_mode,
)
return result
@app.post("/api/bulk-process")
async def bulk_process(
files: list[UploadFile] = File(...),
ocr_engine: str = Query(default="tesseract"),
ocr_model: str = Query(default=""),
sticker_mode: bool = Query(default=False),
):
"""Process multiple photos: OCR each, extract EXIF GPS, look up matching assets.
Query params:
- ocr_engine: 'tesseract' (default) or 'llm'
- ocr_model: model name override
- sticker_mode: if true, uses sticker-specific prompt
Returns a list of results, each with:
- filename, exif (gps), ocr match, matched asset (if found)
- needs_gps: true if asset exists AND has no coordinates AND photo has GPS
"""
if not files:
return {"results": [], "summary": {"total": 0, "has_gps": 0, "matched": 0, "needs_gps": 0}}
# Read all files first
all_files: list[tuple[str, bytes]] = []
for file in files:
contents = await file.read()
all_files.append((file.filename or "photo.jpg", contents))
ocr_use_llm = ocr_engine == "llm"
ocr_use_google = ocr_engine == "google"
# Batch OCR if using LLM mode
if ocr_use_llm:
llm_ocr_results = run_ocr_llm_batch(
[(f, b) for f, b in all_files],
ocr_model or None,
sticker_mode=sticker_mode,
)
else:
llm_ocr_results = None
results = []
summary = {"total": len(all_files), "has_gps": 0, "matched": 0, "needs_gps": 0, "duplicates": 0}
for i, (orig_fname, contents) in enumerate(all_files):
file_size = len(contents)
fhash = _file_hash(contents)
# Check dup
dup_row = None
db_conn = _get_photos_db()
if db_conn:
try:
dup_row = db_conn.execute(
"SELECT id, saved_as FROM photos WHERE file_hash = ?", (fhash,)
).fetchone()
finally:
db_conn.close()
if dup_row:
saved_name = dup_row["saved_as"]
is_dup = True
photo_id = dup_row["id"]
else:
ext = Path(orig_fname or "photo.jpg").suffix.lower()
if ext not in {".jpg", ".jpeg", ".png", ".webp", ".bmp", ".tiff", ".tif", ".dng", ".heic", ".heif"}:
ext = ".jpg"
saved_name = f"{uuid.uuid4().hex}{ext}"
(UPLOADS / saved_name).write_bytes(contents)
is_dup = False
photo_id = None
exif_result = extract_exif(contents)
if ocr_use_llm:
ocr_result = llm_ocr_results[i] if i < len(llm_ocr_results) else {"available": False, "engine": "none", "error": "missing batch result"}
elif ocr_use_google:
ocr_result = run_ocr_google(contents, ocr_model or None, sticker_mode=sticker_mode)
elif HAS_TESSERACT:
ocr_result = run_ocr(contents)
ocr_result["engine"] = "tesseract"
else:
ocr_result = {"available": False, "text": "", "engine": "none"}
has_gps = exif_result.get("gps") is not None
if has_gps:
summary["has_gps"] += 1
machine_id = None
asset = None
needs_gps = False
if ocr_result.get("match_5dash6"):
machine_id = re.sub(r"\D", "", ocr_result["match_5dash6"])[-5:]
asset = lookup_machine_id(machine_id)
if asset:
summary["matched"] += 1
if (asset["latitude"] is None or asset["longitude"] is None) and has_gps:
needs_gps = True
summary["needs_gps"] += 1
# Save to DB if not dup
if not is_dup and not dup_row:
photo_id = _save_photo_to_db(
orig_fname, fhash, saved_name, file_size,
exif_result, ocr_result, machine_id,
)
if is_dup:
summary["duplicates"] += 1
results.append({
"filename": orig_fname,
"saved_as": saved_name,
"photo_id": photo_id or (dup_row["id"] if dup_row else None),
"file_size_kb": round(file_size / 1024, 1),
"duplicate": is_dup,
"exif": exif_result,
"ocr": ocr_result,
"machine_id": machine_id,
"asset": asset,
"needs_gps": needs_gps,
})
return {"results": results, "summary": summary}
@app.get("/api/photos")
async def list_photos(limit: int = Query(default=50, le=200)):
"""List previously processed photos from DB (newest first)."""
conn = _get_photos_db()
try:
rows = conn.execute(
"SELECT * FROM photos ORDER BY created_at DESC LIMIT ?", (limit,)
).fetchall()
return {"photos": [_row_to_dict(r) for r in rows]}
finally:
conn.close()
@app.get("/api/photos/{photo_id}")
async def get_photo(photo_id: int):
"""Get a single photo record from DB."""
conn = _get_photos_db()
try:
row = conn.execute("SELECT * FROM photos WHERE id = ?", (photo_id,)).fetchone()
if not row:
raise HTTPException(404, "Photo not found")
return _row_to_dict(row)
finally:
conn.close()
@app.get("/api/photos/{photo_id}/file")
async def get_photo_file(photo_id: int):
"""Serve the saved image file for a photo record."""
conn = _get_photos_db()
try:
row = conn.execute("SELECT saved_as FROM photos WHERE id = ?", (photo_id,)).fetchone()
if not row:
raise HTTPException(404, "Photo not found")
finally:
conn.close()
filepath = UPLOADS / row["saved_as"]
if not filepath.exists():
raise HTTPException(404, "File not found on disk")
return FileResponse(str(filepath))
@app.post("/api/photos/{photo_id}/reprocess")
async def reprocess_photo(
photo_id: int,
ocr_engine: str = Query(default="tesseract"),
ocr_model: str = Query(default=""),
sticker_mode: bool = Query(default=False),
):
"""Re-run OCR on a previously saved photo with different engine/model.
Query params:
- ocr_engine: 'tesseract' (default) or 'llm'
- ocr_model: model name override (e.g. 'mimo-v2-omni', 'glm-5.1')
- sticker_mode: if true, uses sticker-specific prompt
"""
conn = _get_photos_db()
try:
row = conn.execute(
"SELECT saved_as, orig_filename FROM photos WHERE id = ?", (photo_id,)
).fetchone()
if not row:
raise HTTPException(404, "Photo not found")
finally:
conn.close()
filepath = UPLOADS / row["saved_as"]
if not filepath.exists():
raise HTTPException(404, "Uploaded file not found on disk")
contents = filepath.read_bytes()
if ocr_engine == "llm":
ocr_result = run_ocr_llm(contents, ocr_model or None, sticker_mode=sticker_mode)
elif ocr_engine == "google":
ocr_result = run_ocr_google(contents, ocr_model or None, sticker_mode=sticker_mode)
elif HAS_TESSERACT:
ocr_result = run_ocr(contents)
ocr_result["engine"] = "tesseract"
else:
ocr_result = {"available": False, "text": "", "engine": "none"}
machine_id = None
asset = None
if ocr_result.get("match_5dash6"):
machine_id = re.sub(r"\D", "", ocr_result["match_5dash6"])[-5:]
asset = lookup_machine_id(machine_id)
# Update DB record with new OCR results
db2 = _get_photos_db()
try:
db2.execute(
"""UPDATE photos SET ocr_engine=?, ocr_model=?, ocr_raw_text=?,
ocr_match_5dash6=?, ocr_match_5plus=?, machine_id=?
WHERE id=?""",
(
ocr_result.get("engine"),
ocr_result.get("llm_model"),
ocr_result.get("raw_text"),
ocr_result.get("match_5dash6"),
ocr_result.get("match_5plus"),
machine_id,
photo_id,
),
)
db2.commit()
finally:
db2.close()
return {
"photo_id": photo_id,
"ocr_engine": ocr_result.get("engine"),
"llm_model": ocr_result.get("llm_model"),
"ocr": ocr_result,
"machine_id": machine_id,
"asset": asset,
}
@app.get("/api/lookup")
async def lookup_asset(machine_id: str = ""):
"""Look up an asset by machine_id in the canteen assets database."""
if not machine_id or not machine_id.strip():
return {"found": False, "reason": "No machine_id provided"}
asset = lookup_machine_id(machine_id.strip())
if asset:
return {"found": True, "asset": asset}
return {"found": False, "reason": f"No asset with machine_id '{machine_id}'"}
@app.post("/api/push-gps")
async def push_gps(request: dict):
"""Update an asset's GPS coordinates from photo EXIF data.
Body: {"asset_id": 123, "latitude": 40.7417, "longitude": -73.9292}
Only updates assets that currently have NULL lat/lng.
"""
asset_id = request.get("asset_id")
latitude = request.get("latitude")
longitude = request.get("longitude")
if not asset_id or latitude is None or longitude is None:
raise HTTPException(400, "asset_id, latitude, and longitude are required")
conn = _get_canteen_db()
if not conn:
raise HTTPException(500, "Database not available")
try:
row = conn.execute(
"SELECT latitude, longitude FROM assets WHERE id = ?", (int(asset_id),)
).fetchone()
if not row:
conn.close()
raise HTTPException(404, f"Asset {asset_id} not found")
if row["latitude"] is not None and row["longitude"] is not None:
conn.close()
return {"updated": False, "reason": "Asset already has GPS coordinates", "asset_id": asset_id}
conn.execute(
"UPDATE assets SET latitude = ?, longitude = ?, updated_at = datetime('now') WHERE id = ?",
(float(latitude), float(longitude), int(asset_id)),
)
conn.commit()
conn.close()
return {"updated": True, "asset_id": asset_id, "latitude": float(latitude), "longitude": float(longitude)}
except Exception as e:
try:
conn.close()
except Exception:
pass
raise HTTPException(500, str(e))
@app.get("/api/uploads")
async def list_uploads():
"""List previously uploaded files."""
files = sorted(UPLOADS.glob("*"), key=lambda p: p.stat().st_mtime, reverse=True)
return [
{"name": f.name, "size_kb": round(f.stat().st_size / 1024, 1)}
for f in files[:20]
]
# Mount static LAST
app.mount("/", StaticFiles(directory="static", html=True), name="static")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8903)
Reference in New Issue View Git Blame Copy Permalink