| | |
| | | bisheng_service = BishengService(base_url=settings.bisheng_base_url) |
| | | try: |
| | | token = get_bisheng_token(db, current_user.id) |
| | | result = await bisheng_service.upload(token, file_content) |
| | | result = await bisheng_service.upload(token, file.filename, file_content) |
| | | except Exception as e: |
| | | raise HTTPException(status_code=500, detail=str(e)) |
| | | return Response(code=200, msg="", data=result) |
| | |
| | | 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) |