jingrowlocal/jingrow-framework
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化api_adapter.py

Browse Source
This commit is contained in:
jingrow 2025-11-02 00:40:53 +08:00
parent acf87603f4
commit 7665619e01
1 changed files with 7 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
apps/jingrow/jingrow/core/adapters/api_adapter.py
View File

@ -20,9 +20,8 @@ class ApiAdapter:
self.api_url = Config.jingrow_server_url self.api_url = Config.jingrow_server_url
self.api_key = Config.jingrow_api_key self.api_key = Config.jingrow_api_key
self.api_secret = Config.jingrow_api_secret self.api_secret = Config.jingrow_api_secret
self.session_cookie = None # Session cookie 由系统自动管理，不再从配置中读取
def _get_headers(self): def _get_headers(self, session_cookie: Optional[str] = None):
"""获取请求头""" """获取请求头"""
headers = { headers = {
'Content-Type': 'application/json', 'Content-Type': 'application/json',
@ -33,8 +32,8 @@ class ApiAdapter:
if self.api_key and self.api_secret: if self.api_key and self.api_secret:
headers['Authorization'] = f'token {self.api_key}:{self.api_secret}' headers['Authorization'] = f'token {self.api_key}:{self.api_secret}'
if self.session_cookie: if session_cookie:
headers['Cookie'] = f'sid={self.session_cookie}' headers['Cookie'] = f'sid={session_cookie}'
return headers return headers
@ -65,13 +64,12 @@ class ApiAdapter:
def get_logged_user(self, session_cookie: Optional[str] = None) -> Optional[str]: def get_logged_user(self, session_cookie: Optional[str] = None) -> Optional[str]:
try: try:
api_url = f"{self.api_url}/api/action/jingrow.auth.get_logged_user" api_url = f"{self.api_url}/api/action/jingrow.auth.get_logged_user"
cookie = session_cookie or self.session_cookie if not session_cookie:
if not cookie:
return None return None
headers = { headers = {
"Content-Type": "application/json", "Content-Type": "application/json",
"Accept": "application/json", "Accept": "application/json",
"Cookie": f"sid={cookie}" "Cookie": f"sid={session_cookie}"
} }
resp = requests.get(api_url, headers=headers, timeout=8) resp = requests.get(api_url, headers=headers, timeout=8)
if resp.status_code == 200: if resp.status_code == 200:
@ -145,7 +143,7 @@ class ApiAdapter:
def get_agent_detail(self, name: str, session_cookie: Optional[str] = None) -> Optional[Dict[str, Any]]: def get_agent_detail(self, name: str, session_cookie: Optional[str] = None) -> Optional[Dict[str, Any]]:
try: try:
api_url = f"{self.api_url}/api/action/jingrow.ai.utils.jlocal.get_local_ai_agent_detail" api_url = f"{self.api_url}/api/action/jingrow.ai.utils.jlocal.get_local_ai_agent_detail"
headers = self._get_headers() headers = self._get_headers(session_cookie=session_cookie)
payload = {"name": name} payload = {"name": name}
response = requests.post(api_url, headers=headers, json=payload, timeout=15) response = requests.post(api_url, headers=headers, json=payload, timeout=15)
if response.status_code == 200: if response.status_code == 200:
@ -163,7 +161,7 @@ class ApiAdapter:
def get_pg(self, pagetype: str, name: str, session_cookie: Optional[str] = None) -> Dict[str, Any]: def get_pg(self, pagetype: str, name: str, session_cookie: Optional[str] = None) -> Dict[str, Any]:
try: try:
api_url = f"{self.api_url}/api/data/{pagetype}/{name}" api_url = f"{self.api_url}/api/data/{pagetype}/{name}"
headers = self._get_headers() headers = self._get_headers(session_cookie=session_cookie)
response = requests.get(api_url, headers=headers, timeout=15) response = requests.get(api_url, headers=headers, timeout=15)
if response.status_code == 200: if response.status_code == 200:
data = response.json() data = response.json()