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) }