gitlab-instance-0a899031_cnomadcna/backend/main.py

from __future__ import annotations

import asyncio
import io
import re
import secrets
import uuid
import zipfile
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo
from html import escape
from pathlib import Path
from typing import Any

import httpx
from fastapi import FastAPI, File, HTTPException, Request, Response, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.responses import Response as FastAPIResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field

from .avatar_assets import avatar_for_name
from .integrations import subscribe_newsletter
from .ntfy_push import dispatch_ntfy_for_user, ntfy_subscribe_url, ntfy_topic_for_user
from .dm import (
    MAX_MESSAGE_LENGTH,
    conversation_by_connection,
    conversation_for_user,
    ensure_conversation_for_match,
    ensure_direct_conversation,
    list_conversations_for_user,
    list_messages,
    mark_conversation_read,
    other_participant_id,
    public_record as dm_public_record,
    send_message as send_dm_message,
    sync_connection_conversation,
    unread_count_for,
)
from .meetup_live import (
    access_lock_reason,
    build_lounge_chat_url,
    build_mirotalk_join_url,
    can_access_meetup,
    create_lounge_channel,
    create_mirotalk_room_slug,
    viewer_display_name,
)
from .payment import XorPayProvider, ZPayProvider, provider_for
from .pb_client import PocketBaseError, pb, pb_quote
from .settings import get_settings
from .weather import fetch_weather_for_cities
from .weather import fetch_weather_for_cities

settings = get_settings()
settings.upload_dir.mkdir(parents=True, exist_ok=True)
processed_orders: set[str] = set()

app = FastAPI(title="NomadCNA Local API", version="0.1.0")


@app.middleware("http")
async def normalize_double_api_prefix(request: Request, call_next):
    if request.scope.get("path", "").startswith("/api/api/"):
        request.scope["path"] = request.scope["path"][4:]
        raw_path = request.scope.get("raw_path")
        if isinstance(raw_path, bytes) and raw_path.startswith(b"/api/api/"):
            request.scope["raw_path"] = raw_path[4:]
    return await call_next(request)


app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.frontend_origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.mount("/uploads", StaticFiles(directory=settings.upload_dir), name="uploads")


class AuthPayload(BaseModel):
    email: str


class GoogleAuthPayload(BaseModel):
    id_token: str


class SyncSessionPayload(BaseModel):
    token: str
    record: dict[str, Any]


class EnsureUserPayload(BaseModel):
    email: str


class CheckUserPayload(BaseModel):
    email: str


class CompleteOrderPayload(BaseModel):
    order_id: str


class JoinPayload(BaseModel):
    nickname: str = ""
    gender: str = ""
    single: str = ""
    profession: str = ""
    intro: str = ""
    wechat: str = ""
    education: str = ""
    graduationYear: str = ""
    phone: str = ""
    media: str | None = None
    location: str = ""
    citySlug: str = ""
    lookingFor: list[str] = Field(default_factory=list)


class FeedbackPayload(BaseModel):
    type: str
    email: str = ""
    title: str
    content: str
    targetId: str = ""


class CityVolunteerApplicationPayload(BaseModel):
    citySlug: str = ""
    cityName: str = ""
    applicantName: str = ""
    email: str = ""
    wechat: str = ""
    phone: str = ""
    currentCity: str = ""
    availability: str = ""
    roles: list[str] = Field(default_factory=list)
    localExperience: str = ""
    motivation: str = ""
    links: str = ""


class SwipePayload(BaseModel):
    profileId: str
    action: str
    intent: str = "friends"


class ChatMessagePayload(BaseModel):
    body: str = Field(..., min_length=1, max_length=MAX_MESSAGE_LENGTH)


class DiscussionPayload(BaseModel):
    title: str
    author: str = "NomadCNA"
    city: str = ""
    tags: list[str] = Field(default_factory=list)
    excerpt: str = ""


class DiscussionReplyPayload(BaseModel):
    body: str = Field(..., min_length=1, max_length=2000)


class StartConversationPayload(BaseModel):
    targetUserId: str = ""
    profileId: str = ""
    intent: str = "friends"
    opener: str = ""


class GigPayload(BaseModel):
    title: str
    category: str = "任务"
    budget: int = 300
    location: str = "远程"
    description: str = ""


class ServiceLeadPayload(BaseModel):
    serviceSlug: str
    email: str = ""
    note: str = ""


class ContentSubmissionPayload(BaseModel):
    type: str
    title: str
    subtitle: str = ""
    description: str = ""
    coverImage: str = ""
    mediaUrl: str = ""
    authorName: str = ""
    authorEmail: str = ""
    citySlugs: list[str] = Field(default_factory=list)
    chapters: list[dict[str, Any]] = Field(default_factory=list)


class NotificationCreatePayload(BaseModel):
    title: str
    body: str = ""
    category: str = "system"
    priority: str = "normal"
    targetUserId: str = ""
    audience: str = "all"
    actionLabel: str = ""
    actionUrl: str = ""
    entityType: str = ""
    entityId: str = ""
    icon: str = "🔔"
    channels: list[str] = Field(default_factory=lambda: ["in_app"])
    metadata: dict[str, Any] = Field(default_factory=dict)


class NotificationPreferencePayload(BaseModel):
    channels: dict[str, bool] = Field(default_factory=dict)
    categories: dict[str, bool] = Field(default_factory=dict)
    quietHoursStart: str = "22:00"
    quietHoursEnd: str = "08:00"
    digest: str = "daily"
    allowMarketing: bool = False


class RecommendationPayload(BaseModel):
    budget: int = 6000
    internet: int = 50
    climate: str = "any"
    tags: list[str] = Field(default_factory=list)
    limit: int = 12


def set_session_cookie(response: Response, token: str) -> None:
    response.set_cookie(
        settings.cookie_name,
        token,
        httponly=True,
        samesite="lax",
        secure=False,
        max_age=60 * 60 * 24 * 30,
        path="/",
    )


def clear_session_cookie(response: Response) -> None:
    response.delete_cookie(settings.cookie_name, path="/")


def token_from_request(request: Request) -> str | None:
    auth = request.headers.get("authorization", "")
    if auth.lower().startswith("bearer "):
        return auth.split(" ", 1)[1].strip()
    return request.cookies.get(settings.cookie_name)


async def current_user(request: Request) -> tuple[str, dict[str, Any]] | tuple[None, None]:
    token = token_from_request(request)
    if not token:
        return None, None
    try:
        data = await pb.request("POST", "/api/collections/users/auth-refresh", token=token)
        return data.get("token") or token, data.get("record")
    except Exception:
        return None, None


async def find_user_by_email(email: str) -> dict[str, Any] | None:
    return await pb.first_record("users", filter=f"email={pb_quote(email.lower())}")


async def find_membership(user_id: str) -> dict[str, Any] | None:
    membership = await pb.first_record(
        "memberships",
        filter=f'userId={pb_quote(user_id)} && status="active"',
    )
    if membership:
        return membership
    try:
        site_vip = await pb.first_record(
            "site_vip",
            filter=f'user_id={pb_quote(user_id)} && site_id={pb_quote(settings.site_id)}',
        )
        if site_vip and not site_vip.get("expires_at"):
            return site_vip
        if site_vip and str(site_vip.get("expires_at")) > utc_now():
            return site_vip
    except Exception:
        return None
    return None


def internal_auth_password() -> str:
    password = settings.email_auth_password.strip()
    return password if len(password) >= 8 else "12345678"


def auth_password_for_email(email: str) -> str:
    if email.lower().strip() == settings.pb_admin_email.lower().strip():
        return settings.pb_admin_password
    return internal_auth_password()


async def ensure_profile_for_user(
    user: dict[str, Any],
    *,
    email: str,
    name: str = "",
    photo: str = "",
) -> None:
    profile = await pb.first_record("profiles", filter=f'userId={pb_quote(user["id"])}')
    if profile:
        return
    profile_payload = {
        "userId": user["id"],
        "email": email,
        "name": name or user.get("name") or email.split("@")[0],
        "age": 0,
        "location": "",
        "tags": ["新成员"],
        "bio": "",
        "photo": photo or avatar_for_name(email),
        "lookingFor": ["friends", "explore"],
    }
    await pb.create_record("profiles", profile_payload)


async def login_with_internal_password(email: str) -> dict[str, Any]:
    return await pb.request(
        "POST",
        "/api/collections/users/auth-with-password",
        json={"identity": email, "password": auth_password_for_email(email)},
    )


async def ensure_user(
    email: str,
    *,
    name: str = "",
    photo: str = "",
) -> tuple[dict[str, Any], str, bool]:
    normalized = email.lower().strip()
    if "@" not in normalized or "." not in normalized.split("@")[-1]:
        raise HTTPException(status_code=400, detail="请输入有效邮箱")
    existing = await find_user_by_email(normalized)
    if existing:
        update_payload: dict[str, Any] = {
            "password": auth_password_for_email(normalized),
            "passwordConfirm": auth_password_for_email(normalized),
        }
        if name and not existing.get("name"):
            update_payload["name"] = name
        await pb.update_record("users", existing["id"], update_payload)
        try:
            auth = await login_with_internal_password(normalized)
        except Exception:
            raise HTTPException(status_code=401, detail="邮箱登录失败")
        await ensure_profile_for_user(auth["record"], email=normalized, name=name, photo=photo)
        return auth["record"], auth["token"], False

    created = await pb.create_record(
        "users",
        {
            "email": normalized,
            "emailVisibility": True,
            "verified": True,
            "password": auth_password_for_email(normalized),
            "passwordConfirm": auth_password_for_email(normalized),
            "name": name or normalized.split("@")[0],
        },
    )
    await ensure_profile_for_user(created, email=normalized, name=name, photo=photo)
    auth = await login_with_internal_password(normalized)
    return auth["record"], auth["token"], True


async def verify_google_id_token(id_token: str) -> dict[str, str]:
    if not settings.google_client_id:
        raise HTTPException(status_code=400, detail="未配置 Google 登录")
    token = id_token.strip()
    if not token:
        raise HTTPException(status_code=400, detail="缺少 Google token")
    try:
        async with httpx.AsyncClient(timeout=10.0, trust_env=False) as client:
            res = await client.get("https://oauth2.googleapis.com/tokeninfo", params={"id_token": token})
        data = res.json() if res.is_success else {}
    except Exception:
        raise HTTPException(status_code=401, detail="Google 登录验证失败")
    if data.get("aud") != settings.google_client_id:
        raise HTTPException(status_code=401, detail="Google 登录来源不匹配")
    verified = data.get("email_verified")
    if verified not in {True, "true", "1", 1}:
        raise HTTPException(status_code=401, detail="Google 邮箱未验证")
    email = str(data.get("email") or "").strip().lower()
    if not email:
        raise HTTPException(status_code=401, detail="Google 账号缺少邮箱")
    return {
        "email": email,
        "name": str(data.get("name") or email.split("@")[0]),
        "photo": str(data.get("picture") or ""),
    }


def public_record(record: dict[str, Any]) -> dict[str, Any]:
    return {k: v for k, v in record.items() if not k.startswith("@")}


MATCH_INTENTS = ("friends", "dating", "partner", "roommate", "cofounder", "explore")
LIKE_ACTIONS = frozenset({"like", "right", "superlike"})
FREE_SWIPE_DAILY_LIMIT = 25
MEMBERSHIP_PAY_TYPES = frozenset({"meetup", "vip", "join"})
PAY_AMOUNT_TOLERANCE_CENTS = 1


def profile_looking_for(record: dict[str, Any]) -> list[str]:
    raw = record.get("lookingFor")
    if isinstance(raw, list) and raw:
        cleaned = [str(item).strip() for item in raw if str(item).strip() in MATCH_INTENTS]
        if cleaned:
            return cleaned
    return ["friends", "explore"]


async def profile_for_user(user_id: str) -> dict[str, Any] | None:
    if not user_id:
        return None
    return await pb.first_record("profiles", filter=f'userId={pb_quote(user_id)}')


async def swiped_profile_ids(user_id: str) -> set[str]:
    data = await pb.list_records("swipes", filter=f'userId={pb_quote(user_id)}', per_page=200)
    return {str(item.get("profileId")) for item in data.get("items", []) if item.get("profileId")}


def enrich_profile(record: dict[str, Any]) -> dict[str, Any]:
    payload = public_record(record)
    payload["lookingFor"] = profile_looking_for(record)
    payload["gender"] = record.get("gender") or ""
    payload["single"] = record.get("single") or ""
    payload["citySlug"] = record.get("citySlug") or ""
    return payload


async def has_reciprocal_like(target_user_id: str, my_profile_id: str) -> bool:
    data = await pb.list_records(
        "swipes",
        filter=f'userId={pb_quote(target_user_id)} && profileId={pb_quote(my_profile_id)}',
        per_page=20,
    )
    return any(str(item.get("action")) in LIKE_ACTIONS for item in data.get("items", []))


async def ensure_match_connection(
    *,
    user_a_id: str,
    user_b_id: str,
    profile_a_id: str,
    profile_b_id: str,
    intent: str,
) -> dict[str, Any]:
    pair_key = "|".join(sorted([user_a_id, user_b_id]))
    existing = await pb.first_record("match_connections", filter=f'pairKey={pb_quote(pair_key)}')
    if existing:
        conversation = await sync_connection_conversation(existing)
        if conversation and str(existing.get("conversationId") or "") != str(conversation.get("id") or ""):
            existing = await pb.update_record(
                "match_connections",
                existing["id"],
                {"conversationId": conversation["id"]},
            )
        return existing
    connection = await pb.create_record(
        "match_connections",
        {
            "pairKey": pair_key,
            "userAId": user_a_id,
            "userBId": user_b_id,
            "profileAId": profile_a_id,
            "profileBId": profile_b_id,
            "intent": intent or "friends",
            "matchedAt": utc_now(),
        },
    )
    conversation = await ensure_conversation_for_match(
        user_a_id=user_a_id,
        user_b_id=user_b_id,
        match_connection_id=str(connection.get("id") or ""),
        intent=intent or "friends",
    )
    return await pb.update_record(
        "match_connections",
        connection["id"],
        {"conversationId": conversation["id"]},
    )


