# -*- coding: utf-8 -*- """CARPOOLING UTEC · prototipo funcional (Emprendimiento I, Grupo 6) Un solo archivo de Python. Hace todo el trabajo "de negocio": usuarios UTEC, viajes, búsqueda por cercanía a la ruta, precios, reservas, calificaciones, ahorro y panel (SQLite). La página (mapa de Google) solo dibuja y pide datos a la API de este mismo archivo. Dónde corre: · Hugging Face (Space de Gradio, cuenta gratuita = ZeroGPU): Gradio levanta el servidor y esta app va dentro. · Google Colab o tu PC: servidor de la librería estándar, no hay nada que instalar (python app.py). Google Maps: la clave se lee del Secret GOOGLE_MAPS_API_KEY del Space. Si falta o Google la rechaza, la página usa OpenStreetMap automáticamente para que la demo no se rompa. """ import inspect import json import math import os import random import re import sqlite3 import sys import threading import time import unicodedata from datetime import datetime, timedelta, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import parse_qs, urlparse PORT = int(os.environ.get("PORT") or 7860) DB_PATH = os.environ.get("DB_PATH") or "/tmp/carpooling_utec_v2.db" LIMA = timezone(timedelta(hours=-5)) _raw_key = (os.environ.get("GOOGLE_MAPS_API_KEY") or "").strip().strip("\"'") KEY_STATUS = "no configurada" GM_KEY = "" if _raw_key: if re.fullmatch(r"[A-Za-z0-9_\-]{20,100}", _raw_key): GM_KEY, KEY_STATUS = _raw_key, "configurada" else: KEY_STATUS = "con formato extraño (se ignora)" # ---------------------------------------------------------------- modelo (valores del informe) UTEC = (-12.1355, -77.0223) DIST = { "Los Olivos": (-11.9700, -77.0700), "San Martín de Porres": (-12.0100, -77.0850), "Comas": (-11.9400, -77.0550), "Independencia": (-11.9900, -77.0540), "San Juan de Lurigancho": (-12.0030, -77.0080), "Ate": (-12.0260, -76.9170), "La Molina": (-12.0850, -76.9400), "Santiago de Surco": (-12.1450, -76.9900), "Villa María del Triunfo": (-12.1650, -76.9400), "Chorrillos": (-12.1750, -77.0150), "Callao": (-12.0560, -77.1180), "Miraflores": (-12.1220, -77.0300), "Chosica": (-11.9330, -76.6970), } HOURS = ["07:00", "08:00", "10:00", "14:00", "18:00"] TRAFFIC = {"07:00": 1.9, "08:00": 1.9, "10:00": 1.25, "14:00": 1.3, "18:00": 1.8} # solo para el modo sin Google FUEL, MIN_PRICE, MAX_PRICE, BUS_DAY, TAXI_KM, TAXI_MIN, WEEKS, JAM_KMH = 0.60, 3, 40, 7, 1.4, 10, 4.3, 12.7 EMAIL_RE = re.compile(r"[^@\s]+@utec\.edu\.pe", re.I) NAMES = ["Ana", "Diego", "Camila", "Luis", "Valeria", "Jorge", "Sofía", "Andrés", "Renzo", "Mía", "Piero", "Daniela", "Marco", "Lucía", "Bruno", "Fiorella", "Rodrigo", "Alexandra", "Mateo", "Paula"] CARS = ["Toyota Yaris", "Kia Rio", "Hyundai Accent", "Nissan Versa", "Suzuki Swift", "Chevrolet Sail"] NOTES = ["Puntual", "Salgo a la hora exacta", "Cabe mochila y laptop", ""] SIM_NAMES = ["Pasajero simulado", "Compañero simulado", "Estudiante simulado"] def hav(a, b): t = math.pi / 180 dl, dn = (b[0] - a[0]) * t, (b[1] - a[1]) * t x = math.sin(dl / 2) ** 2 + math.cos(a[0] * t) * math.cos(b[0] * t) * math.sin(dn / 2) ** 2 return 12742 * math.asin(math.sqrt(x)) def road_km(a, b): return hav(a, b) * 1.35 # la calle real es más larga que la línea recta def seg_dist(p, a, b): """Distancia (km) del punto p al segmento a-b: cuánto se desvía el conductor para recogerte.""" k = math.cos((a[0] + b[0]) / 2 * math.pi / 180) def xy(q): return (q[1] * 111.32 * k, q[0] * 110.57) P, A, B = xy(p), xy(a), xy(b) dx, dy = B[0] - A[0], B[1] - A[1] ll = dx * dx + dy * dy t = 0 if ll == 0 else max(0, min(1, ((P[0] - A[0]) * dx + (P[1] - A[1]) * dy) / ll)) return math.hypot(P[0] - (A[0] + t * dx), P[1] - (A[1] + t * dy)) def suggest(km, seats): """Aporte por pasajero: el combustible del trayecto repartido entre todos los ocupantes, a múltiplos de S/ 0.50.""" return max(MIN_PRICE, round(km * FUEL / (seats + 1) * 2) / 2) def mins(h): a, b = h.split(":") return int(a) * 60 + int(b) def nearest_district(p): return min(DIST, key=lambda k: hav(p, DIST[k])) def now_lima(): return datetime.now(LIMA) def today_str(): return now_lima().strftime("%Y-%m-%d") def next_weekday(after=None): d = (after or now_lima()) + timedelta(days=1) while d.weekday() >= 5: d += timedelta(days=1) return d.strftime("%Y-%m-%d") def slug(s): return re.sub(r"[^a-z]", "", unicodedata.normalize("NFD", s).encode("ascii", "ignore").decode().lower()) # ---------------------------------------------------------------- base de datos LOCK = threading.RLock() def _connect(): for path in (DB_PATH, ":memory:"): try: c = sqlite3.connect(path, check_same_thread=False) c.row_factory = sqlite3.Row c.execute("PRAGMA journal_mode=MEMORY") return c, path except Exception as e: # sin permisos de escritura: seguimos en memoria print(f"[arranque] no pude abrir {path}: {e!r}", flush=True) raise RuntimeError("sin base de datos") DB, DB_USED = _connect() DB.executescript(""" CREATE TABLE IF NOT EXISTS drivers(email TEXT PRIMARY KEY, name TEXT, car TEXT, base REAL DEFAULT 0, base_n INTEGER DEFAULT 0, sum REAL DEFAULT 0, n INTEGER DEFAULT 0, seeded INTEGER DEFAULT 0); CREATE TABLE IF NOT EXISTS trips(id INTEGER PRIMARY KEY AUTOINCREMENT, driver TEXT, from_name TEXT, lat REAL, lon REAL, date TEXT, time TEXT, seats INTEGER, free INTEGER, price REAL, km REAL, notes TEXT, status TEXT, seeded INTEGER DEFAULT 0); CREATE TABLE IF NOT EXISTS bookings(id INTEGER PRIMARY KEY AUTOINCREMENT, trip_id INTEGER, email TEXT, name TEXT, plat REAL, plon REAL, price REAL, code TEXT, status TEXT, stars INTEGER, comment TEXT, pickup TEXT DEFAULT '', created REAL); """) def seed(): """Conductores de ejemplo (ficticios) repartidos por 13 distritos; siempre los mismos.""" r = random.Random(7) date = next_weekday() n = 0 for dn, c in DIST.items(): for j in range(2 + (1 if r.random() < 0.4 else 0)): name = NAMES[n % len(NAMES)] email = f"{slug(name)}{n}@utec.edu.pe" n += 1 lat, lon = c[0] + (r.random() - .5) * .024, c[1] + (r.random() - .5) * .024 seats = 1 + int(r.random() * 3) km = road_km((lat, lon), UTEC) DB.execute("INSERT INTO drivers(email,name,car,base,base_n,seeded) VALUES(?,?,?,?,?,1)", (email, name, r.choice(CARS), round(4.3 + r.random() * .6, 1), 4 + int(r.random() * 27))) hour = "08:00" if j == 0 else r.choice(HOURS) DB.execute("INSERT INTO trips(driver,from_name,lat,lon,date,time,seats,free,price,km,notes,status,seeded) " "VALUES(?,?,?,?,?,?,?,?,?,?,?, 'active',1)", (email, dn, lat, lon, date, hour, seats, seats, suggest(km, seats), round(km, 1), r.choice(NOTES))) DB.commit() with LOCK: if DB.execute("SELECT COUNT(*) FROM trips").fetchone()[0] == 0: seed() # ---------------------------------------------------------------- utilidades de validación class ApiError(Exception): def __init__(self, msg, code=400): super().__init__(msg) self.msg, self.code = msg, code def clean(s, n): """Texto seguro: sin caracteres de control ni < >, espacios normalizados y largo máximo.""" s = "" if s is None else str(s) s = "".join(ch for ch in s if ch.isprintable() and ch not in "<>") return re.sub(r"\s+", " ", s).strip()[:n] def num(q, key, lo, hi, default=None): v = q.get(key, default) try: v = float(v) except (TypeError, ValueError): raise ApiError(f"Dato inválido: {key}.") if math.isnan(v) or v < lo or v > hi: raise ApiError(f"Dato fuera de rango: {key}.") return v def user_email(q): e = clean(q.get("email"), 80).lower() if not EMAIL_RE.fullmatch(e): raise ApiError("Ingresa con tu correo @utec.edu.pe.", 401) return e def point(q, a="lat", b="lng"): """Punto dentro de Lima (la demo solo cubre esa zona).""" la, lo = num(q, a, -13.5, -11.0), num(q, b, -78.0, -76.0) return (la, lo) def rating(row): n = (row["base_n"] or 0) + (row["n"] or 0) return (None if n == 0 else round((row["base"] * row["base_n"] + row["sum"]) / n, 1)), n def trip_json(r, me=None): avg, n = rating(r) return {"id": r["id"], "driver": {"name": r["name"], "car": r["car"], "rating": avg, "n": n}, "from": r["from_name"], "lat": r["lat"], "lon": r["lon"], "date": r["date"], "time": r["time"], "seats": r["seats"], "free": r["free"], "price": r["price"], "km": r["km"], "notes": r["notes"], "status": r["status"], "own": me is not None and r["driver"] == me} TRIP_SQL = ("SELECT t.*, d.name, d.car, d.base, d.base_n, d.sum, d.n FROM trips t " "JOIN drivers d ON d.email = t.driver ") def default_name(e): return e.split("@")[0].split(".")[0].capitalize() # ---------------------------------------------------------------- API def h_login(q): e = clean(q.get("email"), 80).lower() if not EMAIL_RE.fullmatch(e): raise ApiError("Escribe un correo que termine en @utec.edu.pe.") with LOCK: d = DB.execute("SELECT seeded FROM drivers WHERE email=?", (e,)).fetchone() if d and d["seeded"]: raise ApiError("Ese correo es de un conductor de ejemplo. Usa el tuyo.") name = clean(q.get("name"), 30) or default_name(e) DB.execute("UPDATE drivers SET name=? WHERE email=?", (name, e)) DB.commit() return {"user": {"email": e, "name": name}} def h_search(q): la, lo = point(q) hour = clean(q.get("time"), 5) or "*" if hour != "*" and hour not in HOURS: raise ApiError("Hora inválida.") radius = num(q, "radius", 1, 25, 8) me = clean(q.get("email"), 80).lower() or None today = today_str() out = [] booked = set() with LOCK: cur = DB.execute("UPDATE trips SET date=? WHERE seeded=1 AND status='active' AND date=?", (today,)).fetchall() if me: booked = {x[0] for x in DB.execute("SELECT trip_id FROM bookings WHERE email=? AND status='confirmed'", (me,))} for r in rows: if r["free"] < 1 and r["id"] not in booked and r["driver"] != me: continue # un viaje lleno solo lo ven su conductor y quien ya tiene asiento dt = 0 if hour == "*" else abs(mins(r["time"]) - mins(hour)) if dt > 60: continue pk = seg_dist((la, lo), (r["lat"], r["lon"]), UTEC) if pk > radius: continue j = trip_json(r, me) j["pass_km"] = round(pk, 1) j["booked"] = r["id"] in booked j["score"] = pk + dt / 60 * 1.5 out.append(j) out.sort(key=lambda t: t["score"]) return {"trips": out[:25], "total": len(out)} def h_quote(q): """Aporte sugerido para publicar un viaje desde un punto.""" km = road_km(point(q), UTEC) return {"km": round(km, 1), "price": suggest(km, int(num(q, "seats", 1, 4, 2)))} def h_savings(q): """Cuánto cuesta hoy viajar a UTEC y cuánto con carpooling (ida y vuelta, por semana).""" ds = clean(q.get("district"), 40) if ds: if ds not in DIST: raise ApiError("Distrito inválido.") pt = DIST[ds] else: pt = point(q) mode = clean(q.get("mode"), 5) or "bus" if mode not in ("bus", "auto", "taxi"): raise ApiError("Modo de viaje inválido.") days = int(num(q, "days", 1, 7, 5)) riders = int(num(q, "riders", 1, 4, 2)) km = road_km(pt, UTEC) price = suggest(km, riders) now = BUS_DAY * days if mode == "bus" else km * FUEL * 2 * days if mode == "auto" else max(TAXI_MIN, km * TAXI_KM) * 2 * days cp = price * 2 * days gasto = km * FUEL * 2 * days rec = price * riders * 2 * days return {"km": round(km, 1), "price": price, "now": round(now, 2), "carpool": round(cp, 2), "month": round((now - cp) * WEEKS, 2), "fuel": round(gasto, 2), "recovered": round(rec, 2), "net": round(max(0, gasto - rec), 2), "mode": mode, "riders": riders} def h_panel(q): me = clean(q.get("email"), 80).lower() with LOCK: a = DB.execute("SELECT COUNT(*) c, COALESCE(SUM(free),0) s, COALESCE(AVG(price),0) p FROM trips WHERE status='active'").fetchone() books = DB.execute("SELECT COUNT(*) FROM bookings b JOIN trips t ON t.id=b.trip_id WHERE b.status!='cancelled' AND t.status!='deleted'").fetchone()[0] rated = DB.execute("SELECT COUNT(*) FROM bookings b JOIN trips t ON t.id=b.trip_id WHERE b.stars IS NOT NULL AND t.status!='deleted'").fetchone()[0] mine = DB.execute("SELECT COUNT(*) FROM trips WHERE driver=? AND status!='deleted'", (me,)).fetchone()[0] if me else 0 zones = DB.execute("SELECT from_name, COUNT(*) c FROM trips WHERE status='active' GROUP BY from_name ORDER BY c DESC, from_name").fetchall() return {"trips": a["c"], "seats": a["s"], "bookings": books, "avg_price": round(a["p"], 2), "mine": mine, "ratings": rated, "zones": [[z[0], z[1]] for z in zones]} def h_publish(q): e = user_email(q) la, lo = point(q, "lat", "lon") time_ = clean(q.get("time"), 5) date = clean(q.get("date"), 10) try: d = datetime.strptime(date, "%Y-%m-%d") if re.fullmatch(r"\d{4}-\d{2}-\d{2}", date) else None except ValueError: d = None today = datetime.strptime(today_str(), "%Y-%m-%d") if d is None or d < today: raise ApiError("Elige una fecha de hoy en adelante.") if d > today + timedelta(days=60): raise ApiError("Publica con máximo 60 días de anticipación.") if time_ not in HOURS: raise ApiError("Elige una hora de llegada de la lista.") try: seats_f = float(q.get("seats")) except (TypeError, ValueError): seats_f = float("nan") if not (1 <= seats_f <= 4) or seats_f != int(seats_f): raise ApiError("Los asientos libres deben ser entre 1 y 4.") seats = int(seats_f) try: price = float(q.get("price")) except (TypeError, ValueError): price = float("nan") if not (MIN_PRICE <= price <= MAX_PRICE): raise ApiError(f"El aporte por pasajero debe estar entre S/ {MIN_PRICE} y S/ {MAX_PRICE}.") price = round(price * 2) / 2 notes, car = clean(q.get("notes"), 80), clean(q.get("car"), 40) or "Auto particular" name = clean(q.get("name"), 30) or default_name(e) with LOCK: if DB.execute("SELECT 1 FROM drivers WHERE email=? AND seeded=1", (e,)).fetchone(): raise ApiError("Ese correo es de un conductor de ejemplo.") if DB.execute("SELECT COUNT(*) FROM trips").fetchone()[0] > 2000: raise ApiError("La demo está llena, intenta más tarde.", 503) if DB.execute("SELECT 1 FROM trips WHERE driver=? AND date=? AND time=? AND status='active'", (e, date, time_)).fetchone(): raise ApiError("Ya tienes un viaje activo para esa fecha y hora.") if DB.execute("SELECT COUNT(*) FROM trips WHERE driver=? AND status='active'", (e,)).fetchone()[0] >= 5: raise ApiError("Máximo 5 viajes activos por conductor en la demo.") DB.execute("INSERT INTO drivers(email,name,car) VALUES(?,?,?) ON CONFLICT(email) DO UPDATE SET name=excluded.name, car=excluded.car", (e, name, car)) km = road_km((la, lo), UTEC) cur = DB.execute("INSERT INTO trips(driver,from_name,lat,lon,date,time,seats,free,price,km,notes,status,seeded) " "VALUES(?,?,?,?,?,?,?,?,?,?,?, 'active',0)", (e, nearest_district((la, lo)), la, lo, date, time_, seats, seats, price, round(km, 1), notes)) DB.commit() r = DB.execute(TRIP_SQL + "WHERE t.id=?", (cur.lastrowid,)).fetchone() return {"trip": trip_json(r, e)} def h_book(q): e = user_email(q) pla, plo = point(q, "plat", "plng") tid = int(num(q, "trip_id", 1, 10 ** 9)) name = clean(q.get("name"), 30) or default_name(e) pickup = clean(q.get("pickup"), 5) if not re.fullmatch(r"\d\d:\d\d", pickup): pickup = "" with LOCK: t = DB.execute("SELECT * FROM trips WHERE id=?", (tid,)).fetchone() if not t or t["status"] != "active": raise ApiError("Ese viaje ya no está disponible.", 409) if t["driver"] == e: raise ApiError("No puedes reservar tu propio viaje.") if t["free"] < 1: raise ApiError("El viaje se llenó.", 409) if DB.execute("SELECT 1 FROM bookings WHERE trip_id=? AND email=? AND status='confirmed'", (tid, e)).fetchone(): raise ApiError("Ya tienes un asiento en este viaje.") if DB.execute("SELECT COUNT(*) FROM bookings").fetchone()[0] > 5000: raise ApiError("La demo está llena, intenta más tarde.", 503) code = "UTEC-" + str(random.randint(1000, 9999)) DB.execute("INSERT INTO bookings(trip_id,email,name,plat,plon,price,code,status,pickup,created) VALUES(?,?,?,?,?,?,?, 'confirmed',?,?)", (tid, e, name, pla, plo, t["price"], code, pickup, time.time())) DB.execute("UPDATE trips SET free=free-1 WHERE id=?", (tid,)) DB.commit() d = DB.execute("SELECT name FROM drivers WHERE email=?", (t["driver"],)).fetchone() return {"code": code, "price": t["price"], "driver": d["name"] if d else ""} def _own_booking(e, bid): b = DB.execute("SELECT * FROM bookings WHERE id=? AND email=?", (bid, e)).fetchone() if not b: raise ApiError("No encontré esa reserva en tu cuenta.", 404) return b def h_cancel(q): e = user_email(q) with LOCK: b = _own_booking(e, int(num(q, "booking_id", 1, 10 ** 9))) if b["status"] != "confirmed": raise ApiError("Esa reserva ya no se puede cancelar.") DB.execute("UPDATE bookings SET status='cancelled' WHERE id=?", (b["id"],)) DB.execute("UPDATE trips SET free=MIN(seats, free+1) WHERE id=? AND status='active'", (b["trip_id"],)) DB.commit() return {} def h_finish(q): """Simula que el viaje terminó (en la demo no hay GPS): habilita calificar.""" e = user_email(q) with LOCK: b = _own_booking(e, int(num(q, "booking_id", 1, 10 ** 9))) if b["status"] != "confirmed": raise ApiError("Solo se puede completar una reserva confirmada.") DB.execute("UPDATE bookings SET status='completed' WHERE id=?", (b["id"],)) DB.commit() return {} def h_rate(q): e = user_email(q) stars = int(num(q, "stars", 1, 5)) comment = clean(q.get("comment"), 120) with LOCK: b = _own_booking(e, int(num(q, "booking_id", 1, 10 ** 9))) if b["status"] != "completed": raise ApiError("Solo puedes calificar viajes completados.") if b["stars"]: raise ApiError("Ya calificaste este viaje.") t = DB.execute("SELECT t.driver, d.name FROM trips t JOIN drivers d ON d.email=t.driver WHERE t.id=?", (b["trip_id"],)).fetchone() DB.execute("UPDATE bookings SET stars=?, comment=? WHERE id=?", (stars, comment, b["id"])) DB.execute("UPDATE drivers SET sum=sum+?, n=n+1 WHERE email=?", (stars, t["driver"])) DB.commit() return {"driver": t["name"]} def _own_trip(e, tid): t = DB.execute("SELECT * FROM trips WHERE id=? AND driver=?", (tid, e)).fetchone() if not t: raise ApiError("No encontré ese viaje en tu cuenta.", 404) if t["status"] != "active": raise ApiError("Ese viaje ya no está activo.") return t def h_trip_cancel(q): e = user_email(q) tid = int(num(q, "trip_id", 1, 10 ** 9)) with LOCK: _own_trip(e, tid) DB.execute("UPDATE trips SET status='cancelled' WHERE id=?", (tid,)) DB.execute("UPDATE bookings SET status='cancelled' WHERE trip_id=? AND status='confirmed'", (tid,)) DB.commit() return {} def h_trip_complete(q): e = user_email(q) tid = int(num(q, "trip_id", 1, 10 ** 9)) with LOCK: _own_trip(e, tid) DB.execute("UPDATE trips SET status='completed' WHERE id=?", (tid,)) DB.execute("UPDATE bookings SET status='completed' WHERE trip_id=? AND status='confirmed'", (tid,)) DB.commit() return {} def h_sim_passenger(q): """Para demostrar: suma un pasajero ficticio al viaje del conductor.""" e = user_email(q) tid = int(num(q, "trip_id", 1, 10 ** 9)) with LOCK: t = _own_trip(e, tid) if t["free"] < 1: raise ApiError("Ese viaje no tiene asientos libres.") k = DB.execute("SELECT COUNT(*) FROM bookings WHERE trip_id=?", (tid,)).fetchone()[0] name = SIM_NAMES[t["seats"] - t["free"]] if t["seats"] - t["free"] < len(SIM_NAMES) else SIM_NAMES[0] DB.execute("INSERT INTO bookings(trip_id,email,name,plat,plon,price,code,status,pickup,created) VALUES(?,?,?,?,?,?,?, 'confirmed','',?)", (tid, f"sim{tid}x{k}@utec.edu.pe", name, t["lat"], t["lon"], t["price"], "UTEC-" + str(random.randint(1000, 9999)), time.time())) DB.execute("UPDATE trips SET free=free-1 WHERE id=?", (tid,)) DB.commit() return {} def h_mine(q): e = user_email(q) with LOCK: rider = DB.execute( "SELECT b.id, b.code, b.status, b.stars, b.comment, b.price, b.pickup, t.id AS tid, t.from_name, t.date, t.time, t.status AS tstatus, " "d.name AS dname, d.car FROM bookings b JOIN trips t ON t.id=b.trip_id JOIN drivers d ON d.email=t.driver " "WHERE b.email=? ORDER BY b.id DESC LIMIT 30", (e,)).fetchall() trips = DB.execute("SELECT * FROM trips WHERE driver=? AND status!='deleted' ORDER BY id DESC LIMIT 20", (e,)).fetchall() riders = {} if trips: ids = [t["id"] for t in trips] for b in DB.execute("SELECT trip_id, name FROM bookings WHERE status!='cancelled' AND trip_id IN (%s) ORDER BY id" % ",".join("?" * len(ids)), ids): riders.setdefault(b["trip_id"], []).append(b["name"]) return {"rider": [dict(r) for r in rider], "driver": [dict({k: t[k] for k in ("id", "from_name", "date", "time", "seats", "free", "price", "status")}, riders=riders.get(t["id"], [])) for t in trips]} def h_reset_me(q): """Borra SOLO los datos de quien lo pide: sus reservas, sus calificaciones y sus viajes.""" e = user_email(q) with LOCK: for b in DB.execute("SELECT b.stars, t.driver FROM bookings b JOIN trips t ON t.id=b.trip_id WHERE b.email=? AND b.stars IS NOT NULL", (e,)).fetchall(): DB.execute("UPDATE drivers SET sum=sum-?, n=MAX(0, n-1) WHERE email=?", (b["stars"], b["driver"])) for b in DB.execute("SELECT trip_id FROM bookings WHERE email=? AND status IN ('confirmed','completed')", (e,)).fetchall(): DB.execute("UPDATE trips SET free=MIN(seats, free+1) WHERE id=? AND status='active'", (b["trip_id"],)) DB.execute("DELETE FROM bookings WHERE email=?", (e,)) DB.execute("UPDATE bookings SET status='cancelled' WHERE status='confirmed' AND trip_id IN (SELECT id FROM trips WHERE driver=?)", (e,)) DB.execute("UPDATE trips SET status='deleted' WHERE driver=?", (e,)) DB.execute("UPDATE drivers SET sum=0, n=0 WHERE email=?", (e,)) DB.commit() return {} GET_ROUTES = {"/api/search": h_search, "/api/quote": h_quote, "/api/savings": h_savings, "/api/panel": h_panel, "/api/mine": h_mine} POST_ROUTES = {"/api/login": h_login, "/api/publish": h_publish, "/api/book": h_book, "/api/cancel": h_cancel, "/api/finish": h_finish, "/api/rate": h_rate, "/api/trip_cancel": h_trip_cancel, "/api/trip_complete": h_trip_complete, "/api/sim_passenger": h_sim_passenger, "/api/reset_me": h_reset_me} # ---------------------------------------------------------------- página PAGE = r""" CARPOOLING UTEC
""" def render_page(): cfg = {"gmKey": GM_KEY, "utec": UTEC, "districts": DIST, "hours": HOURS, "traffic": TRAFFIC, "jamKmh": JAM_KMH, "model": {"fuel": FUEL, "bus": BUS_DAY, "taxiKm": TAXI_KM, "taxiMin": TAXI_MIN, "weeks": WEEKS}, "today": today_str(), "nextDay": next_weekday()} blob = json.dumps(cfg, ensure_ascii=False).replace("<", "\\u003c").replace(">", "\\u003e").replace("&", "\\u0026") return PAGE.replace("/*__CFG__*/null", blob).encode("utf-8") # ---------------------------------------------------------------- respuestas (las usan los dos servidores) SEC = {"X-Content-Type-Options": "nosniff", "Referrer-Policy": "strict-origin-when-cross-origin", "Cache-Control": "no-store"} MAX_BODY = 16384 JSON_T = "application/json; charset=utf-8" def _json(code, obj): return code, JSON_T, json.dumps(obj, ensure_ascii=False).encode("utf-8") def _run(fn, q): try: res = fn(q) res["ok"] = True return _json(200, res) except ApiError as e: return _json(e.code, {"ok": False, "error": e.msg}) except Exception as e: # nunca devolvemos trazas al navegador print("[error]", repr(e), flush=True) return _json(500, {"ok": False, "error": "Error interno del servidor."}) def route(method, path, query, raw): """Único lugar donde se decide qué responder (lo usan el servidor de Python y el de Gradio). raw = cuerpo ya leído (bytes), o None si superaba el límite. Devuelve (código, tipo, cuerpo).""" if method in ("GET", "HEAD"): if path in ("/", "/index.html"): return 200, "text/html; charset=utf-8", render_page() if path == "/api/health": return _json(200, {"ok": True}) if path == "/favicon.ico": return 204, "image/x-icon", b"" fn = GET_ROUTES.get(path) if fn: return _run(fn, {k: v[0] for k, v in parse_qs(query, keep_blank_values=True).items()}) return _json(404, {"ok": False, "error": "No existe."}) if method == "POST": if raw is None: return _json(413, {"ok": False, "error": "Solicitud demasiado grande."}) fn = POST_ROUTES.get(path) if not fn: return _json(404, {"ok": False, "error": "No existe."}) try: q = json.loads(raw.decode("utf-8") or "{}") if not isinstance(q, dict): raise ValueError except ValueError: return _json(400, {"ok": False, "error": "JSON inválido."}) return _run(fn, q) return _json(405, {"ok": False, "error": "Método no permitido."}) # ---------------------------------------------------------------- servidor 1: librería estándar (Colab, tu PC) class Handler(BaseHTTPRequestHandler): # HTTP/1.0 a propósito: cada respuesta cierra la conexión, así no hay choques con proxies server_version = "CarpoolingUTEC/2.0" timeout = 30 def log_message(self, fmt, *args): if "/api/health" not in (args[0] if args else ""): sys.stdout.write("[web] %s\n" % (fmt % args)) sys.stdout.flush() def _answer(self, method): u = urlparse(self.path) raw = b"" if method == "POST": try: n = int(self.headers.get("Content-Length") or 0) except ValueError: n = -1 raw = None if (n < 0 or n > MAX_BODY) else (self.rfile.read(n) if n else b"") code, ctype, body = route(method, u.path, u.query, raw) self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(body))) for k, v in SEC.items(): self.send_header(k, v) self.end_headers() if method != "HEAD": self.wfile.write(body) def do_GET(self): self._answer("GET") def do_HEAD(self): self._answer("HEAD") def do_POST(self): self._answer("POST") class WebServer(ThreadingHTTPServer): daemon_threads = True allow_reuse_address = True request_queue_size = 64 def handle_error(self, request, client_address): # conexiones cortadas por el navegador: sin ruido if isinstance(sys.exc_info()[1], (BrokenPipeError, ConnectionResetError, TimeoutError)): return super().handle_error(request, client_address) def serve_stdlib(): srv = WebServer(("0.0.0.0", PORT), Handler) print(f"[arranque] escuchando en http://0.0.0.0:{PORT}", flush=True) try: srv.serve_forever() except KeyboardInterrupt: pass # ---------------------------------------------------------------- servidor 2: Gradio (Hugging Face, cuenta gratuita) # Con cuenta gratuita, Hugging Face solo deja un Space de Gradio sobre ZeroGPU, y ZeroGPU exige (1) que el servidor # lo levante Gradio y (2) que exista al menos una función @spaces.GPU. Esa función no hace nada ni se usa: # solo está para pasar el chequeo de arranque. Nuestra página y nuestra API van en el mismo servidor. try: import spaces @spaces.GPU def gpu_ping(texto: str) -> str: return texto except Exception: # fuera de Hugging Face no existe la librería "spaces" spaces = None gpu_ping = None def en_hf(): """True en un Space de Hugging Face. USAR_GRADIO=1 / 0 lo fuerza (útil para pruebas).""" forzar = os.environ.get("USAR_GRADIO") if forzar in ("0", "1"): return forzar == "1" return bool(os.environ.get("SPACE_ID") or os.environ.get("SPACES_ZERO_GPU") or os.environ.get("SYSTEM") == "spaces") def _acepta(fn, nombre): try: ps = inspect.signature(fn).parameters except (TypeError, ValueError): return False return nombre in ps or any(p.kind is inspect.Parameter.VAR_KEYWORD for p in ps.values()) def _rutas_asgi(app, al_frente=False): """Registra nuestras rutas en el servidor FastAPI que trae Gradio, para compartir el mismo puerto.""" from starlette.concurrency import run_in_threadpool from starlette.requests import Request from starlette.responses import Response async def leer(request): try: n = int(request.headers.get("content-length") or 0) except ValueError: return None if n < 0 or n > MAX_BODY: return None buf = b"" async for trozo in request.stream(): buf += trozo if len(buf) > MAX_BODY: return None return buf def crear(path): async def endpoint(request: Request): raw = await leer(request) if request.method == "POST" else b"" consulta = request.scope.get("query_string", b"").decode("latin-1") code, ctype, body = await run_in_threadpool(route, request.method, path, consulta, raw) cab = dict(SEC) cab["Content-Type"] = ctype return Response(content=body, status_code=code, headers=cab) return endpoint gets = ["/", "/index.html", "/api/health", "/favicon.ico"] + list(GET_ROUTES) n0 = len(app.router.routes) for p in dict.fromkeys(gets + list(POST_ROUTES)): metodos = (["GET", "HEAD"] if p in gets else []) + (["POST"] if p in POST_ROUTES else []) app.add_api_route(p, crear(p), methods=metodos, include_in_schema=False) if al_frente: # servidores que ya traen su propia página en "/": las nuestras van primero mias = app.router.routes[n0:] del app.router.routes[n0:] app.router.routes[0:0] = mias def serve_gradio(): import gradio as gr print(f"[arranque] Gradio {getattr(gr, '__version__', '?')} · librería spaces: {'sí' if spaces else 'no'}", flush=True) kw = {"server_name": "0.0.0.0", "server_port": PORT} Server = getattr(gr, "Server", None) if Server is not None: # camino oficial: gradio.Server = FastAPI + motor de Gradio (admite ZeroGPU) app = Server() _rutas_asgi(app) if gpu_ping is not None: try: @app.api(name="ping") def api_ping(texto: str) -> str: return gpu_ping(texto) except Exception as e: print(f"[arranque] no pude registrar /ping ({e!r}); sigo", flush=True) if _acepta(app.launch, "ssr_mode"): kw["ssr_mode"] = False print("[arranque] el servidor es gradio.Server (sirve la página y la API)", flush=True) app.launch(**kw) return # Gradio más antiguo, sin gradio.Server: lanzamos un Blocks mínimo y ponemos nuestras rutas por delante print("[arranque] esta versión de Gradio no tiene gradio.Server; uso gr.Blocks", flush=True) with gr.Blocks(title="Carpooling UTEC") as demo: gr.Markdown("# Carpooling UTEC") if gpu_ping is not None: caja = gr.Textbox(visible=False) boton = gr.Button("ping", visible=False) boton.click(gpu_ping, caja, caja, api_name="ping") kw["prevent_thread_lock"] = True if _acepta(demo.launch, "ssr_mode"): kw["ssr_mode"] = False try: app = demo.launch(**kw)[0] _rutas_asgi(app, al_frente=True) except Exception: try: demo.close() except Exception: pass raise demo.block_thread() def main(): print(f"[arranque] Python {sys.version.split()[0]} · puerto {PORT} · base de datos: {DB_USED} · " f"clave de Google Maps: {KEY_STATUS}", flush=True) if en_hf(): try: serve_gradio() return except KeyboardInterrupt: return except Exception as e: print(f"[arranque] Gradio no pudo arrancar ({e!r}); uso el servidor estándar de Python", flush=True) serve_stdlib() if __name__ == "__main__": main()