10 Commits
cpay ... memos

Author SHA1 Message Date
eric
18d098f5f8 'memos' 2026-03-15 04:26:05 -05:00
eric
fcf990c133 remove pycache from git tracking 2026-03-14 00:47:09 +08:00
eric
8f9bf47713 sss 2026-03-14 00:47:00 +08:00
eric
32bfc89792 ss 2026-03-14 00:36:22 +08:00
eric
4b6955fc93 s 2026-03-14 00:36:22 +08:00
eric
4c618af91a s 2026-03-14 00:36:22 +08:00
eric
0326ab2875 'env' 2026-03-13 10:59:27 -05:00
eric
4634b16cf8 'c1' 2026-03-13 05:52:35 -05:00
eric
53cacfac46 'c1' 2026-03-13 05:15:00 -05:00
eric
90378502cf 's' 2026-03-12 19:50:50 -05:00
50 changed files with 2911 additions and 166 deletions

23
.env Normal file
View File

@@ -0,0 +1,23 @@
# ========== 生产环境配置 ==========
BASE_URL=https://api.hackrobot.cn
PAYMENT_PROVIDER=zpay
# ZPAY请填入实际值
ZPAY_PID=2025121809351743
ZPAY_KEY=tpEi7wWIWI2kXiYVTpIG6j7it0mjVW89
# PocketBasecheck-user、ensure-user 必需,否则报「服务配置错误」)
PB_URL=https://pocketbase.hackrobot.cn
PB_ADMIN_EMAIL=xiaoshuang.eric@gmail.com
PB_ADMIN_PASSWORD=Xiao4669805@
# Memosnomadvip 支付成功后创建账号)
MEMOS_API=https://qun.hackrobot.cn/api/v1/users
MEMOS_TOKEN=eyJhbGciOiJIUzI1NiIsImtpZCI6InYxIiwidHlwIjoiSldUIn0.eyJuYW1lIjoiaGFja3JvYm90IiwiaXNzIjoibWVtb3MiLCJzdWIiOiIxIiwiYXVkIjpbInVzZXIuYWNjZXNzLXRva2VuIl0sImlhdCI6MTc1NzQwNTE0OX0.Idn7kBlxE-CoSOXwWuZ1tHGIRKHAIeDyXSafGS5OHsg
# MinIO可选
MINIO_ENDPOINT=minioweb.hackrobot.cn
MINIO_PORT=443
MINIO_BUCKET=hackrobot
MINIO_UPLOAD_PREFIX=joins

View File

@@ -1,20 +1,12 @@
# ========== 本地调试 ==========
# PORT: 默认 8007与前端 PAYMENT_API_URL 一致
# BASE_URL 需为 ZPAY 可访问的回调地址。本地调试时 ZPAY 无法回调 localhost/内网,
# 可用 ngrok 等隧道BASE_URL=https://xxx.ngrok.io
# 或直接使用生产地址测试回调(需 payjsapi 已部署到公网)
# nomadvip/digital/cnomadcna/nomadlms 的 zpay_notify 均依赖此地址,未正确配置会导致 PocketBase/Memos 不更新
# 本地调试ngrok 暴露 8700 后设置 BASE_URL=https://xxx.ngrok.io
# 线上部署BASE_URL=https://api.hackrobot.cn
# ========== 生产部署Ubuntu 等)==========
# 线上部署必须正确配置,否则会显示「支付失败」:
# 1. BASE_URL 必须为 payjsapi 公网地址ZPAY 需能访问此地址做异步回调
# 错误示例http://127.0.0.1:8700、http://localhostZPAY 无法访问)
# 正确示例https://api.hackrobot.cn域名需解析到本机并配置 nginx 反向代理)
# 2. 服务器需能访问 zpayz.cn检查防火墙、DNS可访问 /health?check_zpay=1 自检
PORT=8007
BASE_URL=https://api.hackrobot.cn
# 支付渠道zpay(已替代 xorpay| xorpay
# 支付渠道zpay | xorpay
PAYMENT_PROVIDER=zpay
# XorPay
@@ -25,14 +17,14 @@ XORPAY_SECRET=your_secret
ZPAY_PID=2025121809351743
ZPAY_KEY=tpEi7wWIWI2kXiYVTpIG6j7it0mjVW89
# PocketBase
# PocketBasecheck-user、ensure-user、支付回调、nomadvip/digital/cnomadcna 共用)
PB_URL=https://pocketbase.hackrobot.cn
PB_ADMIN_EMAIL=your@email.com
PB_ADMIN_PASSWORD=your_password
PB_ADMIN_EMAIL=xiaoshuang.eric@gmail.com
PB_ADMIN_PASSWORD=Xiao4669805@
# Memosnomadvip 使用
# Memosnomadvip 支付成功后创建账号,与 payjsapi 同源
MEMOS_API=https://qun.hackrobot.cn/api/v1/users
MEMOS_TOKEN=your_token
MEMOS_TOKEN=eyJhbGciOiJIUzI1NiIsImtpZCI6InYxIiwidHlwIjoiSldUIn0.eyJuYW1lIjoiaGFja3JvYm90IiwiaXNzIjoibWVtb3MiLCJzdWIiOiIxIiwiYXVkIjpbInVzZXIuYWNjZXNzLXRva2VuIl0sImlhdCI6MTc1NzQwNTE0OX0.Idn7kBlxE-CoSOXwWuZ1tHGIRKHAIeDyXSafGS5OHsg
# MinIO 对象存储(可选)
MINIO_ENDPOINT=minioweb.hackrobot.cn

17
.env.local Normal file
View File

@@ -0,0 +1,17 @@
# 本地调试 - 端口 8007、zpay 渠道
PORT=8007
PAYMENT_PROVIDER=zpay
BASE_URL=https://api.hackrobot.cn
# PocketBase支付回调、nomadvip/digital/cnomadcna 共用)
PB_URL=https://pocketbase.hackrobot.cn
PB_ADMIN_EMAIL=xiaoshuang.eric@gmail.com
PB_ADMIN_PASSWORD=Xiao4669805@
# Memosnomadvip 支付成功后创建账号)
MEMOS_API=https://qun.hackrobot.cn/api/v1/users
MEMOS_TOKEN=eyJhbGciOiJIUzI1NiIsImtpZCI6InYxIiwidHlwIjoiSldUIn0.eyJuYW1lIjoiaGFja3JvYm90IiwiaXNzIjoibWVtb3MiLCJzdWIiOiIxIiwiYXVkIjpbInVzZXIuYWNjZXNzLXRva2VuIl0sImlhdCI6MTc1NzQwNTE0OX0.Idn7kBlxE-CoSOXwWuZ1tHGIRKHAIeDyXSafGS5OHsg
# ZPAY
ZPAY_PID=2025121809351743
ZPAY_KEY=tpEi7wWIWI2kXiYVTpIG6j7it0mjVW89

4
.gitignore vendored
View File

@@ -9,5 +9,5 @@ __pycache__/
*.py[cod]
venv/
.venv/
.env
.env.local
# .env
# .env.local

View File

@@ -84,13 +84,19 @@ PAYMENT_PROVIDER=zpay
## 线上「支付失败」排查
本地正常、Ubuntu 部署后显示支付失败,常见原因:
本地正常、Ubuntu 部署后 api.hackrobot.cn 调用支付失败,常见原因:
1. **BASE_URL 配置错误**:必须为 ZPAY 可公网访问的地址。错误示例:`http://127.0.0.1:8700``http://localhost`。正确:`https://api.hackrobot.cn`域名需解析到本机nginx 反向代理 8700 端口)。
1. **BASE_URL 配置错误**:必须为 ZPAY 可公网访问的地址。错误示例:`http://127.0.0.1:8700``http://localhost`。正确:`https://api.hackrobot.cn`域名需解析到本机nginx 反向代理对应端口)。
2. **服务器无法访问 zpayz.cn**防火墙、DNS 或网络限制。访问 `https://你的域名/health?check_zpay=1` 自检,若 `zpayz_reachable: false` 则需开放出站或检查网络。
2. **服务器无法访问 zpayz.cn**防火墙、DNS 或网络限制。访问 `https://api.hackrobot.cn/health?check_zpay=1` 自检
- `zpayz_reachable: false` → 开放出站或检查网络
- `zpayz_error` 含 "SSL" → 执行 `apt install ca-certificates` 后重启
3. **启动时控制台有 BASE_URL 警告**:说明配置了内网地址,需修改 `.env` 中的 `BASE_URL` 为公网地址后重启
3. **client_ip 内网导致 ZPAY 风控拒单**前端digital/cnomadcna/nomadvip通过 nginx 代理时,需配置 `proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for`,否则 payjsapi 收到 127.0.0.1 会触发风控。已做兼容:内网 IP 时自动省略 clientip 参数
4. **启动时控制台有 BASE_URL 警告**:说明配置了内网地址,需修改 `.env` 中的 `BASE_URL` 为公网地址后重启。
5. **查看详细错误**支付失败时payjsapi 会输出 `ZPAY create_order_api 失败` 日志,含 status/code/msg便于定位。
## Getting started

Binary file not shown.

View File

