Python 后端提交

Python 后端（FastAPI + FastMCP + ...）的初始版本号设定为 0.1.0，这是 uv 在 pypriject.toml
里给我自动设置的，我觉得有道理。

router/__init__.py

@@ -0,0 +1 @@
+from .__version__ import __version__

router/__version__.py

@@ -0,0 +1,6 @@
+from importlib.metadata import PackageNotFoundError, version
+
+try:
+    __version__ = version("njupt-suan-api")
+except PackageNotFoundError:
+    __version__ = "dev"

router/admin_router.py

@@ -0,0 +1,123 @@
+from pathlib import Path
+from typing import Annotated, Sequence
+
+import aiofiles
+from fastapi import APIRouter, Depends
+from pydantic import BaseModel
+from sqlmodel import Session, delete, select
+
+from njupt_api.baselib import config, logger
+from njupt_api.zhengfang import ZhengFang, course_list_serializer
+from router.enhance.auth import verify_token
+from router.enhance.lib import AliasDto, ReturnDto, TestDto, get_session
+from router.enhance.model import Alias, Course
+
+
+class ValidateTokenDto(BaseModel):
+    token: str
+
+
+admin_router = APIRouter(prefix="/admin", tags=["admin"])
+
+
+@admin_router.post("/validateToken")
+async def validate_token(vtd: ValidateTokenDto) -> ReturnDto:
+    """
+    验证 Token 是否正确，以此判断是否允许登录 WebUI。
+    验证时无需使用 HTTP Bearer，直接作为 body 传入即可。
+
+    Returns:
+        ReturnDto，以 success 字段表明是否有效。
+    """
+    async with aiofiles.open(file=Path.cwd() / "data/token.txt", mode="r") as f:
+        if (await f.readline()).strip() == vtd.token:
+            return ReturnDto(success=True)
+        return ReturnDto(success=False)
+
+
+@admin_router.post("/schedule/test", dependencies=[Depends(verify_token)])
+async def post_schedule_test(test: TestDto, session: Annotated[Session, Depends(get_session)]) -> ReturnDto:
+    async with ZhengFang() as zf:
+        if await zf.login(test.username, test.password):
+            if test.scheduleType == "class":
+                final_course_list = course_list_serializer(
+                    await zf.get_class_schedule(),
+                )
+                session.exec(delete(Course))
+                for course in final_course_list:
+                    session.add(Course(**course))
+                session.commit()
+
+                logger.success(
+                    f"{test.username} | 获取 {test.scheduleType} 课表成功，已保存到数据库。",
+                )
+                return ReturnDto(success=True, result=final_course_list)
+            if test.scheduleType == "student":
+                final_course_list = course_list_serializer(
+                    await zf.get_student_schedule(),
+                )
+                logger.success(
+                    f"{test.username} | 获取 {test.scheduleType} 课表成功。个人课表不保存。",
+                )
+                return ReturnDto(success=True, result=final_course_list)
+            logger.error(
+                f"{test.username} | scheduleType 参数错误。给定的 schedule={test.scheduleType}",
+            )
+            return ReturnDto(
+                success=False,
+                message="参数错误，请检查 scheduleType 参数。",
+            )
+        logger.error(
+            f"{test.username} | 获取课程表失败，请检查账号密码是否正确后再试。",
+        )
+        return ReturnDto(
+            success=False,
+            message="获取课程表失败，请检查账号密码是否正确后再试。",
+        )
+
+
+@admin_router.get("/schedule/test", dependencies=[Depends(verify_token)])
+async def get_schedule_test(session: Annotated[Session, Depends(get_session)]) -> ReturnDto:
+    course_dtos: Sequence[Course] = session.exec(select(Course)).all()
+    return ReturnDto(
+        success=True,
+        result=[course.model_dump() for course in course_dtos],
+    )
+
+
+@admin_router.post("/schedule/alias", dependencies=[Depends(verify_token)])
+async def post_schedule_alias(alias: AliasDto, session: Annotated[Session, Depends(get_session)]) -> ReturnDto:
+    for alia in session.exec(select(Alias)).all():
+        if alias.originalName == alia.originalName:
+            logger.error(
+                f"课程 {alia.originalName} 已经在数据库中存在，不允许重复添加。",
+            )
+            return ReturnDto(
+                success=False,
+                message=f"课程 {alia.originalName} 已经在数据库中存在，不允许重复添加。",
+            )
+
+    session.add(Alias(originalName=alias.originalName, aliasName=alias.aliasName))
+    session.commit()
+    logger.success(f"已添加课程别名 | {alias.originalName} => {alias.aliasName}")
+    return ReturnDto(success=True)
+
+
+@admin_router.get("/schedule/alias", dependencies=[Depends(verify_token)])
+async def get_schedule_alias(session: Annotated[Session, Depends(get_session)]) -> ReturnDto:
+    aliases: Sequence[Alias] = session.exec(select(Alias)).all()
+    return ReturnDto(success=True, result=[alias.model_dump() for alias in aliases])
+
+
+@admin_router.post("/config", dependencies=[Depends(verify_token)])
+async def post_config(data: dict) -> ReturnDto:
+    data_ = data.get("data")
+    logger.debug(f"接收到配置字典 - {data_}")
+    config.from_dict(data_)
+    await config.save_json()
+    return ReturnDto(success=True, result=config.to_dict())
+
+
+@admin_router.get("/config", dependencies=[Depends(verify_token)])
+async def get_config() -> ReturnDto:
+    return ReturnDto(success=True, result=config.to_dict())

