| | |
|---|
| | | json={"user_name": username, "password": password}, |
|---|
| | | headers={'Content-Type': 'application/json'} |
|---|
| | | ) |
|---|
| | | self._check_response(response) |
|---|
| | | return self._check_response(response) |
|---|
| | | |
|---|
| | | async def login(self, username: str, password: str) -> 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) |
|---|
| | |
|---|
| | | } |
|---|
| | | |
|---|
| | | return result |
|---|
| | | |
|---|
| | | async def user_list(self, token: str) -> list: |
|---|
| | | url = f"{self.base_url}/api/v1/user/list" |
|---|
| | | headers = {'cookie': f"access_token_cookie={token};"} |
|---|
| | | async with httpx.AsyncClient() as client: |
|---|
| | | response = await client.get(url, headers=headers) |
|---|
| | | data = self._check_response(response) |
|---|
| | | return data |
|---|
