shawn
064740ef2e
Add env var overrides matching main app pattern: - CANTEEN_IMPORT_PORT — port override (default 8091) - CANTEEN_IMPORT_DB_PATH — database path override - CANTEEN_IMPORT_SEED_DIR — seed data directory override - CANTEEN_IMPORT_UPLOAD_DIR — upload directory override Create start scripts in repo: - start.sh — production (default port 8091) - start-dev.sh — development (port 8093, assets.dev.db) Fix: dev clone now pulls from Gitea (not local prod dir)
626 lines
21 KiB
Python
626 lines
21 KiB
Python
"""
|
|
🌱 Canteen Seed Import — Web UX Server
|
|
|
|
FastAPI server wrapping the seed import pipeline:
|
|
Upload Excel → Preview parse → Import with backup → View report
|
|
|
|
Usage:
|
|
cd ~/projects/canteen-seed-import
|
|
python3 web/server.py --port 8091
|
|
"""
|
|
|
|
import argparse
|
|
import csv
|
|
import glob
|
|
import gzip
|
|
import io
|
|
import json
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import traceback
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from fastapi import FastAPI, HTTPException, Query, Request, UploadFile, File
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import FileResponse, JSONResponse, Response
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
# ─── Import pipeline modules (relative to project root) ──────────────────────
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
from parser import parse_excel
|
|
from db_writer import write_seed, write_update, ALL_COLUMNS, PRESERVE_FIELDS, AUTO_FIELDS
|
|
from backup import backup as do_backup, compare as compare_backup
|
|
from sync import push as sync_push, TARGETS as SYNC_TARGETS
|
|
|
|
|
|
# ─── Helpers ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _list_backup_files() -> list[dict]:
|
|
"""List backup files in the seed-data directory OR project root."""
|
|
patterns = [
|
|
str(PROJECT_ROOT / "seed-data" / "backup_*.json.gz"),
|
|
str(PROJECT_ROOT / "backup_*.json.gz"),
|
|
]
|
|
files = set()
|
|
for pat in patterns:
|
|
for f in glob.glob(pat):
|
|
files.add(f)
|
|
files = sorted(files, reverse=True)
|
|
result = []
|
|
for f in files:
|
|
try:
|
|
with gzip.open(f, "rb") as gz:
|
|
meta = json.loads(gz.read().decode("utf-8"))
|
|
result.append({
|
|
"path": f,
|
|
"filename": os.path.basename(f),
|
|
"created": meta.get("created", "unknown"),
|
|
"records": len(meta.get("records", [])),
|
|
"fields": meta.get("fields", []),
|
|
})
|
|
except Exception:
|
|
mtime = os.path.getmtime(f)
|
|
result.append({
|
|
"path": f,
|
|
"filename": os.path.basename(f),
|
|
"created": datetime.fromtimestamp(mtime).isoformat(),
|
|
"records": 0,
|
|
"fields": [],
|
|
})
|
|
return result
|
|
|
|
# ─── Config ──────────────────────────────────────────────────────────────────
|
|
|
|
DEFAULT_DB_PATH = os.environ.get(
|
|
"CANTEEN_IMPORT_DB_PATH",
|
|
str(PROJECT_ROOT / "seed-data" / "assets.db"),
|
|
)
|
|
DEFAULT_SEED_DIR = os.environ.get(
|
|
"CANTEEN_IMPORT_SEED_DIR",
|
|
str(PROJECT_ROOT / "seed-data"),
|
|
)
|
|
UPLOAD_DIR = Path(os.environ.get(
|
|
"CANTEEN_IMPORT_UPLOAD_DIR",
|
|
"/tmp/canteen-import-uploads",
|
|
))
|
|
REPORT_PATH = Path(DEFAULT_SEED_DIR) / "validation_report.md"
|
|
HEADERS_FILE = Path(DEFAULT_SEED_DIR) / "headers_modified.csv"
|
|
REFERENCE_FILE = Path(DEFAULT_SEED_DIR) / "reference_handbook.md"
|
|
|
|
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
# In-memory state for the current session
|
|
_session_state = {
|
|
"machines": None, # parsed machine list
|
|
"file_name": None, # uploaded file name
|
|
"parse_stats": None, # parse summary stats
|
|
"last_import": None, # result of last import
|
|
"last_backup": None, # path to last backup
|
|
}
|
|
|
|
# ─── App ─────────────────────────────────────────────────────────────────────
|
|
|
|
app = FastAPI(title="Canteen Seed Import Tool")
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
# ─── Helpers ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _make_serializable(obj):
|
|
"""Convert any object to JSON-serializable form."""
|
|
if obj is None:
|
|
return None
|
|
if isinstance(obj, (str, int, float, bool)):
|
|
return obj
|
|
if isinstance(obj, datetime):
|
|
return obj.isoformat()
|
|
if isinstance(obj, (list, tuple)):
|
|
return [_make_serializable(x) for x in obj]
|
|
if isinstance(obj, dict):
|
|
return {k: _make_serializable(v) for k, v in obj.items()}
|
|
return str(obj)
|
|
|
|
|
|
def _parse_stats(machines: list) -> dict:
|
|
"""Build summary stats from parsed machines."""
|
|
total = len(machines)
|
|
disney = sum(1 for m in machines if m.get("is_disney"))
|
|
unknown = sum(1 for m in machines if m.get("is_unknown_machine"))
|
|
ocr = sum(1 for m in machines if m.get("serial_ocr_corrected"))
|
|
hi = sum(1 for m in machines if m.get("priority") == "High")
|
|
mid = sum(1 for m in machines if m.get("priority") == "Mid")
|
|
low = sum(1 for m in machines if m.get("priority") == "Low")
|
|
|
|
# Telemetry breakdown
|
|
telemetry = {}
|
|
for m in machines:
|
|
prov = m.get("telemetry_provider", "None") or "None"
|
|
telemetry[prov] = telemetry.get(prov, 0) + 1
|
|
|
|
# Class breakdown
|
|
classes = {}
|
|
for m in machines:
|
|
cls = m.get("class", "Unknown") or "Unknown"
|
|
classes[cls] = classes.get(cls, 0) + 1
|
|
|
|
return {
|
|
"total": total,
|
|
"disney": disney,
|
|
"unknown_machines": unknown,
|
|
"ocr_corrections": ocr,
|
|
"priority": {"High": hi, "Mid": mid, "Low": low},
|
|
"telemetry": telemetry,
|
|
"classes": classes,
|
|
}
|
|
|
|
|
|
def _format_number(n):
|
|
"""Format a number with commas."""
|
|
if n is None:
|
|
return "—"
|
|
return f"{n:,}"
|
|
|
|
|
|
# ─── API Routes ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
@app.get("/api/health")
|
|
async def health():
|
|
return {"status": "ok", "time": datetime.now(timezone.utc).isoformat()}
|
|
|
|
|
|
@app.get("/api/status")
|
|
async def status():
|
|
"""Dashboard stats: DB health, machine counts, GPS coverage."""
|
|
db_path = DEFAULT_DB_PATH
|
|
result = {
|
|
"db_path": db_path,
|
|
"exists": os.path.exists(db_path),
|
|
"total_machines": 0,
|
|
"by_class": {},
|
|
"by_make": {},
|
|
"by_priority": {},
|
|
"gps_coverage": {"with_gps": 0, "without_gps": 0},
|
|
"last_modified": None,
|
|
"last_backup": None,
|
|
"backup_count": 0,
|
|
}
|
|
if not result["exists"]:
|
|
return result
|
|
|
|
if os.path.exists(db_path):
|
|
mtime = os.path.getmtime(db_path)
|
|
result["last_modified"] = datetime.fromtimestamp(mtime).isoformat()
|
|
|
|
try:
|
|
import sqlite3
|
|
conn = sqlite3.connect(db_path)
|
|
conn.row_factory = sqlite3.Row
|
|
cur = conn.cursor()
|
|
|
|
cur.execute("SELECT COUNT(*) as c FROM assets")
|
|
result["total_machines"] = cur.fetchone()["c"]
|
|
|
|
cur.execute("SELECT COALESCE(NULLIF(class,''),'Unknown') as cls, COUNT(*) as c FROM assets GROUP BY cls ORDER BY c DESC")
|
|
result["by_class"] = {r["cls"]: r["c"] for r in cur.fetchall()}
|
|
|
|
cur.execute("SELECT COALESCE(NULLIF(make,''),'Unknown') as mk, COUNT(*) as c FROM assets GROUP BY mk ORDER BY c DESC")
|
|
result["by_make"] = {r["mk"]: r["c"] for r in cur.fetchall()}
|
|
|
|
cur.execute("SELECT COALESCE(NULLIF(disney_area,''),'Non-Disney') as area, COUNT(*) as c FROM assets GROUP BY area ORDER BY c DESC")
|
|
result["by_disney_area"] = {r["area"]: r["c"] for r in cur.fetchall()}
|
|
|
|
cur.execute("SELECT COALESCE(NULLIF(priority,''),'None') as pri, COUNT(*) as c FROM assets GROUP BY pri ORDER BY pri")
|
|
result["by_priority"] = {r["pri"]: r["c"] for r in cur.fetchall()}
|
|
|
|
cur.execute("SELECT COUNT(*) as c FROM assets WHERE latitude IS NOT NULL AND latitude != 0")
|
|
result["gps_coverage"]["with_gps"] = cur.fetchone()["c"]
|
|
result["gps_coverage"]["without_gps"] = result["total_machines"] - result["gps_coverage"]["with_gps"]
|
|
|
|
conn.close()
|
|
except Exception:
|
|
pass
|
|
|
|
backups = _list_backup_files()
|
|
result["backup_count"] = len(backups)
|
|
if backups:
|
|
result["last_backup"] = backups[0].get("created", "")
|
|
return result
|
|
|
|
|
|
@app.get("/api/backups")
|
|
async def list_backups():
|
|
"""List available backup files."""
|
|
return {"backups": _list_backup_files()}
|
|
|
|
|
|
@app.get("/api/columns")
|
|
async def columns():
|
|
"""Return column definitions: which are auto vs preserve vs primary key."""
|
|
return {
|
|
"all": ALL_COLUMNS,
|
|
"preserve": sorted(PRESERVE_FIELDS),
|
|
"auto": sorted(AUTO_FIELDS),
|
|
"primary_key": "machine_id",
|
|
}
|
|
|
|
|
|
@app.get("/api/headers")
|
|
async def headers():
|
|
"""Return the header mapping CSV as JSON."""
|
|
if not HEADERS_FILE.exists():
|
|
return {"headers": [], "note": f"{HEADERS_FILE} not found"}
|
|
rows = []
|
|
with open(HEADERS_FILE, newline="", encoding="utf-8") as f:
|
|
reader = csv.DictReader(f)
|
|
for row in reader:
|
|
rows.append(row)
|
|
return {"headers": rows, "count": len(rows)}
|
|
|
|
|
|
@app.get("/api/reference")
|
|
async def reference():
|
|
"""Return the reference handbook markdown."""
|
|
if not REFERENCE_FILE.exists():
|
|
return {"content": "", "note": f"{REFERENCE_FILE} not found"}
|
|
with open(REFERENCE_FILE, encoding="utf-8") as f:
|
|
content = f.read()
|
|
return {"content": content, "size": len(content)}
|
|
|
|
|
|
@app.post("/api/upload")
|
|
async def upload(file: UploadFile = File(...)):
|
|
"""Upload an Excel file and parse it. Returns preview + stats."""
|
|
if not file.filename:
|
|
raise HTTPException(400, "No file provided")
|
|
|
|
suffix = Path(file.filename).suffix.lower()
|
|
if suffix not in (".xlsx", ".xls", ".csv"):
|
|
raise HTTPException(400, f"Unsupported file type: {suffix}. Use .xlsx, .xls, or .csv")
|
|
|
|
# Save to temp
|
|
upload_path = UPLOAD_DIR / f"{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}_{file.filename}"
|
|
with open(upload_path, "wb") as f:
|
|
content = await file.read()
|
|
f.write(content)
|
|
|
|
# Parse
|
|
try:
|
|
machines = parse_excel(str(upload_path))
|
|
except Exception as e:
|
|
traceback.print_exc()
|
|
raise HTTPException(500, f"Parse error: {e}")
|
|
|
|
# Build stats
|
|
stats = _parse_stats(machines)
|
|
|
|
# Build preview (first 50 rows, key columns)
|
|
preview_cols = [
|
|
"machine_id", "serial_number", "name", "class", "make", "model",
|
|
"company", "place", "building_name", "floor", "zone",
|
|
"priority", "telemetry_provider", "disney_park", "disney_area", "is_disney",
|
|
"yearly_sales", "has_cashless", "alerts_summary",
|
|
]
|
|
preview = []
|
|
for m in machines[:50]:
|
|
row = {}
|
|
for col in preview_cols:
|
|
row[col] = _make_serializable(m.get(col))
|
|
preview.append(row)
|
|
|
|
# Store in session
|
|
_session_state["machines"] = machines
|
|
_session_state["file_name"] = file.filename
|
|
_session_state["parse_stats"] = stats
|
|
|
|
# Cleanup temp file (keep parsed data in memory)
|
|
try:
|
|
os.unlink(str(upload_path))
|
|
except OSError:
|
|
pass
|
|
|
|
return {
|
|
"file_name": file.filename,
|
|
"stats": stats,
|
|
"preview": preview,
|
|
"preview_count": len(preview),
|
|
"total_rows": len(machines),
|
|
"columns_previewed": preview_cols,
|
|
}
|
|
|
|
|
|
@app.post("/api/import")
|
|
async def run_import(
|
|
mode: str = Query("seed", pattern="^(seed|update)$"),
|
|
backup: bool = Query(True),
|
|
db_path: str = Query(DEFAULT_DB_PATH),
|
|
):
|
|
"""Run the import — seed or update mode. Optionally backup first."""
|
|
if _session_state["machines"] is None:
|
|
raise HTTPException(400, "No data parsed. Upload a file first.")
|
|
|
|
machines = _session_state["machines"]
|
|
backup_path = None
|
|
|
|
# Backup if requested
|
|
if backup and os.path.exists(db_path):
|
|
try:
|
|
backup_path = do_backup(db_path)
|
|
_session_state["last_backup"] = backup_path
|
|
except Exception as e:
|
|
traceback.print_exc()
|
|
# Don't fail the whole import if backup fails
|
|
backup_path = f"ERROR: {e}"
|
|
|
|
# Run import
|
|
try:
|
|
if mode == "seed":
|
|
stats = write_seed(machines, db_path)
|
|
else:
|
|
stats = write_update(machines, db_path)
|
|
except Exception as e:
|
|
traceback.print_exc()
|
|
raise HTTPException(500, f"Import error: {e}")
|
|
|
|
_session_state["last_import"] = {
|
|
"mode": mode,
|
|
"db_path": db_path,
|
|
"stats": stats,
|
|
"backup_path": backup_path,
|
|
"time": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
|
|
return {
|
|
"mode": mode,
|
|
"stats": stats,
|
|
"backup_path": backup_path,
|
|
"machine_count": len(machines),
|
|
}
|
|
|
|
|
|
@app.get("/api/report")
|
|
async def report(refresh: bool = Query(False)):
|
|
"""Return the latest validation report."""
|
|
if not REPORT_PATH.exists():
|
|
return {"content": "", "note": "No report generated yet. Run an import first."}
|
|
with open(REPORT_PATH, encoding="utf-8") as f:
|
|
content = f.read()
|
|
return {
|
|
"content": content,
|
|
"size": len(content),
|
|
"path": str(REPORT_PATH),
|
|
}
|
|
|
|
|
|
@app.get("/api/session")
|
|
async def session():
|
|
"""Return current session state."""
|
|
return {
|
|
"file_name": _session_state["file_name"],
|
|
"parse_stats": _session_state["parse_stats"],
|
|
"last_import": _session_state["last_import"],
|
|
"last_backup": _session_state["last_backup"],
|
|
"has_data": _session_state["machines"] is not None,
|
|
"machine_count": len(_session_state["machines"]) if _session_state["machines"] else 0,
|
|
}
|
|
|
|
|
|
@app.post("/api/backup")
|
|
async def api_backup(db_path: str = Query(DEFAULT_DB_PATH)):
|
|
"""Create a backup of the current database."""
|
|
if not os.path.exists(db_path):
|
|
raise HTTPException(404, f"Database not found: {db_path}")
|
|
try:
|
|
path = do_backup(db_path)
|
|
_session_state["last_backup"] = path
|
|
return {"backup_path": path, "status": "ok"}
|
|
except Exception as e:
|
|
traceback.print_exc()
|
|
raise HTTPException(500, f"Backup error: {e}")
|
|
|
|
|
|
@app.post("/api/compare")
|
|
async def api_compare(
|
|
backup_path: str = Query(...),
|
|
db_path: str = Query(DEFAULT_DB_PATH),
|
|
):
|
|
"""Compare a backup against the live database."""
|
|
if not os.path.exists(backup_path):
|
|
raise HTTPException(404, f"Backup not found: {backup_path}")
|
|
try:
|
|
changes = compare_backup(backup_path, db_path)
|
|
return {"changes": changes, "backup": backup_path, "db": db_path}
|
|
except Exception as e:
|
|
traceback.print_exc()
|
|
raise HTTPException(500, f"Compare error: {e}")
|
|
|
|
|
|
@app.post("/api/backup/restore")
|
|
async def api_restore(
|
|
backup_path: str = Query(...),
|
|
db_path: str = Query(DEFAULT_DB_PATH),
|
|
):
|
|
"""Restore GPS/photo fields from a backup."""
|
|
if not os.path.exists(backup_path):
|
|
raise HTTPException(404, f"Backup not found: {backup_path}")
|
|
try:
|
|
from backup import restore as do_restore
|
|
do_restore(backup_path, db_path)
|
|
return {"status": "ok", "restored_from": os.path.basename(backup_path), "db": db_path}
|
|
except Exception as e:
|
|
traceback.print_exc()
|
|
raise HTTPException(500, f"Restore error: {e}")
|
|
|
|
|
|
# ─── GPS-only endpoints ───────────────────────────────────────────────────────
|
|
|
|
|
|
@app.post("/api/gps/backup")
|
|
async def api_gps_backup(db_path: str = Query(DEFAULT_DB_PATH)):
|
|
"""Backup GPS data only (latitude, longitude) — ideal before wiping."""
|
|
if not os.path.exists(db_path):
|
|
raise HTTPException(404, f"Database not found: {db_path}")
|
|
try:
|
|
from backup import backup_gps as do_backup_gps
|
|
path = do_backup_gps(db_path)
|
|
_session_state["last_backup"] = path
|
|
return {"backup_path": path, "status": "ok", "type": "gps_only"}
|
|
except Exception as e:
|
|
traceback.print_exc()
|
|
raise HTTPException(500, f"GPS backup error: {e}")
|
|
|
|
|
|
@app.post("/api/gps/restore")
|
|
async def api_gps_restore(
|
|
backup_path: str = Query(...),
|
|
db_path: str = Query(DEFAULT_DB_PATH),
|
|
):
|
|
"""Restore GPS data only (latitude, longitude) from a GPS-only backup."""
|
|
if not os.path.exists(backup_path):
|
|
raise HTTPException(404, f"Backup not found: {backup_path}")
|
|
try:
|
|
from backup import restore_gps as do_restore_gps
|
|
do_restore_gps(backup_path, db_path)
|
|
return {"status": "ok", "restored_from": os.path.basename(backup_path), "db": db_path}
|
|
except Exception as e:
|
|
traceback.print_exc()
|
|
raise HTTPException(500, f"GPS restore error: {e}")
|
|
|
|
|
|
@app.post("/api/gps/restore/upload")
|
|
async def api_gps_restore_upload(
|
|
backup_file: UploadFile = File(...),
|
|
db_path: str = Query(DEFAULT_DB_PATH),
|
|
):
|
|
"""Upload a GPS backup file and restore it into the database."""
|
|
if not backup_file.filename:
|
|
raise HTTPException(400, "No file provided")
|
|
suffix = Path(backup_file.filename).suffix.lower()
|
|
if suffix not in (".gz", ".json"):
|
|
raise HTTPException(400, f"Unsupported file type: {suffix}. Upload .json.gz or .json")
|
|
try:
|
|
# Save to temp
|
|
import tempfile
|
|
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".json.gz")
|
|
content = await backup_file.read()
|
|
tmp.write(content)
|
|
tmp_path = tmp.name
|
|
tmp.close()
|
|
|
|
from backup import restore_gps as do_restore_gps
|
|
do_restore_gps(tmp_path, db_path)
|
|
|
|
os.unlink(tmp_path)
|
|
|
|
return {"status": "ok", "restored_from": backup_file.filename, "db": db_path}
|
|
except Exception as e:
|
|
traceback.print_exc()
|
|
raise HTTPException(500, f"GPS restore error: {e}")
|
|
|
|
|
|
# ─── Database management ───────────────────────────────────────────────────────
|
|
|
|
|
|
@app.post("/api/db/wipe")
|
|
async def api_db_wipe(db_path: str = Query(DEFAULT_DB_PATH)):
|
|
"""Wipe all asset records from the database. Does NOT delete the file."""
|
|
if not os.path.exists(db_path):
|
|
raise HTTPException(404, f"Database not found: {db_path}")
|
|
try:
|
|
import sqlite3
|
|
conn = sqlite3.connect(db_path)
|
|
cur = conn.execute("SELECT COUNT(*) FROM assets")
|
|
before = cur.fetchone()[0]
|
|
conn.execute("DELETE FROM assets")
|
|
conn.commit()
|
|
conn.close()
|
|
return {
|
|
"status": "ok",
|
|
"records_deleted": before,
|
|
"db_path": db_path,
|
|
"note": "All asset records wiped. Database file preserved.",
|
|
}
|
|
except Exception as e:
|
|
traceback.print_exc()
|
|
raise HTTPException(500, f"Wipe error: {e}")
|
|
|
|
|
|
# ─── Sync / Push to Main App ──────────────────────────────────────────────────
|
|
|
|
|
|
@app.post("/api/sync")
|
|
async def api_sync(
|
|
target: str = Query("dev"),
|
|
dry_run: bool = Query(False),
|
|
):
|
|
"""Push seed import data to the main app database."""
|
|
if target not in SYNC_TARGETS:
|
|
raise HTTPException(400, f"Unknown target: {target}. Options: {list(SYNC_TARGETS.keys())}")
|
|
target_db = SYNC_TARGETS[target]["path"]
|
|
source_db = DEFAULT_DB_PATH
|
|
report = sync_push(source_db, target_db, dry_run=dry_run)
|
|
if "error" in report:
|
|
raise HTTPException(500, report["error"])
|
|
return report
|
|
|
|
|
|
@app.get("/api/sync/targets")
|
|
async def api_sync_targets():
|
|
"""Return available sync targets."""
|
|
return {
|
|
"targets": {
|
|
name: {
|
|
"path": info["path"],
|
|
"exists": os.path.exists(info["path"]),
|
|
}
|
|
for name, info in SYNC_TARGETS.items()
|
|
},
|
|
"source": {
|
|
"path": DEFAULT_DB_PATH,
|
|
"exists": os.path.exists(DEFAULT_DB_PATH),
|
|
},
|
|
}
|
|
|
|
|
|
# ─── Static files ─────────────────────────────────────────────────────────────
|
|
|
|
STATIC_DIR = Path(__file__).parent / "static"
|
|
STATIC_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
if STATIC_DIR.exists():
|
|
app.mount("/", StaticFiles(directory=str(STATIC_DIR), html=True), name="static")
|
|
|
|
|
|
# ─── Entry point ─────────────────────────────────────────────────────────────
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
|
|
parser = argparse.ArgumentParser(description="Canteen Seed Import Web Server")
|
|
parser.add_argument("--port", type=int,
|
|
default=int(os.environ.get("CANTEEN_IMPORT_PORT", "8091")),
|
|
help="Port to listen on")
|
|
parser.add_argument("--host", type=str, default="0.0.0.0", help="Host to bind to")
|
|
args = parser.parse_args()
|
|
|
|
print(f"🌱 Canteen Seed Import Web Server")
|
|
print(f" http://localhost:{args.port}")
|
|
print(f" DB: {DEFAULT_DB_PATH}")
|
|
print(f" Seed dir: {DEFAULT_SEED_DIR}")
|
|
uvicorn.run(app, host=args.host, port=args.port, log_level="info")
|