Spaces:
Sleeping
Sleeping
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
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")}