router/api_router.py

@@ -0,0 +1,100 @@
+from pathlib import Path
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.responses import FileResponse
+from sqlmodel import Session, select
+
+from njupt_api.baselib import logger
+from njupt_api.zhengfang import (
+    ZhengFang,
+    course_dict_serializer,
+    course_list_serializer,
+)
+from router.enhance.lib import ReturnDto, ScheduleQueryDto, apply_enhance, get_session
+from router.enhance.model import Course
+
+TEMP_DIR = Path.cwd() / "temp"
+
+
+api_router = APIRouter(prefix="/api", tags=["API"])
+
+
+@api_router.post("/schedule/class")
+async def post_schedule_class(
+    student: ScheduleQueryDto,
+    session: Annotated[Session, Depends(get_session)],
+) -> ReturnDto:
+    if student.username is None and student.password is None:
+        logger.debug("未提供学号和密码参数，尝试从数据库中返回一次性存储的班级课表。")
+        course_dtos = session.exec(select(Course)).all()
+        course_list: list[dict] = [course_dict_serializer(course) for course in course_dtos]
+        logger.success(f"{student.week=} 从数据库中返回一次性存储的班级课表。")
+        return await apply_enhance(course_list, student.week, student.img)
+    if student.username and student.password:
+        async with ZhengFang() as zf:
+            if await zf.login(student.username, student.password):
+                course_list = course_list_serializer(await zf.get_class_schedule())
+                logger.success(
+                    f"{student.username} | {student.week=} 获取指定学生的班级课表成功。",
+                )
+                return await apply_enhance(course_list, student.week, student.img)
+            logger.error(
+                f"{student.username} | 获取课程表失败，请检查账号密码是否正确后再试。",
+            )
+            return ReturnDto(
+                success=False,
+                message="获取课程表失败，请检查账号密码是否正确后再试。",
+            )
+    else:
+        logger.error(
+            f"参数错误，请同时携带或同时不携带学号和密码参数: {student.username=} | {student.password=}",
+        )
+        return ReturnDto(
+            success=False,
+            message="参数错误，请同时携带或同时不携带学号和密码参数。",
+        )
+
+
+@api_router.post("/schedule/student")
+async def post_schedule_student(student: ScheduleQueryDto) -> ReturnDto:
+    if student.username is None or student.password is None:
+        logger.error("查询学生课表需要同时提供学号和密码参数。")
+        return ReturnDto(
+            success=False,
+            message="查询学生课表需要同时提供学号和密码参数。",
+        )
+
+    async with ZhengFang() as zf:
+        if await zf.login(student.username, student.password):
+            course_list = course_list_serializer(await zf.get_student_schedule())
+            logger.success(f"{student.username} | 获取学生个人课表成功。")
+            return await apply_enhance(course_list, student.week, student.img)
+        logger.error(
+            f"{student.username} | 获取课程表失败，请检查账号密码是否正确后再试。",
+        )
+        return ReturnDto(
+            success=False,
+            message="获取课程表失败，请检查账号密码是否正确后再试。",
+        )
+
+
+@api_router.get("/schedule/img/{name}")
+async def get_schedule_img(name: str) -> FileResponse:
+    """
+    从 temp 工作目录中读取指定图片并返回。如果图片不存在则报 404。
+
+    Returns:
+        FileResponse: 图片。
+
+    Raises:
+        HTTPException: 404 - 查找的图片不存在。
+    """
+    image_file = TEMP_DIR / name
+    logger.debug(f"尝试获取 {image_file!s}")
+    if image_file.exists():
+        return FileResponse(path=str(image_file), media_type="image/png")
+    raise HTTPException(
+        status_code=status.HTTP_404_NOT_FOUND,
+        detail="Name wrong or too late.",
+    )

