xbxs/services/services.py

import base64
from typing import List
from uuid import uuid4
from fastapi import UploadFile
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,
        remark: str,
        address: str,
        location: str,
        images: List[UploadFile],
    ):
        files = {}
        for i, image in enumerate(images):
            # 为每个图片生成一个唯一的 UUID 文件名
            files[f"file{i}"] = (
                str(uuid4()),  # 这里使用 UUID 作为文件名
                await image.read(),
                "image/jpeg",  # 假设是 jpeg 格式，按需修改
            )
        resp = await self.client.post(
            "/xiaobei-api/student/know/updateStudentKnow",
            data={
                "address": address,
                "remark": remark,
                "knowingId": knowingID,
                "location": location,
                # "location": "106.5346788194445:29.343291015625",
                "size": len(images),
            },
            files=files,
        )
        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