def chat_action_url(conversation_id: str) -> str:
    return f"/chat/{conversation_id}"


async def peer_profile_for_conversation(conversation: dict[str, Any], user_id: str) -> dict[str, Any] | None:
    other_user_id = other_participant_id(conversation, user_id)
    if not other_user_id:
        return None
    profile = await profile_for_user(other_user_id)
    if not profile:
        return None
    return enrich_profile(profile)


async def decorate_conversation(conversation: dict[str, Any], user_id: str) -> dict[str, Any]:
    peer = await peer_profile_for_conversation(conversation, user_id)
    payload = dm_public_record(conversation)
    payload["peer"] = peer
    payload["unreadCount"] = unread_count_for(conversation, user_id)
    return payload


def profile_matches_city(record: dict[str, Any], city_query: str) -> bool:
    query = city_query.strip().lower()
    if not query or query in {"all", "any"}:
        return True
    location = str(record.get("location") or "").lower()
    slug = str(record.get("citySlug") or "").lower()
    return query in location or query == slug


def profile_matches_advanced(record: dict[str, Any], gender: str, single: str) -> bool:
    if gender and gender not in {"", "all", "不限"}:
        if str(record.get("gender") or "") != gender:
            return False
    if single and single not in {"", "all", "不限"}:
        if str(record.get("single") or "") != single:
            return False
    return True


def compute_is_upcoming(record: dict[str, Any]) -> bool:
    date_part = str(record.get("date") or "")[:10]
    if not date_part:
        return bool(record.get("isUpcoming", True))
    time_part = str(record.get("time") or "23:59")[:5]
    try:
        event_at = datetime.strptime(f"{date_part} {time_part}", "%Y-%m-%d %H:%M").replace(
            tzinfo=ZoneInfo("Asia/Shanghai")
        )
        return event_at > datetime.now(ZoneInfo("Asia/Shanghai"))
    except ValueError:
        return bool(record.get("isUpcoming", True))


def public_meetup_record(record: dict[str, Any]) -> dict[str, Any]:
    payload = public_record(record)
    payload["isUpcoming"] = compute_is_upcoming(record)
    return payload


async def swipe_count_today(user_id: str) -> int:
    today_prefix = datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y-%m-%d")
    data = await pb.list_records(
        "swipes",
        filter=f'userId={pb_quote(user_id)} && created >= "{today_prefix} 00:00:00"',
        per_page=200,
        sort="-created",
    )
    return len(data.get("items", []))


def utc_now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.000Z")


def clean_submission_text(value: Any, limit: int, *, multiline: bool = False) -> str:
    text = str(value or "").strip()
    if not multiline:
        text = re.sub(r"\s+", " ", text)
    else:
        text = "\n".join(line.strip() for line in text.splitlines()).strip()
    return text[:limit]


def slugify_content_title(value: str) -> str:
    slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-")
    return slug[:70].strip("-") or "submission"


def validate_submission_url(value: str, field: str) -> str:
    url = str(value or "").strip()
    if not url:
        raise HTTPException(status_code=400, detail=f"缺少{field}")
    if len(url) > 500:
        raise HTTPException(status_code=400, detail=f"{field}链接过长")
    if not (url.startswith("https://") or url.startswith("http://") or url.startswith("/")):
        raise HTTPException(status_code=400, detail=f"{field}必须是有效链接")
    return url


def normalize_submission_chapters(chapters: list[dict[str, Any]], content_type: str) -> list[dict[str, Any]]:
    normalized: list[dict[str, Any]] = []
    for chapter in chapters[:30]:
        title = clean_submission_text(chapter.get("title", ""), 160)
        if not title:
            continue
        item: dict[str, Any] = {"title": title}
        if content_type == "video":
            time = clean_submission_text(chapter.get("time", "") or chapter.get("minutes", ""), 20)
            if time:
                item["time"] = time
        else:
            try:
                minutes = int(chapter.get("minutes") or 8)
            except Exception:
                minutes = 8
            item["minutes"] = max(1, min(minutes, 999))
        normalized.append(item)
    return normalized


def parse_order_id(order_id: str) -> tuple[str, str]:
    if "_order_" not in order_id:
        return "", "unknown"
    prefix = order_id.split("_order_", 1)[0]
    if "_" not in prefix:
        return prefix, "unknown"
    user_id, pay_type = prefix.rsplit("_", 1)
    return user_id, pay_type.lower()


def decode_pending_email(user_id: str) -> str | None:
    if not user_id.startswith("pending_"):
        return None
    try:
        return bytes.fromhex(user_id.removeprefix("pending_")).decode("utf-8")
    except Exception:
        return None


def client_ip_from_request(request: Request) -> str:
    forwarded = request.headers.get("x-forwarded-for") or ""
    if forwarded:
        return forwarded.split(",", 1)[0].strip()
    real_ip = request.headers.get("x-real-ip")
    if real_ip:
        return real_ip.strip()
    return request.client.host if request.client else ""


def append_order_id(return_url: str, order_id: str) -> str:
    if not return_url:
        return f"/zh/join/paid?order_id={order_id}"
    if "order_id=" in return_url or "out_trade_no=" in return_url:
        return return_url
    sep = "&" if "?" in return_url else "?"
    return f"{return_url}{sep}order_id={order_id}"


def payment_name(pay_type: str) -> str:
    return {
        "book": "购买书籍",
        "meetup": "数字游民社区报名",
        "salon": "沙龙报名茶歇费",
        "video": "视频解锁",
        "vip": "VIP会员充值",
    }.get(pay_type, "商品支付")


def resolve_order_amount(pay_type: str) -> int:
    normalized = pay_type.lower().strip()
    if normalized == settings.payment_join_type.lower():
        return settings.payment_join_amount
    return settings.payment_default_amount


def pay_type_grants_membership(pay_type: str) -> bool:
    normalized = pay_type.lower().strip()
    return normalized in MEMBERSHIP_PAY_TYPES or normalized == settings.payment_join_type.lower()


def pay_amount_matches(order: dict[str, Any], pay_price: str, *, tolerance_cents: int = PAY_AMOUNT_TOLERANCE_CENTS) -> bool:
    if settings.dev_auto_pay:
        return True
    expected = int(order.get("totalFee") or 0)
    if expected <= 0:
        return True
    try:
        paid_cents = int(round(float(pay_price) * 100))
    except (TypeError, ValueError):
        return False
    return abs(paid_cents - expected) <= tolerance_cents


def pay_order_cookie_matches(request: Request, order_id: str) -> bool:
    raw = str(request.cookies.get("join_pay_order") or "")
    cookie_order_id = raw.split("|", 1)[0].strip()
    return bool(cookie_order_id and cookie_order_id == order_id)


async def assert_can_complete_order(
    request: Request,
    order_id: str,
    order: dict[str, Any] | None,
) -> None:
    order_user_id = str((order or {}).get("userId") or parse_order_id(order_id)[0] or "")
    _, user = await current_user(request)
    if order_user_id and not order_user_id.startswith("pending_"):
        if not user or str(user.get("id")) != order_user_id:
            raise HTTPException(status_code=403, detail="无权操作该订单")
        return
    order_email = str((order or {}).get("email") or decode_pending_email(order_user_id) or "").lower().strip()
    if user and order_email and str(user.get("email") or "").lower().strip() == order_email:
        return
    if pay_order_cookie_matches(request, order_id):
        return
    raise HTTPException(status_code=403, detail="无权操作该订单")


def set_pay_order_cookie(response: Response, order_id: str) -> None:
    response.set_cookie(
        "join_pay_order",
        f"{order_id}|{int(datetime.now().timestamp() * 1000)}",
        path="/",
        max_age=900,
        httponly=True,
        samesite="lax",
    )


def build_auto_submit_html(pay_url: str, params: dict[str, Any], order_id: str) -> str:
    inputs = "\n".join(
        f'<input type="hidden" name="{escape(str(key), quote=True)}" value="{escape(str(value), quote=True)}" />'
        for key, value in params.items()
        if value is not None and str(value).strip() != ""
    )
    order_value = escape(f"{order_id}|{int(datetime.now().timestamp() * 1000)}", quote=True)
    return f"""<!doctype html><html><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>跳转支付</title></head>
<body style="font-family:system-ui;display:flex;align-items:center;justify-content:center;min-height:100vh;margin:0;background:#f0f4f8;color:#64748b">
<p>正在跳转到支付页面...</p>
<form id="payForm" method="POST" action="{escape(pay_url, quote=True)}">{inputs}</form>
<script>
try{{localStorage.setItem("join_pay_order","{order_value}");}}catch(e){{}}
document.cookie="join_pay_order="+encodeURIComponent("{order_value}")+"; path=/; max-age=900";
document.getElementById("payForm").submit();
</script></body></html>"""


def build_redirect_html(pay_url: str, order_id: str) -> str:
    order_value = escape(f"{order_id}|{int(datetime.now().timestamp() * 1000)}", quote=True)
    return f"""<!doctype html><html><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>跳转支付</title></head>
<body style="font-family:system-ui;display:flex;align-items:center;justify-content:center;min-height:100vh;margin:0;background:#f0f4f8;color:#64748b">
<p>正在跳转到支付页面...</p>
<script>
try{{localStorage.setItem("join_pay_order","{order_value}");}}catch(e){{}}
document.cookie="join_pay_order="+encodeURIComponent("{order_value}")+"; path=/; max-age=900";
location.href={pay_url!r};
</script></body></html>"""


def build_qr_html(img_src: str, return_url: str, order_id: str) -> str:
    order_value = escape(f"{order_id}|{int(datetime.now().timestamp() * 1000)}", quote=True)
    return f"""<!doctype html><html><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>微信扫码支付</title>
<style>body{{font-family:system-ui,-apple-system,sans-serif;display:flex;align-items:center;justify-content:center;min-height:100vh;margin:0;background:linear-gradient(135deg,#667eea,#764ba2)}}.card{{background:#fff;border-radius:20px;padding:36px;text-align:center;box-shadow:0 20px 50px rgba(0,0,0,.25)}}img{{width:220px;height:220px;border-radius:12px;border:1px solid #e2e8f0}}p{{color:#64748b}}</style></head>
<body><div class="card"><h2>微信扫码支付</h2><p id="tip">请使用微信扫描二维码完成支付</p><img src="{escape(img_src, quote=True)}" alt="支付二维码" /></div>
<script>
try{{localStorage.setItem("join_pay_order","{order_value}");}}catch(e){{}}
document.cookie="join_pay_order="+encodeURIComponent("{order_value}")+"; path=/; max-age=900";
var orderId={order_id!r};var returnUrl={return_url!r};
function poll(){{fetch("/api/pay/status?order_id="+encodeURIComponent(orderId),{{cache:"no-store"}}).then(function(r){{return r.json()}}).then(function(d){{if(d&&d.paid){{document.getElementById("tip").textContent="支付成功，正在返回...";location.href=returnUrl;return;}}setTimeout(poll,1000);}}).catch(function(){{setTimeout(poll,1500);}})}}
setTimeout(poll,1000);
</script></body></html>"""


async def find_order(order_id: str) -> dict[str, Any] | None:
    return await pb.first_record("orders", filter=f'orderId={pb_quote(order_id)}')


async def upsert_order_record(
    *,
    order_id: str,
    user_id: str,
    email: str = "",
    total_fee: int = 0,
    pay_type: str = "meetup",
    channel: str = "wxpay",
    status: str = "pending",
    return_url: str = "",
) -> dict[str, Any] | None:
    payload = {
        "orderId": order_id,
        "userId": user_id,
        "email": email,
        "totalFee": total_fee,
        "type": pay_type,
        "channel": channel,
        "status": status,
        "returnUrl": return_url,
        "isNew": user_id.startswith("pending_"),
    }
    existing = await find_order(order_id)
    if existing:
        return await pb.update_record("orders", existing["id"], payload)
    return await pb.create_record("orders", payload)


async def mark_order_paid(order_id: str, pay_price: str = "") -> dict[str, Any] | None:
    order = await find_order(order_id)
    if order:
        return await pb.update_record("orders", order["id"], {"status": "paid"})
    user_id, pay_type = parse_order_id(order_id)
    total_fee = int(float(pay_price or 0) * 100) if pay_price else 0
    email = decode_pending_email(user_id) or ""
    return await upsert_order_record(
        order_id=order_id,
        user_id=user_id,
        email=email,
        total_fee=total_fee,
        pay_type=pay_type,
        status="paid",
    )


async def query_gateway_paid(order_id: str) -> tuple[bool, str]:
    zpay, xorpay = await asyncio.gather(
        ZPayProvider().query_order_status(order_id),
        XorPayProvider().query_order_status(order_id),
    )
    for result in (zpay, xorpay):
        if result.get("paid"):
            return True, str(result.get("pay_price") or "")
    return False, ""


async def ensure_site_vip(user_id: str, order_id: str) -> None:
    try:
        payload = {"user_id": user_id, "site_id": settings.site_id, "order_id": order_id, "expires_at": ""}
        existing = await pb.first_record("site_vip", filter=f'user_id={pb_quote(user_id)} && site_id={pb_quote(settings.site_id)}')
        if existing:
            await pb.update_record("site_vip", existing["id"], payload)
        else:
            await pb.create_record("site_vip", payload)
    except Exception:
        return


async def ensure_paid_membership(user: dict[str, Any], order_id: str) -> None:
    expires = (datetime.now(timezone.utc) + timedelta(days=365)).strftime("%Y-%m-%d %H:%M:%S.000Z")
    payload_membership = {
        "userId": user["id"],
        "email": user.get("email", ""),
        "plan": "pro",
        "status": "active",
        "expiresAt": expires,
    }
    membership = await pb.first_record("memberships", filter=f'userId={pb_quote(user["id"])}')
    if membership:
        await pb.update_record("memberships", membership["id"], payload_membership)
    else:
        await pb.create_record("memberships", payload_membership)
    await ensure_site_vip(user["id"], order_id)


