from __future__ import annotations from collections.abc import Sequence from dataclasses import dataclass from datetime import UTC, datetime, timedelta import secrets from app.extensions import db from app.models import ShareLink DEFAULT_SHARE_LINK_EXPIRE_DAYS = 30 MIN_SHARE_LINK_EXPIRE_DAYS = 1 MAX_SHARE_LINK_EXPIRE_DAYS = 365 SHARE_LINK_TOKEN_BYTES = 16 SHARE_EDITABLE_FIELDS = ( "phone", "attendance_status", "actual_attendee_count", "child_count", "red_packet_child_count", "total_gift_amount", "gift_method_option_id", "gift_scene_option_id", "favor_status", "candy_status", "child_red_packet_status", "note", ) @dataclass(frozen=True) class ShareTokenCheckResult: ok: bool reason: str link: ShareLink | None = None def list_share_links() -> Sequence[ShareLink]: result = db.session.execute( db.select(ShareLink).order_by(ShareLink.created_at.desc(), ShareLink.id.desc()), ) return result.scalars().all() def get_share_link_by_token(token: str) -> ShareLink | None: normalized = token.strip() if not normalized: return None result = db.session.execute(db.select(ShareLink).where(ShareLink.token == normalized).limit(1)) return result.scalar_one_or_none() def get_share_link_or_none(link_id: int) -> ShareLink | None: return db.session.get(ShareLink, link_id) def normalize_expire_days(raw_days: int) -> int: if raw_days < MIN_SHARE_LINK_EXPIRE_DAYS: return MIN_SHARE_LINK_EXPIRE_DAYS if raw_days > MAX_SHARE_LINK_EXPIRE_DAYS: return MAX_SHARE_LINK_EXPIRE_DAYS return raw_days def parse_share_link_create_form(form: dict[str, str]) -> tuple[dict[str, object], list[str]]: errors: list[str] = [] label = form.get("label", "").strip() or None raw_expire_days = form.get("expire_days", "").strip() or str(DEFAULT_SHARE_LINK_EXPIRE_DAYS) try: expire_days = int(raw_expire_days) except ValueError: return {}, ["有效期天数必须是整数。"] if expire_days < MIN_SHARE_LINK_EXPIRE_DAYS or expire_days > MAX_SHARE_LINK_EXPIRE_DAYS: errors.append(f"有效期天数必须在 {MIN_SHARE_LINK_EXPIRE_DAYS}-{MAX_SHARE_LINK_EXPIRE_DAYS} 之间。") if errors: return {}, errors return {"label": label, "expire_days": expire_days}, [] def create_share_link(*, created_by: int | None, label: str | None, expire_days: int) -> ShareLink: normalized_days = normalize_expire_days(expire_days) token = _generate_unique_share_token() link = ShareLink( token=token, label=(label or "").strip() or None, expires_at=_build_expire_at(normalized_days), revoked_at=None, created_by=created_by, ) db.session.add(link) db.session.flush() return link def revoke_share_link(link: ShareLink, *, now: datetime | None = None) -> bool: if link.revoked_at is not None: return False link.revoked_at = (now or datetime.now(UTC)).replace(tzinfo=None) return True def check_share_token(token: str, *, now: datetime | None = None) -> ShareTokenCheckResult: link = get_share_link_by_token(token) if link is None: return ShareTokenCheckResult(ok=False, reason="not_found") current = (now or datetime.now(UTC)).replace(tzinfo=None) if link.revoked_at is not None: return ShareTokenCheckResult(ok=False, reason="revoked", link=link) if link.expires_at <= current: return ShareTokenCheckResult(ok=False, reason="expired", link=link) return ShareTokenCheckResult(ok=True, reason="ok", link=link) def build_share_link_url(token: str) -> str: return f"/s/{token}" def _build_expire_at(expire_days: int) -> datetime: return (datetime.now(UTC) + timedelta(days=expire_days)).replace(tzinfo=None) def _generate_share_token() -> str: return secrets.token_urlsafe(SHARE_LINK_TOKEN_BYTES) def _generate_unique_share_token() -> str: while True: token = _generate_share_token() exists = db.session.execute(db.select(ShareLink.id).where(ShareLink.token == token).limit(1)).scalar_one_or_none() if exists is None: return token