908 lines
32 KiB
Python
908 lines
32 KiB
Python
from datetime import datetime, timezone, timedelta
|
|
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 datetime import datetime
|
|
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
|
|
|
|
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
|
|
|
|
|
|
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 _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
|
|
|
|
|
|
# ---------- 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
|
|
|
|
|
|
@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")
|
|
codes = db.query(InviteCode).order_by(InviteCode.created_at.desc()).all()
|
|
return templates.TemplateResponse("admin_invites.html", {
|
|
"request": request, "user": user, "codes": codes,
|
|
})
|
|
|
|
|
|
@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:
|
|
codes = db.query(InviteCode).order_by(InviteCode.created_at.desc()).all()
|
|
return templates.TemplateResponse("admin_invites.html", {
|
|
"request": request, "user": user, "codes": codes,
|
|
"error": "邀请码必须为8个字符",
|
|
})
|
|
if db.query(InviteCode).filter(InviteCode.code == code).first():
|
|
codes = db.query(InviteCode).order_by(InviteCode.created_at.desc()).all()
|
|
return templates.TemplateResponse("admin_invites.html", {
|
|
"request": request, "user": user, "codes": codes,
|
|
"error": "邀请码已存在",
|
|
})
|
|
# 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()
|
|
codes = db.query(InviteCode).order_by(InviteCode.created_at.desc()).all()
|
|
return templates.TemplateResponse("admin_invites.html", {
|
|
"request": request, "user": user, "codes": codes,
|
|
"success": f"邀请码 {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()
|
|
codes = db.query(InviteCode).order_by(InviteCode.created_at.desc()).all()
|
|
return templates.TemplateResponse("admin_invites.html", {
|
|
"request": request, "user": user, "codes": codes,
|
|
"success": f"邀请码 {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/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 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
|
|
|
|
# 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:
|
|
continue
|
|
if c.status in ("sent", "delivered"):
|
|
sent_country_counts[c.country] = sent_country_counts.get(c.country, 0) + 1
|
|
elif c.status == "received":
|
|
received_country_counts[c.country] = received_country_counts.get(c.country, 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 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")
|
|
|
|
|
|
@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 == 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(""),
|
|
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()
|
|
if db.query(Postcard).filter(Postcard.card_number == cn).first():
|
|
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=(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),
|
|
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: 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).filter(Postcard.card_number == cn, Postcard.id != postcard_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 = 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)
|
|
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).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(""),
|
|
user: User = Depends(get_current_user_web),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
from app.models import ApiKey
|
|
from secrets import token_hex
|
|
key = f"mv_{token_hex(24)}"
|
|
ak = ApiKey(key=key, name=name.strip(), user_id=user.id)
|
|
db.add(ak)
|
|
db.commit()
|
|
db.refresh(ak)
|
|
# Re-render settings with apikeys tab active and the new key shown
|
|
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")
|
|
|
|
|
|
# ---------- 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,
|
|
})
|