65ec4e6a52
- 将上课日从周六调整为周日，新增 CLASS_WEEKDAY 常量 - 更新 get_class_date 和 get_sunday_of_week 函数以支持周日计算 - 修改 generate_week_list 函数生成周日日期而非周六日期 - 重构月计划和年计划的时间窗口逻辑，增加提前提交功能 - 新增 monthly_window 和 yearly_window 函数定义准确的提交窗口期 - 调整默认周期选择逻辑，支持跨月年的时间窗口 - 添加 period_labels 字段提供前端周期标签显示 - 移除旧的 _validate_current_period 函数 - 完善计划提交验证逻辑，支持多种时间周期类型 chore(miniapp): 更新小程序项目配置 - 补充 project.config.json 中的设置选项 - 调整 project.private.config.json 配置结构 - 移除冗余的页面调试条件配置 - 更新编辑器和编译相关配置项
374 lines
12 KiB
Python
374 lines
12 KiB
Python
import logging
|
||
import re
|
||
from datetime import date, datetime, timedelta
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException
|
||
from pydantic import BaseModel
|
||
|
||
from .. import db
|
||
from .auth import get_current_user
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/api/v1/checkin", tags=["checkin"])
|
||
|
||
CHECKIN_STATUSES = {"正常", "迟到", "请假", "不参加讨论", "早退"}
|
||
DEFAULT_STATUS = "正常"
|
||
MAX_NOTE_LEN = 500
|
||
DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
|
||
SELECT_COLS = "id,user_id,class_date,checked_at,status,note"
|
||
|
||
# 周期列表起始年份
|
||
PERIOD_START_YEAR = 2022
|
||
|
||
# 上课日：每周周日（date.weekday(): 周一=0 ... 周六=5, 周日=6）
|
||
CLASS_WEEKDAY = 6
|
||
|
||
|
||
def get_class_date(ref_date=None):
|
||
"""获取指定日期所在自然周的周日日期（上课日）。"""
|
||
today = ref_date or date.today()
|
||
return today + timedelta(days=(CLASS_WEEKDAY - today.weekday()))
|
||
|
||
|
||
def get_sunday_of_week(week_monday: date) -> date:
|
||
"""给定某周周一，返回该周周日（上课日）。"""
|
||
return week_monday + timedelta(days=CLASS_WEEKDAY)
|
||
|
||
|
||
def generate_week_list(start_year: int = PERIOD_START_YEAR) -> list:
|
||
"""
|
||
生成从 start_year 到今天的所有上课周（每周日）列表，倒序排列。
|
||
返回格式: ["2026-09-20", "2026-09-13", ...]
|
||
"""
|
||
today = date.today()
|
||
# 找到 start_year 第一个周日（1月1日之后的第一个周日）
|
||
d = date(start_year, 1, 1)
|
||
days_until_sunday = (CLASS_WEEKDAY - d.weekday()) % 7
|
||
first_sunday = d + timedelta(days=days_until_sunday)
|
||
|
||
weeks = []
|
||
current = first_sunday
|
||
while current <= today + timedelta(days=7): # 多生成一周
|
||
weeks.append(current.strftime("%Y-%m-%d"))
|
||
current += timedelta(days=7)
|
||
|
||
weeks.reverse() # 倒序，最新的在前
|
||
return weeks
|
||
|
||
|
||
def _is_valid_date(value):
|
||
if not isinstance(value, str) or not DATE_RE.match(value):
|
||
return False
|
||
try:
|
||
return datetime.strptime(value, "%Y-%m-%d").date().isoformat() == value
|
||
except ValueError as exc:
|
||
logger.warning("日期校验失败: %s", exc)
|
||
return False
|
||
|
||
|
||
def fmt_dt(value):
|
||
if isinstance(value, datetime):
|
||
return value.strftime("%Y-%m-%d %H:%M:%S")
|
||
return str(value) if value else ""
|
||
|
||
|
||
def fmt_rec(rec):
|
||
return {
|
||
"id": rec["id"],
|
||
"user_id": rec["user_id"],
|
||
"class_date": str(rec["class_date"]),
|
||
"checked_at": fmt_dt(rec.get("checked_at")),
|
||
"status": rec.get("status", DEFAULT_STATUS),
|
||
"note": rec.get("note", ""),
|
||
}
|
||
|
||
|
||
def _is_teacher(user):
|
||
"""判断当前用户是否为老师角色。"""
|
||
return (user.get("role") or "").strip().lower() == "teacher"
|
||
|
||
|
||
def _groups_stats(class_date, group_filter=None):
|
||
"""
|
||
统计各小组在指定上课日的打卡情况。
|
||
返回 [{group_name, checked_count, total}]，按 group_name 排序。
|
||
group_filter: 只统计指定小组
|
||
"""
|
||
where_clauses = []
|
||
params = []
|
||
|
||
if group_filter:
|
||
where_clauses.append("m.`group` = %s")
|
||
params.append(group_filter)
|
||
else:
|
||
where_clauses.append("m.`group` IS NOT NULL AND m.`group` != ''")
|
||
|
||
where_sql = ("WHERE " + " AND ".join(where_clauses)) if where_clauses else ""
|
||
|
||
rows = db.fetch_all(
|
||
"SELECT m.`group` AS group_name, "
|
||
" COUNT(*) AS total, "
|
||
" SUM(CASE WHEN r.id IS NOT NULL THEN 1 ELSE 0 END) AS checked_count "
|
||
"FROM auth_users m "
|
||
"LEFT JOIN checkin_records_old r "
|
||
" ON r.user_id = m.id AND r.class_date = %s "
|
||
+ where_sql + " "
|
||
"GROUP BY m.`group` "
|
||
"ORDER BY m.`group`",
|
||
(class_date, *params),
|
||
)
|
||
return [
|
||
{
|
||
"group_name": r["group_name"] or "",
|
||
"total": r["total"] or 0,
|
||
"checked_count": r["checked_count"] or 0,
|
||
}
|
||
for r in rows
|
||
]
|
||
|
||
|
||
class CheckinRequest(BaseModel):
|
||
status: str = DEFAULT_STATUS
|
||
class_date: str = ""
|
||
note: str = ""
|
||
|
||
|
||
@router.get("/weeks")
|
||
def weeks(user=Depends(get_current_user)):
|
||
"""
|
||
返回所有上课周列表（每周日日期），从 2022 年到当前，倒序排列。
|
||
用于前端做历史周切换选择器。
|
||
"""
|
||
week_list = generate_week_list()
|
||
current = str(get_class_date())
|
||
return {
|
||
"weeks": week_list,
|
||
"current_week": current,
|
||
}
|
||
|
||
|
||
@router.get("/week")
|
||
def week(class_date: str = "", user=Depends(get_current_user)):
|
||
"""
|
||
[已弃用] 获取指定上课日的个人打卡状态。
|
||
请使用 /groups 接口，其返回值已包含 checked + record。
|
||
保留此接口仅为兼容旧客户端。
|
||
"""
|
||
if class_date:
|
||
class_date = class_date.strip()
|
||
if not _is_valid_date(class_date):
|
||
raise HTTPException(status_code=400, detail="无效的上课日期")
|
||
else:
|
||
class_date = str(get_class_date())
|
||
rec = db.fetch_one(
|
||
"SELECT " + SELECT_COLS + " FROM checkin_records_old WHERE user_id=%s AND class_date=%s",
|
||
(user["id"], class_date),
|
||
)
|
||
return {"class_date": class_date, "checked": rec is not None, "record": fmt_rec(rec) if rec else None}
|
||
|
||
|
||
@router.get("/groups")
|
||
def groups_view(class_date: str = "", user=Depends(get_current_user)):
|
||
"""
|
||
老师：返回所有小组统计 + 个人状态
|
||
学员：返回自己所在小组统计 + 个人状态
|
||
"""
|
||
if class_date:
|
||
class_date = class_date.strip()
|
||
if not _is_valid_date(class_date):
|
||
raise HTTPException(status_code=400, detail="无效的上课日期")
|
||
else:
|
||
class_date = str(get_class_date())
|
||
|
||
# 个人打卡状态
|
||
rec = db.fetch_one(
|
||
"SELECT " + SELECT_COLS + " FROM checkin_records_old WHERE user_id=%s AND class_date=%s",
|
||
(user["id"], class_date),
|
||
)
|
||
my_checked = rec is not None
|
||
my_record = fmt_rec(rec) if rec else None
|
||
|
||
is_teacher = _is_teacher(user)
|
||
my_group = (user.get("group") or "").strip()
|
||
|
||
if is_teacher:
|
||
groups_list = _groups_stats(class_date)
|
||
# 自己所在小组排最前
|
||
if my_group:
|
||
mine = [g for g in groups_list if g["group_name"] == my_group]
|
||
others = [g for g in groups_list if g["group_name"] != my_group]
|
||
groups_list = mine + others
|
||
else:
|
||
# 学员只看自己小组
|
||
if not my_group:
|
||
groups_list = []
|
||
else:
|
||
groups_list = _groups_stats(class_date, group_filter=my_group)
|
||
|
||
# 未分组人数（仅老师可见）
|
||
unassigned_count = 0
|
||
if is_teacher:
|
||
unassigned_row = db.fetch_one(
|
||
"SELECT COUNT(*) AS cnt FROM auth_users WHERE `group` IS NULL OR `group` = ''"
|
||
)
|
||
unassigned_count = unassigned_row["cnt"] if unassigned_row else 0
|
||
|
||
return {
|
||
"class_date": class_date,
|
||
"checked": my_checked,
|
||
"record": my_record,
|
||
"is_teacher": is_teacher,
|
||
"my_group": my_group,
|
||
"groups": groups_list,
|
||
"unassigned_count": unassigned_count,
|
||
}
|
||
|
||
|
||
def _can_checkin_date(class_date_str):
|
||
"""判断指定上课日是否还可以打卡。
|
||
|
||
规则：上课日所在自然周（周一~周日）与当前周相同，或为未来周，则可以打卡。
|
||
上课日是周日，同一自然周内（周一到周日）都可以打卡、可以修改。
|
||
"""
|
||
try:
|
||
class_date = datetime.strptime(class_date_str, "%Y-%m-%d").date()
|
||
except ValueError:
|
||
return False, "无效的日期"
|
||
today = date.today()
|
||
# 找到上课日所在周的周一
|
||
class_monday = class_date - timedelta(days=class_date.isoweekday() - 1)
|
||
# 找到今天所在周的周一
|
||
today_monday = today - timedelta(days=today.isoweekday() - 1)
|
||
if class_monday < today_monday:
|
||
return False, "已过打卡时间"
|
||
return True, ""
|
||
|
||
|
||
@router.post("")
|
||
def checkin(req: CheckinRequest, user=Depends(get_current_user)):
|
||
status = req.status.strip()
|
||
if status not in CHECKIN_STATUSES:
|
||
raise HTTPException(status_code=400, detail="无效的打卡状态")
|
||
|
||
class_date = req.class_date.strip() or str(get_class_date())
|
||
if not _is_valid_date(class_date):
|
||
raise HTTPException(status_code=400, detail="无效的上课日期")
|
||
|
||
can_submit, msg = _can_checkin_date(class_date)
|
||
if not can_submit:
|
||
raise HTTPException(status_code=400, detail=msg)
|
||
|
||
note = req.note.strip() if status != DEFAULT_STATUS else ""
|
||
if len(note) > MAX_NOTE_LEN:
|
||
raise HTTPException(status_code=400, detail="备注过长")
|
||
|
||
now = datetime.now()
|
||
# 根据 user_id 判断当天是否已打卡：已打卡 → 更新；未打卡 → 新增
|
||
existing = db.fetch_one(
|
||
"SELECT id FROM checkin_records_old WHERE user_id=%s AND class_date=%s",
|
||
(user["id"], class_date),
|
||
)
|
||
if existing:
|
||
db.execute(
|
||
"UPDATE checkin_records_old SET checked_at=%s, status=%s, note=%s WHERE id=%s",
|
||
(now, status, note, existing["id"]),
|
||
)
|
||
record_id = existing["id"]
|
||
else:
|
||
record_id = db.execute(
|
||
"INSERT INTO checkin_records_old (user_id, class_date, checked_at, status, note) "
|
||
"VALUES (%s, %s, %s, %s, %s)",
|
||
(user["id"], class_date, now, status, note),
|
||
)
|
||
# 直接构造返回，省一次 SELECT
|
||
record = {
|
||
"id": record_id,
|
||
"user_id": user["id"],
|
||
"class_date": class_date,
|
||
"status": status,
|
||
"checked_at": fmt_dt(now),
|
||
"note": note,
|
||
}
|
||
return {"success": True, "record": record}
|
||
|
||
|
||
@router.get("/list")
|
||
def list_records(page: int = 1, size: int = 10, user=Depends(get_current_user)):
|
||
page = max(1, page)
|
||
size = max(1, min(50, size))
|
||
offset = (page - 1) * size
|
||
total_row = db.fetch_one(
|
||
"SELECT COUNT(*) AS cnt FROM checkin_records_old WHERE user_id=%s", (user["id"],)
|
||
)
|
||
total = total_row["cnt"] if total_row else 0
|
||
rows = db.fetch_all(
|
||
"SELECT " + SELECT_COLS + " FROM checkin_records_old "
|
||
"WHERE user_id=%s ORDER BY class_date DESC LIMIT %s OFFSET %s",
|
||
(user["id"], size, offset),
|
||
)
|
||
return {"list": [fmt_rec(r) for r in rows], "total": total, "page": page, "size": size}
|
||
|
||
|
||
@router.get("/group")
|
||
def group_view(class_date: str = "", group_name: str = "", user=Depends(get_current_user)):
|
||
"""
|
||
查看某小组的打卡成员列表。
|
||
- 老师：可以传 group_name 查看任意小组；不传则看自己所在小组
|
||
- 学员：只能看自己所在小组，传其他小组名返回 403
|
||
"""
|
||
class_date = class_date.strip() or str(get_class_date())
|
||
if not _is_valid_date(class_date):
|
||
raise HTTPException(status_code=400, detail="无效的上课日期")
|
||
|
||
my_group = (user.get("group") or "").strip()
|
||
is_teacher = _is_teacher(user)
|
||
|
||
if group_name:
|
||
group_name = group_name.strip()
|
||
# 权限校验：学员只能看自己小组
|
||
if not is_teacher and group_name != my_group:
|
||
raise HTTPException(status_code=403, detail="无权限查看其他小组")
|
||
else:
|
||
group_name = my_group
|
||
|
||
if not group_name:
|
||
return {"class_date": class_date, "group_name": "", "members": []}
|
||
|
||
# 校验小组是否存在
|
||
g_row = db.fetch_one("SELECT `group` FROM auth_users WHERE `group`=%s LIMIT 1", (group_name,))
|
||
if not g_row:
|
||
return {"class_date": class_date, "group_name": group_name, "members": []}
|
||
# 单查询：成员 + 同组成员该上课日的打卡记录，LEFT JOIN 一次取回
|
||
rows = db.fetch_all(
|
||
"SELECT m.id AS user_id, m.student_id, m.name, "
|
||
" r.id AS record_id, r.status, r.checked_at, r.note "
|
||
"FROM auth_users m "
|
||
"LEFT JOIN checkin_records_old r "
|
||
" ON r.user_id = m.id AND r.class_date = %s "
|
||
"WHERE m.`group` = %s "
|
||
"ORDER BY m.student_id",
|
||
(class_date, group_name),
|
||
)
|
||
|
||
result_members = []
|
||
for row in rows:
|
||
rec = None
|
||
if row.get("status") is not None:
|
||
rec = {
|
||
"id": row["record_id"],
|
||
"user_id": row["user_id"],
|
||
"status": row["status"],
|
||
"checked_at": fmt_dt(row.get("checked_at")),
|
||
"note": row.get("note", ""),
|
||
}
|
||
result_members.append({
|
||
"user_id": row["user_id"],
|
||
"student_id": row["student_id"],
|
||
"name": row["name"],
|
||
"checked": rec is not None,
|
||
"record": rec,
|
||
})
|
||
return {"class_date": class_date, "group_name": group_name, "members": result_members}
|