router/enhance/alias.py

@@ -0,0 +1,41 @@
+"""为课程提供一个简短的别名，便于在空间有限的课程表图片中辨认。
+
+别名是单独保存的，在最终输出阶段才会被装饰在原有的课表输出上。
+"""
+
+from typing import Sequence
+
+from sqlmodel import Session, select
+
+from njupt_api.baselib import logger
+from router.enhance.model import Alias, engine
+
+
+def apply_alias(courses: list[dict]) -> list[dict]:
+    with Session(engine) as session:
+        aliases: Sequence[Alias] = session.exec(select(Alias)).all()
+
+        # 否则不做任何更改
+        if len(aliases) == 0:
+            return courses
+
+    alias_count = 0
+    apply_count = 0
+
+    alias_dict = {}
+    for alias in aliases:
+        m = alias.model_dump()
+        alias_dict[m["originalName"]] = m["aliasName"]
+        alias_count += 1
+
+    for course in courses:
+        if course["name"] in alias_dict:
+            course["alias"] = alias_dict[course["name"]]
+            apply_count += 1
+        else:
+            course["alias"] = None
+
+    logger.debug(
+        f"课程别名 | 将 {alias_count} 个别名应用在了 {apply_count} 门输出的课程上。",
+    )
+    return courses

router/enhance/auth.py

@@ -0,0 +1,21 @@
+from pathlib import Path
+from typing import Annotated
+
+import aiofiles
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+
+security = HTTPBearer()
+
+
+async def verify_token(credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)]) -> str:
+    token = credentials.credentials
+
+    async with aiofiles.open(file=Path.cwd() / "data" / "token.txt", mode="r") as f:
+        if (await f.readline()).strip() == token:
+            return token
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid or missing token. (Suan API WebUI)",
+            headers={"WWW-Authenticate": "Bearer"},
+        )

router/enhance/lib.py

