shawn/canteen-asset-tracker
1
0
Fork 0
Code Issues 8 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
874e15c01ea70a0267e9c16449d8c21a5060faa1
canteen-asset-tracker/server.py
T
shawn 874e15c01e Add disney_group grouped park filter categories 
- New disney_group query param on /api/assets (park, resort,
  office, springs, other)
- Both list and map filter dropdowns show group options above
  individual parks
- Backend uses IN queries for grouped park values
- Map view handles groups via client-side PARK_GROUP_MAP

Closes #69
2026-06-03 00:57:55 -04:00

3941 lines
150 KiB
Python
Raw Blame History

"""
Canteen Asset Geolocation Tool — FastAPI server.
Single-file backend: SQLite storage, asset CRUD, machine_id search,
check-ins with GPS, stats, and CSV export.
v2 schema: full asset management with customers, locations, keys, badges, etc.
"""
import csv
import hashlib
import io
import json as _json
import math
import os
import re
import secrets
import sqlite3
import urllib.error
import urllib.request
import uuid
from contextlib import asynccontextmanager
from datetime import date
from pathlib import Path
_HAS_TESSERACT: bool
try:
import pytesseract
_HAS_TESSERACT = True
except ImportError:
pytesseract = None # type: ignore
_HAS_TESSERACT = False
# Ollama vision model (Windows PC via SSH tunnel)
_OLLAMA_BASE = "http://127.0.0.1:11434"
_OLLAMA_VISION_MODEL = "qwen2.5vl:3b"
_HAS_OLLAMA: bool = False
try:
import urllib.request as _ollama_urllib
import json as _ollama_json
_req = _ollama_urllib.Request(
f"{_OLLAMA_BASE}/api/generate",
data=_ollama_json.dumps({"model": _OLLAMA_VISION_MODEL, "prompt": "ping", "stream": False}).encode(),
headers={"Content-Type": "application/json"},
)
_resp = _ollama_urllib.urlopen(_req, timeout=5)
_HAS_OLLAMA = True
except Exception:
_HAS_OLLAMA = False
import piexif
from PIL import Image as PILImage
# ─── Asset matcher (photo OCR → DB lookup) ─────────────────────────────────
from classify_makes import normalize_identifier, find_asset_by_normalized_id
from fastapi import FastAPI, HTTPException, Query, Request, UploadFile, File, Form, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from typing import Optional, List
# ─── Config ─────────────────────────────────────────────────────────────────
DB_PATH = os.environ.get("CANTEEN_DB_PATH", str(Path(__file__).parent / "assets.db"))
UPLOADS_DIR = Path(os.environ.get("CANTEEN_UPLOADS_DIR", str(Path(__file__).parent / "uploads")))
STATIC_DIR = Path(__file__).parent / "static"
# ─── MSFS Data (merged asset enrichment from Dynamics 365 Field Service) ───
MSFS_DATA_PATH = os.environ.get(
"MSFS_DATA_PATH",
str(Path.home() / "projects/ms-field-service-extraction/web/static/data/merged-assets.json"),
)
_MSFS_LOOKUP: dict[str, dict] = {} # keyed by canteen_machine_id
_MSFS_BY_ID: dict[int, dict] = {} # keyed by canteen_id (assets.id)
if os.path.exists(MSFS_DATA_PATH):
try:
with open(MSFS_DATA_PATH, "r") as _f:
_msfs_all = _json.load(_f)
for _entry in _msfs_all.get("assets", []):
_mid = _entry.get("canteen_machine_id")
_cid = _entry.get("canteen_id")
if _mid:
_MSFS_LOOKUP[_mid] = _entry
if _cid:
_MSFS_BY_ID[_cid] = _entry
except Exception:
pass # Non-fatal — MSFS enrichment just won't be available
def _load_categories() -> set:
"""Load valid category names from the categories lookup table."""
try:
conn = sqlite3.connect(DB_PATH)
names = {r[0] for r in conn.execute("SELECT name FROM categories").fetchall()}
conn.close()
return names if names else {"Other"}
except Exception:
return {"Furniture", "Appliances", "Utensils & Serveware", "Equipment", "Other"}
VALID_CATEGORIES = _load_categories()
VALID_STATUSES = {"active", "maintenance", "retired"}
# ─── Database ───────────────────────────────────────────────────────────────
def get_db() -> sqlite3.Connection:
"""Return a new DB connection with WAL + foreign keys enabled."""
conn = sqlite3.connect(DB_PATH)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
conn.row_factory = sqlite3.Row
return conn
def _create_v2_tables(conn: sqlite3.Connection):
"""Create all v2 tables if they don't exist."""
conn.executescript("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
password_hash TEXT NOT NULL,
role TEXT NOT NULL DEFAULT 'technician',
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS customers (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS customer_contacts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
customer_id INTEGER NOT NULL REFERENCES customers(id) ON DELETE CASCADE,
name TEXT,
phone TEXT,
email TEXT
);
CREATE TABLE IF NOT EXISTS locations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
customer_id INTEGER REFERENCES customers(id),
name TEXT,
address TEXT DEFAULT '',
building_name TEXT DEFAULT '',
building_number TEXT DEFAULT '',
floor TEXT DEFAULT '',
trailer_number TEXT DEFAULT '',
site_hours TEXT DEFAULT '',
access_notes TEXT DEFAULT '',
walking_directions TEXT DEFAULT '',
map_link TEXT DEFAULT '',
latitude REAL DEFAULT NULL,
longitude REAL DEFAULT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS rooms (
id INTEGER PRIMARY KEY AUTOINCREMENT,
location_id INTEGER NOT NULL REFERENCES locations(id) ON DELETE CASCADE,
name TEXT,
floor TEXT DEFAULT '',
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS categories (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE NOT NULL,
icon TEXT DEFAULT ''
);
CREATE TABLE IF NOT EXISTS key_names (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE NOT NULL
);
CREATE TABLE IF NOT EXISTS key_types (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE NOT NULL
);
CREATE TABLE IF NOT EXISTS badge_types (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE NOT NULL
);
CREATE TABLE IF NOT EXISTS makes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE NOT NULL
);
CREATE TABLE IF NOT EXISTS models (
id INTEGER PRIMARY KEY AUTOINCREMENT,
make_id INTEGER NOT NULL REFERENCES makes(id),
name TEXT NOT NULL,
icon_path TEXT
);
CREATE TABLE IF NOT EXISTS assets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
machine_id TEXT NOT NULL UNIQUE,
serial_number TEXT DEFAULT '',
name TEXT NOT NULL,
description TEXT DEFAULT '',
category TEXT NOT NULL DEFAULT 'Other',
status TEXT NOT NULL DEFAULT 'active',
make TEXT DEFAULT '',
model TEXT DEFAULT '',
connect_id TEXT DEFAULT '',
equipment_id TEXT DEFAULT '',
barcode TEXT DEFAULT '',
address TEXT DEFAULT '',
building_name TEXT DEFAULT '',
building_number TEXT DEFAULT '',
floor TEXT DEFAULT '',
room TEXT DEFAULT '',
trailer_number TEXT DEFAULT '',
walking_directions TEXT DEFAULT '',
map_link TEXT DEFAULT '',
parking_location TEXT DEFAULT '',
photo_path TEXT,
customer_id INTEGER REFERENCES customers(id),
location_id INTEGER REFERENCES locations(id),
assigned_to INTEGER REFERENCES users(id),
latitude REAL DEFAULT NULL,
longitude REAL DEFAULT NULL,
geofence_radius_meters INTEGER DEFAULT 50,
disney_park TEXT DEFAULT NULL,
is_disney INTEGER DEFAULT 0,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
company TEXT DEFAULT '',
customer_name TEXT DEFAULT '',
place TEXT DEFAULT '',
location_area TEXT DEFAULT ''
);
CREATE TABLE IF NOT EXISTS service_entrances (
id INTEGER PRIMARY KEY AUTOINCREMENT,
location_id INTEGER NOT NULL REFERENCES locations(id) ON DELETE CASCADE,
name TEXT NOT NULL,
latitude REAL NOT NULL,
longitude REAL NOT NULL,
notes TEXT DEFAULT '',
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS checkins (
id INTEGER PRIMARY KEY AUTOINCREMENT,
asset_id INTEGER NOT NULL REFERENCES assets(id) ON DELETE CASCADE,
user_id INTEGER REFERENCES users(id),
latitude REAL,
longitude REAL,
accuracy REAL,
photo_path TEXT,
notes TEXT DEFAULT '',
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS asset_keys (
id INTEGER PRIMARY KEY AUTOINCREMENT,
asset_id INTEGER NOT NULL REFERENCES assets(id) ON DELETE CASCADE,
key_name TEXT,
key_type TEXT
);
CREATE TABLE IF NOT EXISTS asset_badges (
id INTEGER PRIMARY KEY AUTOINCREMENT,
asset_id INTEGER NOT NULL REFERENCES assets(id) ON DELETE CASCADE,
badge_name TEXT
);
CREATE TABLE IF NOT EXISTS settings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
key TEXT UNIQUE NOT NULL,
value TEXT
);
CREATE TABLE IF NOT EXISTS activity_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER REFERENCES users(id),
action TEXT,
entity_type TEXT,
entity_id INTEGER,
details TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS geofences (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT,
points TEXT,
color TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS geofence_users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
geofence_id INTEGER NOT NULL REFERENCES geofences(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
UNIQUE(geofence_id, user_id)
);
CREATE TABLE IF NOT EXISTS visits (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER REFERENCES users(id),
asset_id INTEGER REFERENCES assets(id) ON DELETE CASCADE,
checkin_time TEXT,
checkout_time TEXT,
duration_minutes INTEGER,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS sessions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER REFERENCES users(id) ON DELETE CASCADE,
token TEXT UNIQUE NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
expires_at TEXT NOT NULL DEFAULT (datetime('now', '+1 day'))
);
CREATE INDEX IF NOT EXISTS idx_checkins_asset_id ON checkins(asset_id);
CREATE INDEX IF NOT EXISTS idx_checkins_created_at ON checkins(created_at);
""")
def _seed_if_empty(conn: sqlite3.Connection, table: str, columns: tuple, rows: list):
"""Insert seed rows if table is empty."""
existing = conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
if existing == 0:
placeholders = ", ".join(["?"] * len(columns))
col_names = ", ".join(columns)
conn.executemany(
f"INSERT INTO {table} ({col_names}) VALUES ({placeholders})",
rows,
)
def _seed_data(conn: sqlite3.Connection):
"""Insert default seed data for lookup tables."""
_seed_if_empty(conn, "categories", ("name", "icon"), [
("Bev", "🥤"), ("Snack", "🍿"),
("Food", "🍔"), ("Equipment", "🔧"),
("Appliances", "🔌"), ("Other", "📦"),
])
_seed_if_empty(conn, "key_names", ("name",), [
("MK500",), ("Green Dot",), ("Red Key",), ("Blue Key",),
("Master Key",), ("Padlock Key",),
])
_seed_if_empty(conn, "key_types", ("name",), [
("Round Short",), ("Barrel",), ("Standard",), ("Flat",), ("Tubular",),
])
_seed_if_empty(conn, "badge_types", ("name",), [
("Disney Contractor Base",), ("Visitor Badge",), ("Employee Badge",),
("Contractor Badge",), ("Temporary Pass",),
])
_seed_if_empty(conn, "makes", ("name",), [
("Canteen",), ("Hobart",), ("Vollrath",), ("Metro",),
("Rubbermaid",), ("Cambro",), ("Other",),
])
# Ensure default admin users exist
conn.execute(
"INSERT OR IGNORE INTO users (username, password_hash, role) VALUES (?, ?, ?)",
("shawn", "887f732280a2d74a8468b04ebce6feb1a4968cc7d77411d996802ef22a907da5", "admin"),
)
conn.execute(
"INSERT OR IGNORE INTO users (username, password_hash, role) VALUES (?, ?, ?)",
("admin", "057ba03d6c44104863dc7361fe4578965d1887360f90a0895882e58a6248fc86", "admin"),
)
def _migrate_sessions_expires_at(conn: sqlite3.Connection):
"""Add expires_at column to sessions table if it doesn't exist yet."""
try:
conn.execute("ALTER TABLE sessions ADD COLUMN expires_at TEXT")
except sqlite3.OperationalError:
pass # Column already exists
# Set default expiry for any sessions with NULL expires_at
conn.execute(
"UPDATE sessions SET expires_at = datetime(created_at, '+1 day') "
"WHERE expires_at IS NULL"
)
conn.commit()
def _ensure_unique_machine_id(conn: sqlite3.Connection):
"""Remove duplicate machine_ids (keep oldest), then create UNIQUE index."""
dupes = conn.execute("""
SELECT machine_id FROM assets
GROUP BY machine_id
HAVING COUNT(*) > 1
""").fetchall()
for dupe in dupes:
mid = dupe["machine_id"]
ids = conn.execute(
"SELECT id FROM assets WHERE machine_id = ? ORDER BY id", (mid,)
).fetchall()
keep_id = ids[0]["id"]
for row in ids[1:]:
conn.execute("DELETE FROM assets WHERE id = ?", (row["id"],))
conn.execute("DROP INDEX IF EXISTS idx_assets_machine_id")
conn.execute("CREATE UNIQUE INDEX idx_assets_machine_id ON assets(machine_id)")
def _migrate_v1_to_v2(conn: sqlite3.Connection):
"""Migrate existing v1 database (barcode-based) to v2 schema."""
# Create all new tables first
_create_v2_tables(conn)
# Create assets_v2 with new schema, copying old data
conn.execute("""
CREATE TABLE assets_v2 (
id INTEGER PRIMARY KEY AUTOINCREMENT,
machine_id TEXT NOT NULL UNIQUE,
serial_number TEXT DEFAULT '',
name TEXT NOT NULL,
description TEXT DEFAULT '',
category TEXT NOT NULL DEFAULT 'Other',
status TEXT NOT NULL DEFAULT 'active',
make TEXT DEFAULT '',
model TEXT DEFAULT '',
address TEXT DEFAULT '',
building_name TEXT DEFAULT '',
building_number TEXT DEFAULT '',
floor TEXT DEFAULT '',
room TEXT DEFAULT '',
trailer_number TEXT DEFAULT '',
walking_directions TEXT DEFAULT '',
map_link TEXT DEFAULT '',
parking_location TEXT DEFAULT '',
photo_path TEXT,
customer_id INTEGER REFERENCES customers(id),
location_id INTEGER REFERENCES locations(id),
assigned_to INTEGER REFERENCES users(id),
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
)
""")
conn.execute("""
INSERT INTO assets_v2 (id, machine_id, name, description, category, status, photo_path, created_at, updated_at)
SELECT id, barcode, name, description, category, status, photo_path, created_at, updated_at
FROM assets
""")
conn.execute("DROP TABLE assets")
conn.execute("ALTER TABLE assets_v2 RENAME TO assets")
_ensure_unique_machine_id(conn)
conn.execute("CREATE INDEX IF NOT EXISTS idx_assets_category ON assets(category)")
# Add user_id to checkins if not present
cursor = conn.execute("PRAGMA table_info(checkins)")
checkin_cols = {row[1] for row in cursor.fetchall()}
if "user_id" not in checkin_cols:
conn.execute("ALTER TABLE checkins ADD COLUMN user_id INTEGER REFERENCES users(id)")
# Seed lookup data
_seed_data(conn)
# Schema migrations for existing databases
_migrate_sessions_expires_at(conn)
conn.commit()
def init_db(conn: sqlite3.Connection):
"""Create tables and indexes if they don't exist. Runs v1→v2 migration if needed."""
# Check if assets table exists and has old schema
cursor = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='assets'"
)
if cursor.fetchone():
col_cursor = conn.execute("PRAGMA table_info(assets)")
columns = {row[1] for row in col_cursor.fetchall()}
if "barcode" in columns and "machine_id" not in columns:
_migrate_v1_to_v2(conn)
return
# Fresh install or already migrated — create all tables
_create_v2_tables(conn)
_seed_data(conn)
_migrate_sessions_expires_at(conn)
# Asset indexes — created here (not in _create_v2_tables) to avoid
# failing during migration when old v1 assets table lacks machine_id.
_ensure_unique_machine_id(conn)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_assets_category ON assets(category)"
)
conn.commit()
# Add lat/lng to assets if not present (v3 migration)
cursor = conn.execute("PRAGMA table_info(assets)")
asset_cols = {row[1] for row in cursor.fetchall()}
if "latitude" not in asset_cols:
conn.execute("ALTER TABLE assets ADD COLUMN latitude REAL DEFAULT NULL")
if "longitude" not in asset_cols:
conn.execute("ALTER TABLE assets ADD COLUMN longitude REAL DEFAULT NULL")
if "geofence_radius_meters" not in asset_cols:
conn.execute("ALTER TABLE assets ADD COLUMN geofence_radius_meters INTEGER DEFAULT 50")
cursor = conn.execute("PRAGMA table_info(locations)")
loc_cols = {row[1] for row in cursor.fetchall()}
if "latitude" not in loc_cols:
conn.execute("ALTER TABLE locations ADD COLUMN latitude REAL DEFAULT NULL")
if "longitude" not in loc_cols:
conn.execute("ALTER TABLE locations ADD COLUMN longitude REAL DEFAULT NULL")
conn.commit()
# v4 migration: disney_park column + service_entrances table
if "disney_park" not in asset_cols:
conn.execute("ALTER TABLE assets ADD COLUMN disney_park TEXT DEFAULT NULL")
conn.commit()
# v5 migration: dex_report_date + install_date
if "dex_report_date" not in asset_cols:
conn.execute("ALTER TABLE assets ADD COLUMN dex_report_date TEXT DEFAULT NULL")
if "install_date" not in asset_cols:
conn.execute("ALTER TABLE assets ADD COLUMN install_date TEXT DEFAULT NULL")
conn.commit()
# Ensure service_entrances table exists with correct schema
se_exists = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='service_entrances'"
).fetchone()
if se_exists:
# Check if name column has NOT NULL constraint
se_cols = conn.execute("PRAGMA table_info(service_entrances)").fetchall()
se_name_col = next((c for c in se_cols if c[1] == 'name'), None)
if se_name_col and not se_name_col[3]: # notnull == 0
# Table exists but with wrong schema — recreate it (new table, safe to drop)
conn.execute("DROP TABLE service_entrances")
se_exists = False
if not se_exists:
conn.execute("""
CREATE TABLE service_entrances (
id INTEGER PRIMARY KEY AUTOINCREMENT,
location_id INTEGER NOT NULL REFERENCES locations(id) ON DELETE CASCADE,
name TEXT NOT NULL,
latitude REAL NOT NULL,
longitude REAL NOT NULL,
notes TEXT DEFAULT '',
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
)
""")
conn.commit()
# ─── App / Middleware ───────────────────────────────────────────────────────
@asynccontextmanager
async def lifespan(app: FastAPI):
UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
conn = get_db()
init_db(conn)
conn.close()
yield
app = FastAPI(title="Canteen Asset Tracker", lifespan=lifespan)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# ─── Auth Middleware ──────────────────────────────────────────────────────────
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
"""Require valid Bearer token for all /api/* routes except login.
Falls back to admin user if no token provided (field tool mode)."""
path = request.url.path
# Skip auth enforcement in test mode
if os.environ.get("CANTEEN_SKIP_AUTH") == "1":
return await call_next(request)
# Public paths — no auth required
if not path.startswith("/api/") or path == "/api/auth/login":
return await call_next(request)
# Extract optional token
auth_header = request.headers.get("Authorization", "")
if auth_header.startswith("Bearer "):
token = auth_header[7:]
conn = get_db()
row = conn.execute(
"SELECT u.id, u.username, u.role FROM users u "
"JOIN sessions s ON u.id = s.user_id WHERE s.token = ? AND s.expires_at > datetime('now')",
(token,),
).fetchone()
conn.close()
if row is not None:
request.state.current_user = {
"id": row["id"],
"username": row["username"],
"role": row["role"],
}
return await call_next(request)
# No valid token — use default admin user (field tool mode)
request.state.current_user = {
"id": 1,
"username": "admin",
"role": "admin",
}
return await call_next(request)
# ─── Global Error Handling ───────────────────────────────────────────────────
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
"""Return structured JSON with status detail for all HTTP exceptions."""
return JSONResponse(
status_code=exc.status_code,
content={"detail": exc.detail},
)
@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
"""Catch-all for unhandled exceptions — log and return 500."""
import traceback
traceback.print_exc()
return JSONResponse(
status_code=500,
content={"detail": "Internal server error"},
)
# ─── Input sanitization helpers ──────────────────────────────────────────────
def _sanitize_machine_id(machine_id: str) -> str:
"""Strip whitespace and reject empty machine IDs."""
clean = machine_id.strip()
if not clean:
raise HTTPException(status_code=422, detail="Machine ID must not be empty")
if len(clean) > 256:
raise HTTPException(status_code=422, detail="Machine ID too long (max 256 chars)")
return clean
def _sanitize_name(name: str) -> str:
"""Trim name and reject empty/massive names."""
clean = name.strip()
if not clean:
raise HTTPException(status_code=422, detail="Name must not be empty")
if len(clean) > 512:
raise HTTPException(status_code=422, detail="Name too long (max 512 chars)")
return clean
# ─── Pydantic Models ────────────────────────────────────────────────────────
class AssetKey(BaseModel):
key_name: str
key_type: Optional[str] = ""
class AssetBadge(BaseModel):
badge_name: str
class AssetCreate(BaseModel):
machine_id: str
name: str
serial_number: Optional[str] = ""
description: Optional[str] = ""
category: Optional[str] = "Other"
status: Optional[str] = "active"
make: Optional[str] = ""
model: Optional[str] = ""
address: Optional[str] = ""
building_name: Optional[str] = ""
building_number: Optional[str] = ""
floor: Optional[str] = ""
room: Optional[str] = ""
trailer_number: Optional[str] = ""
walking_directions: Optional[str] = ""
map_link: Optional[str] = ""
parking_location: Optional[str] = ""
photo_path: Optional[str] = None
customer_id: Optional[int] = None
location_id: Optional[int] = None
assigned_to: Optional[int] = None
latitude: Optional[float] = None
longitude: Optional[float] = None
geofence_radius_meters: Optional[int] = 50
is_disney: Optional[bool] = None
keys: Optional[List[AssetKey]] = []
badges: Optional[List[str]] = []
dex_report_date: Optional[str] = None
class AssetUpdate(BaseModel):
machine_id: Optional[str] = None
name: Optional[str] = None
serial_number: Optional[str] = None
description: Optional[str] = None
category: Optional[str] = None
status: Optional[str] = None
make: Optional[str] = None
model: Optional[str] = None
address: Optional[str] = None
building_name: Optional[str] = None
building_number: Optional[str] = None
floor: Optional[str] = None
room: Optional[str] = None
trailer_number: Optional[str] = None
walking_directions: Optional[str] = None
map_link: Optional[str] = None
parking_location: Optional[str] = None
photo_path: Optional[str] = None
customer_id: Optional[int] = None
location_id: Optional[int] = None
assigned_to: Optional[int] = None
latitude: Optional[float] = None
longitude: Optional[float] = None
geofence_radius_meters: Optional[int] = None
is_disney: Optional[bool] = None
keys: Optional[List[AssetKey]] = None
badges: Optional[List[str]] = None
class CheckinCreate(BaseModel):
asset_id: int
latitude: Optional[float] = None
longitude: Optional[float] = None
accuracy: Optional[float] = None
photo_path: Optional[str] = None
notes: Optional[str] = ""
user_id: Optional[int] = None
class CheckinUpdate(BaseModel):
latitude: Optional[float] = None
longitude: Optional[float] = None
accuracy: Optional[float] = None
photo_path: Optional[str] = None
notes: Optional[str] = None
user_id: Optional[int] = None
class VisitCreate(BaseModel):
user_id: Optional[int] = None
asset_id: int
latitude: Optional[float] = None
longitude: Optional[float] = None
duration_minutes: Optional[int] = None
# ─── Helpers ────────────────────────────────────────────────────────────────
def row_to_dict(row: sqlite3.Row) -> dict:
return dict(row)
def _geofence_row(row: sqlite3.Row, conn: Optional[sqlite3.Connection] = None) -> dict:
"""Serialize a geofence row, parsing points JSON and including assigned users."""
d = dict(row)
if isinstance(d.get("points"), str):
try:
d["points"] = _json.loads(d["points"])
except (_json.JSONDecodeError, TypeError):
pass
# Include assigned users
close_conn = False
if conn is None:
conn = get_db()
close_conn = True
try:
user_rows = conn.execute(
"""SELECT u.id, u.username, u.role FROM geofence_users gu
JOIN users u ON u.id = gu.user_id
WHERE gu.geofence_id = ? ORDER BY u.username""",
(d["id"],),
).fetchall()
d["assigned_users"] = [dict(r) for r in user_rows]
finally:
if close_conn:
conn.close()
return d
def _sync_geofence_users(conn: sqlite3.Connection, geofence_id: int, user_ids: list[int]):
"""Replace assigned users for a geofence with the given list of user IDs.
Validates all user_ids exist before modifying the junction table.
Does NOT commit — caller manages transaction boundaries.
"""
# Validate all user IDs exist
if user_ids:
placeholders = ", ".join(["?"] * len(user_ids))
existing = conn.execute(
f"SELECT id FROM users WHERE id IN ({placeholders})",
user_ids,
).fetchall()
existing_ids = {r["id"] for r in existing}
for uid in user_ids:
if uid not in existing_ids:
raise HTTPException(
status_code=422,
detail=f"User with id {uid} not found",
)
conn.execute("DELETE FROM geofence_users WHERE geofence_id = ?", (geofence_id,))
for uid in user_ids:
conn.execute(
"INSERT INTO geofence_users (geofence_id, user_id) VALUES (?, ?)",
(geofence_id, uid),
)
def _validate_category(category: str):
if category not in VALID_CATEGORIES:
raise HTTPException(
status_code=422,
detail=f"Invalid category '{category}'. Must be one of: {', '.join(sorted(VALID_CATEGORIES))}",
)
def _validate_status(status: str):
if status not in VALID_STATUSES:
raise HTTPException(
status_code=422,
detail=f"Invalid status '{status}'. Must be one of: {', '.join(sorted(VALID_STATUSES))}",
)
def _validate_ref(conn: sqlite3.Connection, table: str, id_val: int, label: str):
"""Validate a foreign key reference exists."""
if id_val is not None:
row = conn.execute(f"SELECT id FROM {table} WHERE id = ?", (id_val,)).fetchone()
if row is None:
raise HTTPException(status_code=422, detail=f"{label} with id {id_val} not found")
def _validate_enum_table(conn: sqlite3.Connection, table: str, value: str, label: str):
"""Validate that a value exists in a lookup table (name column)."""
if value:
row = conn.execute(
f"SELECT id FROM {table} WHERE name = ?", (value,)
).fetchone()
if row is None:
raise HTTPException(
status_code=422,
detail=f"Invalid {label} '{value}'. Must exist in {table} table.",
)
# ─── Task 3: Health ────────────────────────────────────────────────────────
@app.get("/health")
def health():
return {"status": "ok"}
# ─── Customer & Location listing (for filter dropdowns) ─────────────────────
@app.get("/api/customers")
def list_customers():
conn = get_db()
rows = conn.execute(
"SELECT id, name FROM customers ORDER BY name"
).fetchall()
conn.close()
return [dict(r) for r in rows]
@app.get("/api/locations")
def list_locations(customer_id: Optional[int] = Query(None)):
conn = get_db()
if customer_id:
rows = conn.execute(
"SELECT id, customer_id, name FROM locations WHERE customer_id = ? ORDER BY name",
(customer_id,),
).fetchall()
else:
rows = conn.execute(
"SELECT id, customer_id, name FROM locations ORDER BY customer_id, name"
).fetchall()
conn.close()
return [dict(r) for r in rows]
@app.get("/api/users")
def list_users():
conn = get_db()
rows = conn.execute(
"SELECT id, username, role FROM users ORDER BY username"
).fetchall()
conn.close()
return [dict(r) for r in rows]
@app.get("/api/categories")
def list_categories():
conn = get_db()
rows = conn.execute(
"SELECT DISTINCT category FROM assets WHERE category IS NOT NULL AND category != '' ORDER BY category"
).fetchall()
conn.close()
return [r["category"] for r in rows]
@app.get("/api/makes")
def list_makes():
conn = get_db()
rows = conn.execute(
"SELECT DISTINCT make FROM assets WHERE make IS NOT NULL AND make != '' ORDER BY make"
).fetchall()
conn.close()
return [r["make"] for r in rows]
# ─── Task 4: POST /api/assets ──────────────────────────────────────────────
def _build_asset_insert(body: AssetCreate, machine_id: str, name: str):
"""Build column names and values tuple for asset INSERT."""
columns = [
"machine_id", "serial_number", "name", "description", "category", "status",
"make", "model", "address", "building_name", "building_number", "floor",
"room", "trailer_number", "walking_directions", "map_link",
"parking_location", "photo_path", "customer_id", "location_id",
"assigned_to", "latitude", "longitude", "geofence_radius_meters", "is_disney",
"dex_report_date",
]
values = (
machine_id, body.serial_number or "", name, body.description or "",
body.category or "Other", body.status or "active",
body.make or "", body.model or "", body.address or "",
body.building_name or "", body.building_number or "",
body.floor or "", body.room or "", body.trailer_number or "",
body.walking_directions or "", body.map_link or "",
body.parking_location or "", body.photo_path,
body.customer_id, body.location_id, body.assigned_to,
body.latitude, body.longitude,
body.geofence_radius_meters if body.geofence_radius_meters is not None else 50,
body.is_disney if body.is_disney is not None else None,
body.dex_report_date,
)
return columns, values
def _get_asset_keys(conn: sqlite3.Connection, asset_id: int) -> list:
"""Load all keys for an asset."""
rows = conn.execute(
"SELECT id, asset_id, key_name, key_type FROM asset_keys WHERE asset_id = ?",
(asset_id,),
).fetchall()
return [row_to_dict(r) for r in rows]
def _get_asset_badges(conn: sqlite3.Connection, asset_id: int) -> list:
"""Load all badges for an asset."""
rows = conn.execute(
"SELECT id, asset_id, badge_name FROM asset_badges WHERE asset_id = ?",
(asset_id,),
).fetchall()
return [row_to_dict(r) for r in rows]
@app.post("/api/assets", status_code=201)
def create_asset(body: AssetCreate):
machine_id = _sanitize_machine_id(body.machine_id)
name = _sanitize_name(body.name)
_validate_status(body.status or "active")
conn = get_db()
# DB-based validation for reference fields
_validate_enum_table(conn, "categories", body.category or "Other", "category")
_validate_enum_table(conn, "makes", body.make or "", "make")
if body.customer_id is not None:
_validate_ref(conn, "customers", body.customer_id, "Customer")
if body.location_id is not None:
_validate_ref(conn, "locations", body.location_id, "Location")
if body.assigned_to is not None:
_validate_ref(conn, "users", body.assigned_to, "User")
# Auto-populate address from GPS coordinates
if body.latitude is not None and body.longitude is not None and not (body.address or "").strip():
geo = reverse_geocode(body.latitude, body.longitude)
if geo:
body.address = geo.get("address", "")
body.building_name = geo.get("building_name", body.building_name or "")
body.building_number = geo.get("building_number", body.building_number or "")
columns, values = _build_asset_insert(body, machine_id, name)
placeholders = ", ".join(["?"] * len(columns))
col_names = ", ".join(columns)
try:
cursor = conn.execute(
f"INSERT INTO assets ({col_names}) VALUES ({placeholders})",
values,
)
except sqlite3.IntegrityError:
conn.close()
raise HTTPException(
status_code=409,
detail=f"Asset with machine_id '{machine_id}' already exists",
)
asset_id = cursor.lastrowid
# Insert keys
if body.keys:
for k in body.keys:
conn.execute(
"INSERT INTO asset_keys (asset_id, key_name, key_type) VALUES (?, ?, ?)",
(asset_id, k.key_name, k.key_type or ""),
)
# Insert badges
if body.badges:
for b_name in body.badges:
conn.execute(
"INSERT INTO asset_badges (asset_id, badge_name) VALUES (?, ?)",
(asset_id, b_name),
)
conn.commit()
_log_activity(conn, "created", "asset", asset_id,
f"Asset '{name}' (machine_id: {machine_id}) created")
conn.commit()
row = conn.execute("SELECT * FROM assets WHERE id = ?", (asset_id,)).fetchone()
result = row_to_dict(row)
result["keys"] = _get_asset_keys(conn, asset_id)
result["badges"] = _get_asset_badges(conn, asset_id)
conn.close()
return result
# ─── Task 5: GET /api/assets ───────────────────────────────────────────────
@app.get("/api/assets")
def list_assets(
category: Optional[str] = Query(None),
status: Optional[str] = Query(None),
make: Optional[str] = Query(None),
model: Optional[str] = Query(None),
customer_id: Optional[int] = Query(None),
location_id: Optional[int] = Query(None),
assigned_to: Optional[int] = Query(None),
disney_park: Optional[str] = Query(None),
disney_filter: Optional[str] = Query(None),
disney_group: Optional[str] = Query(None, description="Grouped park categories: park, resort, office, springs, other"),
q: Optional[str] = Query(None),
no_dex_days: Optional[int] = Query(None, ge=1),
has_gps: Optional[bool] = Query(None),
branch: Optional[str] = Query(None, description="Filter by branch/territory name"),
limit: int = Query(100, ge=1, le=2000),
offset: int = Query(0, ge=0),
):
conn = get_db()
conditions = []
params = []
if category:
conditions.append("category = ?")
params.append(category)
if status:
conditions.append("status = ?")
params.append(status)
if make:
conditions.append("make = ?")
params.append(make)
if disney_park:
conditions.append("disney_park = ?")
params.append(disney_park)
if disney_filter == "disney":
conditions.append("is_disney = 1")
elif disney_filter == "non_disney":
conditions.append("(is_disney IS NULL OR is_disney = 0)")
if disney_group:
park_groups = {
"park": ["magic-kingdom", "epcot", "hollywood-studios", "animal-kingdom"],
"resort": ["resort"],
"office": ["office"],
"springs": ["disney-springs"],
"other": ["other"],
}
parks = park_groups.get(disney_group)
if parks:
placeholders = ",".join("?" * len(parks))
conditions.append(f"disney_park IN ({placeholders})")
params.extend(parks)
if model:
conditions.append("model = ?")
params.append(model)
if customer_id is not None:
conditions.append("customer_id = ?")
params.append(customer_id)
if location_id is not None:
conditions.append("location_id = ?")
params.append(location_id)
if assigned_to is not None:
conditions.append("assigned_to = ?")
params.append(assigned_to)
if q:
conditions.append("(a.name LIKE ? OR a.machine_id LIKE ? OR a.serial_number LIKE ? OR a.description LIKE ? OR a.company LIKE ?)")
like = f"%{q}%"
params.extend([like, like, like, like, like])
if no_dex_days is not None:
conditions.append("(dex_report_date IS NULL OR REPLACE(dex_report_date, 'T', ' ') < datetime('now', '-' || ? || ' days'))")
params.append(no_dex_days)
if has_gps is not None:
if has_gps:
conditions.append("a.latitude IS NOT NULL AND a.longitude IS NOT NULL")
else:
conditions.append("(a.latitude IS NULL OR a.longitude IS NULL)")
if branch:
cache = _build_asset_branch_cache()
# Find all connect_id prefixes that map to this branch name
matching_prefixes = [
pfx for pfx, br in cache.items() if br == branch
]
if matching_prefixes:
placeholders = ",".join(["?"] * len(matching_prefixes))
conditions.append(
f"SUBSTR(a.connect_id, 1, 8) IN ({placeholders})"
)
params.extend(matching_prefixes)
else:
# Branch name not found — return empty result
conditions.append("1=0")
where = " AND ".join(conditions)
from_clause = "FROM assets a LEFT JOIN customers c ON a.customer_id = c.id"
sql = f"SELECT a.*, COALESCE(c.name, a.customer_name) AS customer_name {from_clause}"
if where:
sql += f" WHERE {where}"
sql += " ORDER BY a.created_at DESC LIMIT ? OFFSET ?"
params.extend([limit, offset])
rows = conn.execute(sql, params).fetchall()
# Get total count for pagination (use same FROM with alias for unambiguous conditions)
count_sql = f"SELECT COUNT(a.id) {from_clause}"
if where:
count_sql += f" WHERE {where}"
total = conn.execute(count_sql, params[:len(params)-2]).fetchone()[0]
conn.close()
response = JSONResponse([row_to_dict(r) for r in rows])
response.headers["X-Total-Count"] = str(total)
return response
@app.get("/api/assets/search")
def search_by_machine_id(machine_id: str = Query(...)):
conn = get_db()
row = conn.execute(
"SELECT * FROM assets WHERE machine_id = ? OR connect_id = ? OR serial_number = ?",
(machine_id, machine_id, machine_id),
).fetchone()
conn.close()
if row is None:
raise HTTPException(status_code=404, detail="Asset not found")
return row_to_dict(row)
# ── Task 5b: GET /api/branches ─────────────────────────────────────────────
@app.get("/api/branches")
def list_branches():
"""Return distinct branch/territory names that assets belong to.
Uses the connect_id prefix → branch mapping from the extraction DB.
Returns sorted list of {name, machine_count} objects.
"""
cache = _build_asset_branch_cache()
# Count how many assets per branch
conn = get_db()
prefix_counts: dict[str, int] = {}
try:
cur = conn.execute(
"SELECT SUBSTR(connect_id, 1, 8) AS prefix, COUNT(*) AS cnt "
"FROM assets WHERE connect_id IS NOT NULL AND connect_id != '' "
"GROUP BY prefix"
)
for r in cur.fetchall():
prefix_counts[r[0]] = r[1]
finally:
conn.close()
# Aggregate by branch name
branch_map: dict[str, int] = {}
for prefix, machine_count in prefix_counts.items():
branch = cache.get(prefix, prefix)
branch_map[branch] = branch_map.get(branch, 0) + machine_count
branches = sorted(
[{"name": k, "machine_count": v} for k, v in branch_map.items()],
key=lambda b: -b["machine_count"], # type: ignore[arg-type]
)
return {"branches": branches, "total": len(branches)}
@app.get("/api/assets/{asset_id}")
def get_asset(asset_id: int):
conn = get_db()
row = conn.execute("SELECT * FROM assets WHERE id = ?", (asset_id,)).fetchone()
if row is None:
conn.close()
raise HTTPException(status_code=404, detail="Asset not found")
result = row_to_dict(row)
result["keys"] = _get_asset_keys(conn, asset_id)
result["badges"] = _get_asset_badges(conn, asset_id)
# Add joined names
if result.get("customer_id"):
cust = conn.execute(
"SELECT name FROM customers WHERE id = ?", (result["customer_id"],)
).fetchone()
result["customer_name"] = cust["name"] if cust else None
if result.get("location_id"):
loc = conn.execute(
"SELECT name FROM locations WHERE id = ?", (result["location_id"],)
).fetchone()
result["location_name"] = loc["name"] if loc else None
# Enrich with MSFS data (from Dynamics 365 Field Service)
msfs = _MSFS_LOOKUP.get(result.get("machine_id", "")) or _MSFS_BY_ID.get(asset_id)
if msfs:
result["msfs"] = {
"equipment_id": msfs.get("msfs_equipment_id"),
"connect_id": msfs.get("msfs_connect_id"),
"compass_asset_number": msfs.get("msfs_compass_asset_number"),
"manufacturer": msfs.get("msfs_manufacturer"),
"model_text": msfs.get("msfs_model_text"),
"dex_model": msfs.get("msfs_dex_model"),
"install_date": msfs.get("msfs_install_date"),
"manufacture_date": msfs.get("msfs_manufacture_date"),
"software_version": msfs.get("msfs_software_version"),
"canteen_connect_guid": msfs.get("msfs_canteen_connect_guid"),
"account_name": msfs.get("msfs_account_name"),
"work_order_count": msfs.get("work_order_count", 0),
"work_orders": msfs.get("work_orders", []),
}
# Enrich with Seed data (from mycantaloupe.com)
seed_row = conn.execute(
"SELECT raw_data, imported_at FROM seed_data WHERE asset_id = ?",
(asset_id,),
).fetchone()
if seed_row:
try:
result["seed"] = _json.loads(seed_row["raw_data"])
result["seed_imported_at"] = seed_row["imported_at"]
except Exception:
result["seed"] = None
conn.close()
return result
# ─── Navigation endpoint ────────────────────────────────────────────────────
@app.get("/api/assets/{asset_id}/navigation")
def get_navigation(
asset_id: int,
lat: float = Query(...),
lng: float = Query(...),
mode: str = Query("driving", pattern="^(driving|walking)$"),
):
"""Return navigation info from GPS to asset using OSRM (falls back to Haversine).
Accepts optional mode: 'driving' (default) or 'walking'.
On success returns route_coords, route_duration_s, walking_directions, route_mode
plus the standard distance/bearing fields.
"""
import math
conn = get_db()
row = conn.execute(
"SELECT id, name, latitude, longitude, machine_id, address, walking_directions FROM assets WHERE id = ?",
(asset_id,),
).fetchone()
conn.close()
if row is None:
raise HTTPException(status_code=404, detail="Asset not found")
if row["latitude"] is None or row["longitude"] is None:
raise HTTPException(
status_code=400,
detail="Asset has no GPS coordinates set",
)
dest_lat, dest_lng = row["latitude"], row["longitude"]
# ── Try OSRM first ──────────────────────────────────────────────────
osrm_profile = "driving" if mode == "driving" else "foot"
route_coords = None
route_duration_s = None
route_distance_m = None
walking_directions = None
try:
url = (
f"https://router.project-osrm.org/route/v1/{osrm_profile}/{lng},{lat};{dest_lng},{dest_lat}"
"?overview=full&geometries=geojson&steps=true&alternatives=false"
)
req = urllib.request.Request(url, headers={"User-Agent": "CanteenAssetTracker/1.0"})
with urllib.request.urlopen(req, timeout=10) as resp:
data = _json.loads(resp.read().decode())
if data.get("code") == "Ok" and data.get("routes"):
route = data["routes"][0]
coords_raw = route.get("geometry", {}).get("coordinates", [])
route_coords = [{"lat": c[1], "lng": c[0]} for c in coords_raw]
route_duration_s = round(route.get("duration", 0))
route_distance_m = round(route.get("distance", 0))
# Build step-by-step walking directions (only for walking mode)
walking_directions = None
if mode == "walking":
steps = []
for leg in route.get("legs", []):
for step in leg.get("steps", []):
steps.append({
"instruction": step.get("maneuver", {}).get("instruction", step.get("name", "")),
"distance_m": round(step.get("distance", 0)),
"duration_s": round(step.get("duration", 0)),
})
if steps:
walking_directions = steps
except Exception:
# OSRM failed — fall through to Haversine below
pass
# ── Haversine distance (always computed for fallback / reference) ────
R = 6371000 # Earth radius in meters
phi1 = math.radians(lat)
phi2 = math.radians(dest_lat)
dphi = math.radians(dest_lat - lat)
dlambda = math.radians(dest_lng - lng)
a_val = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2) ** 2
c_val = 2 * math.atan2(math.sqrt(a_val), math.sqrt(1 - a_val))
haversine_distance_m = R * c_val
# Bearing
y = math.sin(dlambda) * math.cos(phi2)
x = math.cos(phi1) * math.sin(phi2) - math.sin(phi1) * math.cos(phi2) * math.cos(dlambda)
bearing = (math.degrees(math.atan2(y, x)) + 360) % 360
# Cardinal direction
dirs = ["N", "NE", "E", "SE", "S", "SW", "W", "NW"]
idx = round(bearing / 45) % 8
cardinal = dirs[idx]
result = {
"asset_id": asset_id,
"asset_name": row["name"],
"asset_lat": dest_lat,
"asset_lng": dest_lng,
"asset_address": row["address"] or None,
"origin_lat": lat,
"origin_lng": lng,
"distance_meters": round(route_distance_m if route_distance_m else haversine_distance_m, 1),
"distance_km": round((route_distance_m if route_distance_m else haversine_distance_m) / 1000, 2),
"bearing": round(bearing, 1),
"cardinal": cardinal,
"route_mode": mode,
"google_maps_url": (
f"https://www.google.com/maps/dir/?api=1"
f"&origin={lat},{lng}"
f"&destination={dest_lat},{dest_lng}"
),
}
if route_coords:
result["route_coords"] = route_coords
result["route_duration_s"] = route_duration_s
result["route_distance_m"] = route_distance_m
result["route_mode"] = mode
result["osrm_source"] = "osrm"
if mode == "walking" and walking_directions:
result["walking_directions"] = walking_directions
else:
result["osrm_source"] = "haversine"
result["walking_directions"] = row["walking_directions"] or None
return result
# ─── Task 6: PUT / DELETE /api/assets/{id} ─────────────────────────────────
_TEXT_FIELDS = [
"name", "machine_id", "serial_number", "description", "make", "model",
"address", "building_name", "building_number", "floor", "room",
"trailer_number", "walking_directions", "map_link", "parking_location",
]
@app.put("/api/assets/{asset_id}")
def update_asset(asset_id: int, body: AssetUpdate):
conn = get_db()
existing = conn.execute("SELECT * FROM assets WHERE id = ?", (asset_id,)).fetchone()
if existing is None:
conn.close()
raise HTTPException(status_code=404, detail="Asset not found")
updates = {}
for field in _TEXT_FIELDS:
val = getattr(body, field, None)
if val is not None:
if field == "name":
updates[field] = _sanitize_name(val)
elif field == "machine_id":
updates[field] = _sanitize_machine_id(val)
else:
updates[field] = val
if body.category is not None:
_validate_enum_table(conn, "categories", body.category, "category")
updates["category"] = body.category
if body.status is not None:
_validate_status(body.status)
updates["status"] = body.status
if body.photo_path is not None:
updates["photo_path"] = body.photo_path
if body.customer_id is not None:
updates["customer_id"] = body.customer_id
if body.location_id is not None:
updates["location_id"] = body.location_id
if body.assigned_to is not None:
updates["assigned_to"] = body.assigned_to
if body.latitude is not None:
updates["latitude"] = body.latitude
if body.longitude is not None:
updates["longitude"] = body.longitude
if body.geofence_radius_meters is not None:
updates["geofence_radius_meters"] = body.geofence_radius_meters
if body.is_disney is not None:
updates["is_disney"] = 1 if body.is_disney else 0
# Auto-populate address from GPS when lat/lng are provided
lat = updates.get("latitude") if "latitude" in updates else None
lng = updates.get("longitude") if "longitude" in updates else None
if lat is not None and lng is not None:
geo = reverse_geocode(lat, lng)
if geo:
updates["address"] = geo.get("address", "")
updates["building_name"] = geo.get("building_name", "")
updates["building_number"] = geo.get("building_number", "")
if updates:
updates["updated_at"] = "datetime('now')"
set_clause = ", ".join(
f"{k} = {v}" if k == "updated_at" else f"{k} = ?"
for k, v in updates.items()
)
values = [v for k, v in updates.items() if k != "updated_at"]
conn.execute(
f"UPDATE assets SET {set_clause} WHERE id = ?",
values + [asset_id],
)
conn.commit()
row = conn.execute("SELECT * FROM assets WHERE id = ?", (asset_id,)).fetchone()
_log_activity(conn, "updated", "asset", asset_id,
f"Asset '{row['name']}' updated")
conn.commit()
conn.close()
return row_to_dict(row)
@app.delete("/api/assets/{asset_id}", status_code=204)
def delete_asset(asset_id: int):
conn = get_db()
existing = conn.execute("SELECT id FROM assets WHERE id = ?", (asset_id,)).fetchone()
if existing is None:
conn.close()
raise HTTPException(status_code=404, detail="Asset not found")
# Delete dependent rows first (visits, asset_keys, asset_badges lack ON DELETE CASCADE)
conn.execute("DELETE FROM visits WHERE asset_id = ?", (asset_id,))
conn.execute("DELETE FROM asset_keys WHERE asset_id = ?", (asset_id,))
conn.execute("DELETE FROM asset_badges WHERE asset_id = ?", (asset_id,))
conn.execute("DELETE FROM assets WHERE id = ?", (asset_id,))
_log_activity(conn, "deleted", "asset", asset_id,
f"Asset {asset_id} deleted")
conn.commit()
conn.close()
# ─── Task 8: POST /api/checkins ─────────────────────────────────────────────
@app.post("/api/checkins", status_code=201)
def create_checkin(body: CheckinCreate, request: Request):
conn = get_db()
existing = conn.execute("SELECT id FROM assets WHERE id = ?", (body.asset_id,)).fetchone()
if existing is None:
conn.close()
raise HTTPException(status_code=404, detail="Asset not found")
# Auto-resolve user from auth token if not provided in body
user_id = body.user_id
if user_id is None:
user_id = request.state.current_user.get("id")
cursor = conn.execute(
"""INSERT INTO checkins (asset_id, user_id, latitude, longitude, accuracy, photo_path, notes)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(body.asset_id, user_id, body.latitude, body.longitude,
body.accuracy, body.photo_path, body.notes or ""),
)
conn.commit()
checkin_id = cursor.lastrowid
# Auto-update asset's last known location and address
if body.latitude is not None and body.longitude is not None:
conn.execute(
"UPDATE assets SET latitude = ?, longitude = ? WHERE id = ?",
(body.latitude, body.longitude, body.asset_id),
)
geo = reverse_geocode(body.latitude, body.longitude)
if geo:
conn.execute(
"UPDATE assets SET address = ?, building_name = ?, building_number = ? WHERE id = ?",
(geo.get("address", ""), geo.get("building_name", ""), geo.get("building_number", ""), body.asset_id),
)
conn.commit()
# Auto-log visit
row = conn.execute("SELECT created_at FROM checkins WHERE id = ?", (checkin_id,)).fetchone()
_auto_log_visit(conn, user_id, body.asset_id, row["created_at"])
# Activity log — include machine_id and name for clarity
asset_info = conn.execute(
"SELECT machine_id, name FROM assets WHERE id = ?",
(body.asset_id,)
).fetchone()
asset_label = f"asset {body.asset_id}"
if asset_info:
mid = asset_info["machine_id"] or ""
nm = asset_info["name"] or ""
if mid and nm:
asset_label = f"{nm} (#{mid})"
elif mid:
asset_label = f"asset #{mid}"
_log_activity(conn, "created", "checkin", checkin_id,
f"Check-in for {asset_label}",
user_id=user_id)
conn.commit()
row = conn.execute("SELECT * FROM checkins WHERE id = ?", (checkin_id,)).fetchone()
conn.close()
return row_to_dict(row)
# ─── Task 9: GET /api/checkins ──────────────────────────────────────────────
@app.get("/api/checkins")
def list_checkins(
asset_id: Optional[int] = Query(None),
user_id: Optional[int] = Query(None),
limit: int = Query(100, ge=1, le=1000),
offset: int = Query(0, ge=0),
):
conn = get_db()
conditions = []
params = []
if asset_id is not None:
conditions.append("c.asset_id = ?")
params.append(asset_id)
if user_id is not None:
conditions.append("c.user_id = ?")
params.append(user_id)
where = " AND ".join(conditions)
sql = "SELECT c.*, u.username FROM checkins c LEFT JOIN users u ON c.user_id = u.id"
if where:
sql += f" WHERE {where}"
sql += " ORDER BY c.created_at DESC, c.id DESC LIMIT ? OFFSET ?"
params.extend([limit, offset])
rows = conn.execute(sql, params).fetchall()
conn.close()
return [row_to_dict(r) for r in rows]
@app.get("/api/checkins/{checkin_id}")
def get_checkin(checkin_id: int):
conn = get_db()
row = conn.execute("SELECT c.*, u.username FROM checkins c LEFT JOIN users u ON c.user_id = u.id WHERE c.id = ?", (checkin_id,)).fetchone()
if row is None:
conn.close()
raise HTTPException(status_code=404, detail="Checkin not found")
conn.close()
return row_to_dict(row)
@app.put("/api/checkins/{checkin_id}")
def update_checkin(checkin_id: int, body: CheckinUpdate):
conn = get_db()
existing = conn.execute("SELECT id FROM checkins WHERE id = ?", (checkin_id,)).fetchone()
if existing is None:
conn.close()
raise HTTPException(status_code=404, detail="Checkin not found")
updates = {}
for field in ("latitude", "longitude", "accuracy", "photo_path", "notes", "user_id"):
val = getattr(body, field, None)
if val is not None:
updates[field] = val
if updates:
set_clause = ", ".join(f"{k} = ?" for k in updates)
values = list(updates.values())
conn.execute(
f"UPDATE checkins SET {set_clause} WHERE id = ?",
values + [checkin_id],
)
conn.commit()
row = conn.execute("SELECT * FROM checkins WHERE id = ?", (checkin_id,)).fetchone()
conn.close()
return row_to_dict(row)
@app.delete("/api/checkins/{checkin_id}", status_code=204)
def delete_checkin(checkin_id: int):
conn = get_db()
existing = conn.execute("SELECT id FROM checkins WHERE id = ?", (checkin_id,)).fetchone()
if existing is None:
conn.close()
raise HTTPException(status_code=404, detail="Checkin not found")
conn.execute("DELETE FROM checkins WHERE id = ?", (checkin_id,))
conn.commit()
conn.close()
# ─── Task 10: GET /api/stats ────────────────────────────────────────────────
# ─── Phase C: Helpers ────────────────────────────────────────────────────────
VALID_ROLES = {"admin", "technician", "readonly"}
def _hash_password(password: str) -> str:
"""Simple SHA-256 password hashing."""
return hashlib.sha256(password.encode()).hexdigest()
def _generate_token() -> str:
"""Generate a random session token."""
return secrets.token_hex(32)
def _log_activity(conn: sqlite3.Connection, action: str, entity_type: str,
entity_id: int, details: str = "", user_id: int = None):
"""Insert an activity log entry."""
conn.execute(
"""INSERT INTO activity_log (user_id, action, entity_type, entity_id, details)
VALUES (?, ?, ?, ?, ?)""",
(user_id, action, entity_type, entity_id, details),
)
def _user_to_dict(row: sqlite3.Row) -> dict:
"""Convert user row to dict, excluding password_hash."""
d = row_to_dict(row)
d.pop("password_hash", None)
return d
class LoginRequest(BaseModel):
username: str
password: str
remember_me: bool = False
# ─── Phase C: Auth API ───────────────────────────────────────────────────────
@app.post("/api/auth/login")
def login(body: LoginRequest):
conn = get_db()
# Clean up any expired sessions for this user
conn.execute(
"DELETE FROM sessions WHERE user_id = ? AND expires_at < datetime('now')",
(body.username,),
)
row = conn.execute(
"SELECT * FROM users WHERE username = ?", (body.username,)
).fetchone()
if row is None:
conn.close()
raise HTTPException(status_code=401, detail="Invalid username or password")
password_hash = _hash_password(body.password)
if password_hash != row["password_hash"]:
conn.close()
raise HTTPException(status_code=401, detail="Invalid username or password")
token = _generate_token()
# Set expiry based on remember_me
if body.remember_me:
expires = "datetime('now', '+30 days')"
else:
expires = "datetime('now', '+1 day')"
conn.execute(
"INSERT INTO sessions (user_id, token, expires_at) VALUES (?, ?, " + expires + ")",
(row["id"], token),
)
conn.commit()
conn.close()
result = _user_to_dict(row)
result["token"] = token
return result
@app.get("/api/auth/me")
def auth_me(request: Request):
"""Return the current authenticated user from the Authorization header."""
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid Authorization header")
token = auth_header[7:]
conn = get_db()
row = conn.execute(
"SELECT u.* FROM users u JOIN sessions s ON u.id = s.user_id WHERE s.token = ? AND s.expires_at > datetime('now')",
(token,),
).fetchone()
conn.close()
if row is None:
raise HTTPException(status_code=401, detail="Invalid or expired token")
return _user_to_dict(row)
@app.post("/api/auth/logout")
def logout(request: Request):
"""Invalidate the current session token."""
auth_header = request.headers.get("Authorization", "")
if auth_header.startswith("Bearer "):
token = auth_header[7:]
conn = get_db()
conn.execute("DELETE FROM sessions WHERE token = ?", (token,))
conn.commit()
conn.close()
return {"detail": "Logged out"}
# ─── Phase 0: Proximity Check ─────────────────────────────────────────────────
@app.get("/api/proximity")
def proximity_check(
lat: float = Query(...),
lng: float = Query(...),
radius_meters: int = Query(200, ge=1, le=50000),
):
"""
Return assets within radius_meters of (lat, lng), sorted by distance.
Uses Haversine formula for accurate spherical distance.
"""
conn = get_db()
rows = conn.execute("""
SELECT * FROM (
SELECT *, (
6371000 * acos(
cos(radians(?)) * cos(radians(latitude)) *
cos(radians(longitude) - radians(?)) +
sin(radians(?)) * sin(radians(latitude))
)
) AS distance_meters
FROM assets
WHERE latitude IS NOT NULL AND longitude IS NOT NULL
)
WHERE distance_meters <= ?
ORDER BY distance_meters
LIMIT 50
""", (lat, lng, lat, radius_meters)).fetchall()
conn.close()
results = [row_to_dict(r) for r in rows]
return results
# ─── Phase C: Visits API & Auto-visit Logging ────────────────────────────────
def _auto_log_visit(conn: sqlite3.Connection, user_id: int, asset_id: int,
checkin_time: str):
"""Check if a visit should be logged for recent check-ins at same asset by same user."""
if user_id is None:
return
# Find the most recent visit for this user+asset
prev = conn.execute(
"""SELECT id, checkin_time FROM visits
WHERE user_id = ? AND asset_id = ?
ORDER BY checkin_time DESC LIMIT 1""",
(user_id, asset_id),
).fetchone()
if prev is None:
# No prior visit — check if there are at least 2 check-ins in the window
rows = conn.execute(
"""SELECT id, created_at FROM checkins
WHERE user_id = ? AND asset_id = ?
ORDER BY created_at DESC LIMIT 2""",
(user_id, asset_id),
).fetchall()
if len(rows) >= 2:
t1 = rows[-1]["created_at"]
t2 = rows[0]["created_at"]
conn.execute(
"""INSERT INTO visits (user_id, asset_id, checkin_time, checkout_time)
VALUES (?, ?, ?, ?)""",
(user_id, asset_id, t1, t2),
)
else:
conn.execute(
"UPDATE visits SET checkout_time = ? WHERE id = ?",
(checkin_time, prev["id"]),
)
@app.post("/api/visits", status_code=201)
def create_visit(body: VisitCreate):
conn = get_db()
existing = conn.execute("SELECT id FROM assets WHERE id = ?", (body.asset_id,)).fetchone()
if existing is None:
conn.close()
raise HTTPException(status_code=404, detail="Asset not found")
cursor = conn.execute(
"""INSERT INTO visits (user_id, asset_id, checkin_time, checkout_time, duration_minutes)
VALUES (?, ?, datetime('now'), datetime('now'), ?)""",
(body.user_id, body.asset_id, body.duration_minutes or 0),
)
conn.commit()
visit_id = cursor.lastrowid
# Activity log
_log_activity(conn, "created", "visit", visit_id,
f"Visit logged for asset {body.asset_id}",
user_id=body.user_id)
conn.commit()
row = conn.execute("SELECT * FROM visits WHERE id = ?", (visit_id,)).fetchone()
conn.close()
return row_to_dict(row)
@app.get("/api/visits")
def list_visits(
asset_id: Optional[int] = Query(None),
user_id: Optional[int] = Query(None),
date_from: Optional[str] = Query(None),
date_to: Optional[str] = Query(None),
limit: int = Query(100, ge=1, le=1000),
offset: int = Query(0, ge=0),
):
conn = get_db()
conditions = []
params = []
if asset_id is not None:
conditions.append("v.asset_id = ?")
params.append(asset_id)
if user_id is not None:
conditions.append("v.user_id = ?")
params.append(user_id)
if date_from:
conditions.append("v.checkin_time >= ?")
params.append(date_from)
if date_to:
conditions.append("v.checkin_time <= ?")
params.append(date_to + " 23:59:59")
where = " AND ".join(conditions)
sql = """SELECT v.*, a.name AS asset_name, a.machine_id,
u.username AS user_name
FROM visits v
LEFT JOIN assets a ON v.asset_id = a.id
LEFT JOIN users u ON v.user_id = u.id"""
if where:
sql += f" WHERE {where}"
sql += " ORDER BY v.checkin_time DESC LIMIT ? OFFSET ?"
params.extend([limit, offset])
rows = conn.execute(sql, params).fetchall()
conn.close()
return [row_to_dict(r) for r in rows]
@app.get("/api/visits/stats")
def get_visit_stats():
conn = get_db()
total_visits = conn.execute("SELECT COUNT(*) FROM visits").fetchone()[0]
per_asset = conn.execute(
"""SELECT a.name, a.machine_id, COUNT(*) AS cnt
FROM visits v JOIN assets a ON v.asset_id = a.id
GROUP BY v.asset_id ORDER BY cnt DESC"""
).fetchall()
visits_per_asset = [
{"name": r["name"], "machine_id": r["machine_id"], "count": r["cnt"]}
for r in per_asset
]
per_user = conn.execute(
"""SELECT u.username, COUNT(*) AS visit_count,
SUM(
CASE WHEN v.checkout_time IS NOT NULL AND v.checkin_time IS NOT NULL
THEN (julianday(v.checkout_time) - julianday(v.checkin_time)) * 1440
ELSE 0 END
) AS total_minutes
FROM visits v JOIN users u ON v.user_id = u.id
GROUP BY v.user_id ORDER BY total_minutes DESC"""
).fetchall()
time_on_site = [
{"username": r["username"], "visit_count": r["visit_count"],
"total_minutes": round(r["total_minutes"] or 0, 1)}
for r in per_user
]
conn.close()
return {
"total_visits": total_visits,
"visits_per_asset": visits_per_asset,
"time_on_site": time_on_site,
}
# ─── File Uploads ───────────────────────────────────────────────────────────
ICON_MAX_SIZE = 2 * 1024 * 1024 # 2 MB
PHOTO_MAX_SIZE = 20 * 1024 * 1024 # 20 MB
ICON_ALLOWED_EXTS = {".png", ".jpg", ".jpeg", ".svg"}
PHOTO_ALLOWED_EXTS = {".png", ".jpg", ".jpeg", ".dng"}
def _dms_to_decimal(dms_tuple: tuple, ref: str) -> float:
"""Convert GPS DMS (degrees, minutes, seconds) tuple to decimal degrees."""
try:
degrees = float(dms_tuple[0])
minutes = float(dms_tuple[1])
seconds = float(dms_tuple[2])
decimal = degrees + minutes / 60.0 + seconds / 3600.0
if ref in ("S", "W"):
decimal = -decimal
return round(decimal, 6)
except (TypeError, ValueError, IndexError):
return None
def _extract_gps_from_bytes(image_bytes: bytes) -> dict | None:
"""Extract GPS coordinates from image EXIF data.
Reads the raw image bytes with PIL, extracts the GPS IFD,
and converts DMS coordinates to decimal degrees.
Returns {lat, lng} or None if no GPS data found.
"""
try:
img = PILImage.open(io.BytesIO(image_bytes))
exif = img.getexif()
# GPSInfo IFD tag = 0x8825 = 34853
gps_ifd = exif.get_ifd(0x8825)
except Exception:
return None
if not gps_ifd:
return None
# GPSLatitudeRef = tag 1, GPSLatitude = tag 2
# GPSLongitudeRef = tag 3, GPSLongitude = tag 4
lat_ref = gps_ifd.get(1)
lat_dms = gps_ifd.get(2)
lng_ref = gps_ifd.get(3)
lng_dms = gps_ifd.get(4)
if lat_dms is None or lng_dms is None:
return None
lat = _dms_to_decimal(lat_dms, lat_ref or "N")
lng = _dms_to_decimal(lng_dms, lng_ref or "E")
if lat is None or lng is None:
return None
return {"lat": lat, "lng": lng}
def _extract_serial_from_text(text: str) -> str | None:
"""Try to extract a plausible serial number from OCR/vision text.
Looks for explicit label prefixes (S/N, Serial#, ID#, Machine ID)
and returns the value portion. Avoids false-positive matches on
short numbers (< 6 chars) that are likely Connect-ID fragments.
Returns the raw serial value (with punctuation preserved) or None.
"""
if not text:
return None
patterns = [
r'(?:S/N|SN|SERIAL\s*NO|SERIAL|SERIAL\s*#)\s*[:=#]?\s*([A-Za-z0-9.\-/]{6,})',
r'(?:EQUIPMENT\s*ID|EQ\s*ID|ASSET\s*ID)\s*[:=]?\s*([A-Za-z0-9.\-/]{6,})',
r'(?:MODEL\s*NO|MODEL\s*#|PART\s*NO|P/N)\s*[:=]?\s*([A-Za-z0-9.\-/]{6,})',
]
for pat in patterns:
m = re.search(pat, text, re.IGNORECASE)
if m:
val = m.group(1).strip().rstrip('.')
# Filter out pure-number strings that look like Connect IDs (XXXXX-XXXXXX)
if re.match(r'^\d{4,}$', val.replace('-', '').replace('.', '')):
continue # Probably a Connect ID, not a serial
clean = sum(1 for c in val if c.isalnum())
if clean >= 6:
return val
return None
def _save_upload_bytes(contents: bytes, filename: str | None, subdir: str, allowed_exts: set, max_size: int) -> str:
"""Save raw bytes to uploads/{subdir}/ with a UUID filename.
Like _save_upload but accepts pre-read bytes instead of an UploadFile,
so callers can extract EXIF before saving.
"""
ext = Path(filename or "").suffix.lower()
if not ext or ext not in allowed_exts:
allowed = ", ".join(sorted(allowed_exts))
raise HTTPException(400, f"Invalid file type. Allowed: {allowed}")
if len(contents) > max_size:
mb = max_size // (1024 * 1024)
raise HTTPException(413, f"File too large. Maximum size: {mb} MB")
dest_dir = UPLOADS_DIR / subdir
dest_dir.mkdir(parents=True, exist_ok=True)
fname = f"{uuid.uuid4().hex}{ext}"
(dest_dir / fname).write_bytes(contents)
return f"/uploads/{subdir}/{fname}"
def _re_embed_exif(filepath: Path, exif_json: str):
"""Re-embed EXIF data (from client-side exifr.parse) into a saved JPEG.
This is a defense against EXIF being stripped during upload — the client
reads EXIF from the original file before sending, and we write it back.
"""
try:
exif_data = _json.loads(exif_json)
except (_json.JSONDecodeError, TypeError):
return
if filepath.suffix.lower() not in (".jpg", ".jpeg"):
return
try:
exif_dict: dict = {"0th": {}, "Exif": {}, "GPS": {}, "Interop": {}, "1st": {}, "thumbnail": None}
if exif_data.get("Make"):
exif_dict["0th"][piexif.ImageIFD.Make] = str(exif_data["Make"]).encode()
if exif_data.get("Model"):
exif_dict["0th"][piexif.ImageIFD.Model] = str(exif_data["Model"]).encode()
if exif_data.get("Orientation"):
exif_dict["0th"][piexif.ImageIFD.Orientation] = int(exif_data["Orientation"])
dto = exif_data.get("DateTimeOriginal") or exif_data.get("DateTime")
if dto:
exif_dict["Exif"][piexif.ExifIFD.DateTimeOriginal] = str(dto).encode()
gps = {}
if exif_data.get("GPSLatitudeRef"):
gps[piexif.GPSIFD.GPSLatitudeRef] = str(exif_data["GPSLatitudeRef"]).encode()
if exif_data.get("GPSLatitude"):
vals = exif_data["GPSLatitude"]
if isinstance(vals, list) and len(vals) == 3:
gps[piexif.GPSIFD.GPSLatitude] = [
_piexif_rational(vals[0]),
_piexif_rational(vals[1]),
_piexif_rational(vals[2]),
]
if exif_data.get("GPSLongitudeRef"):
gps[piexif.GPSIFD.GPSLongitudeRef] = str(exif_data["GPSLongitudeRef"]).encode()
if exif_data.get("GPSLongitude"):
vals = exif_data["GPSLongitude"]
if isinstance(vals, list) and len(vals) == 3:
gps[piexif.GPSIFD.GPSLongitude] = [
_piexif_rational(vals[0]),
_piexif_rational(vals[1]),
_piexif_rational(vals[2]),
]
if exif_data.get("GPSAltitudeRef") is not None:
gps[piexif.GPSIFD.GPSAltitudeRef] = bytes([int(exif_data["GPSAltitudeRef"])])
if exif_data.get("GPSAltitude") is not None:
gps[piexif.GPSIFD.GPSAltitude] = _piexif_rational(exif_data["GPSAltitude"])
if exif_data.get("GPSTimeStamp"):
vals = exif_data["GPSTimeStamp"]
if isinstance(vals, list) and len(vals) == 3:
gps[piexif.GPSIFD.GPSTimeStamp] = [
_piexif_rational(vals[0]),
_piexif_rational(vals[1]),
_piexif_rational(vals[2]),
]
if exif_data.get("GPSDateStamp"):
gps[piexif.GPSIFD.GPSDateStamp] = str(exif_data["GPSDateStamp"]).encode()
if gps:
exif_dict["GPS"] = gps
exif_bytes = piexif.dump(exif_dict)
piexif.insert(exif_bytes, str(filepath))
except Exception:
pass
def _piexif_rational(val) -> tuple:
"""Convert a float/int to a piexif rational ((numerator, denominator))."""
if isinstance(val, (int, float)):
return (round(val * 10000), 10000)
return (int(val), 1)
def _save_upload(upload: UploadFile, subdir: str, allowed_exts: set, max_size: int) -> str:
"""Save uploaded file to uploads/{subdir}/ with a UUID filename.
Returns the relative URL path, e.g. /uploads/icons/abc123.png.
"""
ext = Path(upload.filename or "").suffix.lower()
if not ext or ext not in allowed_exts:
allowed = ", ".join(sorted(allowed_exts))
raise HTTPException(400, f"Invalid file type. Allowed: {allowed}")
contents = upload.file.read()
if len(contents) > max_size:
mb = max_size // (1024 * 1024)
raise HTTPException(413, f"File too large. Maximum size: {mb} MB")
dest_dir = UPLOADS_DIR / subdir
dest_dir.mkdir(parents=True, exist_ok=True)
fname = f"{uuid.uuid4().hex}{ext}"
(dest_dir / fname).write_bytes(contents)
return f"/uploads/{subdir}/{fname}"
@app.post("/api/upload/photo", status_code=201)
async def upload_photo(file: UploadFile = File(...), exif_data: str = Form(None)):
# Read bytes for EXIF extraction before saving
contents = await file.read()
exif_gps = _extract_gps_from_bytes(contents) if contents else None
# Save using the raw bytes
path = _save_upload_bytes(contents, file.filename, "photos", PHOTO_ALLOWED_EXTS, PHOTO_MAX_SIZE)
# If client sent EXIF data, re-embed it into the saved file (defense against EXIF stripping)
if exif_data:
_re_embed_exif(UPLOADS_DIR / path.split("/", 2)[-1], exif_data)
result = {"path": path}
if exif_gps:
result["exif_gps"] = exif_gps
return result
@app.post("/api/ocr", status_code=200)
async def ocr_sticker(file: UploadFile = File(...), exif_data: str = Form(None)):
"""OCR a sticker photo to extract machine_id from XXXXX-XXXXXX format.
Accepts an image upload, runs Tesseract OCR, and looks for a pattern like
'12345-678901'. Also extracts EXIF GPS data from the raw file bytes before
processing, returning both in one response.
When exif_data is provided (gallery upload), saves the photo permanently and
re-embeds the client-side EXIF data for a proper round-trip.
Returns {machine_id, exif_gps (if found), path (if permanent), raw_text, ...}.
"""
# Read file contents first (validate size before saving)
contents = await file.read()
if len(contents) > 20 * 1024 * 1024: # 20MB max
raise HTTPException(status_code=400, detail="Image too large (max 20MB)")
# Extract EXIF GPS from raw bytes before any processing
exif_gps = _extract_gps_from_bytes(contents) if contents else None
# If client sent EXIF data (gallery upload), also try to extract GPS from JSON
# (fallback for when transport strips EXIF from raw bytes)
if not exif_gps and exif_data:
try:
ed = _json.loads(exif_data)
lat = ed.get("GPSLatitude")
lng = ed.get("GPSLongitude")
lat_ref = ed.get("GPSLatitudeRef", "N")
lng_ref = ed.get("GPSLongitudeRef", "W")
if lat and lng and isinstance(lat, list) and isinstance(lng, list) and len(lat) == 3 and len(lng) == 3:
def _dms_to_dd(parts, ref):
dd = parts[0] + parts[1] / 60.0 + parts[2] / 3600.0
if ref in ("S", "W"):
dd = -dd
return dd
exif_gps = {
"lat": _dms_to_dd(lat, lat_ref),
"lng": _dms_to_dd(lng, lng_ref),
}
except Exception:
pass
# If client sent EXIF data (gallery upload), save permanently with EXIF round-trip
saved_path: str | None = None
if exif_data:
try:
saved_path = _save_upload_bytes(contents, file.filename, "photos", PHOTO_ALLOWED_EXTS, PHOTO_MAX_SIZE)
_re_embed_exif(UPLOADS_DIR / saved_path.split("/", 2)[-1], exif_data)
except Exception:
saved_path = None # fall through to temp-file path below
# Save file for OCR — use permanent path if available, otherwise temp
if saved_path:
ocr_path = UPLOADS_DIR / saved_path.split("/", 2)[-1]
else:
ext = file.filename.rsplit(".", 1)[-1].lower() if "." in (file.filename or "") else "jpg"
if ext not in {"png", "jpg", "jpeg", "webp", "bmp", "dng", "tiff", "tif"}:
ext = "jpg"
temp_dir = Path(UPLOADS_DIR / "ocr")
temp_dir.mkdir(parents=True, exist_ok=True)
ocr_path = temp_dir / f"{uuid.uuid4().hex}.{ext or 'jpg'}"
ocr_path.write_bytes(contents)
text = ""
ocr_source = "none"
ollama_text = ""
# Try Ollama vision model first (Windows PC) — most accurate
if _HAS_OLLAMA:
try:
import urllib.request as _ourl
import base64 as _b64
img_vision = PILImage.open(ocr_path)
img_vision.thumbnail((640, 480))
temp_vision = ocr_path.parent / f"vis_{ocr_path.name}"
img_vision.save(temp_vision)
with open(temp_vision, "rb") as _f:
_b64_data = _b64.b64encode(_f.read()).decode()
temp_vision.unlink(missing_ok=True)
_vdata = _json.dumps({
"model": _OLLAMA_VISION_MODEL,
"prompt": "Extract all text, numbers, serial numbers, and IDs visible on this sticker or label. Return ONLY the raw text content, one item per line. Do not describe the image.",
"images": [_b64_data],
"stream": False,
}).encode()
_vreq = _ourl.Request(
f"{_OLLAMA_BASE}/api/generate",
data=_vdata,
headers={"Content-Type": "application/json"},
)
_vresp = _ourl.urlopen(_vreq, timeout=120)
_vresult = _json.loads(_vresp.read().decode())
ollama_text = _vresult.get("response", "").strip()
if ollama_text:
ocr_source = "ollama"
except Exception:
pass # Fall through to Tesseract
# Try Tesseract if Ollama didn't yield usable text
if ocr_source == "none":
try:
img = PILImage.open(ocr_path)
img_gray = img.convert("L")
if _HAS_TESSERACT:
tess_text = pytesseract.image_to_string(img_gray, config="--psm 6")
tess_text = tess_text.strip()
clean_chars = sum(1 for c in tess_text if c.isalnum() or c in ' \n/-.')
if len(tess_text) > 0 and clean_chars >= 10:
ocr_source = "tesseract"
except Exception:
pass
# Prefer Ollama text when available (much more accurate)
if ollama_text:
text = ollama_text
elif ocr_source == "tesseract":
text = tess_text
else:
text = ""
# Clean up temp file
if not saved_path:
ocr_path.unlink(missing_ok=True)
if not text:
if _HAS_OLLAMA:
detail = "Could not read any text from the image via Ollama vision model."
elif _HAS_TESSERACT:
detail = "Could not read any text from the image via Tesseract. Try a clearer photo of the label."
else:
detail = "No OCR service available (Tesseract not installed, Ollama not connected)."
raise HTTPException(status_code=422, detail=detail)
# Build response — search for identifiers in the OCR text
result: dict = {
"raw_text": text.strip()[:1000],
"ocr_source": ocr_source,
}
# 1. Legacy pattern: XXXXX-XXXXXX (5 digits - 6+ digits = Connect ID)
match = re.search(r"(\d{5})[-\s]*(\d{6,})", text)
if match:
full_match = match.group(0)
digits_only = re.sub(r"\D", "", full_match)
machine_id = digits_only[-5:]
result["machine_id"] = machine_id
result["raw_match"] = full_match
result["confidence"] = "high"
else:
# Try looser: any 5+ digit number, take the last 5 digits
loose = re.search(r"(\d{5,})", text)
if loose:
digits = loose.group(1)
machine_id = digits[-5:] if len(digits) > 5 else digits
result["machine_id"] = machine_id
result["confidence"] = "low"
else:
result["machine_id"] = None
result["confidence"] = "none"
result["detail"] = "No machine ID pattern found in image. Try again with better lighting."
# 2. Cross-reference OCR text against DB — find matched assets by
# serial_number, connect_id, equipment_id, or barcode
db_path = DB_PATH
db_matches = []
seen_ids = set()
for line in text.strip().split('\n'):
line = line.strip()
if not line or len(line) < 4:
continue
# Try full line
norm = normalize_identifier(line)
if norm and len(norm) >= 4:
assets = find_asset_by_normalized_id(db_path, norm)
for a in assets:
if a['id'] not in seen_ids:
seen_ids.add(a['id'])
db_matches.append({
"asset_id": a['id'],
"machine_id": a['machine_id'],
"name": a['name'],
"serial_number": a['serial_number'],
"matched_on": norm,
"source_text": line,
})
# Try individual tokens on the line
tokens = re.findall(r'[A-Za-z0-9]{4,}', line)
for token in tokens:
norm = normalize_identifier(token)
if norm and len(norm) >= 4:
assets = find_asset_by_normalized_id(db_path, norm)
for a in assets:
if a['id'] not in seen_ids:
seen_ids.add(a['id'])
db_matches.append({
"asset_id": a['id'],
"machine_id": a['machine_id'],
"name": a['name'],
"serial_number": a['serial_number'],
"matched_on": norm,
"source_text": token,
})
if db_matches:
result["matched_assets"] = db_matches
# Auto-save photo_path to matched assets when photo was saved permanently
if saved_path and db_matches:
try:
conn = get_db()
updated_count = 0
for m in db_matches:
aid = m["asset_id"]
existing = conn.execute(
"SELECT photo_path FROM assets WHERE id = ?",
(aid,),
).fetchone()
if existing and not existing["photo_path"]:
conn.execute(
"UPDATE assets SET photo_path = ?, updated_at = datetime('now') WHERE id = ?",
(saved_path, aid),
)
updated_count += 1
conn.commit()
conn.close()
if updated_count > 0:
result["photo_saved"] = updated_count
except Exception:
pass # Non-critical — don't fail OCR if photo save fails
# Auto-update serial_number on matched assets from OCR text
if db_matches and any(m.get("serial_number") == "" for m in db_matches):
serial = _extract_serial_from_text(text)
if serial:
try:
conn = get_db()
updated_count = 0
for m in db_matches:
if m.get("serial_number") == "":
conn.execute(
"UPDATE assets SET serial_number = ?, updated_at = datetime('now') WHERE id = ?",
(serial, m["asset_id"]),
)
updated_count += 1
conn.commit()
conn.close()
if updated_count > 0:
result["serial_saved"] = serial
except Exception:
pass # Non-critical
if exif_gps:
result["exif_gps"] = exif_gps
# Auto-save GPS to asset if lat/lng is blank
machine_id = result.get("machine_id")
if machine_id:
try:
conn = get_db()
asset = conn.execute(
"SELECT id, latitude, longitude FROM assets WHERE machine_id = ?",
(machine_id,),
).fetchone()
if asset and (asset["latitude"] is None or asset["longitude"] is None):
conn.execute(
"UPDATE assets SET latitude = ?, longitude = ?, updated_at = datetime('now') WHERE id = ?",
(exif_gps["lat"], exif_gps["lng"], asset["id"]),
)
conn.commit()
result["gps_saved"] = True
conn.close()
except Exception:
pass # Non-critical — don't fail OCR if GPS save fails
if saved_path:
result["path"] = saved_path
return result
# ─── Match raw text against DB (for barcode scanner / client-side OCR) ────
@app.post("/api/match-text", status_code=200)
async def match_text(text: str = Form(...)):
"""
Accept raw text (from barcode scanner, QR reader, or client-side vision),
normalize it, and search the DB for matching assets.
Returns matched_assets if any found.
"""
if not text or len(text.strip()) < 4:
return {"matched_assets": [], "detail": "Text too short to match"}
db_path = DB_PATH
db_matches = []
seen_ids = set()
raw = text.strip()
for line in raw.split('\n'):
line = line.strip()
if not line or len(line) < 4:
continue
norm = normalize_identifier(line)
if norm and len(norm) >= 4:
assets = find_asset_by_normalized_id(db_path, norm)
for a in assets:
if a['id'] not in seen_ids:
seen_ids.add(a['id'])
db_matches.append({
"asset_id": a['id'],
"machine_id": a['machine_id'],
"name": a['name'],
"serial_number": a['serial_number'],
"connect_id": a['connect_id'],
"make": a['make'],
"model": a['model'],
"category": a['category'],
"matched_on": norm,
"source_text": line,
})
# Also check individual tokens
tokens = re.findall(r'[A-Za-z0-9]{4,}', line)
for token in tokens:
norm = normalize_identifier(token)
if norm and len(norm) >= 4:
assets = find_asset_by_normalized_id(db_path, norm)
for a in assets:
if a['id'] not in seen_ids:
seen_ids.add(a['id'])
db_matches.append({
"asset_id": a['id'],
"machine_id": a['machine_id'],
"name": a['name'],
"serial_number": a['serial_number'],
"connect_id": a['connect_id'],
"make": a['make'],
"model": a['model'],
"category": a['category'],
"matched_on": norm,
"source_text": token,
})
return {
"raw_text": raw[:1000],
"matched_assets": db_matches,
"match_count": len(db_matches),
}
# ─── T4: Connect Label — unified photo + OCR + GPS endpoint ─────────────────
class ConnectLabelRequest(BaseModel):
"""Request body for the connect-label endpoint."""
machine_id: str
name: str = ""
latitude: Optional[float] = None
longitude: Optional[float] = None
category: Optional[str] = "Other"
photo_path: Optional[str] = None
@app.post("/api/connect-label", status_code=201)
async def connect_label(
photo: Optional[UploadFile] = File(None),
machine_id: str = Form(""),
name: str = Form(""),
latitude: Optional[float] = Form(None),
longitude: Optional[float] = Form(None),
category: str = Form("Other"),
photo_path: str = Form(""),
):
"""Unified endpoint for Connect Label workflow.
Accepts multipart form data with optional photo upload.
Creates an asset with the provided machine_id, name, GPS coords.
If photo is uploaded, it saves the file and sets photo_path.
Validates machine_id format (XXXXX-XXXXXX → last 5 digits).
"""
# If machine_id/name come as form fields, use those; otherwise try query params
machine_id = _sanitize_machine_id(machine_id)
name = _sanitize_name(name or "Scanned Asset")
_validate_status("active")
# Save photo if provided
saved_photo_path = photo_path or ""
if photo and photo.filename:
saved_photo_path = _save_upload(photo, "connect_labels", PHOTO_ALLOWED_EXTS, PHOTO_MAX_SIZE)
conn = get_db()
_validate_enum_table(conn, "categories", category or "Other", "category")
# Build insert
columns: list[str] = ["machine_id", "name", "category", "status"]
values: list = [machine_id, name, category or "Other", "active"]
if saved_photo_path:
columns.append("photo_path")
values.append(saved_photo_path)
if latitude is not None:
columns.append("latitude")
values.append(latitude)
if longitude is not None:
columns.append("longitude")
values.append(longitude)
# Auto-populate address from GPS
if latitude is not None and longitude is not None:
geo = reverse_geocode(latitude, longitude)
if geo:
if "address" not in columns:
columns.append("address")
values.append(geo.get("address", geo.get("formatted", "")))
if "building_name" not in columns and geo.get("building_name"):
columns.append("building_name")
values.append(geo.get("building_name", ""))
if "building_number" not in columns and geo.get("building_number"):
columns.append("building_number")
values.append(geo.get("building_number", ""))
placeholders = ", ".join(["?"] * len(columns))
col_names = ", ".join(columns)
try:
cursor = conn.execute(
f"INSERT INTO assets ({col_names}) VALUES ({placeholders})",
values,
)
except sqlite3.IntegrityError:
conn.close()
raise HTTPException(
status_code=409,
detail=f"Asset with machine_id '{machine_id}' already exists",
)
raw_id = cursor.lastrowid
assert raw_id is not None, "INSERT did not return lastrowid"
asset_id: int = raw_id
_log_activity(conn, "created", "asset", asset_id,
f"Asset '{name}' (machine_id: {machine_id}) created via Connect Label")
conn.commit()
row = conn.execute("SELECT * FROM assets WHERE id = ?", (asset_id,)).fetchone()
result = row_to_dict(row)
result["keys"] = _get_asset_keys(conn, asset_id)
result["badges"] = _get_asset_badges(conn, asset_id)
conn.close()
return result
# ─── Reverse Geocode (Nominatim) ────────────────────────────────────────────
def reverse_geocode(lat: float, lng: float) -> dict | None:
"""Call OpenStreetMap Nominatim to reverse geocode GPS coords.
Returns a dict with address fields mapped to our schema, or None on failure.
Does NOT raise HTTPException — callers handle the None case.
"""
try:
url = (
f"https://nominatim.openstreetmap.org/reverse"
f"?format=json&lat={lat}&lon={lng}&zoom=18&addressdetails=1"
)
req = urllib.request.Request(url, headers={"User-Agent": "CanteenAssetTracker/2.0"})
with urllib.request.urlopen(req, timeout=5) as resp:
data = _json.loads(resp.read())
except (urllib.error.URLError, _json.JSONDecodeError, OSError):
return None
addr = data.get("address", {})
display_name = data.get("display_name", "")
# Map Nominatim address parts to our schema
building = addr.get("building", "") or addr.get("house_number", "")
building_name = addr.get("building_name", "") or addr.get("amenity", "")
street = addr.get("road", "") or addr.get("pedestrian", "") or addr.get("path", "")
house_number = addr.get("house_number", "")
city = addr.get("city", "") or addr.get("town", "") or addr.get("village", "")
state = addr.get("state", "")
postcode = addr.get("postcode", "")
# Build a formatted address string
parts = []
if house_number and street:
parts.append(f"{house_number} {street}")
elif street:
parts.append(street)
if city:
parts.append(city)
if state:
parts.append(state)
if postcode:
parts.append(postcode)
formatted = ", ".join(parts) if parts else display_name
return {
"display_name": display_name,
"formatted": formatted,
"building_number": house_number or building,
"building_name": building_name,
"street": street,
"city": city,
"state": state,
"postcode": postcode,
"address": formatted,
}
@app.get("/api/geocode")
def geocode(lat: float = Query(...), lng: float = Query(...)):
"""Reverse geocode GPS coordinates to address using OpenStreetMap Nominatim."""
result = reverse_geocode(lat, lng)
if result is None:
raise HTTPException(status_code=502, detail="Geocoding failed: unable to reach Nominatim")
return result
# ─── Disney Park Classification ──────────────────────────────────────────────
@app.get("/api/disney/stats")
def disney_stats():
"""Get Disney classification statistics."""
conn = get_db()
rows = conn.execute(
"SELECT disney_park, COUNT(*) as count FROM assets WHERE disney_park IS NOT NULL GROUP BY disney_park ORDER BY count DESC"
).fetchall()
total = conn.execute("SELECT COUNT(*) as c FROM assets").fetchone()['c']
classified = conn.execute("SELECT COUNT(*) as c FROM assets WHERE disney_park IS NOT NULL").fetchone()['c']
conn.close()
from disney_classify import DISNEY_PARK_NAMES, DISNEY_PARK_ICONS, DISNEY_PARK_COLORS
return {
"total_assets": total,
"classified": classified,
"by_park": {
row['disney_park']: {
"count": row['count'],
"name": DISNEY_PARK_NAMES.get(row['disney_park'], row['disney_park']),
"icon": DISNEY_PARK_ICONS.get(row['disney_park'], "📍"),
"color": DISNEY_PARK_COLORS.get(row['disney_park'], "#95a5a6"),
}
for row in rows
},
}
@app.post("/api/disney/classify")
def classify_disney_assets():
"""Run auto-classification on all assets. Returns report."""
from disney_classify import classify_all_assets
result = classify_all_assets(DB_PATH)
return result
@app.put("/api/assets/{asset_id}/disney-park")
def set_asset_disney_park(asset_id: int, body: dict):
"""Manually set the disney_park classification for an asset."""
park = body.get("disney_park")
valid = {"magic-kingdom", "epcot", "hollywood-studios", "animal-kingdom",
"disney-springs", "resort", "office", "other", None}
if park not in valid and park is not None:
raise HTTPException(400, f"Invalid park. Valid: {', '.join(v for v in valid if v)}")
conn = get_db()
conn.execute(
"UPDATE assets SET disney_park = ?, updated_at = datetime('now') WHERE id = ?",
(park, asset_id)
)
conn.commit()
conn.close()
return {"status": "ok", "asset_id": asset_id, "disney_park": park}
# ─── Service Entrances ────────────────────────────────────────────────────────
class ServiceEntranceCreate(BaseModel):
location_id: int
name: str = ""
latitude: float
longitude: float
notes: str = ""
class ServiceEntranceUpdate(BaseModel):
name: Optional[str] = None
latitude: Optional[float] = None
longitude: Optional[float] = None
notes: Optional[str] = None
@app.get("/api/locations/{location_id}/service-entrances")
def list_service_entrances(location_id: int):
"""List service entrances for a location."""
conn = get_db()
rows = conn.execute(
"SELECT se.*, l.name as location_name FROM service_entrances se "
"JOIN locations l ON se.location_id = l.id "
"WHERE se.location_id = ? ORDER BY se.name",
(location_id,)
).fetchall()
conn.close()
return [dict(r) for r in rows]
@app.post("/api/service-entrances", status_code=201)
def create_service_entrance(body: ServiceEntranceCreate):
"""Create a new service entrance."""
conn = get_db()
# Verify location exists
loc = conn.execute("SELECT id FROM locations WHERE id = ?", (body.location_id,)).fetchone()
if not loc:
conn.close()
raise HTTPException(404, "Location not found")
cursor = conn.execute(
"INSERT INTO service_entrances (location_id, name, latitude, longitude, notes) VALUES (?, ?, ?, ?, ?)",
(body.location_id, body.name, body.latitude, body.longitude, body.notes)
)
conn.commit()
row = conn.execute("SELECT * FROM service_entrances WHERE id = ?", (cursor.lastrowid,)).fetchone()
conn.close()
return dict(row)
@app.put("/api/service-entrances/{entrance_id}")
def update_service_entrance(entrance_id: int, body: ServiceEntranceUpdate):
"""Update a service entrance."""
conn = get_db()
existing = conn.execute("SELECT * FROM service_entrances WHERE id = ?", (entrance_id,)).fetchone()
if not existing:
conn.close()
raise HTTPException(404, "Service entrance not found")
updates = {}
if body.name is not None:
updates['name'] = body.name
if body.latitude is not None:
updates['latitude'] = body.latitude
if body.longitude is not None:
updates['longitude'] = body.longitude
if body.notes is not None:
updates['notes'] = body.notes
if updates:
updates['updated_at'] = "datetime('now')"
set_clause = ", ".join(f"{k} = ?" for k in updates)
values = list(updates.values())
# Handle updated_at specially
set_clause = ", ".join(
f"{k} = datetime('now')" if k == 'updated_at' else f"{k} = ?"
for k in updates
)
values = [v for k, v in updates.items() if k != 'updated_at']
conn.execute(
f"UPDATE service_entrances SET {set_clause} WHERE id = ?",
(*values, entrance_id)
)
conn.commit()
row = conn.execute("SELECT se.*, l.name as location_name FROM service_entrances se "
"JOIN locations l ON se.location_id = l.id "
"WHERE se.id = ?", (entrance_id,)).fetchone()
conn.close()
return dict(row)
@app.delete("/api/service-entrances/{entrance_id}", status_code=204)
def delete_service_entrance(entrance_id: int):
"""Delete a service entrance."""
conn = get_db()
cursor = conn.execute("DELETE FROM service_entrances WHERE id = ?", (entrance_id,))
conn.commit()
conn.close()
if cursor.rowcount == 0:
raise HTTPException(404, "Service entrance not found")
return None
@app.get("/api/service-entrances")
def list_all_service_entrances():
"""List all service entrances."""
conn = get_db()
rows = conn.execute(
"SELECT se.*, l.name as location_name, l.address as location_address "
"FROM service_entrances se "
"LEFT JOIN locations l ON se.location_id = l.id "
"ORDER BY l.name, se.name"
).fetchall()
conn.close()
return [dict(r) for r in rows]
# ═══════════════════════════════════════════════════════════════════════════
# WORK ORDER & ROUTE OPTIMIZATION — reads extraction DB directly
# ═══════════════════════════════════════════════════════════════════════════
# ── Extraction DB (MS Field Service sync) ────────────────────────────────
EXTRACTION_PROJECT = Path.home() / "projects" / "ms-field-service-extraction"
EXTRACTION_DB = (
EXTRACTION_PROJECT
/ "data-samples"
/ "msfs_backup"
/ "fieldservicecanteen.crm.dynamics.com_9fc6c50c-b097-f011-b4cb-7ced8d1b15a2_data.db"
)
STATUS_LABELS = {
"690970000": "Unscheduled",
"690970001": "Scheduled",
"690970002": "In Progress",
"690970003": "Completed",
"690970004": "Posted",
"690970005": "Cancelled",
"690959000": "Postponed",
}
def _status_label(code) -> str:
"""Map a status code integer to a human-readable label."""
return STATUS_LABELS.get(str(code), f"Unknown ({code})")
def _get_extraction_db() -> sqlite3.Connection | None:
"""Return read-only connection to the MS Field Service extraction DB."""
if not EXTRACTION_DB.exists():
return None
conn = sqlite3.connect(str(EXTRACTION_DB))
conn.row_factory = sqlite3.Row
return conn
# ── Asset Branch Cache (connect_id prefix → branch/territory name) ────────
_BRANCH_CACHE: dict[str, str] | None = None
def _build_asset_branch_cache() -> dict[str, str]:
"""Build a mapping from connect_id prefix (first 8 chars) to branch name.
Queries the extraction DB for each customer asset, linking it through
its account to work order territories (msdyn_serviceterritory!name).
Returns {connect_id_prefix: branch_name}.
"""
global _BRANCH_CACHE
if _BRANCH_CACHE is not None:
return _BRANCH_CACHE
result: dict[str, str] = {}
conn = _get_extraction_db()
if not conn:
print("⚠️ Extraction DB not available — branch cache will be empty")
_BRANCH_CACHE = result
return result
try:
cur = conn.cursor()
# For each customer asset (connect_id + account), find the most common
# work order territory for that account.
cur.execute("""
SELECT
SUBSTR(ca.hsl_connectid, 1, 8) AS prefix,
wo."msdyn_serviceterritory!name" AS territory,
COUNT(*) AS cnt
FROM msdyn_customerasset ca
LEFT JOIN account a ON ca."msdyn_account!id" = a.accountid
LEFT JOIN msdyn_workorder wo ON a.accountid = wo."msdyn_serviceaccount!id"
WHERE ca.hsl_connectid IS NOT NULL AND ca.hsl_connectid != ''
AND wo."msdyn_serviceterritory!name" IS NOT NULL
AND wo."msdyn_serviceterritory!name" != ''
GROUP BY prefix, wo."msdyn_serviceterritory!name"
ORDER BY cnt DESC
""")
# Take the most frequent territory per prefix
prefix_map: dict[str, list[tuple[str, int]]] = {}
for r in cur.fetchall():
prefix = r[0]
territory = r[1]
cnt = r[2]
if prefix not in prefix_map:
prefix_map[prefix] = []
prefix_map[prefix].append((territory, cnt))
for prefix, territories in prefix_map.items():
# Sort by count descending, pick the top one
territories.sort(key=lambda x: -x[1])
result[prefix] = territories[0][0]
print(f"📡 Built branch cache: {len(result)} connect_id prefixes mapped")
except Exception as e:
print(f"⚠️ Failed to build branch cache: {e}")
finally:
conn.close()
_BRANCH_CACHE = result
return result
# ── TSP Solver ───────────────────────────────────────────────────────────
def _haversine(lat1, lon1, lat2, lon2):
"""Haversine distance in km between two lat/lng points."""
R = 6371
dlat = math.radians(lat2 - lat1)
dlon = math.radians(lon2 - lon1)
a = (
math.sin(dlat / 2) ** 2
+ math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlon / 2) ** 2
)
return R * 2 * math.asin(math.sqrt(a))
def _solve_tsp(points, origin_idx=0):
"""Nearest-neighbor TSP with 2-opt improvement.
Args:
points: list of (lat, lng) tuples.
origin_idx: index of the starting point (default 0).
Returns list of indices in TSP-optimized order.
"""
n = len(points)
if n <= 2:
return list(range(n))
# Precompute distance matrix
dist = [
[_haversine(points[i][0], points[i][1], points[j][0], points[j][1]) for j in range(n)]
for i in range(n)
]
# Nearest-neighbor
unvisited = set(range(n))
unvisited.remove(origin_idx)
route = [origin_idx]
current = origin_idx
while unvisited:
nearest = min(unvisited, key=lambda j: dist[current][j])
route.append(nearest)
unvisited.remove(nearest)
current = nearest
# 2-opt improvement
improved = True
while improved:
improved = False
for i in range(n - 1):
for j in range(i + 2, n):
j1 = (j + 1) % n
old_dist = dist[route[i]][route[i + 1]] + dist[route[j]][route[j1]]
new_dist = dist[route[i]][route[j]] + dist[route[i + 1]][route[j1]]
if new_dist < old_dist - 0.001:
route[i + 1 : j + 1] = reversed(route[i + 1 : j + 1])
improved = True
return route
# ── Work Order Search ──────────────────────────────────────────────────────
@app.get("/api/workorders/search")
async def workorders_search(
q: str = Query("", description="Search term"),
limit: int = Query(50, ge=1, le=200),
offset: int = Query(0, ge=0),
):
"""Search work orders by name, account, or city."""
conn = _get_extraction_db()
if not conn:
raise HTTPException(503, "Database not available — sync extraction DB missing")
try:
cur = conn.cursor()
like = f"%{q}%"
cur.execute(
"""
SELECT
w.msdyn_workorderid, w.msdyn_name,
w."msdyn_serviceaccount!name" AS account_name,
w.msdyn_workordersummary,
w.msdyn_datewindowstart, w.msdyn_timewindowstart, w.msdyn_timewindowend,
w.hsl_onsiteduration,
w."msdyn_priority!name" AS priority,
w.msdyn_systemstatus,
a.address1_line1, a.address1_city, a.address1_stateorprovince, a.address1_postalcode,
a.address1_latitude, a.address1_longitude
FROM msdyn_workorder w
LEFT JOIN account a ON w."msdyn_serviceaccount!id" = a.accountid
WHERE w.msdyn_name LIKE ?
OR a.name LIKE ?
OR a.address1_city LIKE ?
ORDER BY w.msdyn_name
LIMIT ? OFFSET ?
""",
(like, like, like, limit, offset),
)
rows = cur.fetchall()
cur.execute(
"""
SELECT COUNT(*) FROM msdyn_workorder w
LEFT JOIN account a ON w."msdyn_serviceaccount!id" = a.accountid
WHERE w.msdyn_name LIKE ?
OR a.name LIKE ?
OR a.address1_city LIKE ?
""",
(like, like, like),
)
total = cur.fetchone()[0]
return {
"results": [
{
"id": r["msdyn_workorderid"],
"name": r["msdyn_name"],
"account_name": r["account_name"],
"summary": r["msdyn_workordersummary"],
"status_code": r["msdyn_systemstatus"],
"status": _status_label(r["msdyn_systemstatus"]),
"priority": r["priority"],
"onsite_duration": r["hsl_onsiteduration"],
"address": {
"line1": r["address1_line1"],
"city": r["address1_city"],
"state": r["address1_stateorprovince"],
"postal": r["address1_postalcode"],
},
"gps": (
{
"lat": r["address1_latitude"],
"lng": r["address1_longitude"],
}
if r["address1_latitude"] and r["address1_longitude"]
else None
),
}
for r in rows
],
"total": total,
"offset": offset,
"limit": limit,
}
finally:
conn.close()
# ── Work Order Lookup (by ID) ──────────────────────────────────────────────
class WorkorderLookupRequest(BaseModel):
ids: list[str]
@app.post("/api/workorders/lookup")
async def workorders_lookup(body: WorkorderLookupRequest):
"""Lookup work orders by ID or name."""
ids = body.ids
if not ids:
raise HTTPException(400, "No work order IDs provided")
if len(ids) > 100:
raise HTTPException(400, "Max 100 work orders per request")
conn = _get_extraction_db()
if not conn:
raise HTTPException(503, "Database not available — sync extraction DB missing")
try:
cur = conn.cursor()
placeholders = ",".join(["?"] * len(ids))
cur.execute(
f"""
SELECT
w.msdyn_workorderid, w.msdyn_name,
w."msdyn_serviceaccount!name" AS account_name,
w."msdyn_serviceaccount!id" AS account_id,
w.msdyn_workordersummary,
w.msdyn_datewindowstart, w.msdyn_timewindowstart, w.msdyn_timewindowend,
w.hsl_onsiteduration,
w."msdyn_priority!name" AS priority,
w.msdyn_systemstatus,
w."msdyn_workordertype!name" AS work_type,
w."msdyn_primaryincidenttype!name" AS incident_type,
w."msdyn_serviceterritory!name" AS territory,
w.msdyn_instructions,
w."msdyn_customerasset!name" AS customer_asset_name,
w.msdyn_completedon,
w.msdyn_datewindowend,
w.msdyn_city, w.msdyn_stateorprovince,
-- Timestamps
w.createdon,
w.modifiedon,
w.msdyn_firstarrivedon,
w.msdyn_timeclosed,
w.hsl_pausedon, w.hsl_pausedend,
w.hsl_statuschangedate,
w.hsl_completedawaitingparts, w.hsl_completedcustomerhold,
w.hsl_totalpauseduration, w.hsl_totalduration,
w.hsl_travelduration,
w.msdyn_totalestimatedduration,
w.msdyn_primaryincidentestimatedduration,
-- More info fields
w.hsl_additionalnotes,
w.hsl_signature, w.hsl_signedby,
w.hsl_equipmentserialnumbermachineidin,
w.hsl_equipmentserialnumbermachineidout,
w.msdyn_address1,
w.msdyn_postalcode, w.msdyn_country,
w."msdyn_substatus!name" AS substatus,
w."msdyn_closedby!name" AS closed_by,
w."msdyn_primaryresolution!name" AS primary_resolution,
w."msdyn_trade!name" AS trade,
w."msdyn_functionallocation!name" AS functional_location,
w."hsl_lineofbusiness!name" AS line_of_business,
w.msdyn_bookingsummary,
(SELECT GROUP_CONCAT(DISTINCT t)
FROM (
SELECT b2."resource!name" AS t FROM bookableresourcebooking b2 WHERE b2."msdyn_workorder!name" = w.msdyn_name
UNION
SELECT w2."hsl_bookedresource1!name" FROM msdyn_workorder w2 WHERE w2.msdyn_workorderid = w.msdyn_workorderid AND w2."hsl_bookedresource1!name" IS NOT NULL
UNION
SELECT w2."hsl_bookedresource2!name" FROM msdyn_workorder w2 WHERE w2.msdyn_workorderid = w.msdyn_workorderid AND w2."hsl_bookedresource2!name" IS NOT NULL
UNION
SELECT w2."hsl_bookedresource3!name" FROM msdyn_workorder w2 WHERE w2.msdyn_workorderid = w.msdyn_workorderid AND w2."hsl_bookedresource3!name" IS NOT NULL
)) AS technicians,
a.address1_line1, a.address1_city, a.address1_stateorprovince, a.address1_postalcode,
a.address1_latitude, a.address1_longitude, a.name AS account_display_name
FROM msdyn_workorder w
LEFT JOIN account a ON w."msdyn_serviceaccount!id" = a.accountid
WHERE w.msdyn_workorderid IN ({placeholders})
OR w.msdyn_name IN ({placeholders})
""",
ids + ids,
)
rows = cur.fetchall()
found = []
id_set = set(ids)
found_set = set()
for r in rows:
rid = r["msdyn_workorderid"]
rname = r["msdyn_name"]
matched = rid if rid in id_set else rname if rname in id_set else None
if matched:
found_set.add(matched)
lat = r["address1_latitude"]
lng = r["address1_longitude"]
gps = None
if lat and lng:
try:
gps = {"lat": float(lat), "lng": float(lng)}
except (ValueError, TypeError):
pass
found.append(
{
"id": rid,
"name": rname,
"account_name": r["account_name"],
"account_display_name": r["account_display_name"],
"summary": r["msdyn_workordersummary"],
"status_code": r["msdyn_systemstatus"],
"status": _status_label(r["msdyn_systemstatus"]),
"priority": r["priority"],
"work_type": r["work_type"],
"incident_type": r["incident_type"],
"territory": r["territory"],
"instructions": r["msdyn_instructions"],
"customer_asset_name": r["customer_asset_name"],
"completed_on": r["msdyn_completedon"],
"datewindow_start": r["msdyn_datewindowstart"],
"datewindow_end": r["msdyn_datewindowend"],
"timewindow_start": r["msdyn_timewindowstart"],
"timewindow_end": r["msdyn_timewindowend"],
"city": r["msdyn_city"] or r["address1_city"],
"state": r["msdyn_stateorprovince"] or r["address1_stateorprovince"],
"onsite_duration": r["hsl_onsiteduration"],
"created_on": r["createdon"],
"modified_on": r["modifiedon"],
"first_arrived_on": r["msdyn_firstarrivedon"],
"time_closed": r["msdyn_timeclosed"],
"paused_on": r["hsl_pausedon"],
"paused_end": r["hsl_pausedend"],
"status_changed_on": r["hsl_statuschangedate"],
"awaiting_parts": bool(r["hsl_completedawaitingparts"]),
"customer_hold": bool(r["hsl_completedcustomerhold"]),
"total_pause_duration": r["hsl_totalpauseduration"],
"total_duration": r["hsl_totalduration"],
"travel_duration": r["hsl_travelduration"],
"estimated_duration": r["msdyn_totalestimatedduration"],
"incident_estimated_duration": r["msdyn_primaryincidentestimatedduration"],
"substatus": r["substatus"],
"closed_by": r["closed_by"],
"primary_resolution": r["primary_resolution"],
"trade": r["trade"],
"functional_location": r["functional_location"],
"line_of_business": r["line_of_business"],
"booking_summary": r["msdyn_bookingsummary"],
"additional_notes": r["hsl_additionalnotes"],
"signature": r["hsl_signature"],
"signed_by": r["hsl_signedby"],
"equipment_serial_in": r["hsl_equipmentserialnumbermachineidin"],
"equipment_serial_out": r["hsl_equipmentserialnumbermachineidout"],
"address_line1": r["msdyn_address1"],
"postal_code": r["msdyn_postalcode"],
"country": r["msdyn_country"],
"address": {
"line1": r["address1_line1"],
"city": r["address1_city"],
"state": r["address1_stateorprovince"],
"postal": r["address1_postalcode"],
},
"gps": gps,
"technicians": r["technicians"],
}
)
not_found = [uid for uid in ids if uid not in found_set and uid not in {f["id"] for f in found}]
return {
"found": found,
"not_found": not_found,
"total_found": len(found),
"total_not_found": len(not_found),
}
finally:
conn.close()
# ── Today's Active Work Orders ──────────────────────────────────────────────
@app.get("/api/workorders/today")
async def workorders_today(
tech: list[str] = Query([], description="Filter by technician names"),
limit: int = Query(100, ge=1, le=500),
status: str = Query("", description="Filter by work order status"),
priority: str = Query("", description="Filter by priority (Urgent/High/Medium/Low)"),
work_type: str = Query("", description="Filter by work type (Install/Repair/PM/Emergency)"),
date_range: str = Query("today", description="Date range: today, week, month, all"),
):
"""Fetch today's active work orders via bookableresourcebooking.
Pass ?tech=Shawn+Canada&tech=John+Doe to filter by one or more technicians.
Omit tech entirely to return all technicians.
Supports optional ?status=, ?priority=, ?work_type=, ?date_range= filters.
"""
conn = _get_extraction_db()
if not conn:
raise HTTPException(503, "Database not available — sync extraction DB missing")
try:
cur = conn.cursor()
today = date.today().isoformat()
# Determine date window
date_window_start = f"{today}T00:00:00Z"
date_window_end = f"{today}T23:59:59Z"
if date_range == "week":
from datetime import timedelta
week_ago = (date.today() - timedelta(days=7)).isoformat()
date_window_start = f"{week_ago}T00:00:00Z"
elif date_range == "month":
from datetime import timedelta
month_ago = (date.today() - timedelta(days=30)).isoformat()
date_window_start = f"{month_ago}T00:00:00Z"
elif date_range == "all":
date_window_start = "2000-01-01T00:00:00Z"
date_window_end = "2099-12-31T23:59:59Z"
active_booking_statuses = ",".join(
f"'{s}'"
for s in ["Scheduled", "Unscheduled", "On Site", "Traveling", "On Break", "Hold", "Committed"]
)
params = [date_window_start, date_window_end]
tech_clause = ""
if tech:
placeholders_t = ",".join(["?" for _ in tech])
tech_clause = f'AND b."resource!name" IN ({placeholders_t})'
params.extend(tech)
# Additional filters
filter_clauses = []
if status:
filter_clauses.append("w.msdyn_systemstatus = ?")
params.append(status)
if priority:
filter_clauses.append('w."msdyn_priority!name" = ?')
params.append(priority)
if work_type:
filter_clauses.append('w."msdyn_workordertype!name" = ?')
params.append(work_type)
filter_sql = " ".join(filter_clauses)
params.append(limit)
cur.execute(
f"""
SELECT
w.msdyn_workorderid,
COALESCE(b."msdyn_workorder!name", w.msdyn_name) AS msdyn_name,
COALESCE(b."hsl_serviceaccountid!name", w."msdyn_serviceaccount!name") AS account_name,
COALESCE(b."hsl_serviceaccountid!id", w."msdyn_serviceaccount!id") AS account_id,
b."bookingstatus!name" AS booking_status,
w.msdyn_systemstatus,
w.msdyn_workordersummary,
b.starttime, b.endtime,
w.hsl_onsiteduration,
w."msdyn_priority!name" AS priority,
w."msdyn_workordertype!name" AS work_type,
w."msdyn_primaryincidenttype!name" AS incident_type,
b."resource!name" AS technician,
w."msdyn_serviceterritory!name" AS territory,
a.address1_line1, a.address1_city, a.address1_stateorprovince, a.address1_postalcode,
a.address1_latitude, a.address1_longitude, a.name AS account_display_name
FROM bookableresourcebooking b
LEFT JOIN msdyn_workorder w ON b."msdyn_workorder!name" = w.msdyn_name
LEFT JOIN account a ON COALESCE(b."hsl_serviceaccountid!id", w."msdyn_serviceaccount!id") = a.accountid
WHERE b.starttime >= ?
AND b.starttime <= ?
AND b."bookingstatus!name" IN ({active_booking_statuses})
{tech_clause}
{filter_sql}
ORDER BY b.starttime ASC
LIMIT ?
""",
params,
)
rows = cur.fetchall()
workorders = []
for r in rows:
lat = r["address1_latitude"]
lng = r["address1_longitude"]
gps = None
if lat and lng:
try:
gps = {"lat": float(lat), "lng": float(lng)}
except (ValueError, TypeError):
pass
workorders.append(
{
"id": r["msdyn_workorderid"],
"name": r["msdyn_name"],
"account_name": r["account_name"],
"account_display_name": r["account_display_name"],
"summary": r["msdyn_workordersummary"],
"status_code": r["msdyn_systemstatus"],
"status": _status_label(r["msdyn_systemstatus"]),
"booking_status": r["booking_status"],
"priority": r["priority"],
"work_type": r["work_type"],
"incident_type": r["incident_type"],
"technician": r["technician"],
"territory": r["territory"],
"onsite_duration": r["hsl_onsiteduration"],
"starttime": r["starttime"],
"endtime": r["endtime"],
"address": {
"line1": r["address1_line1"],
"city": r["address1_city"],
"state": r["address1_stateorprovince"],
"postal": r["address1_postalcode"],
},
"gps": gps,
}
)
return {
"date": today,
"total": len(workorders),
"with_gps": sum(1 for w in workorders if w["gps"]),
"without_gps": sum(1 for w in workorders if not w["gps"]),
"workorders": workorders,
"booking_statuses": ["Scheduled", "Unscheduled", "On Site", "Traveling", "On Break", "Hold", "Committed"],
}
finally:
conn.close()
# ── List Technicians ──────────────────────────────────────────────────────
@app.get("/api/workorders/technicians")
async def workorders_technicians():
"""Return distinct technician names from bookings."""
conn = _get_extraction_db()
if not conn:
raise HTTPException(503, "Database not available — sync extraction DB missing")
try:
cur = conn.cursor()
cur.execute(
"""SELECT DISTINCT name AS technician FROM bookableresource WHERE name IS NOT NULL AND name != ''
UNION
SELECT DISTINCT b."resource!name" AS technician FROM bookableresourcebooking b
WHERE b."resource!name" IS NOT NULL AND b."resource!name" != ''
UNION
SELECT DISTINCT w."hsl_bookedresource1!name" FROM msdyn_workorder w
WHERE w."hsl_bookedresource1!name" IS NOT NULL AND w."hsl_bookedresource1!name" != ''
UNION
SELECT DISTINCT w."hsl_bookedresource2!name" FROM msdyn_workorder w
WHERE w."hsl_bookedresource2!name" IS NOT NULL AND w."hsl_bookedresource2!name" != ''
UNION
SELECT DISTINCT w."hsl_bookedresource3!name" FROM msdyn_workorder w
WHERE w."hsl_bookedresource3!name" IS NOT NULL AND w."hsl_bookedresource3!name" != ''
ORDER BY technician"""
)
techs = [r["technician"] for r in cur.fetchall()]
return {"technicians": techs, "total": len(techs)}
finally:
conn.close()
# ── Work Order List (filterable, paginated) ─────────────────────────────────
@app.get("/api/workorders/list")
async def workorders_list(
q: str = Query("", description="Search term"),
status: str = Query("", description="Comma-separated status codes to include"),
tech: str = Query("", description="Comma-separated technician names"),
date_from: str = Query("", description="ISO date filter start (inclusive)"),
date_to: str = Query("", description="ISO date filter end (inclusive)"),
date_field: str = Query("createdon", description="Date field to filter on: createdon, msdyn_completedon"),
limit: int = Query(50, ge=1, le=500),
offset: int = Query(0, ge=0),
):
"""Fetch a filtered, paginated list of work orders from the extraction DB.
Supports text search, status filter, technician filter, and date range.
Date range can be applied to different date fields (default: date window start).
"""
conn = _get_extraction_db()
if not conn:
raise HTTPException(503, "Database not available — sync extraction DB missing")
try:
cur = conn.cursor()
params: list = []
# Date field validation
valid_date_fields = {"createdon", "msdyn_completedon", "msdyn_datewindowstart"}
if date_field not in valid_date_fields:
date_field = "createdon"
where_clauses: list[str] = []
# Text search
if q:
like = f"%{q}%"
where_clauses.append(
"(w.msdyn_name LIKE ? OR a.name LIKE ? OR a.address1_city LIKE ?)"
)
params.extend([like, like, like])
# Status filter
if status:
status_codes = [s.strip() for s in status.split(",") if s.strip()]
if status_codes:
placeholders = ",".join(["?" for _ in status_codes])
where_clauses.append(f"w.msdyn_systemstatus IN ({placeholders})")
params.extend(status_codes)
# Technician filter — check bookings table AND hsl_bookedresource fields
if tech:
tech_names = [t.strip() for t in tech.split(",") if t.strip()]
if tech_names:
placeholders = ",".join(["?" for _ in tech_names])
tech_conditions = [
f'EXISTS (SELECT 1 FROM bookableresourcebooking b '
f'WHERE b."msdyn_workorder!name" = w.msdyn_name '
f'AND b."resource!name" IN ({placeholders}))',
f'w."hsl_bookedresource1!name" IN ({placeholders})',
f'w."hsl_bookedresource2!name" IN ({placeholders})',
f'w."hsl_bookedresource3!name" IN ({placeholders})',
]
where_clauses.append("(" + " OR ".join(tech_conditions) + ")")
params.extend(tech_names * 4)
# Date range filter
if date_from:
where_clauses.append(f"w.{date_field} >= ?")
params.append(f"{date_from}T00:00:00Z")
if date_to:
where_clauses.append(f"w.{date_field} <= ?")
params.append(f"{date_to}T23:59:59Z")
where_sql = " AND ".join(where_clauses) if where_clauses else "1=1"
# Count query
count_sql = f"""
SELECT COUNT(*) FROM msdyn_workorder w
LEFT JOIN account a ON w."msdyn_serviceaccount!id" = a.accountid
WHERE {where_sql}
"""
cur.execute(count_sql, params)
total = cur.fetchone()[0]
# Data query
data_sql = f"""
SELECT
w.msdyn_workorderid,
w.msdyn_name,
w."msdyn_serviceaccount!name" AS account_name,
w.msdyn_workordersummary,
w.msdyn_datewindowstart,
w.msdyn_datewindowend,
w.msdyn_timewindowstart,
w.msdyn_timewindowend,
w.msdyn_completedon,
w.msdyn_systemstatus,
w."msdyn_priority!name" AS priority,
w."msdyn_workordertype!name" AS work_type,
w."msdyn_primaryincidenttype!name" AS incident_type,
w."msdyn_serviceterritory!name" AS territory,
w.msdyn_instructions,
w.hsl_onsiteduration,
w.msdyn_address1,
w.msdyn_city,
w.msdyn_stateorprovince,
w.msdyn_postalcode,
w.msdyn_latitude,
w.msdyn_longitude,
w."msdyn_customerasset!name" AS customer_asset_name,
(SELECT GROUP_CONCAT(DISTINCT t)
FROM (
SELECT b2."resource!name" AS t FROM bookableresourcebooking b2 WHERE b2."msdyn_workorder!name" = w.msdyn_name
UNION
SELECT w2."hsl_bookedresource1!name" FROM msdyn_workorder w2 WHERE w2.msdyn_workorderid = w.msdyn_workorderid AND w2."hsl_bookedresource1!name" IS NOT NULL
UNION
SELECT w2."hsl_bookedresource2!name" FROM msdyn_workorder w2 WHERE w2.msdyn_workorderid = w.msdyn_workorderid AND w2."hsl_bookedresource2!name" IS NOT NULL
UNION
SELECT w2."hsl_bookedresource3!name" FROM msdyn_workorder w2 WHERE w2.msdyn_workorderid = w.msdyn_workorderid AND w2."hsl_bookedresource3!name" IS NOT NULL
)) AS technicians,
a.name AS account_display_name,
a.address1_line1,
a.address1_city,
a.address1_stateorprovince,
a.address1_postalcode,
a.address1_latitude,
a.address1_longitude
FROM msdyn_workorder w
LEFT JOIN account a ON w."msdyn_serviceaccount!id" = a.accountid
WHERE {where_sql}
ORDER BY w.msdyn_datewindowstart DESC, w.msdyn_name
LIMIT ? OFFSET ?
"""
data_params = params + [limit, offset]
cur.execute(data_sql, data_params)
rows = cur.fetchall()
workorders = []
for r in rows:
lat = r["msdyn_latitude"] or r["address1_latitude"]
lng = r["msdyn_longitude"] or r["address1_longitude"]
gps = None
if lat and lng:
try:
gps = {"lat": float(lat), "lng": float(lng)}
except (ValueError, TypeError):
pass
workorders.append(
{
"id": r["msdyn_workorderid"],
"name": r["msdyn_name"],
"account_name": r["account_name"],
"account_display_name": r["account_display_name"],
"summary": r["msdyn_workordersummary"],
"status_code": str(r["msdyn_systemstatus"]),
"status": _status_label(r["msdyn_systemstatus"]),
"priority": r["priority"],
"work_type": r["work_type"],
"incident_type": r["incident_type"],
"territory": r["territory"],
"customer_asset_name": r["customer_asset_name"],
"instructions": r["msdyn_instructions"],
"onsite_duration": r["hsl_onsiteduration"],
"datewindow_start": r["msdyn_datewindowstart"],
"datewindow_end": r["msdyn_datewindowend"],
"timewindow_start": r["msdyn_timewindowstart"],
"timewindow_end": r["msdyn_timewindowend"],
"completed_on": r["msdyn_completedon"],
"address_line1": r["msdyn_address1"] or r["address1_line1"],
"city": r["msdyn_city"] or r["address1_city"],
"state": r["msdyn_stateorprovince"] or r["address1_stateorprovince"],
"postal": r["msdyn_postalcode"] or r["address1_postalcode"],
"gps": gps,
"technicians": r["technicians"],
}
)
# Build status summary
status_counts = {}
for wo in workorders:
s = wo["status"]
status_counts[s] = status_counts.get(s, 0) + 1
return {
"results": workorders,
"total": total,
"offset": offset,
"limit": limit,
"status_summary": status_counts,
"available_statuses": [
{"code": k, "label": v} for k, v in STATUS_LABELS.items()
],
}
finally:
conn.close()
# ── Route Optimization ──────────────────────────────────────────────────────
def _lookup_asset_gps_by_machine_id(machine_id: str) -> dict | None:
"""Lookup GPS from the canteen-asset-tracker assets table by machine_id."""
try:
conn = get_db()
row = conn.execute(
"SELECT latitude, longitude FROM assets WHERE machine_id = ? "
"AND latitude IS NOT NULL AND latitude != 0 "
"AND longitude IS NOT NULL AND longitude != 0",
(machine_id,),
).fetchone()
conn.close()
if row:
return {"lat": float(row["latitude"]), "lng": float(row["longitude"])}
except Exception:
pass
return None
class RouteOptimizeRequest(BaseModel):
work_order_ids: list[str]
origin: dict | None = None # {"lat": 28.5, "lng": -81.3} or null
start_at: str | None = None # optional name/label for origin stop
@app.post("/api/route/optimize")
async def route_optimize(body: RouteOptimizeRequest):
"""Optimize a route for a set of work orders using TSP (nearest-neighbor + 2-opt)."""
wo_ids = body.work_order_ids
origin = body.origin
if not wo_ids:
raise HTTPException(400, "No work order IDs provided")
if len(wo_ids) > 50:
raise HTTPException(400, "Max 50 work orders per route")
conn = _get_extraction_db()
if not conn:
raise HTTPException(503, "Database not available — sync extraction DB missing")
try:
cur = conn.cursor()
placeholders = ",".join(["?"] * len(wo_ids))
cur.execute(
f"""
SELECT
w.msdyn_workorderid, w.msdyn_name,
w."msdyn_serviceaccount!name" AS account_name,
w."msdyn_serviceaccount!id" AS account_id,
w.msdyn_workordersummary,
w.msdyn_datewindowstart, w.msdyn_timewindowstart, w.msdyn_timewindowend,
w.hsl_onsiteduration,
w."msdyn_priority!name" AS priority,
w.msdyn_systemstatus,
w."msdyn_workordertype!name" AS work_type,
w."msdyn_primaryincidenttype!name" AS incident_type,
w."msdyn_serviceterritory!name" AS territory,
w."msdyn_customerasset!id" AS asset_id,
w."msdyn_customerasset!name" AS asset_name,
ca.msdyn_latitude AS asset_lat, ca.msdyn_longitude AS asset_lng, ca.hsl_lobassetnumber,
ca."msdyn_functionallocation!name" AS asset_location_in_facility,
ca."msdyn_account!name" AS asset_account_name,
a.address1_line1, a.address1_city, a.address1_stateorprovince, a.address1_postalcode,
a.address1_latitude, a.address1_longitude, a.name AS account_display_name
FROM msdyn_workorder w
LEFT JOIN account a ON w."msdyn_serviceaccount!id" = a.accountid
LEFT JOIN msdyn_customerasset ca ON w."msdyn_customerasset!id" = ca.msdyn_customerassetid
WHERE w.msdyn_name IN ({placeholders})
OR w.msdyn_workorderid IN ({placeholders})
""",
wo_ids + wo_ids,
)
rows = cur.fetchall()
finally:
conn.close()
if not rows:
raise HTTPException(404, "No matching work orders found in DB")
# Build stop list
stops = []
for r in rows:
addr_lat = r["address1_latitude"]
addr_lng = r["address1_longitude"]
address_gps = None
if addr_lat and addr_lng:
try:
address_gps = {"lat": float(addr_lat), "lng": float(addr_lng)}
except (ValueError, TypeError):
pass
# Machine GPS — prefer msdyn_customerasset GPS, fall back to canteen assets table
machine_gps = None
asset_lat = r["asset_lat"]
asset_lng = r["asset_lng"]
if asset_lat and asset_lng:
try:
machine_gps = {"lat": float(asset_lat), "lng": float(asset_lng)}
except (ValueError, TypeError):
pass
if not machine_gps:
lob = r["hsl_lobassetnumber"]
if lob:
machine_gps = _lookup_asset_gps_by_machine_id(str(lob).strip())
gps = machine_gps or address_gps
stops.append(
{
"work_order_id": r["msdyn_workorderid"],
"name": r["msdyn_name"],
"account_name": r["account_name"],
"account_display_name": r["account_display_name"],
"summary": r["msdyn_workordersummary"],
"address": {
"line1": r["address1_line1"],
"city": r["address1_city"],
"state": r["address1_stateorprovince"],
"postal": r["address1_postalcode"],
},
"gps": gps,
"address_gps": address_gps,
"date_window": r["msdyn_datewindowstart"],
"time_window_start": r["msdyn_timewindowstart"],
"time_window_end": r["msdyn_timewindowend"],
"onsite_duration": r["hsl_onsiteduration"],
"priority": r["priority"],
"status_code": r["msdyn_systemstatus"],
"status": _status_label(r["msdyn_systemstatus"]),
"work_type": r["work_type"],
"incident_type": r["incident_type"],
"territory": r["territory"],
"asset_id": r["asset_id"],
"asset_name": r["asset_name"],
"asset_location_in_facility": r["asset_location_in_facility"],
"asset_account_name": r["asset_account_name"],
}
)
# Filter stops with GPS
stops_with_gps = [s for s in stops if s["gps"]]
stops_without_gps = [s for s in stops if not s["gps"]]
if not stops_with_gps:
return {
"route": [],
"error": "No stops have GPS coordinates",
"stops_without_gps": stops_without_gps,
"summary": {"total_stops": 0, "with_gps": 0, "without_gps": len(stops_without_gps)},
}
# TSP
points = [(s["gps"]["lat"], s["gps"]["lng"]) for s in stops_with_gps]
has_origin = origin and "lat" in origin and "lng" in origin
if has_origin:
points.insert(0, (origin["lat"], origin["lng"]))
route_indices = _solve_tsp(points, origin_idx=0)
route_stops = [stops_with_gps[i - 1] for i in route_indices if i > 0]
else:
route_indices = _solve_tsp(points, origin_idx=0)
route_stops = [stops_with_gps[i] for i in route_indices]
# Calculate distances and drive times
total_distance_km = 0
total_drive_min = 0
for i in range(len(route_stops) - 1):
p1 = (route_stops[i]["gps"]["lat"], route_stops[i]["gps"]["lng"])
p2 = (route_stops[i + 1]["gps"]["lat"], route_stops[i + 1]["gps"]["lng"])
d = _haversine(*p1, *p2)
drive_min = (d / 50) * 60 # avg 50 km/h
route_stops[i]["distance_to_next_km"] = round(d, 1)
route_stops[i]["drive_min_to_next"] = round(drive_min)
total_distance_km += d
total_drive_min += drive_min
if route_stops:
route_stops[-1]["distance_to_next_km"] = 0
route_stops[-1]["drive_min_to_next"] = 0
# Cumulative time from start
cumulative_min = 0
if has_origin and route_stops:
o_lat, o_lng = origin["lat"], origin["lng"]
f_lat, f_lng = route_stops[0]["gps"]["lat"], route_stops[0]["gps"]["lng"]
first_leg = _haversine(o_lat, o_lng, f_lat, f_lng)
cumulative_min = round((first_leg / 50) * 60)
for stop in route_stops:
stop["cumulative_min"] = cumulative_min
onsite = round((stop.get("onsite_duration") or 0) / 60)
cumulative_min += (stop.get("drive_min_to_next") or 0) + onsite
total_onsite_min = sum(round((s.get("onsite_duration") or 0) / 60) for s in route_stops)
result = {
"route": route_stops,
"stops_without_gps": stops_without_gps,
"origin": origin,
"summary": {
"total_stops": len(route_stops),
"with_gps": len(stops_with_gps),
"without_gps": len(stops_without_gps),
"total_distance_km": round(total_distance_km, 1),
"total_drive_min": round(total_drive_min),
"total_onsite_min": total_onsite_min,
"total_estimated_min": round(total_drive_min + total_onsite_min),
"origin": origin,
},
}
return result
# ─── Static Files (mounted last to not shadow routes) ──────────────────────
app.mount("/uploads", StaticFiles(directory=str(UPLOADS_DIR)), name="uploads")
if STATIC_DIR.exists():
class NoCacheStaticFiles(StaticFiles):
async def get_response(self, path, scope):
resp = await super().get_response(path, scope)
# No-cache for all static assets (HTML, images, JS, CSS)
resp.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
return resp
app.mount("/", NoCacheStaticFiles(directory=str(STATIC_DIR), html=True), name="static")
Reference in New Issue View Git Blame Copy Permalink