import sqlite3 import json import uuid import os from fastapi import FastAPI, Request, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles app = FastAPI(title="AttendAI Pro API", version="2.0.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) DB_FILE = "database.sqlite" def get_db(): conn = sqlite3.connect(DB_FILE, check_same_thread=False) conn.row_factory = sqlite3.Row return conn def init_db(): conn = get_db() c = conn.cursor() # Create tables exactly as specified in the README c.execute('''CREATE TABLE IF NOT EXISTS users ( id text PRIMARY KEY, email text, name text, role text, roll_number text, department text, institution text, is_active boolean, face_data text, last_login datetime, password text )''') # FORCE INIT: Ensure the student exists and has the correct password on every start email = 'ugs23020_csm.gowthami@cbit.org.in' c.execute("SELECT id FROM users WHERE LOWER(email) = ?", (email,)) if not c.fetchone(): c.execute("INSERT INTO users (id, email, name, role, roll_number, is_active, password) VALUES (?, ?, ?, ?, ?, ?, ?)", (str(uuid.uuid4()), email, 'Gowthami', 'student', 'UGS23020', True, 'CBIT2024')) print(f"DEBUG: Created new student account for {email}") else: c.execute("UPDATE users SET password = 'CBIT2024' WHERE LOWER(email) = ?", (email,)) print(f"DEBUG: Updated password for existing account {email}") conn.commit() c.execute('''CREATE TABLE IF NOT EXISTS sessions ( id text PRIMARY KEY, teacher_id text, subject text, qr_code text, qr_expires_at datetime, geofence_lat real, geofence_lng real, geofence_radius real, is_active boolean )''') c.execute('''CREATE TABLE IF NOT EXISTS attendance ( id text PRIMARY KEY, session_id text, student_id text, roll_number text, status text, marked_at datetime, gps_lat real, gps_lng real, face_match_score real, device_fingerprint text, fraud_flags text, verified boolean )''') c.execute('''CREATE TABLE IF NOT EXISTS fraud_alerts ( id text PRIMARY KEY, user_id text, session_id text, reason text, created_at datetime, is_resolved boolean )''') c.execute('''CREATE TABLE IF NOT EXISTS system_logs ( id text PRIMARY KEY, user_id text, action text, resource text, details text, ip_address text, user_agent text, status text, timestamp integer )''') conn.commit() conn.close() init_db() def dict_factory(cursor, row): d = {} for idx, col in enumerate(cursor.description): val = row[idx] col_name = col[0] # Auto-parse booleans explicitly if col_name in ['is_active', 'verified', 'is_resolved'] and val is not None: val = bool(val) # Attempt to parse json lists/dicts if possible if isinstance(val, str) and (val.startswith('[') or val.startswith('{')): try: val = json.loads(val) except: pass d[col_name] = val return d @app.api_route("/tables/{table_name}", methods=["GET", "POST"]) async def handle_table(table_name: str, request: Request): conn = get_db() conn.row_factory = dict_factory c = conn.cursor() c.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table_name,)) if not c.fetchone(): raise HTTPException(status_code=404, detail="Table not found") if request.method == "GET": c.execute(f"PRAGMA table_info({table_name})") valid_columns = [col['name'] for col in c.fetchall()] query_params = dict(request.query_params) where_clauses = [] values = [] limit = None for k, v in query_params.items(): if k == "limit": try: limit = int(v) except: pass elif k in valid_columns: where_clauses.append(f"{k} = ?") values.append(v) sql = f"SELECT * FROM {table_name}" if where_clauses: sql += " WHERE " + " AND ".join(where_clauses) if limit is not None: sql += f" LIMIT {limit}" c.execute(sql, values) items = c.fetchall() return {"data": items, "total": len(items)} elif request.method == "POST": data = await request.json() if 'id' not in data: data['id'] = str(uuid.uuid4()) c.execute(f"PRAGMA table_info({table_name})") valid_columns = [col['name'] for col in c.fetchall()] # Serialize lists/dicts to strings for sqlite serialized_data: dict = {} for k, v in data.items(): if k not in valid_columns: continue if isinstance(v, (list, dict)): serialized_data[k] = json.dumps(v) else: serialized_data[k] = v keys = list(serialized_data.keys()) values = list(serialized_data.values()) placeholders = ",".join(["?"] * len(keys)) sql = f"INSERT INTO {table_name} ({','.join(keys)}) VALUES ({placeholders})" try: c.execute(sql, values) conn.commit() except Exception as e: raise HTTPException(status_code=400, detail=str(e)) return data @app.api_route("/tables/{table_name}/{item_id}", methods=["GET", "PUT", "PATCH", "DELETE"]) async def handle_item(table_name: str, item_id: str, request: Request): conn = get_db() conn.row_factory = dict_factory c = conn.cursor() c.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table_name,)) if not c.fetchone(): raise HTTPException(status_code=404, detail="Table not found") if request.method == "GET": c.execute(f"SELECT * FROM {table_name} WHERE id=?", (item_id,)) row = c.fetchone() if not row: raise HTTPException(status_code=404, detail="Item not found") return row elif request.method in ["PUT", "PATCH"]: data = await request.json() # Prevent ID update data.pop("id", None) if not data: return await handle_item(table_name, item_id, Request(scope={"type": "http", "method": "GET"})) c.execute(f"PRAGMA table_info({table_name})") valid_columns = [col['name'] for col in c.fetchall()] serialized_data: dict = {} for k, v in data.items(): if k not in valid_columns: continue if isinstance(v, (list, dict)): serialized_data[k] = json.dumps(v) else: serialized_data[k] = v set_clauses = [] values = [] for k, v in serialized_data.items(): set_clauses.append(f"{k} = ?") values.append(v) values.append(item_id) sql = f"UPDATE {table_name} SET {','.join(set_clauses)} WHERE id=?" try: c.execute(sql, values) conn.commit() if c.rowcount == 0: raise HTTPException(status_code=404, detail="Item not found") except Exception as e: raise HTTPException(status_code=400, detail=str(e)) c.execute(f"SELECT * FROM {table_name} WHERE id=?", (item_id,)) return c.fetchone() elif request.method == "DELETE": c.execute(f"DELETE FROM {table_name} WHERE id=?", (item_id,)) conn.commit() if c.rowcount == 0: raise HTTPException(status_code=404, detail="Item not found") return {"success": True} @app.post("/login") async def login(request: Request): data = await request.json() email = data.get("email", "").lower().strip() password = data.get("password", "").strip() print(f"DEBUG: Login attempt for {email}") conn = get_db() conn.row_factory = dict_factory c = conn.cursor() # Query user by email c.execute("SELECT * FROM users WHERE LOWER(email) = ?", (email,)) user = c.fetchone() if not user: print(f"DEBUG: User {email} not found") raise HTTPException(status_code=404, detail="Email not registered in database") db_password = str(user.get("password") or "").strip() if db_password != password: print(f"DEBUG: Password mismatch for {email}") raise HTTPException(status_code=401, detail="Incorrect password. Please try 'CBIT2024'.") if not user.get("is_active"): raise HTTPException(status_code=403, detail="Your account is deactivated") # Exclude password from response user_data = dict(user) user_data.pop("password", None) return user_data # Mount static files correctly # This should happen LAST so API routes take precedence app.mount("/", StaticFiles(directory=".", html=True), name="static") if __name__ == "__main__": import uvicorn # Make sure to run in current working directory to serve proper static files print("Serving from directory:", os.getcwd()) print("Running API on http://localhost:7860") uvicorn.run("server:app", host="0.0.0.0", port=7860, reload=True)