fastapi demo

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)

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="登录并获取token,以下所有接口都需要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)

ecjtu/ecjtu_api/auth.py

@@ -0,0 +1,109 @@
+import base64
+import datetime
+
+from cushy_storage import CushyOrmCache
+
+from ecjtu.client import ECJTU
+from ecjtu.utils.cookie import cookies_tolist, list_tocookie
+from ecjtu.utils.logger import get_path
+
+from . import schema
+
+
+def encode_data(data):
+    # 将数据编码为base64字符串
+    return base64.b64encode(data.encode()).decode()
+
+
+def decode_data(encoded_data):
+    # 解码base64字符串
+    return base64.b64decode(encoded_data.encode()).decode()
+
+
+# 双token加密
+def create_tokens(stud_id, pwd):
+    access_time = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
+        minutes=60
+    )
+    access_data = f"{stud_id}:access_token:{access_time}"
+    refresh_time = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
+        days=7
+    )
+    refresh_data = f"{stud_id}:{pwd}:{refresh_time}"
+
+    access_token = encode_data(access_data)
+    refresh_token = encode_data(refresh_data)
+    client = ECJTU(stud_id, pwd)
+    try:
+        client.login()
+    except Exception as e:
+        raise e
+
+    cookie_list = cookies_tolist(client.cookies)
+    stud_file = CushyOrmCache(get_path())
+    stud = stud_file.query("FileAuth").filter(stud_id=stud_id).first()
+
+    # 创建用户储存信息
+    if not stud:
+        stud = schema.FileAuth(stud_id, access_token, cookie_list)
+        stud_file.add(stud)
+        return access_token, refresh_token
+    # 更新用户信息
+    stud.cookie = cookie_list
+    stud.token = access_token
+    stud_file.update_obj(stud)
+    return access_token, refresh_token
+
+
+# 刷新access_token
+def refresh_access_token(refresh_token):
+    data = decode_data(refresh_token)
+    print(data)
+    stud_id = data.split(":")[0]
+    pwd = data.split(":")[1]
+    token_time = data.split(":")[2]
+
+    expire_time = datetime.datetime.fromisoformat(token_time)
+    expire_time = expire_time.replace(tzinfo=datetime.timezone.utc)
+    current_time = datetime.datetime.now(datetime.timezone.utc)
+
+    if current_time > expire_time:
+        raise Exception("令牌已过期，请重新登录")
+    else:
+        try:
+            create_tokens(stud_id, pwd)
+        except Exception:
+            raise Exception("令牌有误，请重新登录")
+        return create_tokens(stud_id, pwd)[0]
+
+
+# 验证和读取access_token的stud_id
+def get_stud_id(access_token):
+    try:
+        data = decode_data(access_token)
+        stud_file = CushyOrmCache(get_path())
+        stud = stud_file.query("FileAuth").filter(token=access_token).first()
+        if not stud:
+            raise Exception("无效令牌")
+    except Exception as e:
+        # raise Exception("无效令牌")
+        raise e
+    stud_id = data.split(":")[0]
+    token_time = data.split(":")[2]
+
+    expire_time = datetime.datetime.fromisoformat(token_time)
+    expire_time = expire_time.replace(tzinfo=datetime.timezone.utc)
+    current_time = datetime.datetime.now(datetime.timezone.utc)
+
+    if current_time > expire_time:
+        raise Exception("令牌已过期，请刷新")
+    return stud_id
+    # return data.split(':')[0]
+
+
+# 读取cookie
+def get_cookie(stud_id):
+    stud_file = CushyOrmCache(get_path())
+    stud = stud_file.query("FileAuth").filter(stud_id=stud_id).first()
+    cookies = list_tocookie(stud.cookie)
+    return cookies

ecjtu/ecjtu_api/middle.py

@@ -0,0 +1,39 @@
+from starlette.middleware.base import BaseHTTPMiddleware
+
+from . import auth, respose_result
+
+
+class MyMiddleware(BaseHTTPMiddleware):
+    async def dispatch(self, request, call_next):
+        # 不被拦截的路由
+        secure_routes = [
+            "/refresh_token",
+            "/login",
+            "/",
+            "/docs",
+            "/docs/",
+            "/openapi.json",
+            "/favicon.ico",
+        ]
+
+        response = respose_result.ResponseResult.auth_error()
+        path = request.url.path
+
+        if path not in secure_routes:
+            header = request.headers.get("token")
+            # 是否存在token
+            if not header:
+                return response
+            # 是否过期
+            try:
+                stud_id = auth.get_stud_id(header)
+            except Exception as e:
+                return respose_result.ResponseResult.auth_error(str(e))
+            if not stud_id:
+                response = respose_result.ResponseResult.auth_error(
+                    "access_token令牌已过期，请刷新"
+                )
+                return response
+        response = await call_next(request)
+
+        return response