shawn/exif-test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Backend: Export CSV/KML/clipboard, Assign Machine ID, GPS Proximity, Session Management, Batch Assign

Browse Source
This commit is contained in:
Shawn
2026-05-25 19:08:07 -04:00
parent fa894183fc
commit e11ac3c6f2
3 changed files with 415 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
__pycache__/server.cpython-311.pyc
BIN
View File
Binary file not shown.
new_endpoints.py
View File
server.py
+415 -4
View File
@@ -1,9 +1,9 @@
"""EXIF + OCR test backend — validate that GPS survives upload pipeline."""
import hashlib, io, json, os, re, uuid, sqlite3, urllib.request
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
from fastapi.responses import FileResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from PIL import Image as PILImage
@@ -141,6 +141,7 @@ def _file_hash(data: bytes) -> str:
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()
@@ -149,8 +150,8 @@ def _save_photo_to_db(
"""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)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
ocr_match_5dash6, ocr_match_5plus, machine_id, sticker_color, has_barcode, session_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
orig_filename,
file_hash,
@@ -167,6 +168,7 @@ def _save_photo_to_db(
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()
@@ -693,6 +695,7 @@ async def bulk_process(
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.
@@ -793,6 +796,7 @@ async def bulk_process(
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:
@@ -1051,9 +1055,416 @@ async def list_uploads():
]
# 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)