feat: 第一次上传

services/services.py

@@ -0,0 +1,166 @@
+import base64
+from uuid import uuid4
+import httpx
+import config
+from services.student_cache import get_student_by_token, save_student
+
+
+SEX_MAP = {
+    0: "男",
+    1: "女",
+}
+
+
+class XBXS:
+    def __init__(
+        self,
+        token: str | None = None,
+        UA: str = config.DEFAULT_USER_AGENT,
+    ) -> None:
+        self.token = token
+
+        self.client = httpx.AsyncClient(
+            base_url=config.BASE_URL,
+            timeout=config.HTTPX_TIMEOUT,
+            headers={
+                "user-agent": UA,
+                "accept": "*/*",
+                "accept-language": "zh-CN,zh-Hans;q=0.9",
+                "accept-encoding": "gzip, deflate, br",
+            },
+            proxy="http://127.0.0.1:9000" if config.DEBUG else None,
+            verify=False if config.DEBUG else True,
+        )
+
+        if token:
+            self.client.headers["Authorization"] = f"Bearer {token}"
+
+    async def get_captcha(self):
+        resp = await self.client.get("/xiaobei-api/captchaImage")
+        resp.raise_for_status()
+        result = resp.json()
+        if "uuid" not in result:
+            raise RuntimeError(f"captcha error: {result}")
+        return {
+            "showCode": result.get("showCode", ""),
+            "uuid": result.get("uuid", ""),
+        }
+
+    async def login(self, username: str, password: str):
+        data = await self.get_captcha()
+        resp = await self.client.post(
+            "/xiaobei-api/userLogin",
+            json={
+                "username": username,
+                "password": base64.b64encode(password.encode()).decode(),
+                "code": data.get("showCode", ""),
+                "uuid": data.get("uuid", ""),
+                "appUuid": "",
+            },
+        )
+        resp.raise_for_status()
+
+        result = resp.json()
+
+        self.token = result.get("token")
+
+        self.client.headers["Authorization"] = f"Bearer {self.token}"
+
+        return result
+
+    async def get_info(self):
+        resp = await self.client.get("/xiaobei-api/getInfo")
+        resp.raise_for_status()
+        result = resp.json()
+        return {
+            "userName": result.get("user", {}).get("userName", ""),
+            "nickName": result.get("user", {}).get("nickName", ""),
+            "roles": result.get("roles", []),
+            "phonenumber": result.get("user", {}).get("phonenumber", ""),
+            "bindphonenumber": result.get("user", {}).get("bindphonenumber", ""),
+            "idCard": result.get("user", {}).get("idCard", ""),
+            "trtcId": result.get("user", {}).get("trtcId", ""),
+        }
+
+    async def get_student_info(self):
+        resp = await self.client.get("/xiaobei-api/student/studentInfo")
+        resp.raise_for_status()
+        result = resp.json()
+
+        sex = result.get("data", {}).get("studentSex", "")
+        sex_text = SEX_MAP.get(sex, "未知")
+
+        return {
+            "id": result.get("data", {}).get("id", ""),
+            "studentNumber": result.get("data", {}).get("studentNumber", ""),
+            "deptId": result.get("data", {}).get("deptId", ""),
+            "deptName": result.get("data", {}).get("deptName", ""),
+            "studentName": result.get("data", {}).get("studentName", ""),
+            "studentIpone": result.get("data", {}).get("studentIpone", ""),
+            "studentIdCard": result.get("data", {}).get("studentIdCard", ""),
+            "studentSex": sex_text,
+            "studentSchoolTime": result.get("data", {}).get("studentSchoolTime", ""),
+            "schoolId": result.get("data", {}).get("schoolId", ""),
+            "schoolName": result.get("data", {}).get("schoolName", ""),
+            "collegeId": result.get("data", {}).get("collegeId", ""),
+            "collegeName": result.get("data", {}).get("collegeName", ""),
+            "professionalId": result.get("data", {}).get("professionalId", ""),
+            "professionalName": result.get("data", {}).get("professionalName", ""),
+        }
+
+    async def dep_id(self):
+        resp = await self.client.get("/xiaobei-api/student/deptId")
+        resp.raise_for_status()
+        result = resp.json()
+        return result
+
+    async def get_user_sig2(self):
+        resp = await self.client.get("/xiaobei-api/trtc/genUserSig2")
+        resp.raise_for_status()
+        result = resp.json()
+        return result
+
+    async def get_know_list(self):
+        resp = await self.client.get("/xiaobei-api/student/know/list")
+        resp.raise_for_status()
+        result = resp.json()
+        if result.get("total", 0) > 0:
+            pass
+        return result
+
+    async def get_know(self, know: str):
+        resp = await self.client.get(f"/xiaobei-api/student/know/{know}")
+        resp.raise_for_status()
+        result = resp.json()
+        if result.get("total", 0) > 0:
+            pass
+        return result
+
+    async def update_student_know(self, knowingID: str):
+        files = {
+            "file0": (
+                str(uuid4()),
+                open("./image.jpg", "rb"),
+                "image/jpeg",
+            )
+        }
+        resp = await self.client.post(
+            "/xiaobei-api/student/know/updateStudentKnow",
+            data={
+                "address": "1388号",
+                "remark": "无",
+                "knowingId": knowingID,
+                "location": "106.5346788194445:29.343291015625",
+                "size": "1",
+            },
+        )
+        resp.raise_for_status()
+
+    async def get_student_info_cached(self):
+        cached = get_student_by_token(str(self.token))
+        if cached:
+            return cached
+
+        student = await self.get_student_info()
+        save_student(str(self.token), student)
+        return student