fastapi demo

README.md

@@ -261,6 +261,67 @@ async def main():
 asyncio.run(main())
 ```
 
+## 提供 web 服务器，提供 API 服务
+
+### 启动方法
+1. 通过python代码启动
+    ```py
+    from ecjtu.ecjtu_api.api import start_api_server
+
+    def main():
+        start_api_server(port=8080)
+
+    if __name__ == "__main__":
+        main()
+    ```
+2. 通过命令行启动
+    ```shell
+    python ecjtu/server.py --port 8080
+    ```
+
+### 使用方法
+1. 启动之后，命令行会显示如下内容
+    ```shell
+    INFO:     Started server process [2545]
+    INFO:     Waiting for application startup.
+    INFO:     Application startup complete.
+    INFO:     Uvicorn running on http://127.0.0.1:8080 (Press CTRL+C to quit)
+    ```
+2. 此时通过浏览器访问 http://127.0.0.1:8080 可以看到api在线调试文档
+
+### 本项目提供的api接口
+
+详细信息可以参考源代码当中examples/ecjtu-api.md当中
+
+1. 登录
+    * post /login 
+    通过学号和密码进行登录，获取access_token和refresh_token,access_token用于之后的所有请求,refresh_token用于刷新access_token
+
+    * post /refresh_token
+    当access_token过期时，可以使用refresh_token刷新access_token。
+2. gpa
+
+    * get /gpa
+    获取当前gpa情况
+3. 课表
+    * get /schedule
+        获取今日课表
+    * get /schedule/{date}
+        获取指定日期课表 date格式为2024-05-01
+    * get /schedule/week
+        获取本周课表
+4. 成绩
+    * get /score
+        获取目前成绩
+    * /score/{semester}
+        获取指定学期成绩 semester格式为2023.1
+5. 选课情况
+    * get /elective_courses
+        获取当前选课信息
+    * get /elective_courses/{semester}
+        获取指定学期选课信息 semester格式为2023.1
+
+
 ## 🧰 本地开发
 
 欢迎贡献代码与二次开发，你可以通过以下方式安装依赖，推荐使用 Conda 作为环境管理工具，首先创建一个新的环境并激活：
@@ -281,7 +342,6 @@ poetry install
 
 下面列举了一些未来可能添加的功能，欢迎贡献代码，提出建议。
 
-- [ ] 添加 web 服务器，提供 API 服务
 - [ ] 提供 vercel 一键部署
 - [ ] 提供 docker 快速服务部署
 - [ ] 增加考试查询

ecjtu/client.py

@@ -63,7 +63,11 @@ def has_login(self) -> bool:
 
 class ECJTU(BaseClient[httpx.Client], httpx.Client):
     def __init__(
-        self, stud_id: Optional[str] = None, password: Optional[str] = None, **kwargs
+        self,
+        stud_id: Optional[str] = None,
+        password: Optional[str] = None,
+        cookie: Optional[CookieTypes] = None,
+        **kwargs,
     ) -> None:
         """Initialize ECJTU client.
 
@@ -73,9 +77,12 @@ def __init__(
         """
         super().__init__(verify=False, **kwargs)
 
-        self.stud_id: str = stud_id or os.environ.get("ECJTU_STUDENT_ID")
-        self.password: str = password or os.environ.get("ECJTU_PASSWORD")
-        self.enc_password: str = _get_enc_password(self.password)
+        if cookie:
+            self.cookies = cookie
+        else:
+            self.stud_id: str = stud_id or os.environ.get("ECJTU_STUDENT_ID")
+            self.password: str = password or os.environ.get("ECJTU_PASSWORD")
+            self.enc_password: str = _get_enc_password(self.password)
 
         self.scheduled_courses = crud.ScheduledCourseCRUD(self)
         self.scores = crud.ScoreCRUD(self)
@@ -217,10 +224,6 @@ def login(self) -> None:
 
         logger.info("Login successful")
 
-    def start_api_server(self, port: int = 8000):
-        # TODO: Start a FastAPI server
-        pass
-
 
 class AsyncECJTU(BaseClient[httpx.AsyncClient], httpx.AsyncClient):
     def __init__(self, stud_id: str, password: str, **kwargs) -> None:
@@ -377,7 +380,3 @@ async def login(self) -> None:
             )
 
         logger.info("Login successful")
-
-    async def start_api_server(self):
-        # TODO: Start a FastAPI server
-        pass

ecjtu/ecjtu_api/api.py

@@ -0,0 +1,215 @@
+import datetime
+import re
+
+from fastapi import FastAPI, Header
+from fastapi.responses import RedirectResponse
+
+from ecjtu.client import ECJTU
+
+from . import auth, middle, respose_result, schema
+
+app = FastAPI(title="ECJTU API", description="API for ECJTU")
+
+app.add_middleware(middle.MyMiddleware)
+
+
+@app.get("/", include_in_schema=False)
+def push_docs():
+    respose = RedirectResponse(url="/docs")
+    return respose
+
+
+@app.post(
+    "/login",
+    tags=["登录"],
+    summary="登录",
+    description="登录获取access_token和refresh_token,access_token用于之后的所有请求,refresh_token用于刷新access_token",
+)
+def login(user: schema.UserLoginSchema):
+    try:
+        access_token, refresh_token = auth.create_tokens(user.stud_id, user.password)
+    except Exception as e:
+        return respose_result.ResponseResult.error(str(e))
+
+    return respose_result.ResponseResult.success(
+        {"access_token": access_token, "refresh_token": refresh_token}
+    )
+
+
+@app.post(
+    "/refresh_token",
+    tags=["登录"],
+    summary="刷新access_token",
+    description="刷新access_token",
+)
+def refresh_token(data: str = None):
+    try:
+        access_token = auth.refresh_access_token(data)
+    except Exception as e:
+        return respose_result.ResponseResult.error(str(e))
+    return respose_result.ResponseResult.success({"access_token": access_token})
+
+
+# gpa接口
+@app.get("/gpa", tags=["GPA"], summary="获取GPA", description="获取当学期GPA")
+def gpa(token: str = Header(None)):
+    stud_id = auth.get_stud_id(token)
+    cookie = auth.get_cookie(stud_id)
+    client = ECJTU(cookie=cookie)
+    try:
+        gpa = client.gpa.today()
+    except Exception as e:
+        return respose_result.ResponseResult.error(str(e))
+    return respose_result.ResponseResult.success(dict(gpa))
+
+
+# 课表接口
+@app.get("/schedule", tags=["课表"], summary="获取当天课表", description="获取当天课表")
+def schedule(token: str = Header(None)):
+    stud_id = auth.get_stud_id(token)
+    cookie = auth.get_cookie(stud_id)
+    client = ECJTU(cookie=cookie)
+    try:
+        schedule = client.scheduled_courses.today()
+    except Exception as e:
+        return respose_result.ResponseResult.error(str(e))
+    schedule_list = []
+    for sublist in schedule:
+        schedule_list.append(dict(sublist))
+    return respose_result.ResponseResult.success(schedule_list)
+
+
+@app.get(
+    "/schedule/{date}",
+    tags=["课表"],
+    summary="获取指定日期课表",
+    description="获取指定日期课表,指定日期格式为yyyy-mm-dd",
+)
+def schedule_date(token: str = Header(None), date: str = None):
+    # date(str): The date to filter, eg: 2023-01-01
+    try:
+        valid_date = datetime.datetime.strptime(date, "%Y-%m-%d").date()
+    except ValueError:
+        return respose_result.ResponseResult.param_error("日期格式错误")
+
+    stud_id = auth.get_stud_id(token)
+    cookie = auth.get_cookie(stud_id)
+    client = ECJTU(cookie=cookie)
+    try:
+        scheduled_courses = client.scheduled_courses.filter(date=valid_date)
+    except Exception as e:
+        return respose_result.ResponseResult.error(str(e))
+    schedule_list = []
+    for sublist in scheduled_courses:
+        schedule_list.append(dict(sublist))
+    return respose_result.ResponseResult.success(schedule_list)
+
+
+@app.get(
+    "/schedule_week", tags=["课表"], summary="获取本周课表", description="获取本周课表"
+)
+def schedule_week(token: str = Header(None)):
+    stud_id = auth.get_stud_id(token)
+    cookie = auth.get_cookie(stud_id)
+    client = ECJTU(cookie=cookie)
+    try:
+        schedule = client.scheduled_courses.this_week()
+    except Exception as e:
+        return respose_result.ResponseResult.error(str(e))
+    dict_list = []
+    for sublist in schedule:
+        dict_sublist = []
+        for item in sublist:
+            dict_sublist.append(dict(item))
+        dict_list.append(dict_sublist)
+    return respose_result.ResponseResult.success(dict_list)
+
+
+# 成绩接口
+@app.get("/score", tags=["成绩"], summary="获取当前成绩", description="获取当学期成绩")
+def score(token: str = Header(None)):
+    stud_id = auth.get_stud_id(token)
+    cookie = auth.get_cookie(stud_id)
+    client = ECJTU(cookie=cookie)
+    try:
+        score = client.scores.today()
+    except Exception as e:
+        return respose_result.ResponseResult.error(str(e))
+    score_list = []
+    for sublist in score:
+        score_list.append(dict(sublist))
+    return respose_result.ResponseResult.success(score_list)
+
+
+@app.get(
+    "/score/{semester}",
+    tags=["成绩"],
+    summary="获取指定学期成绩",
+    description="获取指定学期成绩,semester格式为yyyy.1或yyyy.2",
+)
+def score_semester(token: str = Header(None), semester: str = None):
+    # semester(Optional[str]): The semester to filter, eg: 2023.1, 2023.2
+    if not re.match(r"\d{4}\.[12]", semester):
+        return respose_result.ResponseResult.param_error("学期格式错误")
+    stud_id = auth.get_stud_id(token)
+    cookie = auth.get_cookie(stud_id)
+    client = ECJTU(cookie=cookie)
+    try:
+        scores = client.scores.filter(semester=semester)
+    except Exception as e:
+        return respose_result.ResponseResult.error(str(e))
+    score_list = []
+    for sublist in scores:
+        score_list.append(dict(sublist))
+    return respose_result.ResponseResult.success(score_list)
+
+
+# 选课情况接口
+@app.get(
+    "/elective_courses",
+    tags=["选课情况"],
+    summary="获取当前选课情况",
+    description="获取当前学期选课情况",
+)
+def elective_courses(token: str = Header(None)):
+    stud_id = auth.get_stud_id(token)
+    cookie = auth.get_cookie(stud_id)
+    client = ECJTU(cookie=cookie)
+    try:
+        elective_courses = client.elective_courses.today()
+    except Exception as e:
+        return respose_result.ResponseResult.error(str(e))
+    elective_courses_list = []
+    for sublist in elective_courses:
+        elective_courses_list.append(dict(sublist))
+    return respose_result.ResponseResult.success(elective_courses_list)
+
+
+@app.get(
+    "/elective_courses/{semester}",
+    tags=["选课情况"],
+    summary="获取指定学期选课情况",
+    description="获取指定学期选课情况,semester格式为yyyy.1或yyyy.2",
+)
+def elective_courses_semester(token: str = Header(None), semester: str = None):
+    # semester(Optional[str]): The semester to filter, eg: 2023.1, 2023.2
+    if not re.match(r"\d{4}\.[12]", semester):
+        return respose_result.ResponseResult.param_error("学期格式错误")
+    stud_id = auth.get_stud_id(token)
+    cookie = auth.get_cookie(stud_id)
+    client = ECJTU(cookie=cookie)
+    try:
+        elective_courses = client.elective_courses.filter(semester=semester)
+    except Exception as e:
+        return respose_result.ResponseResult.error(str(e))
+    elective_courses_list = []
+    for sublist in elective_courses:
+        elective_courses_list.append(dict(sublist))
+    return respose_result.ResponseResult.success(elective_courses_list)
+
+
+# 启动api服务
+def start_api_server(port=8080):
+    import uvicorn
+
+    uvicorn.run(app, host="127.0.0.1", port=port)