import json import os import uuid import random import time import base64 import requests import threading from datetime import datetime from zoneinfo import ZoneInfo from typing import Optional, List, Dict from fastapi import FastAPI, HTTPException, Request from pydantic import BaseModel app = FastAPI( title="Drivers Log-Book", docs_url=None, # This disables /docs redoc_url=None, # This disables /redoc openapi_url=None # This disables the underlying json schema ) # ------------------------- # Get Date-Time UTILS # ------------------------- def date_time_now(): tz = ZoneInfo("Asia/Karachi") return datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S") # ------------------------- # SITE SETTINGS # ------------------------- site_settings = { "site_name": "DriveLedger", "tagline": "Smart vehicle Logging System.", "photo_required_start": True, "photo_required_point": True, "photo_required_end": True, "force_camera_only": True, "allow_gallery_upload": True, "allow_delete_session": True, "allow_delete_fuel": True, "allow_delete_user": True, "allow_ai": False, "logs": [], "available_companies": [ "snip" ], "img_host": "https://whynotasif-telecloud-cdn.hf.space/upload", "img_viewer": "https://whynotasif-telecloud-cdn.hf.space/view/" } # ------------------------- # JSON LOAD # ------------------------- users = {} sessions = {} fuel_record = {} data_lock = threading.Lock() # ------------------------- # IMAGE UPLOAD (TeleCloud CDN) # ------------------------- TELE_CLOUD_URL = "https://whynotasif-telecloud-cdn.hf.space" def upload_pict(base64_img: str, username: str = "driveledger", caption_type: str = "default"): """ Uploads image to TeleCloud CDN (Telegram backend). Returns full JSON object with file_id for display via CDN. caption_type options: - "profile" - New user profile picture - "profile_changed" - Profile picture changed - "session_start" - Session start photo - "session_end" - Session end photo - "point" - Added point photo - "default" - Generic upload """ # Convert base64 to bytes img_bytes = base64.b64decode(base64_img) # Build caption based on type caption_map = { "profile": f"Username: {username}\nType: User Profile\nTime: {date_time_now()}", "profile_changed": f"Username: {username}\nType: Profile Changed\nTime: {date_time_now()}", "session_start": f"Username: {username}\nType: Session Start\nTime: {date_time_now()}", "session_end": f"Username: {username}\nType: Session End\nTime: {date_time_now()}", "point": f"Username: {username}\nType: Added Point\nTime: {date_time_now()}", "default": f"Username: {username}\nType: DriveLedger Upload\nTime: {date_time_now()}" } caption = caption_map.get(caption_type, caption_map["default"]) url = f"{TELE_CLOUD_URL}/upload" files = {"file": ("image.jpg", img_bytes, "image/jpeg")} data = { "username": username, "caption": caption } try: response = requests.post(url, files=files, data=data, timeout=30) if response.status_code != 200: raise HTTPException(500, f"Image upload failed: {response.status_code}") result = response.json() # Return the full JSON object (will be stored in database) # Frontend will use: https://whynotasif-telecloud-cdn.hf.space/view/{file_id} return result except requests.exceptions.RequestException as e: raise HTTPException(500, f"Upload error: {str(e)}") def get_image_url(photo_data): """ Converts stored photo JSON object to display URL. photo_data can be: - New format: {chat_id, file_id, message_id, date_time} - Old format: direct URL string (for backward compatibility) Returns: URL string for display """ if not photo_data: return None # If it's a string (old ImgBB format), return as-is if isinstance(photo_data, str): return photo_data # If it's a dict with file_id (new TeleCloud format) if isinstance(photo_data, dict) and photo_data.get("file_id"): return f"{TELE_CLOUD_URL}/view/{photo_data['file_id']}" return None # ------------------------- # HELPERS # ------------------------- def user_has_active_session(username): for sid, s in sessions.items(): if s["username"] == username and s["ongoing"] == True: return sid return None def get_user_company(username: str) -> Optional[str]: """Get company of a user""" if username in users: return users[username].get("company") return None def can_access_all(request_by: str) -> bool: """Check if user can access all data (super owner with company='all')""" if request_by in users: user_company = users[request_by].get("company", "") return users[request_by].get("is_owner", False) and user_company == "all" return False def get_requester_company(request_by: str) -> Optional[str]: """Get the company to filter by for the requester""" if can_access_all(request_by): return None # Super owner can see all return get_user_company(request_by) def filter_sessions_by_company(request_by: str, target_company: str = None) -> Dict: """Filter sessions by company for admin/owner view""" result = {} # Only allow if requester is admin/owner if request_by not in users or not (users[request_by].get("is_admin") or users[request_by].get("is_owner")): raise HTTPException(403, "Not authorized") # If target_company is specified (owner viewing specific company), use that # Super owner (company="all") can see all if can_access_all(request_by): return sessions requester_company = get_user_company(request_by) filter_company = target_company if target_company else requester_company # If no company assigned, return empty if not filter_company: return result for sid, session in sessions.items(): session_company = get_user_company(session["username"]) if session_company == filter_company: result[sid] = session return result def filter_users_by_company(request_by: str, target_company: str = None) -> Dict: """Filter users by company for admin/owner view""" result = {} # Only allow if requester is admin/owner if request_by not in users or not (users[request_by].get("is_admin") or users[request_by].get("is_owner")): raise HTTPException(403, "Not authorized") # Super owner (company="all") can see all if can_access_all(request_by): return users requester_company = get_user_company(request_by) # If target_company is specified (owner viewing specific company), use that filter_company = target_company if target_company else requester_company # If no company assigned, return empty if not filter_company: return result for username, user in users.items(): user_company = user.get("company") if user_company == filter_company: result[username] = user return result def filter_fuel_by_company(request_by: str, target_company: str = None) -> Dict: """Filter fuel records by company for admin/owner view""" result = {} # Only allow if requester is admin/owner if request_by not in users or not (users[request_by].get("is_admin") or users[request_by].get("is_owner")): raise HTTPException(403, "Not authorized") # Super owner (company="all") can see all if can_access_all(request_by): return fuel_record requester_company = get_user_company(request_by) # If target_company is specified (owner viewing specific company), use that filter_company = target_company if target_company else requester_company # If no company assigned, return empty if not filter_company: return result for username, fuel_records in fuel_record.items(): user_company = get_user_company(username) if user_company == filter_company: result[username] = fuel_records return result def is_owner(request_by: str) -> bool: """Check if user is owner""" if request_by in users: return users[request_by].get("is_owner", False) return False def can_delete_sessions(request_by: str) -> bool: """Check if user can delete sessions based on settings""" if is_owner(request_by): return True return site_settings.get("allow_delete_session", False) def can_delete_fuel(request_by: str) -> bool: """Check if user can delete fuel based on settings""" if is_owner(request_by): return True return site_settings.get("allow_delete_fuel", False) def can_delete_user(request_by: str) -> bool: """Check if user can delete users based on settings""" if is_owner(request_by): return True return site_settings.get("allow_delete_user", False) def add_log(request_by: str, action: str, details: str = ""): """Add an action log entry""" log_entry = { "time": date_time_now(), "by": request_by, "action": action, "details": details } site_settings["logs"].append(log_entry) # Keep only last 1000 logs if len(site_settings["logs"]) > 1000: site_settings["logs"] = site_settings["logs"][-1000:] def can_delete_fuel(request_by: str) -> bool: """Check if user can delete fuel based on settings""" if is_owner(request_by): return True return site_settings.get("allow_delete_fuel", False) def can_delete_user(request_by: str) -> bool: """Check if user can delete users based on settings""" if is_owner(request_by): return True return site_settings.get("allow_delete_user", False) # ------------------------- # MODELS # ------------------------- class ChangePictureRequest(BaseModel): username: str picture_base64: str class UserCreate(BaseModel): username: str password: str profile_picture: Optional[str] vehicle_number: Optional[str] ip_address: str cnic_no: str gender: str company: str full_name: str email_address: str contact_number: str country_name: str state_region_province: str district_name: str tehsil_taluka_town: str city_name: str union_council: str full_address: str class SessionCreate(BaseModel): username: str start_km: int start_photo: Optional[str] lat: float lng: float remarks: Optional[str] class PointCreate(BaseModel): session_id: str village: str km: int photo: Optional[str] lat: float lng: float remarks: Optional[str] class EndSession(BaseModel): session_id: str end_km: int photo: Optional[str] lat: float lng: float remarks: Optional[str] class EnergyRecord(BaseModel): fuel_type: str # #Petrol/Diesel #default Petrol km: int liters: Optional[float] = None price_per_liter: Optional[float] = None total_cost: Optional[float] = None station: Optional[str] = None photo: Optional[str] = None lat: Optional[float] = None lng: Optional[float] = None remarks: Optional[str] @app.get("/") async def root(): return {"status": True, "service": "Drivers Log-Book is Active"} # ------------------------- # USER ENDPOINTS (7) # ------------------------- @app.post("/users/add") def add_user(data: UserCreate): if data.username in users: raise HTTPException(400, "User exists") # Validate company against available_companies allowed_companies = site_settings.get("available_companies", []) if data.company and data.company not in allowed_companies: raise HTTPException(400, "Company not authorized") img_url = None if data.profile_picture: img_url = upload_pict(data.profile_picture, data.username, "profile") users[data.username] = { "username": data.username, "password": data.password, "vehicle_number": data.vehicle_number, "profile_picture": img_url, "ip_address": data.ip_address, "cnic_no": data.cnic_no, "gender": data.gender, "full_name": data.full_name, "email_address": data.email_address, "contact_number": data.contact_number, "country_name": data.country_name, "state_region_province": data.state_region_province, "district_name": data.district_name, "tehsil_taluka_town": data.tehsil_taluka_town, "city_name": data.city_name, "union_council": data.union_council, "full_address": data.full_address, "is_admin": False, "is_owner": False, "company": data.company, "joining_time": date_time_now() } return {"msg": "user added"} @app.delete("/users/delete") def delete_user(username: str, backup: bool = True, request_by: str = ""): if username not in users: raise HTTPException(404, "User not found") # Check permission - only owner can delete users if request_by and not can_delete_user(request_by): raise HTTPException(403, "Not authorized to delete users") # If backup is True, prepare data for download before deleting if backup: dr_user = users.get(username, {}) dr_sessions = sessions.get(username, {}) dr_fuel = fuel_record.get(username, {}) data = { "user": dr_user, "sessions": dr_sessions, "fuel_record": dr_fuel } # Delete user data del users[username] if username in sessions: del sessions[username] if username in fuel_record: del fuel_record[username] # Log the action if request_by: add_log(request_by, "delete_user", f"Deleted user {username} with backup") # Return the data for download return { "msg": "deleted with backup", "backup_data": data } else: # Direct delete without backup del users[username] if username in sessions: del sessions[username] if username in fuel_record: del fuel_record[username] # Log the action if request_by: add_log(request_by, "delete_user", f"Deleted user {username} permanently (no backup)") return {"msg": "deleted"} @app.get("/users/download") def download_user_data(username: str): """Fetch and download user data (sessions and fuel records)""" if username not in users: raise HTTPException(404, "User not found") dr_user = users.get(username, {}) dr_sessions = sessions.get(username, {}) dr_fuel = fuel_record.get(username, {}) data = { "user": dr_user, "sessions": dr_sessions, "fuel_record": dr_fuel } return data @app.post("/users/change_password") def change_password(username: str, new_password: str): users[username]["password"] = new_password return {"msg": "updated"} @app.post("/users/change_picture") def change_picture(data: ChangePictureRequest): """Change user profile picture - uses BaseModel to avoid long URL error""" if data.username not in users: raise HTTPException(404, "User not found") img = upload_pict(data.picture_base64, data.username, "profile_changed") users[data.username]["profile_picture"] = img return {"msg": "updated"} @app.post("/users/change_location") def change_location( username: str, state_region_province: Optional[str] = None, district_name: Optional[str] = None, tehsil_taluka_town: Optional[str] = None, city_name: Optional[str] = None, union_council: Optional[str] = None ): """ Update user location details. All fields are optional - only update fields that are provided. If a field is not provided, the existing value will be kept. """ if username not in users: raise HTTPException(status_code=404, detail="User not found") # Only update fields that are provided (not None) if state_region_province is not None: users[username]["state_region_province"] = state_region_province if district_name is not None: users[username]["district_name"] = district_name if tehsil_taluka_town is not None: users[username]["tehsil_taluka_town"] = tehsil_taluka_town if city_name is not None: users[username]["city_name"] = city_name if union_council is not None: users[username]["union_council"] = union_council return {"msg": "updated"} @app.get("/users/get") def get_user(username: str, request_by: str = ""): """Get a specific user - filtered by company""" # Allow users to get their own profile without auth if request_by and request_by == username: return users.get(username) # For admin/owner, filter by company if request_by and request_by in users and (users[request_by].get("is_admin") or users[request_by].get("is_owner")): target_user = users.get(username) if not target_user: raise HTTPException(404, "User not found") # Super owner can see all if can_access_all(request_by): return target_user # Check if target user is in same company requester_company = get_user_company(request_by) target_company = target_user.get("company") if requester_company and target_company == requester_company: return target_user raise HTTPException(403, "Not authorized to view this user") # Default: return user if exists (for public/own profile) return users.get(username) @app.get("/users/all") def get_all_users(request_by: str = ""): """Get all users - filtered by company for admin/owner""" # For admin/owner, filter by company if request_by and request_by in users and (users[request_by].get("is_admin") or users[request_by].get("is_owner")): return filter_users_by_company(request_by) # Default: return empty or all users return {} @app.post("/users/set_admin") def set_admin(username: str, value: bool, request_by: str): """Set user as admin - only owner can do this (filtered by company)""" if not is_owner(request_by): raise HTTPException(403, "Only owners can set admins") # Check company permission target_company = get_user_company(username) requester_company = get_user_company(request_by) # Super owner can set admin for any; others must match company if not can_access_all(request_by): if target_company != requester_company: raise HTTPException(403, "Not authorized to set admin for this user") users[username]["is_admin"] = value # Log the action add_log(request_by, "set_admin", f"Set {username} as admin: {value}") return {"msg": "admin updated"} @app.post("/users/set_owner") def set_owner(username: str, value: bool): """Set user as owner - can only be done directly""" users[username]["is_owner"] = value return {"msg": "owner updated"} @app.post("/users/update") def update_user(username: str, field: str, value, request_by: str): """Generic admin update for arbitrary user fields - only owner can update other users (filtered by company)""" if username not in users: raise HTTPException(404, "User not found") # Only owner can update other users if not is_owner(request_by): raise HTTPException(403, "Only owners can update users") # Check company permission target_company = get_user_company(username) requester_company = get_user_company(request_by) # Super owner can update any; others must match company if not can_access_all(request_by): if target_company != requester_company: raise HTTPException(403, "Not authorized to update this user") users[username][field] = value # Log the action add_log(request_by, "update_user", f"Updated {field} to '{value}' for user {username}") return {"msg": "updated"} # ------------------------- # SESSION ENDPOINTS (6) # ------------------------- @app.get("/session/get") def get_session(session_id: str, request_by: str = ""): """Get a specific session - filtered by company""" session = sessions.get(session_id) if not session: raise HTTPException(404, "Session not found") # Allow if user owns the session if request_by and session.get("username") == request_by: return session # For admin/owner, filter by company if request_by and request_by in users and (users[request_by].get("is_admin") or users[request_by].get("is_owner")): # Super owner can see all if can_access_all(request_by): return session # Check if session owner is in same company session_owner_company = get_user_company(session.get("username")) requester_company = get_user_company(request_by) if requester_company and session_owner_company == requester_company: return session raise HTTPException(403, "Not authorized to view this session") # Default: no access raise HTTPException(403, "Not authorized") @app.get("/session/all") def get_sessions(request_by: str = ""): """Get all sessions - filtered by company for admin/owner""" # For admin/owner, filter by company if request_by and request_by in users and (users[request_by].get("is_admin") or users[request_by].get("is_owner")): return filter_sessions_by_company(request_by) # Default: return empty return {} @app.post("/session/new") def new_session(data: SessionCreate): if data.username not in users: raise HTTPException(404, "User not found") active = user_has_active_session(data.username) if active: raise HTTPException(400, f"Active session exists: {active}") sid = str(uuid.uuid4()) photo = None if data.start_photo: photo = upload_pict(data.start_photo, data.username, "session_start") sessions[sid] = { "ongoing": True, "username": data.username, "start_km": data.start_km, "end_km": None, "start_photo": photo, "end_photo": None, "start_location": { "lat": data.lat, "lng": data.lng }, "end_location": None, "created_at": date_time_now(), "ended_at": None, "total_km": None, "points": [], "remarks": data.remarks } # Log the action add_log(data.username, "session_start", f"Started new session {sid} with {data.start_km}km") return {"session_id": sid} @app.post("/session/add_point") def add_point(data: PointCreate): if data.session_id not in sessions: raise HTTPException(404, "Session not found") s = sessions[data.session_id] # prevent km going backwards relative to start or last point last_km = s["start_km"] if s["points"]: last_km = s["points"][-1]["km"] if data.km < last_km: raise HTTPException(400, "Kilometers cannot be less than previous reading") img = None if data.photo: img = upload_pict(data.photo, s["username"], "point") point = { "village": data.village, "km": data.km, "photo": img, "location": { "lat": data.lat, "lng": data.lng }, "time": date_time_now(), "remarks": data.remarks } s["points"].append(point) # Log the action add_log(s["username"], "session_add_point", f"Added checkpoint at {data.village} ({data.km}km) to session {data.session_id}") return {"msg": "point added"} @app.post("/session/end") def end_session(data: EndSession): if data.session_id not in sessions: raise HTTPException(404, "Session not found") s = sessions[data.session_id] # verify end_km is not less than last point km last_km = s["start_km"] if s["points"]: last_km = s["points"][-1]["km"] if data.end_km < last_km: raise HTTPException(400, "End kilometers cannot be less than last recorded point") total = data.end_km - s["start_km"] if total < 0: raise HTTPException(400, "Negative KM") photo = None if data.photo: photo = upload_pict(data.photo, s["username"], "session_end") s["end_km"] = data.end_km s["end_photo"] = photo s["end_location"] = { "lat": data.lat, "lng": data.lng } s["remarks"] = data.remarks s["ended_at"] = date_time_now() s["total_km"] = total s["ongoing"] = False # Log the action add_log(s["username"], "session_end", f"Ended session {data.session_id} at {data.end_km}km ({total}km total)") return {"total_km": total} #admin-panel from owner - only owner can update sessions (filtered by company) @app.post("/session/update") def update_session(session_id: str, field: str, value, request_by: str): if session_id not in sessions: raise HTTPException(404, "Session not found") # Only owner can update sessions if not is_owner(request_by): raise HTTPException(403, "Only owners can update sessions") # Check company permission session_username = sessions[session_id].get("username") session_company = get_user_company(session_username) requester_company = get_user_company(request_by) # Super owner can update any; others must match company if not can_access_all(request_by): if session_company != requester_company: raise HTTPException(403, "Not authorized to update this session") sessions[session_id][field] = value # Log the action add_log(request_by, "update_session", f"Updated {field} for session {session_id}") return {"msg": "updated"} # administrative helpers for sessions/points - owner only (filtered by company) @app.delete("/session/delete") def delete_session(session_id: str, request_by: str): if session_id not in sessions: raise HTTPException(404, "Session not found") # Only owner can delete sessions if not can_delete_sessions(request_by): raise HTTPException(403, "Not authorized to delete sessions") # Check company permission session_username = sessions[session_id].get("username") session_company = get_user_company(session_username) requester_company = get_user_company(request_by) # Super owner can delete any; others must match company if not can_access_all(request_by): if session_company != requester_company: raise HTTPException(403, "Not authorized to delete this session") # Get session info for logging del sessions[session_id] # Log the action add_log(request_by, "delete_session", f"Deleted session {session_id} for user {session_username}") return {"msg": "deleted"} @app.post("/session/point/update") def update_point(session_id: str, index: int, field: str, value, request_by: str): if session_id not in sessions: raise HTTPException(404, "Session not found") # Only owner can update points if not is_owner(request_by): raise HTTPException(403, "Only owners can update points") # Check company permission session_username = sessions[session_id].get("username") session_company = get_user_company(session_username) requester_company = get_user_company(request_by) # Super owner can update any; others must match company if not can_access_all(request_by): if session_company != requester_company: raise HTTPException(403, "Not authorized to update this session") pts = sessions[session_id].get("points", []) if index < 0 or index >= len(pts): raise HTTPException(404, "Point not found") pts[index][field] = value # Log the action add_log(request_by, "update_point", f"Updated point {index} field '{field}' in session {session_id}") return {"msg": "updated"} @app.delete("/session/point/delete") def delete_point(session_id: str, index: int, request_by: str): if session_id not in sessions: raise HTTPException(404, "Session not found") # Only owner can delete points if not is_owner(request_by): raise HTTPException(403, "Only owners can delete points") # Check company permission session_username = sessions[session_id].get("username") session_company = get_user_company(session_username) requester_company = get_user_company(request_by) # Super owner can delete any; others must match company if not can_access_all(request_by): if session_company != requester_company: raise HTTPException(403, "Not authorized to delete this session") pts = sessions[session_id].get("points", []) if index < 0 or index >= len(pts): raise HTTPException(404, "Point not found") pts.pop(index) # Log the action add_log(request_by, "delete_point", f"Deleted point {index} from session {session_id}") return {"msg": "deleted"} @app.get("/session/user") def get_all_sessions_of_user(username: str, request_by: str = ""): """Get all sessions of a user - filtered by company""" if username not in users: raise HTTPException(404, "User not found") # Allow if user owns the sessions if request_by and request_by == username: result = {} for sid, session in sessions.items(): if session["username"] == username: result[sid] = session return result # For admin/owner, filter by company if request_by and request_by in users and (users[request_by].get("is_admin") or users[request_by].get("is_owner")): # Super owner can see all if can_access_all(request_by): result = {} for sid, session in sessions.items(): if session["username"] == username: result[sid] = session return result # Check if target user is in same company target_company = get_user_company(username) requester_company = get_user_company(request_by) if requester_company and target_company == requester_company: result = {} for sid, session in sessions.items(): if session["username"] == username: result[sid] = session return result raise HTTPException(403, "Not authorized to view this user's sessions") # Default: no access raise HTTPException(403, "Not authorized") # ------------------------- # FUEL / ENERGY RECORDS # ------------------------- @app.post("/fuel/add") def add_fuel(username: str, record: EnergyRecord, timestamp: Optional[str] = None): if username not in users: raise HTTPException(404, "User not found") if username not in fuel_record: fuel_record[username] = {} if not timestamp: timestamp = date_time_now() rec = record.dict() # Upload photo to Telegram if present if rec.get("photo"): rec["photo"] = upload_pict(rec["photo"], username, "fuel_record") # auto-calc total cost if needed if rec.get("liters") is not None and rec.get("price_per_liter") is not None and rec.get("total_cost") is None: rec["total_cost"] = rec["liters"] * rec["price_per_liter"] fuel_record[username][timestamp] = rec # Log the action add_log(username, "fuel_add", f"Added fuel record: {rec.get('liters', 0)}L at {rec.get('station', 'Unknown')} (${rec.get('total_cost', 0)})") return {"timestamp": timestamp} @app.get("/fuel/get") def get_fuel(username: Optional[str] = None, request_by: str = ""): """Return fuel records - filtered by company""" # If username provided, return that user's fuel if username: # Allow if user owns the fuel records if request_by and request_by == username: return fuel_record.get(username, {}) # For admin/owner, filter by company if request_by and request_by in users and (users[request_by].get("is_admin") or users[request_by].get("is_owner")): # Super owner can see all if can_access_all(request_by): return fuel_record.get(username, {}) # Check if target user is in same company target_company = get_user_company(username) requester_company = get_user_company(request_by) if requester_company and target_company == requester_company: return fuel_record.get(username, {}) raise HTTPException(403, "Not authorized to view this user's fuel records") # Default: no access raise HTTPException(403, "Not authorized") # No username - return all (filtered by company for admin/owner) if request_by and request_by in users and (users[request_by].get("is_admin") or users[request_by].get("is_owner")): return filter_fuel_by_company(request_by) return {} # mirror the session/user endpoint for fuel records @app.get("/fuel/user") def get_fuel_of_user(username: str, request_by: str = ""): """Return fuel records for a single user - filtered by company""" if username not in users: raise HTTPException(404, "User not found") # Allow if user owns the fuel records if request_by and request_by == username: return fuel_record.get(username, {}) # For admin/owner, filter by company if request_by and request_by in users and (users[request_by].get("is_admin") or users[request_by].get("is_owner")): # Super owner can see all if can_access_all(request_by): return fuel_record.get(username, {}) # Check if target user is in same company target_company = get_user_company(username) requester_company = get_user_company(request_by) if requester_company and target_company == requester_company: return fuel_record.get(username, {}) raise HTTPException(403, "Not authorized to view this user's fuel records") # Default: no access raise HTTPException(403, "Not authorized") @app.post("/fuel/update") def update_fuel(username: str, timestamp: str, field: str, value, request_by: str): if username not in fuel_record or timestamp not in fuel_record[username]: raise HTTPException(404, "Record not found") # Only owner can update fuel if not is_owner(request_by): raise HTTPException(403, "Only owners can update fuel records") # Check company permission target_company = get_user_company(username) requester_company = get_user_company(request_by) # Super owner can update any; others must match company if not can_access_all(request_by): if target_company != requester_company: raise HTTPException(403, "Not authorized to update this fuel record") fuel_record[username][timestamp][field] = value # Log the action add_log(request_by, "update_fuel", f"Updated {field} for {username}'s fuel record at {timestamp}") return {"msg": "updated"} @app.delete("/fuel/delete") def delete_fuel(username: str, timestamp: str, request_by: str): if username not in fuel_record or timestamp not in fuel_record[username]: raise HTTPException(404, "Record not found") # Only owner can delete fuel based on settings if not can_delete_fuel(request_by): raise HTTPException(403, "Not authorized to delete fuel records") # Check company permission target_company = get_user_company(username) requester_company = get_user_company(request_by) # Super owner can delete any; others must match company if not can_access_all(request_by): if target_company != requester_company: raise HTTPException(403, "Not authorized to delete this fuel record") del fuel_record[username][timestamp] # Log the action add_log(request_by, "delete_fuel", f"Deleted fuel record for {username} at {timestamp}") return {"msg": "deleted"} # ------------------------- # SETTINGS ENDPOINTS (2) # ------------------------- @app.get("/settings") def get_settings(): return site_settings @app.post("/settings/update") def update_setting(key: str, value, request_by: str): """Update site settings - only owner can do this""" if not is_owner(request_by): raise HTTPException(403, "Only owners can update settings") site_settings[key] = value # Log the action add_log(request_by, "settings_update", f"Updated setting '{key}' to '{value}'") return {"msg": "updated"} # ------------------------- # COMPANY FILTERED ENDPOINTS (for admin panel) # ------------------------- @app.get("/admin/sessions") def get_admin_sessions(request_by: str): """Get all sessions filtered by requester's company - for admin panel""" return filter_sessions_by_company(request_by) @app.get("/admin/users") def get_admin_users(request_by: str): """Get all users filtered by requester's company - for admin panel""" return filter_users_by_company(request_by) @app.get("/admin/fuel") def get_admin_fuel(request_by: str): """Get all fuel records filtered by requester's company - for admin panel""" return filter_fuel_by_company(request_by) # ------------------------- # COMPANY MANAGEMENT ENDPOINTS # ------------------------- @app.get("/companies") def get_companies(request_by: str): """Get all available companies - only owner can view""" if not is_owner(request_by): raise HTTPException(403, "Only owners can view companies") return site_settings.get("available_companies", []) @app.post("/companies/add") def add_company(company_name: str, request_by: str): """Add a new company to available_companies - only owner can do this""" if not is_owner(request_by): raise HTTPException(403, "Only owners can add companies") companies = site_settings.get("available_companies", []) if company_name in companies: raise HTTPException(400, "Company already exists") companies.append(company_name) site_settings["available_companies"] = companies # Log the action add_log(request_by, "add_company", f"Added company: {company_name}") return {"msg": "company added", "companies": companies} @app.delete("/companies") def delete_company(company_name: str, request_by: str): """Delete a company from available_companies - only owner can do this""" if not is_owner(request_by): raise HTTPException(403, "Only owners can delete companies") companies = site_settings.get("available_companies", []) if company_name not in companies: raise HTTPException(404, "Company not found") companies.remove(company_name) site_settings["available_companies"] = companies # Log the action add_log(request_by, "delete_company", f"Deleted company: {company_name}") return {"msg": "company deleted"} @app.delete("/companies/clear") def clear_companies(request_by: str): """Clear all companies from available_companies - only owner can do this""" if not is_owner(request_by): raise HTTPException(403, "Only owners can clear companies") # Log before clearing companies = site_settings.get("available_companies", []) add_log(request_by, "clear_companies", f"Cleared all companies: {companies}") site_settings["available_companies"] = [] return {"msg": "companies cleared"} # ------------------------- # AI CHAT ENDPOINT (for visitors) # ------------------------- @app.post("/ai/chat") def chat_with_ai(message: str): """Chat with AI - limit is handled on frontend via localStorage""" # Check if AI is allowed if not site_settings.get("allow_ai", False): raise HTTPException(403, "AI chat is not available") # Simple AI response - in production, you would integrate with an actual AI API # For now, return a helpful response based on the message response = generate_ai_response(message) return { "response": response } def generate_ai_response(message: str) -> str: """Generate a simple AI response based on the message""" message_lower = message.lower() # Greetings if any(word in message_lower for word in ["hello", "hi", "hey", "greetings"]): return "Hello! I'm here to help you with DriveLedger. You can ask me about starting a session, recording fuel, or using the features. How can I assist you today?" # Help with sessions if "session" in message_lower or "start" in message_lower or "drive" in message_lower: return "To start a driving session: 1) Login to your account 2) Go to 'Start Session' 3) Take a photo of your odometer 4) Enter the starting KM 5) Click Start. The system will track your trip with GPS and photos." # Help with fuel if "fuel" in message_lower or "petrol" in message_lower or "diesel" in message_lower: return "To record fuel: Go to 'Fuel Records' in your dashboard. Enter the fuel type, liters, price per liter, and optionally upload a receipt photo. The system will calculate the total cost automatically." # Help with login/signup if "login" in message_lower or "sign up" in message_lower or "register" in message_lower: return "To get started: Click 'Sign Up' on the home page. Fill in your details including your company name, vehicle number, and location. You'll need a profile picture and your CNIC number." # Help with admin if "admin" in message_lower or "manage" in message_lower: return "Admin features include: viewing all users and sessions, fuel records management, and data export. Contact your administrator for access." # Help with export if "export" in message_lower or "download" in message_lower: return "To export your data: Go to 'Export Data' in your dashboard. You can download your session history and fuel records in various formats." # Default response return "I'm here to help with DriveLedger! Ask me about:\n- Starting a driving session\n- Recording fuel entries\n- Login and registration\n- Admin features\n- Data export\n\nWhat would you like to know?" # ------------------------- # LOGS ENDPOINTS # ------------------------- @app.get("/logs") def get_logs(request_by: str): """Get all logs - only owner can view""" if not is_owner(request_by): raise HTTPException(403, "Only owners can view logs") return site_settings.get("logs", []) @app.delete("/logs/clear") def clear_logs(request_by: str): """Clear all logs - only owner can do this""" if not is_owner(request_by): raise HTTPException(403, "Only owners can clear logs") site_settings["logs"] = [] add_log(request_by, "clear_logs", "All logs cleared") return {"msg": "logs cleared"} # ------------------------- # REMOTE SYNC HELPERS # ------------------------- def fetch_update_json(method: str = "fetch"): """Synchronize *users*, *sessions* and *site_settings* with the VPS. In ``fetch`` mode we pull from the remote host and replace the three globals. In ``update`` mode we POST the current globals so the VPS can store them. The same helper is exposed via ``/sync`` below and may also be invoked periodically by a background thread. The base URL of the remote service is read from the same GitHub redirect file that the frontend uses. """ # read redirect file for base URL rb = requests.get( "https://raw.githubusercontent.com/WhyNotAsif/redirects/refs/heads/main/smartort_fastapi_baseurl.txt" ) if rb.status_code != 200: raise HTTPException(500, "could not read base-url file") base_url_info = rb.text.strip() #this will fetch json file from vps_main.py backup if method == "fetch": # use POST so VPS (which only exposes POST) doesn't reject the request rs = requests.post(f"{base_url_info}/driverledge/sync?method=fetch") if rs.status_code != 200: # propagate error, caller may choose to ignore raise HTTPException(500, "remote fetch failed") data = rs.json() with data_lock: # update existing dicts rather than rebinding locals if "site_settings" in data: site_settings.clear() site_settings.update(data["site_settings"]) if "users" in data: users.clear() users.update(data["users"]) if "sessions" in data: sessions.clear() sessions.update(data["sessions"]) if "fuel_record" in data: fuel_record.clear() fuel_record.update(data["fuel_record"]) return {"msg": "json files updated with fetched settings"} #this will update driveledger.json by vps_main.py on vps elif method == "update": payload = {"site_settings": site_settings, "users": users, "sessions": sessions, "fuel_record": fuel_record} rs = requests.post(f"{base_url_info}/driverledge/sync?method=update", json=payload) if rs.status_code != 200: raise HTTPException(500, "remote update failed") return rs.json() else: raise HTTPException(400, "invalid method") @app.post("/sync") def sync_json(method: str = "fetch"): """Endpoint to trigger a fetch or update of the three data blobs. ``method`` may be ``fetch`` or ``update``; the return value will contain ``site_settings``, ``users`` and ``sessions`` regardless of direction. """ return fetch_update_json(method) # perform initial fetch but don't crash if it fails (e.g. network unavailable) try: fetch_update_json("fetch") except Exception as e: print("initial fetch failed, continuing without remote data:", e) # automatically push to the VPS once every hour in background @app.on_event("startup") def start_backup_thread(): threading.Thread(target=schedule_hourly_sync, daemon=True).start() def schedule_hourly_sync(): while True: try: fetch_update_json("update") except Exception as e: print("hourly sync failed:", e) time.sleep(3600) #every hour