3162 lines
123 KiB
Python
3162 lines
123 KiB
Python
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]}
|