from datetime import datetime from pathlib import Path from uuid import uuid4 from fastapi import APIRouter, Depends, File, Form, Request, UploadFile from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.templating import Jinja2Templates from sqlalchemy.orm import Session from app.auth import ( create_session, destroy_session, get_current_user_web, hash_password, redirect_if_not_logged_in, verify_password, ) from app.config import SECRET_KEY, SESSION_COOKIE_NAME, UPLOAD_DIR from app.database import get_db from app.models import ApiKey, Postcard, Profile, User router = APIRouter(tags=["web"]) templates = Jinja2Templates(directory=str(Path(__file__).resolve().parent.parent / "templates")) def _redirect(url: str) -> RedirectResponse: return RedirectResponse(url, status_code=303) # ---------- Auth ---------- @router.get("/login", response_class=HTMLResponse) def login_page(request: Request, db: Session = Depends(get_db)): user = redirect_if_not_logged_in(request, db) if user: return _redirect("/dashboard") return templates.TemplateResponse("login.html", {"request": request}) @router.post("/login") def login_submit( request: Request, username: str = Form(...), password: str = Form(...), db: Session = Depends(get_db), ): user = db.query(User).filter(User.username == username).first() if not user or not verify_password(password, user.password_hash): return templates.TemplateResponse( "login.html", {"request": request, "error": "用户名或密码错误"}, status_code=401 ) resp = _redirect("/dashboard") resp.set_cookie(SESSION_COOKIE_NAME, create_session(user.id), httponly=True) return resp @router.get("/register", response_class=HTMLResponse) def register_page(request: Request): return templates.TemplateResponse("register.html", {"request": request}) @router.post("/register") def register_submit( request: Request, username: str = Form(...), password: str = Form(...), password2: str = Form(...), db: Session = Depends(get_db), ): if password != password2: return templates.TemplateResponse( "register.html", {"request": request, "error": "两次密码不一致"}, status_code=400 ) if db.query(User).filter(User.username == username).first(): return templates.TemplateResponse( "register.html", {"request": request, "error": "用户名已存在"}, status_code=400 ) user = User(username=username, password_hash=hash_password(password)) db.add(user) db.commit() resp = _redirect("/dashboard") resp.set_cookie(SESSION_COOKIE_NAME, create_session(user.id), httponly=True) return resp @router.get("/logout") def logout(request: Request): token = request.cookies.get(SESSION_COOKIE_NAME) destroy_session(token or "") resp = _redirect("/login") resp.delete_cookie(SESSION_COOKIE_NAME) return resp # ---------- Dashboard ---------- @router.get("/", response_class=HTMLResponse) def root(user: User = Depends(get_current_user_web)): return _redirect("/dashboard") @router.get("/dashboard", response_class=HTMLResponse) def dashboard(request: Request, user: User = Depends(get_current_user_web), db: Session = Depends(get_db)): profiles = db.query(Profile).filter(Profile.user_id == user.id).all() # stats per profile stats = {} for p in profiles: cards = db.query(Postcard).filter(Postcard.profile_id == p.id).all() stats[p.id] = { "total": len(cards), "sent": sum(1 for c in cards if c.status == "sent"), "delivered": sum(1 for c in cards if c.status == "delivered"), "received": sum(1 for c in cards if c.status == "received"), } return templates.TemplateResponse("dashboard.html", {"request": request, "user": user, "profiles": profiles, "stats": stats}) # ---------- Profiles ---------- @router.get("/profiles", response_class=HTMLResponse) def profile_list(request: Request, user: User = Depends(get_current_user_web), db: Session = Depends(get_db)): profiles = db.query(Profile).filter(Profile.user_id == user.id).all() return templates.TemplateResponse("profiles.html", {"request": request, "user": user, "profiles": profiles}) @router.post("/profiles") def profile_create(request: Request, nickname: str = Form(...), user: User = Depends(get_current_user_web), db: Session = Depends(get_db)): db.add(Profile(user_id=user.id, nickname=nickname)) db.commit() return _redirect("/profiles") @router.post("/profiles/{profile_id}/delete") def profile_delete(profile_id: int, user: User = Depends(get_current_user_web), db: Session = Depends(get_db)): p = db.query(Profile).filter(Profile.id == profile_id, Profile.user_id == user.id).first() if p: db.delete(p) db.commit() return _redirect("/profiles") # ---------- Postcards ---------- @router.get("/profiles/{profile_id}/postcards", response_class=HTMLResponse) def postcard_list( request: Request, profile_id: int, status: str = "", country: str = "", user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): profile = db.query(Profile).filter(Profile.id == profile_id, Profile.user_id == user.id).first() if not profile: return _redirect("/profiles") q = db.query(Postcard).filter(Postcard.profile_id == profile_id) if status: q = q.filter(Postcard.status == status) if country: q = q.filter(Postcard.country.ilike(f"%{country}%")) postcards = q.order_by(Postcard.created_at.desc()).all() return templates.TemplateResponse( "postcards.html", {"request": request, "user": user, "profile": profile, "postcards": postcards, "status": status, "country": country}, ) @router.get("/profiles/{profile_id}/postcards/new", response_class=HTMLResponse) def postcard_new(request: Request, profile_id: int, user: User = Depends(get_current_user_web), db: Session = Depends(get_db)): profile = db.query(Profile).filter(Profile.id == profile_id, Profile.user_id == user.id).first() if not profile: return _redirect("/profiles") return templates.TemplateResponse("postcard_form.html", {"request": request, "user": user, "profile": profile, "postcard": None}) @router.post("/profiles/{profile_id}/postcards") def postcard_create( request: Request, profile_id: int, card_number: str = Form(...), status: str = Form(...), recipient_name: str = Form(""), country: str = Form(""), send_time: str = Form(""), arrival_time: str = Form(""), sender_name: str = Form(""), receive_time: str = Form(""), user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): profile = db.query(Profile).filter(Profile.id == profile_id, Profile.user_id == user.id).first() if not profile: return _redirect("/profiles") def _parse_dt(s: str) -> datetime | None: s = s.strip() if not s: return None try: return datetime.fromisoformat(s) except ValueError: try: return datetime.strptime(s, "%Y-%m-%d") except ValueError: return None pc = Postcard( profile_id=profile_id, card_number=card_number, status=status, recipient_name=recipient_name or None, country=country or None, send_time=_parse_dt(send_time), arrival_time=_parse_dt(arrival_time), sender_name=sender_name or None, receive_time=_parse_dt(receive_time), ) db.add(pc) db.commit() return _redirect(f"/profiles/{profile_id}/postcards") @router.get("/postcards/{postcard_id}", response_class=HTMLResponse) def postcard_detail(request: Request, postcard_id: int, user: User = Depends(get_current_user_web), db: Session = Depends(get_db)): pc = ( db.query(Postcard) .join(Profile) .filter(Postcard.id == postcard_id, Profile.user_id == user.id) .first() ) if not pc: return _redirect("/profiles") return templates.TemplateResponse("postcard_detail.html", {"request": request, "user": user, "postcard": pc}) @router.get("/postcards/{postcard_id}/edit", response_class=HTMLResponse) def postcard_edit(request: Request, postcard_id: int, user: User = Depends(get_current_user_web), db: Session = Depends(get_db)): pc = ( db.query(Postcard) .join(Profile) .filter(Postcard.id == postcard_id, Profile.user_id == user.id) .first() ) if not pc: return _redirect("/profiles") return templates.TemplateResponse("postcard_form.html", {"request": request, "user": user, "profile": pc.profile, "postcard": pc}) @router.post("/postcards/{postcard_id}/edit") def postcard_update( request: Request, postcard_id: int, card_number: str = Form(...), status: str = Form(...), recipient_name: str = Form(""), country: str = Form(""), send_time: str = Form(""), arrival_time: str = Form(""), sender_name: str = Form(""), receive_time: str = Form(""), user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): pc = ( db.query(Postcard) .join(Profile) .filter(Postcard.id == postcard_id, Profile.user_id == user.id) .first() ) if not pc: return _redirect("/profiles") def _parse_dt(s: str) -> datetime | None: s = s.strip() if not s: return None try: return datetime.fromisoformat(s) except ValueError: try: return datetime.strptime(s, "%Y-%m-%d") except ValueError: return None pc.card_number = card_number pc.status = status pc.recipient_name = recipient_name or None pc.country = country or None pc.send_time = _parse_dt(send_time) pc.arrival_time = _parse_dt(arrival_time) pc.sender_name = sender_name or None pc.receive_time = _parse_dt(receive_time) db.commit() return _redirect(f"/postcards/{postcard_id}") @router.post("/postcards/{postcard_id}/delete") def postcard_delete(postcard_id: int, user: User = Depends(get_current_user_web), db: Session = Depends(get_db)): pc = ( db.query(Postcard) .join(Profile) .filter(Postcard.id == postcard_id, Profile.user_id == user.id) .first() ) if not pc: return _redirect("/profiles") pid = pc.profile_id db.delete(pc) db.commit() return _redirect(f"/profiles/{pid}/postcards") # ---------- Image Upload ---------- @router.post("/postcards/{postcard_id}/image/{side}") async def upload_image( postcard_id: int, side: str, file: UploadFile = File(...), user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): if side not in ("front", "back"): return _redirect("/profiles") pc = ( db.query(Postcard) .join(Profile) .filter(Postcard.id == postcard_id, Profile.user_id == user.id) .first() ) if not pc: return _redirect("/profiles") ext = Path(file.filename or "upload.jpg").suffix or ".jpg" filename = f"{uuid4().hex}{ext}" dest = UPLOAD_DIR / filename content = await file.read() dest.write_bytes(content) setattr(pc, f"image_{side}", f"/uploads/{filename}") db.commit() return _redirect(f"/postcards/{postcard_id}") # ---------- API Keys ---------- @router.get("/api-keys", response_class=HTMLResponse) def api_keys_page(request: Request, user: User = Depends(get_current_user_web), db: Session = Depends(get_db)): keys = db.query(ApiKey).filter(ApiKey.user_id == user.id).all() return templates.TemplateResponse("api_keys.html", {"request": request, "user": user, "keys": keys}) @router.post("/api-keys") def api_key_create(user: User = Depends(get_current_user_web), db: Session = Depends(get_db)): ak = ApiKey(user_id=user.id, key=ApiKey.generate_key()) db.add(ak) db.commit() return _redirect("/api-keys") @router.post("/api-keys/{key_id}/delete") def api_key_delete(key_id: int, user: User = Depends(get_current_user_web), db: Session = Depends(get_db)): ak = db.query(ApiKey).filter(ApiKey.id == key_id, ApiKey.user_id == user.id).first() if ak: db.delete(ak) db.commit() return _redirect("/api-keys")