async def record_payment(user: dict[str, Any], order: dict[str, Any], order_id: str, pay_price: str = "") -> None:
    try:
        amount = int(float(pay_price or 0) * 100) if pay_price else int(order.get("totalFee") or 0)
        existing = await pb.first_record("payments", filter=f'order_id={pb_quote(order_id)}')
        payload = {
            "user_id": user["id"],
            "type": order.get("type") or parse_order_id(order_id)[1],
            "order_id": order_id,
            "amount": amount,
            "total_fee": amount,
            "channel": order.get("channel") or "wxpay",
        }
        if existing:
            await pb.update_record("payments", existing["id"], payload)
        else:
            await pb.create_record("payments", payload)
    except Exception:
        return


async def user_for_paid_order(order: dict[str, Any], order_id: str) -> tuple[dict[str, Any] | None, str | None, bool]:
    user_id = order.get("userId") or parse_order_id(order_id)[0]
    if user_id.startswith("pending_"):
        email = order.get("email") or decode_pending_email(user_id)
        if not email:
            return None, None, False
        user, token, is_new = await ensure_user(email)
        return user, token, is_new
    if not user_id:
        return None, None, False
    user = await pb.first_record("users", filter=f'id={pb_quote(user_id)}')
    return user, None, False


async def order_already_fulfilled(order_id: str) -> bool:
    if order_id in processed_orders:
        return True
    existing = await pb.first_record("site_vip", filter=f'order_id={pb_quote(order_id)}')
    return bool(existing)


async def finalize_paid_order(order_id: str, pay_price: str = "") -> tuple[dict[str, Any] | None, str | None, bool]:
    if not order_id:
        return None, None, False
    order = await find_order(order_id)
    if order and await order_already_fulfilled(order_id):
        user, token, is_new = await user_for_paid_order(order, order_id)
        processed_orders.add(order_id)
        return user, token, is_new
    if order and pay_price and not pay_amount_matches(order, pay_price):
        return None, None, False
    order = await mark_order_paid(order_id, pay_price)
    if not order:
        return None, None, False
    if pay_price and not pay_amount_matches(order, pay_price):
        return None, None, False
    user, token, is_new = await user_for_paid_order(order, order_id)
    if user:
        await record_payment(user, order, order_id, pay_price)
        pay_type = str(order.get("type") or parse_order_id(order_id)[1])
        if pay_type_grants_membership(pay_type):
            await ensure_paid_membership(user, order_id)
        processed_orders.add(order_id)
    return user, token, is_new


def default_notification_preferences(user_id: str, email: str = "") -> dict[str, Any]:
    return {
        "userId": user_id,
        "email": email,
        "channels": {"in_app": True, "push": False, "email": False, "wechat": False},
        "categories": {
            "system": True,
            "meetup": True,
            "gig": True,
            "match": True,
            "service": True,
            "community": True,
            "city": True,
        },
        "quietHoursStart": "22:00",
        "quietHoursEnd": "08:00",
        "digest": "daily",
        "allowMarketing": False,
    }


def normalize_notification_preferences(record: dict[str, Any] | None, user_id: str, email: str = "") -> dict[str, Any]:
    defaults = default_notification_preferences(user_id, email)
    if not record:
        return defaults
    public = public_record(record)
    return {
        **defaults,
        **public,
        "channels": {**defaults["channels"], **(public.get("channels") or {})},
        "categories": {**defaults["categories"], **(public.get("categories") or {})},
    }


async def notification_preferences_for_user(user_id: str, user: dict[str, Any] | None = None) -> dict[str, Any]:
    record = await pb.first_record("notification_preferences", filter=f'userId={pb_quote(user_id)}')
    return normalize_notification_preferences(record, user_id, user.get("email", "") if user else "")


def notification_is_muted(item: dict[str, Any], preferences: dict[str, Any]) -> bool:
    category = item.get("category") or "system"
    categories = preferences.get("categories") or {}
    channels = preferences.get("channels") or {}
    item_channels = item.get("channels") or ["in_app"]
    category_enabled = bool(categories.get(category, True))
    channel_enabled = any(bool(channels.get(channel, False)) for channel in item_channels)
    return not category_enabled or not channel_enabled


async def create_notification(
    *,
    title: str,
    body: str = "",
    category: str = "system",
    priority: str = "normal",
    target_user_id: str = "",
    audience: str = "all",
    action_label: str = "",
    action_url: str = "",
    entity_type: str = "",
    entity_id: str = "",
    icon: str = "🔔",
    channels: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any] | None:
    try:
        record = await pb.create_record(
            "notifications",
            {
                "title": title,
                "body": body,
                "category": category,
                "priority": priority,
                "targetUserId": target_user_id,
                "audience": audience,
                "actionLabel": action_label,
                "actionUrl": action_url,
                "entityType": entity_type,
                "entityId": entity_id,
                "icon": icon,
                "status": "published",
                "channels": channels or ["in_app"],
                "metadata": metadata or {},
            },
        )
    except PocketBaseError:
        return None
    if target_user_id:
        try:
            prefs = await notification_preferences_for_user(target_user_id)
            await dispatch_ntfy_for_user(
                user_id=target_user_id,
                preferences=prefs,
                title=title,
                body=body,
                category=category,
                priority=priority,
                action_url=action_url,
                icon=icon,
            )
        except Exception:
            pass
    return record


async def receipt_for(notification_id: str, user_id: str) -> dict[str, Any] | None:
    if not user_id:
        return None
    return await pb.first_record(
        "notification_receipts",
        filter=f'notificationId={pb_quote(notification_id)} && userId={pb_quote(user_id)}',
    )


async def upsert_receipt(notification_id: str, user_id: str, payload: dict[str, Any]) -> dict[str, Any]:
    existing = await receipt_for(notification_id, user_id)
    base = {"notificationId": notification_id, "userId": user_id, "pinned": False}
    if existing:
        return await pb.update_record("notification_receipts", existing["id"], payload)
    return await pb.create_record("notification_receipts", {**base, **payload})


