1471 lines
51 KiB
Python
1471 lines
51 KiB
Python
"""EXIF + OCR test backend — validate that GPS survives upload pipeline."""
|
|
import csv, datetime, hashlib, io, json, math, os, re, uuid, sqlite3, urllib.request
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, File, Form, HTTPException, Query, UploadFile
|
|
from fastapi.responses import FileResponse, StreamingResponse
|
|
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
|
|
|
|
# --- Load .env file automatically ---
|
|
_dotenv = Path(__file__).parent.parent.parent / ".hermes" / ".env"
|
|
if not _dotenv.exists():
|
|
_dotenv = Path.home() / ".hermes" / ".env"
|
|
if _dotenv.exists():
|
|
for _line in _dotenv.read_text().splitlines():
|
|
_line = _line.strip()
|
|
if not _line or _line.startswith("#") or "=" not in _line:
|
|
continue
|
|
_key, _val = _line.split("=", 1)
|
|
_key = _key.strip()
|
|
_val = _val.strip()
|
|
# Only set if not already in environment
|
|
if _key not in os.environ:
|
|
os.environ[_key] = _val
|
|
|
|
# === LLM OCR via OpenCode Go ===
|
|
OPENCODE_GO_KEY = (os.environ.get("OPENCODE_GO_API_KEY") or "").strip()
|
|
OPENCODE_GO_BASE = (os.environ.get("OPENCODE_GO_BASE_URL") or "https://opencode.ai/zen/go/v1").strip()
|
|
LLM_OCR_MODEL = (os.environ.get("LLM_OCR_MODEL") or "mimo-v2-omni").strip()
|
|
|
|
# === Google Gemini OCR (free tier) ===
|
|
GOOGLE_API_KEY = (os.environ.get("GOOGLE_API_KEY") or "").strip()
|
|
GOOGLE_API_BASE = (os.environ.get("GOOGLE_API_BASE_URL") or "https://generativelanguage.googleapis.com/v1beta").strip()
|
|
GOOGLE_OCR_MODEL = (os.environ.get("GOOGLE_OCR_MODEL") or "gemini-2.5-flash").strip()
|
|
|
|
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,
|
|
session_id: int | None = 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, sticker_color, has_barcode, session_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,
|
|
ocr_result.get("sticker_color") if ocr_result else None,
|
|
1 if ocr_result and ocr_result.get("has_barcode") else 0,
|
|
session_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 _extract_machine_id(ocr_result: dict) -> str | None:
|
|
"""Extract machine ID from OCR result with priority logic.
|
|
|
|
1. match_5dash6 (5dash6 format: 12345-678901) → first 5 digits
|
|
2. match_5plus (any 5+ digit run) → first 5 digits
|
|
3. Otherwise → None
|
|
"""
|
|
match_5dash6 = ocr_result.get("match_5dash6")
|
|
match_5plus = ocr_result.get("match_5plus")
|
|
|
|
if match_5dash6:
|
|
# "12345-678901" → strip non-digits → "12345678901" → first 5
|
|
clean = re.sub(r"\D", "", match_5dash6)
|
|
if len(clean) >= 5:
|
|
return clean[:5]
|
|
if match_5plus:
|
|
clean = re.sub(r"\D", "", match_5plus)
|
|
if len(clean) >= 5:
|
|
return clean[:5]
|
|
return None
|
|
|
|
|
|
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
|
|
mid = _extract_machine_id(ocr_result)
|
|
if mid:
|
|
machine_id = mid
|
|
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),
|
|
session_id: int | None = Query(default=None),
|
|
):
|
|
"""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
|
|
|
|
mid = _extract_machine_id(ocr_result)
|
|
if mid:
|
|
machine_id = mid
|
|
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,
|
|
session_id=session_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
|
|
mid = _extract_machine_id(ocr_result)
|
|
if mid:
|
|
machine_id = mid
|
|
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=?,
|
|
sticker_color=?
|
|
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,
|
|
ocr_result.get("sticker_color"),
|
|
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.post("/api/photos/{photo_id}/reset")
|
|
async def reset_photo(photo_id: int):
|
|
"""Delete a photo record and its file, allowing re-upload as new."""
|
|
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")
|
|
|
|
# Delete the file from disk
|
|
filepath = UPLOADS / row["saved_as"]
|
|
if filepath.exists():
|
|
filepath.unlink()
|
|
|
|
conn.execute("DELETE FROM photos WHERE id = ?", (photo_id,))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
return {"ok": True, "reset": photo_id, "message": "Photo entry deleted. Re-upload to process as new."}
|
|
|
|
|
|
@app.delete("/api/photos/{photo_id}")
|
|
async def delete_photo(photo_id: int):
|
|
"""Delete a photo record and its file from disk."""
|
|
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")
|
|
|
|
# Delete the file from disk
|
|
filepath = UPLOADS / row["saved_as"]
|
|
if filepath.exists():
|
|
filepath.unlink()
|
|
|
|
conn.execute("DELETE FROM photos WHERE id = ?", (photo_id,))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
return {"ok": True, "deleted": photo_id}
|
|
|
|
|
|
@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.get("/api/export")
|
|
async def export_photos(
|
|
format: str = Query(default="csv", pattern="^(csv|kml|clipboard)$"),
|
|
session_id: int | None = Query(default=None),
|
|
limit: int = Query(default=500, le=5000),
|
|
):
|
|
"""Export photos as CSV, KML, or clipboard-ready text.
|
|
|
|
Query params:
|
|
- format: 'csv' (default), 'kml', or 'clipboard'
|
|
- session_id: limit export to a single session
|
|
- limit: max rows (default 500, max 5000)
|
|
"""
|
|
conn = _get_photos_db()
|
|
try:
|
|
if session_id:
|
|
rows = conn.execute(
|
|
"SELECT * FROM photos WHERE session_id = ? ORDER BY created_at DESC LIMIT ?",
|
|
(session_id, limit),
|
|
).fetchall()
|
|
else:
|
|
rows = conn.execute(
|
|
"SELECT * FROM photos ORDER BY created_at DESC LIMIT ?", (limit,)
|
|
).fetchall()
|
|
finally:
|
|
conn.close()
|
|
|
|
records = []
|
|
for r in rows:
|
|
d = _row_to_dict(r)
|
|
rec = {
|
|
"filename": d.get("orig_filename", ""),
|
|
"lat": d.get("gps_lat"),
|
|
"lng": d.get("gps_lng"),
|
|
"machine_id": d.get("machine_id", ""),
|
|
"ocr_engine": d.get("ocr_engine", ""),
|
|
"sticker_color": d.get("sticker_color", ""),
|
|
"timestamp": d.get("created_at", ""),
|
|
}
|
|
# Look up asset details
|
|
if d.get("machine_id"):
|
|
asset = lookup_machine_id(d["machine_id"])
|
|
if asset:
|
|
rec.update({
|
|
"asset_name": asset.get("name", ""),
|
|
"building": asset.get("building_name", ""),
|
|
"floor": asset.get("floor", ""),
|
|
"room": asset.get("room", ""),
|
|
"make": asset.get("make", ""),
|
|
"model": asset.get("model", ""),
|
|
})
|
|
records.append(rec)
|
|
|
|
if format == "csv":
|
|
output = io.StringIO()
|
|
writer = csv.writer(output)
|
|
header = ["filename", "lat", "lng", "machine_id", "asset_name",
|
|
"building", "floor", "room", "make", "model",
|
|
"ocr_engine", "sticker_color", "timestamp"]
|
|
bom = "\ufeff"
|
|
output.write(bom)
|
|
writer.writerow(header)
|
|
for rec in records:
|
|
writer.writerow([rec.get(h.replace("-", "_"), "") for h in header])
|
|
csv_content = output.getvalue()
|
|
return StreamingResponse(
|
|
iter([csv_content]),
|
|
media_type="text/csv",
|
|
headers={"Content-Disposition": f"attachment; filename=exif-export.csv"},
|
|
)
|
|
|
|
elif format == "kml":
|
|
kml_parts = [
|
|
'<?xml version="1.0" encoding="UTF-8"?>',
|
|
'<kml xmlns="http://www.opengis.net/kml/2.2">',
|
|
" <Document>",
|
|
f" <name>EXIF Export ({len(records)} photos)</name>",
|
|
]
|
|
for rec in records:
|
|
if rec["lat"] is None or rec["lng"] is None:
|
|
continue
|
|
name = rec.get("asset_name") or rec.get("machine_id") or rec.get("filename", "Photo")
|
|
desc_lines = [
|
|
f"Filename: {rec['filename']}",
|
|
f"Machine ID: {rec.get('machine_id', 'N/A')}",
|
|
f"OCR Engine: {rec.get('ocr_engine', 'N/A')}",
|
|
f"Timestamp: {rec.get('timestamp', 'N/A')}",
|
|
]
|
|
kml_parts.append(" <Placemark>")
|
|
kml_parts.append(f" <name>{name}</name>")
|
|
kml_parts.append(f" <description>{chr(10).join(desc_lines)}</description>")
|
|
kml_parts.append(" <Point>")
|
|
kml_parts.append(f" <coordinates>{rec['lng']},{rec['lat']},0</coordinates>")
|
|
kml_parts.append(" </Point>")
|
|
kml_parts.append(" </Placemark>")
|
|
kml_parts.append(" </Document>")
|
|
kml_parts.append("</kml>")
|
|
kml_content = "\n".join(kml_parts)
|
|
return StreamingResponse(
|
|
iter([kml_content]),
|
|
media_type="application/vnd.google-earth.kml+xml",
|
|
headers={"Content-Disposition": f"attachment; filename=exif-export.kml"},
|
|
)
|
|
|
|
else: # clipboard
|
|
lines = []
|
|
for rec in records:
|
|
if rec["lat"] is not None and rec["lng"] is not None:
|
|
label = rec.get("asset_name") or rec.get("machine_id") or rec.get("filename", "?")
|
|
lines.append(f"{rec['lat']:.6f},{rec['lng']:.6f} # {label}")
|
|
return {"count": len(lines), "text": "\n".join(lines)}
|
|
|
|
|
|
# ---------------------------------------------------------------------------#
|
|
# Inline Machine ID Edit — Feature #5 #
|
|
# ---------------------------------------------------------------------------#
|
|
@app.post("/api/assign-machine-id")
|
|
async def assign_machine_id(request: dict):
|
|
"""Manually assign a machine_id to a photo record.
|
|
|
|
Body: { "photo_id": int, "machine_id": str }
|
|
Returns: { "photo_id": int, "machine_id": str, "asset": dict|null, "needs_gps": bool }
|
|
"""
|
|
photo_id = request.get("photo_id")
|
|
machine_id = (request.get("machine_id") or "").strip()
|
|
if not photo_id or not machine_id:
|
|
raise HTTPException(400, "photo_id and machine_id are required")
|
|
|
|
conn = _get_photos_db()
|
|
try:
|
|
row = conn.execute("SELECT id, gps_lat, gps_lng FROM photos WHERE id = ?", (photo_id,)).fetchone()
|
|
if not row:
|
|
raise HTTPException(404, f"Photo {photo_id} not found")
|
|
|
|
conn.execute(
|
|
"UPDATE photos SET machine_id = ? WHERE id = ?",
|
|
(machine_id, photo_id),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
asset = lookup_machine_id(machine_id)
|
|
has_photo_gps = row["gps_lat"] is not None and row["gps_lng"] is not None
|
|
needs_gps = bool(asset and (asset["latitude"] is None or asset["longitude"] is None) and has_photo_gps)
|
|
|
|
return {
|
|
"photo_id": photo_id,
|
|
"machine_id": machine_id,
|
|
"asset": asset,
|
|
"needs_gps": needs_gps,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------#
|
|
# GPS Proximity — Feature #6 #
|
|
# ---------------------------------------------------------------------------#
|
|
@app.get("/api/nearby")
|
|
async def nearby_assets(
|
|
lat: float = Query(...),
|
|
lng: float = Query(...),
|
|
radius: float = Query(default=10.0, ge=1, le=1000),
|
|
exclude_photo_id: int | None = Query(default=None),
|
|
):
|
|
"""Find canteen assets near a GPS point.
|
|
|
|
Uses Euclidean approximation over lat/lng (accurate at small radii <1km).
|
|
1 deg lat ~ 111,320m, 1 deg lng ~ 111,320 * cos(lat).
|
|
"""
|
|
conn = _get_canteen_db()
|
|
if not conn:
|
|
return {"nearby": [], "count": 0, "radius_m": radius}
|
|
|
|
try:
|
|
lat_rad = lat * 3.14159265 / 180.0
|
|
deg_per_m_lat = 1.0 / 111320.0
|
|
deg_per_m_lng = 1.0 / (111320.0 * math.cos(lat_rad))
|
|
rad_deg = radius * deg_per_m_lat
|
|
|
|
rows = conn.execute(
|
|
"""SELECT id, machine_id, name, latitude, longitude,
|
|
building_name, floor, room, category, status
|
|
FROM assets
|
|
WHERE latitude IS NOT NULL AND longitude IS NOT NULL
|
|
AND (latitude - ?) * (latitude - ?) + (longitude - ?) * (longitude - ?) < ?
|
|
ORDER BY (latitude - ?) * (latitude - ?) + (longitude - ?) * (longitude - ?)
|
|
LIMIT 50""",
|
|
(lat, lat, lng, lng, rad_deg * rad_deg,
|
|
lat, lat, lng, lng),
|
|
).fetchall()
|
|
|
|
nearby = []
|
|
for row in rows:
|
|
dlat = row["latitude"] - lat
|
|
dlng = (row["longitude"] - lng) * math.cos(lat_rad)
|
|
distance_m = (dlat * dlat + dlng * dlng) ** 0.5 * 111320.0
|
|
if distance_m >= radius:
|
|
continue
|
|
nearby.append({
|
|
"asset_id": row["id"],
|
|
"machine_id": row["machine_id"],
|
|
"name": row["name"],
|
|
"distance_m": round(distance_m, 1),
|
|
"latitude": row["latitude"],
|
|
"longitude": row["longitude"],
|
|
"has_gps": True,
|
|
"building": row.get("building_name"),
|
|
"floor": row.get("floor"),
|
|
"room": row.get("room"),
|
|
"category": row.get("category"),
|
|
"status": row.get("status"),
|
|
})
|
|
nearby.sort(key=lambda x: x["distance_m"])
|
|
return {"nearby": nearby, "count": len(nearby), "radius_m": radius}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------#
|
|
# Session Management — Feature #9 #
|
|
# ---------------------------------------------------------------------------#
|
|
def _init_sessions_table():
|
|
"""Migrate photos.db: create sessions table, add session_id column."""
|
|
conn = sqlite3.connect(str(PHOTOS_DB))
|
|
try:
|
|
conn.execute("""CREATE TABLE IF NOT EXISTS sessions (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL DEFAULT '',
|
|
building TEXT DEFAULT '',
|
|
floor TEXT DEFAULT '',
|
|
photo_count INTEGER DEFAULT 0,
|
|
matched_count INTEGER DEFAULT 0,
|
|
needs_gps_count INTEGER DEFAULT 0,
|
|
closed_at TEXT,
|
|
created_at TEXT DEFAULT (datetime('now'))
|
|
)""")
|
|
cols = [r[1] for r in conn.execute("PRAGMA table_info(photos)").fetchall()]
|
|
if "session_id" not in cols:
|
|
conn.execute("ALTER TABLE photos ADD COLUMN session_id INTEGER REFERENCES sessions(id)")
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
_init_sessions_table()
|
|
|
|
|
|
@app.get("/api/sessions")
|
|
async def list_sessions():
|
|
"""List all sessions, newest first."""
|
|
conn = _get_photos_db()
|
|
try:
|
|
rows = conn.execute(
|
|
"SELECT * FROM sessions ORDER BY created_at DESC LIMIT 100"
|
|
).fetchall()
|
|
return {"sessions": [_row_to_dict(r) for r in rows]}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@app.post("/api/sessions")
|
|
async def create_session(request: dict):
|
|
"""Create a new session.
|
|
|
|
Body: { name?, building?, floor? }
|
|
Returns: { id, ...session_fields }
|
|
"""
|
|
name = (request.get("name") or "").strip() or f"Session {datetime.datetime.now().strftime('%Y-%m-%d %H:%M')}"
|
|
building = (request.get("building") or "").strip()
|
|
floor = (request.get("floor") or "").strip()
|
|
conn = _get_photos_db()
|
|
try:
|
|
cur = conn.execute(
|
|
"INSERT INTO sessions (name, building, floor) VALUES (?, ?, ?)",
|
|
(name, building, floor),
|
|
)
|
|
conn.commit()
|
|
session_id = cur.lastrowid
|
|
row = conn.execute("SELECT * FROM sessions WHERE id = ?", (session_id,)).fetchone()
|
|
return _row_to_dict(row)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@app.get("/api/sessions/{session_id}")
|
|
async def get_session(session_id: int):
|
|
"""Get a session with all its photos."""
|
|
conn = _get_photos_db()
|
|
try:
|
|
srow = conn.execute("SELECT * FROM sessions WHERE id = ?", (session_id,)).fetchone()
|
|
if not srow:
|
|
raise HTTPException(404, "Session not found")
|
|
session = _row_to_dict(srow)
|
|
photos = conn.execute(
|
|
"SELECT * FROM photos WHERE session_id = ? ORDER BY created_at ASC",
|
|
(session_id,),
|
|
).fetchall()
|
|
session["photos"] = [_row_to_dict(r) for r in photos]
|
|
total = len(session["photos"])
|
|
matched = sum(1 for p in session["photos"] if p.get("machine_id"))
|
|
needs_gps = sum(1 for p in session["photos"] if p.get("machine_id") and (p.get("gps_lat") is None))
|
|
conn.execute(
|
|
"UPDATE sessions SET photo_count=?, matched_count=?, needs_gps_count=? WHERE id=?",
|
|
(total, matched, needs_gps, session_id),
|
|
)
|
|
conn.commit()
|
|
session["photo_count"] = total
|
|
session["matched_count"] = matched
|
|
session["needs_gps_count"] = needs_gps
|
|
return session
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@app.post("/api/sessions/{session_id}/close")
|
|
async def close_session(session_id: int):
|
|
"""Finalize a session — no more photos can be added."""
|
|
conn = _get_photos_db()
|
|
try:
|
|
row = conn.execute("SELECT id FROM sessions WHERE id = ?", (session_id,)).fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Session not found")
|
|
conn.execute(
|
|
"UPDATE sessions SET closed_at = datetime('now') WHERE id = ?",
|
|
(session_id,),
|
|
)
|
|
conn.commit()
|
|
return {"ok": True, "session_id": session_id, "closed_at": "now"}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------#
|
|
# Batch Machine ID Assign — Feature #8 #
|
|
# ---------------------------------------------------------------------------#
|
|
@app.post("/api/bulk-assign")
|
|
async def bulk_assign_machine_id(request: dict):
|
|
"""Assign the same machine_id to multiple photos at once.
|
|
|
|
Body: { "photo_ids": [int], "machine_id": str }
|
|
Returns: { "updated": N, "gps_pushed": M, "no_gps": [...filenames] }
|
|
"""
|
|
photo_ids = request.get("photo_ids", [])
|
|
machine_id = (request.get("machine_id") or "").strip()
|
|
|
|
if not photo_ids or not isinstance(photo_ids, list):
|
|
raise HTTPException(400, "photo_ids must be a non-empty list")
|
|
if not machine_id:
|
|
raise HTTPException(400, "machine_id is required")
|
|
|
|
asset = lookup_machine_id(machine_id)
|
|
conn = _get_photos_db()
|
|
no_gps_files = []
|
|
gps_pushed = 0
|
|
|
|
try:
|
|
for pid in photo_ids:
|
|
row = conn.execute(
|
|
"SELECT id, orig_filename, gps_lat, gps_lng, machine_id FROM photos WHERE id = ?",
|
|
(pid,),
|
|
).fetchone()
|
|
if not row:
|
|
continue
|
|
|
|
conn.execute(
|
|
"UPDATE photos SET machine_id = ? WHERE id = ?",
|
|
(machine_id, pid),
|
|
)
|
|
|
|
if asset and (asset["latitude"] is None or asset["longitude"] is None):
|
|
gps_lat = row["gps_lat"]
|
|
gps_lng = row["gps_lng"]
|
|
if gps_lat is not None and gps_lng is not None:
|
|
canteen = _get_canteen_db()
|
|
if canteen:
|
|
try:
|
|
canteen.execute(
|
|
"UPDATE assets SET latitude=?, longitude=?, updated_at=datetime('now') WHERE id=?",
|
|
(float(gps_lat), float(gps_lng), asset["id"]),
|
|
)
|
|
canteen.commit()
|
|
gps_pushed += 1
|
|
finally:
|
|
canteen.close()
|
|
else:
|
|
no_gps_files.append(row.get("orig_filename", f"photo_{pid}"))
|
|
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
return {
|
|
"updated": len(photo_ids),
|
|
"gps_pushed": gps_pushed,
|
|
"no_gps": no_gps_files,
|
|
}
|
|
|
|
|
|
# Mount static LAST
|
|
app.mount("/", StaticFiles(directory="static", html=True), name="static")
|
|
|
|
|
|
# ---------------------------------------------------------------------------#
|
|
# Export (CSV / KML / Clipboard) — Feature #4 #
|
|
# ---------------------------------------------------------------------------#
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8903)
|