Spaces:
Sleeping
Sleeping
| """ | |
| ChillTrack API — cold-chain temperature logging for Colcom. | |
| A technician photographs a chiller/freezer's gauge or digital display, Gemini | |
| extracts the reading, the technician confirms/edits it, and it's logged | |
| against one of the 12 fixed daily time slots (see slots.py). | |
| Firebase Admin / Gemini init patterns follow NeoFix-API/main.py and | |
| iristrack-api/main.py. Auth decorator + audit-log shape follow | |
| RentNow-Admin-API/app.py. PDF export follows sqx-api's in-memory BytesIO | |
| pattern (pdf_export.py). | |
| """ | |
| import os | |
| import uuid | |
| import secrets | |
| from datetime import datetime, timedelta | |
| from flask import Flask, request, jsonify, send_file | |
| from flask_cors import CORS | |
| from firebase_admin import auth as fb_auth, firestore, storage | |
| from auth import init_firebase, require_auth, require_role, audit_log | |
| import gemini_client | |
| import slots as slots_mod | |
| import insights | |
| import pdf_export | |
| from seed_units import seed_units | |
| app = Flask(__name__) | |
| # Allow the deployed frontend plus local dev. ALLOWED_ORIGINS can hold a | |
| # comma-separated list to add another origin (a custom domain, a preview | |
| # deploy) without a code change. | |
| DEFAULT_ORIGINS = "https://chilltrack-colcom.web.app,http://localhost:5173,http://localhost:5180" | |
| ALLOWED_ORIGINS = [o.strip() for o in os.environ.get("ALLOWED_ORIGINS", DEFAULT_ORIGINS).split(",") if o.strip()] | |
| CORS(app, origins=ALLOWED_ORIGINS) | |
| db = init_firebase() | |
| # Admin access is granted by Google sign-in from an allow-listed address, not by | |
| # a password or an invite flow. Add or remove an address here to change who can | |
| # reach the admin area; everything else about a technician account is unchanged. | |
| ADMIN_EMAILS = {e.strip().lower() for e in os.environ.get("ADMIN_EMAILS", "").split(",") if e.strip()} | |
| PROFILES = "ctk_profiles" | |
| UNITS = "ctk_units" | |
| READINGS = "ctk_readings" | |
| DIGESTS = "ctk_digests" | |
| SHARES = "ctk_shares" | |
| DIGEST_MAX_AGE = timedelta(hours=24) | |
| SHARE_TTL = timedelta(days=7) | |
| # ============================================================================= | |
| # HELPERS | |
| # ============================================================================= | |
| def _actor(): | |
| """The {uid, email} pair for audit_log, from the verified token on request.user.""" | |
| user = getattr(request, "user", {}) or {} | |
| return {"uid": user.get("uid"), "email": user.get("email")} | |
| def _serialize(doc_snapshot): | |
| if not doc_snapshot.exists: | |
| return None | |
| data = doc_snapshot.to_dict() or {} | |
| data["id"] = doc_snapshot.id | |
| return data | |
| def _get_unit(unit_id): | |
| snap = db.collection(UNITS).document(unit_id).get() | |
| return _serialize(snap) | |
| def _get_profile(uid): | |
| snap = db.collection(PROFILES).document(uid).get() | |
| return _serialize(snap) | |
| def _is_admin_request(): | |
| return (getattr(request, "user", {}) or {}).get("role") == "admin" | |
| def _upload_image(path, image_bytes, content_type): | |
| """Uploads bytes to Firebase Storage. Returns the blob path, or a placeholder | |
| path (still correct/enabled code) if the Storage bucket isn't configured.""" | |
| try: | |
| bucket = storage.bucket() | |
| blob = bucket.blob(path) | |
| blob.upload_from_string(image_bytes, content_type=content_type) | |
| return path | |
| except Exception as e: | |
| # Storage bucket may not be configured yet in this environment — the upload | |
| # call above is the real, enabled code path; this is just a safe fallback | |
| # so capture still works end-to-end without a live bucket. | |
| print(f"_upload_image: storage upload failed for {path}: {e}") | |
| return path | |
| def _move_image(old_path, new_path): | |
| """Moves (copies + deletes) a blob from old_path to new_path. Best-effort.""" | |
| try: | |
| bucket = storage.bucket() | |
| source_blob = bucket.blob(old_path) | |
| if source_blob.exists(): | |
| bucket.copy_blob(source_blob, bucket, new_path) | |
| source_blob.delete() | |
| return new_path | |
| except Exception as e: | |
| print(f"_move_image: could not move {old_path} -> {new_path}: {e}") | |
| return new_path | |
| def _paginate_by_created_at(query, limit, start_after_iso): | |
| """Applies a created_at cursor + limit to a Firestore query.""" | |
| query = query.order_by("created_at", direction=firestore.Query.DESCENDING) | |
| if start_after_iso: | |
| query = query.start_after({"created_at": start_after_iso}) | |
| return query.limit(limit) | |
| # ============================================================================= | |
| # ===== AUTH ===== | |
| # ============================================================================= | |
| def google_session(): | |
| """Called right after a Google sign-in popup succeeds. Grants admin access | |
| if the signed-in address is on ADMIN_EMAILS (or is already an admin), and | |
| otherwise refuses. This is the only path into the admin role: there is no | |
| password and no invite flow, only this allow-list.""" | |
| uid = request.user["uid"] | |
| email = (request.user.get("email") or "").lower() | |
| profile_ref = db.collection(PROFILES).document(uid) | |
| profile_snap = profile_ref.get() | |
| profile = profile_snap.to_dict() if profile_snap.exists else {} | |
| already_admin = request.user.get("role") == "admin" or profile.get("role") == "admin" | |
| if email not in ADMIN_EMAILS and not already_admin: | |
| return jsonify({ | |
| "error": "This Google account is not set up for ChillTrack admin access. Ask an existing admin to add it." | |
| }), 403 | |
| if not already_admin: | |
| fb_auth.set_custom_user_claims(uid, {"role": "admin"}) | |
| profile_ref.set({ | |
| "email": email, | |
| "display_name": profile.get("display_name") or request.user.get("name") or email, | |
| "role": "admin", | |
| "active": True, | |
| "created_at": profile.get("created_at") or firestore.SERVER_TIMESTAMP, | |
| "last_login": firestore.SERVER_TIMESTAMP, | |
| }, merge=True) | |
| audit_log(db, {"uid": uid, "email": email}, "auth.google_session", "profile", uid, {"first_grant": not already_admin}) | |
| return jsonify({"uid": uid, "email": email, "role": "admin"}), 200 | |
| def auth_create_user(): | |
| body = request.get_json(silent=True) or {} | |
| email = body.get("email") | |
| display_name = body.get("display_name") | |
| role = body.get("role", "technician") | |
| if not email: | |
| return jsonify({"error": "email is required."}), 400 | |
| if role == "admin": | |
| return jsonify({ | |
| "error": "Admin access comes from Google sign-in with an address on ADMIN_EMAILS, not from this form." | |
| }), 400 | |
| temp_password = secrets.token_urlsafe(12) | |
| try: | |
| user = fb_auth.create_user(email=email, password=temp_password, display_name=display_name) | |
| fb_auth.set_custom_user_claims(user.uid, {"role": role}) | |
| db.collection(PROFILES).document(user.uid).set({ | |
| "email": email, | |
| "display_name": display_name, | |
| "role": role, | |
| "active": True, | |
| "created_at": firestore.SERVER_TIMESTAMP, | |
| }) | |
| reset_link = fb_auth.generate_password_reset_link(email) | |
| except Exception as e: | |
| return jsonify({"error": str(e)}), 400 | |
| audit_log(db, _actor(), "auth.create_user", "profile", user.uid, {"email": email, "role": role}) | |
| return jsonify({"uid": user.uid, "email": email, "role": role, "password_reset_link": reset_link}), 201 | |
| def auth_update_user(uid): | |
| body = request.get_json(silent=True) or {} | |
| profile_ref = db.collection(PROFILES).document(uid) | |
| if not profile_ref.get().exists: | |
| return jsonify({"error": "Profile not found."}), 404 | |
| if body.get("role") == "admin": | |
| return jsonify({ | |
| "error": "Admin access comes from Google sign-in with an address on ADMIN_EMAILS, not from this form." | |
| }), 400 | |
| updates = {} | |
| for field in ("role", "active", "display_name"): | |
| if field in body: | |
| updates[field] = body[field] | |
| if not updates: | |
| return jsonify({"error": "No valid fields to update."}), 400 | |
| profile_ref.update(updates) | |
| if "role" in updates: | |
| try: | |
| fb_auth.set_custom_user_claims(uid, {"role": updates["role"]}) | |
| except Exception as e: | |
| print(f"auth_update_user: failed to set custom claims for {uid}: {e}") | |
| audit_log(db, _actor(), "auth.update_user", "profile", uid, updates) | |
| return jsonify(_serialize(profile_ref.get())), 200 | |
| def auth_me(): | |
| user = request.user | |
| profile = _get_profile(user["uid"]) or {} | |
| return jsonify({ | |
| **profile, | |
| "uid": user.get("uid"), | |
| "email": user.get("email"), | |
| "role": user.get("role"), | |
| }), 200 | |
| # ============================================================================= | |
| # ===== UNITS ===== | |
| # ============================================================================= | |
| def list_units(): | |
| include_inactive = request.args.get("include_inactive") == "1" | |
| if include_inactive and not _is_admin_request(): | |
| return jsonify({"error": "Only admins may list inactive units."}), 403 | |
| query = db.collection(UNITS) | |
| if not include_inactive: | |
| query = query.where("active", "==", True) | |
| units = [_serialize(doc) for doc in query.stream()] | |
| return jsonify({"units": units}), 200 | |
| def create_unit(): | |
| body = request.get_json(silent=True) or {} | |
| if not body.get("name") or body.get("setpoint_c") is None or body.get("tolerance_c") is None: | |
| return jsonify({"error": "name, setpoint_c and tolerance_c are required."}), 400 | |
| doc = { | |
| "name": body["name"], | |
| "code": body.get("code", ""), | |
| "category": body.get("category", "other"), | |
| "setpoint_c": body["setpoint_c"], | |
| "tolerance_c": body["tolerance_c"], | |
| "display_type": body.get("display_type", "analog_dial"), | |
| "active": True, | |
| "calibration": body.get("calibration"), | |
| "created_at": firestore.SERVER_TIMESTAMP, | |
| } | |
| ref = db.collection(UNITS).add(doc)[1] | |
| audit_log(db, _actor(), "unit.create", "unit", ref.id, {"name": doc["name"]}) | |
| return jsonify(_serialize(ref.get())), 201 | |
| def update_unit(unit_id): | |
| ref = db.collection(UNITS).document(unit_id) | |
| if not ref.get().exists: | |
| return jsonify({"error": "Unit not found."}), 404 | |
| body = request.get_json(silent=True) or {} | |
| allowed = ("name", "code", "category", "setpoint_c", "tolerance_c", "display_type", "active", "calibration") | |
| updates = {field: body[field] for field in allowed if field in body} | |
| if not updates: | |
| return jsonify({"error": "No valid fields to update."}), 400 | |
| ref.update(updates) | |
| audit_log(db, _actor(), "unit.update", "unit", unit_id, updates) | |
| return jsonify(_serialize(ref.get())), 200 | |
| def delete_unit(unit_id): | |
| ref = db.collection(UNITS).document(unit_id) | |
| if not ref.get().exists: | |
| return jsonify({"error": "Unit not found."}), 404 | |
| hard = request.args.get("hard") == "1" | |
| if hard: | |
| ref.delete() | |
| audit_log(db, _actor(), "unit.delete_hard", "unit", unit_id) | |
| else: | |
| ref.update({"active": False}) | |
| audit_log(db, _actor(), "unit.delete_soft", "unit", unit_id) | |
| return jsonify({"success": True, "hard": hard}), 200 | |
| def units_seed_endpoint(): | |
| force = request.args.get("force") == "1" | |
| summary = seed_units(db, force=force) | |
| audit_log(db, _actor(), "unit.seed", "units", "bulk", summary) | |
| return jsonify(summary), 200 | |
| # ============================================================================= | |
| # ===== CAPTURE / READINGS ===== | |
| # ============================================================================= | |
| def extract_reading(): | |
| if "image" not in request.files: | |
| return jsonify({"error": "An 'image' file is required."}), 400 | |
| unit_id = request.form.get("unit_id") | |
| if not unit_id: | |
| return jsonify({"error": "unit_id is required."}), 400 | |
| unit = _get_unit(unit_id) | |
| if not unit: | |
| return jsonify({"error": "Unit not found."}), 404 | |
| image_file = request.files["image"] | |
| image_bytes = image_file.read() | |
| mime_type = image_file.mimetype or "image/jpeg" | |
| try: | |
| ai_raw = gemini_client.extract_reading(image_bytes, mime_type, unit) | |
| except Exception as e: | |
| return jsonify({"error": f"Extraction failed: {e}"}), 502 | |
| value_c = ai_raw.get("value_c") | |
| setpoint_c = unit.get("setpoint_c") | |
| tolerance_c = unit.get("tolerance_c") | |
| within_tolerance = ( | |
| abs(value_c - setpoint_c) <= tolerance_c | |
| if isinstance(value_c, (int, float)) and setpoint_c is not None and tolerance_c is not None | |
| else None | |
| ) | |
| # Sudden-jump check against the unit's most recent reading (if any). | |
| sudden_jump = False | |
| last_readings = list( | |
| db.collection(READINGS) | |
| .where("unit_id", "==", unit_id) | |
| .order_by("created_at", direction=firestore.Query.DESCENDING) | |
| .limit(1) | |
| .stream() | |
| ) | |
| if last_readings and isinstance(value_c, (int, float)) and tolerance_c is not None: | |
| last_value = last_readings[0].to_dict().get("value_c") | |
| if isinstance(last_value, (int, float)): | |
| sudden_jump = abs(value_c - last_value) > (tolerance_c * 2) | |
| anomaly_flags = [] | |
| if isinstance(ai_raw.get("confidence"), (int, float)) and ai_raw["confidence"] < 0.6: | |
| anomaly_flags.append("low_confidence") | |
| if within_tolerance is False: | |
| anomaly_flags.append("out_of_tolerance") | |
| if sudden_jump: | |
| anomaly_flags.append("sudden_jump") | |
| uid = request.user["uid"] | |
| image_path = f"pending/{uid}/{uuid.uuid4()}.jpg" | |
| image_ref = _upload_image(image_path, image_bytes, mime_type) | |
| proposed_slot = slots_mod.resolve_slot(datetime.utcnow()) | |
| proposed_slot = {**proposed_slot, "expected_at": proposed_slot["expected_at"].isoformat()} | |
| return jsonify({ | |
| "unit_id": unit_id, | |
| "ai_raw": ai_raw, | |
| "within_tolerance": within_tolerance, | |
| "anomaly_flags": anomaly_flags, | |
| "proposed_slot": proposed_slot, | |
| "image_ref": image_ref, | |
| }), 200 | |
| def confirm_reading(): | |
| body = request.get_json(silent=True) or {} | |
| required = ("unit_id", "value_c", "ai_raw", "image_ref", "shift_date", "slot_index", "slot_label", "status") | |
| missing = [f for f in required if f not in body] | |
| if missing: | |
| return jsonify({"error": f"Missing required fields: {', '.join(missing)}"}), 400 | |
| unit = _get_unit(body["unit_id"]) | |
| if not unit: | |
| return jsonify({"error": "Unit not found."}), 404 | |
| uid = request.user["uid"] | |
| profile = _get_profile(uid) or {} | |
| value_c = body["value_c"] | |
| ai_raw = body["ai_raw"] or {} | |
| ai_value = ai_raw.get("value_c") | |
| user_edited = ( | |
| abs(value_c - ai_value) > 0.05 | |
| if isinstance(value_c, (int, float)) and isinstance(ai_value, (int, float)) | |
| else True | |
| ) | |
| edit_delta = ( | |
| round(value_c - ai_value, 3) | |
| if isinstance(value_c, (int, float)) and isinstance(ai_value, (int, float)) | |
| else None | |
| ) | |
| setpoint_c = unit.get("setpoint_c") | |
| tolerance_c = unit.get("tolerance_c") | |
| within_tolerance = ( | |
| abs(value_c - setpoint_c) <= tolerance_c | |
| if isinstance(value_c, (int, float)) and setpoint_c is not None and tolerance_c is not None | |
| else None | |
| ) | |
| # Conceptually promote the image from pending/ to confirmed/. Actually moves the | |
| # blob when a real Storage bucket is configured; otherwise _move_image is a no-op | |
| # that still returns a sensible path. | |
| old_ref = body["image_ref"] | |
| new_ref = old_ref.replace("pending/", "confirmed/", 1) if old_ref.startswith("pending/") else old_ref | |
| if new_ref != old_ref: | |
| new_ref = _move_image(old_ref, new_ref) | |
| doc = { | |
| "unit_id": body["unit_id"], | |
| "unit_name": unit.get("name"), | |
| "user_id": uid, | |
| "user_display_name": profile.get("display_name") or request.user.get("email"), | |
| "value_c": value_c, | |
| "ai_raw": ai_raw, | |
| "user_edited": user_edited, | |
| "edit_delta": edit_delta, | |
| "within_tolerance": within_tolerance, | |
| "image_ref": new_ref, | |
| "shift_date": body["shift_date"], | |
| "slot_index": body["slot_index"], | |
| "slot_label": body["slot_label"], | |
| "status": body["status"], | |
| "source": "capture", | |
| "created_at": firestore.SERVER_TIMESTAMP, | |
| } | |
| ref = db.collection(READINGS).add(doc)[1] | |
| audit_log(db, _actor(), "reading.confirm", "reading", ref.id, { | |
| "unit_id": body["unit_id"], "value_c": value_c, "user_edited": user_edited, | |
| }) | |
| return jsonify(_serialize(ref.get())), 201 | |
| def my_readings(): | |
| uid = request.user["uid"] | |
| query = db.collection(READINGS).where("user_id", "==", uid) | |
| start_date = request.args.get("start_date") | |
| end_date = request.args.get("end_date") | |
| unit_id = request.args.get("unit_id") | |
| if start_date: | |
| query = query.where("shift_date", ">=", start_date) | |
| if end_date: | |
| query = query.where("shift_date", "<=", end_date) | |
| if unit_id: | |
| query = query.where("unit_id", "==", unit_id) | |
| limit = int(request.args.get("limit", 50)) | |
| start_after = request.args.get("start_after") | |
| query = _paginate_by_created_at(query, limit, start_after) | |
| readings = [_serialize(doc) for doc in query.stream()] | |
| return jsonify({"readings": readings, "limit": limit}), 200 | |
| def pending_slots(): | |
| now = datetime.utcnow() | |
| current = slots_mod.resolve_slot(now) | |
| units = [_serialize(doc) for doc in db.collection(UNITS).where("active", "==", True).stream()] | |
| existing = { | |
| doc.to_dict().get("unit_id") | |
| for doc in db.collection(READINGS) | |
| .where("shift_date", "==", current["shift_date"]) | |
| .where("slot_index", "==", current["slot_index"]) | |
| .stream() | |
| } | |
| due = [] | |
| for unit in units: | |
| if unit["id"] in existing: | |
| continue | |
| minutes_from_expected = (now - current["expected_at"]).total_seconds() / 60.0 | |
| if abs(minutes_from_expected) <= 30: | |
| due_status = "current" | |
| elif minutes_from_expected < 0: | |
| due_status = "due_soon" | |
| else: | |
| due_status = "overdue" | |
| due.append({ | |
| "unit_id": unit["id"], | |
| "unit_name": unit.get("name"), | |
| "shift_date": current["shift_date"], | |
| "slot_index": current["slot_index"], | |
| "slot_label": current["slot_label"], | |
| "due_status": due_status, | |
| }) | |
| return jsonify({"slot": {**current, "expected_at": current["expected_at"].isoformat()}, "units": due}), 200 | |
| def get_reading(reading_id): | |
| ref = db.collection(READINGS).document(reading_id) | |
| snap = ref.get() | |
| if not snap.exists: | |
| return jsonify({"error": "Reading not found."}), 404 | |
| reading = _serialize(snap) | |
| if reading.get("user_id") != request.user["uid"] and not _is_admin_request(): | |
| return jsonify({"error": "Access denied."}), 403 | |
| return jsonify(reading), 200 | |
| # ============================================================================= | |
| # ===== ADMIN: USERS & ACTIVITY ===== | |
| # ============================================================================= | |
| def admin_readings(): | |
| query = db.collection(READINGS) | |
| user_id = request.args.get("user_id") | |
| unit_id = request.args.get("unit_id") | |
| start_date = request.args.get("start_date") | |
| end_date = request.args.get("end_date") | |
| status = request.args.get("status") | |
| low_confidence = request.args.get("low_confidence") == "1" | |
| if user_id: | |
| query = query.where("user_id", "==", user_id) | |
| if unit_id: | |
| query = query.where("unit_id", "==", unit_id) | |
| if start_date: | |
| query = query.where("shift_date", ">=", start_date) | |
| if end_date: | |
| query = query.where("shift_date", "<=", end_date) | |
| if status: | |
| query = query.where("status", "==", status) | |
| readings = [_serialize(doc) for doc in query.stream()] | |
| if low_confidence: | |
| readings = [r for r in readings if (r.get("ai_raw") or {}).get("confidence", 1) < 0.6] | |
| return jsonify({"readings": readings}), 200 | |
| def admin_users(): | |
| profiles = [_serialize(doc) for doc in db.collection(PROFILES).stream()] | |
| counts = {} | |
| for doc in db.collection(READINGS).stream(): | |
| uid = doc.to_dict().get("user_id") | |
| if uid: | |
| counts[uid] = counts.get(uid, 0) + 1 | |
| for profile in profiles: | |
| profile["reading_count"] = counts.get(profile["id"], 0) | |
| return jsonify({"users": profiles}), 200 | |
| def admin_activity(): | |
| query = db.collection("ctk_audit") | |
| actor_email = request.args.get("actor") | |
| action = request.args.get("action") | |
| start_date = request.args.get("start_date") | |
| end_date = request.args.get("end_date") | |
| if action: | |
| query = query.where("action", "==", action) | |
| limit = int(request.args.get("limit", 50)) | |
| start_after = request.args.get("start_after") | |
| query = query.order_by("ts", direction=firestore.Query.DESCENDING) | |
| if start_after: | |
| query = query.start_after({"ts": start_after}) | |
| query = query.limit(limit) | |
| entries = [_serialize(doc) for doc in query.stream()] | |
| if actor_email: | |
| entries = [e for e in entries if (e.get("actor") or {}).get("email") == actor_email] | |
| if start_date: | |
| entries = [e for e in entries if str(e.get("ts", "")) >= start_date] | |
| if end_date: | |
| entries = [e for e in entries if str(e.get("ts", "")) <= end_date] | |
| return jsonify({"activity": entries}), 200 | |
| # ============================================================================= | |
| # ===== ADMIN: BULK IMPORT ===== | |
| # ============================================================================= | |
| # Backfills historical readings from a photographed paper form: an admin | |
| # uploads a photo (gallery access is fine here, unlike the technician capture | |
| # flow), Gemini transcribes every legible cell it can find, the admin reviews | |
| # and corrects the draft rows in the UI, then commits the finalized set. Every | |
| # row created this way is tagged source="bulk_import" / status="backfilled" so | |
| # it's clearly distinguishable from a live capture in insights and the audit | |
| # trail, while still counting toward historical compliance reporting. | |
| def import_extract(): | |
| if "image" not in request.files: | |
| return jsonify({"error": "An 'image' file is required."}), 400 | |
| image_file = request.files["image"] | |
| image_bytes = image_file.read() | |
| mime_type = image_file.mimetype or "image/jpeg" | |
| units = [_serialize(doc) for doc in db.collection(UNITS).where("active", "==", True).stream()] | |
| unit_names = [u["name"] for u in units] | |
| name_to_id = {u["name"]: u["id"] for u in units} | |
| try: | |
| extraction = gemini_client.extract_form(image_bytes, mime_type, unit_names) | |
| except Exception as e: | |
| return jsonify({"error": f"Extraction failed: {e}"}), 502 | |
| rows = [] | |
| for row in extraction.get("rows", []): | |
| matched_name = row.get("matched_unit_name") | |
| slot_index = row.get("slot_index") | |
| valid_slot = isinstance(slot_index, int) and 0 <= slot_index < len(slots_mod.SLOT_TIMES) | |
| rows.append({ | |
| "unit_name_seen": row.get("unit_name_seen"), | |
| "matched_unit_id": name_to_id.get(matched_name) if matched_name else None, | |
| "matched_unit_name": matched_name, | |
| "slot_index": slot_index if valid_slot else None, | |
| "slot_label": slots_mod.SLOT_TIMES[slot_index] if valid_slot else None, | |
| "value_c": row.get("value_c"), | |
| "confidence": row.get("confidence"), | |
| "raw_text_seen": row.get("raw_text_seen"), | |
| }) | |
| uid = request.user["uid"] | |
| image_path = f"imports/{uid}/{uuid.uuid4()}.jpg" | |
| image_ref = _upload_image(image_path, image_bytes, mime_type) | |
| return jsonify({ | |
| "shift_date_seen": extraction.get("shift_date_seen"), | |
| "rows": rows, | |
| "units": units, | |
| "image_ref": image_ref, | |
| }), 200 | |
| def import_commit(): | |
| body = request.get_json(silent=True) or {} | |
| shift_date = body.get("shift_date") | |
| rows = body.get("rows") | |
| image_ref = body.get("image_ref") | |
| if not shift_date or not isinstance(rows, list) or not rows: | |
| return jsonify({"error": "shift_date and at least one row are required."}), 400 | |
| uid = request.user["uid"] | |
| profile = _get_profile(uid) or {} | |
| admin_actor = {"uid": uid, "email": request.user.get("email")} | |
| created = [] | |
| skipped = 0 | |
| for row in rows: | |
| unit_id = row.get("unit_id") | |
| value_c = row.get("value_c") | |
| slot_index = row.get("slot_index") | |
| if not unit_id or not isinstance(value_c, (int, float)) or not isinstance(slot_index, int): | |
| skipped += 1 | |
| continue | |
| unit = _get_unit(unit_id) | |
| if not unit or not (0 <= slot_index < len(slots_mod.SLOT_TIMES)): | |
| skipped += 1 | |
| continue | |
| setpoint_c = unit.get("setpoint_c") | |
| tolerance_c = unit.get("tolerance_c") | |
| within_tolerance = ( | |
| abs(value_c - setpoint_c) <= tolerance_c | |
| if setpoint_c is not None and tolerance_c is not None | |
| else None | |
| ) | |
| doc = { | |
| "unit_id": unit_id, | |
| "unit_name": unit.get("name"), | |
| "user_id": uid, | |
| "user_display_name": profile.get("display_name") or request.user.get("email"), | |
| "value_c": value_c, | |
| "ai_raw": None, | |
| "user_edited": True, | |
| "edit_delta": None, | |
| "within_tolerance": within_tolerance, | |
| "image_ref": image_ref, | |
| "shift_date": shift_date, | |
| "slot_index": slot_index, | |
| "slot_label": slots_mod.SLOT_TIMES[slot_index], | |
| "status": "backfilled", | |
| "source": "bulk_import", | |
| "imported_by": admin_actor, | |
| "original_recorder_name": row.get("original_recorder_name") or None, | |
| "created_at": firestore.SERVER_TIMESTAMP, | |
| } | |
| ref = db.collection(READINGS).add(doc)[1] | |
| created.append(ref.id) | |
| audit_log(db, admin_actor, "reading.bulk_import", "reading", "bulk", { | |
| "shift_date": shift_date, "created": len(created), "skipped": skipped, | |
| }) | |
| return jsonify({"created": len(created), "skipped": skipped, "reading_ids": created}), 201 | |
| # ============================================================================= | |
| # ===== ADMIN: INSIGHTS ===== | |
| # ============================================================================= | |
| def insights_compliance(): | |
| start_date = request.args.get("start_date") | |
| end_date = request.args.get("end_date") | |
| if not start_date or not end_date: | |
| return jsonify({"error": "start_date and end_date are required."}), 400 | |
| result = insights.compute_compliance( | |
| db, start_date, end_date, | |
| unit_id=request.args.get("unit_id"), | |
| user_id=request.args.get("user_id"), | |
| ) | |
| return jsonify(result), 200 | |
| def insights_anomalies(): | |
| lookback_days = int(request.args.get("lookback_days", 7)) | |
| result = insights.compute_anomalies(db, lookback_days=lookback_days) | |
| return jsonify(result), 200 | |
| def insights_calibration(): | |
| result = insights.compute_calibration_flags(db) | |
| return jsonify(result), 200 | |
| def insights_performance(): | |
| start_date = request.args.get("start_date") | |
| end_date = request.args.get("end_date") | |
| if not start_date or not end_date: | |
| return jsonify({"error": "start_date and end_date are required."}), 400 | |
| result = insights.compute_performance(db, start_date, end_date) | |
| return jsonify(result), 200 | |
| def insights_rework(): | |
| threshold = float(request.args.get("confidence_threshold", 0.6)) | |
| result = insights.compute_rework(db, confidence_threshold=threshold) | |
| return jsonify(result), 200 | |
| def insights_digest(): | |
| today = datetime.utcnow().strftime("%Y-%m-%d") | |
| digest_ref = db.collection(DIGESTS).document(today) | |
| cached = digest_ref.get() | |
| if cached.exists: | |
| data = cached.to_dict() | |
| generated_at = data.get("generated_at") | |
| if generated_at: | |
| try: | |
| age = datetime.utcnow() - datetime.fromisoformat(str(generated_at).replace("Z", "")) | |
| except Exception: | |
| age = DIGEST_MAX_AGE + timedelta(seconds=1) | |
| else: | |
| age = DIGEST_MAX_AGE + timedelta(seconds=1) | |
| if age <= DIGEST_MAX_AGE: | |
| return jsonify(data), 200 | |
| week_ago = (datetime.utcnow() - timedelta(days=7)).strftime("%Y-%m-%d") | |
| stats = { | |
| "compliance": insights.compute_compliance(db, week_ago, today), | |
| "anomalies": insights.compute_anomalies(db, lookback_days=7), | |
| "calibration": insights.compute_calibration_flags(db), | |
| "performance": insights.compute_performance(db, week_ago, today), | |
| } | |
| try: | |
| narrative = insights.generate_digest_narrative(db, stats) | |
| except Exception as e: | |
| narrative = f"Digest narrative unavailable: {e}" | |
| payload = { | |
| "date": today, | |
| "stats": stats, | |
| "narrative": narrative, | |
| "generated_at": datetime.utcnow().isoformat(), | |
| } | |
| digest_ref.set(payload) | |
| return jsonify(payload), 200 | |
| # ============================================================================= | |
| # ===== EXPORT ===== | |
| # ============================================================================= | |
| def _units_and_readings_for_export(shift_date, scope, uid): | |
| units = [_serialize(doc) for doc in db.collection(UNITS).where("active", "==", True).stream()] | |
| query = db.collection(READINGS).where("shift_date", "==", shift_date) | |
| if scope == "mine": | |
| query = query.where("user_id", "==", uid) | |
| readings = [_serialize(doc) for doc in query.stream()] | |
| return units, readings | |
| def export_pdf(): | |
| shift_date = request.args.get("shift_date") | |
| scope = request.args.get("scope", "mine") | |
| if not shift_date: | |
| return jsonify({"error": "shift_date is required."}), 400 | |
| if scope == "all" and not _is_admin_request(): | |
| return jsonify({"error": "Only admins may export scope=all."}), 403 | |
| units, readings = _units_and_readings_for_export(shift_date, scope, request.user["uid"]) | |
| buffer = pdf_export.build_room_temperature_pdf(units, readings, shift_date) | |
| return send_file( | |
| buffer, | |
| mimetype="application/pdf", | |
| as_attachment=request.args.get("download") == "1", | |
| download_name=f"room-temperature-{shift_date}.pdf", | |
| ) | |
| def export_share(): | |
| body = request.get_json(silent=True) or {} | |
| shift_date = body.get("shift_date") | |
| scope = body.get("scope", "mine") | |
| if not shift_date: | |
| return jsonify({"error": "shift_date is required."}), 400 | |
| if not _is_admin_request(): | |
| scope = "mine" | |
| token = secrets.token_urlsafe(32) | |
| expires_at = datetime.utcnow() + SHARE_TTL | |
| db.collection(SHARES).document(token).set({ | |
| "shift_date": shift_date, | |
| "scope": scope, | |
| "user_id": request.user["uid"], | |
| "expires_at": expires_at.isoformat(), | |
| "created_at": firestore.SERVER_TIMESTAMP, | |
| }) | |
| audit_log(db, _actor(), "export.share", "share", token, {"shift_date": shift_date, "scope": scope}) | |
| return jsonify({"share_url": f"/api/export/share/{token}"}), 201 | |
| def export_share_view(token): | |
| # Intentionally NOT behind @require_auth — this is a public, unguessable-token link. | |
| share_ref = db.collection(SHARES).document(token) | |
| snap = share_ref.get() | |
| if not snap.exists: | |
| return jsonify({"error": "Share link not found."}), 404 | |
| share = snap.to_dict() | |
| expires_at = share.get("expires_at") | |
| if expires_at and datetime.utcnow() > datetime.fromisoformat(str(expires_at)): | |
| return jsonify({"error": "Share link has expired."}), 404 | |
| units, readings = _units_and_readings_for_export(share["shift_date"], share["scope"], share.get("user_id")) | |
| buffer = pdf_export.build_room_temperature_pdf(units, readings, share["shift_date"]) | |
| return send_file( | |
| buffer, | |
| mimetype="application/pdf", | |
| as_attachment=False, | |
| download_name=f"room-temperature-{share['shift_date']}.pdf", | |
| ) | |
| # ============================================================================= | |
| # ROOT | |
| # ============================================================================= | |
| def health(): | |
| return jsonify({"status": "ok", "service": "chilltrack-api"}), 200 | |
| if __name__ == "__main__": | |
| app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 7860))) | |