async def decorated_notifications_for_user(
    user_id: str,
    preferences: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    preferences = preferences or await notification_preferences_for_user(user_id)
    data = await pb.list_records("notifications", filter='status="published"', per_page=100)
    notifications = [
        public_record(item)
        for item in data.get("items", [])
        if item.get("audience", "all") == "all" or item.get("targetUserId") == user_id
    ]
    receipts = (await pb.list_records("notification_receipts", filter=f'userId={pb_quote(user_id)}', per_page=200)).get("items", [])
    receipt_by_notification = {receipt.get("notificationId"): public_record(receipt) for receipt in receipts}
    decorated = []
    for item in notifications:
        receipt = receipt_by_notification.get(item["id"], {})
        archived = bool(receipt.get("archivedAt"))
        decorated.append(
            {
                **item,
                "receipt": receipt,
                "read": bool(receipt.get("readAt")),
                "archived": archived,
                "pinned": bool(receipt.get("pinned")),
                "muted": notification_is_muted(item, preferences),
            }
        )
    decorated.sort(key=lambda item: (item.get("pinned", False), item.get("created", "")), reverse=True)
    return decorated


@app.get("/api/health")
async def health() -> dict[str, Any]:
    return {"ok": True, "service": "NomadCNA FastAPI", "pb": settings.pb_url}


@app.post("/api/auth/email")
async def email_login(payload: AuthPayload, response: Response) -> dict[str, Any]:
    user, token, is_new = await ensure_user(payload.email)
    set_session_cookie(response, token)
    return {"ok": True, "token": token, "record": user, "user": user, "is_new": is_new}


@app.post("/api/auth/google")
async def google_login(payload: GoogleAuthPayload, response: Response) -> dict[str, Any]:
    profile = await verify_google_id_token(payload.id_token)
    user, token, is_new = await ensure_user(profile["email"], name=profile["name"], photo=profile["photo"])
    set_session_cookie(response, token)
    return {"ok": True, "token": token, "record": user, "user": user, "is_new": is_new, "provider": "google"}


@app.post("/api/auth/sync-session")
async def sync_session(payload: SyncSessionPayload, response: Response) -> dict[str, Any]:
    token = payload.token.strip()
    if not token:
        raise HTTPException(status_code=400, detail="缺少 token")
    try:
        data = await pb.request("POST", "/api/collections/users/auth-refresh", token=token)
    except Exception:
        raise HTTPException(status_code=401, detail="无效 token")
    refreshed = data.get("token") or token
    set_session_cookie(response, refreshed)
    return {"ok": True, "token": refreshed, "user": data.get("record")}


@app.get("/api/auth/me")
async def me(request: Request, response: Response) -> dict[str, Any]:
    token, user = await current_user(request)
    if not user:
        return {"ok": True, "user": None}
    set_session_cookie(response, token)
    vip = bool(await find_membership(user["id"]))
    return {"ok": True, "token": token, "user": {**user, "vip": vip}}


@app.post("/api/auth/logout")
async def logout(response: Response) -> dict[str, Any]:
    clear_session_cookie(response)
    return {"ok": True}


@app.post("/api/meetup/check-user")
async def check_user(payload: CheckUserPayload) -> dict[str, Any]:
    user = await find_user_by_email(str(payload.email))
    vip = bool(await find_membership(user["id"])) if user else False
    return {"ok": True, "exists": bool(user), "vip": vip, "user_id": user["id"] if user else None}


@app.post("/api/meetup/ensure-user")
async def ensure_user_endpoint(payload: EnsureUserPayload, response: Response) -> dict[str, Any]:
    user, token, is_new = await ensure_user(str(payload.email))
    set_session_cookie(response, token)
    vip = bool(await find_membership(user["id"]))
    return {"ok": True, "user_id": user["id"], "record": user, "token": token, "is_new": is_new, "vip": vip}


@app.post("/api/meetup/complete-order")
async def complete_order(payload: CompleteOrderPayload, request: Request, response: Response) -> dict[str, Any]:
    order_id = payload.order_id.strip()
    if not order_id:
        raise HTTPException(status_code=400, detail="缺少 order_id")
    order = await find_order(order_id)
    await assert_can_complete_order(request, order_id, order)
    if order and await order_already_fulfilled(order_id):
        user, token, is_new = await user_for_paid_order(order, order_id)
        if not user:
            raise HTTPException(status_code=500, detail="会员开通失败")
        if token:
            set_session_cookie(response, token)
        return {"ok": True, "token": token, "record": user, "is_new": is_new}
    if not order:
        paid, pay_price = await query_gateway_paid(order_id)
        if not paid and not settings.dev_auto_pay:
            raise HTTPException(status_code=400, detail="订单不存在或未支付成功")
        order = await find_order(order_id)
        if order and pay_price and not pay_amount_matches(order, pay_price):
            raise HTTPException(status_code=400, detail="支付金额与订单不符")
        order = await mark_order_paid(order_id, pay_price)
    elif order.get("status") != "paid":
        paid, pay_price = await query_gateway_paid(order_id)
        if not paid and not settings.dev_auto_pay:
            raise HTTPException(status_code=400, detail="订单未支付成功")
        if pay_price and not pay_amount_matches(order, pay_price):
            raise HTTPException(status_code=400, detail="支付金额与订单不符")
        order = await mark_order_paid(order_id, pay_price)
    if not order:
        raise HTTPException(status_code=500, detail="订单处理失败")

    user, token, is_new = await finalize_paid_order(order_id)
    if not user:
        raise HTTPException(status_code=500, detail="会员开通失败")
    if token:
        set_session_cookie(response, token)
    return {"ok": True, "token": token, "record": user, "is_new": is_new}


@app.get("/api/vip/check")
async def vip_check(request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        return {"ok": True, "vip": False}
    membership = await find_membership(user["id"])
    return {"ok": True, "vip": bool(membership), "membership": membership}


@app.post("/api/join")
async def join(payload: JoinPayload, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        raise HTTPException(status_code=401, detail="请先登录")
    email = user.get("email", "")
    record_payload = {
        **payload.model_dump(),
        "userId": user["id"],
        "email": email,
        "isComplete": True,
    }
    existing = await pb.first_record("member_applications", filter=f'userId={pb_quote(user["id"])}')
    if existing:
        record = await pb.update_record("member_applications", existing["id"], record_payload)
    else:
        record = await pb.create_record("member_applications", record_payload)
    profile = await pb.first_record("profiles", filter=f'userId={pb_quote(user["id"])}')
    cleaned_looking_for = [item for item in payload.lookingFor if item in MATCH_INTENTS]
    profile_payload = {
        "userId": user["id"],
        "email": email,
        "name": payload.nickname or user.get("name") or email.split("@")[0],
        "location": payload.location,
        "citySlug": payload.citySlug,
        "tags": [payload.profession, payload.education, "数字游民"],
        "bio": payload.intro,
        "photo": payload.media or avatar_for_name(email),
        "gender": payload.gender,
        "single": payload.single,
        "lookingFor": cleaned_looking_for or ["friends", "explore"],
    }
    if profile:
        await pb.update_record("profiles", profile["id"], profile_payload)
    else:
        await pb.create_record("profiles", profile_payload)
    await create_notification(
        title="成员资料已更新",
        body="你的加入资料已经保存，后续匹配、活动推荐和城市建议会基于这份资料联动。",
        category="system",
        priority="normal",
        target_user_id=user["id"],
        audience="user",
        action_label="查看资料",
        action_url="/join",
        entity_type="member_application",
        entity_id=record["id"],
        icon="👤",
    )
    return {"ok": True, "record": record, "user_id": user["id"]}


@app.get("/api/join/status")
async def join_status(request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        return {"ok": True, "hasRecord": False, "isComplete": False}
    record = await pb.first_record("member_applications", filter=f'userId={pb_quote(user["id"])}')
    return {"ok": True, "hasRecord": bool(record), "isComplete": bool(record and record.get("isComplete"))}


@app.post("/api/upload-media")
async def upload_media(request: Request, file: UploadFile = File(...)) -> dict[str, Any]:
    purpose = (request.query_params.get("purpose") or "").lower()
    is_content_upload = purpose in {"content", "ebook", "video", "submission"}
    max_size = 200 * 1024 * 1024 if is_content_upload else 20 * 1024 * 1024
    if file.size and file.size > max_size:
        raise HTTPException(status_code=400, detail=f"文件大小不能超过 {max_size // 1024 // 1024}MB")
    upload = await upload_file_to_minio(file, purpose=purpose or "uploads", max_size=max_size)
    user_id = ""
    _, user = await current_user(request)
    user_id = user["id"] if user else ""
    await pb.create_record(
        "media_assets",
        {
            "userId": user_id,
            "url": upload.url,
            "filename": file.filename or upload.object_key,
            "contentType": file.content_type or "application/octet-stream",
            "size": upload.size,
            "objectKey": upload.object_key,
        },
    )
    return {"ok": True, "url": upload.url, "objectKey": upload.object_key}


@app.post("/api/pay", response_model=None)
async def create_pay(request: Request, response: Response) -> HTMLResponse | dict[str, Any]:
    content_type = request.headers.get("content-type", "")
    if "application/json" in content_type:
        body = await request.json()
    else:
        form = await request.form()
        body = dict(form)

    user_id = str(body.get("user_id") or "").strip()
    if not user_id:
        raise HTTPException(status_code=400, detail="缺少 user_id")
    pay_type = str(body.get("type") or settings.payment_join_type or "meetup").lower()
    total_fee = resolve_order_amount(pay_type)
    channel = str(body.get("channel") or "wxpay")
    device = str(body.get("device") or "pc")
    requested_provider = str(body.get("provider") or "")
    want_html = str(body.get("html") or "0") == "1" or "text/html" in (request.headers.get("accept") or "")
    order_id = f"{user_id}_{pay_type}_order_{int(datetime.now().timestamp())}_{secrets.token_hex(4)}"
    return_url = append_order_id(str(body.get("return_url") or "/zh/join/paid"), order_id)
    email = decode_pending_email(user_id) or ""
    await upsert_order_record(
        order_id=order_id,
        user_id=user_id,
        email=email,
        total_fee=total_fee,
        pay_type=pay_type,
        channel=channel,
        status="paid" if settings.dev_auto_pay else "pending",
        return_url=return_url,
    )
    set_pay_order_cookie(response, order_id)

    if settings.dev_auto_pay:
        redirect_url = return_url
        html_response = HTMLResponse(build_redirect_html(redirect_url, order_id))
        set_pay_order_cookie(html_response, order_id)
        return html_response if want_html else {"ok": True, "order_id": order_id, "paid": True}

    amount_yuan = f"{total_fee / 100:.2f}"
    provider = provider_for(device, requested_provider, request.headers.get("user-agent", ""))
    notify_url = f"{settings.base_url}/api/pay/{provider.name}_notify"
    name = payment_name(pay_type)
    client_ip = client_ip_from_request(request)

    if isinstance(provider, XorPayProvider):
        created = provider.create_order(order_id=order_id, name=name, amount_yuan=amount_yuan, notify_url=notify_url)
        if not want_html:
            return {"ok": True, "order_id": order_id, "provider": provider.name, "params": created["params"], "payUrl": created["pay_url"]}
        html_response = HTMLResponse(build_auto_submit_html(created["pay_url"], created["params"], order_id))
        set_pay_order_cookie(html_response, order_id)
        return html_response

    api_result = await provider.create_order_api(
        order_id=order_id,
        name=name,
        amount_yuan=amount_yuan,
        channel=channel,
        notify_url=notify_url,
        return_url=return_url,
        client_ip=client_ip,
        device=device,
    )
    if api_result.get("status") == "ok":
        if not want_html:
            return {"ok": True, "order_id": order_id, "provider": provider.name, **api_result}
        if device == "pc" and api_result.get("img"):
            html_response = HTMLResponse(build_qr_html(str(api_result["img"]), return_url, order_id))
            set_pay_order_cookie(html_response, order_id)
            return html_response
        pay_url = api_result.get("payurl2") if device == "h5" and api_result.get("payurl2") else api_result.get("payurl")
        if pay_url:
            html_response = HTMLResponse(build_redirect_html(str(pay_url), order_id))
            set_pay_order_cookie(html_response, order_id)
            return html_response

    created = provider.create_order(
        order_id=order_id,
        name=name,
        amount_yuan=amount_yuan,
        channel=channel,
        notify_url=notify_url,
        return_url=return_url,
        client_ip=client_ip,
    )
    if not want_html:
        return {"ok": True, "order_id": order_id, "provider": provider.name, "params": created["params"], "payUrl": created["pay_url"]}
    html_response = HTMLResponse(build_auto_submit_html(created["pay_url"], created["params"], order_id))
    set_pay_order_cookie(html_response, order_id)
    return html_response


@app.get("/api/pay/status")
async def pay_status(order_id: str) -> dict[str, Any]:
    order = await find_order(order_id)
    if order and order.get("status") == "paid":
        return {"ok": True, "paid": True, "order": order}
    paid, pay_price = await query_gateway_paid(order_id)
    if paid and order and pay_price and not pay_amount_matches(order, pay_price):
        return {"ok": True, "paid": False, "order": order, "error": "amount_mismatch"}
    if paid:
        order = await mark_order_paid(order_id, pay_price)
    return {"ok": True, "paid": paid, "order": order}


@app.get("/cnomadcna/order_status")
async def cnomadcna_order_status(order_id: str = "", out_trade_no: str = "") -> dict[str, Any]:
    target = (order_id or out_trade_no).strip()
    if not target:
        return {"paid": False, "error": "missing order_id"}
    result = await pay_status(target)
    return {"paid": bool(result.get("paid")), "order": result.get("order")}


@app.get("/cnomadcna/zpay_order_status")
async def cnomadcna_zpay_order_status(out_trade_no: str) -> dict[str, Any]:
    return await ZPayProvider().query_order_status(out_trade_no)


@app.api_route("/api/pay/zpay_notify", methods=["GET", "POST"])
@app.api_route("/cnomadcna/zpay_notify", methods=["GET", "POST"])
async def zpay_notify(request: Request) -> FastAPIResponse:
    if request.method == "GET":
        data = dict(request.query_params)
    else:
        form = await request.form()
        data = dict(form)
    provider = ZPayProvider()
    order_id = str(data.get("out_trade_no", ""))
    if order_id in processed_orders:
        return FastAPIResponse(provider.success_response(), media_type="text/plain")
    if not provider.verify_notify(data):
        raise HTTPException(status_code=400, detail="sign error")
    parsed = provider.parse_notify(data)
    if parsed.get("trade_status") == "success":
        notify_order_id = str(parsed.get("order_id", ""))
        notify_pay_price = str(parsed.get("pay_price", ""))
        notify_order = await find_order(notify_order_id)
        if notify_order and notify_pay_price and not pay_amount_matches(notify_order, notify_pay_price):
            return FastAPIResponse(provider.success_response(), media_type="text/plain")
        await finalize_paid_order(notify_order_id, notify_pay_price)
    return FastAPIResponse(provider.success_response(), media_type="text/plain")


@app.post("/api/pay/xorpay_notify")
@app.post("/cnomadcna/xorpay_notify")
async def xorpay_notify(request: Request) -> FastAPIResponse:
    form = await request.form()
    data = dict(form)
    provider = XorPayProvider()
    order_id = str(data.get("order_id", ""))
    if order_id in processed_orders:
        return FastAPIResponse(provider.success_response(), media_type="text/plain")
    if not provider.verify_notify(data):
        raise HTTPException(status_code=400, detail="sign error")
    parsed = provider.parse_notify(data)
    notify_order_id = str(parsed.get("order_id", ""))
    notify_pay_price = str(parsed.get("pay_price", ""))
    try:
        paid_amount = float(notify_pay_price)
    except (TypeError, ValueError):
        paid_amount = 0.0
    if paid_amount <= 0:
        return FastAPIResponse(provider.success_response(), media_type="text/plain")
    notify_order = await find_order(notify_order_id)
    if notify_order and not pay_amount_matches(notify_order, notify_pay_price):
        return FastAPIResponse(provider.success_response(), media_type="text/plain")
    await finalize_paid_order(notify_order_id, notify_pay_price)
    return FastAPIResponse(provider.success_response(), media_type="text/plain")


@app.get("/api/cities")
async def list_cities() -> dict[str, Any]:
    data = await pb.list_records("cities", per_page=150, sort="idNumber")
    return {"ok": True, "items": [public_record(x) for x in data.get("items", [])]}


@app.get("/api/weather/cities")
async def city_weather_batch() -> dict[str, Any]:
    """首页城市卡片实时温度（Open-Meteo，服务端缓存 30 分钟）。"""
    data = await pb.list_records("cities", per_page=150, sort="idNumber")
    items = await fetch_weather_for_cities(data.get("items", []))
    return {
        "ok": True,
        "items": items,
        "source": "open-meteo",
        "attribution": "Weather data by Open-Meteo.com",
    }


@app.get("/api/weather/cities")
async def city_weather_batch() -> dict[str, Any]:
    """各城市实时温度（Open-Meteo，服务端缓存 30 分钟）。"""
    data = await pb.list_records("cities", per_page=150, sort="idNumber")
    items = await fetch_weather_for_cities(data.get("items", []))
    return {
        "ok": True,
        "source": "open-meteo",
        "items": items,
    }


@app.get("/api/cities/{slug}")
async def city_detail(slug: str) -> dict[str, Any]:
    city = await pb.first_record("cities", filter=f'slug={pb_quote(slug)}')
    if not city:
        raise HTTPException(status_code=404, detail="城市不存在")
    detail = await pb.first_record("city_details", filter=f'citySlug={pb_quote(slug)}')
    content = await pb.list_records(
        "content_items",
        filter=f'citySlugs ?~ {pb_quote(slug)} && status="published"',
        per_page=20,
        sort="sortOrder",
    )
    meetups = await pb.list_records("meetups", filter=f'city={pb_quote(city["name"])}', per_page=20)
    discussions = await pb.list_records("discussions", filter=f'city={pb_quote(city["name"])}', per_page=20)
    return {
        "ok": True,
        "city": public_record(city),
        "detail": public_record(detail) if detail else None,
        "content": [public_record(x) for x in content.get("items", [])],
        "meetups": [public_record(x) for x in meetups.get("items", [])],
        "discussions": [public_record(x) for x in discussions.get("items", [])],
    }


@app.post("/api/cities/{slug}/volunteer-applications")
async def create_city_volunteer_application(
    slug: str,
    payload: CityVolunteerApplicationPayload,
    request: Request,
) -> dict[str, Any]:
    city_slug = clean_submission_text(payload.citySlug or slug, 120)
    city = await pb.first_record("cities", filter=f'slug={pb_quote(city_slug)}')
    city_name = clean_submission_text(payload.cityName or (city or {}).get("name") or city_slug, 120)
    applicant_name = clean_submission_text(payload.applicantName, 140)
    email = clean_submission_text(payload.email, 255)
    wechat = clean_submission_text(payload.wechat, 140)
    phone = clean_submission_text(payload.phone, 40)
    if not applicant_name:
        raise HTTPException(status_code=400, detail="姓名或昵称不能为空")
    if not (email or wechat or phone):
        raise HTTPException(status_code=400, detail="请至少填写一种联系方式")
    roles = [
        clean_submission_text(role, 80)
        for role in payload.roles
        if clean_submission_text(role, 80)
    ][:8]
    if not roles:
        raise HTTPException(status_code=400, detail="请至少选择一个志愿者方向")
    _, user = await current_user(request)
    submitted_at = utc_now()
    record = await pb.create_record(
        "city_volunteer_applications",
        {
            "citySlug": city_slug,
            "cityName": city_name,
            "userId": user["id"] if user else "",
            "applicantName": applicant_name,
            "email": email,
            "wechat": wechat,
            "phone": phone,
            "currentCity": clean_submission_text(payload.currentCity, 120),
            "availability": clean_submission_text(payload.availability, 255),
            "roles": roles,
            "localExperience": clean_submission_text(payload.localExperience, 2000, multiline=True),
            "motivation": clean_submission_text(payload.motivation, 2000, multiline=True),
            "links": clean_submission_text(payload.links, 1000, multiline=True),
            "status": "pending",
            "reviewNote": "",
            "submittedAt": submitted_at,
            "metadata": {
                "source": "city_volunteer_page",
                "ip": client_ip_from_request(request),
                "userAgent": request.headers.get("user-agent", "")[:300],
            },
        },
    )
    await create_notification(
        title=f"{city_name} 志愿者申请待审核",
        body=f"{applicant_name} 申请成为 {city_name} 城市志愿者，方向：{'、'.join(roles)}。",
        category="community",
        priority="high",
        audience="review",
        action_label="查看消息",
        action_url="/messages",
        entity_type="city_volunteer_application",
        entity_id=record.get("id", ""),
        icon="🧭",
        metadata={"citySlug": city_slug, "cityName": city_name, "status": "pending"},
    )
    return {"ok": True, "status": "pending", "record": public_record(record)}


@app.get("/api/content/home")
async def home_content() -> dict[str, Any]:
    data = await pb.list_records("content_items", filter='status="published"', sort="sortOrder", per_page=20)
    return {"ok": True, "items": [public_record(x) for x in data.get("items", [])]}


@app.post("/api/content/submissions")
async def submit_content(payload: ContentSubmissionPayload, request: Request) -> dict[str, Any]:
    content_type = clean_submission_text(payload.type, 20).lower()
    if content_type not in {"ebook", "video"}:
        raise HTTPException(status_code=400, detail="只支持电子书或访谈视频投稿")

    title = clean_submission_text(payload.title, 255)
    subtitle = clean_submission_text(payload.subtitle, 500) or ("社区电子书投稿" if content_type == "ebook" else "社区访谈视频投稿")
    description = clean_submission_text(payload.description, 2000, multiline=True)
    if not title:
        raise HTTPException(status_code=400, detail="缺少标题")
    if not description:
        raise HTTPException(status_code=400, detail="缺少内容简介")

    cover_image = validate_submission_url(payload.coverImage, "封面图")
    media_url = validate_submission_url(payload.mediaUrl, "内容文件")
    _, user = await current_user(request)
    author_name = clean_submission_text(payload.authorName or (user or {}).get("name") or "", 120)
    author_email = clean_submission_text(payload.authorEmail or (user or {}).get("email") or "", 255).lower()
    if "@" not in author_email or "." not in author_email.split("@")[-1]:
        raise HTTPException(status_code=400, detail="请填写有效联系邮箱，便于审核沟通")

    slug = f"ugc-{content_type}-{slugify_content_title(title)}-{uuid.uuid4().hex[:8]}"
    target_url = f"/ebooks/{slug}" if content_type == "ebook" else f"/videos/{slug}"
    cta_label = "查看电子书" if content_type == "ebook" else "观看访谈"
    chapters = normalize_submission_chapters(payload.chapters, content_type)
    city_slugs = [clean_submission_text(x, 80) for x in payload.citySlugs[:12] if clean_submission_text(x, 80)]
    submitted_by = str((user or {}).get("id") or "")
    submitted_at = utc_now()
    base_record: dict[str, Any] = {
        "slug": slug,
        "type": content_type,
        "title": title,
        "titleEn": "",
        "subtitle": subtitle,
        "subtitleEn": "",
        "description": description,
        "descriptionEn": "",
        "coverImage": cover_image,
        "mediaUrl": media_url,
        "targetUrl": target_url,
        "ctaLabel": cta_label,
        "ctaLabelEn": "View ebook" if content_type == "ebook" else "Watch interview",
        "citySlugs": city_slugs,
        "relatedProfiles": [submitted_by] if submitted_by else [],
        "chapters": chapters,
        "durationMinutes": 0,
        "sortOrder": 1000,
        "status": "pending",
    }
    record_payload = {
        **base_record,
        "authorName": author_name,
        "authorEmail": author_email,
        "submittedBy": submitted_by,
        "reviewStatus": "pending",
        "reviewNote": "",
        "submissionMeta": {
            "source": "ugc",
            "submittedAt": submitted_at,
            "ip": client_ip_from_request(request),
            "userAgent": request.headers.get("user-agent", "")[:300],
        },
    }
    try:
        record = await pb.create_record("content_items", record_payload)
    except PocketBaseError:
        record = await pb.create_record("content_items", base_record)

    await create_notification(
        title="新的社区内容待审核",
        body=f"{author_name or author_email} 提交了《{title}》，类型：{'电子书' if content_type == 'ebook' else '访谈视频'}。",
        category="system",
        priority="high",
        audience="review",
        action_label="查看消息",
        action_url="/messages",
        entity_type="content_submission",
        entity_id=record.get("id", ""),
        icon="📝",
        metadata={"slug": slug, "type": content_type, "status": "pending"},
    )
    return {"ok": True, "status": "pending", "record": public_record(record)}


@app.get("/api/videos")
async def list_videos() -> dict[str, Any]:
    data = await pb.list_records(
        "content_items",
        filter='status="published" && type="video"',
        sort="sortOrder",
        per_page=50,
    )
    items = [public_record(x) for x in data.get("items", [])]
    return {"ok": True, "items": items, "total": len(items)}


@app.get("/api/content/{slug}")
async def content_detail(slug: str, request: Request) -> dict[str, Any]:
    item = await pb.first_record("content_items", filter=f'slug={pb_quote(slug)}')
    if not item:
        raise HTTPException(status_code=404, detail="内容不存在")
    status = str(item.get("status") or "published")
    if status != "published":
        _, user = await current_user(request)
        submitted_by = str(item.get("submittedBy") or "")
        if not (user and submitted_by and user.get("id") == submitted_by):
            raise HTTPException(status_code=404, detail="内容不存在")
    cities: list[dict[str, Any]] = []
    for city_slug in item.get("citySlugs") or []:
        city = await pb.first_record("cities", filter=f'slug={pb_quote(city_slug)}')
        if city:
            cities.append(public_record(city))
    return {"ok": True, "item": public_record(item), "cities": cities}


@app.get("/api/notifications")
async def list_notifications(
    request: Request,
    category: str = "all",
    status: str = "active",
    q: str = "",
    muted: str = "include",
) -> dict[str, Any]:
    _, user = await current_user(request)
    user_id = user["id"] if user else "anonymous"
    preferences = await notification_preferences_for_user(user_id, user)
    all_items = await decorated_notifications_for_user(user_id, preferences)
    items = all_items
    if muted == "exclude":
        items = [item for item in items if not item.get("muted")]
    elif muted == "only":
        items = [item for item in items if item.get("muted")]
    if category != "all":
        items = [item for item in items if item.get("category") == category]
    if status == "unread":
        items = [item for item in items if not item.get("read") and not item.get("archived") and not item.get("muted")]
    elif status == "read":
        items = [item for item in items if item.get("read") and not item.get("archived")]
    elif status == "archived":
        items = [item for item in items if item.get("archived")]
    else:
        items = [item for item in items if not item.get("archived")]
    if q.strip():
        needle = q.strip().lower()
        items = [
            item
            for item in items
            if needle in str(item.get("title", "")).lower() or needle in str(item.get("body", "")).lower()
        ]
    unread = sum(1 for item in all_items if not item.get("read") and not item.get("archived") and not item.get("muted"))
    categories: dict[str, int] = {}
    muted_count = 0
    for item in all_items:
        if not item.get("archived"):
            categories[item.get("category", "system")] = categories.get(item.get("category", "system"), 0) + 1
            if item.get("muted"):
                muted_count += 1
    return {
        "ok": True,
        "items": items,
        "unread": unread,
        "categories": categories,
        "muted": muted_count,
        "preferences": preferences,
    }


@app.get("/api/notifications/unread-count")
async def notification_unread_count(request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    user_id = user["id"] if user else "anonymous"
    preferences = await notification_preferences_for_user(user_id, user)
    items = await decorated_notifications_for_user(user_id, preferences)
    return {"ok": True, "count": sum(1 for item in items if not item.get("read") and not item.get("archived") and not item.get("muted"))}


@app.post("/api/notifications")
async def create_notification_endpoint(payload: NotificationCreatePayload, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    target_user_id = payload.targetUserId or (user["id"] if user and payload.audience == "user" else "")
    record = await create_notification(
        title=payload.title,
        body=payload.body,
        category=payload.category,
        priority=payload.priority,
        target_user_id=target_user_id,
        audience=payload.audience,
        action_label=payload.actionLabel,
        action_url=payload.actionUrl,
        entity_type=payload.entityType,
        entity_id=payload.entityId,
        icon=payload.icon,
        channels=payload.channels,
        metadata=payload.metadata,
    )
    if not record:
        raise HTTPException(status_code=500, detail="通知创建失败")
    return {"ok": True, "record": public_record(record)}


@app.post("/api/notifications/{notification_id}/read")
async def mark_notification_read(notification_id: str, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    user_id = user["id"] if user else "anonymous"
    receipt = await upsert_receipt(notification_id, user_id, {"readAt": utc_now()})
    return {"ok": True, "receipt": public_record(receipt)}


@app.post("/api/notifications/{notification_id}/unread")
async def mark_notification_unread(notification_id: str, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    user_id = user["id"] if user else "anonymous"
    receipt = await upsert_receipt(notification_id, user_id, {"readAt": None, "archivedAt": None})
    return {"ok": True, "receipt": public_record(receipt)}


@app.post("/api/notifications/{notification_id}/archive")
async def archive_notification(notification_id: str, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    user_id = user["id"] if user else "anonymous"
    receipt = await upsert_receipt(notification_id, user_id, {"archivedAt": utc_now(), "readAt": utc_now()})
    return {"ok": True, "receipt": public_record(receipt)}


@app.post("/api/notifications/{notification_id}/restore")
async def restore_notification(notification_id: str, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    user_id = user["id"] if user else "anonymous"
    receipt = await upsert_receipt(notification_id, user_id, {"archivedAt": None, "dismissedAt": None})
    return {"ok": True, "receipt": public_record(receipt)}


@app.post("/api/notifications/{notification_id}/pin")
async def pin_notification(notification_id: str, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    user_id = user["id"] if user else "anonymous"
    existing = await receipt_for(notification_id, user_id)
    pinned = not bool(existing and existing.get("pinned"))
    receipt = await upsert_receipt(notification_id, user_id, {"pinned": pinned})
    return {"ok": True, "receipt": public_record(receipt), "pinned": pinned}


@app.post("/api/notifications/mark-all-read")
async def mark_all_notifications_read(request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    user_id = user["id"] if user else "anonymous"
    preferences = await notification_preferences_for_user(user_id, user)
    items = await decorated_notifications_for_user(user_id, preferences)
    count = 0
    for item in items:
        if not item.get("read") and not item.get("archived") and not item.get("muted"):
            await upsert_receipt(item["id"], user_id, {"readAt": utc_now()})
            count += 1
    return {"ok": True, "count": count}


@app.get("/api/notifications/push-config")
async def notification_push_config(request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        raise HTTPException(status_code=401, detail="请先登录")
    user_id = user["id"]
    return {
        "ok": True,
        "topic": ntfy_topic_for_user(user_id),
        "subscribeUrl": ntfy_subscribe_url(user_id),
        "serverUrl": settings.ntfy_url,
    }


@app.get("/api/notifications/preferences")
async def notification_preferences(request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    user_id = user["id"] if user else "anonymous"
    return {"ok": True, "record": await notification_preferences_for_user(user_id, user)}


@app.put("/api/notifications/preferences")
async def update_notification_preferences(payload: NotificationPreferencePayload, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    user_id = user["id"] if user else "anonymous"
    record_payload = {
        "userId": user_id,
        "email": user.get("email", "") if user else "",
        **payload.model_dump(),
    }
    existing = await pb.first_record("notification_preferences", filter=f'userId={pb_quote(user_id)}')
    if existing:
        record = await pb.update_record("notification_preferences", existing["id"], record_payload)
    else:
        record = await pb.create_record("notification_preferences", record_payload)
    return {"ok": True, "record": public_record(record)}


def city_recommendation_score(city: dict[str, Any], payload: RecommendationPayload) -> tuple[float, list[str]]:
    score = float(city.get("overallScore", 0)) * 20
    reasons: list[str] = []
    cost = int(city.get("costPerMonth") or 999999)
    speed = int(city.get("internetSpeed") or 0)
    temp = int(city.get("temperature") or 0)
    tags = city.get("tags") or []

    if cost <= payload.budget:
        score += 22
        reasons.append("预算内")
    else:
        over_ratio = min(1.0, (cost - payload.budget) / max(payload.budget, 1))
        score -= 18 * over_ratio

    if speed >= payload.internet:
        score += 18
        reasons.append("网络达标")
    else:
        score -= min(18, payload.internet - speed)

    if payload.climate == "warm" and temp >= 20:
        score += 8
        reasons.append("偏温暖")
    elif payload.climate == "cool" and temp <= 20:
        score += 8
        reasons.append("偏凉爽")
    elif payload.climate == "mild" and 14 <= temp <= 24:
        score += 10
        reasons.append("气候温和")

    matched_tags = [tag for tag in payload.tags if tag in tags]
    if matched_tags:
        score += len(matched_tags) * 5
        reasons.append("标签匹配：" + "、".join(matched_tags[:3]))

    if city.get("nomadsNow", 0) >= 300:
        score += 6
        reasons.append("社区活跃")

    return round(score, 1), reasons[:4]


@app.post("/api/recommendations/next-stop")
async def next_stop_recommendation(payload: RecommendationPayload, request: Request) -> dict[str, Any]:
    cities = (await pb.list_records("cities", per_page=150, sort="idNumber")).get("items", [])
    scored = []
    for city in cities:
        score, reasons = city_recommendation_score(city, payload)
        scored.append({**public_record(city), "matchScore": score, "matchReasons": reasons})
    scored.sort(key=lambda item: item["matchScore"], reverse=True)
    picks = scored[: max(1, min(payload.limit, 30))]

    _, user = await current_user(request)
    try:
        await pb.create_record(
            "recommendation_logs",
            {
                "userId": user["id"] if user else "anonymous",
                "budget": payload.budget,
                "internet": payload.internet,
                "climate": payload.climate,
                "tags": payload.tags,
                "resultSlugs": [city.get("slug") for city in picks],
            },
        )
    except PocketBaseError:
        pass
    return {"ok": True, "items": picks}


@app.get("/api/routes")
async def list_routes() -> dict[str, Any]:
    routes = (await pb.list_records("routes", per_page=100)).get("items", [])
    cities = (await pb.list_records("cities", per_page=150, sort="idNumber")).get("items", [])
    city_by_slug = {city.get("slug"): public_record(city) for city in cities}
    items = []
    for route in routes:
        record = public_record(route)
        record["cities"] = [city_by_slug[slug] for slug in route.get("citySlugs", []) if slug in city_by_slug]
        items.append(record)
    return {"ok": True, "items": items}


@app.get("/api/services")
async def list_services() -> dict[str, Any]:
    data = await pb.list_records("services", filter='status="published"', per_page=100)
    return {"ok": True, "items": [public_record(x) for x in data.get("items", [])]}


@app.post("/api/services/leads")
async def create_service_lead(payload: ServiceLeadPayload, request: Request) -> dict[str, Any]:
    if not payload.serviceSlug.strip():
        raise HTTPException(status_code=400, detail="服务不能为空")
    service = await pb.first_record("services", filter=f'slug={pb_quote(payload.serviceSlug)}')
    if not service:
        raise HTTPException(status_code=404, detail="服务不存在")
    _, user = await current_user(request)
    record = await pb.create_record(
        "service_leads",
        {
            "serviceSlug": payload.serviceSlug,
            "userId": user["id"] if user else "",
            "email": payload.email or (user.get("email", "") if user else ""),
            "note": payload.note,
            "status": "new",
        },
    )
    await create_notification(
        title=f"服务咨询已提交：{service.get('title', payload.serviceSlug)}",
        body="我们已经记录你的服务咨询，后续可以按城市、预算、路线和会员状态继续跟进。",
        category="service",
        priority="normal",
        target_user_id=user["id"] if user else "",
        audience="user" if user else "all",
        action_label="查看服务",
        action_url="/services",
        entity_type="service_lead",
        entity_id=record["id"],
        icon=service.get("icon") or "🧭",
    )
    return {"ok": True, "record": public_record(record), "service": public_record(service)}


@app.get("/api/downloads/{filename}")
async def download_asset(filename: str) -> FastAPIResponse:
    async def redirect_to_minio(body: bytes, content_type: str) -> RedirectResponse:
        object_key = f"{settings.minio_upload_prefix}/downloads/{filename}"
        asset = await upload_bytes_to_minio(body, object_key=object_key, content_type=content_type)
        return RedirectResponse(asset.url, status_code=307)

    if filename == "nomad-china-playbook.pdf":
        body = (
            "%PDF-1.4\n"
            "1 0 obj << /Type /Catalog /Pages 2 0 R >> endobj\n"
            "2 0 obj << /Type /Pages /Kids [3 0 R] /Count 1 >> endobj\n"
            "3 0 obj << /Type /Page /Parent 2 0 R /MediaBox [0 0 300 144] /Contents 4 0 R >> endobj\n"
            "4 0 obj << /Length 68 >> stream\n"
            "BT /F1 18 Tf 24 88 Td (NomadCNA China Digital Nomad Playbook) Tj ET\n"
            "endstream endobj\n"
            "xref\n0 5\n0000000000 65535 f \n0000000009 00000 n \n0000000058 00000 n \n0000000115 00000 n \n0000000204 00000 n \n"
            "trailer << /Root 1 0 R /Size 5 >>\nstartxref\n322\n%%EOF\n"
        ).encode("latin-1")
        return await redirect_to_minio(body, "application/pdf")
    if filename == "nomadcna-app-demo.zip":
        buffer = io.BytesIO()
        with zipfile.ZipFile(buffer, "w", compression=zipfile.ZIP_DEFLATED) as archive:
            archive.writestr(
                "README.txt",
                "\n".join(
                    [
                        "Nomadro Web App Package",
                        "",
                        "Open Nomadro-Web-App.html in a browser or visit https://nomadro.cn/zh/digital.",
                        "The web app can be installed from Chrome, Edge, Safari, and other modern browsers.",
                        "",
                    ]
                ),
            )
            archive.writestr(
                "Nomadro-Web-App.html",
                """<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8" />
  <meta name="viewport" content="width=device-width, initial-scale=1" />
  <meta http-equiv="refresh" content="0; url=https://nomadro.cn/zh/digital" />
  <title>Nomadro Web App</title>
  <style>
    body{font-family:system-ui,-apple-system,BlinkMacSystemFont,"Segoe UI",sans-serif;margin:0;min-height:100vh;display:grid;place-items:center;background:#fafafa;color:#111827}
    main{max-width:560px;padding:32px;text-align:center}
    a{display:inline-flex;margin-top:16px;border-radius:999px;background:#ff4d4f;color:#fff;padding:12px 20px;text-decoration:none;font-weight:700}
  </style>
</head>
<body>
  <main>
    <h1>Nomadro Web App</h1>
    <p>正在打开数字游民指南。如果没有自动跳转，请点击下方按钮。</p>
    <a href="https://nomadro.cn/zh/digital">打开 Nomadro</a>
  </main>
</body>
</html>
""",
            )
        body = buffer.getvalue()
        return await redirect_to_minio(body, "application/zip")
    raise HTTPException(status_code=404, detail="下载文件不存在")


@app.get("/api/meetups")
async def list_meetups() -> dict[str, Any]:
    data = await pb.list_records("meetups", sort="date", per_page=100)
    items: list[dict[str, Any]] = []
    for record in data.get("items", []):
        status = str(record.get("status") or "published").lower()
        if status == "pending":
            continue
        items.append(public_meetup_record(record))
    return {"ok": True, "items": items}


async def fetch_meetup_record(meetup_id: str) -> dict[str, Any]:
    record = await pb.first_record("meetups", filter=f'id={pb_quote(meetup_id)}')
    if record:
        return record
    raise HTTPException(status_code=404, detail="活动不存在")


async def sync_meetup_attendees(meetup_id: str) -> dict[str, Any] | None:
    meetup = await pb.first_record("meetups", filter=f'id={pb_quote(meetup_id)}')
    if not meetup:
        return None
    data = await pb.list_records(
        "meetup_rsvps",
        filter=f'meetupId={pb_quote(meetup_id)} && status="going"',
        per_page=100,
    )
    attendees: list[dict[str, Any]] = []
    for rsvp in data.get("items", []):
        user_id = str(rsvp.get("userId") or "")
        if not user_id:
            continue
        profile = await profile_for_user(user_id)
        attendees.append(
            {
                "userId": user_id,
                "profileId": str((profile or {}).get("id") or ""),
                "name": str((profile or {}).get("name") or "成员"),
                "photo": str((profile or {}).get("photo") or avatar_for_name(user_id)),
            }
        )
    return await pb.update_record(
        "meetups",
        meetup["id"],
        {"rsvpCount": len(attendees), "attendees": attendees},
    )


async def user_rsvp_for_meetup(meetup_id: str, user_id: str) -> dict[str, Any] | None:
    if not user_id:
        return None
    return await pb.first_record(
        "meetup_rsvps",
        filter=f'meetupId={pb_quote(meetup_id)} && userId={pb_quote(user_id)}',
    )


@app.get("/api/meetups/{meetup_id}")
async def get_meetup(meetup_id: str) -> dict[str, Any]:
    record = await fetch_meetup_record(meetup_id)
    status = str(record.get("status") or "published").lower()
    if status == "pending":
        raise HTTPException(status_code=404, detail="活动不存在")
    return {"ok": True, "item": public_meetup_record(record)}


class NewsletterSubscribePayload(BaseModel):
    email: str
    name: str = ""
    citySlug: str = ""
    list: str = "general"


@app.post("/api/newsletter/subscribe")
async def newsletter_subscribe(payload: NewsletterSubscribePayload) -> dict[str, Any]:
    email = payload.email.strip()
    if not email or "@" not in email:
        raise HTTPException(status_code=400, detail="邮箱格式不正确")
    result = await subscribe_newsletter(
        email=email,
        name=payload.name,
        city_slug=payload.citySlug.strip(),
    )
    if not result.get("ok"):
        raise HTTPException(status_code=502, detail=result.get("error") or "订阅失败")
    return {"ok": True, "subscriber": result.get("subscriber")}


@app.get("/api/meetups/{meetup_id}/session")
async def meetup_session(meetup_id: str, request: Request) -> dict[str, Any]:
    record = await fetch_meetup_record(meetup_id)
    _, user = await current_user(request)
    vip = bool(await find_membership(user["id"])) if user else False
    mode = str(record.get("mode") or "offline").lower()
    is_live_event = mode in {"online", "hybrid"}
    status = str(record.get("status") or "published").lower()
    if status == "pending":
        raise HTTPException(status_code=404, detail="活动不存在")
    can_join = can_access_meetup(record, user, vip=vip) if is_live_event else False
    display_name = viewer_display_name(user)
    lounge_channel = create_lounge_channel(record)
    session = {
        "canJoin": can_join,
        "lockReason": "" if can_join else access_lock_reason(record),
        "displayName": display_name,
        "videoUrl": build_mirotalk_join_url(record, display_name) if is_live_event and can_join else "",
        "chatUrl": build_lounge_chat_url(record) if is_live_event and can_join else "",
        "replayUrl": str(record.get("replayUrl") or "").strip() if is_live_event and can_join else "",
        "loungeChannel": lounge_channel if can_join else "",
        "mirotalkRoom": str(record.get("mirotalkRoom") or "") if can_join else "",
        "meetingProvider": str(record.get("meetingProvider") or ("mirotalk" if is_live_event else "")),
        "user": None,
    }
    if user:
        session["user"] = {
            "id": user.get("id"),
            "email": user.get("email"),
            "name": display_name,
            "vip": vip,
            "role": user.get("role"),
        }
    return {"ok": True, "meetup": public_meetup_record(record), "session": session}


@app.post("/api/meetups")
async def create_meetup(payload: dict[str, Any], request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        raise HTTPException(status_code=401, detail="请先登录后再发起活动")
    if payload.get("date") and "T" not in str(payload["date"]):
        payload["date"] = f"{str(payload['date'])[:10]} 00:00:00.000Z"
    mode = str(payload.get("mode") or "offline").lower()
    if mode not in {"offline", "online", "hybrid"}:
        mode = "offline"
    access_level = str(payload.get("accessLevel") or "public").lower()
    if access_level not in {"public", "members", "vip", "hosts"}:
        access_level = "public"
    payload["mode"] = mode
    payload["accessLevel"] = access_level
    payload.setdefault("status", "pending")
    payload.setdefault("rsvpCount", 0)
    payload.setdefault("attendees", [])
    payload.setdefault("maxAttendees", 0)
    payload.setdefault("organizer", "NomadCNA")
    payload.setdefault("meetingProvider", "mirotalk" if mode != "offline" else "")
    payload.setdefault("mirotalkRoom", "")
    payload.setdefault("loungeChannel", "")
    payload.setdefault("meetingUrl", "")
    payload.setdefault("replayUrl", "")
    if not isinstance(payload.get("features"), list):
        payload["features"] = []
    record = await pb.create_record("meetups", payload)
    if mode != "offline":
        room_updates: dict[str, Any] = {}
        if not str(record.get("mirotalkRoom") or "").strip():
            room_updates["mirotalkRoom"] = create_mirotalk_room_slug(record)
        if not str(record.get("loungeChannel") or "").strip():
            room_slug = room_updates.get("mirotalkRoom") or create_mirotalk_room_slug(record)
            room_updates["loungeChannel"] = f"#{room_slug}"
        if room_updates:
            record = await pb.update_record("meetups", record["id"], room_updates)
    await create_notification(
        title=f"新活动已提交：{payload.get('city', '同城')} · {payload.get('venue', '活动地点')}",
        body=str(payload.get("description") or "新的同城活动已提交审核，通过后将展示在聚会列表。"),
        category="meetup",
        priority="normal",
        audience="all",
        action_label="查看活动",
        action_url="/meetups",
        entity_type="meetup",
        entity_id=record["id"],
        icon=payload.get("emoji") or "🍹",
    )
    return {"ok": True, "record": record}


async def discussion_liked_by_user(discussion_id: str, user_id: str) -> bool:
    if not user_id:
        return False
    existing = await pb.first_record(
        "discussion_likes",
        filter=f'discussionId={pb_quote(discussion_id)} && userId={pb_quote(user_id)}',
    )
    return bool(existing)


def decorate_discussion(record: dict[str, Any], *, liked: bool = False) -> dict[str, Any]:
    payload = public_record(record)
    payload["likes"] = int(record.get("likes") or 0)
    payload["pinned"] = bool(record.get("pinned"))
    payload["likedByMe"] = liked
    return payload


@app.get("/api/discussions")
async def list_discussions(request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    user_id = user["id"] if user else ""
    data = await pb.list_records("discussions", per_page=100)
    items: list[dict[str, Any]] = []
    for record in data.get("items", []):
        liked = await discussion_liked_by_user(str(record.get("id") or ""), user_id)
        items.append(decorate_discussion(record, liked=liked))
    items.sort(key=lambda item: str(item.get("created") or ""), reverse=True)
    items.sort(key=lambda item: bool(item.get("pinned")), reverse=True)
    return {"ok": True, "items": items}


@app.get("/api/discussions/{discussion_id}")
async def get_discussion(discussion_id: str, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    user_id = user["id"] if user else ""
    record = await pb.first_record("discussions", filter=f'id={pb_quote(discussion_id)}')
    if not record:
        raise HTTPException(status_code=404, detail="讨论不存在")
    views = int(record.get("views") or 0) + 1
    record = await pb.update_record("discussions", record["id"], {"views": views})
    liked = await discussion_liked_by_user(discussion_id, user_id)
    return {"ok": True, "item": decorate_discussion(record, liked=liked)}


@app.post("/api/discussions/{discussion_id}/like")
async def toggle_discussion_like(discussion_id: str, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        raise HTTPException(status_code=401, detail="请先登录")
    record = await pb.first_record("discussions", filter=f'id={pb_quote(discussion_id)}')
    if not record:
        raise HTTPException(status_code=404, detail="讨论不存在")
    existing = await pb.first_record(
        "discussion_likes",
        filter=f'discussionId={pb_quote(discussion_id)} && userId={pb_quote(user["id"])}',
    )
    likes = int(record.get("likes") or 0)
    if existing:
        await pb.delete_record("discussion_likes", existing["id"])
        likes = max(0, likes - 1)
        liked = False
    else:
        await pb.create_record(
            "discussion_likes",
            {"discussionId": discussion_id, "userId": user["id"]},
        )
        likes += 1
        liked = True
    updated = await pb.update_record("discussions", record["id"], {"likes": likes})
    return {"ok": True, "liked": liked, "likes": likes, "item": decorate_discussion(updated, liked=liked)}


@app.post("/api/discussions/{discussion_id}/pin")
async def toggle_discussion_pin(discussion_id: str, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        raise HTTPException(status_code=401, detail="请先登录")
    record = await pb.first_record("discussions", filter=f'id={pb_quote(discussion_id)}')
    if not record:
        raise HTTPException(status_code=404, detail="讨论不存在")
    author_user_id = str(record.get("authorUserId") or "")
    if author_user_id and author_user_id != user["id"]:
        raise HTTPException(status_code=403, detail="只有作者可以置顶该话题")
    pinned = not bool(record.get("pinned"))
    updated = await pb.update_record("discussions", record["id"], {"pinned": pinned})
    liked = await discussion_liked_by_user(discussion_id, user["id"])
    return {"ok": True, "pinned": pinned, "item": decorate_discussion(updated, liked=liked)}


@app.get("/api/discussions/{discussion_id}/replies")
async def list_discussion_replies(discussion_id: str) -> dict[str, Any]:
    record = await pb.first_record("discussions", filter=f'id={pb_quote(discussion_id)}')
    if not record:
        raise HTTPException(status_code=404, detail="讨论不存在")
    data = await pb.list_records(
        "discussion_replies",
        filter=f'discussionId={pb_quote(discussion_id)}',
        per_page=100,
        sort="created",
    )
    return {"ok": True, "items": [public_record(x) for x in data.get("items", [])]}


@app.post("/api/discussions/{discussion_id}/replies")
async def create_discussion_reply(
    discussion_id: str,
    payload: DiscussionReplyPayload,
    request: Request,
) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        raise HTTPException(status_code=401, detail="请先登录")
    record = await pb.first_record("discussions", filter=f'id={pb_quote(discussion_id)}')
    if not record:
        raise HTTPException(status_code=404, detail="讨论不存在")
    profile = await profile_for_user(user["id"])
    author = str((profile or {}).get("name") or user.get("name") or user.get("email", "").split("@")[0] or "成员")
    reply = await pb.create_record(
        "discussion_replies",
        {
            "discussionId": discussion_id,
            "userId": user["id"],
            "author": author,
            "body": payload.body.strip(),
        },
    )
    replies = int(record.get("replies") or 0) + 1
    await pb.update_record("discussions", record["id"], {"replies": replies})
    await create_notification(
        title=f"讨论有新回复：{record.get('title', '社区话题')}",
        body=payload.body.strip()[:180],
        category="community",
        priority="normal",
        audience="all",
        action_label="查看回复",
        action_url=f"/discuss/{discussion_id}",
        entity_type="discussion",
        entity_id=discussion_id,
        icon="💬",
    )
    return {"ok": True, "reply": public_record(reply), "replies": replies}


@app.post("/api/discussions")
async def create_discussion(payload: DiscussionPayload, request: Request) -> dict[str, Any]:
    if not payload.title.strip():
        raise HTTPException(status_code=400, detail="标题不能为空")
    _, user = await current_user(request)
    author = payload.author.strip()
    if user:
        profile = await profile_for_user(user["id"])
        author = str((profile or {}).get("name") or user.get("name") or author or "社区成员")
    record = await pb.create_record(
        "discussions",
        {
            **payload.model_dump(),
            "author": author or "社区成员",
            "authorUserId": user["id"] if user else "",
            "replies": 0,
            "views": 0,
            "likes": 0,
            "pinned": False,
        },
    )
    await create_notification(
        title=f"新讨论：{payload.title}",
        body=payload.excerpt or f"{author} 在 {payload.city or '社区'} 发起了一个新话题。",
        category="community",
        priority="normal",
        audience="all",
        action_label="参与讨论",
        action_url=f"/discuss/{record['id']}",
        entity_type="discussion",
        entity_id=record["id"],
        icon="🔥",
        metadata={"city": payload.city, "tags": payload.tags},
    )
    return {"ok": True, "record": public_record(record)}


@app.post("/api/meetups/{meetup_id}/rsvp")
async def rsvp_meetup(meetup_id: str, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        raise HTTPException(status_code=401, detail="请先登录后再报名活动")
    meetup = await fetch_meetup_record(meetup_id)
    status = str(meetup.get("status") or "published").lower()
    if status == "pending":
        raise HTTPException(status_code=400, detail="活动尚未发布")
    max_attendees = int(meetup.get("maxAttendees") or 0)
    current_count = int(meetup.get("rsvpCount") or 0)
    if max_attendees > 0 and current_count >= max_attendees:
        existing = await user_rsvp_for_meetup(meetup_id, user["id"])
        if not existing or str(existing.get("status")) != "going":
            raise HTTPException(status_code=400, detail="活动名额已满")
    existing = await user_rsvp_for_meetup(meetup_id, user["id"])
    if existing:
        await pb.update_record("meetup_rsvps", existing["id"], {"status": "going"})
    else:
        await pb.create_record(
            "meetup_rsvps",
            {"meetupId": meetup_id, "userId": user["id"], "status": "going"},
        )
    updated = await sync_meetup_attendees(meetup_id)
    profile = await profile_for_user(user["id"])
    display_name = str((profile or {}).get("name") or user.get("name") or "一位成员")
    await create_notification(
        title=f"活动报名成功 · {meetup.get('city', '同城')}",
        body=f"{display_name} 已报名 {meetup.get('venue', '活动')}，记得准时参加。",
        category="meetup",
        priority="normal",
        target_user_id=user["id"],
        audience="user",
        action_label="查看活动",
        action_url=f"/meetups",
        entity_type="meetup",
        entity_id=meetup_id,
        icon=str(meetup.get("emoji") or "📍"),
    )
    return {"ok": True, "rsvp": True, "meetup": public_meetup_record(updated or meetup)}


@app.delete("/api/meetups/{meetup_id}/rsvp")
async def cancel_meetup_rsvp(meetup_id: str, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        raise HTTPException(status_code=401, detail="请先登录")
    existing = await user_rsvp_for_meetup(meetup_id, user["id"])
    if existing:
        await pb.delete_record("meetup_rsvps", existing["id"])
    updated = await sync_meetup_attendees(meetup_id)
    meetup = updated or await fetch_meetup_record(meetup_id)
    return {"ok": True, "rsvp": False, "meetup": public_meetup_record(meetup)}


@app.get("/api/meetups/{meetup_id}/rsvp")
async def meetup_rsvp_status(meetup_id: str, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        return {"ok": True, "rsvp": False}
    existing = await user_rsvp_for_meetup(meetup_id, user["id"])
    return {"ok": True, "rsvp": bool(existing and str(existing.get("status")) == "going")}


@app.get("/api/meetups/{meetup_id}/social-suggestions")
async def meetup_social_suggestions(meetup_id: str, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    meetup = await fetch_meetup_record(meetup_id)
    status = str(meetup.get("status") or "published").lower()
    if status == "pending":
        raise HTTPException(status_code=404, detail="活动不存在")
    city_name = str(meetup.get("city") or "").strip()
    my_profile = await profile_for_user(user["id"]) if user else None
    my_profile_id = str(my_profile.get("id") or "") if my_profile else ""
    swiped = await swiped_profile_ids(user["id"]) if user else set()
    data = await pb.list_records("profiles", per_page=100)
    items: list[dict[str, Any]] = []
    for record in data.get("items", []):
        record_id = str(record.get("id") or "")
        if not record_id or record_id == my_profile_id:
            continue
        if user and record_id in swiped:
            continue
        location = str(record.get("location") or "")
        city_slug = str(record.get("citySlug") or "")
        if city_name and city_name not in location and city_name not in city_slug:
            if city_name != "线上" and city_name not in location:
                continue
        payload = enrich_profile(record)
        payload["rsvpAttendee"] = any(
            str(attendee.get("profileId") or "") == record_id
            for attendee in (meetup.get("attendees") or [])
            if isinstance(attendee, dict)
        )
        items.append(payload)
    items.sort(key=lambda item: (not bool(item.get("rsvpAttendee")), str(item.get("name") or "")))
    return {
        "ok": True,
        "items": items[:12],
        "total": len(items),
        "city": city_name,
        "meetupId": meetup_id,
    }


@app.get("/api/gigs")
async def list_gigs() -> dict[str, Any]:
    data = await pb.list_records("gigs", per_page=100)
    return {"ok": True, "items": [public_record(x) for x in data.get("items", [])]}


@app.post("/api/gigs")
async def create_gig(payload: GigPayload, request: Request) -> dict[str, Any]:
    if not payload.title.strip():
        raise HTTPException(status_code=400, detail="任务标题不能为空")
    _, user = await current_user(request)
    record = await pb.create_record(
        "gigs",
        {
            **payload.model_dump(),
            "status": "open",
        },
    )
    await create_notification(
        title=f"赏金墙新任务：{payload.title}",
        body=f"{payload.location} · {payload.category} · ¥{payload.budget}。{payload.description}",
        category="gig",
        priority="high" if payload.budget >= 1000 else "normal",
        audience="all",
        action_label="查看任务",
        action_url="/gigs",
        entity_type="gig",
        entity_id=record["id"],
        icon="💰",
    )
    return {"ok": True, "record": public_record(record), "user": public_record(user) if user else None}


@app.post("/api/gigs/{gig_id}/apply")
async def apply_gig(gig_id: str, request: Request) -> dict[str, Any]:
    gig = await pb.first_record("gigs", filter=f'id={pb_quote(gig_id)}')
    if not gig:
        raise HTTPException(status_code=404, detail="任务不存在")
    _, user = await current_user(request)
    applicant = user.get("email") if user else "anonymous"
    application = await pb.create_record(
        "gig_applications",
        {
            "gigId": gig_id,
            "userId": user["id"] if user else "",
            "applicant": applicant,
            "message": "用户从赏金墙接单",
            "status": "submitted",
        },
    )
    updated = await pb.update_record("gigs", gig_id, {"status": "in_progress"})
    await create_notification(
        title=f"任务已被接单：{gig.get('title', '赏金任务')}",
        body=f"{applicant} 已接下这个任务，任务状态已进入进行中。",
        category="gig",
        priority="high",
        target_user_id=user["id"] if user else "",
        audience="user" if user else "all",
        action_label="查看赏金墙",
        action_url="/gigs",
        entity_type="gig_application",
        entity_id=application["id"],
        icon="✅",
    )
    return {
        "ok": True,
        "message": "已接单",
        "applicant": applicant,
        "record": public_record(updated),
        "application": public_record(application),
    }


@app.get("/api/profiles")
async def list_profiles() -> dict[str, Any]:
    data = await pb.list_records("profiles", per_page=100)
    return {"ok": True, "items": [enrich_profile(x) for x in data.get("items", [])]}


@app.get("/api/profiles/{profile_id}")
async def get_profile(profile_id: str) -> dict[str, Any]:
    record = await pb.first_record("profiles", filter=f'id={pb_quote(profile_id)}')
    if not record:
        raise HTTPException(status_code=404, detail="资料不存在")
    return {"ok": True, "profile": enrich_profile(record)}


@app.get("/api/matches/candidates")
async def match_candidates(
    request: Request,
    intent: str = "friends",
    city: str = "",
    exclude_swiped: bool = True,
    gender: str = "",
    single: str = "",
) -> dict[str, Any]:
    _, user = await current_user(request)
    user_id = user["id"] if user else ""
    membership = await find_membership(user_id) if user_id else None
    if not membership:
        gender = ""
        single = ""
    selected_intent = intent if intent in MATCH_INTENTS else "friends"
    data = await pb.list_records("profiles", per_page=100)
    my_profile = await profile_for_user(user_id) if user_id else None
    swiped = await swiped_profile_ids(user_id) if user_id and exclude_swiped else set()
    items: list[dict[str, Any]] = []
    for record in data.get("items", []):
        record_id = str(record.get("id") or "")
        if my_profile and record_id == str(my_profile.get("id")):
            continue
        if record_id in swiped:
            continue
        if selected_intent not in profile_looking_for(record):
            continue
        if not profile_matches_city(record, city):
            continue
        if not profile_matches_advanced(record, gender, single):
            continue
        items.append(enrich_profile(record))
    return {
        "ok": True,
        "items": items,
        "total": len(items),
        "intent": selected_intent,
        "city": city.strip(),
        "gender": gender,
        "single": single,
    }


@app.get("/api/matches/profile/me")
async def match_profile_me(request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        return {"ok": True, "profile": None}
    profile = await profile_for_user(user["id"])
    if not profile:
        return {"ok": True, "profile": None}
    return {"ok": True, "profile": enrich_profile(profile)}


@app.get("/api/matches/quota")
async def match_quota(request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        return {
            "ok": True,
            "vip": False,
            "limit": FREE_SWIPE_DAILY_LIMIT,
            "used": 0,
            "remaining": FREE_SWIPE_DAILY_LIMIT,
        }
    membership = await find_membership(user["id"])
    vip = bool(membership)
    used = await swipe_count_today(user["id"])
    limit = 9999 if vip else FREE_SWIPE_DAILY_LIMIT
    return {
        "ok": True,
        "vip": vip,
        "limit": limit,
        "used": used,
        "remaining": max(0, limit - used),
    }


@app.get("/api/matches/likes")
async def match_likes(request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        return {"ok": True, "items": [], "total": 0}
    my_profile = await profile_for_user(user["id"])
    if not my_profile:
        return {"ok": True, "items": [], "total": 0}
    swipe_data = await pb.list_records(
        "swipes",
        filter=f'profileId={pb_quote(my_profile["id"])}',
        per_page=80,
        sort="-created",
    )
    profiles_data = await pb.list_records("profiles", per_page=100)
    profile_by_user = {
        str(item.get("userId")): item for item in profiles_data.get("items", []) if item.get("userId")
    }
    items: list[dict[str, Any]] = []
    seen_users: set[str] = set()
    for swipe in swipe_data.get("items", []):
        if str(swipe.get("action")) not in LIKE_ACTIONS:
            continue
        swiper_id = str(swipe.get("userId") or "")
        if not swiper_id or swiper_id == user["id"] or swiper_id in seen_users:
            continue
        swiper_profile = profile_by_user.get(swiper_id)
        if not swiper_profile:
            continue
        seen_users.add(swiper_id)
        payload = enrich_profile(swiper_profile)
        payload["likedAt"] = swipe.get("created") or ""
        payload["swipeId"] = swipe.get("id") or ""
        items.append(payload)
    return {"ok": True, "items": items, "total": len(items)}


@app.get("/api/matches/mutual")
async def match_mutual(request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        return {"ok": True, "items": [], "total": 0}
    my_profile = await profile_for_user(user["id"])
    if not my_profile:
        return {"ok": True, "items": [], "total": 0}
    data = await pb.list_records(
        "match_connections",
        filter=f'userAId={pb_quote(user["id"])} || userBId={pb_quote(user["id"])}',
        per_page=50,
        sort="-created",
    )
    profiles_data = await pb.list_records("profiles", per_page=100)
    profile_by_id = {str(item.get("id")): item for item in profiles_data.get("items", []) if item.get("id")}
    items: list[dict[str, Any]] = []
    for connection in data.get("items", []):
        other_profile_id = (
            connection.get("profileBId")
            if str(connection.get("userAId")) == user["id"]
            else connection.get("profileAId")
        )
        other_profile = profile_by_id.get(str(other_profile_id))
        if not other_profile:
            continue
        payload = enrich_profile(other_profile)
        payload["matchedAt"] = connection.get("matchedAt") or connection.get("created") or ""
        payload["intent"] = connection.get("intent") or "friends"
        payload["connectionId"] = connection.get("id") or ""
        conversation = await sync_connection_conversation(connection)
        payload["conversationId"] = conversation.get("id") or connection.get("conversationId") or ""
        items.append(payload)
    return {"ok": True, "items": items, "total": len(items)}


@app.post("/api/matches/swipes/undo")
async def undo_last_swipe(request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        raise HTTPException(status_code=401, detail="请先登录")
    data = await pb.list_records("swipes", filter=f'userId={pb_quote(user["id"])}', per_page=1, sort="-created")
    items = data.get("items", [])
    if not items:
        raise HTTPException(status_code=404, detail="没有可撤销的滑动记录")
    last = items[0]
    await pb.delete_record("swipes", last["id"])
    return {"ok": True, "profileId": last.get("profileId"), "action": last.get("action")}


@app.post("/api/matches/swipes")
async def save_swipe(payload: SwipePayload, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        raise HTTPException(status_code=401, detail="请先登录后再进行匹配操作")
    user_id = user["id"]
    my_profile = await profile_for_user(user_id)
    if not my_profile:
        raise HTTPException(status_code=400, detail="请先完善加入资料后再进行匹配")
    membership = await find_membership(user_id)
    vip = bool(membership)
    if payload.action == "superlike" and not vip:
        raise HTTPException(status_code=403, detail="超级赞为会员专属功能")
    if not vip:
        used = await swipe_count_today(user_id)
        if used >= FREE_SWIPE_DAILY_LIMIT:
            raise HTTPException(status_code=429, detail="今日免费滑动次数已用完，升级会员可无限滑动")
    intent = payload.intent if payload.intent in MATCH_INTENTS else "friends"
    target_profile = await pb.first_record("profiles", filter=f'id={pb_quote(payload.profileId)}')
    if not target_profile:
        raise HTTPException(status_code=404, detail="资料不存在")
    if str(target_profile.get("id")) == str(my_profile.get("id")):
        raise HTTPException(status_code=400, detail="不能对自己进行操作")
    existing_swipe = await pb.first_record(
        "swipes",
        filter=f'userId={pb_quote(user_id)} && profileId={pb_quote(payload.profileId)}',
    )
    if existing_swipe:
        raise HTTPException(status_code=409, detail="已经对该用户进行过滑动操作")
    record = await pb.create_record(
        "swipes",
        {
            "userId": user_id,
            "profileId": payload.profileId,
            "action": payload.action,
            "intent": intent,
        },
    )
    matched = False
    match_record: dict[str, Any] | None = None
    if payload.action in LIKE_ACTIONS:
        target_user_id = str(target_profile.get("userId") or "")
        target_name = str(target_profile.get("name") or "一位成员")
        if my_profile and target_user_id and await has_reciprocal_like(target_user_id, str(my_profile["id"])):
            matched = True
            match_record = await ensure_match_connection(
                user_a_id=user_id,
                user_b_id=target_user_id,
                profile_a_id=str(my_profile["id"]),
                profile_b_id=str(target_profile["id"]),
                intent=intent,
            )
            my_name = str(my_profile.get("name") or "一位成员")
            conversation_id = str(match_record.get("conversationId") or "")
            chat_url = chat_action_url(conversation_id) if conversation_id else "/chat"
            await create_notification(
                title=f"互相喜欢 · {target_name}",
                body=f"你和 {target_name} 互相喜欢了，可以发消息约咖啡或同城见面。",
                category="match",
                priority="high",
                target_user_id=user_id,
                audience="user",
                action_label="发消息",
                action_url=chat_url,
                entity_type="match_connection",
                entity_id=match_record["id"],
                icon="💕",
            )
            await create_notification(
                title=f"互相喜欢 · {my_name}",
                body=f"你和 {my_name} 互相喜欢了，可以发消息约咖啡或同城见面。",
                category="match",
                priority="high",
                target_user_id=target_user_id,
                audience="user",
                action_label="发消息",
                action_url=chat_url,
                entity_type="match_connection",
                entity_id=match_record["id"],
                icon="💕",
            )
        else:
            await create_notification(
                title=f"已喜欢 {target_name}",
                body="对方也喜欢你的话，会在这里收到互相喜欢提醒。",
                category="match",
                priority="normal",
                target_user_id=user_id,
                audience="user",
                action_label="继续匹配",
                action_url="/dating",
                entity_type="swipe",
                entity_id=record["id"],
                icon="💚",
            )
            if target_user_id:
                swiper_name = str(my_profile.get("name") if my_profile else user.get("name") or "一位成员")
                is_superlike = payload.action == "superlike"
                await create_notification(
                    title=f"{swiper_name} {'超级赞了你' if is_superlike else '喜欢了你'}",
                    body="超级赞代表对方对你特别感兴趣，去看看资料并回一个吧。"
                    if is_superlike
                    else "去看看对方的资料，如果喜欢就回一个吧。",
                    category="match",
                    priority="high",
                    target_user_id=target_user_id,
                    audience="user",
                    action_label="去匹配",
                    action_url="/dating",
                    entity_type="swipe",
                    entity_id=record["id"],
                    icon="⭐" if is_superlike else "💚",
                )
    return {
        "ok": True,
        "record": public_record(record),
        "matched": matched,
        "match": public_record(match_record) if match_record else None,
    }


@app.post("/api/conversations/start")
async def start_conversation(payload: StartConversationPayload, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        raise HTTPException(status_code=401, detail="请先登录")
    target_user_id = payload.targetUserId.strip()
    if not target_user_id and payload.profileId.strip():
        target_profile = await pb.first_record("profiles", filter=f'id={pb_quote(payload.profileId.strip())}')
        if not target_profile:
            raise HTTPException(status_code=404, detail="资料不存在")
        target_user_id = str(target_profile.get("userId") or "")
    if not target_user_id:
        raise HTTPException(status_code=400, detail="对方尚未绑定社区账号，暂时无法私信")
    if target_user_id == user["id"]:
        raise HTTPException(status_code=400, detail="不能给自己发消息")
    intent = payload.intent if payload.intent in MATCH_INTENTS else "friends"
    conversation = await ensure_direct_conversation(
        user["id"],
        target_user_id,
        conv_type="city",
        intent=intent,
    )
    opener = payload.opener.strip()
    if opener:
        try:
            await send_dm_message(conversation=conversation, sender_id=user["id"], body=opener)
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        conversation = await conversation_for_user(conversation["id"], user["id"]) or conversation
    return {"ok": True, "conversation": await decorate_conversation(conversation, user["id"])}


@app.get("/api/conversations")
async def list_conversations(request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        raise HTTPException(status_code=401, detail="请先登录")
    conversations = await list_conversations_for_user(user["id"])
    items: list[dict[str, Any]] = []
    unread_total = 0
    for conversation in conversations:
        decorated = await decorate_conversation(conversation, user["id"])
        unread_total += int(decorated.get("unreadCount") or 0)
        items.append(decorated)
    items.sort(
        key=lambda item: str(item.get("lastMessageAt") or item.get("updated") or item.get("created") or ""),
        reverse=True,
    )
    return {"ok": True, "items": items, "total": len(items), "unread": unread_total}


@app.get("/api/conversations/by-connection/{connection_id}")
async def conversation_for_connection(connection_id: str, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        raise HTTPException(status_code=401, detail="请先登录")
    conversation = await conversation_by_connection(connection_id, user["id"])
    if not conversation:
        raise HTTPException(status_code=404, detail="会话不存在")
    return {"ok": True, "conversation": await decorate_conversation(conversation, user["id"])}


@app.get("/api/conversations/{conversation_id}")
async def get_conversation(conversation_id: str, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        raise HTTPException(status_code=401, detail="请先登录")
    conversation = await conversation_for_user(conversation_id, user["id"])
    if not conversation:
        raise HTTPException(status_code=404, detail="会话不存在")
    return {"ok": True, "conversation": await decorate_conversation(conversation, user["id"])}


@app.get("/api/conversations/{conversation_id}/messages")
async def get_conversation_messages(
    conversation_id: str,
    request: Request,
    limit: int = 50,
    before: str = "",
) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        raise HTTPException(status_code=401, detail="请先登录")
    conversation = await conversation_for_user(conversation_id, user["id"])
    if not conversation:
        raise HTTPException(status_code=404, detail="会话不存在")
    messages = await list_messages(conversation_id, limit=limit, before=before)
    await mark_conversation_read(conversation, user["id"])
    items = [
        {
            **dm_public_record(message),
            "mine": str(message.get("senderId") or "") == user["id"],
        }
        for message in messages
    ]
    return {"ok": True, "items": items, "total": len(items)}


@app.post("/api/conversations/{conversation_id}/messages")
async def post_conversation_message(
    conversation_id: str,
    payload: ChatMessagePayload,
    request: Request,
) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        raise HTTPException(status_code=401, detail="请先登录")
    conversation = await conversation_for_user(conversation_id, user["id"])
    if not conversation:
        raise HTTPException(status_code=404, detail="会话不存在")
    try:
        message, recipient_id = await send_dm_message(
            conversation=conversation,
            sender_id=user["id"],
            body=payload.body,
        )
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc

    sender_profile = await profile_for_user(user["id"])
    sender_name = str((sender_profile or {}).get("name") or user.get("name") or "一位成员")
    preview = payload.body.strip().replace("\n", " ")
    if len(preview) > 120:
        preview = f"{preview[:117]}..."

    if recipient_id:
        await create_notification(
            title=f"新私信 · {sender_name}",
            body=preview,
            category="match",
            priority="high",
            target_user_id=recipient_id,
            audience="user",
            action_label="回复",
            action_url=chat_action_url(conversation_id),
            entity_type="conversation",
            entity_id=conversation_id,
            icon="💬",
        )

    return {
        "ok": True,
        "message": {
            **dm_public_record(message),
            "mine": True,
        },
    }


@app.post("/api/conversations/{conversation_id}/read")
async def mark_conversation_as_read(conversation_id: str, request: Request) -> dict[str, Any]:
    _, user = await current_user(request)
    if not user:
        raise HTTPException(status_code=401, detail="请先登录")
    conversation = await conversation_for_user(conversation_id, user["id"])
    if not conversation:
        raise HTTPException(status_code=404, detail="会话不存在")
    updated = await mark_conversation_read(conversation, user["id"])
    return {"ok": True, "conversation": await decorate_conversation(updated, user["id"])}


@app.post("/api/feedback")
async def feedback(payload: FeedbackPayload) -> dict[str, Any]:
    record = await pb.create_record("feedback", {**payload.model_dump(), "status": "new"})
    await create_notification(
        title=f"反馈已收到：{payload.title}",
        body=payload.content[:220],
        category="system",
        priority="normal",
        audience="all",
        action_label="查看反馈页",
        action_url="/feedback",
        entity_type="feedback",
        entity_id=record["id"],
        icon="💡",
    )
    return {"ok": True, "record": record}


@app.get("/api/stats/member-map")
async def member_map() -> dict[str, Any]:
    data = await pb.list_records("cities", per_page=150, sort="idNumber")
    items = [
        {
            "id": city.get("idNumber"),
            "name": city.get("name"),
            "nameEn": city.get("nameEn"),
            "emoji": city.get("icon"),
            "lat": city.get("lat"),
            "lng": city.get("lng"),
            "nomadsNow": city.get("nomadsNow"),
        }
        for city in data.get("items", [])
    ]
    return {"ok": True, "items": items}


@app.get("/api/stats/overview")
async def stats_overview() -> dict[str, Any]:
    cities_data = await pb.list_records("cities", per_page=150, sort="idNumber")
    meetups_data = await pb.list_records("meetups", per_page=1)
    discussions_data = await pb.list_records("discussions", per_page=1)
    profiles_data = await pb.list_records("profiles", per_page=1)
    gigs_data = await pb.list_records("gigs", per_page=1)
    cities = cities_data.get("items", [])
    total_nomads = sum(int(city.get("nomadsNow") or 0) for city in cities)
    min_cost = min((int(city.get("costPerMonth") or 0) for city in cities), default=0)
    avg_cost = round(sum(int(city.get("costPerMonth") or 0) for city in cities) / max(len(cities), 1))
    avg_speed = round(sum(int(city.get("internetSpeed") or 0) for city in cities) / max(len(cities), 1))
    return {
        "ok": True,
        "stats": {
            "cities": len(cities),
            "nomads": total_nomads,
            "meetups": meetups_data.get("totalItems", 0),
            "discussions": discussions_data.get("totalItems", 0),
            "profiles": profiles_data.get("totalItems", 0),
            "gigs": gigs_data.get("totalItems", 0),
            "minCost": min_cost,
            "avgCost": avg_cost,
            "avgInternet": avg_speed,
        },
    }


@app.get("/api/stats/city-ranking")
async def city_ranking(metric: str = "overallScore", limit: int = 12) -> dict[str, Any]:
    allowed = {
        "overallScore": ("overallScore", True),
        "cost": ("costPerMonth", False),
        "internet": ("internetSpeed", True),
        "community": ("nomadsNow", True),
        "temperature": ("temperature", True),
    }
    field, desc = allowed.get(metric, allowed["overallScore"])
    cities = (await pb.list_records("cities", per_page=150, sort="idNumber")).get("items", [])
    cities.sort(key=lambda city: float(city.get(field) or 0), reverse=desc)
    size = max(1, min(limit, 50))
    return {"ok": True, "items": [public_record(city) for city in cities[:size]], "metric": metric}


@app.get("/api/geo/china")
async def china_geo() -> dict[str, Any]:
    return {"type": "FeatureCollection", "features": []}


@app.post("/api/ai/assistant")
async def ai_assistant(payload: dict[str, Any]) -> dict[str, Any]:
    question = str(payload.get("question", "")).lower()
    cities = (await pb.list_records("cities", per_page=150, sort="-overallScore")).get("items", [])
    picks = cities[:3]
    if "便宜" in question or "budget" in question or "预算" in question:
        picks = sorted(cities, key=lambda c: c.get("costPerMonth", 999999))[:3]
    elif "网" in question or "internet" in question:
        picks = sorted(cities, key=lambda c: c.get("internetSpeed", 0), reverse=True)[:3]
    answer = "根据当前城市数据，推荐：" + "、".join(
        f"{c['name']}（¥{c['costPerMonth']}/月，{c['internetSpeed']}Mbps）" for c in picks
    )
    return {"ok": True, "answer": answer, "cities": [public_record(c) for c in picks]}