from pathlib import Path from datetime import datetime, timezone from fastapi import FastAPI, Request, Form, Depends, HTTPException from fastapi.responses import RedirectResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from sqlalchemy.orm import Session from starlette.middleware.sessions import SessionMiddleware from . import security, license_online, auth from .database import init_db, get_db, SessionLocal from .models import Truck, MaintenanceItem, MaintenanceRecord, User from .maintenance_logic import compute_item_status, worst_status from .geotab_client import get_geotab_client from .sync_service import run_sync, seed_default_maintenance_items from .scheduler import start_scheduler, reschedule, shutdown_scheduler APP_DIR = Path(__file__).resolve().parent templates = Jinja2Templates(directory=str(APP_DIR / "templates")) templates.env.globals["header_info"] = lambda: security.load_active_license() or {} app = FastAPI(title="Frota Trucks") app.mount("/static", StaticFiles(directory=str(APP_DIR / "static")), name="static") @app.on_event("startup") def on_startup(): init_db() db = SessionLocal() try: auth.seed_default_user(db) finally: db.close() license_online.refresh_from_server() start_scheduler() @app.on_event("shutdown") def on_shutdown(): shutdown_scheduler() PUBLIC_PREFIXES = ("/static", "/ativacao", "/api/ativacao") SETUP_PREFIXES = ("/setup", "/api/setup") RENOVACAO_PREFIXES = ("/renovacao", "/api/licenca/verificar-agora") LOGIN_PREFIXES = ("/login", "/api/login") @app.middleware("http") async def gate_middleware(request: Request, call_next): path = request.url.path if path.startswith(PUBLIC_PREFIXES) or path.startswith(RENOVACAO_PREFIXES): return await call_next(request) if not security.load_active_license(): if license_online.refresh_from_server(): pass # renovou sozinho com a chave publicada, segue o fluxo normal elif license_online.known_activation_data(): return RedirectResponse("/renovacao") else: return RedirectResponse("/ativacao") elif license_online.is_server_blocked(): security.BLOCK_FLAG_FILE.touch() dest = "/renovacao" if license_online.known_activation_data() else "/ativacao" return RedirectResponse(dest) if path.startswith(SETUP_PREFIXES): return await call_next(request) if not security.has_geotab_credentials(): return RedirectResponse("/setup") if path.startswith(LOGIN_PREFIXES): return await call_next(request) if not request.session.get("logged_in"): return RedirectResponse("/login") return await call_next(request) # Precisa ser adicionado DEPOIS do gate_middleware acima: no Starlette, o # ultimo middleware adicionado fica por fora (processa a request primeiro), # entao isso garante que request.session ja existe quando o gate roda. app.add_middleware(SessionMiddleware, secret_key=security.get_session_secret()) # ---------- Ativação ---------- @app.get("/ativacao") def ativacao_form(request: Request): if security.load_active_license(): return RedirectResponse("/setup" if not security.has_geotab_credentials() else "/") return templates.TemplateResponse("activation.html", {"request": request, "error": None}) @app.post("/api/ativacao") def ativacao_submit(request: Request, nome_empresa: str = Form(...), codigo: str = Form(...), license_key: str = Form("")): machine_id = codigo.strip() license_key = license_key.strip() if license_key: chave = license_key else: chave = license_online.fetch_published_key(machine_id) if not chave: return templates.TemplateResponse("activation.html", { "request": request, "error": "Nenhuma chave encontrada pra esse código (sem internet, ou o fornecedor ainda não " "publicou). Tente de novo em instantes, ou digite a chave manualmente se já recebeu uma.", "submitted_empresa": nome_empresa, "submitted_codigo": machine_id, }) ok, info_or_error = security.validate_license_key(machine_id, chave) if not ok: return templates.TemplateResponse("activation.html", { "request": request, "error": info_or_error, "submitted_key": license_key, "submitted_empresa": nome_empresa, "submitted_codigo": machine_id, }) security.save_license(chave, nome_empresa, info_or_error) return RedirectResponse("/setup", status_code=303) # ---------- Renovação da licença ---------- @app.get("/renovacao") def renovacao_page(request: Request): dados = license_online.known_activation_data() return templates.TemplateResponse("renovacao.html", { "request": request, "codigo": dados["machine_id"] if dados else None, }) @app.post("/api/licenca/verificar-agora") def verificar_licenca_agora(): ok = license_online.refresh_from_server() return RedirectResponse("/" if ok else "/renovacao", status_code=303) # ---------- Setup Geotab ---------- @app.get("/setup") def setup_form(request: Request): if security.has_geotab_credentials(): return RedirectResponse("/") return templates.TemplateResponse("setup.html", {"request": request, "error": None}) @app.post("/api/setup") def setup_submit(request: Request, username: str = Form(...), password: str = Form(...), database: str = Form(""), server: str = Form("my.geotab.com"), db: Session = Depends(get_db)): creds = {"username": username, "password": password, "database": database, "server": server} try: client = get_geotab_client(credentials=creds) client.get_devices() # forca autenticacao + smoke test except Exception as exc: return templates.TemplateResponse("setup.html", { "request": request, "error": f"Não foi possível conectar: {exc}", "username": username, "database": database, "server": server, }) security.save_geotab_credentials(username, password, database, server) seed_default_maintenance_items(db) run_sync() return RedirectResponse("/", status_code=303) # ---------- Login do painel ---------- @app.get("/login") def login_form(request: Request): return templates.TemplateResponse("login.html", {"request": request, "error": None}) @app.post("/api/login") def login_submit(request: Request, username: str = Form(...), password: str = Form(...), db: Session = Depends(get_db)): if not auth.verify_user(db, username, password): return templates.TemplateResponse("login.html", { "request": request, "error": "Usuário ou senha incorretos.", "username": username, }) request.session["logged_in"] = True request.session["username"] = username return RedirectResponse("/", status_code=303) @app.post("/api/logout") def logout(request: Request): request.session.clear() return RedirectResponse("/login", status_code=303) # ---------- Usuários ---------- @app.get("/usuarios") def usuarios_page(request: Request, db: Session = Depends(get_db)): users = db.query(User).order_by(User.username).all() return templates.TemplateResponse("usuarios.html", { "request": request, "users": users, "error": None, }) @app.post("/api/usuarios") def criar_usuario(request: Request, username: str = Form(...), password: str = Form(...), confirmar_senha: str = Form(...), db: Session = Depends(get_db)): users = db.query(User).order_by(User.username).all() if len(password) < 4: return templates.TemplateResponse("usuarios.html", { "request": request, "users": users, "error": "A senha precisa ter pelo menos 4 caracteres.", }) if password != confirmar_senha: return templates.TemplateResponse("usuarios.html", { "request": request, "users": users, "error": "As senhas não conferem.", }) if db.query(User).filter_by(username=username).first(): return templates.TemplateResponse("usuarios.html", { "request": request, "users": users, "error": "Já existe um usuário com esse nome.", }) auth.create_user(db, username, password) return RedirectResponse("/usuarios", status_code=303) @app.post("/api/usuarios/{user_id}/senha") def trocar_senha_usuario(request: Request, user_id: int, nova_senha: str = Form(...), confirmar_nova_senha: str = Form(...), db: Session = Depends(get_db)): users = db.query(User).order_by(User.username).all() if len(nova_senha) < 4: return templates.TemplateResponse("usuarios.html", { "request": request, "users": users, "error": "A senha precisa ter pelo menos 4 caracteres.", }) if nova_senha != confirmar_nova_senha: return templates.TemplateResponse("usuarios.html", { "request": request, "users": users, "error": "As senhas não conferem.", }) auth.change_password(db, user_id, nova_senha) return RedirectResponse("/usuarios", status_code=303) @app.post("/api/usuarios/{user_id}/excluir") def excluir_usuario(user_id: int, db: Session = Depends(get_db)): if db.query(User).count() <= 1: raise HTTPException(400, "Não é possível excluir o último usuário.") auth.delete_user(db, user_id) return RedirectResponse("/usuarios", status_code=303) # ---------- Dashboard ---------- @app.get("/") def dashboard(request: Request, status: str = "todos", db: Session = Depends(get_db)): filtro = status if status in ("ok", "vencendo", "vencido") else "todos" trucks = db.query(Truck).filter_by(active=True).order_by(Truck.name).all() items = db.query(MaintenanceItem).filter_by(active=True).all() rows = [] counts = {"total": len(trucks), "ok": 0, "vencendo": 0, "vencido": 0} for truck in trucks: statuses = [] for item in items: last_record = ( db.query(MaintenanceRecord) .filter_by(truck_id=truck.id, item_id=item.id) .order_by(MaintenanceRecord.service_date.desc()) .first() ) result = compute_item_status(truck, item, last_record) statuses.append(result["status"]) tracked = [s for s in statuses if s != "sem_registro"] status = worst_status(tracked) if tracked else ("sem_registro" if statuses else "sem_configuracao") if status in counts: counts[status] += 1 rows.append({"truck": truck, "status": status}) if filtro != "todos": rows = [r for r in rows if r["status"] == filtro] last_sync_truck = db.query(Truck).filter(Truck.last_synced_at.isnot(None)).order_by(Truck.last_synced_at.desc()).first() last_sync_local = None if last_sync_truck: last_sync_local = last_sync_truck.last_synced_at.replace(tzinfo=timezone.utc).astimezone() mensalidade_aviso = None license_info = security.load_active_license() if license_info and license_info["expires_epoch"] != 0: expira_em = datetime.fromtimestamp(license_info["expires_epoch"], tz=timezone.utc).date() dias_restantes = (expira_em - datetime.now(timezone.utc).date()).days if dias_restantes <= 5: mensalidade_aviso = f"Faltam {dias_restantes} dia(s) para o fim da licença." return templates.TemplateResponse("dashboard.html", { "request": request, "trucks": rows, "counts": counts, "filtro": filtro, "last_sync": last_sync_local.strftime("%d/%m/%Y %H:%M") if last_sync_local else None, "mensalidade_aviso": mensalidade_aviso, }) @app.post("/api/sync") def api_sync(): run_sync() return RedirectResponse("/", status_code=303) # ---------- Detalhe do truck ---------- @app.get("/trucks/{truck_id}") def truck_detail(request: Request, truck_id: int, db: Session = Depends(get_db)): truck = db.get(Truck, truck_id) if truck is None: raise HTTPException(404, "Truck não encontrado") items = db.query(MaintenanceItem).filter_by(active=True).all() rows = [] for item in items: last_record = ( db.query(MaintenanceRecord) .filter_by(truck_id=truck.id, item_id=item.id) .order_by(MaintenanceRecord.service_date.desc()) .first() ) result = compute_item_status(truck, item, last_record) rows.append({"item": item, "last_record": last_record, **result}) history = ( db.query(MaintenanceRecord) .filter_by(truck_id=truck.id) .order_by(MaintenanceRecord.service_date.desc()) .limit(30) .all() ) return templates.TemplateResponse("truck_detail.html", { "request": request, "truck": truck, "rows": rows, "items": items, "history": history, }) @app.post("/api/trucks/{truck_id}") def update_truck(truck_id: int, engine_model: str = Form(""), db: Session = Depends(get_db)): truck = db.get(Truck, truck_id) if truck is None: raise HTTPException(404, "Truck não encontrado") truck.engine_model = engine_model or None db.commit() return RedirectResponse(f"/trucks/{truck_id}", status_code=303) @app.post("/api/trucks/{truck_id}/manutencao") def register_maintenance( truck_id: int, item_id: int = Form(...), service_date: str = Form(...), mileage_at_service: str = Form(""), hours_at_service: str = Form(""), parts_cost: str = Form("0"), labor_cost: str = Form("0"), downtime_days: str = Form("0"), notes: str = Form(""), db: Session = Depends(get_db), ): truck = db.get(Truck, truck_id) if truck is None: raise HTTPException(404, "Truck não encontrado") record = MaintenanceRecord( truck_id=truck_id, item_id=item_id, service_date=datetime.strptime(service_date, "%Y-%m-%d").replace(tzinfo=timezone.utc), mileage_at_service=float(mileage_at_service) if mileage_at_service else truck.odometer_miles, hours_at_service=float(hours_at_service) if hours_at_service else truck.engine_hours, parts_cost=float(parts_cost or 0), labor_cost=float(labor_cost or 0), downtime_days=float(downtime_days or 0), notes=notes or None, ) db.add(record) db.commit() return RedirectResponse(f"/trucks/{truck_id}", status_code=303) # ---------- Itens de manutenção ---------- @app.get("/manutencao") def maintenance_items_page(request: Request, db: Session = Depends(get_db)): items = db.query(MaintenanceItem).order_by(MaintenanceItem.name).all() return templates.TemplateResponse("manutencao.html", {"request": request, "items": items}) WARNING_FRACTION = 0.20 # "vencendo" comeca com 80% do intervalo consumido (20% restante) @app.post("/api/manutencao") def create_maintenance_item( name: str = Form(...), interval_miles: str = Form(""), interval_hours: str = Form(""), interval_days: str = Form(""), db: Session = Depends(get_db), ): miles = float(interval_miles) if interval_miles else None hours = float(interval_hours) if interval_hours else None days = int(interval_days) if interval_days else None db.add(MaintenanceItem( name=name, interval_miles=miles, interval_hours=hours, interval_days=days, warning_miles=round(miles * WARNING_FRACTION) if miles else 1500, warning_hours=round(hours * WARNING_FRACTION) if hours else 25, warning_days=round(days * WARNING_FRACTION) if days else 15, )) db.commit() return RedirectResponse("/manutencao", status_code=303) @app.post("/api/manutencao/{item_id}") def update_maintenance_item( item_id: int, name: str = Form(...), interval_miles: str = Form(""), interval_hours: str = Form(""), interval_days: str = Form(""), warning_miles: str = Form("1500"), warning_hours: str = Form("25"), warning_days: str = Form("15"), active: str = Form(""), db: Session = Depends(get_db), ): item = db.get(MaintenanceItem, item_id) if item is None: raise HTTPException(404, "Item não encontrado") item.name = name item.interval_miles = float(interval_miles) if interval_miles else None item.interval_hours = float(interval_hours) if interval_hours else None item.interval_days = int(interval_days) if interval_days else None item.warning_miles = float(warning_miles or 1500) item.warning_hours = float(warning_hours or 25) item.warning_days = int(warning_days or 15) item.active = bool(active) db.commit() return RedirectResponse("/manutencao", status_code=303) @app.post("/api/manutencao/{item_id}/excluir") def excluir_maintenance_item(item_id: int, db: Session = Depends(get_db)): item = db.get(MaintenanceItem, item_id) if item is None: raise HTTPException(404, "Item não encontrado") tem_registro = db.query(MaintenanceRecord).filter_by(item_id=item_id).first() is not None if tem_registro: raise HTTPException(400, "Não é possível excluir um item com histórico de manutenção registrado. Desative-o em vez disso.") db.delete(item) db.commit() return RedirectResponse("/manutencao", status_code=303) # ---------- Configurações ---------- @app.get("/configuracoes") def settings_page(request: Request): license_info = security.load_active_license() if license_info: license_info = dict(license_info) license_info["expires_display"] = ( "sem expiração" if license_info["expires_epoch"] == 0 else datetime.fromtimestamp(license_info["expires_epoch"], tz=timezone.utc).strftime("%d/%m/%Y") ) creds = security.load_geotab_credentials() or {} settings = security.load_settings() return templates.TemplateResponse("configuracoes.html", { "request": request, "license_info": license_info, "geotab_username": creds.get("username"), "sync_interval_minutes": settings["sync_interval_minutes"], }) @app.post("/api/configuracoes/sync-interval") def update_sync_interval(sync_interval_minutes: int = Form(...)): settings = security.load_settings() settings["sync_interval_minutes"] = max(5, sync_interval_minutes) security.save_settings(settings) reschedule(settings["sync_interval_minutes"]) return RedirectResponse("/configuracoes", status_code=303) @app.post("/api/configuracoes/reconectar") def reconectar_geotab(): security.CREDENTIALS_FILE.unlink(missing_ok=True) return RedirectResponse("/setup", status_code=303)