@@ -0,0 +1,77 @@
+from datetime import date, timedelta
+from typing import Any, Generator, Literal
+
+from pydantic import BaseModel
+from sqlmodel import Session
+
+from njupt_api.baselib import config
+
+from .alias import apply_alias
+from .model import engine
+from .screenshot import generate_img
+
+
+class ScheduleQueryDto(BaseModel):
+    username: str | None = None
+    password: str | None = None
+    week: int = 0
+    img: bool = False
+
+
+class TestDto(BaseModel):
+    username: str
+    password: str
+    scheduleType: Literal["class", "student"]  # noqa: N815
+
+
+class AliasDto(BaseModel):
+    originalName: str  # noqa: N815
+    aliasName: str | None  # noqa: N815
+
+
+class ReturnDto(BaseModel):
+    success: bool
+    message: str | None = None
+    result: Any | None = None
+    img_url: str | None = None
+
+
+def get_session() -> Generator[Session, None, None]:
+    with Session(engine) as session:
+        yield session
+
+
+async def apply_enhance(course_list: list[dict], week: int, img: bool) -> ReturnDto:
+    """
+    在一个方法中集成了 应用别名 和 生成课表图片 功能。此为异步方法，需要 await。
+
+    Example:
+        return await apply_enhance(course_list, week, img)
+
+    Returns:
+        返回应用别名和图片完毕的 ReturnDto。
+    """
+    final_course_list = [course for course in course_list if week in course["weeks"]] if week > 0 else course_list
+
+    final_course_list = apply_alias(final_course_list)
+
+    # 获取课表图片设置
+    title_template = config.get("schedule", "schedule_title_template", "芒果酸的课程表")
+    subtitle_template = config.get("schedule", "schedule_subtitle_template", "")
+    semester_start_date = date.fromisoformat(config.get("schedule", "semester_start_date", "2026-03-02"))
+
+    # 可用变量
+    week_start_day = semester_start_date + timedelta(weeks=week)
+    vars_ = {
+        "week": week,
+        "week_start_day": week_start_day.isoformat(),
+        "week_end_day": (week_start_day + timedelta(days=6)).isoformat(),
+    }
+
+    img_url = None
+    if img:
+        img_url = f"http://172.28.143.24:8000/api/schedule/img/{
+            await generate_img(final_course_list, title_template.format(**vars_), subtitle_template.format(**vars_))
+        }"
+
+    return ReturnDto(success=True, result=final_course_list, img_url=img_url)

router/enhance/model.py

