| | |
|---|
| | | status_code = response_data.get("status_code", 0) |
|---|
| | | if status_code != 200: |
|---|
| | | raise Exception(f"Failed to fetch data from Bisheng API: {response.text}") |
|---|
| | | return response_data.get("data", {}) |
|---|
| | | # 检查返回的数据类型 |
|---|
| | | if isinstance(response_data.get("data"), dict): |
|---|
| | | return response_data.get("data", {}) |
|---|
| | | elif isinstance(response_data.get("data"), list): |
|---|
| | | return response_data.get("data", []) |
|---|
| | | else: |
|---|
| | | return {} |
|---|
| | | |
|---|
| | | async def register(self, username: str, password: str): |
|---|
| | | public_key = await self.get_public_key_api() |
|---|
| | |
|---|
| | | data = self._check_response(response) |
|---|
| | | return data |
|---|
| | | |
|---|
| | | async def upload(self, token: str, file: bytes) -> dict: |
|---|
| | | async def upload(self, token: str, filename: str, file: bytes) -> dict: |
|---|
| | | url = f"{self.base_url}/api/v1/knowledge/upload" |
|---|
| | | headers = {'cookie': f"access_token_cookie={token};"} |
|---|
| | | |
|---|
| | | # 创建表单数据，包含文件 |
|---|
| | | files = {"file": ("file", file)} # 使用默认文件名 "file" |
|---|
| | | |
|---|
| | | files = {"file": (filename, file)} |
|---|
| | | async with httpx.AsyncClient() as client: |
|---|
| | | response = await client.post(url, headers=headers, files=files) |
|---|
| | | data = self._check_response(response) |
|---|
