Commit 65250db5 authored by uuo00_n's avatar uuo00_n

feat(仪表盘): 添加后端仪表盘功能模块

添加仪表盘路由和服务模块,包含学生今日摘要、班主任当前课程摘要、部门概览和校园概览功能
修复初始化数据库脚本中未关联学生与用户的问题
恢复之前移除的仪表盘路由配置
parent 6bc71b9a
home = /Applications/ServBay/package/python/3.9/3.9.21/bin
home = /Users/uu/Desktop/dles_prj/llm-filter/.venv/bin
include-system-site-packages = false
version = 3.9.21
from fastapi import APIRouter, Depends
from app.api.deps import require_edition_for_mode, require_role
from app.services.dashboard import (
student_today_summary,
homeroom_current_summary,
department_overview,
campus_overview,
)
router = APIRouter(dependencies=[Depends(require_edition_for_mode())])
@router.get("/student/today")
async def student_today(current_user: dict = Depends(require_role(1))):
return await student_today_summary(current_user)
@router.get("/homeroom/current")
async def homeroom_current(current_user: dict = Depends(require_role(2))):
return await homeroom_current_summary(current_user)
@router.get("/department/overview")
async def department(current_user: dict = Depends(require_role(3))):
return await department_overview(current_user)
@router.get("/campus/overview")
async def campus(current_user: dict = Depends(require_role(5))):
return await campus_overview(current_user)
\ No newline at end of file
from fastapi import APIRouter
from app.api.v1 import auth, conversation, admin
from app.api.v1 import auth, conversation, admin, dashboard
api_router = APIRouter()
......@@ -7,4 +7,4 @@ api_router = APIRouter()
api_router.include_router(auth.router, prefix="/auth", tags=["认证"])
api_router.include_router(conversation.router, prefix="/conversations", tags=["对话"])
api_router.include_router(admin.router, prefix="/admin", tags=["管理员"])
# 需求变更:移除后端仪表盘路由,仪表盘由前端实现,无需后端接口。
\ No newline at end of file
api_router.include_router(dashboard.router, prefix="/dashboard", tags=["仪表盘"])
\ No newline at end of file
from datetime import datetime
from typing import Dict, List, Any
from bson import ObjectId
from app.db.mongodb import db
async def _today_iso() -> str:
return datetime.now().date().isoformat()
async def _weekday() -> int:
return datetime.now().isoweekday()
async def _current_period() -> int:
h = datetime.now().hour
if h < 10:
return 1
if h < 11:
return 2
if h < 12:
return 3
return 4
async def _get_student_by_user(user_id: ObjectId) -> Dict[str, Any]:
s = await db.db.students.find_one({"user_id": user_id})
if s:
return s
return await db.db.students.find_one({})
async def student_today_summary(current_user: Dict[str, Any]) -> Dict[str, Any]:
today = await _today_iso()
weekday = await _weekday()
student = await _get_student_by_user(ObjectId(current_user["_id"]))
class_id = student.get("class_id") if student else None
schedules: List[Dict[str, Any]] = []
if class_id:
cursor = db.db.schedules.find({"weekday": weekday, "classes.class_id": class_id}).sort("period", 1)
async for doc in cursor:
loc = None
for c in doc.get("classes", []):
if c.get("class_id") == class_id:
loc = c.get("location")
break
schedules.append({
"lesson_id": doc.get("lesson_id"),
"period": doc.get("period"),
"course_name": doc.get("course_name"),
"location": loc,
})
attendance = []
if student:
cursor = db.db.attendance.find({"student_id": student.get("student_id"), "date": today})
async for a in cursor:
attendance.append({
"lesson_id": a.get("lesson_id"),
"status": a.get("status"),
})
conduct = await db.db.conduct.find_one({"student_id": student.get("student_id") if student else None, "date": today})
return {
"student": {
"student_id": student.get("student_id") if student else None,
"name": student.get("name") if student else None,
"class_id": class_id,
},
"today_schedule": schedules,
"today_attendance": attendance,
"today_conduct": conduct or {},
}
async def homeroom_current_summary(current_user: Dict[str, Any]) -> Dict[str, Any]:
today = await _today_iso()
weekday = await _weekday()
period = await _current_period()
uid = ObjectId(current_user["_id"])
classes = []
cursor = db.db.classes.find({"head_teacher_id": uid})
async for c in cursor:
classes.append(c)
class_ids = [c.get("class_id") for c in classes]
lessons: List[Dict[str, Any]] = []
for cid in class_ids:
doc = await db.db.schedules.find_one({"weekday": weekday, "period": period, "classes.class_id": cid})
if doc:
loc = None
for cc in doc.get("classes", []):
if cc.get("class_id") == cid:
loc = cc.get("location")
break
lessons.append({
"class_id": cid,
"course_name": doc.get("course_name"),
"location": loc,
})
rates: List[Dict[str, Any]] = []
for c in classes:
total = c.get("students_count", 0)
present = await db.db.attendance.count_documents({"class_id": c.get("class_id"), "date": today, "status": "出勤"})
rate = (present / total) if total else 0
rates.append({"class_id": c.get("class_id"), "present": present, "total": total, "rate": rate})
leaves: List[Dict[str, Any]] = []
for cid in class_ids:
cursor = db.db.leaves.find({"class_id": cid, "from_date": {"$lte": today}, "to_date": {"$gte": today}})
async for l in cursor:
leaves.append({
"student_id": l.get("student_id"),
"class_id": cid,
"reason": l.get("reason"),
"status": l.get("status"),
})
directives: List[Dict[str, Any]] = []
cursor = db.db.directives.find({"level": "department"}).sort("created_at", -1).limit(5)
async for d in cursor:
directives.append({
"content": d.get("content"),
"created_at": d.get("created_at"),
})
return {
"current_lessons": lessons,
"attendance_rates": rates,
"leaves": leaves,
"directives": directives,
}
async def department_overview(current_user: Dict[str, Any]) -> Dict[str, Any]:
today = await _today_iso()
weekday = await _weekday()
total_students = await db.db.students.count_documents({})
present = await db.db.attendance.count_documents({"date": today, "status": "出勤"})
absent = await db.db.attendance.count_documents({"date": today, "status": {"$in": ["缺勤", "请假"]}})
ratio = (present / total_students) if total_students else 0
anomalies: List[Dict[str, Any]] = []
cursor = db.db.classes.find({})
async for c in cursor:
cid = c.get("class_id")
total = c.get("students_count", 0)
a = await db.db.attendance.count_documents({"class_id": cid, "date": today, "status": {"$in": ["缺勤", "请假"]}})
r = (a / total) if total else 0
if r > 0.3:
anomalies.append({"class_id": cid, "anomaly_rate": r})
teachers: Dict[str, int] = {}
cursor = db.db.schedules.find({"weekday": weekday})
async for s in cursor:
tid = s.get("teacher_id")
if tid:
key = str(tid)
teachers[key] = teachers.get(key, 0) + 1
return {
"student_present_ratio": ratio,
"anomalies": anomalies,
"teacher_timeslots": teachers,
}
async def campus_overview(current_user: Dict[str, Any]) -> Dict[str, Any]:
today = await _today_iso()
total_students = await db.db.students.count_documents({})
present = await db.db.attendance.count_documents({"date": today, "status": "出勤"})
leaves = await db.db.leaves.count_documents({"from_date": {"$lte": today}, "to_date": {"$gte": today}})
directives_count = await db.db.directives.count_documents({})
return {
"total_students": total_students,
"present": present,
"leaves": leaves,
"directives": directives_count,
"term_goals": [],
"department_progress": [],
}
\ No newline at end of file
......@@ -403,6 +403,9 @@ async def seed_school_data(db, mode: str):
return docs
students_docs = gen_students("SW22-1", 12) + gen_students("SW22-2", 12)
sample_user = await db.users.find_one({"username": "user", "edition": mode})
if sample_user and students_docs:
students_docs[0]["user_id"] = sample_user["_id"]
await db.students.insert_many(students_docs)
# 更新班级人数
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment