You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
115 lines
4.2 KiB
115 lines
4.2 KiB
"""
|
|
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")
|
|
|