"""
Noir & Grain rezervasyon demo API'si + admin masa paneli.
Şablon vitrini için örnek uçtan uca akış:
- POST /api/reservations → talep oluştur (masa otomatik atanır) + örnek SMTP bildirimi
- GET /api/reservations?date= → gün listesi
- GET /api/reservations/tables → date+time için masa durumu (dolu/boş)
- PATCH /api/reservations/{id} → durum güncelle (confirmed/cancelled/seated)
- GET /admin/reservations → tek dosyalık karanlık admin paneli (masa haritası)
Depolama: DATA_DIR/reservations.json (HF Spaces'te kalıcı /data'ya işaret edin).
Mail: SMTP_HOST/USER/PASSWORD tanımlıysa gönderir; değilse sadece loglar (örnek entegrasyon).
"""
from __future__ import annotations
import json
import os
import smtplib
import threading
import uuid
from datetime import date as date_type
from datetime import datetime, timezone
from email.mime.text import MIMEText
from pathlib import Path
from typing import Literal
from fastapi import APIRouter, HTTPException, Query
from fastapi.responses import HTMLResponse
from pydantic import BaseModel, Field, field_validator
from ..services.logging_config import get_logger
logger = get_logger(__name__)
router = APIRouter(tags=["reservations"])
DATA_DIR = Path(os.getenv("DATA_DIR", "data"))
STORE_PATH = DATA_DIR / "reservations.json"
TOTAL_TABLES = int(os.getenv("RESTAURANT_TOTAL_TABLES", "12"))
_lock = threading.Lock()
# ── Depolama ─────────────────────────────────────────────────────────
def _load() -> list[dict]:
if not STORE_PATH.exists():
return []
try:
return json.loads(STORE_PATH.read_text(encoding="utf-8"))
except json.JSONDecodeError:
logger.error("reservations.json bozuk, boş listeyle devam ediliyor")
return []
def _save(items: list[dict]) -> None:
DATA_DIR.mkdir(parents=True, exist_ok=True)
STORE_PATH.write_text(json.dumps(items, ensure_ascii=False, indent=2), encoding="utf-8")
# ── Şemalar ──────────────────────────────────────────────────────────
class ReservationIn(BaseModel):
name: str = Field(min_length=2, max_length=80)
phone: str = Field(min_length=10, max_length=20)
date: str # YYYY-MM-DD
time: str # HH:MM
guests: int = Field(ge=1, le=12)
note: str = Field(default="", max_length=500)
@field_validator("date")
@classmethod
def date_not_past(cls, v: str) -> str:
try:
picked = date_type.fromisoformat(v)
except ValueError as exc:
raise ValueError("Tarih YYYY-MM-DD formatında olmalı") from exc
if picked < date_type.today():
raise ValueError("Geçmiş bir tarih seçilemez")
return v
@field_validator("time")
@classmethod
def time_format(cls, v: str) -> str:
try:
datetime.strptime(v, "%H:%M")
except ValueError as exc:
raise ValueError("Saat HH:MM formatında olmalı") from exc
return v
class Reservation(ReservationIn):
id: str
table: int
status: Literal["pending", "confirmed", "seated", "cancelled"]
created_at: str
class StatusUpdate(BaseModel):
status: Literal["pending", "confirmed", "seated", "cancelled"]
# ── Örnek mail bildirimi ─────────────────────────────────────────────
def _notify_by_mail(res: dict) -> None:
"""SMTP env değişkenleri tanımlıysa restorana bildirim maili atar.
Tanımlı değilse sessizce loglar — demo ortamında kırılma olmaz.
"""
host = os.getenv("SMTP_HOST")
user = os.getenv("SMTP_USER")
password = os.getenv("SMTP_PASSWORD")
to_addr = os.getenv("RESERVATION_MAIL_TO", user or "")
if not (host and user and password and to_addr):
logger.info("SMTP yapılandırılmamış — mail bildirimi atlandı (demo modu)")
return
body = (
f"Yeni rezervasyon talebi\n\n"
f"Ad: {res['name']}\nTelefon: {res['phone']}\n"
f"Tarih: {res['date']} Saat: {res['time']}\n"
f"Kişi: {res['guests']} Masa: {res['table']}\n"
f"Not: {res['note'] or '-'}"
)
msg = MIMEText(body, _charset="utf-8")
msg["Subject"] = f"Rezervasyon: {res['name']} — {res['date']} {res['time']}"
msg["From"] = os.getenv("SMTP_FROM", user)
msg["To"] = to_addr
try:
with smtplib.SMTP(host, int(os.getenv("SMTP_PORT", "587")), timeout=10) as smtp:
smtp.starttls()
smtp.login(user, password)
smtp.send_message(msg)
logger.info("Rezervasyon maili gönderildi: %s", res["id"])
except Exception as exc: # demo: mail hatası rezervasyonu düşürmesin
logger.error("Mail gönderilemedi: %s", exc)
# ── Yardımcılar ──────────────────────────────────────────────────────
def _occupied_tables(items: list[dict], date: str, time: str) -> dict[int, dict]:
"""Aynı gün + saat dilimindeki iptal edilmemiş rezervasyonların masaları."""
return {
r["table"]: r
for r in items
if r["date"] == date and r["time"] == time and r["status"] != "cancelled"
}
# ── API ──────────────────────────────────────────────────────────────
@router.post("/api/reservations", response_model=Reservation, status_code=201)
async def create_reservation(payload: ReservationIn) -> Reservation:
with _lock:
items = _load()
occupied = _occupied_tables(items, payload.date, payload.time)
free = [n for n in range(1, TOTAL_TABLES + 1) if n not in occupied]
if not free:
raise HTTPException(status_code=409, detail="Bu saat için boş masa kalmadı")
record = {
**payload.model_dump(),
"id": uuid.uuid4().hex[:10],
"table": free[0],
"status": "pending",
"created_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
}
items.append(record)
_save(items)
_notify_by_mail(record)
return Reservation(**record)
@router.get("/api/reservations", response_model=list[Reservation])
async def list_reservations(date: str | None = Query(default=None)) -> list[Reservation]:
items = _load()
if date:
items = [r for r in items if r["date"] == date]
items.sort(key=lambda r: (r["date"], r["time"], r["table"]))
return [Reservation(**r) for r in items]
@router.get("/api/reservations/tables")
async def table_map(date: str, time: str) -> dict:
items = _load()
occupied = _occupied_tables(items, date, time)
return {
"date": date,
"time": time,
"total": TOTAL_TABLES,
"tables": [
{
"table": n,
"occupied": n in occupied,
"reservation": (
{k: occupied[n][k] for k in ("id", "name", "guests", "status")}
if n in occupied else None
),
}
for n in range(1, TOTAL_TABLES + 1)
],
}
@router.patch("/api/reservations/{reservation_id}", response_model=Reservation)
async def update_status(reservation_id: str, payload: StatusUpdate) -> Reservation:
with _lock:
items = _load()
for r in items:
if r["id"] == reservation_id:
r["status"] = payload.status
_save(items)
return Reservation(**r)
raise HTTPException(status_code=404, detail="Rezervasyon bulunamadı")
# ── Admin paneli (tek dosyalık, Noir & Grain paletinde) ─────────────
ADMIN_HTML = """
Noir & Grain — Servis Paneli
Noir & Grain · Servis Paneli
Günün Rezervasyonları
"""
@router.get("/admin/reservations", response_class=HTMLResponse, include_in_schema=False)
async def admin_panel() -> HTMLResponse:
return HTMLResponse(ADMIN_HTML)