""" API客户端工具类 用于与FastAPI后端通信 """ import requests from typing import Optional, Dict, List import json class APIClient: """API客户端""" def __init__(self, base_url: str = "http://127.0.0.1:8000"): self.base_url = base_url.rstrip('/') def _get(self, endpoint: str, params: Optional[Dict] = None) -> Dict: """GET请求""" url = f"{self.base_url}{endpoint}" try: response = requests.get(url, params=params, timeout=5) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: raise Exception(f"API请求失败: {str(e)}") def _post(self, endpoint: str, data: Optional[Dict] = None, files: Optional[Dict] = None) -> Dict: """POST请求""" url = f"{self.base_url}{endpoint}" try: if files: response = requests.post(url, files=files, timeout=30) else: response = requests.post(url, json=data, timeout=5) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: raise Exception(f"API请求失败: {str(e)}") def _delete(self, endpoint: str) -> Dict: """DELETE请求""" url = f"{self.base_url}{endpoint}" try: response = requests.delete(url, timeout=5) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: raise Exception(f"API请求失败: {str(e)}") # 学生相关API def get_students(self) -> List[Dict]: """获取学生列表""" return self._get("/api/students/") def get_student(self, student_id: int) -> Dict: """获取单个学生""" return self._get(f"/api/students/{student_id}") def create_student(self, student_data: Dict) -> Dict: """创建学生""" return self._post("/api/students/", data=student_data) def import_students_from_excel(self, file_path: str) -> Dict: """从Excel导入学生""" with open(file_path, 'rb') as f: files = {'file': (file_path.split('/')[-1], f, 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')} return self._post("/api/students/import-excel", files=files) def delete_student(self, student_id: int) -> Dict: """删除学生""" return self._delete(f"/api/students/{student_id}") # 点名相关API def start_rollcall(self, rollcall_type: str, student_id: Optional[int] = None) -> Dict: """开始点名""" data = {"rollcall_type": rollcall_type} if student_id: data["student_id"] = student_id return self._post("/api/rollcall/start", data=data) def record_rollcall(self, record_data: Dict) -> Dict: """记录点名结果""" return self._post("/api/rollcall/record", data=record_data) def get_rollcall_records(self, skip: int = 0, limit: int = 100) -> List[Dict]: """获取点名记录""" return self._get("/api/rollcall/records", params={"skip": skip, "limit": limit}) def transfer_rollcall(self, from_student_id: int, to_student_id: int) -> Dict: """转移点名""" return self._post("/api/rollcall/transfer", data={ "from_student_id": from_student_id, "to_student_id": to_student_id }) # 积分相关API def get_ranking(self, top_n: int = 10) -> Dict: """获取积分排名""" return self._get("/api/scores/ranking", params={"top_n": top_n}) def export_scores(self, save_path: str) -> bool: """导出积分详单""" url = f"{self.base_url}/api/scores/export" try: response = requests.get(url, timeout=30) response.raise_for_status() with open(save_path, 'wb') as f: f.write(response.content) return True except requests.exceptions.RequestException as e: raise Exception(f"导出失败: {str(e)}") def get_statistics(self) -> Dict: """获取统计信息""" return self._get("/api/scores/statistics")