from datetime import datetime, timezone, timedelta from pathlib import Path from random import shuffle from uuid import uuid4 from fastapi import APIRouter, Depends, File, Form, Query, Request, UploadFile from fastapi.responses import HTMLResponse, JSONResponse, 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 random import choice from string import ascii_letters, digits from app.models import InviteCode, Postcard, Profile, User from app.i18n import t as _t, get_languages as _get_languages 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 _CST = timezone(timedelta(hours=8)) def _to_local(dt: datetime) -> datetime: """Convert UTC datetime to Asia/Shanghai (CST).""" if dt is None: return None if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(_CST) def _to_utc(dt_str: str) -> datetime: """Parse local datetime-local string and convert to UTC.""" dt = datetime.fromisoformat(dt_str) if dt.tzinfo is None: dt = dt.replace(tzinfo=_CST) return dt.astimezone(timezone.utc).replace(tzinfo=None) templates.env.filters["to_local"] = _to_local def _strftime(dt: datetime, fmt: str) -> str: """Format a datetime with strftime.""" return dt.strftime(fmt) if dt else "" templates.env.filters["strftime"] = _strftime import json as _json def _json_loads(raw: str) -> list: """Parse JSON string for Jinja2 templates.""" try: return _json.loads(raw) if raw else [] except (ValueError, TypeError): return [] templates.env.filters["json_loads"] = _json_loads def _translate_filter(key: str, lang: str = "zh") -> str: return _t(key, lang) templates.env.filters["t"] = _translate_filter from app.i18n import get_languages as _get_languages templates.env.globals["available_languages"] = _get_languages() def _get_copyright() -> str: from app.models import SystemConfig as _SC from app.database import SessionLocal try: db = SessionLocal() cfg = db.query(_SC).filter(_SC.key == "copyright_text").first() db.close() return cfg.value if cfg else "" except Exception: return "" templates.env.globals["get_copyright"] = _get_copyright def _redirect(url: str) -> RedirectResponse: return RedirectResponse(url, status_code=303) # ---------- Root / Index ---------- @router.get("/", response_class=HTMLResponse) def index_page(request: Request, db: Session = Depends(get_db)): user = redirect_if_not_logged_in(request, db) if user: return _redirect("/dashboard") # Collect all public postcards from all users with showcase enabled all_postcards: list[Postcard] = [] users = db.query(User).filter(User.showcase_mode.isnot(None)).all() for u in users: profiles = db.query(Profile).filter( Profile.user_id == u.id, Profile.showcase_enabled == True ).all() for p in profiles: cards = db.query(Postcard).filter( Postcard.profile_id == p.id, Postcard.showcase_hidden == False, Postcard.image_front.isnot(None), Postcard.image_front != "" ).all() for c in cards: visible = True if c.status in ("sent", "delivered") and p.showcase_visible not in ("sent", "both"): visible = False if c.status == "received" and p.showcase_visible not in ("received", "both"): visible = False if visible: all_postcards.append(c) shuffle(all_postcards) lang = request.query_params.get("lang") avail = _get_languages() if not lang or lang not in avail: lang = request.cookies.get("showcase_lang") if not lang or lang not in avail: lang = "zh" return templates.TemplateResponse("index.html", { "request": request, "postcards": all_postcards[:20], "lang": lang, "has_more": len(all_postcards) > 20, }) @router.get("/api/public/cards") def api_public_cards( request: Request, page: int = Query(1, ge=1), limit: int = Query(20, ge=1, le=100), db: Session = Depends(get_db), ): all_cards: list[Postcard] = [] users = db.query(User).filter(User.showcase_mode.isnot(None)).all() for u in users: profiles = db.query(Profile).filter( Profile.user_id == u.id, Profile.showcase_enabled == True ).all() for p in profiles: cards = db.query(Postcard).filter( Postcard.profile_id == p.id, Postcard.showcase_hidden == False, Postcard.image_front.isnot(None), Postcard.image_front != "" ).all() for c in cards: vis = True if c.status in ("sent", "delivered") and p.showcase_visible not in ("sent", "both"): vis = False if c.status == "received" and p.showcase_visible not in ("received", "both"): vis = False if vis: all_cards.append(c) shuffle(all_cards) total = len(all_cards) start = (page - 1) * limit batch = all_cards[start:start + limit] return JSONResponse({ "total": total, "page": page, "has_more": start + limit < total, "cards": [{ "image_front": c.image_front, "country_from": c.country_from, "country_to": c.country_to, "status": c.status, } for c in batch], }) @router.get("/api/public/cards/{username}") def api_public_cards_user( username: str, request: Request, page: int = Query(1, ge=1), limit: int = Query(20, ge=1, le=100), tab: str = Query("sent"), db: Session = Depends(get_db), ): user = db.query(User).filter(User.username == username).first() if not user: return JSONResponse({"error": "not found"}, status_code=404) profiles = db.query(Profile).filter( Profile.user_id == user.id, Profile.showcase_enabled == True ).all() all_cards: list[Postcard] = [] for p in profiles: cards = db.query(Postcard).filter( Postcard.profile_id == p.id, Postcard.showcase_hidden == False, Postcard.image_front.isnot(None), Postcard.image_front != "" ).all() if tab == "sent" and p.showcase_visible in ("sent", "both"): all_cards.extend([c for c in cards if c.status in ("sent", "delivered")]) elif tab == "received" and p.showcase_visible in ("received", "both"): all_cards.extend([c for c in cards if c.status == "received"]) all_cards.sort(key=lambda c: c.send_time or c.created_at, reverse=True) total = len(all_cards) start = (page - 1) * limit batch = all_cards[start:start + limit] return JSONResponse({ "total": total, "page": page, "has_more": start + limit < total, "cards": [{ "image_front": c.image_front, "country_from": c.country_from, "country_to": c.country_to, } for c in batch], }) # ---------- 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 # ---------- Register ---------- @router.get("/register", response_class=HTMLResponse) def register_page(request: Request, db: Session = Depends(get_db)): user = redirect_if_not_logged_in(request, db) if user: return _redirect("/dashboard") return templates.TemplateResponse("register.html", {"request": request}) @router.post("/register") def register_submit( request: Request, invite_code: str = Form(...), username: str = Form(...), password: str = Form(...), confirm_password: str = Form(...), db: Session = Depends(get_db), ): # Validate invite code code_obj = db.query(InviteCode).filter(InviteCode.code == invite_code).first() if not code_obj: return templates.TemplateResponse( "register.html", {"request": request, "error": "邀请码无效"}, status_code=400 ) if code_obj.expires_at and code_obj.expires_at < datetime.utcnow(): return templates.TemplateResponse( "register.html", {"request": request, "error": "邀请码已过期"}, status_code=400 ) if code_obj.max_uses is not None and code_obj.use_count >= code_obj.max_uses: return templates.TemplateResponse( "register.html", {"request": request, "error": "邀请码使用次数已用尽"}, status_code=400 ) # Validate passwords if password != confirm_password: return templates.TemplateResponse( "register.html", {"request": request, "error": "两次输入的密码不一致"}, status_code=400 ) if len(password) < 6: return templates.TemplateResponse( "register.html", {"request": request, "error": "密码至少6个字符"}, status_code=400 ) # Check username if db.query(User).filter(User.username == username).first(): return templates.TemplateResponse( "register.html", {"request": request, "error": "用户名已存在"}, status_code=400 ) # Create user new_user = User(username=username, password_hash=hash_password(password), invited_by_code_id=code_obj.id) db.add(new_user) db.flush() # Update invite code usage code_obj.use_count += 1 db.commit() # Auto-login token = create_session(new_user.id, db) resp = _redirect("/dashboard") resp.set_cookie(SESSION_COOKIE_NAME, token, httponly=True, samesite="lax") return resp # ---------- Admin ---------- def _generate_code() -> str: """Generate a random 8-character alphanumeric code.""" return "".join(choice(ascii_letters + digits) for _ in range(8)) def _require_admin(user: User) -> bool: return user.is_admin def _admin_context(user, db, active_tab="invites", **extra): """Build context for admin.html with invite codes and system config.""" from app.models import SystemConfig codes = db.query(InviteCode).order_by(InviteCode.created_at.desc()).all() config = {r.key: r.value for r in db.query(SystemConfig).all()} ctx = {"request": extra.get("request"), "user": user, "codes": codes, "active_tab": active_tab, "config": config} ctx.update({k: v for k, v in extra.items() if k != "request"}) return ctx def _admin_render(request, user, db, active_tab="invites", **extra): return templates.TemplateResponse("admin.html", _admin_context(user, db, active_tab, request=request, **extra)) @router.get("/admin", response_class=HTMLResponse) def admin_page( request: Request, active_tab: str = Query("invites", alias="tab"), user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): if not _require_admin(user): return _redirect("/dashboard") return _admin_render(request, user, db, active_tab) @router.get("/admin/invites", response_class=HTMLResponse) def admin_invites_page( request: Request, user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): if not _require_admin(user): return _redirect("/dashboard") return _admin_render(request, user, db, "invites") @router.post("/admin/config") def admin_save_config( request: Request, copyright_text: str = Form(""), user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): if not _require_admin(user): return _redirect("/dashboard") from app.models import SystemConfig cfg = db.query(SystemConfig).filter(SystemConfig.key == "copyright_text").first() if cfg: cfg.value = copyright_text.strip() else: cfg = SystemConfig(key="copyright_text", value=copyright_text.strip()) db.add(cfg) db.commit() return _admin_render(request, user, db, "config", success=_t("admin.config_saved", user.language)) @router.post("/admin/invites/add") def admin_add_invite( request: Request, code: str = Form(""), limit_type: str = Form("unlimited"), max_uses: int = Form(None), expires_at: str = Form(None), user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): if not _require_admin(user): return _redirect("/dashboard") # Generate or validate code if not code.strip(): code = _generate_code() else: code = code.strip() if len(code) != 8: return _admin_render(request, user, db, "invites", error=_t("admin.invite_length_error", user.language)) if db.query(InviteCode).filter(InviteCode.code == code).first(): return _admin_render(request, user, db, "invites", error=_t("admin.invite_exists_error", user.language)) # Parse limits m_uses = None exp_at = None if limit_type == "uses" and max_uses: m_uses = max_uses elif limit_type == "expires" and expires_at: exp_at = _to_utc(expires_at) new_code = InviteCode( code=code, max_uses=m_uses, expires_at=exp_at, use_count=0, created_by=user.id, ) db.add(new_code) db.commit() return _admin_render(request, user, db, "invites", success=_t("admin.invite_created", user.language).format(code=code)) @router.post("/admin/invites/{code_id}/edit") def admin_edit_invite( request: Request, code_id: int, limit_type: str = Form("unlimited"), max_uses: int = Form(None), expires_at: str = Form(None), user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): if not _require_admin(user): return _redirect("/dashboard") code_obj = db.query(InviteCode).filter(InviteCode.id == code_id).first() if not code_obj: return _redirect("/admin/invites") if limit_type == "uses" and max_uses: code_obj.max_uses = max_uses code_obj.expires_at = None elif limit_type == "expires" and expires_at: code_obj.expires_at = _to_utc(expires_at) code_obj.max_uses = None else: code_obj.max_uses = None code_obj.expires_at = None db.commit() return _admin_render(request, user, db, "invites", success=_t("admin.invite_updated", user.language).format(code=code_obj.code)) @router.post("/admin/invites/{code_id}/delete") def admin_delete_invite( request: Request, code_id: int, user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): if not _require_admin(user): return _redirect("/dashboard") code_obj = db.query(InviteCode).filter(InviteCode.id == code_id).first() if code_obj: db.delete(code_obj) db.commit() return _redirect("/admin?tab=invites") # ---------- 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), ): from app.models import ApiKey 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() api_keys = db.query(ApiKey).filter(ApiKey.user_id == user.id).order_by(ApiKey.created_at.desc()).all() return templates.TemplateResponse("settings.html", { "request": request, "user": user, "profiles": profiles, "profile_cards": profile_cards, "active_tab": active_tab, "api_keys": api_keys, }) @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": "密码修改成功"} ) @router.post("/settings/language") def change_language( request: Request, language: str = Form(...), user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): if language in ("zh", "en"): user.language = language db.commit() return _redirect(request.headers.get("referer", "/settings")) # ---------- 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_to for c in all_cards if c.country_to}) # recent postcards across all profiles (last 8), sorted by most recently updated all_cards.sort(key=lambda c: c.updated_at or 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 # country stats split by sent/received sent_country_counts: dict[str, int] = {} received_country_counts: dict[str, int] = {} for c in all_cards: if not c.country_to: continue if c.status in ("sent", "delivered"): sent_country_counts[c.country_to] = sent_country_counts.get(c.country_to, 0) + 1 elif c.status == "received": received_country_counts[c.country_to] = received_country_counts.get(c.country_to, 0) + 1 sent_country_stats = sorted(sent_country_counts.items(), key=lambda x: -x[1]) received_country_stats = sorted(received_country_counts.items(), key=lambda x: -x[1]) 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, "sent_country_stats": sent_country_stats, "received_country_stats": received_country_stats, "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_to for c in cards if c.country_to}), } return templates.TemplateResponse("profiles.html", {"request": request, "user": user, "profiles": profiles, "extras": extras}) @router.post("/profiles") def profile_create(request: Request, nickname: str = Form(...), country: str = Form(""), user: User = Depends(get_current_user_web), db: Session = Depends(get_db)): c = country.strip().upper() if country else None db.add(Profile(user_id=user.id, nickname=nickname, country=c)) 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(...), country: 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 p.country = country.strip().upper() if country else None db.commit() return _redirect("/profiles") @router.post("/profiles/{profile_id}/notes") def profile_notes(profile_id: int, notes: str = Form(""), redirect_to: str = Form("/profiles"), 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.notes = notes db.commit() return _redirect(redirect_to) # ---------- 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_to == country) postcards = q.all() # collect existing countries for dropdown all_countries = sorted({pc.country_to for pc in db.query(Postcard.country_to).filter(Postcard.profile_id == profile_id, Postcard.country_to.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_to: str = Form(""), country_from: str = Form(""), send_time: str = Form(""), arrival_time: str = Form(""), sender_name: str = Form(""), receive_time: str = Form(""), notes: 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") cn = card_number.strip() dup = db.query(Postcard).join(Profile).filter( Postcard.card_number == cn, Profile.user_id == user.id ).first() if dup: return templates.TemplateResponse("postcard_form.html", { "request": request, "user": user, "profile": profile, "postcard": None, "error": _t("pc_list.duplicate", user.language), }) 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.strip(), status=status, recipient_name=recipient_name or None, country_from=(country_from.strip().upper() if country_from else profile.country), country_to=(country_to.strip().upper() if country_to 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), notes=notes.strip(), ) 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_to: str = Form(""), country_from: str = Form(""), send_time: str = Form(""), arrival_time: str = Form(""), sender_name: str = Form(""), receive_time: str = Form(""), notes: 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") cn = card_number.strip() dup = db.query(Postcard).join(Profile).filter( Postcard.card_number == cn, Postcard.id != postcard_id, Profile.user_id == user.id ).first() if dup: return templates.TemplateResponse("postcard_form.html", { "request": request, "user": user, "profile": pc.profile, "postcard": pc, "error": _t("pc_list.duplicate", user.language), }) 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.strip() pc.status = status pc.recipient_name = recipient_name or None pc.country_from = country_from.strip().upper() if country_from else pc.country_from pc.country_to = country_to 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) pc.notes = notes.strip() 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) # Language: query param > cookie > owner's default from app.i18n import get_languages _avail = get_languages() visitor_lang = request.query_params.get("lang") if not visitor_lang or visitor_lang not in _avail: visitor_lang = request.cookies.get("showcase_lang") if not visitor_lang or visitor_lang not in _avail: visitor_lang = user.language 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, Postcard.image_front.isnot(None), Postcard.image_front != "" ).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) resp = 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, "lang": visitor_lang, "available_languages": _avail, }) resp.set_cookie("showcase_lang", visitor_lang, max_age=365 * 24 * 3600, httponly=True, samesite="lax") return resp # ---------- API Keys ---------- @router.post("/settings/apikeys/create") def apikey_create( request: Request, name: str = Form(""), permissions: list[str] = Form([]), user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): from app.models import ApiKey, ALL_PERMISSIONS from secrets import token_hex import json key = f"mv_{token_hex(24)}" # Filter to valid scopes only valid = [p for p in permissions if p in ALL_PERMISSIONS] perms_json = json.dumps(valid) if valid else "[]" ak = ApiKey(key=key, name=name.strip(), user_id=user.id, permissions=perms_json) db.add(ak) db.commit() db.refresh(ak) api_keys = db.query(ApiKey).filter(ApiKey.user_id == user.id).order_by(ApiKey.created_at.desc()).all() return templates.TemplateResponse("settings.html", { "request": request, "user": user, "active_tab": "apikeys", "api_keys": api_keys, "new_key": ak.key, "success": "API Key created", }) @router.post("/settings/apikeys/{key_id}/delete") def apikey_delete( key_id: int, user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): from app.models import ApiKey ak = db.query(ApiKey).filter(ApiKey.id == key_id, ApiKey.user_id == user.id).first() if ak: db.delete(ak) db.commit() return _redirect("/settings?tab=apikeys") @router.post("/settings/apikeys/{key_id}/edit") def apikey_edit( key_id: int, request: Request, name: str = Form(""), permissions: list[str] = Form([]), user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): from app.models import ApiKey, ALL_PERMISSIONS import json ak = db.query(ApiKey).filter(ApiKey.id == key_id, ApiKey.user_id == user.id).first() if not ak: return _redirect("/settings?tab=apikeys") ak.name = name.strip() valid = [p for p in permissions if p in ALL_PERMISSIONS] ak.permissions = json.dumps(valid) if valid else "[]" db.commit() return _redirect("/settings?tab=apikeys") # ---------- API Docs ---------- @router.get("/api/docs", response_class=HTMLResponse) def api_docs_page( request: Request, user: User = Depends(get_current_user_web), db: Session = Depends(get_db), ): return templates.TemplateResponse("api_docs.html", { "request": request, "user": user, })