import logging import time from urllib.parse import quote from fastapi import FastAPI, Request from fastapi.responses import JSONResponse, RedirectResponse from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.gzip import GZipMiddleware from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from starlette.middleware.sessions import SessionMiddleware from starlette.middleware.trustedhost import TrustedHostMiddleware from app.auth import obter_sessao_usuario from starlette.middleware.base import BaseHTTPMiddleware from app.time_utils import agora_local from app.db import init_db, jobs_usam_banco, storage_usa_banco from app.routes import router from app.logging_config import setup_logging from app.settings import ( APP_ALLOWED_HOSTS, APP_ENABLE_AUTH, APP_AUTH_SESSION_VALIDATION_SECONDS, APP_CORS_ORIGINS, APP_ENV, APP_MODE, APP_SESSION_COOKIE_NAME, APP_SESSION_SECRET, APP_STORAGE_BACKEND, BASE_DIR, CAMINHO_MODELO_CARTAO, FRONTEND_PUBLIC_URL, HOST, KEEPEDU_API_KEY, KEEPEDU_LOGIN_SCHOOL, KEEPEDU_READING_BASE_URL, PORT, PASTA_BACKUPS, PASTA_LOGS, REPO_DIR, app_modo_api, ambiente_local, app_session_same_site, garantir_pastas, gestor_home_url_para_sessao, host_frontend_publico, nome_banco_para_escola, nome_escola_para_sessao, tenant_debug_para_sessao, url_gestor_para_escola, ) from app.sso import sso_habilitado from app.tenant import apply_tenant_from_session from omr_reader import pyzbar_decode, zxingcpp garantir_pastas() setup_logging() logger = logging.getLogger("omrcheck.main") try: DEPLOY_VERSION = (REPO_DIR / ".deploy-version").read_text(encoding="utf-8").strip() except OSError: DEPLOY_VERSION = "local" def _normalizar_caminho_auth(caminho: str) -> str: caminho_limpo = str(caminho or "").strip() if caminho_limpo != "/": caminho_limpo = caminho_limpo.rstrip("/") return caminho_limpo def _requisicao_espera_json(request: Request, caminho: str) -> bool: # No modo API o frontend React trata 401; não redirecionar HTML. if app_modo_api() and caminho not in {"/login", "/sso", "/logout"}: return True if request.headers.get("x-requested-with", "").lower() == "xmlhttprequest": return True accept = str(request.headers.get("accept") or "").lower() if "application/json" in accept and "text/html" not in accept.split(",")[0]: return True if caminho.endswith("-ajax") or "/ajax/" in caminho: return True if request.method.upper() != "GET": if caminho.startswith(( "/reprocessar-", "/deletar-", "/importar-", "/correcao/enviar-", )): return True prefixos_api = ( "/api/", "/status-processamento/", "/leituras/", "/correcoes/", ) return any(caminho.startswith(prefixo) for prefixo in prefixos_api) def _url_login_gestor(request: Request) -> str: school = str( request.session.get("school") or request.session.get("keepedu_school") or KEEPEDU_LOGIN_SCHOOL or "proposito" ).strip().lower() login_sessao = str(request.session.get("gestor_login_url") or "").strip() if login_sessao and f"{school}.keepedu.com.br" in login_sessao.lower(): return login_sessao return f"{url_gestor_para_escola(school)}login" def _resposta_nao_autenticado(request: Request, caminho: str): if _requisicao_espera_json(request, caminho): return JSONResponse( status_code=401, content={ "status": "erro", "codigo": "sessao_expirada", "mensagem": "Sessão expirada. Faça login no gestor KeepEdu novamente.", }, ) if sso_habilitado() and APP_ENABLE_AUTH: destino = f"{KEEPEDU_READING_BASE_URL}{caminho}" return RedirectResponse( url=f"{_url_login_gestor(request)}?next={quote('reading-sso')}", status_code=303, ) return RedirectResponse( url=f"/login?next={quote(caminho)}", status_code=303, ) def _renovar_sessao(request: Request) -> None: request.session["_atividade"] = agora_local().isoformat() def _sessao_precisa_validacao_no_banco(sessao: dict) -> bool: try: ultima_validacao = float(sessao.get("_sessao_validada_em") or 0) except (TypeError, ValueError): return True return (time.time() - ultima_validacao) >= APP_AUTH_SESSION_VALIDATION_SECONDS # Telas HTML legadas — no modo API redirecionam para o React (sem exigir sessão aqui). _PREFIXOS_UI_LEGADA = ( "/painel", "/kanban", "/historico", "/bernoulli", "/avaliacao", "/validacao", "/correcao", "/processando", "/processamento-react", "/correcao-react", "/validacao-react", "/kanban-react", "/historico-react", "/bernoulli-react", "/painel-react", ) def _caminho_ui_legada(caminho: str) -> bool: """Telas HTML/React — não inclui ajax, downloads nem arquivos.""" if any( marcador in caminho for marcador in ( "-ajax", "/ajax/", "/download", "/visualizar/", "/folha-keepedu", "/api/", ) ): return False return any( caminho == prefixo or caminho.startswith(f"{prefixo}/") for prefixo in _PREFIXOS_UI_LEGADA ) class AutenticacaoMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): caminho = _normalizar_caminho_auth(request.url.path) rotas_livres = { "/", "/login", "/logout", "/sso", "/healthz", "/favicon.ico", } if ( not APP_ENABLE_AUTH or caminho in rotas_livres or caminho.startswith("/static") or caminho.startswith("/assets") or (app_modo_api() and request.method.upper() == "GET" and _caminho_ui_legada(caminho)) ): return await call_next(request) sessao = request.session apply_tenant_from_session(sessao) usuario_id = sessao.get("usuario_id") session_id = sessao.get("session_id") if not usuario_id or not session_id: return _resposta_nao_autenticado(request, caminho) if _sessao_precisa_validacao_no_banco(sessao): try: sessao_banco = obter_sessao_usuario(usuario_id) except Exception: # Falha transitória no MySQL não deve derrubar a sessão do usuário. logger.exception( "Falha ao validar sessao ativa usuario_id=%s caminho=%s", usuario_id, caminho, ) else: if not sessao_banco or sessao_banco != session_id: request.session.clear() return _resposta_nao_autenticado(request, caminho) sessao["_sessao_validada_em"] = time.time() _renovar_sessao(request) return await call_next(request) app = FastAPI(title="Reading KeepEdu API" if app_modo_api() else "OMRCheck Web") @app.on_event("startup") async def _configurar_capacidade_threadpool() -> None: """Permite mais I/O paralelo (galeria + Kanban + histórico) sem travar o loop.""" try: import anyio limiter = anyio.to_thread.current_default_thread_limiter() # Padrão do anyio costuma ser 40; sobe para caber miniaturas + board juntos. limiter.total_tokens = max(int(getattr(limiter, "total_tokens", 40) or 40), 64) except Exception: logger.exception("Falha ao ajustar capacidade do threadpool") class ApiOnlyMiddleware(BaseHTTPMiddleware): """Garante que este host se comporte como API para o frontend React.""" async def dispatch(self, request: Request, call_next): if not app_modo_api(): return await call_next(request) caminho = _normalizar_caminho_auth(request.url.path) if caminho == "/": return JSONResponse( { "status": "ok", "servico": "reading-keepedu-api", "app_mode": APP_MODE, "frontend": FRONTEND_PUBLIC_URL, "healthz": "/healthz", } ) if request.method.upper() == "GET" and _caminho_ui_legada(caminho): host_req = (request.url.hostname or request.headers.get("host") or "").split(":")[0].lower() host_front = host_frontend_publico() if not host_front or host_front == host_req: return JSONResponse( status_code=503, content={ "status": "erro", "codigo": "frontend_nao_configurado", "mensagem": ( "FRONTEND_PUBLIC_URL deve apontar para o React " "(ex.: https://develop.frontend-reading-keepedu.pages.dev), " "não para esta API." ), "frontend": FRONTEND_PUBLIC_URL, }, ) return RedirectResponse( url=f"{FRONTEND_PUBLIC_URL}{request.url.path}", status_code=307, ) return await call_next(request) app.add_middleware(ApiOnlyMiddleware) app.add_middleware(AutenticacaoMiddleware) app.add_middleware( GZipMiddleware, minimum_size=1024 ) _SESSION_SAME_SITE = app_session_same_site() app.add_middleware( SessionMiddleware, secret_key=APP_SESSION_SECRET, session_cookie=APP_SESSION_COOKIE_NAME, same_site=_SESSION_SAME_SITE, # SameSite=None exige Secure; em produção local XAMPP use APP_SESSION_SAME_SITE=lax https_only=_SESSION_SAME_SITE == "none" or not ambiente_local(), max_age=60 * 60 * 5 ) # HOSTS if APP_ALLOWED_HOSTS: app.add_middleware( TrustedHostMiddleware, allowed_hosts=APP_ALLOWED_HOSTS ) if APP_CORS_ORIGINS: app.add_middleware( CORSMiddleware, allow_origins=APP_CORS_ORIGINS, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) app.mount( "/static", StaticFiles(directory=str(BASE_DIR / "web" / "static")), name="static" ) app.mount( "/assets", StaticFiles(directory=str(BASE_DIR / "assets")), name="assets" ) app.state.templates = Jinja2Templates( directory=str(BASE_DIR / "web" / "templates") ) app.state.templates.env.globals["gestor_home_url"] = gestor_home_url_para_sessao app.state.templates.env.globals["nome_escola"] = nome_escola_para_sessao app.state.templates.env.globals["nome_banco_para_escola"] = nome_banco_para_escola app.state.templates.env.globals["tenant_debug"] = tenant_debug_para_sessao @app.middleware("http") async def adicionar_headers_de_seguranca(request: Request, call_next): resposta = await call_next(request) resposta.headers.setdefault("X-Frame-Options", "DENY") resposta.headers.setdefault("X-Content-Type-Options", "nosniff") resposta.headers.setdefault("Referrer-Policy", "strict-origin-when-cross-origin") resposta.headers.setdefault( "Permissions-Policy", "camera=(), microphone=(), geolocation=()" ) return resposta @app.get("/healthz") async def healthcheck(): return { "status": "ok", "deploy_version": DEPLOY_VERSION, "app_env": APP_ENV, "app_mode": APP_MODE, "frontend_public_url": FRONTEND_PUBLIC_URL, "storage_backend": APP_STORAGE_BACKEND, "auth_enabled": APP_ENABLE_AUTH, "modelo_cartao_existe": CAMINHO_MODELO_CARTAO.exists(), "barcode_decoder_available": pyzbar_decode is not None or zxingcpp is not None, "keepedu_api_configurada": bool(str(KEEPEDU_API_KEY).strip()), "jobs_store_existe": jobs_usam_banco(), "logs_dir": str(PASTA_LOGS), "backups_dir": str(PASTA_BACKUPS), } @app.on_event("startup") async def registrar_inicio(): init_db() logger.info( "Aplicacao iniciada | env=%s | mode=%s | auth=%s | storage=%s | frontend=%s | jobs=%s", APP_ENV, APP_MODE, APP_ENABLE_AUTH, APP_STORAGE_BACKEND, FRONTEND_PUBLIC_URL, jobs_usam_banco(), ) app.include_router(router) if __name__ == "__main__": import uvicorn uvicorn.run("main:app", host=HOST, port=PORT, reload=False)