@@ -3,8 +3,16 @@
支付渠道可通过 PAYMENT_PROVIDER 切换xorpay | zpay
"""
import os
from pathlib import Path
from typing import Literal
from dotenv import load_dotenv
# 强制从 payjsapi 项目根目录加载 .env、.env.local避免 cwd 不同导致加载失败
_root = Path(__file__).resolve().parent.parent
load_dotenv(_root / ".env")
load_dotenv(_root / ".env.local", override=True)
PaymentChannel = Literal["xorpay", "zpay"]
# 当前使用的支付渠道zpay 替代 xorpaypayjs2 接口业务保留)
@@ -19,10 +27,14 @@ XORPAY_CASHIER_URL = os.getenv("XORPAY_CASHIER_URL", "https://xorpay.com/api/cas
ZPAY_PID = os.getenv("ZPAY_PID", "2025121809351743")
ZPAY_KEY = os.getenv("ZPAY_KEY", "tpEi7wWIWI2kXiYVTpIG6j7it0mjVW89")
# PocketBase
PB_URL = os.getenv("PB_URL", "https://pocketbase.hackrobot.cn")
PB_ADMIN_EMAIL = os.getenv("PB_ADMIN_EMAIL", "xiaoshuang.eric@gmail.com")
PB_ADMIN_PASSWORD = os.getenv("PB_ADMIN_PASSWORD", "Xiao4669805@")
# PocketBase(支持 PB_ADMIN_* 或 POCKETBASE_*,空值时用默认)
PB_URL = os.getenv("PB_URL") or os.getenv("POCKETBASE_URL") or "https://pocketbase.hackrobot.cn"
PB_ADMIN_EMAIL = (
os.getenv("PB_ADMIN_EMAIL") or os.getenv("POCKETBASE_EMAIL") or "xiaoshuang.eric@gmail.com"
)
PB_ADMIN_PASSWORD = (
os.getenv("PB_ADMIN_PASSWORD") or os.getenv("POCKETBASE_PASSWORD") or "Xiao4669805@"
)
# Memos
MEMOS_API = os.getenv("MEMOS_API", "https://qun.hackrobot.cn/api/v1/users")

1
app/db/__init__.py Normal file
View File

@@ -0,0 +1 @@
# Database modules

753
app/db/community_sqlite.py Normal file
View File

@@ -0,0 +1,753 @@
"""
SQLite storage for the private community module.
"""
from __future__ import annotations
from collections import Counter
from pathlib import Path
from typing import Optional
import os
import re
import sqlite3
import time
import uuid
_ROOT = Path(__file__).resolve().parent.parent.parent
DATA_DIR = Path(os.getenv("COMMUNITY_DATA_DIR", str(_ROOT / "data")))
DB_PATH = DATA_DIR / "community.db"
UPLOADS_DIR = DATA_DIR / "community_uploads"
UPVOTE_REACTION = "UPVOTE"
def _ensure_dirs() -> None:
DATA_DIR.mkdir(parents=True, exist_ok=True)
UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
def _get_conn() -> sqlite3.Connection:
_ensure_dirs()
conn = sqlite3.connect(str(DB_PATH))
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
return conn
def _now_iso() -> str:
return time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.gmtime())
def _normalize_memo_id(memo_id: str) -> str:
return memo_id.replace("memos/", "") if memo_id.startswith("memos/") else memo_id
def _extract_tags(content: str) -> list[str]:
matches = re.findall(r"#[\w\u4e00-\u9fa5]+", content or "")
return list(dict.fromkeys(match.strip("#") for match in matches))
def _extract_attachments(content: str) -> list[dict]:
items = []
for match in re.finditer(r"!\[([^\]]*)\]\(([^)]+)\)", content or ""):
alt, url = match.group(1).strip(), match.group(2).strip()
if url:
items.append({"url": url, "alt": alt})
return items
def init_schema() -> None:
_ensure_dirs()
conn = _get_conn()
try:
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS community_memos (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
content TEXT NOT NULL,
pinned INTEGER NOT NULL DEFAULT 0,
visibility TEXT NOT NULL DEFAULT 'PRIVATE',
state TEXT NOT NULL DEFAULT 'NORMAL',
created TEXT NOT NULL,
updated TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_memos_user ON community_memos(user_id);
CREATE INDEX IF NOT EXISTS idx_memos_state ON community_memos(state);
CREATE INDEX IF NOT EXISTS idx_memos_created ON community_memos(created);
CREATE INDEX IF NOT EXISTS idx_memos_pinned ON community_memos(pinned);
CREATE TABLE IF NOT EXISTS community_uploads (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
original_name TEXT NOT NULL,
filename TEXT NOT NULL UNIQUE,
content_type TEXT NOT NULL,
size INTEGER NOT NULL DEFAULT 0,
url TEXT NOT NULL,
created TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_uploads_user ON community_uploads(user_id);
CREATE INDEX IF NOT EXISTS idx_uploads_created ON community_uploads(created);
CREATE TABLE IF NOT EXISTS community_comments (
id TEXT PRIMARY KEY,
memo_id TEXT NOT NULL,
user_id TEXT NOT NULL,
content TEXT NOT NULL,
created TEXT NOT NULL,
updated TEXT NOT NULL,
FOREIGN KEY (memo_id) REFERENCES community_memos(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_comments_memo ON community_comments(memo_id, created);
CREATE INDEX IF NOT EXISTS idx_comments_user ON community_comments(user_id);
CREATE TABLE IF NOT EXISTS community_reactions (
id TEXT PRIMARY KEY,
memo_id TEXT NOT NULL,
user_id TEXT NOT NULL,
reaction_type TEXT NOT NULL,
created TEXT NOT NULL,
UNIQUE(memo_id, user_id, reaction_type),
FOREIGN KEY (memo_id) REFERENCES community_memos(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_reactions_memo ON community_reactions(memo_id);
CREATE INDEX IF NOT EXISTS idx_reactions_user ON community_reactions(user_id);
"""
)
conn.commit()
finally:
conn.close()
def _base_memo_from_row(row: sqlite3.Row) -> dict:
content = row["content"]
return {
"id": row["id"],
"user_id": row["user_id"],
"content": content,
"pinned": bool(row["pinned"]),
"visibility": row["visibility"] or "PRIVATE",
"state": row["state"] or "NORMAL",
"created": row["created"],
"updated": row["updated"],
"tags": _extract_tags(content),
"attachments": _extract_attachments(content),
}
def _comment_from_row(row: sqlite3.Row) -> dict:
return {
"id": row["id"],
"memo_id": row["memo_id"],
"user_id": row["user_id"],
"content": row["content"],
"created": row["created"],
"updated": row["updated"],
}
def _comment_count(conn: sqlite3.Connection, memo_id: str) -> int:
cursor = conn.execute("SELECT COUNT(*) FROM community_comments WHERE memo_id = ?", (memo_id,))
return int(cursor.fetchone()[0] or 0)
def _reaction_count(conn: sqlite3.Connection, memo_id: str) -> int:
cursor = conn.execute(
"SELECT COUNT(*) FROM community_reactions WHERE memo_id = ? AND reaction_type = ?",
(memo_id, UPVOTE_REACTION),
)
return int(cursor.fetchone()[0] or 0)
def _liked_by_user(conn: sqlite3.Connection, memo_id: str, viewer_id: str = "") -> bool:
if not viewer_id:
return False
cursor = conn.execute(
"""
SELECT 1
FROM community_reactions
WHERE memo_id = ? AND user_id = ? AND reaction_type = ?
LIMIT 1
""",
(memo_id, viewer_id, UPVOTE_REACTION),
)
return cursor.fetchone() is not None
def list_comments(memo_id: str) -> list[dict]:
uid = _normalize_memo_id(memo_id)
conn = _get_conn()
try:
rows = conn.execute(
"""
SELECT id, memo_id, user_id, content, created, updated
FROM community_comments
WHERE memo_id = ?
ORDER BY created ASC
""",
(uid,),
).fetchall()
return [_comment_from_row(row) for row in rows]
finally:
conn.close()
def _enrich_memo(conn: sqlite3.Connection, memo: dict, viewer_id: str = "", include_comments: bool = False) -> dict:
enriched = dict(memo)
enriched["comment_count"] = _comment_count(conn, memo["id"])
enriched["reaction_count"] = _reaction_count(conn, memo["id"])
enriched["liked_by_me"] = _liked_by_user(conn, memo["id"], viewer_id)
if include_comments:
enriched["comments"] = list_comments(memo["id"])
return enriched
def _to_memo_format(memo: dict) -> dict:
payload = {
"name": f"memos/{memo['id']}",
"id": memo["id"],
"creator": f"users/{memo['user_id']}",
"user_id": memo["user_id"],
"content": memo["content"],
"pinned": memo["pinned"],
"visibility": memo["visibility"],
"state": memo["state"],
"create_time": memo["created"],
"update_time": memo["updated"],
"display_time": memo["created"],
"tags": memo["tags"],
"attachments": memo["attachments"],
"comment_count": memo["comment_count"],
"reaction_count": memo["reaction_count"],
"liked_by_me": memo["liked_by_me"],
}
if "comments" in memo:
payload["comments"] = memo["comments"]
return payload
def _normalize_memo(memo: dict) -> dict:
payload = {
"id": memo["id"],
"user_id": memo["user_id"],
"content": memo["content"],
"pinned": memo["pinned"],
"visibility": memo["visibility"],
"state": memo["state"],
"created": memo["created"],
"updated": memo["updated"],
"tags": memo["tags"],
"attachments": memo["attachments"],
"comment_count": memo["comment_count"],
"reaction_count": memo["reaction_count"],
"liked_by_me": memo["liked_by_me"],
}
if "comments" in memo:
payload["comments"] = memo["comments"]
return payload
def _build_where_clause(state: str = "NORMAL", query: str = "") -> tuple[str, list[str]]:
clauses: list[str] = []
params: list[str] = []
normalized_state = (state or "NORMAL").upper()
if normalized_state != "ALL":
clauses.append("state = ?")
params.append(normalized_state)
if query:
lowered = query.lower()
clauses.append("LOWER(content) LIKE ?")
params.append(f"%{lowered}%")
if not clauses:
return "", params
return f"WHERE {' AND '.join(clauses)}", params
def list_memos(
page: int = 1,
per_page: int = 50,
state: str = "NORMAL",
order_by: str = "-pinned,-created",
memo_format: bool = False,
query: str = "",
viewer_id: str = "",
) -> dict:
conn = _get_conn()
try:
order_clause = "pinned DESC, created DESC"
if "created" in order_by and "pinned" not in order_by:
order_clause = "created DESC"
offset = (page - 1) * per_page
where_clause, params = _build_where_clause(state=state, query=query)
total = int(
conn.execute(f"SELECT COUNT(*) FROM community_memos {where_clause}", params).fetchone()[0] or 0
)
rows = conn.execute(
f"""
SELECT id, user_id, content, pinned, visibility, state, created, updated
FROM community_memos
{where_clause}
ORDER BY {order_clause}
LIMIT ? OFFSET ?
""",
[*params, per_page, offset],
).fetchall()
items = []
for row in rows:
enriched = _enrich_memo(conn, _base_memo_from_row(row), viewer_id=viewer_id)
items.append(_to_memo_format(enriched) if memo_format else _normalize_memo(enriched))
next_token = "" if len(items) < per_page else str(page + 1)
return {
"memos": items,
"items": items,
"totalItems": total,
"page": page,
"pageSize": per_page,
"next_page_token": next_token,
}
finally:
conn.close()
def get_memo(memo_id: str, viewer_id: str = "", include_comments: bool = True) -> dict | None:
uid = _normalize_memo_id(memo_id)
conn = _get_conn()
try:
row = conn.execute(
"""
SELECT id, user_id, content, pinned, visibility, state, created, updated
FROM community_memos
WHERE id = ?
""",
(uid,),
).fetchone()
if not row:
return None
enriched = _enrich_memo(
conn,
_base_memo_from_row(row),
viewer_id=viewer_id,
include_comments=include_comments,
)
return _to_memo_format(enriched)
finally:
conn.close()
def create_memo(
user_id: str,
content: str,
pinned: bool = False,
visibility: str = "PRIVATE",
state: str = "NORMAL",
) -> dict:
content = (content or "").strip()
if not content:
raise ValueError("content cannot be empty")
if len(content) > 10000:
raise ValueError("content is too long")
memo_id = uuid.uuid4().hex[:16]
now = _now_iso()
conn = _get_conn()
try:
conn.execute(
"""
INSERT INTO community_memos (id, user_id, content, pinned, visibility, state, created, updated)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(memo_id, user_id, content, 1 if pinned else 0, visibility, state, now, now),
)
conn.commit()
memo = _enrich_memo(
conn,
{
"id": memo_id,
"user_id": user_id,
"content": content,
"pinned": pinned,
"visibility": visibility,
"state": state,
"created": now,
"updated": now,
"tags": _extract_tags(content),
"attachments": _extract_attachments(content),
},
viewer_id=user_id,
)
return _to_memo_format(memo)
finally:
conn.close()
def update_memo(
memo_id: str,
user_id: str,
pinned: Optional[bool] = None,
content: Optional[str] = None,
visibility: Optional[str] = None,
state: Optional[str] = None,
) -> dict | None:
uid = _normalize_memo_id(memo_id)
conn = _get_conn()
try:
row = conn.execute("SELECT * FROM community_memos WHERE id = ?", (uid,)).fetchone()
if not row:
return None
memo = _base_memo_from_row(row)
if memo["user_id"] != user_id:
return None
updates = []
params = []
if pinned is not None:
updates.append("pinned = ?")
params.append(1 if pinned else 0)
if content is not None:
content = content.strip()
if not content:
raise ValueError("content cannot be empty")
if len(content) > 10000:
raise ValueError("content is too long")
updates.append("content = ?")
params.append(content)
if visibility is not None:
updates.append("visibility = ?")
params.append(visibility)
if state is not None:
updates.append("state = ?")
params.append(state)
if not updates:
return _to_memo_format(_enrich_memo(conn, memo, viewer_id=user_id))
now = _now_iso()
updates.append("updated = ?")
params.append(now)
params.append(uid)
conn.execute(f"UPDATE community_memos SET {', '.join(updates)} WHERE id = ?", params)
conn.commit()
latest_content = content if content is not None else memo["content"]
memo.update(
{
"pinned": pinned if pinned is not None else memo["pinned"],
"content": latest_content,
"visibility": visibility if visibility is not None else memo["visibility"],
"state": state if state is not None else memo["state"],
"updated": now,
"tags": _extract_tags(latest_content),
"attachments": _extract_attachments(latest_content),
}
)
return _to_memo_format(_enrich_memo(conn, memo, viewer_id=user_id))
finally:
conn.close()
def delete_memo(memo_id: str, user_id: str) -> bool:
uid = _normalize_memo_id(memo_id)
conn = _get_conn()
try:
row = conn.execute("SELECT user_id FROM community_memos WHERE id = ?", (uid,)).fetchone()
if not row or row["user_id"] != user_id:
return False
conn.execute("DELETE FROM community_memos WHERE id = ?", (uid,))
conn.commit()
return True
finally:
conn.close()
def create_comment(memo_id: str, user_id: str, content: str) -> dict:
uid = _normalize_memo_id(memo_id)
content = (content or "").strip()
if not content:
raise ValueError("comment content cannot be empty")
if len(content) > 3000:
raise ValueError("comment content is too long")
conn = _get_conn()
try:
memo_exists = conn.execute("SELECT 1 FROM community_memos WHERE id = ?", (uid,)).fetchone()
if not memo_exists:
raise ValueError("memo does not exist")
comment_id = uuid.uuid4().hex[:16]
now = _now_iso()
conn.execute(
"""
INSERT INTO community_comments (id, memo_id, user_id, content, created, updated)
VALUES (?, ?, ?, ?, ?, ?)
""",
(comment_id, uid, user_id, content, now, now),
)
conn.commit()
return {
"id": comment_id,
"memo_id": uid,
"user_id": user_id,
"content": content,
"created": now,
"updated": now,
}
finally:
conn.close()
def delete_comment(comment_id: str, user_id: str) -> bool:
conn = _get_conn()
try:
row = conn.execute(
"""
SELECT c.user_id AS comment_user_id, m.user_id AS memo_user_id
FROM community_comments c
JOIN community_memos m ON m.id = c.memo_id
WHERE c.id = ?
""",
(comment_id,),
).fetchone()
if not row:
return False
if row["comment_user_id"] != user_id and row["memo_user_id"] != user_id:
return False
conn.execute("DELETE FROM community_comments WHERE id = ?", (comment_id,))
conn.commit()
return True
finally:
conn.close()
def toggle_reaction(memo_id: str, user_id: str, reaction_type: str = UPVOTE_REACTION) -> dict:
uid = _normalize_memo_id(memo_id)
reaction_type = (reaction_type or UPVOTE_REACTION).upper()
conn = _get_conn()
try:
memo_exists = conn.execute("SELECT 1 FROM community_memos WHERE id = ?", (uid,)).fetchone()
if not memo_exists:
raise ValueError("memo does not exist")
existing = conn.execute(
"""
SELECT id
FROM community_reactions
WHERE memo_id = ? AND user_id = ? AND reaction_type = ?
""",
(uid, user_id, reaction_type),
).fetchone()
liked = False
if existing:
conn.execute("DELETE FROM community_reactions WHERE id = ?", (existing["id"],))
else:
conn.execute(
"""
INSERT INTO community_reactions (id, memo_id, user_id, reaction_type, created)
VALUES (?, ?, ?, ?, ?)
""",
(uuid.uuid4().hex[:16], uid, user_id, reaction_type, _now_iso()),
)
liked = True
conn.commit()
return {
"memo_id": uid,
"reaction_type": reaction_type,
"liked": liked,
"reaction_count": _reaction_count(conn, uid),
}
finally:
conn.close()
def save_upload(data: bytes, filename: str, content_type: str, user_id: str, base_url: str) -> dict:
_ensure_dirs()
ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else "jpg"
if ext not in ("jpg", "jpeg", "png", "gif", "webp"):
ext = "jpg"
upload_id = uuid.uuid4().hex[:16]
stored_name = f"{int(time.time() * 1000)}_{upload_id[:8]}.{ext}"
path = UPLOADS_DIR / stored_name
path.write_bytes(data)
now = _now_iso()
url = f"{base_url.rstrip('/')}/community/files/{stored_name}"
conn = _get_conn()
try:
conn.execute(
"""
INSERT INTO community_uploads (id, user_id, original_name, filename, content_type, size, url, created)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(upload_id, user_id, filename, stored_name, content_type, len(data), url, now),
)
conn.commit()
finally:
conn.close()
return {
"id": upload_id,
"url": url,
"markdown": f"![{filename}]({url})",
"filename": stored_name,
"original_name": filename,
"content_type": content_type,
"size": len(data),
"created": now,
}
def list_uploads(page: int = 1, per_page: int = 24, user_id: str = "") -> dict:
offset = (page - 1) * per_page
conn = _get_conn()
try:
where_clause = ""
params: list[str] = []
if user_id:
where_clause = "WHERE user_id = ?"
params.append(user_id)
total = int(
conn.execute(f"SELECT COUNT(*) FROM community_uploads {where_clause}", params).fetchone()[0] or 0
)
rows = conn.execute(
f"""
SELECT id, user_id, original_name, filename, content_type, size, url, created
FROM community_uploads
{where_clause}
ORDER BY created DESC
LIMIT ? OFFSET ?
""",
[*params, per_page, offset],
).fetchall()
items = [
{
"id": row["id"],
"user_id": row["user_id"],
"filename": row["filename"],
"original_name": row["original_name"],
"content_type": row["content_type"],
"size": row["size"],
"url": row["url"],
"created": row["created"],
}
for row in rows
]
next_token = "" if len(items) < per_page else str(page + 1)
return {
"items": items,
"totalItems": total,
"page": page,
"pageSize": per_page,
"next_page_token": next_token,
}
finally:
conn.close()
def get_stats(member_count: int | None = None) -> dict:
conn = _get_conn()
try:
totals = conn.execute(
"""
SELECT
COUNT(*) AS total_memos,
SUM(CASE WHEN state = 'NORMAL' THEN 1 ELSE 0 END) AS normal_memos,
SUM(CASE WHEN state = 'ARCHIVED' THEN 1 ELSE 0 END) AS archived_memos,
SUM(CASE WHEN pinned = 1 THEN 1 ELSE 0 END) AS pinned_memos,
COUNT(DISTINCT user_id) AS contributors
FROM community_memos
"""
).fetchone()
total_uploads = int(conn.execute("SELECT COUNT(*) FROM community_uploads").fetchone()[0] or 0)
total_comments = int(conn.execute("SELECT COUNT(*) FROM community_comments").fetchone()[0] or 0)
total_reactions = int(conn.execute("SELECT COUNT(*) FROM community_reactions").fetchone()[0] or 0)
today_prefix = _now_iso()[:10]
today_memos = int(
conn.execute("SELECT COUNT(*) FROM community_memos WHERE created LIKE ?", (f"{today_prefix}%",)).fetchone()[0]
or 0
)
top_authors_rows = conn.execute(
"""
SELECT user_id, COUNT(*) AS memo_count, MAX(created) AS latest_created
FROM community_memos
WHERE state = 'NORMAL'
GROUP BY user_id
ORDER BY memo_count DESC, latest_created DESC
LIMIT 6
"""
).fetchall()
top_authors = [
{
"user_id": row["user_id"],
"memo_count": row["memo_count"],
"latest_created": row["latest_created"],
}
for row in top_authors_rows
]
tag_counter: Counter[str] = Counter()
for row in conn.execute("SELECT content FROM community_memos WHERE state = 'NORMAL'").fetchall():
tag_counter.update(_extract_tags(row["content"]))
top_tags = [{"tag": tag, "count": count} for tag, count in tag_counter.most_common(10)]
latest_rows = conn.execute(
"""
SELECT id, user_id, content, pinned, visibility, state, created, updated
FROM community_memos
WHERE state = 'NORMAL'
ORDER BY pinned DESC, created DESC
LIMIT 3
"""
).fetchall()
latest_memos = [_normalize_memo(_enrich_memo(conn, _base_memo_from_row(row))) for row in latest_rows]
return {
"stats": {
"members": member_count,
"total_memos": int(totals["total_memos"] or 0),
"normal_memos": int(totals["normal_memos"] or 0),
"archived_memos": int(totals["archived_memos"] or 0),
"pinned_memos": int(totals["pinned_memos"] or 0),
"contributors": int(totals["contributors"] or 0),
"attachments": total_uploads,
"comments": total_comments,
"reactions": total_reactions,
"today_memos": today_memos,
},
"top_tags": top_tags,
"top_authors": top_authors,
"latest_memos": latest_memos,
}
finally:
conn.close()
def get_upload_path(filename: str) -> Path | None:
if not filename or ".." in filename or "/" in filename or "\\" in filename:
return None
path = UPLOADS_DIR / filename
if not path.exists() or not path.is_file():
return None
return path

