diff --git a/app/training/db.py b/app/training/db.py index a8ef063..2b5135d 100644 --- a/app/training/db.py +++ b/app/training/db.py @@ -13,7 +13,7 @@ os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) def get_conn(): - conn = sqlite3.connect(DB_PATH) + conn = sqlite3.connect(DB_PATH, check_same_thread=False) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") return conn diff --git a/app/training/main.py b/app/training/main.py index 72829c9..1c4c32a 100644 --- a/app/training/main.py +++ b/app/training/main.py @@ -3,6 +3,7 @@ 由 server-core/dispatcher.py 统一对外(opc.pinesound.cn):`/api/*` 路由到本子应用, `/auth`、`/opc`、`/admin` 等由平台应用(app.main)服务。独立使用 data/opc.db 业务库。 """ +import asyncio import json import os import time @@ -15,7 +16,16 @@ from . import db, opc_engine, policy_engine, policy_data, survey_data from .auth import hash_password, make_token, verify_token, send_sms_code, check_sms_code, PHONE_RE from .opc_engine import SECTIONS, AXIS_NAMES, ADAPT_LABELS, PROFILES -app = FastAPI(title="云超服 OPC 培训站后端", version="0.1") +from contextlib import asynccontextmanager + + +@asynccontextmanager +async def lifespan(app): + await asyncio.to_thread(db.init_db) + yield + + +app = FastAPI(title="云超服 OPC 培训站后端", version="0.1", lifespan=lifespan) ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # server-core 根 @@ -82,7 +92,6 @@ app.add_middleware( allow_headers=["*"], ) -db.init_db() def now_iso(): @@ -145,14 +154,14 @@ def jsonify(obj): # ================= 域名验证文件(微信/腾讯域名校验,须在根路径返回) ================= @app.get("/SpXvScDiDT.txt") -def domain_verify(): +async def domain_verify(): from fastapi.responses import Response return Response(content="d813c7bbab566807a833fbc3b0e20a9c", media_type="text/plain") # ================= 认证 ================= @app.get("/api/health") -def health(): +async def health(): return {"ok": True, "service": "opc-fastapi"} @@ -167,12 +176,12 @@ async def register(req: Request): raise HTTPException(400, "账号至少 2 个字符") if len(password) < 4: raise HTTPException(400, "密码至少 4 个字符") - if db.fetch_one("accounts", username=username): + if await asyncio.to_thread(db.fetch_one, "accounts", username=username): raise HTTPException(400, "账号已存在,请直接登录") entry = {"id": db.gen_id("U"), "username": username, "password": hash_password(password), "name": name or username, "contact": contact, "identities": json.dumps(["opc_member|certified"]), "created_at": now_iso()} - db.insert("accounts", entry) + await asyncio.to_thread(db.insert, "accounts", entry) return {"ok": True, "token": make_token(username), "username": username, "name": name or username} @@ -185,7 +194,7 @@ async def login(req: Request): if password == "123456": return {"ok": True, "token": make_token("pine"), "username": "pine", "name": "Pine"} raise HTTPException(401, "密码错误,请重新输入") - acct = db.fetch_one("accounts", username=username) + acct = await asyncio.to_thread(db.fetch_one, "accounts", username=username) if not acct: raise HTTPException(404, "账号不存在,请先注册") if acct["password"] != hash_password(password): @@ -275,18 +284,18 @@ def auth_by_phone(body: dict): @app.post("/api/auth/phone-login") async def phone_login(req: Request): b = await req.json() - acct, is_new = auth_by_phone(b) + acct, is_new = await asyncio.to_thread(auth_by_phone, b) return {"ok": True, "token": make_token(acct["username"]), "username": acct["username"], "name": acct.get("name") or acct["username"], "isNew": is_new} @app.post("/api/auth/logout") -def logout(): +async def logout(): return {"ok": True} @app.get("/api/auth/verify") -def verify(authorization: str = Header(default="")): +async def verify(authorization: str = Header(default="")): payload = require_auth(authorization) return {"ok": True, "username": payload["username"]} @@ -311,22 +320,22 @@ async def wx_login(req: Request): raise HTTPException(401, f"微信登录失败:{errmsg}") openid = wresp.get("openid") - acct = db.fetch_one("accounts", wxid=openid) + acct = await asyncio.to_thread(db.fetch_one, "accounts", wxid=openid) if not acct: name = str(b.get("nickName", "")).strip() or "微信用户" acct = {"id": db.gen_id("U"), "username": openid, "wxid": openid, "phone": "", "name": name, "avatar": str(b.get("avatarUrl", "")), "phone_bound": 0, "identities": json.dumps(["wx"]), "created_at": now_iso()} - db.insert("accounts", acct) + await asyncio.to_thread(db.insert, "accounts", acct) return {"ok": True, "token": make_token(acct["username"]), "username": acct["username"], "name": acct.get("name") or "微信用户", "avatar": abs_url(acct.get("avatar")), "phoneBound": bool(acct.get("phone_bound")), "configured": WX_HAS_CONFIG} @app.get("/api/auth/me") -def me(authorization: str = Header(default="")): +async def me(authorization: str = Header(default="")): payload = require_auth(authorization) - acct = db.fetch_one("accounts", username=payload["username"]) + acct = await asyncio.to_thread(db.fetch_one, "accounts", username=payload["username"]) if not acct: raise HTTPException(404, "账号不存在") return {"ok": True, "user": _user_payload(acct)} @@ -337,7 +346,7 @@ async def update_profile(req: Request, authorization: str = Header(default="")): """更新当前用户资料:昵称/头像 + 报名资料(status/topics/source)""" payload = require_auth(authorization) b = await req.json() - acct = db.fetch_one("accounts", username=payload["username"]) + acct = await asyncio.to_thread(db.fetch_one, "accounts", username=payload["username"]) if not acct: raise HTTPException(404, "账号不存在") patch = {} @@ -352,8 +361,8 @@ async def update_profile(req: Request, authorization: str = Header(default="")): if b.get("topics") is not None: patch["topics"] = json.dumps(b["topics"], ensure_ascii=False) if isinstance(b["topics"], list) else str(b["topics"]) if patch: - db.update_row("accounts", acct["id"], patch) - acct = db.fetch_one("accounts", username=payload["username"]) + await asyncio.to_thread(db.update_row, "accounts", acct["id"], patch) + acct = await asyncio.to_thread(db.fetch_one, "accounts", username=payload["username"]) return {"ok": True, "user": _user_payload(acct)} @@ -379,11 +388,11 @@ async def wx_phone(req: Request, authorization: str = Header(default="")): phone = (data.get("phone_info") or {}).get("purePhoneNumber", "") if not phone: raise HTTPException(400, "未获取到手机号") - acct = db.fetch_one("accounts", username=payload["username"]) + acct = await asyncio.to_thread(db.fetch_one, "accounts", username=payload["username"]) if not acct: raise HTTPException(404, "账号不存在") # 绑号即合并:若该手机号下有其它账号(如网页手机号登录账号),把其数据并入当前微信账号 - acct = _adopt_phone(acct, phone) + acct = await asyncio.to_thread(_adopt_phone, acct, phone) return {"ok": True, "phoneBound": True, "phone": phone} @@ -397,11 +406,11 @@ async def bind_phone(req: Request, authorization: str = Header(default="")): raise HTTPException(400, "请输入正确的 11 位手机号") if not check_sms_code(phone, code): raise HTTPException(401, "验证码错误或已过期") - acct = db.fetch_one("accounts", username=payload["username"]) + acct = await asyncio.to_thread(db.fetch_one, "accounts", username=payload["username"]) if not acct: raise HTTPException(404, "账号不存在") # 短信绑定同样合并同一手机号下的其它账号 - acct = _adopt_phone(acct, phone) + acct = await asyncio.to_thread(_adopt_phone, acct, phone) return {"ok": True, "phoneBound": True, "phone": phone} @@ -449,29 +458,29 @@ def event_out(e): @app.get("/api/events") -def events(request: Request, current: Optional[str] = None, bookable: Optional[str] = None): - all_events = db.list_all("events") +async def events(request: Request, current: Optional[str] = None, bookable: Optional[str] = None): + all_events = await asyncio.to_thread(db.list_all, "events") if current is not None or bookable is not None: upcoming = _upcoming(all_events) if current is not None: free = next((e for e in upcoming if e["type"] == "free"), None) salon = next((e for e in upcoming if e["type"] == "salon"), None) - return {"ok": True, "next": event_out(upcoming[0] if upcoming else None), - "free": event_out(free), "salon": event_out(salon), - "upcoming": [event_out(e) for e in upcoming[:6]]} + return {"ok": True, "next": await asyncio.to_thread(event_out, upcoming[0] if upcoming else None), + "free": await asyncio.to_thread(event_out, free), "salon": await asyncio.to_thread(event_out, salon), + "upcoming": [await asyncio.to_thread(event_out, e) for e in upcoming[:6]]} if bookable is not None: # 可公开报名:排除 待开放(pending)/邀请中(invite)/已结束(done)/已满员(full) lst = [e for e in upcoming if e["status"] not in ("closed", "done", "pending", "invite", "full")][:30] - return {"ok": True, "list": [event_out(e) for e in lst]} - return {"ok": True, "list": [event_out(e) for e in all_events]} + return {"ok": True, "list": [await asyncio.to_thread(event_out, e) for e in lst]} + return {"ok": True, "list": [await asyncio.to_thread(event_out, e) for e in all_events]} @app.get("/api/events/{eid}") -def event_detail(eid: str): - e = db.fetch_by_id("events", eid) +async def event_detail(eid: str): + e = await asyncio.to_thread(db.fetch_by_id, "events", eid) if not e: raise HTTPException(404, "活动不存在") - return {"ok": True, "event": event_out(e)} + return {"ok": True, "event": await asyncio.to_thread(event_out, e)} @app.post("/api/events") @@ -500,8 +509,8 @@ async def create_event(req: Request, authorization: str = Header(default="")): "audit_mode": 'manual' if b.get("auditMode") == "manual" else "auto", "show_capacity": 1 if b.get("showCapacity") else 0, } - db.insert("events", entry) - return {"ok": True, "entry": event_out(entry)} + await asyncio.to_thread(db.insert, "events", entry) + return {"ok": True, "entry": await asyncio.to_thread(event_out, entry)} @app.put("/api/events/{eid}") @@ -529,16 +538,16 @@ async def update_event(eid: str, req: Request, authorization: str = Header(defau patch["show_capacity"] = 1 if b["showCapacity"] else 0 if not patch: raise HTTPException(400, "无更新字段") - if not db.fetch_by_id("events", eid): + if not await asyncio.to_thread(db.fetch_by_id, "events", eid): raise HTTPException(404, "排期不存在") - db.update_row("events", eid, patch) - return {"ok": True, "entry": event_out(db.fetch_by_id("events", eid))} + await asyncio.to_thread(db.update_row, "events", eid, patch) + return {"ok": True, "entry": await asyncio.to_thread(event_out, await asyncio.to_thread(db.fetch_by_id, "events", eid))} @app.delete("/api/events/{eid}") -def delete_event(eid: str, authorization: str = Header(default="")): +async def delete_event(eid: str, authorization: str = Header(default="")): require_auth(authorization) - db.delete_row("events", eid) + await asyncio.to_thread(db.delete_row, "events", eid) return {"ok": True} @@ -546,7 +555,7 @@ def delete_event(eid: str, authorization: str = Header(default="")): @app.post("/api/bookings") async def create_booking(req: Request, authorization: str = Header(default="")): b = await req.json() - event = db.fetch_by_id("events", str(b.get("eventId", "")).strip()) + event = await asyncio.to_thread(db.fetch_by_id, "events", str(b.get("eventId", "")).strip()) # 校验活动可报名:公开报名仅 open/full;待开放/邀请中/已结束不可报名 if not event: raise HTTPException(404, "活动不存在") @@ -556,7 +565,7 @@ async def create_booking(req: Request, authorization: str = Header(default="")): # 强制登录:必须携带有效 token(无登录 → 401) payload = require_auth(authorization) - acct = db.fetch_one("accounts", username=payload["username"]) + acct = await asyncio.to_thread(db.fetch_one, "accounts", username=payload["username"]) if not acct: raise HTTPException(401, "登录状态异常,请重新登录") # 报名必须已绑定手机号(登录后自动/引导绑定),保证报名有联系方式 @@ -580,7 +589,7 @@ async def create_booking(req: Request, authorization: str = Header(default="")): "question": b.get("question", ""), "source": source, "audit_status": "approved" if event.get("audit_mode") != "manual" else "pending", } - db.insert("bookings", entry) + await asyncio.to_thread(db.insert, "bookings", entry) return {"ok": True, "id": entry["id"], "createdAt": entry["created_at"], "username": username, "name": name, "auditStatus": entry["audit_status"]} @@ -596,13 +605,13 @@ def _decode_bk(row): @app.get("/api/bookings/mine") -def my_bookings(authorization: str = Header(default="")): +async def my_bookings(authorization: str = Header(default="")): payload = require_auth(authorization) - acct = db.fetch_one("accounts", username=payload["username"]) + acct = await asyncio.to_thread(db.fetch_one, "accounts", username=payload["username"]) ids = {payload["username"]} if acct and acct.get("phone"): ids.add(acct["phone"]) - conn = db.get_conn() + conn = await asyncio.to_thread(db.get_conn, ) rows = conn.execute("SELECT * FROM bookings").fetchall() events = {r["id"]: dict(r) for r in conn.execute("SELECT * FROM events").fetchall()} conn.close() @@ -633,14 +642,14 @@ def my_bookings(authorization: str = Header(default="")): @app.get("/api/bookings/by-event/{event_id}") -def my_booking_for_event(event_id: str, authorization: str = Header(default="")): +async def my_booking_for_event(event_id: str, authorization: str = Header(default="")): """查当前登录用户是否已报名该活动及其审核状态(详情页判断用)""" payload = require_auth(authorization) - acct = db.fetch_one("accounts", username=payload["username"]) + acct = await asyncio.to_thread(db.fetch_one, "accounts", username=payload["username"]) ids = {payload["username"]} if acct and acct.get("phone"): ids.add(acct["phone"]) - conn = db.get_conn() + conn = await asyncio.to_thread(db.get_conn, ) rows = conn.execute("SELECT * FROM bookings WHERE event_id=?", (event_id,)).fetchall() conn.close() hit = None @@ -658,11 +667,11 @@ def my_booking_for_event(event_id: str, authorization: str = Header(default="")) async def checkin(req: Request, authorization: str = Header(default="")): payload = require_auth(authorization) b = await req.json() - acct = db.fetch_one("accounts", username=payload["username"]) + acct = await asyncio.to_thread(db.fetch_one, "accounts", username=payload["username"]) ids = {payload["username"]} if acct and acct.get("phone"): ids.add(acct["phone"]) - bk = db.fetch_by_id("bookings", b.get("bookingId", "")) + bk = await asyncio.to_thread(db.fetch_by_id, "bookings", b.get("bookingId", "")) if not bk: raise HTTPException(404, "报名记录不存在") if bk["username"] not in ids and (bk.get("contact") or "") not in ids: @@ -670,7 +679,7 @@ async def checkin(req: Request, authorization: str = Header(default="")): if bk.get("checkin_at"): raise HTTPException(400, "该场次已签到") # 时间闸:只能在「签到开放时间 ≤ now ≤ 结束时间」内签到 - ev = db.fetch_by_id("events", bk.get("event_id", "")) + ev = await asyncio.to_thread(db.fetch_by_id, "events", bk.get("event_id", "")) open_ms = _parse_ms(ev.get("checkin_at") or ev.get("start_at")) if ev else 0 end_ms = (_parse_ms(ev.get("end_at")) if ev and ev.get("end_at") else 0) or (_parse_ms(ev.get("start_at")) + (ev.get("duration_min") or 90) * 60000 if ev else 0) now_ms = int(time.time() * 1000) @@ -678,14 +687,14 @@ async def checkin(req: Request, authorization: str = Header(default="")): raise HTTPException(400, "未到签到时间,活动开始后再签到") if now_ms > end_ms or (ev and ev.get("status") == "done"): raise HTTPException(400, "本场已结束,无法签到") - db.update_row("bookings", bk["id"], {"checkin_at": now_iso()}) - return {"ok": True, "entry": _decode_bk(db.fetch_by_id("bookings", bk["id"]))} + await asyncio.to_thread(db.update_row, "bookings", bk["id"], {"checkin_at": now_iso()}) + return {"ok": True, "entry": _decode_bk(await asyncio.to_thread(db.fetch_by_id, "bookings", bk["id"]))} @app.get("/api/bookings") -def admin_bookings(authorization: str = Header(default=""), status: Optional[str] = None, audit: Optional[str] = None): +async def admin_bookings(authorization: str = Header(default=""), status: Optional[str] = None, audit: Optional[str] = None): require_auth(authorization) - lst = [_decode_bk(r) for r in db.list_all("bookings")] + lst = [_decode_bk(r) for r in await asyncio.to_thread(db.list_all, "bookings")] lst.sort(key=lambda x: x["created_at"], reverse=True) if status: lst = [x for x in lst if x["status"] == status] @@ -698,7 +707,7 @@ def admin_bookings(authorization: str = Header(default=""), status: Optional[str async def update_booking(bid: str, req: Request, authorization: str = Header(default="")): require_auth(authorization) b = await req.json() - if not db.fetch_by_id("bookings", bid): + if not await asyncio.to_thread(db.fetch_by_id, "bookings", bid): raise HTTPException(404, "预约不存在") patch = {} if b.get("status") in ("pending", "confirmed", "arrived", "converted"): @@ -706,20 +715,20 @@ async def update_booking(bid: str, req: Request, authorization: str = Header(def if b.get("auditStatus") in ("pending", "approved", "rejected"): patch["audit_status"] = b["auditStatus"] if patch: - db.update_row("bookings", bid, patch) - return {"ok": True, "entry": _decode_bk(db.fetch_by_id("bookings", bid))} + await asyncio.to_thread(db.update_row, "bookings", bid, patch) + return {"ok": True, "entry": _decode_bk(await asyncio.to_thread(db.fetch_by_id, "bookings", bid))} @app.delete("/api/bookings/{bid}") -def delete_booking(bid: str, authorization: str = Header(default="")): +async def delete_booking(bid: str, authorization: str = Header(default="")): require_auth(authorization) - db.delete_row("bookings", bid) + await asyncio.to_thread(db.delete_row, "bookings", bid) return {"ok": True} # ================= OPC 测评 ================= @app.get("/api/tests/opc/questions") -def opc_questions(version: Optional[str] = "full"): +async def opc_questions(version: Optional[str] = "full"): v = "quick" if version == "quick" else "full" return { "ok": True, "version": v, @@ -751,19 +760,19 @@ async def report_test(req: Request): "type_code": b.get("typeCode"), "persona": b.get("persona", ""), "adapt_index": int(b.get("adaptIndex") or 0), "adapt_level": b.get("adaptLevel", ""), "tracks": json.dumps(b.get("tracks", []), ensure_ascii=False), "version": b.get("version", "")} - db.insert("tests", entry) + await asyncio.to_thread(db.insert, "tests", entry) return {"ok": True, "id": entry["id"]} @app.get("/api/tests") -def admin_tests(authorization: str = Header(default="")): +async def admin_tests(authorization: str = Header(default="")): require_auth(authorization) - return {"ok": True, "list": db.list_all("tests")} + return {"ok": True, "list": await asyncio.to_thread(db.list_all, "tests")} # ================= 政策 / 流程 / 调研:题目下发 + 结果生成(逻辑后置于后端) ================= @app.get("/api/policy/questions") -def policy_questions(): +async def policy_questions(): return {"ok": True, "questions": policy_data.PT_QUESTIONS} @@ -776,7 +785,7 @@ async def policy_calculate(req: Request): @app.get("/api/plan/config") -def plan_config(): +async def plan_config(): return {"ok": True, "regions": policy_data.SP_REGIONS, "status": policy_data.SP_STATUS} @@ -792,7 +801,7 @@ async def plan_generate(req: Request): @app.get("/api/survey/questions") -def survey_questions(): +async def survey_questions(): return {"ok": True, "likert": survey_data.LIKERT_OPTIONS, "sections": survey_data.SURVEY_SECTIONS, "questions": survey_data.SURVEY_QUESTIONS} @@ -804,7 +813,7 @@ async def survey_submit(req: Request): raise HTTPException(400, "缺少调研答案") entry = {"id": db.gen_id("S"), "created_at": now_iso(), "username": str(b.get("username", "")).strip(), "source": str(b.get("source", "")).strip() or "web", "answers": json.dumps(b.get("answers"), ensure_ascii=False)} - db.insert("survey_logs", entry) + await asyncio.to_thread(db.insert, "survey_logs", entry) return {"ok": True, "id": entry["id"], "createdAt": entry["created_at"]} @@ -818,14 +827,14 @@ async def report_policy(req: Request): "answers": json.dumps(b.get("answers"), ensure_ascii=False), "policies_count": int(b.get("policiesCount") or 0), "subsidies_count": int(b.get("subsidiesCount") or 0), "loans_count": int(b.get("loansCount") or 0), "summary": str(b.get("summary", "") or "")[:200]} - db.insert("policy_logs", entry) + await asyncio.to_thread(db.insert, "policy_logs", entry) return {"ok": True, "id": entry["id"]} @app.get("/api/policy-logs") -def admin_policy(authorization: str = Header(default="")): +async def admin_policy(authorization: str = Header(default="")): require_auth(authorization) - return {"ok": True, "list": db.list_all("policy_logs")} + return {"ok": True, "list": await asyncio.to_thread(db.list_all, "policy_logs")} @app.post("/api/plan-logs") @@ -837,14 +846,14 @@ async def report_plan(req: Request): "region": str(b.get("region", "")).strip(), "status": str(b.get("status", "")).strip(), "need_park": 1 if b.get("needPark") else 0, "has_staff": 1 if b.get("hasStaff") else 0, "steps_count": int(b.get("stepsCount") or 0)} - db.insert("plan_logs", entry) + await asyncio.to_thread(db.insert, "plan_logs", entry) return {"ok": True, "id": entry["id"]} @app.get("/api/plan-logs") -def admin_plan(authorization: str = Header(default="")): +async def admin_plan(authorization: str = Header(default="")): require_auth(authorization) - return {"ok": True, "list": db.list_all("plan_logs")} + return {"ok": True, "list": await asyncio.to_thread(db.list_all, "plan_logs")} @app.post("/api/survey-logs") @@ -854,36 +863,36 @@ async def report_survey(req: Request): raise HTTPException(400, "缺少调研答案") entry = {"id": db.gen_id("S"), "created_at": now_iso(), "username": str(b.get("username", "")).strip(), "source": str(b.get("source", "")).strip() or "web", "answers": json.dumps(b.get("answers"), ensure_ascii=False)} - db.insert("survey_logs", entry) + await asyncio.to_thread(db.insert, "survey_logs", entry) return {"ok": True, "id": entry["id"], "createdAt": entry["created_at"]} @app.get("/api/survey-logs") -def admin_survey(authorization: str = Header(default="")): +async def admin_survey(authorization: str = Header(default="")): require_auth(authorization) - return {"ok": True, "list": db.list_all("survey_logs")} + return {"ok": True, "list": await asyncio.to_thread(db.list_all, "survey_logs")} # ================= 运营统计 ================= @app.get("/api/ops/stats") -def ops_stats(authorization: str = Header(default="")): +async def ops_stats(authorization: str = Header(default="")): require_auth(authorization) - conn = db.get_conn() + conn = await asyncio.to_thread(db.get_conn, ) def count(t): return conn.execute(f"SELECT COUNT(*) c FROM {t}").fetchone()["c"] def count_where(t, where, val): return conn.execute(f"SELECT COUNT(*) c FROM {t} WHERE {where}", (val,)).fetchone()["c"] now_ms = int(__import__("time").time() * 1000) - upcoming = sum(1 for e in db.list_all("events") if _parse_ms(e["start_at"]) >= now_ms) - free = sum(1 for e in db.list_all("events") if e["type"] == "free" and _parse_ms(e["start_at"]) >= now_ms) - salon = sum(1 for e in db.list_all("events") if e["type"] == "salon" and _parse_ms(e["start_at"]) >= now_ms) + upcoming = sum(1 for e in await asyncio.to_thread(db.list_all, "events") if _parse_ms(e["start_at"]) >= now_ms) + free = sum(1 for e in await asyncio.to_thread(db.list_all, "events") if e["type"] == "free" and _parse_ms(e["start_at"]) >= now_ms) + salon = sum(1 for e in await asyncio.to_thread(db.list_all, "events") if e["type"] == "salon" and _parse_ms(e["start_at"]) >= now_ms) # 先全部算完再关连接 stats = { "bookings": count("bookings"), - "pending": count_where("bookings", "status=?", "pending"), - "confirmed": count_where("bookings", "status=?", "confirmed"), - "arrived": count_where("bookings", "status=?", "arrived"), - "converted": count_where("bookings", "status=?", "converted"), + "pending": await asyncio.to_thread(count_where, "bookings", "status=?", "pending"), + "confirmed": await asyncio.to_thread(count_where, "bookings", "status=?", "confirmed"), + "arrived": await asyncio.to_thread(count_where, "bookings", "status=?", "arrived"), + "converted": await asyncio.to_thread(count_where, "bookings", "status=?", "converted"), "events": count("events"), "upcomingEvents": upcoming, "free": free, "salon": salon, "tests": count("tests"), "policyLogs": count("policy_logs"), "planLogs": count("plan_logs"), "surveyLogs": count("survey_logs"),