Spaces:
Sleeping
Sleeping
File size: 4,408 Bytes
e62c239 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 |
import sqlite3
import os
import datetime
from typing import List, Dict, Any, Optional, Tuple
class CarSharingDB:
def __init__(self, db_path="car_sharing.db"):
"""Initialize the car sharing database."""
self.db_path = db_path
self.initialize_db()
def initialize_db(self):
"""Create the database tables if they don't exist."""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# Create trips table
cursor.execute('''
CREATE TABLE IF NOT EXISTS trips (
id INTEGER PRIMARY KEY,
user_name TEXT NOT NULL,
start_km INTEGER NOT NULL,
end_km INTEGER NOT NULL,
km_traveled INTEGER NOT NULL,
trip_date TEXT NOT NULL
)
''')
conn.commit()
conn.close()
def record_trip(self, user_name: str, start_km: int, end_km: int) -> bool:
"""Record a new trip in the database."""
try:
km_traveled = end_km - start_km
if km_traveled <= 0:
return False
trip_date = datetime.datetime.now().strftime("%Y-%m-%d")
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute(
"INSERT INTO trips (user_name, start_km, end_km, km_traveled, trip_date) VALUES (?, ?, ?, ?, ?)",
(user_name, start_km, end_km, km_traveled, trip_date)
)
conn.commit()
conn.close()
return True
except Exception as e:
print(f"Error recording trip: {str(e)}")
return False
def get_monthly_stats(self, user_name: str, month: Optional[int] = None, year: Optional[int] = None) -> Dict[str, Any]:
"""Get monthly statistics for a user."""
try:
if month is None or year is None:
# Use current month and year if not specified
now = datetime.datetime.now()
month = now.month
year = now.year
# Format date range for the specified month
start_date = f"{year}-{month:02d}-01"
# Calculate the last day of the month
if month == 12:
next_month_year = year + 1
next_month = 1
else:
next_month_year = year
next_month = month + 1
end_date = f"{next_month_year}-{next_month:02d}-01"
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# Get total kilometers and trip count
cursor.execute(
"SELECT SUM(km_traveled), COUNT(*) FROM trips WHERE user_name = ? AND trip_date >= ? AND trip_date < ?",
(user_name, start_date, end_date)
)
total_km, trip_count = cursor.fetchone()
# Get all trips in the month
cursor.execute(
"SELECT start_km, end_km, km_traveled, trip_date FROM trips WHERE user_name = ? AND trip_date >= ? AND trip_date < ? ORDER BY trip_date",
(user_name, start_date, end_date)
)
trips = []
for row in cursor.fetchall():
trips.append({
"start_km": row[0],
"end_km": row[1],
"km_traveled": row[2],
"trip_date": row[3]
})
conn.close()
# Default to 0 if no trips found
total_km = total_km or 0
trip_count = trip_count or 0
return {
"user_name": user_name,
"month": month,
"year": year,
"total_km": total_km,
"trip_count": trip_count,
"trips": trips
}
except Exception as e:
print(f"Error getting monthly stats: {str(e)}")
return {
"user_name": user_name,
"month": month,
"year": year,
"total_km": 0,
"trip_count": 0,
"trips": [],
"error": str(e)
} |