View File

@@ -14,6 +14,7 @@
线上部署BASE_URL=https://api.hackrobot.cn必须为 ZPAY 可公网访问的地址)
"""
import logging
import sys
import requests
from fastapi import FastAPI, Request, HTTPException
@@ -28,6 +29,8 @@ from .routers import (
nomadlms_router,
legacy_router,
common_router,
meetup_router,
community_router,
)
app = FastAPI(
@@ -45,21 +48,57 @@ app.add_middleware(
)
class _SafeStderrProxy:
def write(self, message: str) -> int:
try:
return sys.stderr.write(message)
except (OSError, ValueError):
return 0
def flush(self) -> None:
try:
sys.stderr.flush()
except (OSError, ValueError):
pass
def _safe_log(level: int, message: str, *args, **kwargs) -> None:
try:
logging.log(level, message, *args, **kwargs)
except Exception:
pass
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
"""捕获未处理异常,记录日志后返回 500HTTPException 交由 FastAPI 默认处理)"""
if isinstance(exc, HTTPException):
raise exc
logging.error(f"未处理异常: {request.url.path} - {exc}", exc_info=True)
_safe_log(logging.ERROR, "未处理异常: %s - %s", request.url.path, exc, exc_info=True)
return JSONResponse(status_code=500, content={"detail": str(exc)})
# 日志(仅控制台,不写文件,避免 git 本地/线上不一致)
logging.raiseExceptions = False
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(message)s",
handlers=[logging.StreamHandler()],
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
handlers=[logging.StreamHandler(_SafeStderrProxy())],
force=True,
)
def _safe_stderr_write(message: str) -> None:
try:
sys.stderr.write(message)
sys.stderr.flush()
except (OSError, ValueError):
pass
def _should_trace_request(path: str) -> bool:
return any(
segment in path for segment in ("/nomadvip", "/digital", "/cnomadcna", "/nomadlms")
)
# 启动时校验 BASE_URL线上部署必须为公网可访问地址否则 ZPAY 无法回调)
def _check_base_url():
url = (BASE_URL or "").strip().lower()
@@ -80,18 +119,45 @@ async def startup_event():
@app.get("/health")
async def health(check_zpay: str = ""):
"""健康检查。check_zpay=1 时额外检测 zpayz.cn 连通性"""
"""健康检查。check_zpay=1 时额外检测 zpayz.cn 连通性(含 SSL、mapi 接口)"""
out = {"status": "ok", "service": "payjsapi", "base_url": BASE_URL}
if str(check_zpay).lower() in ("1", "true", "yes"):
# 1) 检测 zpayz.cn 首页
try:
r = requests.get("https://zpayz.cn", timeout=5)
out["zpayz_reachable"] = r.status_code in (200, 301, 302, 404)
except Exception as e:
except requests.exceptions.SSLError as e:
out["zpayz_reachable"] = False
out["zpayz_error"] = f"SSL 错误: {e}Ubuntu 请执行: apt install ca-certificates"
except requests.RequestException as e:
out["zpayz_reachable"] = False
out["zpayz_error"] = str(e)
# 2) 检测 mapi 接口(实际支付调用地址)
if out.get("zpayz_reachable"):
try:
r2 = requests.post("https://zpayz.cn/mapi.php", data={"pid": "test"}, timeout=5)
out["zpayz_mapi_reachable"] = True
out["zpayz_mapi_status"] = r2.status_code
except requests.exceptions.SSLError as e:
out["zpayz_mapi_reachable"] = False
out["zpayz_mapi_error"] = f"SSL: {e}"
except requests.RequestException as e:
out["zpayz_mapi_reachable"] = False
out["zpayz_mapi_error"] = str(e)
return out
# 请求日志中间件:每个请求都打印到 stderr确保终端可见
@app.middleware("http")
async def log_requests(request, call_next):
path = request.url.path
if _should_trace_request(path):
_safe_stderr_write(f"[payjsapi] >>> {request.method} {path}\n")
resp = await call_next(request)
if _should_trace_request(path):
_safe_stderr_write(f"[payjsapi] <<< {request.method} {path} {resp.status_code}\n")
return resp
# 挂载路由
app.include_router(nomadvip_router)
app.include_router(digital_router)
@@ -99,6 +165,8 @@ app.include_router(cnomadcna_router)
app.include_router(nomadlms_router)
app.include_router(legacy_router)
app.include_router(common_router)
app.include_router(meetup_router)
app.include_router(community_router)
@app.get("/")
@@ -108,11 +176,12 @@ async def root():
"status": "ok",
"service": "payjsapi",
"routes": {
"nomadvip": "/nomadvip/payh5, /nomadvip/payh5/redirect, /nomadvip/zpay_order_status, /nomadvip/zpay_notify, /nomadvip/xorpay_notify",
"nomadvip": "/nomadvip/payh5, /nomadvip/payh5/redirect, /nomadvip/check_vip, /nomadvip/check_vip_by_email, /nomadvip/order_status, /nomadvip/zpay_order_status, /nomadvip/zpay_notify, /nomadvip/xorpay_notify",
"digital": "/digital/payh5, /digital/payh5/redirect, /digital/zpay_order_status, /digital/zpay_notify, /digital/xorpay_notify",
"cnomadcna": "/cnomadcna/payh5, ...",
"nomadlms": "/nomadlms/payh5, ...",
"legacy": "/payh5, /payh5/redirect, /zpay_notify, /xorpay_notify, /zpay_order_status",
"common": "/submit_meetup_application, /create_user",
"community": "/community/memos, /community/stats, /community/attachments, /community/upload",
},
}

View File

@@ -5,7 +5,13 @@
from .base import PaymentProvider
from .xorpay import XorPayProvider
from .zpay import ZPayProvider
from .factory import get_payment_provider, get_payment_provider_by_name, reset_provider
from .factory import (
get_payment_provider,
get_payment_provider_by_name,
reset_provider,
resolve_provider,
resolve_channel,
)
__all__ = [
"PaymentProvider",
@@ -14,4 +20,6 @@ __all__ = [
"get_payment_provider",
"get_payment_provider_by_name",
"reset_provider",
"resolve_provider",
"resolve_channel",
]

View File

@@ -49,3 +49,31 @@ def reset_provider():
"""重置 Provider用于测试或切换渠道"""
global _provider_instance
_provider_instance = None
def resolve_provider(
device: str | None = None,
req_provider: str | None = None,
user_agent: str | None = None,
) -> PaymentProvider:
"""
根据设备解析支付渠道:
- 微信内(device=wechat 或 UA 含 MicroMessenger) → xorpay
- PC/手机浏览器 → zpay
显式传入的 provider 优先。
"""
req = (req_provider or "").strip().lower()
if req in ("xorpay", "zpay"):
return get_payment_provider_by_name(req)
dev = (device or "pc").lower()
if dev == "wechat":
return get_payment_provider_by_name("xorpay")
ua = (user_agent or "").lower()
if "micromessenger" in ua:
return get_payment_provider_by_name("xorpay")
return get_payment_provider_by_name("zpay")
def resolve_channel(item_channel: str | None) -> str:
"""支付方式固定为微信支付,支付宝暂不提供"""
return "wxpay"

View File

@@ -5,6 +5,8 @@ XorPay 支付渠道实现
import hashlib
from typing import Any, Dict, Optional
import requests
from .base import PaymentProvider
@@ -24,6 +26,10 @@ class XorPayProvider(PaymentProvider):
raw = name + pay_type + price + order_id + notify_url + self.secret
return hashlib.md5(raw.encode("utf-8")).hexdigest().lower()
def _query2_sign(self, order_id: str) -> str:
raw = order_id + self.secret
return hashlib.md5(raw.encode("utf-8")).hexdigest().lower()
def create_order(
self,
*,
@@ -80,3 +86,23 @@ class XorPayProvider(PaymentProvider):
def success_response(self) -> str:
return "ok"
def query_order_status(self, order_id: str, timeout: int = 10) -> Dict[str, Any]:
url = f"https://xorpay.com/api/query2/{self.aid}"
params = {
"order_id": order_id,
"sign": self._query2_sign(order_id),
}
try:
response = requests.get(url, params=params, timeout=timeout)
response.raise_for_status()
data = response.json() or {}
status = str(data.get("status", "")).strip().lower()
return {
"paid": status in ("payed", "success"),
"status": status,
"pay_price": data.get("pay_price") or data.get("price") or "",
"raw": data,
}
except Exception as error:
return {"paid": False, "error": str(error)}

View File

@@ -5,12 +5,27 @@ ZPAY 支付渠道实现
"""
import hashlib
import json
import logging
from typing import Any, Dict, Optional
import requests
from .base import PaymentProvider
logger = logging.getLogger(__name__)
def _is_internal_ip(ip: Optional[str]) -> bool:
"""判断是否为内网 IPZPAY 可能拒收导致 Ubuntu 部署失败"""
if not ip or not ip.strip():
return True
ip = ip.strip()
if ip in ("127.0.0.1", "localhost", "::1"):
return True
if ip.startswith("10.") or ip.startswith("192.168.") or ip.startswith("172.16."):
return True
return False
class ZPayProvider(PaymentProvider):
"""ZPAY 支付(易支付兼容接口)"""
@@ -64,8 +79,10 @@ class ZPayProvider(PaymentProvider):
"money": total_fee_yuan,
"return_url": return_url or notify_url,
}
if client_ip:
if client_ip and not _is_internal_ip(client_ip):
params["clientip"] = client_ip
elif client_ip and _is_internal_ip(client_ip):
logger.warning("ZPAY create_order 跳过内网 client_ip=%s", client_ip)
if extra and extra.get("param"):
params["param"] = str(extra["param"])
params["sign_type"] = "MD5"
@@ -98,6 +115,14 @@ class ZPayProvider(PaymentProvider):
适用于需要二维码或 H5 跳转的场景
"""
channel_type = "wxpay" if pay_type.lower() in ("wx", "wechat", "wxpay") else "alipay"
effective_ip = (client_ip or "").strip() or "127.0.0.1"
# 内网 IP127.0.0.1/10.x/192.168.x可能导致 ZPAY 风控拒单Ubuntu 部署时常见
if _is_internal_ip(effective_ip):
logger.warning(
"ZPAY create_order_api client_ip=%s 为内网地址,已省略 clientip 参数(避免风控拒单)",
effective_ip,
)
effective_ip = "" # 不传 clientip由 ZPAY 从连接获取
params = {
"pid": self.pid,
"type": channel_type,
@@ -105,9 +130,10 @@ class ZPayProvider(PaymentProvider):
"notify_url": notify_url,
"name": name,
"money": total_fee_yuan,
"clientip": client_ip or "127.0.0.1",
"device": device,
}
if effective_ip:
params["clientip"] = effective_ip
if return_url:
params["return_url"] = return_url
if extra and extra.get("param"):
@@ -117,7 +143,15 @@ class ZPayProvider(PaymentProvider):
try:
resp = requests.post(self.MAPI_URL, data=params, timeout=15)
except requests.exceptions.SSLError as e:
logger.error("ZPAY SSL 错误Ubuntu 可能 CA 证书不全): %s", e, exc_info=True)
return {
"status": "error",
"provider": self.name,
"msg": f"连接 zpayz.cn SSL 失败: {e}。Ubuntu 请安装 ca-certificates: apt install ca-certificates",
}
except requests.RequestException as e:
logger.error("ZPAY 连接失败: %s", e, exc_info=True)
return {
"status": "error",
"provider": self.name,
@@ -160,10 +194,18 @@ class ZPayProvider(PaymentProvider):
"O_id": result.get("O_id"),
"trade_no": result.get("trade_no"),
}
err_msg = result.get("msg", resp.text)
logger.error(
"ZPAY create_order_api 失败 status=%s code=%s msg=%s resp_text=%s",
resp.status_code,
code,
err_msg,
resp.text[:500] if resp.text else "",
)
return {
"status": "error",
"provider": self.name,
"msg": result.get("msg", resp.text),
"msg": err_msg,
}
def verify_notify(self, data: Dict[str, Any]) -> bool:

View File

@@ -12,6 +12,8 @@ from .cnomadcna import router as cnomadcna_router
from .nomadlms import router as nomadlms_router
from .legacy import router as legacy_router
from .common import router as common_router
from .meetup import router as meetup_router
from .community import router as community_router
__all__ = [
"nomadvip_router",
@@ -20,4 +22,6 @@ __all__ = [
"nomadlms_router",
"legacy_router",
"common_router",
"meetup_router",
"community_router",
]

View File

@@ -10,11 +10,23 @@ import time
from fastapi import APIRouter, Request, HTTPException
from ..config import BASE_URL
from ..payment import get_payment_provider
from ..payment import (
get_payment_provider,
get_payment_provider_by_name,
resolve_provider,
resolve_channel,
)
from ..payment.zpay import ZPayProvider
from ..services import handle_payment_success, processed_orders
def _safe_log(level: int, message: str, *args, **kwargs) -> None:
try:
logging.log(level, message, *args, **kwargs)
except Exception:
pass
def create_digital_router(prefix: str, project_name: str, site_id: str | None = None) -> APIRouter:
"""
创建 digital/cnomadcna/nomadlms 的支付路由
@@ -31,15 +43,36 @@ def create_digital_router(prefix: str, project_name: str, site_id: str | None =
return f"{base}{path}"
@router.post("/payh5")
async def payh5(item: dict):
"""创建 H5 支付订单"""
print(f"{project_name}/payh5 item:", item)
total_fee_yuan = f"{item['total_fee'] / 100:.2f}"
async def payh5(item: dict, request: Request):
"""创建 H5 支付订单。默认微信支付PC/手机用 zpay微信内用 xorpay支付宝暂不提供"""
req_provider = (item.get("provider") or "").strip().lower() or None
device = (item.get("device") or "pc").lower()
ua = (request.headers.get("user-agent") or "").strip()
provider = resolve_provider(device, req_provider, ua)
channel = resolve_channel(item.get("channel"))
user_id = item.get("user_id", "unknown")
pay_type = item.get("type", "unknown").lower()
pay_type = (item.get("type") or "unknown").lower()
_safe_log(
logging.INFO,
"[payjsapi] %s/payh5 user_id=%s type=%s provider=%s channel=%s",
project_name,
user_id,
pay_type,
provider.name,
channel,
)
try:
total_fee = int(item.get("total_fee", 80))
except (TypeError, ValueError):
total_fee = 80
total_fee_yuan = f"{total_fee / 100:.2f}"
random_suffix = "".join(random.choices(string.hexdigits.lower(), k=8))
timestamp = int(time.time())
order_id = f"{user_id}_{pay_type}_order_{timestamp}_{random_suffix}"
return_url = item.get("return_url")
if return_url and "order_id=" not in return_url and "out_trade_no=" not in return_url:
sep = "&" if "?" in return_url else "?"
return_url = f"{return_url}{sep}order_id={order_id}"
type_map = {
"book": "购买书籍",
"meetup": "数字游民社区报名",
@@ -47,22 +80,30 @@ def create_digital_router(prefix: str, project_name: str, site_id: str | None =
"vip": "VIP会员充值",
}
name = type_map.get(pay_type, "商品支付")
provider = get_payment_provider()
notify_path = "/xorpay_notify" if provider.name == "xorpay" else "/zpay_notify"
notify_url = _notify_path(notify_path)
channel = item.get("channel", "alipay")
result = provider.create_order(
order_id=order_id,
name=name,
total_fee_yuan=total_fee_yuan,
pay_type=channel,
notify_url=notify_url,
return_url=item.get("return_url"),
return_url=return_url,
client_ip=item.get("client_ip"),
)
_safe_log(
logging.INFO,
"[payjsapi] %s/payh5 created order_id=%s provider=%s notify_url=%s return_url=%s",
project_name,
order_id,
provider.name,
notify_url,
return_url or "",
)
if provider.name == "xorpay":
return {
"status": "ok",
"order_id": order_id,
"xorpay_params": result.get("params", {}),
"xorpay_url": result.get("pay_url", ""),
"provider": provider.name,
@@ -79,26 +120,39 @@ def create_digital_router(prefix: str, project_name: str, site_id: str | None =
@router.post("/payh5/redirect")
async def payh5_redirect(item: dict, request: Request):
"""ZPAY mapi 接口,返回 302 或二维码"""
"""ZPAY mapi 接口,返回 302 或二维码。PC/手机用 zpay默认微信支付支付宝暂不提供"""
from fastapi.responses import RedirectResponse, JSONResponse
provider = get_payment_provider()
if provider.name != "zpay":
raise HTTPException(status_code=400, detail="payh5/redirect 仅支持 zpay 渠道")
total_fee = float(item.get("total_fee", 80))
provider = get_payment_provider_by_name("zpay")
channel = resolve_channel(item.get("channel"))
_safe_log(
logging.INFO,
"[payjsapi] %s/payh5/redirect user_id=%s type=%s device=%s channel=%s",
project_name,
item.get("user_id", ""),
item.get("type", ""),
item.get("device", "pc"),
channel,
)
try:
total_fee = float(item.get("total_fee", 80))
except (TypeError, ValueError):
total_fee = 80.0
total_fee_yuan = f"{total_fee / 100:.2f}"
user_id = item.get("user_id", "unknown")
pay_type = item.get("type", "unknown").lower()
channel = item.get("channel", "alipay")
return_url = item.get("return_url")
device = item.get("device", "pc")
random_suffix = "".join(random.choices(string.hexdigits.lower(), k=8))
timestamp = int(time.time())
order_id = f"{user_id}_{pay_type}_order_{timestamp}_{random_suffix}"
if return_url and "order_id=" not in return_url and "out_trade_no=" not in return_url:
sep = "&" if "?" in return_url else "?"
return_url = f"{return_url}{sep}order_id={order_id}"
client_ip = item.get("client_ip") or (request.client.host if request.client else "127.0.0.1")
forwarded = request.headers.get("x-forwarded-for")
if forwarded:
client_ip = forwarded.split(",")[0].strip()
random_suffix = "".join(random.choices(string.hexdigits.lower(), k=8))
timestamp = int(time.time())
order_id = f"{user_id}_{pay_type}_order_{timestamp}_{random_suffix}"
type_map = {
"book": "购买书籍",
"meetup": "数字游民社区报名",
@@ -107,8 +161,6 @@ def create_digital_router(prefix: str, project_name: str, site_id: str | None =
}
name = type_map.get(pay_type, "商品支付")
notify_url = _notify_path("/zpay_notify")
if not isinstance(provider, ZPayProvider):
raise HTTPException(status_code=400, detail="payh5/redirect 仅支持 zpay")
result = provider.create_order_api(
order_id=order_id,
name=name,
@@ -121,6 +173,14 @@ def create_digital_router(prefix: str, project_name: str, site_id: str | None =
)
if result.get("status") != "ok":
raise HTTPException(status_code=500, detail=result.get("msg", "ZPAY mapi 创建订单失败"))
_safe_log(
logging.INFO,
"[payjsapi] %s/payh5/redirect created order_id=%s device=%s channel=%s",
project_name,
order_id,
device,
channel,
)
payurl = result.get("payurl")
payurl2 = result.get("payurl2")
img = result.get("img")
@@ -146,23 +206,66 @@ def create_digital_router(prefix: str, project_name: str, site_id: str | None =
raise HTTPException(status_code=400, detail="仅 zpay 渠道支持")
result = provider.query_order_status(out_trade_no, timeout=15)
if result.get("error"):
logging.warning(f"{project_name} ZPAY 订单查询失败: {result.get('error')}")
_safe_log(logging.WARNING, "%s ZPAY 订单查询失败: %s", project_name, result.get("error"))
return result
@router.get("/order_status")
async def order_status(order_id: str = "", out_trade_no: str = ""):
order_id = (order_id or out_trade_no or "").strip()
if not order_id:
return {"paid": False, "error": "missing order_id"}
try:
from ..services.pb import get_pb_client
pb = get_pb_client()
records = pb.collection("payments").get_list(
1, 1, {"filter": f'order_id = "{order_id}"'}
)
if records.items:
return {"paid": True, "source": "pocketbase"}
except Exception as error:
_safe_log(logging.WARNING, "%s order_status PocketBase failed: %s", project_name, error)
try:
xorpay = get_payment_provider_by_name("xorpay")
xorpay_result = xorpay.query_order_status(order_id, timeout=8)
if xorpay_result.get("paid"):
return {
"paid": True,
"source": "xorpay",
"status": xorpay_result.get("status", ""),
}
except Exception as error:
_safe_log(logging.WARNING, "%s order_status XorPay failed: %s", project_name, error)
try:
zpay = get_payment_provider_by_name("zpay")
if isinstance(zpay, ZPayProvider):
zpay_result = zpay.query_order_status(order_id, timeout=8)
if zpay_result.get("paid"):
return {
"paid": True,
"source": "zpay",
"status": zpay_result.get("status", ""),
}
except Exception as error:
_safe_log(logging.WARNING, "%s order_status ZPAY failed: %s", project_name, error)
return {"paid": False}
@router.post("/xorpay_notify")
async def xorpay_notify(request: Request):
"""XorPay 异步通知"""
form_data = await request.form()
data = {k: v for k, v in form_data.items()}
logging.info(f"{project_name} XorPay 异步通知: {data}")
_safe_log(logging.INFO, "%s XorPay 异步通知: %s", project_name, data)
order_id = data.get("order_id", "")
if order_id in processed_orders:
return "ok"
provider = get_payment_provider()
if provider.name != "xorpay":
raise HTTPException(status_code=400, detail="xorpay_notify only for xorpay")
provider = get_payment_provider_by_name("xorpay")
if not provider.verify_notify(data):
logging.error(f"{project_name} XorPay 验签失败: {data}")
_safe_log(logging.ERROR, "%s XorPay 验签失败: %s", project_name, data)
raise HTTPException(status_code=400, detail="sign error")
parsed = provider.parse_notify(data)
if parsed.get("trade_status") != "success":
@@ -185,15 +288,13 @@ def create_digital_router(prefix: str, project_name: str, site_id: str | None =
else:
form_data = await request.form()
data = {k: v for k, v in form_data.items()}
logging.info(f"{project_name} ZPAY 异步通知: {data}")
_safe_log(logging.INFO, "%s ZPAY 异步通知: %s", project_name, data)
order_id = data.get("out_trade_no", "")
if order_id in processed_orders:
return "success"
provider = get_payment_provider()
if provider.name != "zpay":
raise HTTPException(status_code=400, detail="zpay_notify only for zpay")
provider = get_payment_provider_by_name("zpay")
if not provider.verify_notify(data):
logging.error(f"{project_name} ZPAY 验签失败: {data}")
_safe_log(logging.ERROR, "%s ZPAY 验签失败: %s", project_name, data)
raise HTTPException(status_code=400, detail="sign error")
parsed = provider.parse_notify(data)
if parsed.get("trade_status") != "success":

View File

@@ -15,7 +15,7 @@ router = APIRouter(tags=["common"])
@router.post("/submit_meetup_application")
async def submit_meetup_application(item: dict):
"""数字游民社区报名表单"""
print("submit_meetup_application item:", item)
logging.info("submit_meetup_application item: %s", item)
user_id = item.get("user_id")
if not user_id:
raise HTTPException(status_code=400, detail="missing user_id")
@@ -41,7 +41,7 @@ async def submit_meetup_application(item: dict):
logging.info(f"meetup 申请表单提交成功(待支付) - user_id: {user_id}, record_id: {record.id}")
return {"status": "ok", "record_id": record.id}
except Exception as e:
print(f"meetup 申请提交失败: {e}")
logging.error("meetup 申请提交失败: %s", e)
logging.error(f"meetup 申请提交失败 - user_id: {user_id}, error: {e}")
raise HTTPException(status_code=500, detail="submit failed")
@@ -59,7 +59,7 @@ async def create_user(item: CreateUserRequest):
return {"error": "Memos 未配置"}
timenew = str(int(time.time()))
username = item.username or f"user{timenew}"
password = item.password or "123456"
password = item.password or "12345678"
result = sdk.create_user_by_username(username=username, password=password)
print("memos_result-", result)
logging.info("memos_result- %s", result)
return result

359
app/routers/community.py Normal file
View File

@@ -0,0 +1,359 @@
"""
Private community API.
"""
from __future__ import annotations
from datetime import datetime
import logging
from typing import Any
from fastapi import APIRouter, File, Form, HTTPException, UploadFile
from fastapi.responses import FileResponse
import requests
from ..config import BASE_URL, PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD, PB_URL
from ..db import community_sqlite as sqlite_db
from ..services import community as community_svc
ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"}
MAX_IMAGE_SIZE = 10 * 1024 * 1024
SITE_ID = "vip"
router = APIRouter(prefix="/community", tags=["community"])
logger = logging.getLogger(__name__)
def _get_pb_admin_token() -> str | None:
if not PB_ADMIN_EMAIL or not PB_ADMIN_PASSWORD:
return None
base = (PB_URL or "").rstrip("/")
for path in (
"/api/collections/_superusers/auth-with-password",
"/api/admins/auth-with-password",
):
try:
response = requests.post(
f"{base}{path}",
json={"identity": PB_ADMIN_EMAIL, "password": PB_ADMIN_PASSWORD},
timeout=10,
)
if response.status_code == 200:
return response.json().get("token")
except Exception:
continue
return None
def _pb_escape(value: str) -> str:
return str(value).replace("\\", "\\\\").replace('"', '\\"')
def _uid_has_vip(uid: str) -> bool:
uid = (uid or "").strip()
if not uid:
return False
token = _get_pb_admin_token()
if not token:
return False
base = (PB_URL or "").rstrip("/")
headers = {"Authorization": f"Bearer {token}"}
safe_uid = _pb_escape(uid)
try:
response = requests.get(
f"{base}/api/collections/site_vip/records",
params={"filter": f'user_id = "{safe_uid}" && site_id = "{SITE_ID}"', "perPage": 1},
headers=headers,
timeout=10,
)
if response.status_code == 200:
items = response.json().get("items") or []
if items:
record = items[0]
expires = record.get("expires_at") or ""
if not expires:
return True
try:
expires_at = datetime.fromisoformat(expires.replace("Z", "+00:00"))
now = datetime.now(expires_at.tzinfo) if expires_at.tzinfo else datetime.now()
return expires_at.timestamp() >= now.timestamp()
except Exception:
return True
except Exception:
pass
try:
response = requests.get(
f"{base}/api/collections/payments/records",
params={"filter": f'user_id = "{safe_uid}" && type = "vip"', "sort": "-created", "perPage": 1},
headers=headers,
timeout=10,
)
if response.status_code == 200 and (response.json().get("items") or []):
return True
except Exception:
pass
return False
def _count_vip_members() -> int | None:
token = _get_pb_admin_token()
if not token:
return None
base = (PB_URL or "").rstrip("/")
headers = {"Authorization": f"Bearer {token}"}
try:
response = requests.get(
f"{base}/api/collections/site_vip/records",
params={"filter": f'site_id = "{SITE_ID}"', "perPage": 1},
headers=headers,
timeout=10,
)
if response.status_code == 200:
payload = response.json()
return int(payload.get("totalItems") or len(payload.get("items") or []))
except Exception:
return None
return None
def _require_vip_user(user_id: str, detail: str = "Only VIP members can access the private community.") -> str:
normalized = (user_id or "").strip()
if not normalized:
raise HTTPException(status_code=401, detail="Missing user_id")
if not _uid_has_vip(normalized):
raise HTTPException(status_code=403, detail=detail)
return normalized
def _extract_payload(item: dict[str, Any]) -> dict[str, Any]:
nested = item.get("memo") or item.get("comment") or item.get("reaction")
if isinstance(nested, dict):
return nested
return item
@router.get("/health")
async def community_health():
return {"ok": True, "service": "community"}
@router.get("/files/{filename:path}")
async def serve_upload_file(filename: str):
path = sqlite_db.get_upload_path(filename)
if path is None:
raise HTTPException(status_code=404, detail="File not found.")
return FileResponse(path, filename=filename)
@router.get("/stats")
async def get_stats(user_id: str = ""):
normalized = _require_vip_user(user_id)
member_count = _count_vip_members()
return {"ok": True, "user_id": normalized, **community_svc.get_stats(member_count=member_count)}
@router.get("/attachments")
async def list_attachments(user_id: str = "", page: int = 1, per_page: int = 24):
_require_vip_user(user_id)
per_page = min(max(1, per_page), 60)
page = max(1, page)
return {"ok": True, **community_svc.list_uploads(page=page, per_page=per_page)}
@router.get("/memos")
async def list_memos(
user_id: str = "",
page: int = 1,
per_page: int = 50,
page_size: int = 0,
page_token: str = "",
state: str = "NORMAL",
order_by: str = "-pinned,-created",
memo_format: str = "1",
query: str = "",
):
normalized = _require_vip_user(user_id)
per_page = per_page or page_size or 50
per_page = min(max(1, per_page), 100)
page = max(1, int(page_token) if page_token and page_token.isdigit() else page)
try:
result = community_svc.list_memos(
page=page,
per_page=per_page,
state=(state or "NORMAL").upper(),
order_by=order_by or "-pinned,-created",
memo_format=memo_format != "0",
query=(query or "").strip(),
viewer_id=normalized,
)
return {"ok": True, **result}
except RuntimeError as exc:
logger.error("community list_memos config error: %s", exc)
raise HTTPException(status_code=503, detail=str(exc))
except Exception as exc:
logger.error("community list_memos error: %s", exc)
raise HTTPException(status_code=500, detail=str(exc))
@router.get("/memos/{memo_id}")
async def get_memo(memo_id: str, user_id: str = ""):
normalized = _require_vip_user(user_id)
result = community_svc.get_memo(memo_id, viewer_id=normalized, include_comments=True)
if result is None:
raise HTTPException(status_code=404, detail="Memo not found.")
return {"ok": True, "memo": result}
@router.post("/memos")
async def create_memo(item: dict[str, Any]):
payload = _extract_payload(item)
user_id = _require_vip_user(payload.get("user_id") or item.get("user_id"), "Only VIP members can publish posts.")
content = str(payload.get("content") or item.get("content") or "").strip()
pinned = bool(payload.get("pinned") if payload.get("pinned") is not None else item.get("pinned"))
visibility = str(payload.get("visibility") or item.get("visibility") or "PRIVATE").upper()
state = str(payload.get("state") or item.get("state") or "NORMAL").upper()
try:
memo_record = community_svc.create_memo(
user_id=user_id,
content=content,
pinned=pinned,
visibility=visibility,
state=state,
)
return {"ok": True, "memo": memo_record}
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
except RuntimeError as exc:
logger.error("community create_memo config error: %s", exc)
raise HTTPException(status_code=503, detail=str(exc))
except Exception as exc:
logger.error("community create_memo error: %s", exc)
raise HTTPException(status_code=500, detail=str(exc))
@router.patch("/memos/{memo_id}")
async def update_memo(memo_id: str, item: dict[str, Any]):
payload = _extract_payload(item)
user_id = _require_vip_user(payload.get("user_id") or item.get("user_id"), "Only VIP members can update posts.")
pinned = payload.get("pinned") if payload.get("pinned") is not None else item.get("pinned")
content = payload.get("content") if payload.get("content") is not None else item.get("content")
visibility = payload.get("visibility") or item.get("visibility")
state = payload.get("state") or item.get("state")
try:
result = community_svc.update_memo(
memo_id.replace("memos/", ""),
user_id,
pinned=bool(pinned) if pinned is not None else None,
content=str(content) if content is not None else None,
visibility=str(visibility).upper() if visibility else None,
state=str(state).upper() if state else None,
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
if result is None:
raise HTTPException(status_code=404, detail="Memo not found or permission denied.")
return {"ok": True, "memo": result}
@router.delete("/memos/{memo_id}")
async def delete_memo(memo_id: str, user_id: str = ""):
normalized = _require_vip_user(user_id, "Only VIP members can delete posts.")
ok = community_svc.delete_memo(memo_id, normalized)
if not ok:
raise HTTPException(status_code=404, detail="Memo not found or permission denied.")
return {"ok": True}
@router.get("/memos/{memo_id}/comments")
async def get_comments(memo_id: str, user_id: str = ""):
_require_vip_user(user_id)
return {"ok": True, "comments": community_svc.list_comments(memo_id)}
@router.post("/memos/{memo_id}/comments")
async def create_comment(memo_id: str, item: dict[str, Any]):
payload = _extract_payload(item)
user_id = _require_vip_user(
payload.get("user_id") or item.get("user_id"),
"Only VIP members can reply in the private community.",
)
content = str(payload.get("content") or item.get("content") or "").strip()
try:
comment = community_svc.create_comment(memo_id, user_id, content)
return {"ok": True, "comment": comment}
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
except Exception as exc:
logger.error("community create_comment error: %s", exc)
raise HTTPException(status_code=500, detail=str(exc))
@router.delete("/comments/{comment_id}")
async def delete_comment(comment_id: str, user_id: str = ""):
normalized = _require_vip_user(user_id, "Only VIP members can manage comments.")
ok = community_svc.delete_comment(comment_id, normalized)
if not ok:
raise HTTPException(status_code=404, detail="Comment not found or permission denied.")
return {"ok": True}
@router.post("/memos/{memo_id}/reactions")
async def toggle_reaction(memo_id: str, item: dict[str, Any]):
payload = _extract_payload(item)
user_id = _require_vip_user(
payload.get("user_id") or item.get("user_id"),
"Only VIP members can react in the private community.",
)
reaction_type = str(payload.get("reaction_type") or item.get("reaction_type") or sqlite_db.UPVOTE_REACTION)
try:
reaction = community_svc.toggle_reaction(memo_id, user_id, reaction_type)
return {"ok": True, "reaction": reaction}
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
except Exception as exc:
logger.error("community toggle_reaction error: %s", exc)
raise HTTPException(status_code=500, detail=str(exc))
@router.post("/upload")
async def upload_image(user_id: str = Form(""), file: UploadFile = File(...)):
normalized = _require_vip_user(user_id, "Only VIP members can upload images.")
if not file.content_type or file.content_type.lower() not in ALLOWED_IMAGE_TYPES:
raise HTTPException(status_code=400, detail="Only jpeg, png, gif, and webp images are supported.")
data = await file.read()
if len(data) > MAX_IMAGE_SIZE:
raise HTTPException(status_code=400, detail="Image size must be 10MB or smaller.")
try:
base = (BASE_URL or "http://localhost:8000").rstrip("/")
result = community_svc.upload_image(
data=data,
filename=file.filename or "image",
content_type=file.content_type or "image/jpeg",
user_id=normalized,
base_url=base,
)
return {"ok": True, **result}
except RuntimeError as exc:
raise HTTPException(status_code=503, detail=str(exc))
except Exception as exc:
logger.error("community upload error: %s", exc)
raise HTTPException(status_code=500, detail=str(exc))

328
app/routers/meetup.py Normal file
View File

@@ -0,0 +1,328 @@
"""
meetup 加入流程check-user、ensure-user、complete-order
供 cnomadcna 等前端调用,当 /api/* 代理到 payjsapi 时使用
"""
import logging
import time
import requests
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from ..config import PB_URL, PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD
router = APIRouter(prefix="/api/meetup", tags=["meetup"])
DEFAULT_PASSWORD = "12345678" # PocketBase 默认要求至少 8 位
# Admin token 缓存,避免每次请求都向 PocketBase 认证PB 远程时易超时)
_ADMIN_TOKEN_CACHE: tuple[str, float] | None = None
_TOKEN_CACHE_TTL = 300 # 5 分钟
def _get_admin_token() -> tuple[str | None, str | None]:
"""返回 (token, error_detail)。token 为 None 时 error_detail 为失败原因。带 5 分钟缓存。"""
global _ADMIN_TOKEN_CACHE
now = time.time()
if _ADMIN_TOKEN_CACHE and _ADMIN_TOKEN_CACHE[1] > now:
return _ADMIN_TOKEN_CACHE[0], None
if not PB_ADMIN_EMAIL or not PB_ADMIN_PASSWORD:
return None, "PB_ADMIN_EMAIL 或 PB_ADMIN_PASSWORD 未配置"
base = (PB_URL or "").rstrip("/")
# PocketBase v0.23+ 使用 _superusers旧版用 admins
for path in ["/api/collections/_superusers/auth-with-password", "/api/admins/auth-with-password"]:
try:
r = requests.post(
f"{base}{path}",
json={"identity": PB_ADMIN_EMAIL, "password": PB_ADMIN_PASSWORD},
timeout=4,
)
if r.status_code == 200:
data = r.json()
token = data.get("token")
if token:
_ADMIN_TOKEN_CACHE = (token, now + _TOKEN_CACHE_TTL)
return token, None
if r.status_code == 404:
continue # 尝试下一个路径
try:
msg = r.json().get("message", r.text[:100])
except Exception:
msg = r.text[:100]
logging.warning(f"PocketBase admin auth failed: status={r.status_code}, msg={msg}")
return None, f"PocketBase 管理员认证失败: {msg}"
except requests.exceptions.ConnectionError as e:
logging.error(f"PocketBase 连接失败: {e}")
return None, f"无法连接 {PB_URL},请检查网络或 PB_URL"
except Exception as e:
logging.warning(f"PocketBase auth try {path}: {e}")
return None, "PocketBase 管理员认证失败: 请确认 PB_URL 及管理员账号密码正确"
class CheckUserRequest(BaseModel):
email: str
class EnsureUserRequest(BaseModel):
email: str
password: str | None = None
def _check_site_vip(user_id: str, site_id: str = "meetup", token: str | None = None) -> bool:
"""检查用户是否有有效 site_vip未过期。可传入 token 避免重复认证。"""
if not token:
token, _ = _get_admin_token()
if not token:
return False
try:
from datetime import datetime
r = requests.get(
f"{PB_URL}/api/collections/site_vip/records",
params={
"filter": f'user_id = "{user_id}" && site_id = "{site_id}"',
"perPage": 1,
"sort": "-expires_at",
},
headers={"Authorization": f"Bearer {token}"},
timeout=4,
)
if r.status_code != 200:
return False
data = r.json()
items = data.get("items") or []
if not items:
return False
rec = items[0]
exp = rec.get("expires_at")
if not exp:
return True
try:
from datetime import datetime as _dt, timezone
exp_dt = _dt.fromisoformat(exp.replace("Z", "+00:00"))
now = _dt.now(timezone.utc)
if exp_dt.tzinfo is None:
exp_dt = exp_dt.replace(tzinfo=timezone.utc)
return exp_dt > now
except Exception:
return True
except Exception:
return False
@router.post("/check-user")
async def check_user(item: CheckUserRequest):
"""检查邮箱是否已在 users 表存在,以及 VIP 状态(新/老用户判断)"""
email = (item.email or "").strip()
if not email:
raise HTTPException(status_code=400, detail="请输入邮箱")
token, err = _get_admin_token()
if not token:
raise HTTPException(status_code=500, detail=f"服务配置错误:{err}")
try:
filter_val = f'email="{email}"'
r = requests.get(
f"{PB_URL}/api/collections/users/records",
params={"filter": filter_val, "perPage": 1},
headers={"Authorization": f"Bearer {token}"},
timeout=4,
)
if r.status_code != 200:
raise HTTPException(status_code=500, detail="查询失败")
data = r.json()
items = data.get("items") or []
exists = len(items) > 0
user_id = items[0].get("id") if items else None
vip = _check_site_vip(user_id, "meetup", token) if user_id else False
return {"ok": True, "exists": exists, "vip": vip, "user_id": user_id}
except HTTPException:
raise
except Exception as e:
logging.error(f"check-user error: {e}")
raise HTTPException(status_code=500, detail="操作失败")
@router.post("/ensure-user")
async def ensure_user(item: EnsureUserRequest):
"""确保用户存在:老用户验证密码,新用户用默认密码 12345678 创建"""
email = (item.email or "").strip()
if not email:
raise HTTPException(status_code=400, detail="请输入邮箱")
password = (item.password or "").strip() or DEFAULT_PASSWORD
try:
# 尝试用密码登录
login_r = requests.post(
f"{PB_URL}/api/collections/users/auth-with-password",
json={"identity": email, "password": password},
timeout=10,
)
if login_r.status_code == 200:
data = login_r.json()
return {
"ok": True,
"user_id": data.get("record", {}).get("id"),
"token": data.get("token"),
"record": data.get("record"),
"is_new": False,
}
try:
err_data = login_r.json()
except Exception:
err_data = {}
is_not_found = login_r.status_code == 400 and (
"Invalid" in str(err_data.get("message", ""))
or "identity" in str(err_data.get("message", "")).lower()
)
if not is_not_found and item.password:
raise HTTPException(status_code=400, detail="密码错误")
# 新用户:创建
token, err = _get_admin_token()
if not token:
raise HTTPException(status_code=500, detail=f"服务配置错误:{err}")
create_r = requests.post(
f"{PB_URL}/api/collections/users/records",
json={
"email": email,
"password": DEFAULT_PASSWORD,
"passwordConfirm": DEFAULT_PASSWORD,
},
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {token}",
},
timeout=10,
)
if create_r.status_code != 200:
try:
create_err = create_r.json()
except Exception:
create_err = {}
data = create_err.get("data", {})
if (
create_r.status_code == 400
and "already" in str(data.get("email", {}).get("message", "")).lower()
):
raise HTTPException(status_code=400, detail="该邮箱已注册,请输入密码登录")
msg = create_err.get("message", "创建账号失败")
logging.warning(f"ensure-user create failed: status={create_r.status_code}, msg={msg}, data={data}")
raise HTTPException(status_code=500, detail=msg)
# 登录新用户
final_r = requests.post(
f"{PB_URL}/api/collections/users/auth-with-password",
json={"identity": email, "password": DEFAULT_PASSWORD},
timeout=10,
)
if final_r.status_code != 200:
raise HTTPException(status_code=500, detail="登录失败")
auth_data = final_r.json()
return {
"ok": True,
"user_id": auth_data.get("record", {}).get("id"),
"token": auth_data.get("token"),
"record": auth_data.get("record"),
"is_new": True,
}
except HTTPException:
raise
except Exception as e:
logging.error(f"ensure-user error: {e}")
raise HTTPException(status_code=500, detail="操作失败")
def _decode_pending_email(user_id: str) -> str | None:
if not user_id.startswith("pending_"):
return None
try:
return bytes.fromhex(user_id[8:]).decode("utf-8")
except Exception:
return None
def _query_zpay_paid(order_id: str) -> bool:
"""查询 ZPAY 订单是否已支付"""
try:
from ..payment import get_payment_provider
from ..payment.zpay import ZPayProvider
provider = get_payment_provider()
if not isinstance(provider, ZPayProvider):
return False
result = provider.query_order_status(order_id, timeout=10)
return bool(result.get("paid"))
except Exception as e:
logging.warning(f"complete-order zpay query failed: {e}")
return False
@router.post("/complete-order")
async def complete_order(item: dict):
"""
支付成功后完成订单:验证订单、为新用户同步 session。
支持 pending_ 订单新用户和已存在用户订单user_id 为 PB id
异步通知可能晚于用户回跳,故在 site_vip 未找到时轮询重试ZPAY 已确认支付时)。
"""
order_id = (item.get("order_id") or "").strip()
if not order_id:
raise HTTPException(status_code=400, detail="缺少 order_id")
base = (PB_URL or "").rstrip("/")
token, err = _get_admin_token()
if not token:
raise HTTPException(status_code=500, detail=f"服务配置错误:{err}")
from ..services.payment import parse_order_id
user_id, pay_type = parse_order_id(order_id)
if not user_id or pay_type != "meetup":
raise HTTPException(status_code=400, detail="订单格式无效")
# 验证订单site_vip 中需存在该 order_id异步通知可能晚于用户回跳ZPAY 已支付时轮询重试)
max_retries = 4
for attempt in range(max_retries):
r = requests.get(
f"{base}/api/collections/site_vip/records",
params={"filter": f'order_id = "{order_id}"', "perPage": 1},
headers={"Authorization": f"Bearer {token}"},
timeout=10,
)
total = (r.json().get("totalItems") or 0) if r.status_code == 200 else 0
if total > 0:
break
if attempt < max_retries - 1 and _query_zpay_paid(order_id):
time.sleep(2)
continue
resp_preview = r.text[:200] if r.ok else f"status={r.status_code}"
logging.warning(f"complete-order 未找到订单: order_id={order_id}, attempt={attempt + 1}, pb_resp={resp_preview}")
raise HTTPException(status_code=400, detail="订单不存在或未支付成功")
# pending_ 新用户:用邮箱+默认密码登录返回 token
if user_id.startswith("pending_"):
email = _decode_pending_email(user_id)
if not email:
raise HTTPException(status_code=400, detail="订单格式无效")
login_r = requests.post(
f"{base}/api/collections/users/auth-with-password",
json={"identity": email, "password": DEFAULT_PASSWORD},
timeout=10,
)
if login_r.status_code != 200:
raise HTTPException(status_code=500, detail="登录失败,请稍后重试")
auth_data = login_r.json()
return {
"ok": True,
"token": auth_data.get("token"),
"record": auth_data.get("record"),
"is_new": True,
}
# 已存在用户PB user_id订单已验证返回 ok前端已有 session 则无需 token
return {"ok": True, "token": None, "record": None, "is_new": False}

View File

@@ -1,14 +1,21 @@
"""nomadvip 项目支付接口:/nomadvip/*,支持 xorpay/zpayPAYMENT_PROVIDER"""
"""nomadvip 项目支付接口:/nomadvip/*,支持 xorpay / zpay"""
import logging
import random
import string
import time
from fastapi import APIRouter, Request, HTTPException
from fastapi.responses import RedirectResponse, JSONResponse
import requests
from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import JSONResponse, RedirectResponse
from ..config import BASE_URL
from ..payment import get_payment_provider, get_payment_provider_by_name
from ..config import BASE_URL, PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD, PB_URL
from ..payment import (
get_payment_provider,
get_payment_provider_by_name,
resolve_provider,
resolve_channel,
)
from ..payment.zpay import ZPayProvider
from ..services import handle_payment_success, processed_orders
@@ -16,15 +23,27 @@ router = APIRouter(prefix="/nomadvip", tags=["nomadvip"])
def _get_notify_path(provider_name: str) -> str:
return f"{BASE_URL.rstrip('/')}/nomadvip/xorpay_notify" if provider_name == "xorpay" else f"{BASE_URL.rstrip('/')}/nomadvip/zpay_notify"
if provider_name == "xorpay":
return f"{BASE_URL.rstrip('/')}/nomadvip/xorpay_notify"
return f"{BASE_URL.rstrip('/')}/nomadvip/zpay_notify"
@router.post("/payh5")
async def nomadvip_payh5(item: dict):
"""nomadvip 专用:创建支付订单,支持 per-request provider 覆盖(微信用 xorpayPC/H5 用 zpay"""
print("nomadvip/payh5 item:", item)
req_provider = (item.get("provider") or "").strip().lower()
provider = get_payment_provider_by_name(req_provider) if req_provider in ("xorpay", "zpay") else get_payment_provider()
async def nomadvip_payh5(item: dict, request: Request):
"""创建支付订单。微信用 xorpayPC/手机用 zpay,默认微信支付,支付宝暂不提供"""
req_provider = (item.get("provider") or "").strip().lower() or None
device = (item.get("device") or "pc").lower()
ua = (request.headers.get("user-agent") or "").strip()
provider = resolve_provider(device, req_provider, ua)
channel = resolve_channel(item.get("channel"))
logging.info(
"[payjsapi] payh5 user_id=%s type=%s provider=%s channel=%s",
item.get("user_id", ""),
item.get("type", ""),
provider.name,
channel,
)
total_fee = int(item.get("total_fee", 880))
total_fee_yuan = f"{total_fee / 100:.2f}"
user_id = item.get("user_id", "unknown")
@@ -32,6 +51,7 @@ async def nomadvip_payh5(item: dict):
random_suffix = "".join(random.choices(string.hexdigits.lower(), k=8))
timestamp = int(time.time())
order_id = f"{user_id}_{pay_type}_order_{timestamp}_{random_suffix}"
type_map = {
"book": "购买书籍",
"meetup": "数字游民社区报名",
@@ -40,7 +60,6 @@ async def nomadvip_payh5(item: dict):
}
name = type_map.get(pay_type, "VIP会员充值")
notify_url = _get_notify_path(provider.name)
channel = item.get("channel", "alipay")
result = provider.create_order(
order_id=order_id,
name=name,
@@ -50,15 +69,27 @@ async def nomadvip_payh5(item: dict):
return_url=item.get("return_url"),
client_ip=item.get("client_ip"),
)
logging.info(
"[payjsapi] payh5 created order_id=%s provider=%s notify_url=%s return_url=%s",
order_id,
provider.name,
notify_url,
item.get("return_url") or "",
)
if provider.name == "xorpay":
return {
"status": "ok",
"order_id": order_id,
"xorpay_params": result.get("params", {}),
"xorpay_url": result.get("pay_url", ""),
"provider": provider.name,
}
params = result.get("params", {})
params_str = {k: str(v) for k, v in params.items() if v is not None and str(v).strip() != ""}
params_str = {
k: str(v) for k, v in params.items() if v is not None and str(v).strip() != ""
}
return {
"status": "ok",
"params": params_str,
@@ -70,35 +101,49 @@ async def nomadvip_payh5(item: dict):
@router.post("/payh5/redirect")
async def nomadvip_payh5_redirect(item: dict, request: Request):
"""nomadvip 专用ZPAY mapi 接口xorpay 无需 redirect直接 payh5 即可)"""
req_provider = (item.get("provider") or "").strip().lower()
provider = get_payment_provider_by_name("zpay") if req_provider == "zpay" else get_payment_provider()
if not isinstance(provider, ZPayProvider):
raise HTTPException(status_code=400, detail="payh5/redirect 仅支持 zpayxorpay 请直接使用 payh5")
logging.info(f"nomadvip payh5/redirect 请求: item={item}")
"""ZPAY mapi 接口PC/手机用 zpay默认微信支付支付宝暂不提供"""
provider = get_payment_provider_by_name("zpay")
channel = resolve_channel(item.get("channel"))
logging.info("nomadvip payh5/redirect request: item=%s", item)
try:
total_fee = float(item.get("total_fee", 880))
except (TypeError, ValueError):
logging.warning(f"nomadvip payh5/redirect 无效 total_fee: {item.get('total_fee')}")
logging.warning(
"nomadvip payh5/redirect invalid total_fee=%s", item.get("total_fee")
)
total_fee = 880.0
total_fee_yuan = f"{total_fee / 100:.2f}"
user_id = item.get("user_id", "unknown")
pay_type = (item.get("type") or "vip").lower()
channel = item.get("channel", "alipay")
return_url = item.get("return_url")
device = item.get("device", "pc")
client_ip = item.get("client_ip") or (request.client.host if request.client else "127.0.0.1")
client_ip = item.get("client_ip") or (
request.client.host if request.client else "127.0.0.1"
)
forwarded = request.headers.get("x-forwarded-for")
if forwarded:
client_ip = forwarded.split(",")[0].strip()
random_suffix = "".join(random.choices(string.hexdigits.lower(), k=8))
timestamp = int(time.time())
order_id = f"{user_id}_{pay_type}_order_{timestamp}_{random_suffix}"
type_map = {"book": "购买书籍", "meetup": "数字游民社区报名", "video": "视频解锁", "vip": "VIP会员充值"}
# 追加 order_id供 paid 页在 cookie 丢失时恢复 user_id。
if return_url and "order_id=" not in return_url and "out_trade_no=" not in return_url:
sep = "&" if "?" in return_url else "?"
return_url = f"{return_url}{sep}order_id={order_id}"
type_map = {
"book": "购买书籍",
"meetup": "数字游民社区报名",
"video": "视频解锁",
"vip": "VIP会员充值",
}
name = type_map.get(pay_type, "VIP会员充值")
notify_url = f"{BASE_URL.rstrip('/')}/nomadvip/zpay_notify"
if not isinstance(provider, ZPayProvider):
raise HTTPException(status_code=400, detail="payh5/redirect 仅支持 zpay")
result = provider.create_order_api(
order_id=order_id,
name=name,
@@ -111,82 +156,684 @@ async def nomadvip_payh5_redirect(item: dict, request: Request):
)
if result.get("status") != "ok":
err_msg = result.get("msg", "ZPAY mapi 创建订单失败")
logging.error(f"nomadvip payh5/redirect ZPAY 创建订单失败: {err_msg}, result={result}")
logging.error(
"nomadvip payh5/redirect ZPAY create failed: %s result=%s",
err_msg,
result,
)
raise HTTPException(status_code=500, detail=err_msg)
payurl = result.get("payurl")
payurl2 = result.get("payurl2")
img = result.get("img")
if device == "pc" and img:
return JSONResponse(status_code=200, content={
"status": "ok", "qrcode_mode": True, "order_id": order_id,
"img": img, "payurl": payurl, "return_url": return_url, "channel": channel,
})
return JSONResponse(
status_code=200,
content={
"status": "ok",
"qrcode_mode": True,
"order_id": order_id,
"img": img,
"payurl": payurl,
"return_url": return_url,
"channel": channel,
},
)
if device == "wechat" and channel.lower() in ("wxpay", "wx", "wechat"):
return JSONResponse(status_code=200, content={"use_form": True, "order_id": order_id})
return JSONResponse(
status_code=200, content={"use_form": True, "order_id": order_id}
)
headers = {"X-Pay-Order-Id": order_id}
if device == "h5" and channel.lower() in ("wxpay", "wx", "wechat") and payurl2:
return RedirectResponse(url=payurl2, status_code=302, headers=headers)
if payurl:
return RedirectResponse(url=payurl, status_code=302, headers=headers)
logging.error(f"nomadvip payh5/redirect ZPAY 未返回支付链接: device={device}, channel={channel}, result keys={list(result.keys())}")
logging.error(
"nomadvip payh5/redirect ZPAY missing pay url device=%s channel=%s keys=%s",
device,
channel,
list(result.keys()),
)
raise HTTPException(status_code=500, detail="ZPAY 未返回支付链接")
@router.get("/zpay_order_status")
async def nomadvip_zpay_order_status(out_trade_no: str):
"""nomadvip 专用:查询 ZPAY 订单支付状态xorpay 无此接口)"""
"""查询 ZPAY 订单支付状态"""
provider = get_payment_provider()
if not isinstance(provider, ZPayProvider):
raise HTTPException(status_code=400, detail="zpay_order_status 仅支持 zpay")
result = provider.query_order_status(out_trade_no, timeout=15)
if result.get("error"):
logging.warning(f"nomadvip ZPAY 订单查询失败: {result.get('error')}")
logging.warning(
"nomadvip ZPAY order status failed: %s", result.get("error")
)
return result
@router.get("/order_status")
async def nomadvip_order_status(order_id: str):
"""nomadvip 通用查询订单支付状态。zpay 先查 ZPAY 实时xorpay 或失败时查 PocketBase"""
zpay = get_payment_provider_by_name("zpay")
if isinstance(zpay, ZPayProvider):
result = zpay.query_order_status(order_id, timeout=15)
if not result.get("error") and result.get("paid"):
return result
# zpay 查不到或 xorpay查 PocketBasenotify 后写入)
def _get_linked_email_for_user(pb, user_id: str) -> str | None:
"""按 user_id 查询 vip_email_link 关联邮箱。"""
try:
link_rec = pb.collection("vip_email_link").get_list(
1, 1, {"filter": f'anonymous_user_id = "{user_id}"'}
)
if link_rec.items:
return link_rec.items[0].get("email") or ""
if user_id and user_id.startswith("user") and user_id[4:].isdigit():
link_rec = pb.collection("vip_email_link").get_list(
1, 1, {"filter": f'user_id = "{user_id}"'}
)
if link_rec.items:
return link_rec.items[0].get("email") or ""
except Exception:
pass
return None
def _uid_has_vip(pb, uid: str, site_id: str = "vip") -> bool:
if not uid:
return False
sv = pb.collection("site_vip").get_list(
1, 1, {"filter": f'user_id = "{uid}" && site_id = "{site_id}"'}
)
if sv.items:
rec = sv.items[0]
expires = rec.get("expires_at") or ""
if not expires:
return True
from datetime import datetime
try:
exp_dt = datetime.fromisoformat(expires.replace("Z", "+00:00"))
now = datetime.now(exp_dt.tzinfo) if exp_dt.tzinfo else datetime.now()
return exp_dt.timestamp() >= now.timestamp()
except Exception:
return True
pm = pb.collection("payments").get_list(
1,
1,
{"filter": f'user_id = "{uid}" && type = "vip"', "sort": "-created"},
)
return bool(pm.items)
@router.get("/check_vip")
async def nomadvip_check_vip(user_id: str = ""):
"""Return vip status for a user_id."""
user_id = (user_id or "").strip()
logging.info("[payjsapi] check_vip user_id=%s", user_id or "(empty)")
if not user_id:
return {"vip": False, "error": "missing user_id"}
try:
from ..services.pb import get_pb_client
pb = get_pb_client()
records = pb.collection("payments").get_list(1, 1, {"filter": f'order_id = "{order_id}"'})
return {"paid": len(records.items) > 0}
except Exception as e:
logging.warning(f"nomadvip PocketBase 查询失败: {e}")
return {"paid": False, "error": str(e)}
if not _uid_has_vip(pb, user_id):
logging.info("[payjsapi] check_vip miss user_id=%s vip=False", user_id)
return {"vip": False}
email = _get_linked_email_for_user(pb, user_id)
if not email:
admin_token = _get_pb_admin_token()
if admin_token and user_id.startswith("user") and user_id[4:].isdigit():
linked_member = _pb_find_linked_member_from_order(admin_token, user_id)
if linked_member:
linked_user_id, linked_email = linked_member
email = linked_email
_pb_upsert_email_link(admin_token, linked_email, linked_user_id, user_id)
logging.info(
"[payjsapi] check_vip hit user_id=%s vip=True email=%s",
user_id,
email or "(none)",
)
return {"vip": True, "email": email}
except Exception as error:
logging.warning("[payjsapi] check_vip failed: %s", error)
return {"vip": False, "error": str(error)}
def _get_pb_admin_token() -> str | None:
"""获取 PocketBase 管理员 token。"""
if not PB_ADMIN_EMAIL or not PB_ADMIN_PASSWORD:
return None
base = (PB_URL or "").rstrip("/")
for path in [
"/api/collections/_superusers/auth-with-password",
"/api/admins/auth-with-password",
]:
try:
response = requests.post(
f"{base}{path}",
json={"identity": PB_ADMIN_EMAIL, "password": PB_ADMIN_PASSWORD},
timeout=10,
)
if response.status_code == 200:
return response.json().get("token")
except Exception:
pass
return None
def _pb_escape(value: str) -> str:
return str(value).replace("\\", "\\\\").replace('"', '\\"')
def _pb_list_email_links(admin_token: str, email: str) -> list[dict]:
base = (PB_URL or "").rstrip("/")
headers = {"Authorization": f"Bearer {admin_token}"}
for candidate in [email, email.lower()]:
safe_email = _pb_escape(candidate)
response = requests.get(
f"{base}/api/collections/vip_email_link/records",
params={"filter": f'email = "{safe_email}"', "perPage": 1},
headers=headers,
timeout=10,
)
if response.status_code != 200:
continue
items = (response.json() or {}).get("items") or []
if items:
return items
return []
def _pb_upsert_email_link(
admin_token: str,
email: str,
pb_user_id: str,
anonymous_user_id: str | None = None,
):
base = (PB_URL or "").rstrip("/")
headers = {
"Authorization": f"Bearer {admin_token}",
"Content-Type": "application/json",
}
items = _pb_list_email_links(admin_token, email)
payload = {"email": email.lower(), "user_id": pb_user_id}
if anonymous_user_id:
payload["anonymous_user_id"] = anonymous_user_id
if items:
record_id = items[0].get("id")
if record_id:
requests.patch(
f"{base}/api/collections/vip_email_link/records/{record_id}",
json=payload,
headers=headers,
timeout=10,
)
return
requests.post(
f"{base}/api/collections/vip_email_link/records",
json=payload,
headers=headers,
timeout=10,
)
def _pb_move_site_vip(
admin_token: str,
from_user_id: str,
to_user_id: str,
site_id: str = "vip",
):
if not from_user_id or not to_user_id or from_user_id == to_user_id:
return
base = (PB_URL or "").rstrip("/")
safe_from_user_id = _pb_escape(from_user_id)
safe_site_id = _pb_escape(site_id)
response = requests.get(
f"{base}/api/collections/site_vip/records",
params={
"filter": f'user_id = "{safe_from_user_id}" && site_id = "{safe_site_id}"',
"perPage": 500,
},
headers={"Authorization": f"Bearer {admin_token}"},
timeout=10,
)
if response.status_code != 200:
return
headers = {
"Authorization": f"Bearer {admin_token}",
"Content-Type": "application/json",
}
items = (response.json() or {}).get("items") or []
for item in items:
record_id = item.get("id")
if not record_id:
continue
requests.patch(
f"{base}/api/collections/site_vip/records/{record_id}",
json={"user_id": to_user_id},
headers=headers,
timeout=10,
)
def _pb_latest_vip_payment(admin_token: str, user_id: str) -> dict | None:
base = (PB_URL or "").rstrip("/")
safe_user_id = _pb_escape(user_id)
response = requests.get(
f"{base}/api/collections/payments/records",
params={
"filter": f'user_id = "{safe_user_id}" && type = "vip"',
"perPage": 1,
"sort": "-created",
},
headers={"Authorization": f"Bearer {admin_token}"},
timeout=10,
)
if response.status_code != 200:
return None
items = (response.json() or {}).get("items") or []
return items[0] if items else None
def _pb_list_site_vip_by_order(
admin_token: str,
order_id: str,
site_id: str = "vip",
) -> list[dict]:
base = (PB_URL or "").rstrip("/")
safe_order_id = _pb_escape(order_id)
safe_site_id = _pb_escape(site_id)
response = requests.get(
f"{base}/api/collections/site_vip/records",
params={
"filter": f'order_id = "{safe_order_id}" && site_id = "{safe_site_id}"',
"perPage": 500,
"sort": "-created",
},
headers={"Authorization": f"Bearer {admin_token}"},
timeout=10,
)
if response.status_code != 200:
return []
return (response.json() or {}).get("items") or []
def _pb_get_user_email(admin_token: str, user_id: str) -> str | None:
base = (PB_URL or "").rstrip("/")
response = requests.get(
f"{base}/api/collections/users/records/{user_id}",
headers={"Authorization": f"Bearer {admin_token}"},
timeout=10,
)
if response.status_code != 200:
return None
email = (response.json() or {}).get("email") or ""
return email.strip().lower() or None
def _pb_find_linked_member_from_order(
admin_token: str,
anonymous_user_id: str,
) -> tuple[str, str] | None:
base = (PB_URL or "").rstrip("/")
safe_user_id = _pb_escape(anonymous_user_id)
response = requests.get(
f"{base}/api/collections/payments/records",
params={
"filter": f'user_id = "{safe_user_id}" && type = "vip"',
"perPage": 20,
"sort": "-created",
},
headers={"Authorization": f"Bearer {admin_token}"},
timeout=10,
)
if response.status_code != 200:
return None
items = (response.json() or {}).get("items") or []
seen_orders = set()
for item in items:
order_id = (item.get("order_id") or "").strip()
if not order_id or order_id in seen_orders:
continue
seen_orders.add(order_id)
for site_vip in _pb_list_site_vip_by_order(admin_token, order_id):
linked_user_id = (site_vip.get("user_id") or "").strip()
if (
not linked_user_id
or linked_user_id == anonymous_user_id
or (linked_user_id.startswith("user") and linked_user_id[4:].isdigit())
):
continue
email = _pb_get_user_email(admin_token, linked_user_id)
if email:
return linked_user_id, email
return None
def _pb_ensure_site_vip(
admin_token: str,
from_user_id: str,
to_user_id: str,
site_id: str = "vip",
):
if not from_user_id or not to_user_id or from_user_id == to_user_id:
return
base = (PB_URL or "").rstrip("/")
safe_from_user_id = _pb_escape(from_user_id)
safe_to_user_id = _pb_escape(to_user_id)
safe_site_id = _pb_escape(site_id)
source_records = requests.get(
f"{base}/api/collections/site_vip/records",
params={
"filter": f'user_id = "{safe_from_user_id}" && site_id = "{safe_site_id}"',
"perPage": 500,
"sort": "-created",
},
headers={"Authorization": f"Bearer {admin_token}"},
timeout=10,
)
to_records = requests.get(
f"{base}/api/collections/site_vip/records",
params={
"filter": f'user_id = "{safe_to_user_id}" && site_id = "{safe_site_id}"',
"perPage": 500,
"sort": "-created",
},
headers={"Authorization": f"Bearer {admin_token}"},
timeout=10,
)
payment = _pb_latest_vip_payment(admin_token, from_user_id)
order_id = (payment or {}).get("order_id") or ""
order_records = (
_pb_list_site_vip_by_order(admin_token, order_id, site_id=site_id)
if order_id
else []
)
source_items = (source_records.json() or {}).get("items") or [] if source_records.status_code == 200 else []
target_items = (to_records.json() or {}).get("items") or [] if to_records.status_code == 200 else []
canonical = (target_items or source_items or order_records or [None])[0]
headers = {
"Authorization": f"Bearer {admin_token}",
"Content-Type": "application/json",
}
if canonical and canonical.get("id"):
patch_body = {"user_id": to_user_id, "site_id": site_id}
next_order_id = order_id or (canonical.get("order_id") or "").strip()
if next_order_id:
patch_body["order_id"] = next_order_id
requests.patch(
f"{base}/api/collections/site_vip/records/{canonical['id']}",
json=patch_body,
headers=headers,
timeout=10,
)
duplicate_ids = set()
for record in [*target_items, *source_items, *order_records]:
record_id = record.get("id")
if record_id and record_id != canonical["id"]:
duplicate_ids.add(record_id)
for record_id in duplicate_ids:
requests.delete(
f"{base}/api/collections/site_vip/records/{record_id}",
headers={"Authorization": f"Bearer {admin_token}"},
timeout=10,
)
return
if not order_id:
return
requests.post(
f"{base}/api/collections/site_vip/records",
json={"user_id": to_user_id, "site_id": site_id, "order_id": order_id},
headers=headers,
timeout=10,
)
@router.post("/check_vip_by_email")
async def nomadvip_check_vip_by_email(item: dict):
"""Recover VIP by email/password, optionally with anonymous user_id."""
email = (item.get("email") or "").strip().lower()
password = (item.get("password") or "").strip()
anonymous_user_id = (
item.get("user_id") or item.get("anonymousUserId") or ""
).strip()
logging.info(
"[payjsapi] check_vip_by_email email=%s user_id=%s",
email or "(empty)",
anonymous_user_id or "(empty)",
)
if not email or not password:
return {"vip": False, "error": "missing email or password"}
base = (PB_URL or "").rstrip("/")
login_res = requests.post(
f"{base}/api/collections/users/auth-with-password",
json={"identity": email, "password": password},
timeout=10,
)
if login_res.status_code != 200:
try:
error = login_res.json() or {}
except Exception:
error = {}
logging.info(
"[payjsapi] check_vip_by_email login failed status=%s message=%s",
login_res.status_code,
error.get("message", ""),
)
return {
"vip": False,
"error": error.get("message") or "invalid email or password",
}
login_data = login_res.json() or {}
token = login_data.get("token")
record = login_data.get("record") or {}
pb_user_id = (record.get("id") or "").strip()
if not token or not pb_user_id:
return {"vip": False, "error": "login failed"}
admin_token = _get_pb_admin_token()
if not admin_token:
return {"vip": False, "error": "PocketBase admin auth failed"}
try:
from ..services.pb import get_pb_client
pb = get_pb_client()
matched_user_id = None
recovered_anonymous_user_id = (
anonymous_user_id
if anonymous_user_id.startswith("user") and anonymous_user_id[4:].isdigit()
else None
)
recovery_source = None
if recovered_anonymous_user_id and _uid_has_vip(pb, recovered_anonymous_user_id):
matched_user_id = recovered_anonymous_user_id
recovery_source = "request_user_id"
else:
items = _pb_list_email_links(admin_token, email)
logging.info(
"[payjsapi] check_vip_by_email vip_email_link email=%s count=%s",
email,
len(items),
)
for link_record in items:
linked_ids = []
linked_pb_user_id = (link_record.get("user_id") or "").strip()
linked_anonymous_user_id = (
link_record.get("anonymous_user_id") or ""
).strip()
if linked_pb_user_id:
linked_ids.append(linked_pb_user_id)
if linked_anonymous_user_id and linked_anonymous_user_id not in linked_ids:
linked_ids.append(linked_anonymous_user_id)
for linked_uid in linked_ids:
if not _uid_has_vip(pb, linked_uid):
continue
matched_user_id = linked_uid
if linked_anonymous_user_id.startswith("user") and linked_anonymous_user_id[4:].isdigit():
recovered_anonymous_user_id = linked_anonymous_user_id
elif linked_uid.startswith("user") and linked_uid[4:].isdigit():
recovered_anonymous_user_id = linked_uid
recovery_source = "vip_email_link"
break
if matched_user_id:
break
if not matched_user_id and _uid_has_vip(pb, pb_user_id):
matched_user_id = pb_user_id
recovered_anonymous_user_id = None
recovery_source = "pb_user_id"
if not matched_user_id:
logging.info("[payjsapi] check_vip_by_email no vip match for email=%s", email)
return {
"vip": False,
"error": "no linked vip record found; provide the paid user_id if this is a historical anonymous order",
}
if recovered_anonymous_user_id:
_pb_ensure_site_vip(admin_token, recovered_anonymous_user_id, pb_user_id)
_pb_upsert_email_link(
admin_token,
email,
pb_user_id,
recovered_anonymous_user_id,
)
else:
_pb_upsert_email_link(admin_token, email, pb_user_id)
logging.info(
"[payjsapi] check_vip_by_email recovered by %s matched_user_id=%s effectiveUserId=%s anonymousUserId=%s",
recovery_source or "unknown",
matched_user_id,
pb_user_id,
recovered_anonymous_user_id or "(none)",
)
return {
"vip": True,
"token": token,
"record": record,
"effectiveUserId": pb_user_id,
"anonymousUserId": recovered_anonymous_user_id,
}
except Exception as error:
logging.warning("[payjsapi] check_vip_by_email failed: %s", error)
return {"vip": False, "error": str(error)}
@router.get("/order_status")
async def nomadvip_order_status(order_id: str = "", out_trade_no: str = ""):
"""查询订单支付状态,优先查 ZPAY失败时回退 PocketBase。"""
order_id = (order_id or out_trade_no or "").strip()
logging.info("[payjsapi] order_status order_id=%s", order_id or "(空)")
if not order_id:
return {"paid": False, "error": "缺少 order_id 或 out_trade_no"}
try:
from ..services.pb import get_pb_client
pb = get_pb_client()
records = pb.collection("payments").get_list(
1, 1, {"filter": f'order_id = "{order_id}"'}
)
paid = len(records.items) > 0
logging.info("[payjsapi] order_status PocketBase paid=%s", paid)
if paid:
return {"paid": True, "source": "pocketbase"}
except Exception as error:
logging.warning("[payjsapi] order_status PocketBase failed: %s", error)
xorpay = get_payment_provider_by_name("xorpay")
xorpay_result = xorpay.query_order_status(order_id, timeout=8)
if xorpay_result.get("error"):
logging.warning(
"[payjsapi] order_status XorPay query failed order_id=%s error=%s",
order_id,
xorpay_result.get("error"),
)
else:
logging.info(
"[payjsapi] order_status XorPay status=%s paid=%s",
xorpay_result.get("status", ""),
xorpay_result.get("paid", False),
)
if xorpay_result.get("paid"):
return {
"paid": True,
"source": "xorpay",
"status": xorpay_result.get("status"),
"pay_price": xorpay_result.get("pay_price") or "",
}
zpay = get_payment_provider_by_name("zpay")
if isinstance(zpay, ZPayProvider):
result = zpay.query_order_status(order_id, timeout=8)
if not result.get("error") and result.get("paid"):
logging.info("[payjsapi] order_status ZPAY paid=True")
return result
return {"paid": False}
@router.post("/xorpay_notify")
async def nomadvip_xorpay_notify(request: Request):
"""nomadvip 专用XorPay 异步通知(路径即渠道,不依赖全局 PAYMENT_PROVIDER"""
"""XorPay 异步通知。"""
form_data = await request.form()
data = {k: v for k, v in form_data.items()}
logging.info(f"nomadvip XorPay 异步通知: {data}")
logging.info("nomadvip XorPay notify: %s", data)
order_id = data.get("order_id", "")
if order_id in processed_orders:
logging.info(f"nomadvip 订单已处理,跳过: {order_id}")
logging.info("nomadvip order already processed, skip: %s", order_id)
return "ok"
provider = get_payment_provider_by_name("xorpay")
if not provider.verify_notify(data):
logging.error(f"nomadvip XorPay 验签失败: {data}")
logging.error("nomadvip XorPay verify failed: %s", data)
raise HTTPException(status_code=400, detail="sign error")
parsed = provider.parse_notify(data)
if parsed.get("trade_status") != "success":
return "ok"
pay_price = parsed.get("pay_price", data.get("pay_price", ""))
try:
handle_payment_success(order_id, pay_price, "nomadvip-xorpay", use_memos=True, site_id="vip")
logging.info(f"nomadvip 业务处理完成: order_id={order_id}")
except Exception as e:
logging.error(f"nomadvip PocketBase/Memos 处理异常: {e}", exc_info=True)
handle_payment_success(
order_id,
pay_price,
"nomadvip-xorpay",
use_memos=True,
site_id="vip",
)
logging.info("nomadvip business done: order_id=%s", order_id)
except Exception as error:
logging.error(
"nomadvip PocketBase/Memos failed: %s", error, exc_info=True
)
raise
return provider.success_response()
@@ -194,31 +841,43 @@ async def nomadvip_xorpay_notify(request: Request):
@router.get("/zpay_notify")
@router.post("/zpay_notify")
async def nomadvip_zpay_notify(request: Request):
"""nomadvip 专用ZPAY 异步通知(路径即渠道,不依赖全局 PAYMENT_PROVIDER"""
"""ZPAY 异步通知。"""
provider = get_payment_provider_by_name("zpay")
if request.method == "GET":
data = dict(request.query_params)
else:
form_data = await request.form()
data = {k: v for k, v in form_data.items()}
logging.info(f"nomadvip ZPAY 异步通知: {data}")
logging.info("nomadvip ZPAY notify: %s", data)
order_id = data.get("out_trade_no", "")
if order_id in processed_orders:
logging.info(f"nomadvip 订单已处理,跳过: {order_id}")
logging.info("nomadvip order already processed, skip: %s", order_id)
return "success"
if not provider.verify_notify(data):
logging.error(f"nomadvip ZPAY 验签失败: {data}")
logging.error("nomadvip ZPAY verify failed: %s", data)
raise HTTPException(status_code=400, detail="sign error")
parsed = provider.parse_notify(data)
trade_status = parsed.get("trade_status") or data.get("trade_status", "")
if trade_status != "success" and trade_status != "TRADE_SUCCESS":
logging.info(f"nomadvip 非成功状态,跳过: trade_status={trade_status}")
if trade_status not in ("success", "TRADE_SUCCESS"):
logging.info("nomadvip non-success trade status, skip: %s", trade_status)
return "success"
pay_price = parsed.get("pay_price") or data.get("money", "")
try:
handle_payment_success(order_id, pay_price, "nomadvip-zpay", use_memos=True, site_id="vip")
logging.info(f"nomadvip 业务处理完成: order_id={order_id}")
except Exception as e:
logging.error(f"nomadvip PocketBase/Memos 处理异常: {e}", exc_info=True)
handle_payment_success(
order_id,
pay_price,
"nomadvip-zpay",
use_memos=True,
site_id="vip",
)
logging.info("nomadvip business done: order_id=%s", order_id)
except Exception as error:
logging.error(
"nomadvip PocketBase/Memos failed: %s", error, exc_info=True
)
raise
return provider.success_response()

View File

@@ -9,7 +9,12 @@ from fastapi import APIRouter, Request, HTTPException
from fastapi.responses import RedirectResponse, JSONResponse
from ..config import BASE_URL
from ..payment import get_payment_provider, get_payment_provider_by_name
from ..payment import (
get_payment_provider,
get_payment_provider_by_name,
resolve_provider,
resolve_channel,
)
from ..payment.zpay import ZPayProvider
from ..services import handle_payment_success, processed_orders
@@ -61,10 +66,13 @@ def _sync_wxh5_vip_to_supabase(user_id: str, plan: str):
@router.post("/payh5")
async def wxh5_payh5(item: dict):
"""wxH5 专用:创建支付订单,支持 per-request provider 覆盖"""
req_provider = (item.get("provider") or "").strip().lower()
provider = get_payment_provider_by_name(req_provider) if req_provider in ("xorpay", "zpay") else get_payment_provider()
async def wxh5_payh5(item: dict, request: Request):
"""wxH5 专用:创建支付订单。微信内用 xorpayPC/手机用 zpay默认微信支付支付宝暂不提供"""
req_provider = (item.get("provider") or "").strip().lower() or None
device = (item.get("device") or "pc").lower()
ua = (request.headers.get("user-agent") or "").strip()
provider = resolve_provider(device, req_provider, ua)
channel = resolve_channel(item.get("channel"))
plan = (item.get("plan") or "year").lower()
total_fee = PLAN_MAP.get(plan, PLAN_MAP["year"])[0]
total_fee = int(item.get("total_fee", total_fee))
@@ -81,7 +89,6 @@ async def wxh5_payh5(item: dict):
type_map = {"book": "购买书籍", "meetup": "社区报名", "video": "视频解锁", "vip": "VIP会员充值"}
name = type_map.get(pay_type, "VIP会员充值")
notify_url = _get_notify_path(provider.name)
channel = item.get("channel", "alipay")
result = provider.create_order(
order_id=order_id,
name=name,
@@ -113,11 +120,9 @@ async def wxh5_payh5(item: dict):
@router.post("/payh5/redirect")
async def wxh5_payh5_redirect(item: dict, request: Request):
"""wxH5 专用ZPAY mapi 接口xorpay 无需 redirect"""
req_provider = (item.get("provider") or "").strip().lower()
provider = get_payment_provider_by_name("zpay") if req_provider == "zpay" else get_payment_provider()
if not isinstance(provider, ZPayProvider):
raise HTTPException(status_code=400, detail="payh5/redirect 仅支持 zpay")
"""wxH5 专用ZPAY mapi 接口PC/手机用 zpay默认微信支付支付宝暂不提供"""
provider = get_payment_provider_by_name("zpay")
channel = resolve_channel(item.get("channel"))
plan = (item.get("plan") or "year").lower()
total_fee = PLAN_MAP.get(plan, PLAN_MAP["year"])[0]
try:
@@ -127,7 +132,6 @@ async def wxh5_payh5_redirect(item: dict, request: Request):
total_fee_yuan = f"{total_fee / 100:.2f}"
user_id = item.get("user_id", "unknown")
pay_type = (item.get("type") or "vip").lower()
channel = item.get("channel", "alipay")
device = item.get("device", "pc")
client_ip = item.get("client_ip") or (request.client.host if request.client else "127.0.0.1")
forwarded = request.headers.get("x-forwarded-for")

View File

@@ -24,7 +24,7 @@ class MemosSDK:
def create_user(
self,
user_id: str,
password: str = "123456",
password: str = "12345678",
) -> dict:
"""
创建 Memos 用户
@@ -72,7 +72,7 @@ class MemosSDK:
def create_user_by_username(
self,
username: str,
password: str = "123456",
password: str = "12345678",
) -> dict:
"""按用户名直接创建(用于 /create_user 接口)"""
if not self.enabled:

116
app/services/community.py Normal file
View File

@@ -0,0 +1,116 @@
"""
Private community service layer.
"""
from __future__ import annotations
import logging
import os
from ..config import BASE_URL
from ..db import community_sqlite as sqlite_db
logger = logging.getLogger(__name__)
try:
sqlite_db.init_schema()
logger.info("community: SQLite initialized at %s", sqlite_db.DB_PATH)
except Exception as exc:
logger.warning("community SQLite init failed: %s", exc)
def list_memos(
page: int = 1,
per_page: int = 50,
state: str = "NORMAL",
order_by: str = "-pinned,-created",
memo_format: bool = False,
query: str = "",
viewer_id: str = "",
) -> dict:
return sqlite_db.list_memos(
page=page,
per_page=per_page,
state=state,
order_by=order_by,
memo_format=memo_format,
query=query,
viewer_id=viewer_id,
)
def get_memo(memo_id: str, viewer_id: str = "", include_comments: bool = True) -> dict | None:
return sqlite_db.get_memo(memo_id, viewer_id=viewer_id, include_comments=include_comments)
def create_memo(
user_id: str,
content: str,
pinned: bool = False,
visibility: str = "PRIVATE",
state: str = "NORMAL",
) -> dict:
return sqlite_db.create_memo(
user_id=user_id,
content=content,
pinned=pinned,
visibility=visibility,
state=state,
)
def update_memo(
memo_id: str,
user_id: str,
pinned: bool | None = None,
content: str | None = None,
visibility: str | None = None,
state: str | None = None,
) -> dict | None:
return sqlite_db.update_memo(
memo_id,
user_id,
pinned=pinned,
content=content,
visibility=visibility,
state=state,
)
def delete_memo(memo_id: str, user_id: str) -> bool:
return sqlite_db.delete_memo(memo_id, user_id)
def create_comment(memo_id: str, user_id: str, content: str) -> dict:
return sqlite_db.create_comment(memo_id, user_id, content)
def list_comments(memo_id: str) -> list[dict]:
return sqlite_db.list_comments(memo_id)
def delete_comment(comment_id: str, user_id: str) -> bool:
return sqlite_db.delete_comment(comment_id, user_id)
def toggle_reaction(memo_id: str, user_id: str, reaction_type: str = sqlite_db.UPVOTE_REACTION) -> dict:
return sqlite_db.toggle_reaction(memo_id, user_id, reaction_type)
def upload_image(
data: bytes,
filename: str,
content_type: str,
user_id: str,
base_url: str = "",
) -> dict:
base = base_url or (os.getenv("BASE_URL") or BASE_URL or "http://localhost:8000")
return sqlite_db.save_upload(data, filename, content_type, user_id, base)
def list_uploads(page: int = 1, per_page: int = 24, user_id: str = "") -> dict:
return sqlite_db.list_uploads(page=page, per_page=per_page, user_id=user_id)
def get_stats(member_count: int | None = None) -> dict:
return sqlite_db.get_stats(member_count=member_count)

View File

@@ -16,6 +16,6 @@ def _get_memos_sdk() -> MemosSDK:
return _memos_sdk
def create_memos_user(user_id: str, password: str = "123456"):
def create_memos_user(user_id: str, password: str = "12345678"):
"""创建 Memos 用户"""
return _get_memos_sdk().create_user(user_id=user_id, password=password)

View File

@@ -1,11 +1,15 @@
"""支付成功后的统一业务处理(基于 SDK"""
import logging
import requests
from ..config import PB_URL, PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD
from .pb import get_pb_client
from .memos import create_memos_user
# 已处理订单(防重复)
processed_orders: set[str] = set()
DEFAULT_PASSWORD = "12345678"
def parse_order_id(order_id: str) -> tuple[str, str]:
@@ -19,6 +23,77 @@ def parse_order_id(order_id: str) -> tuple[str, str]:
return prefix, "unknown"
def _decode_pending_email(user_id: str) -> str | None:
"""从 pending_hex 格式解析邮箱"""
if not user_id.startswith("pending_"):
return None
hex_part = user_id[8:] # 去掉 "pending_"
try:
return bytes.fromhex(hex_part).decode("utf-8")
except Exception:
return None
def _create_user_by_email(email: str) -> str | None:
"""用管理员创建用户,返回 user_id。若已存在则返回已有用户 id"""
token, _ = _get_admin_token()
if not token:
return None
base = (PB_URL or "").rstrip("/")
r = requests.post(
f"{base}/api/collections/users/records",
json={
"email": email,
"password": DEFAULT_PASSWORD,
"passwordConfirm": DEFAULT_PASSWORD,
},
headers={"Content-Type": "application/json", "Authorization": f"Bearer {token}"},
timeout=10,
)
if r.status_code == 200:
return r.json().get("id")
if r.status_code == 400:
try:
err = r.json()
if "already" in str(err.get("data", {}).get("email", {}).get("message", "")).lower():
# 用户已存在,查询获取 id
get_r = requests.get(
f"{base}/api/collections/users/records",
params={"filter": f'email="{email}"', "perPage": 1},
headers={"Authorization": f"Bearer {token}"},
timeout=10,
)
if get_r.status_code == 200:
items = get_r.json().get("items", [])
if items:
return items[0].get("id")
except Exception:
pass
logging.warning(f"create user failed: {r.status_code} {r.text[:200]}")
return None
def _get_admin_token() -> tuple[str | None, str | None]:
"""获取 PocketBase 管理员 token"""
if not PB_ADMIN_EMAIL or not PB_ADMIN_PASSWORD:
return None, "未配置"
base = (PB_URL or "").rstrip("/")
for path in ["/api/collections/_superusers/auth-with-password", "/api/admins/auth-with-password"]:
try:
r = requests.post(
f"{base}{path}",
json={"identity": PB_ADMIN_EMAIL, "password": PB_ADMIN_PASSWORD},
timeout=10,
)
if r.status_code == 200:
return r.json().get("token"), None
if r.status_code == 404:
continue
except Exception:
pass
return None, "认证失败"
def handle_payment_success(
order_id: str,
pay_price: str,
@@ -36,10 +111,23 @@ def handle_payment_success(
if not user_id:
return
print(f"支付成功 [{provider_name}] 用户ID: {user_id},类型: {pay_type},站点: {site_id},订单: {order_id}")
# meetup 新用户:支付成功后才创建用户
if pay_type == "meetup" and user_id.startswith("pending_"):
email = _decode_pending_email(user_id)
if email:
real_id = _create_user_by_email(email)
if real_id:
user_id = real_id
logging.info(f"meetup 支付成功,已创建用户: {email}")
else:
logging.error(f"meetup 支付成功但创建用户失败: {email}")
return
logging.info("[payjsapi] 支付成功 provider=%s user_id=%s pay_type=%s site_id=%s order_id=%s",
provider_name, user_id, pay_type, site_id, order_id)
if pay_type == "vip" and use_memos:
create_memos_user(user_id)
create_memos_user(user_id, password="12345678")
if pay_type == "meetup":
logging.info(f"meetup 类型支付成功,不创建 memos 用户 - user_id: {user_id}")
@@ -75,12 +163,19 @@ def handle_payment_success(
})
logging.info(f"meetup 完整记录创建(兜底) - user_id: {user_id}")
else:
pb_client.collection("payments").create({
"user_id": user_id,
"type": pay_type,
"order_id": order_id,
"amount": amount_cents,
existing = pb_client.collection("payments").get_list(1, 1, {
"filter": f'order_id = "{order_id}"',
})
if not existing.items:
payload = {
"user_id": user_id,
"type": pay_type,
"order_id": order_id,
"amount": amount_cents,
"total_fee": amount_cents,
"channel": "wxpay",
}
pb_client.collection("payments").create(payload)
# 写入 site_vip站点独立 VIP。meetup 类型统一写 site_id=meetup
effective_site_id = "meetup" if pay_type == "meetup" else site_id
@@ -109,6 +204,5 @@ def handle_payment_success(
processed_orders.add(order_id)
except Exception as e:
print(f"PocketBase 操作失败: {e}")
logging.error(f"PocketBase 操作失败 - order_id: {order_id}, error: {e}")
logging.error("PocketBase 操作失败 order_id=%s error=%s", order_id, e)
raise

View File

@@ -1,7 +1,10 @@
"""PocketBase 客户端(基于 SDK"""
import logging
from app.config import PB_URL, PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD
from app.sdk.pocketbase import PocketBaseSDK
logger = logging.getLogger(__name__)
_pb_sdk: PocketBaseSDK | None = None
@@ -20,5 +23,5 @@ def get_pb_client():
"""获取已认证的 PocketBase 客户端"""
sdk = _get_pb_sdk()
client = sdk.get_client()
print("PocketBase admin 登录成功")
logger.info("PocketBase admin 登录成功")
return client

BIN
data/community.db Normal file

Binary file not shown.

Binary file not shown.

After

Width:  |  Height:  |  Size: 20 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 20 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 20 KiB

View File

@@ -0,0 +1,21 @@
# 私密社群 PocketBase 集合 schema
在 PocketBase 管理后台创建集合 `community_memos`,字段如下:
| 字段名 | 类型 | 必填 | 说明 |
|--------|------|------|------|
| user_id | Text | ✓ | 发布者 user_idPocketBase 用户 ID 或匿名 user123456 |
| content | Text | ✓ | 动态内容,支持 Markdown、#标签、图片 `![](url)` |
| pinned | Bool | | 是否置顶,默认 false |
| visibility | Text | | 可见性PRIVATE/PROTECTED/PUBLIC默认 PRIVATE |
| state | Text | | 状态NORMAL/ARCHIVED默认 NORMAL |
- 集合名:`community_memos`
- 列表/查看权限:仅服务端 admin 操作,前端通过 payjsapi `/community/memos` 接口访问
- 创建/更新/删除:仅 VIP 用户,由 payjsapi 校验
创建步骤:
1. 登录 PocketBase 管理后台
2. Collections → New collection → 名称 `community_memos`
3. 添加字段user_id (Text), content (Editor 或 Text), pinned (Bool)
4. 保存

45
docs/COMMUNITY_SQLITE.md Normal file
View File

@@ -0,0 +1,45 @@
# 社群模块 SQLite 存储
默认使用 SQLite无需 PocketBase、MinIO 等外部依赖。图片上传保存到本地目录。
## 配置
| 环境变量 | 说明 | 默认 |
|----------|------|------|
| COMMUNITY_STORAGE | `sqlite` \| `pocketbase` | `sqlite` |
| COMMUNITY_DATA_DIR | 数据目录(含 db、uploads | `./data` |
| BASE_URL | 用于拼接图片 URL | 见主配置 |
## 文件结构
```
data/
├── community.db # SQLite 数据库
└── community_uploads/ # 上传图片
└── 1731234567_abc123.jpg
```
## 表结构community_memos
| 字段 | 类型 | 说明 |
|------|------|------|
| id | TEXT | 主键 |
| user_id | TEXT | 发布者 |
| content | TEXT | 内容 |
| pinned | INTEGER | 置顶 0/1 |
| visibility | TEXT | PRIVATE/PROTECTED/PUBLIC |
| state | TEXT | NORMAL/ARCHIVED |
| created | TEXT | ISO 时间 |
| updated | TEXT | ISO 时间 |
## 启动时自动建表
`app/services/community.py` 加载时自动执行 `init_schema()`,无需手动迁移。
## 切换 PocketBase
```bash
COMMUNITY_STORAGE=pocketbase
```
需配置 PocketBase 和 MinIO图片上传

View File

@@ -1,5 +1,6 @@
fastapi
uvicorn
python-dotenv
sqlalchemy
pydantic
python-multipart

4
run-with-logs.bat Normal file
View File

@@ -0,0 +1,4 @@
@echo off
REM 强制无缓冲输出,确保日志立即显示在终端
set PYTHONUNBUFFERED=1
python -u run.py

4
run-with-logs.sh Normal file
View File

@@ -0,0 +1,4 @@
#!/bin/bash
# 强制无缓冲输出,确保日志立即显示在终端
export PYTHONUNBUFFERED=1
exec python -u run.py

31
run.py
View File

@@ -1,6 +1,31 @@
import os
import sys
os.environ.setdefault("PYTHONUNBUFFERED", "1")
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(line_buffering=True)
if hasattr(sys.stderr, "reconfigure"):
sys.stderr.reconfigure(line_buffering=True)
import uvicorn
from dotenv import load_dotenv
load_dotenv()
if __name__ == "__main__":
uvicorn.run("app.main:app", host="0.0.0.0", port=8700, reload=True)
# uvicorn.run("app.main:app", host="0.0.0.0", port=8200, reload=True)
port = int(os.getenv("PORT", "8007"))
reload_enabled = os.getenv("UVICORN_RELOAD", "0").strip().lower() in ("1", "true", "yes")
try:
sys.stderr.write(f"\n[payjsapi] startup http://0.0.0.0:{port}\n")
sys.stderr.flush()
except (OSError, ValueError):
pass
uvicorn.run(
"app.main:app",
host="0.0.0.0",
port=port,
reload=reload_enabled,
log_level="info",
)