from datetime import datetime from pathlib import Path from uuid import uuid4 from fastapi import APIRouter, Depends, File, Form, Query, 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 Postcard, Profile, User router = APIRouter(tags=["web"]) templates = Jinja2Templates(directory=str(Path(__file__).resolve().parent.parent / "templates")) def _country_flag(code: str) -> str: if not code or len(code) != 2: return code or "" return chr(0x1F1E6 + ord(code[0].upper()) - ord("A")) + chr(0x1F1E6 + ord(code[1].upper()) - ord("A")) templates.env.filters["flag"] = _country_flag 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(...), remember_me: str = Form(""), db: Session = Depends(get_db), ): from app.auth import cleanup_expired_sessions cleanup_expired_sessions(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 ) is_remember = remember_me == "on" token = create_session(user.id, db, remember_me=is_remember) resp = _redirect("/dashboard") cookie_kwargs = {"httponly": True, "samesite": "lax"} if is_remember: cookie_kwargs["max_age"] = 365 * 24 * 3600 resp.set_cookie(SESSION_COOKIE_NAME, token, **cookie_kwargs) return resp # ---------- Settings ---------- @router.get("/settings", response_class=HTMLResponse) def settings_page( request: Request, active_tab: str = Query("account", alias="tab"), user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): profiles = db.query(Profile).filter(Profile.user_id == user.id).all() profile_cards = {} for p in profiles: profile_cards[p.id] = db.query(Postcard).filter(Postcard.profile_id == p.id).all() return templates.TemplateResponse("settings.html", { "request": request, "user": user, "profiles": profiles, "profile_cards": profile_cards, "active_tab": active_tab, }) @router.post("/settings/change-username") def change_username( request: Request, new_username: str = Form(...), user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): if db.query(User).filter(User.username == new_username, User.id != user.id).first(): return templates.TemplateResponse( "settings.html", {"request": request, "user": user, "error": "用户名已存在"}, status_code=400 ) user.username = new_username db.commit() return templates.TemplateResponse( "settings.html", {"request": request, "user": user, "success": "用户名修改成功"} ) @router.post("/settings/change-password") def change_password( request: Request, old_password: str = Form(...), new_password: str = Form(...), confirm_password: str = Form(...), user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): if not verify_password(old_password, user.password_hash): return templates.TemplateResponse( "settings.html", {"request": request, "user": user, "error": "当前密码错误"}, status_code=400 ) if new_password != confirm_password: return templates.TemplateResponse( "settings.html", {"request": request, "user": user, "error": "两次输入的新密码不一致"}, status_code=400 ) if len(new_password) < 6: return templates.TemplateResponse( "settings.html", {"request": request, "user": user, "error": "新密码长度至少6位"}, status_code=400 ) user.password_hash = hash_password(new_password) db.commit() return templates.TemplateResponse( "settings.html", {"request": request, "user": user, "success": "密码修改成功"} ) # ---------- Auth ---------- @router.get("/logout") def logout(request: Request, db: Session = Depends(get_db)): token = request.cookies.get(SESSION_COOKIE_NAME) destroy_session(token or "", db) 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() all_cards = [] for p in profiles: cards = db.query(Postcard).filter(Postcard.profile_id == p.id).all() all_cards.extend(cards) total = len(all_cards) sent = sum(1 for c in all_cards if c.status == "sent") delivered = sum(1 for c in all_cards if c.status == "delivered") received = sum(1 for c in all_cards if c.status == "received") countries = len({c.country for c in all_cards if c.country}) # recent postcards across all profiles (last 8) all_cards.sort(key=lambda c: c.created_at, reverse=True) recent = all_cards[:8] # status distribution for progress bar if total > 0: sent_pct = round(sent / total * 100) delivered_pct = round(delivered / total * 100) received_pct = round(received / total * 100) else: sent_pct = delivered_pct = received_pct = 0 from datetime import datetime, timezone as _tz return templates.TemplateResponse("dashboard.html", { "request": request, "user": user, "profiles": profiles, "total": total, "sent": sent, "delivered": delivered, "received": received, "countries": countries, "recent": recent, "sent_pct": sent_pct, "delivered_pct": delivered_pct, "received_pct": received_pct, "now": datetime.now(_tz.utc), }) # ---------- 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() extras = {} for p in profiles: cards = db.query(Postcard).filter(Postcard.profile_id == p.id).all() extras[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"), "countries": len({c.country for c in cards if c.country}), } return templates.TemplateResponse("profiles.html", {"request": request, "user": user, "profiles": profiles, "extras": extras}) @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") @router.post("/profiles/{profile_id}/rename") def profile_rename(profile_id: int, nickname: str = Form(...), 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: p.nickname = nickname db.commit() return _redirect("/profiles") # ---------- Postcards ---------- @router.get("/profiles/{profile_id}/postcards", response_class=HTMLResponse) def postcard_list( request: Request, profile_id: int, 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 country: q = q.filter(Postcard.country == country) postcards = q.all() # collect existing countries for dropdown all_countries = sorted({pc.country for pc in db.query(Postcard.country).filter(Postcard.profile_id == profile_id, Postcard.country.isnot(None)).all()}) # status sort order: sent=0, delivered=1, received=2 _status_order = {"sent": 0, "delivered": 1, "received": 2} postcards.sort(key=lambda pc: (_status_order.get(pc.status, 9), -(pc.send_time or pc.arrival_time or pc.receive_time or pc.created_at).timestamp())) # split into sent and received sent = [pc for pc in postcards if pc.status in ("sent", "delivered")] received = [pc for pc in postcards if pc.status == "received"] return templates.TemplateResponse( "postcards.html", {"request": request, "user": user, "profile": profile, "sent": sent, "received": received, "countries": all_countries, "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.strip().upper() if country else 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 = None if status == "sent" else _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") @router.post("/postcards/{postcard_id}/mark-delivered") def mark_delivered( 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") pc.status = "delivered" if not pc.arrival_time: pc.arrival_time = datetime.now() db.commit() return _redirect(f"/postcards/{postcard_id}") # ---------- 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}") # ---------- Showcase ---------- @router.get("/showcase/manage") def showcase_manage_redirect(user: User = Depends(get_current_user_web)): return _redirect("/settings?tab=showcase") @router.post("/showcase/manage") def showcase_manage_save(request: Request, showcase_mode: str = Form("profile"), user: User = Depends(get_current_user_web), db: Session = Depends(get_db)): if showcase_mode in ("profile", "flat"): user.showcase_mode = showcase_mode db.commit() return _redirect("/settings?tab=showcase") @router.get("/profiles/{profile_id}/showcase", response_class=HTMLResponse) def profile_showcase_page( request: Request, 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 not p: return _redirect("/settings?tab=showcase") cards = db.query(Postcard).filter(Postcard.profile_id == profile_id).all() return templates.TemplateResponse("profile_showcase.html", { "request": request, "user": user, "profile": p, "cards": cards, }) @router.post("/profiles/{profile_id}/showcase") def profile_showcase_save( profile_id: int, showcase_enabled: str = Form(""), showcase_visible: str = Form("both"), 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: p.showcase_enabled = showcase_enabled == "1" if showcase_visible in ("sent", "received", "both"): p.showcase_visible = showcase_visible db.commit() return _redirect("/settings?tab=showcase") @router.post("/postcards/{postcard_id}/showcase-toggle") def postcard_showcase_toggle(postcard_id: int, showcase_visible: str = Form(""), user: User = Depends(get_current_user_web), db: Session = Depends(get_db)): pc = db.query(Postcard).filter(Postcard.id == postcard_id).first() if pc: profile = db.query(Profile).filter(Profile.id == pc.profile_id, Profile.user_id == user.id).first() if profile and profile.showcase_enabled: pc.showcase_hidden = showcase_visible != "1" db.commit() return _redirect("/settings?tab=showcase") @router.get("/u/{username}", response_class=HTMLResponse) def public_showcase(request: Request, username: str, db: Session = Depends(get_db)): user = db.query(User).filter(User.username == username).first() if not user: return HTMLResponse("用户不存在", status_code=404) profiles = db.query(Profile).filter(Profile.user_id == user.id, Profile.showcase_enabled == True).all() sent_all, received_all, profile_groups = [], [], [] for p in profiles: cards = db.query(Postcard).filter(Postcard.profile_id == p.id, Postcard.showcase_hidden == False).all() if p.showcase_visible in ("sent", "both"): sent_all.extend([c for c in cards if c.status in ("sent", "delivered")]) if p.showcase_visible in ("received", "both"): received_all.extend([c for c in cards if c.status == "received"]) if p.showcase_visible in ("sent", "both"): p_sent = [c for c in cards if c.status in ("sent", "delivered")] else: p_sent = [] if p.showcase_visible in ("received", "both"): p_received = [c for c in cards if c.status == "received"] else: p_received = [] if p_sent or p_received: p_sent.sort(key=lambda c: c.send_time or c.created_at, reverse=True) p_received.sort(key=lambda c: c.receive_time or c.created_at, reverse=True) profile_groups.append({"profile": p, "sent": p_sent, "received": p_received}) sent_all.sort(key=lambda c: c.send_time or c.created_at, reverse=True) received_all.sort(key=lambda c: c.receive_time or c.created_at, reverse=True) return templates.TemplateResponse("showcase.html", { "request": request, "profile_user": user, "mode": user.showcase_mode, "profile_groups": profile_groups, "sent_all": sent_all, "received_all": received_all, })