@@ -0,0 +1,29 @@
+from typing import Optional
+
+from sqlalchemy import JSON, Column
+from sqlmodel import Field, SQLModel, create_engine
+
+sqlite_file_name = "data/njupt_api.db"
+sqlite_url = f"sqlite:///{sqlite_file_name}"
+
+engine = create_engine(sqlite_url, connect_args={"check_same_thread": False})
+
+
+class Course(SQLModel, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    name: str
+    teacher: Optional[str] = Field(default=None, nullable=True)
+    classroom: Optional[str] = Field(default=None, nullable=True)
+    weeks: list[int] = Field(default=[], sa_column=Column(JSON))
+    day: int
+    classes: list[int] = Field(default=[], sa_column=Column(JSON))
+
+
+class Alias(SQLModel, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    originalName: Optional[str] = Field(default=None, nullable=True)  # noqa: N815
+    aliasName: Optional[str] = Field(default=None, nullable=True)  # noqa: N815
+
+
+def create_db_and_tables() -> None:
+    SQLModel.metadata.create_all(engine)

router/enhance/screenshot.py

@@ -0,0 +1,49 @@
+"""使用 Playwright 截图"""
+
+from json import dumps
+from pathlib import Path
+from urllib.parse import urlencode
+from uuid import uuid4
+
+from playwright.async_api import ViewportSize
+
+from njupt_api.baselib import PlayContextManager, logger
+
+TEMP_DIR = Path.cwd() / "temp"
+
+
+class ScreenShot(PlayContextManager):
+    def __init__(self) -> None:
+        super().__init__()
+
+    async def goto(self, url: str) -> bool:
+        await self.page.set_viewport_size(ViewportSize(width=1200, height=900))
+        res = await self.page.goto(url)
+        logger.debug(f"截图 | {res.ok=} - {url=:.{50}}...")
+        return res.ok
+
+    async def shot(self, save_path: str) -> None:
+        await self.page.mouse.move(0, 0)
+        await self.page.wait_for_load_state("networkidle")
+        await self.page.screenshot(path=save_path)
+        logger.debug(f"截图 | 截图已经保存在 {save_path=}")
+        return
+
+
+async def generate_img(courses: list[dict], title: str, subtitle: str) -> str:
+    """
+    方法将生成课程表图片并保存在临时目录中，返回图片的完整名称。图片位于 temp 工作目录。
+
+    Returns:
+        字符串，表明生成图片的文件名，格式为 `schedule-{uuid4()}.png`
+    """
+    t_name = f"schedule-{uuid4()}.png"
+    async with ScreenShot() as ss:
+        await ss.goto(
+            f"127.0.0.1:8000/webui/schedule#/?{
+                urlencode({'data': dumps(courses), 'title': title, 'subtitle': subtitle})
+            }",
+        )
+        await ss.shot(str(TEMP_DIR / t_name))
+    logger.debug(f"截图 | 生成临时图片 - {t_name}")
+    return t_name

router/enhance/week.py

@@ -0,0 +1,29 @@
+"""学期-星期计算"""
+
+from datetime import date, timedelta
+
+
+def get_semester_week_info(start: date, target: date) -> tuple[int, int]:
+    """
+    给定学期开始日期（第一周周一）和另一指定日期，计算指定的日期是第几周的星期几。
+    Args:
+        start: 学期开始的日期，date
+        target: 指定日期，date
+
+    Returns:
+        包含两个数字的元组，分别为第几周和星期几。
+    """
+    return (target - start).days // 7 + 1, target.isoweekday()
+
+
+def get_week_day_info(target: date) -> tuple[date, date]:
+    """
+    给定一个指定日期，获取该日所在的星期的周一和周日的日期。
+    Args:
+        target: 指定日期，date
+
+    Returns:
+        包含两个 date 的元组，分别为周一和周日。
+    """
+    weekday_int = target.weekday()
+    return target - timedelta(days=weekday_int), target + timedelta(days=6 - weekday_int)

router/mcp_router.py

@@ -0,0 +1,139 @@
+from pathlib import Path
+from typing import Annotated
+
+from fastmcp import FastMCP
+from fastmcp.utilities.types import Image
+from mcp.types import ToolAnnotations
+from pydantic import Field
+from sqlmodel import Session, select
+
+from njupt_api.baselib import LoggingMiddleware, logger
+from njupt_api.zhengfang import (
+    ZhengFang,
+    course_dict_serializer,
+    course_list_serializer,
+)
+from router.enhance.lib import ReturnDto, apply_enhance
+from router.enhance.model import Course, engine
+
+mcp = FastMCP("NJUPT API Suan")
+
+mcp.add_middleware(LoggingMiddleware())
+
+mcp_app = mcp.http_app("/")
+
+
+# 统一参数文档
+USERNAME_TYPE = Annotated[str, Field(description="用户名，也即学号，一般是一位字母接八位数字，字母需要大写。")]
+PASSWORD_TYPE = Annotated[str, Field(description="密码，字符串。")]
+WEEK_TYPE = Annotated[int, Field(description="获取第几周的课表，默认为 0 即获取全部。")]
+IMG_TYPE = Annotated[
+    bool,
+    Field(
+        description="是否需要同时生成图片，默认为 False。如果为 True，图片链接将在 img_url 中提供，链接有效时间为两小时。",  # noqa: E501
+    ),
+]
+
+
+@mcp.tool(
+    name="tool_schedule_class",
+    title="获取默认班级课表",
+    description="获取存储在酸 API 中的默认班级课表，返回值包含 success result message 和 img_url 四个字段。",
+    annotations=ToolAnnotations(
+        title="获取默认课表",
+        readOnlyHint=True,
+        destructiveHint=False,
+        idempotentHint=True,
+        openWorldHint=False,
+    ),
+)
+async def tool_schedule_class(
+    week: WEEK_TYPE = 0,
+    img: IMG_TYPE = False,
+) -> ReturnDto:
+    with Session(engine) as session:
+        course_dtos = session.exec(select(Course)).all()
+    logger.success("从数据库中返回一次性存储的班级课表。")
+    course_list: list[dict] = [course_dict_serializer(course) for course in course_dtos]
+    return await apply_enhance(course_list, week, img)
+
+
+@mcp.tool(
+    name="tool_schedule_class_special",
+    title="获取指定学生的班级课表",
+    description="获取指定学生的班级课表。需要提供学号和密码。返回值包含 success result message 和 img_url 四个字段。",
+    annotations=ToolAnnotations(
+        title="获取指定学生的班级课表",
+        readOnlyHint=True,
+        destructiveHint=False,
+        idempotentHint=True,
+        openWorldHint=False,
+    ),
+)
+async def tool_schedule_class_special(
+    username: USERNAME_TYPE,
+    password: PASSWORD_TYPE,
+    week: WEEK_TYPE = 0,
+    img: IMG_TYPE = False,
+) -> ReturnDto:
+    async with ZhengFang() as zf:
+        if await zf.login(username, password):
+            final_course_list = course_list_serializer(await zf.get_class_schedule())
+            logger.success(f"{username} | 获取指定学生的班级课表成功。")
+            return await apply_enhance(final_course_list, week, img)
+        logger.error(f"{username} | 获取课程表失败，请检查账号密码是否正确后再试。")
+        return ReturnDto(
+            success=False,
+            message="获取课程表失败，请检查账号密码是否正确后再试。",
+        )
+
+
+@mcp.tool(
+    name="tool_schedule_student_special",
+    title="获取指定学生的个人课表",
+    description="获取指定学生的个人课表。需要提供学号和密码。返回值包含 success result message 和 img_url 四个字段。",
+    annotations=ToolAnnotations(
+        title="获取指定学生的个人课表",
+        readOnlyHint=True,
+        destructiveHint=False,
+        idempotentHint=True,
+        openWorldHint=False,
+    ),
+)
+async def tool_schedule_student_special(
+    username: USERNAME_TYPE,
+    password: PASSWORD_TYPE,
+    week: WEEK_TYPE = 0,
+    img: IMG_TYPE = False,
+) -> ReturnDto:
+    async with ZhengFang() as zf:
+        if await zf.login(username, password):
+            final_course_list = course_list_serializer(await zf.get_student_schedule())
+            logger.success(f"{username} | 获取指定学生的个人课表成功。")
+            return await apply_enhance(final_course_list, week, img)
+        logger.error(f"{username} | 获取课程表失败，请检查账号密码是否正确后再试。")
+        return ReturnDto(
+            success=False,
+            message="获取课程表失败，请检查账号密码是否正确后再试。",
+        )
+
+
+@mcp.tool(
+    name="tool_schedule_image",
+    title="直接获取课表图片",
+    description="接收使用其他课表工具得到的图片的文件名，返回 ImageContent。",
+    annotations=ToolAnnotations(
+        title="直接获取课表图片",
+        readOnlyHint=True,
+        destructiveHint=False,
+        idempotentHint=True,
+        openWorldHint=False,
+    ),
+)
+async def tool_schedule_image(
+    img_name: Annotated[str, Field(description="课表图片的文件名，形如 schedule-{uuid4}.png")],
+) -> Image | ReturnDto:
+    img_path = Path.cwd() / "temp" / img_name
+    if img_path.exists():
+        return Image(path=img_path)
+    return ReturnDto(success=False, message=f"未找到指定的课表图片 {img_name}")

router/webui_router.py

@@ -0,0 +1,22 @@
+from pathlib import Path
+
+import aiofiles
+from fastapi import APIRouter
+from fastapi.responses import HTMLResponse
+
+WEBUI_INDEX = Path.cwd() / "webui" / "dist" / "index.html"
+SCHEDULE_INDEX = Path.cwd() / "webui" / "dist" / "index-schedule.html"
+
+webui_router = APIRouter(prefix="/webui")
+
+
+@webui_router.get("/", response_class=HTMLResponse)
+async def get_webui() -> HTMLResponse:
+    async with aiofiles.open(file=WEBUI_INDEX, mode="r", encoding="utf-8") as f:
+        return HTMLResponse(content=await f.read(), status_code=200)
+
+
+@webui_router.get("/schedule", response_class=HTMLResponse)
+async def get_webui_schedule() -> HTMLResponse:
+    async with aiofiles.open(file=SCHEDULE_INDEX, mode="r", encoding="utf-8") as f:
+        return HTMLResponse(content=await f.read(), status_code=200)