modelo-fastapi/main.py

"""Encurtador de links em FastAPI, pronto para publicar na Dado.

Rodar local:  uvicorn main:app --reload   (ou: python main.py)
"""

import html
import os
import platform
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path

import fastapi
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, Response
from pydantic import BaseModel

from links import FORMATO_CODIGO, ErroDeValidacao, criar_link, listar_links, registrar_clique

# Variável de ambiente de exemplo: na Dado, ela se configura no painel do app.
MENSAGEM = os.environ.get("MENSAGEM_BOAS_VINDAS") or "Defina a variável MENSAGEM_BOAS_VINDAS para trocar este texto."
INICIO = datetime.now(timezone.utc)
# Brasília não tem horário de verão desde 2019: UTC-3 o ano todo.
_inicio_brasilia = INICIO.astimezone(timezone(timedelta(hours=-3)))
INICIO_FORMATADO = _inicio_brasilia.strftime("%d/%m/%Y às %H:%M") + " (horário de Brasília)"

PASTA_PAGINAS = Path(__file__).parent / "paginas"

# A página é montada uma vez só: nada nela muda enquanto o processo vive.
PAGINA_INICIAL = (
    (PASTA_PAGINAS / "index.html")
    .read_text(encoding="utf-8")
    .replace("{{RUNTIME}}", html.escape(f"FastAPI {fastapi.__version__} · Python {platform.python_version()}"))
    .replace("{{INICIO_ISO}}", INICIO.isoformat(timespec="seconds").replace("+00:00", "Z"))
    .replace("{{INICIO}}", html.escape(INICIO_FORMATADO))
    .replace("{{MENSAGEM}}", html.escape(MENSAGEM))
)
PAGINA_NAO_ENCONTRADA = (PASTA_PAGINAS / "nao-encontrado.html").read_text(encoding="utf-8")

# Freio simples contra abuso: no máximo 30 links novos a cada 10 minutos por IP.
LIMITE_POR_IP = 30
JANELA_SEGUNDOS = 10 * 60
_tentativas: dict[str, tuple[int, float]] = {}


def passou_do_limite(ip: str) -> bool:
    agora = time.monotonic()
    if len(_tentativas) > 10_000:
        _tentativas.clear()
    total, expira_em = _tentativas.get(ip, (0, 0.0))
    if expira_em <= agora:
        _tentativas[ip] = (1, agora + JANELA_SEGUNDOS)
        return False
    _tentativas[ip] = (total + 1, expira_em)
    return total + 1 > LIMITE_POR_IP


app = FastAPI(title="Encurtador de links", description="Encurta links e conta os cliques.")

CORS = {
    "Access-Control-Allow-Origin": "*",
    "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
    "Access-Control-Allow-Headers": "Content-Type",
}


@app.middleware("http")
async def cors_so_na_api(request: Request, call_next):
    """CORS aberto só na API: qualquer site (ou outro app seu) pode chamar /api/*."""
    if not request.url.path.startswith("/api/"):
        return await call_next(request)
    if request.method == "OPTIONS":
        resposta = Response(status_code=204)
    else:
        resposta = await call_next(request)
    resposta.headers.update(CORS)
    return resposta


def endereco_base(request: Request) -> str:
    # O app roda atrás de um proxy HTTPS, que avisa o protocolo e o host originais.
    protocolo = request.headers.get("x-forwarded-proto", request.url.scheme).split(",")[0].strip()
    host = request.headers.get("x-forwarded-host") or request.headers.get("host") or request.url.netloc
    return f"{protocolo}://{host}"


def para_json(link: dict, request: Request) -> dict:
    return {
        "codigo": link["codigo"],
        "url": link["url"],
        "curto": f"{endereco_base(request)}/{link['codigo']}",
        "cliques": link["cliques"],
        "criadoEm": link["criadoEm"],
    }


class NovoLink(BaseModel):
    url: str | None = None


@app.exception_handler(RequestValidationError)
async def corpo_invalido(request: Request, erro: RequestValidationError):
    return JSONResponse(
        status_code=400,
        content={"erro": 'Corpo inválido. Envie JSON no formato {"url": "https://..."}.'},
    )


@app.get("/", response_class=HTMLResponse, include_in_schema=False)
def pagina_inicial():
    return PAGINA_INICIAL


@app.get("/api/links", summary="Lista os últimos 20 links")
def listar(request: Request):
    return JSONResponse(
        {"links": [para_json(link, request) for link in listar_links()]},
        headers={"Cache-Control": "no-store"},
    )


@app.post("/api/links", status_code=201, summary="Encurta um link")
def criar(novo: NovoLink, request: Request):
    ip = request.headers.get("cf-connecting-ip") or (request.client.host if request.client else "?")
    if passou_do_limite(ip):
        return JSONResponse(
            status_code=429,
            content={"erro": "Muitos links em pouco tempo. Espere alguns minutos e tente de novo."},
        )
    try:
        link = criar_link(novo.url)
    except ErroDeValidacao as erro:
        return JSONResponse(status_code=400, content={"erro": str(erro)})
    return JSONResponse(status_code=201, content=para_json(link, request))


# O redirecionamento é feito pelo servidor: GET /<código> → 302 para o link.
# Fica por último: as rotas fixas acima têm prioridade.
@app.get("/{codigo}", include_in_schema=False)
def redirecionar(codigo: str):
    link = registrar_clique(codigo) if FORMATO_CODIGO.match(codigo) else None
    if link is None:
        return HTMLResponse(PAGINA_NAO_ENCONTRADA, status_code=404)
    return RedirectResponse(link["url"], status_code=302, headers={"Cache-Control": "no-store"})


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", "8000")))