# ARCANUM SALIS (El Secreto de la Sal)
# Una aplicación de E-commerce completa en un solo archivo, diseñada por IA.
# Backend: FastAPI | DB: SQLAlchemy (SQLite) | Frontend: HTML/TailwindCSS | Pagos: MercadoPago
import os
import uvicorn
import mercadopago
from datetime import datetime
from typing import List, Optional, Dict, Any
# --- Imports de FastAPI y relacionados ---
from fastapi import FastAPI, Request, Depends, HTTPException, Form, Response
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.security import OAuth2PasswordRequestForm
from starlette.middleware.sessions import SessionMiddleware
from starlette.datastructures import URL
# --- Imports de SQLAlchemy (Base de Datos) ---
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, ForeignKey, Enum as DBEnum
from sqlalchemy.orm import sessionmaker, relationship, Session, declarative_base
# --- Imports de Pydantic (Validación de Modelos) ---
from pydantic import BaseModel
# --- Imports de Seguridad (Login de Admin) ---
from passlib.context import CryptContext
import secrets
# ======================================================================================
# 1. CONFIGURACIÓN INICIAL
# ======================================================================================
# --- Configuración de Seguridad ---
# En un entorno de producción real, estos valores DEBEN venir de variables de entorno.
# Para Hugging Face Spaces, puedes setearlos en los "Secrets" del Space.
# (Usamos valores default para la demo)
ADMIN_USER = os.environ.get("ADMIN_USER", "gran.maestro@arcanum.sal")
ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", "gadU_357") # GADU = Gran Arquitecto del Universo
SESSION_SECRET_KEY = os.environ.get("SESSION_SECRET_KEY", secrets.token_hex(32))
# --- Configuración de MercadoPago ---
# ¡¡¡IMPORTANTE!!! Reemplaza esto con tu ACCESS TOKEN de Producción o Test.
# Lo puedes obtener de tu cuenta de MercadoPago.
MP_ACCESS_TOKEN = os.environ.get("MP_ACCESS_TOKEN", "TEST-SAMPLE-TOKEN") # <--- ¡CAMBIAR ESTO!
if MP_ACCESS_TOKEN == "TEST-SAMPLE-TOKEN":
print("="*50)
print("ADVERTENCIA: Usando Access Token de MercadoPago de PRUEBA.")
print("El checkout NO funcionará. Debes setear tu propio MP_ACCESS_TOKEN.")
print("="*50)
# --- Configuración de la Base de Datos (SQLite) ---
# Se creará un archivo 'arcanum.db' en la raíz de tu Space.
DATABASE_URL = "sqlite:///./arcanum.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# --- Configuración de Hashing de Contraseñas ---
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# --- Inicialización de FastAPI ---
app = FastAPI(title="Arcanum Salis API", description="El secreto de las cajas saladas.")
# --- Inicialización del SDK de MercadoPago ---
mp_sdk = mercadopago.SDK(MP_ACCESS_TOKEN)
# ======================================================================================
# 2. MODELOS DE BASE DE DATOS (SQLAlchemy)
# ======================================================================================
class Admin(Base):
"""Modelo para el Administrador de 'La Logia'."""
__tablename__ = "admins"
id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
class Product(Base):
"""Modelo para nuestros 'Grados' (las cajas)."""
__tablename__ = "products"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True, nullable=False)
description = Column(String, nullable=False)
price = Column(Float, nullable=False)
image_url = Column(String, nullable=False) # Usaremos placeholders por ahora
ingredients = Column(String, nullable=False) # JSON-like string o separado por comas
class Order(Base):
"""Modelo para un pedido 'Ritual'."""
__tablename__ = "orders"
id = Column(Integer, primary_key=True, index=True)
customer_name = Column(String, nullable=False)
customer_email = Column(String, nullable=False)
customer_phone = Column(String)
total_amount = Column(Float, nullable=False)
status = Column(DBEnum("pending_payment", "paid", "preparing", "shipped", "failed", name="order_status_enum"), default="pending_payment")
created_at = Column(DateTime, default=datetime.utcnow)
mp_preference_id = Column(String, index=True)
items = relationship("OrderItem", back_populates="order")
class OrderItem(Base):
"""Modelo para un item dentro de un pedido."""
__tablename__ = "order_items"
id = Column(Integer, primary_key=True, index=True)
order_id = Column(Integer, ForeignKey("orders.id"))
product_id = Column(Integer, ForeignKey("products.id"))
quantity = Column(Integer, nullable=False)
unit_price = Column(Float, nullable=False) # Precio al momento de la compra
order = relationship("Order", back_populates="items")
product = relationship("Product")
# ======================================================================================
# 3. MODELOS DE DATOS (Pydantic) - Para validación
# ======================================================================================
# (No los usamos mucho en este enfoque de "templates", pero son buenas prácticas)
class ProductSchema(BaseModel):
id: int
name: str
description: str
price: float
image_url: str
ingredients: str
class Config:
orm_mode = True
class CartItem(BaseModel):
product: ProductSchema
quantity: int
class CartSchema(BaseModel):
items: List[CartItem]
total: float
# ======================================================================================
# 4. PLANTILLAS HTML (El Frontend completo como Strings de Python)
# ======================================================================================
# Esto es lo que hace que sea un solo archivo. Es una locura, pero cumple el requisito.
# Usamos TailwindCSS para el estilo.
# --- Símbolo Masónico Secreto (SVG) ---
# Un "Nivel de Plomada" estilizado. Sutil.
MASONIC_SYMBOL_SVG = """
"""
# --- Plantilla Base (Header y Footer) ---
HTML_BASE_TEMPLATE = """
{title} | Arcanum Salis
{content}
"""
# --- Contenido para la Página Principal (Home) ---
HTML_HOME_CONTENT = """
El Ritual de la "Saladeña"
Más que una caja. Es el fin del pan dulce y el comienzo de un nuevo ágape.
El secreto mejor guardado de las fiestas argentinas.
"""
# --- Página de Login de Admin ("La Logia") ---
HTML_ADMIN_LOGIN_CONTENT = """
{MASONIC_SYMBOL_SVG}
Acceso a La Logia
Solo para Maestros del Arcanum.
{error_message}
"""
# --- Página del Dashboard de Admin ---
HTML_ADMIN_DASHBOARD_CONTENT = """
Tablero de la Logia
Bienvenido, Gran Maestro. Aquí están los rituales pendientes y completados.
Pagado (En Cola)
{orders_paid}
En Preparación
{orders_preparing}
Enviado
{orders_shipped}
Pendiente / Fallido
{orders_failed}
"""
# --- Plantilla para una Card de Pedido en Admin ---
HTML_ADMIN_ORDER_CARD = """
Orden #{id}
${total_amount:,.0f}
{customer_name}
{customer_email}
{created_at}
{items_list}
"""
# ======================================================================================
# 5. LÓGICA DE NEGOCIO Y HELPERS
# ======================================================================================
# --- Dependencia de Base de Datos ---
def get_db():
"""Función de dependencia para obtener una sesión de DB."""
db = SessionLocal()
try:
yield db
finally:
db.close()
# --- Funciones de Seguridad ---
def verify_password(plain_password, hashed_password):
"""Verifica una contraseña contra su hash."""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password):
"""Genera un hash para una contraseña."""
return pwd_context.hash(password)
def get_admin_user(request: Request, db: Session = Depends(get_db)):
"""
Dependencia de FastAPI para proteger rutas de admin.
Verifica si 'admin_user' está en la sesión.
"""
username = request.session.get("admin_user")
if not username:
# Usamos una excepción que redirige al login
raise HTTPException(status_code=307, detail="Not authenticated", headers={"Location": "/admin/login"})
user = db.query(Admin).filter(Admin.username == username).first()
if user is None:
raise HTTPException(status_code=307, detail="User not found", headers={"Location": "/admin/login"})
return user
# --- Funciones del Carrito ---
def get_cart_data(request: Request, db: Session) -> Dict[str, Any]:
"""Obtiene y procesa los datos del carrito desde la sesión."""
cart_session = request.session.get("cart", {}) # cart_session = {product_id: quantity}
cart_items = []
total = 0.0
if not cart_session:
return {"items": [], "total": 0.0, "count": 0}
product_ids = cart_session.keys()
products = db.query(Product).filter(Product.id.in_(product_ids)).all()
products_dict = {str(p.id): p for p in products}
for product_id_str, quantity in cart_session.items():
product = products_dict.get(product_id_str)
if product:
subtotal = product.price * quantity
total += subtotal
cart_items.append({
"product_id": product.id,
"name": product.name,
"price": product.price,
"quantity": quantity,
"subtotal": subtotal,
"image_url": product.image_url
})
return {"items": cart_items, "total": total, "count": sum(cart_session.values())}
# --- Función de Renderizado ---
def render_template(
template: str,
context: dict,
request: Request,
db: Session = None
) -> HTMLResponse:
"""
Nuestro "motor de plantillas" súper simple.
Inserta el contenido en la base y añade valores globales.
"""
# Añadir valores globales
context["current_year"] = datetime.utcnow().year
context["MASONIC_SYMBOL_SVG"] = MASONIC_SYMBOL_SVG
# Añadir el contador del carrito
cart_count = 0
if db: # Si la DB está disponible, calcula el carrito real
cart_data = get_cart_data(request, db)
cart_count = cart_data["count"]
cart_badge = ""
if cart_count > 0:
cart_badge = f'{cart_count}'
# Formatea el contenido principal
content_html = template.format(**context)
# Inserta el contenido en la plantilla base
full_html = HTML_BASE_TEMPLATE.format(
title=context.get("title", "El Secreto de la Sal"),
cart_badge=cart_badge,
content=content_html,
current_year=datetime.utcnow().year
)
return HTMLResponse(content=full_html)
# ======================================================================================
# 6. EVENTOS DE STARTUP (Crear DB y Admin)
# ======================================================================================
@app.on_event("startup")
def on_startup():
"""Se ejecuta al iniciar la aplicación."""
print("Iniciando Arcanum Salis...")
# 1. Crear todas las tablas de la base de datos
Base.metadata.create_all(bind=engine)
print("Tablas de la base de datos verificadas/creadas.")
# 2. Crear el Admin por defecto (si no existe)
db = SessionLocal()
try:
admin = db.query(Admin).filter(Admin.username == ADMIN_USER).first()
if not admin:
hashed_pass = get_password_hash(ADMIN_PASSWORD)
new_admin = Admin(username=ADMIN_USER, hashed_password=hashed_pass)
db.add(new_admin)
db.commit()
print(f"Usuario admin '{ADMIN_USER}' creado.")
else:
print(f"Usuario admin '{ADMIN_USER}' ya existe.")
# 3. Crear los Productos (Cajas "Grados") si no existen
if db.query(Product).count() == 0:
print("Poblando la base de datos con los 3 Grados...")
# Placeholders de imágenes (Usando placehold.co, más sutil)
img1 = "https://placehold.co/600x400/374151/D4AF37?text=Grado+Aprendiz"
img2 = "https://placehold.co/600x400/374151/D4AF37?text=Grado+Compa%C3%B1ero"
img3 = "https://placehold.co/600x400/374151/D4AF37?text=Grado+Maestro"
# Ingredientes (como string separado por comas)
ing1 = "Salamín Picado Fino,Queso Mar del Plata,Maní Salado,Papas Pay,Grisines"
ing2 = "Jamón Crudo (Estilo Parma),Queso Brie,Aceitunas Rellenas,Almendras Ahumadas,Mini Focaccias con Romero"
ing3 = "Lomo Ahumado,Queso Azul,Pistachos Pelados,Tomates Secos en Oliva,Pan de Masa Madre,Cerveza Artesanal (Porrón)"
p1 = Product(
name="Grado Aprendiz",
description="El primer paso hacia el sabor. Clásicos infalibles para una picada que cumple. Simple, contundente y honesto.",
price=15000,
image_url=img1,
ingredients=ing1
)
p2 = Product(
name="Grado Compañero",
description="El equilibrio perfecto. Un ascenso en complejidad y texturas. Para paladares que buscan algo más.",
price=25000,
image_url=img2,
ingredients=ing2
)
p3 = Product(
name="Grado Maestro",
description="El conocimiento total del sabor. Lujo, sofisticación y combinaciones audaces. La cúspide del ágape.",
price=40000,
image_url=img3,
ingredients=ing3
)
db.add_all([p1, p2, p3])
db.commit()
print("Los 3 Grados han sido creados.")
finally:
db.close()
# ======================================================================================
# 7. MIDDLEWARE (Para manejar sesiones)
# ======================================================================================
# Añadimos el middleware de sesión a FastAPI
app.add_middleware(SessionMiddleware, secret_key=SESSION_SECRET_KEY)
# ======================================================================================
# 8. ENDPOINTS DEL FRONTEND (Lado Usuario)
# ======================================================================================
@app.get("/", response_class=HTMLResponse)
async def get_home(request: Request, db: Session = Depends(get_db)):
"""Muestra la página principal con los productos."""
products = db.query(Product).all()
product_cards_html = ""
for prod in products:
# Formatear lista de ingredientes
ingredients_list_html = "".join([f"
{item}
" for item in prod.ingredients.split(',')])
product_cards_html += HTML_PRODUCT_CARD.format(
id=prod.id,
name=prod.name,
description=prod.description,
price=prod.price,
image_url=prod.image_url,
ingredients_list=ingredients_list_html
)
context = {
"title": "El Ritual de la Saladeña",
"product_cards": product_cards_html
}
return render_template(HTML_HOME_CONTENT, context, request, db)
@app.get("/cart", response_class=HTMLResponse)
async def get_cart(request: Request, db: Session = Depends(get_db)):
"""Muestra la página del carrito de compras."""
cart_data = get_cart_data(request, db)
if not cart_data["items"]:
return render_template(HTML_CART_EMPTY, {"title": "Carrito Vacío"}, request, db)
cart_items_html = ""
for item in cart_data["items"]:
cart_items_html += HTML_CART_ITEM_ROW.format(**item)
context = {
"title": "Tu Ritual de Compra",
"cart_items_html": cart_items_html,
"total": cart_data["total"],
"checkout_button": HTML_CHECKOUT_BUTTON # Añadimos el formulario de checkout
}
return render_template(HTML_CART_CONTENT, context, request, db)
@app.post("/cart/add")
async def post_add_to_cart(request: Request, product_id: int = Form(...)):
"""Añade un producto al carrito en la sesión."""
cart = request.session.get("cart", {}) # { "1": 1, "2": 3 }
# Usamos strings para las claves de sesión, es más seguro
product_id_str = str(product_id)
cart[product_id_str] = cart.get(product_id_str, 0) + 1
request.session["cart"] = cart
# Redirigimos de vuelta al carrito
return RedirectResponse(url="/cart", status_code=303)
@app.post("/cart/remove")
async def post_remove_from_cart(request: Request, product_id: int = Form(...)):
"""Reduce o elimina un producto del carrito en la sesión."""
cart = request.session.get("cart", {})
product_id_str = str(product_id)
if product_id_str in cart:
cart[product_id_str] -= 1
if cart[product_id_str] <= 0:
del cart[product_id_str] # Elimina si la cantidad es 0 o menos
request.session["cart"] = cart
return RedirectResponse(url="/cart", status_code=303)
# ======================================================================================
# 9. ENDPOINTS DE PAGO (Checkout y MercadoPago)
# ======================================================================================
@app.post("/checkout")
async def post_checkout(
request: Request,
db: Session = Depends(get_db),
customer_name: str = Form(...),
customer_email: str = Form(...)
):
"""
1. Obtiene el carrito.
2. Crea una Orden en la DB como 'pending_payment'.
3. Crea una Preferencia de Pago en MercadoPago.
4. Redirige al usuario al checkout de MP.
"""
cart_data = get_cart_data(request, db)
if not cart_data["items"]:
return RedirectResponse(url="/cart", status_code=303)
# 1. Crear la Orden en la DB
new_order = Order(
customer_name=customer_name,
customer_email=customer_email,
total_amount=cart_data["total"],
status="pending_payment"
)
db.add(new_order)
db.commit() # Hacemos commit para obtener el new_order.id
mp_items = []
for item in cart_data["items"]:
# Añadir items a la Orden
order_item_db = OrderItem(
order_id=new_order.id,
product_id=item["product_id"],
quantity=item["quantity"],
unit_price=item["price"]
)
db.add(order_item_db)
# Preparar items para MP
mp_items.append({
"id": str(item["product_id"]),
"title": item["name"],
"quantity": item["quantity"],
"unit_price": item["price"],
"currency_id": "ARS" # Moneda Argentina
})
db.commit() # Guardamos los OrderItems
# 2. Crear Preferencia de MercadoPago
# Usamos str(request.base_url) para construir URLs absolutas (requerido por MP)
base_url = str(request.base_url)
preference_data = {
"items": mp_items,
"payer": {
"name": customer_name,
"email": customer_email,
},
"back_urls": {
"success": f"{base_url}payment/success",
"failure": f"{base_url}payment/failure",
"pending": f"{base_url}payment/failure" # Tratamos pendiente como fallo por simpleza
},
"auto_return": "approved", # Redirige automáticamente al aprobar
"notification_url": f"{base_url}payment/webhook",
"external_reference": str(new_order.id), # ¡Clave! Vinculamos la Orden con MP
"metadata": {
"order_id": new_order.id # Metadatos extra
}
}
try:
preference_response = mp_sdk.preference().create(preference_data)
preference = preference_response["response"]
# Guardamos el ID de la preferencia en nuestra orden
new_order.mp_preference_id = preference["id"]
db.commit()
# Limpiamos el carrito
request.session["cart"] = {}
# Guardamos la orden en la sesión para mostrarla en la página de éxito
request.session["last_order_id"] = new_order.id
# 3. Redirigir al checkout
# Usamos el init_point del sandbox si estamos en modo test
checkout_url = preference["init_point"]
if "sandbox_init_point" in preference:
checkout_url = preference["sandbox_init_point"]
return RedirectResponse(url=checkout_url, status_code=303)
except Exception as e:
print(f"Error creando preferencia de MercadoPago: {e}")
# Revertir la orden
new_order.status = "failed"
db.commit()
# Idealmente, mostrar un mensaje de error al usuario
return RedirectResponse(url="/cart?error=mp_failed", status_code=303)
@app.get("/payment/success", response_class=HTMLResponse)
async def get_payment_success(request: Request, db: Session = Depends(get_db)):
"""Página de éxito. Se muestra después de un pago aprobado."""
# Obtenemos la orden de la sesión (no del query param, por seguridad)
order_id = request.session.get("last_order_id")
if not order_id:
return RedirectResponse(url="/")
order = db.query(Order).filter(Order.id == order_id).first()
if not order:
return RedirectResponse(url="/")
# Verificamos los parámetros que SÍ envía MP en la URL
payment_status = request.query_params.get("status")
external_reference = request.query_params.get("external_reference")
# Doble chequeo
if payment_status == "approved" and external_reference == str(order.id):
# Marcamos la orden como pagada (si el webhook no lo hizo ya)
if order.status == "pending_payment":
order.status = "paid"
db.commit()
else:
# Algo raro pasó, el usuario no debería estar aquí
return RedirectResponse(url="/")
context = {
"title": "Ritual Completado",
"order_id": order.id,
"email": order.customer_email
}
# Limpiamos el ID de la sesión
request.session["last_order_id"] = None
return render_template(HTML_ORDER_SUCCESS_CONTENT, context, request, db)
@app.get("/payment/failure", response_class=HTMLResponse)
async def get_payment_failure(request: Request, db: Session = Depends(get_db)):
"""Página de fallo. Se muestra si el pago es rechazado."""
order_id = request.session.get("last_order_id")
if order_id:
order = db.query(Order).filter(Order.id == order_id).first()
if order and order.status == "pending_payment":
order.status = "failed"
db.commit()
# Limpiamos el ID de la sesión
request.session["last_order_id"] = None
context = {"title": "Ritual Interrumpido"}
return render_template(HTML_ORDER_FAILURE_CONTENT, context, request)
@app.post("/payment/webhook")
async def post_payment_webhook(request: Request, db: Session = Depends(get_db)):
"""
Webhook de MercadoPago. Se ejecuta en segundo plano.
Es la forma MÁS SEGURA de confirmar un pago.
"""
try:
data = await request.json()
if data.get("type") == "payment":
payment_id = data.get("data", {}).get("id")
if not payment_id:
return Response(status_code=400)
# Obtenemos la info del pago desde MP
payment_info_response = mp_sdk.payment().get(payment_id)
payment_info = payment_info_response["response"]
order_id = payment_info.get("external_reference")
status = payment_info.get("status")
if order_id:
order = db.query(Order).filter(Order.id == int(order_id)).first()
if order:
if status == "approved" and order.status != "paid":
order.status = "paid"
print(f"[Webhook] Orden {order_id} marcada como PAGADA.")
elif status in ["rejected", "cancelled"] and order.status == "pending_payment":
order.status = "failed"
print(f"[Webhook] Orden {order_id} marcada como FALLIDA.")
db.commit()
return Response(status_code=200) # Siempre responder 200 a MP
except Exception as e:
print(f"Error en Webhook: {e}")
return Response(status_code=500)
# ======================================================================================
# 10. ENDPOINTS DE ADMINISTRACIÓN ("La Logia")
# ======================================================================================
@app.get("/admin", response_class=RedirectResponse)
@app.get("/admin/login", response_class=HTMLResponse)
async def get_admin_login(request: Request, error: Optional[str] = None):
"""Muestra la página de login de admin."""
error_message = ""
if error:
error_message = '
Palabra de Paso o Email incorrectos.
'
context = {
"title": "Acceso a La Logia",
"error_message": error_message
}
# Usamos la plantilla base, pero el contenido de login es "full page"
# así que el {content} será todo el HTML.
return HTMLResponse(content=HTML_ADMIN_LOGIN_CONTENT.format(**context))
@app.post("/admin/login", response_class=RedirectResponse)
async def post_admin_login(
request: Request,
db: Session = Depends(get_db),
form_data: OAuth2PasswordRequestForm = Depends()
):
"""Procesa el formulario de login de admin."""
user = db.query(Admin).filter(Admin.username == form_data.username).first()
if not user or not verify_password(form_data.password, user.hashed_password):
# Redirige a la misma página con un mensaje de error
return RedirectResponse(url="/admin/login?error=1", status_code=303)
# Éxito: Guardamos al admin en la sesión
request.session["admin_user"] = user.username
return RedirectResponse(url="/admin/dashboard", status_code=303)
@app.post("/admin/logout", response_class=RedirectResponse)
async def post_admin_logout(request: Request):
"""Cierra la sesión del admin."""
request.session.pop("admin_user", None)
return RedirectResponse(url="/admin/login", status_code=303)
@app.get("/admin/dashboard", response_class=HTMLResponse)
async def get_admin_dashboard(
request: Request,
db: Session = Depends(get_db),
admin: Admin = Depends(get_admin_user) # ¡Ruta Protegida!
):
"""Muestra el dashboard de admin con todas las órdenes."""
# Obtenemos órdenes y las separamos por estado
orders = db.query(Order).order_by(Order.created_at.desc()).all()
orders_by_status = {
"paid": [],
"preparing": [],
"shipped": [],
"failed": [] # Incluye pending_payment y failed
}
for order in orders:
if order.status == "paid":
orders_by_status["paid"].append(order)
elif order.status == "preparing":
orders_by_status["preparing"].append(order)
elif order.status == "shipped":
orders_by_status["shipped"].append(order)
else: # "pending_payment" o "failed"
orders_by_status["failed"].append(order)
# --- Función interna para renderizar las cards de órdenes ---
def render_order_cards(order_list: List[Order]) -> str:
html = ""
for order in order_list:
items_html = ""
for item in order.items:
product_name = db.query(Product.name).filter(Product.id == item.product_id).scalar() or "Producto Eliminado"
items_html += f"
{item.quantity}x {product_name}
"
html += HTML_ADMIN_ORDER_CARD.format(
id=order.id,
total_amount=order.total_amount,
customer_name=order.customer_name,
customer_email=order.customer_email,
created_at=order.created_at.strftime("%d/%m/%y %H:%M"),
items_list=items_html,
selected_paid='selected' if order.status == 'paid' else '',
selected_preparing='selected' if order.status == 'preparing' else '',
selected_shipped='selected' if order.status == 'shipped' else '',
selected_failed='selected' if order.status in ['failed', 'pending_payment'] else '',
)
if not html:
return '
Nada por aquí.
'
return html
# --- Fin de la función interna ---
context = {
"title": "Tablero de la Logia",
"orders_paid": render_order_cards(orders_by_status["paid"]),
"orders_preparing": render_order_cards(orders_by_status["preparing"]),
"orders_shipped": render_order_cards(orders_by_status["shipped"]),
"orders_failed": render_order_cards(orders_by_status["failed"]),
}
return render_template(HTML_ADMIN_DASHBOARD_CONTENT, context, request, db)
@app.post("/admin/order/update", response_class=RedirectResponse)
async def post_update_order_status(
request: Request,
db: Session = Depends(get_db),
admin: Admin = Depends(get_admin_user), # ¡Ruta Protegida!
order_id: int = Form(...),
new_status: str = Form(...)
):
"""Actualiza el estado de una orden."""
order = db.query(Order).filter(Order.id == order_id).first()
if order:
# Validar que el estado sea uno de los permitidos
allowed_statuses = [s.value for s in Order.status.type.enums]
if new_status in allowed_statuses:
order.status = new_status
db.commit()
print(f"[Admin] Orden {order_id} actualizada a {new_status} por {admin.username}")
else:
print(f"[Admin] Intento de actualizar a estado inválido: {new_status}")
return RedirectResponse(url="/admin/dashboard", status_code=303)
# ======================================================================================
# 11. INICIAR LA APLICACIÓN (Para correr localmente)
# ======================================================================================
if __name__ == "__main__":
print("Iniciando servidor Uvicorn localmente en http://127.0.0.1:8000")
print(f"Admin User: {ADMIN_USER}")
print(f"Admin Pass: {ADMIN_PASSWORD}")
uvicorn.run(app, host="127.0.0.1", port=8000)