feat(init): 添加完整的认证和业务路由模块

- 实现微信登录、手机号绑定、token验证和登出功能
- 添加打卡系统,支持正常、迟到、请假等状态记录
- 集成课程投票功能,支持多课程选择和评分评论
- 开发反馈系统,支持图文反馈提交和管理员处理
- 构建首页页面模块,动态展示学习资源和个人信息
- 设计计划管理系统,支持周/月/年计划制定和跟踪
- 实现小组计划统计和历史记录查询功能
- 完善用户认证会话管理和数据库交互逻辑
This commit is contained in:
王宏豪
2026-09-03 16:43:11 +08:00
commit f3fbd0377a
24 changed files with 1918 additions and 0 deletions

97
db.py Normal file
View File

@@ -0,0 +1,97 @@
"""Database access with a PyMySQL connection pool.
Each fetch/execute borrows a pooled connection and returns it on close().
Pool settings are read from environment (DB_POOL_*), so nothing here depends
on config.py and the pool can be shared safely across threads via DBUtils.
"""
import os
import logging
import threading
import pymysql
from dbutils.pooled_db import PooledDB
from . import config
logger = logging.getLogger(__name__)
def _pool_args(database):
return dict(
mincached=int(os.environ.get("DB_POOL_MIN_CACHED", "1")),
maxcached=int(os.environ.get("DB_POOL_MAX_CACHED", "5")),
maxconnections=int(os.environ.get("DB_POOL_MAX_CONNECTIONS", "10")),
blocking=True,
maxusage=int(os.environ.get("DB_POOL_MAX_USAGE", "100")),
ping=1,
host=config.MYSQL_HOST,
port=config.MYSQL_PORT,
user=config.MYSQL_USER,
password=config.MYSQL_PASSWORD,
database=database,
charset="utf8mb4",
connect_timeout=5,
read_timeout=5,
)
_pools = {}
_pools_lock = threading.Lock()
def _get_pool(database):
key = database or config.MYSQL_DB
with _pools_lock:
pool = _pools.get(key)
if pool is None:
pool = PooledDB(creator=pymysql, **_pool_args(key))
_pools[key] = pool
return pool
def _connect(db=None):
"""Borrow a pooled connection (close() returns it to the pool)."""
return _get_pool(db).connection()
def fetch_one(sql, args=None, db=None):
conn = _connect(db)
try:
with conn.cursor(pymysql.cursors.DictCursor) as cur:
cur.execute(sql, args or ())
return cur.fetchone()
finally:
conn.close()
def fetch_all(sql, args=None, db=None):
conn = _connect(db)
try:
with conn.cursor(pymysql.cursors.DictCursor) as cur:
cur.execute(sql, args or ())
return cur.fetchall()
finally:
conn.close()
def execute(sql, args=None, db=None):
conn = _connect(db)
try:
with conn.cursor() as cur:
cur.execute(sql, args or ())
lastrowid = cur.lastrowid
conn.commit()
return lastrowid
finally:
conn.close()
def close_all():
"""Close every pooled connection (used by tests / reload hooks)."""
with _pools_lock:
for key in list(_pools):
try:
_pools.pop(key).close()
except Exception:
logger.exception("关闭数据库连接池失败: %s", key)