from datetime import datetime import httpx from app.config.config import settings from app.utils.rsa_crypto import BishengCrypto class BishengService: def __init__(self, base_url: str): self.base_url = base_url def _check_response(self, response: httpx.Response): if response.status_code not in [200, 201]: raise Exception(f"Failed to fetch data from Bisheng API: {response.text}") response_data = response.json() status_code = response_data.get("status_code", 0) if status_code != 200: raise Exception(f"Failed to fetch data from Bisheng API: {response.text}") # 检查返回的数据类型 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() password = BishengCrypto(public_key, settings.PRIVATE_KEY).encrypt(password) async with httpx.AsyncClient() as client: response = await client.post( f"{self.base_url}/api/v1/user/regist", json={"user_name": username, "password": password}, headers={'Content-Type': 'application/json'} ) return self._check_response(response) async def login(self, username: str, password: str) -> str: public_key = await self.get_public_key_api() password = BishengCrypto(public_key, settings.PRIVATE_KEY).encrypt(password) async with httpx.AsyncClient() as client: response = await client.post( f"{self.base_url}/api/v1/user/login", json={"user_name": username, "password": password}, headers={'Content-Type': 'application/json'} ) data = self._check_response(response) return data.get('access_token') async def get_public_key_api(self) -> dict: async with httpx.AsyncClient() as client: response = await client.get( f"{self.base_url}/api/v1/user/public_key", headers={'Content-Type': 'application/json'} ) data = self._check_response(response) return data.get('public_key') async def get_chat_sessions(self, token: str) -> list: url = f"{self.base_url}/api/v1/chat/list?page=1&limit=40" 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) result = [ { "id": item["chat_id"], "name": item["latest_message"]["message"], "updated_time": int(datetime.strptime(item["update_time"], "%Y-%m-%dT%H:%M:%S").timestamp() * 1000) } for item in data ] return result async def get_session_log(self, token: str, agent_id: str, conversation_id: str): url = ( f"{self.base_url}/api/v1/chat/history?" f"flow_id={agent_id}&" f"chat_id={conversation_id}&page_size=30&id=" ) headers = {'cookie': f"access_token_cookie={token};"} async with httpx.AsyncClient() as client: response = await client.get(url, headers=headers) response.raise_for_status() data = self._check_response(response) session_log = [ { "message": message.get("message"), "role": message.get("category"), "ts": message.get("create_time") } for message in data ] # 把session_log 按ts 升序排序 session_log.sort(key=lambda x: x['ts']) return session_log async def variable_list(self, token: str, agent_id: str) -> list: url = f"{self.base_url}/api/v1/variable/list?flow_id={agent_id}" 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 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": (filename, file)} async with httpx.AsyncClient() as client: response = await client.post(url, headers=headers, files=files) data = self._check_response(response) file_path = data.get("file_path", "") result = { "file_path": file_path } 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 async def change_password_public(self, token: str, username: str, password: str, new_password:str) -> dict: url = f"{self.base_url}/api/v1/user/change_password_public" headers = {'cookie': f"access_token_cookie={token};"} public_key = await self.get_public_key_api() password = BishengCrypto(public_key, settings.PRIVATE_KEY).encrypt(password) new_password = BishengCrypto(public_key, settings.PRIVATE_KEY).encrypt(new_password) json = {"username": username, "password": password, "new_password": new_password} async with httpx.AsyncClient() as client: response = await client.post(url, headers=headers, json=json) data = self._check_response(response) return data