558 lines
20 KiB
Python
558 lines
20 KiB
Python
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,
|
|
})
|