237 lines
7.5 KiB
Python
237 lines
7.5 KiB
Python
import threading
|
||
import sys
|
||
import os
|
||
import requests
|
||
from typing import Optional
|
||
|
||
# 导入配置
|
||
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))))
|
||
from jingrow.config import Config
|
||
|
||
_thread_local = threading.local()
|
||
|
||
def set_context(context):
|
||
"""设置当前线程的context"""
|
||
_thread_local.context = context
|
||
|
||
def get_jingrow_api_headers():
|
||
"""获取 Jingrow API 认证头部，用于系统级调用"""
|
||
api_key = Config.jingrow_api_key
|
||
api_secret = Config.jingrow_api_secret
|
||
|
||
if not api_key or not api_secret:
|
||
return None
|
||
|
||
return {
|
||
"Authorization": f"token {api_key}:{api_secret}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
def get_session_api_headers():
|
||
"""获取session认证的API头部，类似jingrow系统中的get_jingrow_api_headers()"""
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json"
|
||
}
|
||
|
||
# 从线程本地存储获取context
|
||
context = getattr(_thread_local, 'context', None)
|
||
if context:
|
||
session_cookie = context.get('session_cookie')
|
||
if session_cookie:
|
||
headers["Cookie"] = f"sid={session_cookie}"
|
||
|
||
return headers
|
||
|
||
def get_jingrow_cloud_api_url():
|
||
"""获取jingrow cloud API URL"""
|
||
return Config.jingrow_cloud_api_url
|
||
|
||
def get_jingrow_cloud_url():
|
||
"""获取jingrow cloud API URL"""
|
||
return Config.jingrow_cloud_url
|
||
|
||
def get_jingrow_cloud_api_headers():
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json"
|
||
}
|
||
|
||
api_key = Config.jingrow_cloud_api_key
|
||
api_secret = Config.jingrow_cloud_api_secret
|
||
|
||
if api_key and api_secret:
|
||
headers["Authorization"] = f"token {api_key}:{api_secret}"
|
||
|
||
return headers
|
||
|
||
def _extract_session_cookie_from_response(response: requests.Response, session: requests.Session) -> Optional[str]:
|
||
"""从响应中提取session cookie"""
|
||
session_cookie = None
|
||
if 'Set-Cookie' in response.headers:
|
||
cookies = response.headers['Set-Cookie']
|
||
if 'sid=' in cookies:
|
||
session_cookie = cookies.split('sid=')[1].split(';')[0]
|
||
|
||
# 如果没有从Set-Cookie获取，尝试从session的cookies获取
|
||
if not session_cookie:
|
||
session_cookie = session.cookies.get('sid')
|
||
|
||
return session_cookie
|
||
|
||
def _call_jingrow_api(method: str, endpoint: str, session_cookie: Optional[str] = None, **kwargs) -> requests.Response:
|
||
"""调用Jingrow API的通用函数"""
|
||
url = f"{Config.jingrow_server_url}{endpoint}"
|
||
headers = kwargs.pop('headers', {})
|
||
|
||
# 如果没有提供session_cookie，尝试从context获取
|
||
if not session_cookie:
|
||
context = getattr(_thread_local, 'context', None)
|
||
if context:
|
||
session_cookie = context.get('session_cookie')
|
||
|
||
if session_cookie:
|
||
headers["Cookie"] = f"sid={session_cookie}"
|
||
|
||
return requests.request(method.upper(), url, headers=headers, timeout=30, **kwargs)
|
||
|
||
def login(username: str, password: str) -> dict:
|
||
"""
|
||
登录函数，调用jingrow框架的登录API
|
||
|
||
Args:
|
||
username: 用户名
|
||
password: 密码
|
||
|
||
Returns:
|
||
dict: 登录结果，包含success和message字段，成功时还包含session_cookie
|
||
"""
|
||
try:
|
||
url = f"{Config.jingrow_server_url}/api/action/login"
|
||
headers = {
|
||
"Content-Type": "application/x-www-form-urlencoded",
|
||
"Accept": "application/json"
|
||
}
|
||
|
||
data = {
|
||
"cmd": "login",
|
||
"usr": username,
|
||
"pwd": password
|
||
}
|
||
|
||
# 使用session保持cookie
|
||
session = requests.Session()
|
||
response = session.post(url, headers=headers, data=data, timeout=30)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get("message") == "Logged In":
|
||
session_cookie = _extract_session_cookie_from_response(response, session)
|
||
return {
|
||
"success": True,
|
||
"message": result.get("message", "Logged In"),
|
||
"session_cookie": session_cookie
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"message": result.get("message", "登录失败")
|
||
}
|
||
else:
|
||
error_msg = f"登录请求失败 (HTTP {response.status_code})"
|
||
try:
|
||
error_data = response.json()
|
||
error_msg = error_data.get("message", error_data.get("exc", error_msg))
|
||
except:
|
||
error_msg = response.text or error_msg
|
||
|
||
return {
|
||
"success": False,
|
||
"message": error_msg
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
"success": False,
|
||
"message": f"登录异常: {str(e)}"
|
||
}
|
||
|
||
def logout(session_cookie: Optional[str] = None) -> dict:
|
||
"""
|
||
登出函数，调用jingrow框架的登出API
|
||
|
||
Args:
|
||
session_cookie: 会话cookie，如果为None则从context获取
|
||
|
||
Returns:
|
||
dict: 登出结果，包含success和message字段
|
||
"""
|
||
try:
|
||
response = _call_jingrow_api('GET', '/api/action/logout', session_cookie, headers={"Accept": "application/json"})
|
||
|
||
if response.status_code == 200:
|
||
return {
|
||
"success": True,
|
||
"message": "登出成功"
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"message": f"登出请求失败 (HTTP {response.status_code})"
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
"success": False,
|
||
"message": f"登出异常: {str(e)}"
|
||
}
|
||
|
||
def get_user_info(session_cookie: Optional[str] = None) -> dict:
|
||
"""
|
||
获取用户信息函数，调用jingrow框架的get_user_info API
|
||
|
||
Args:
|
||
session_cookie: 会话cookie，如果为None则从context获取
|
||
|
||
Returns:
|
||
dict: 用户信息，包含success和user_info字段
|
||
"""
|
||
try:
|
||
response = _call_jingrow_api(
|
||
'GET',
|
||
'/api/action/jingrow.realtime.get_user_info',
|
||
session_cookie,
|
||
headers={
|
||
"Accept": "application/json",
|
||
"Content-Type": "application/json"
|
||
}
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
user_info = data.get('message', data)
|
||
|
||
# 格式化用户信息
|
||
formatted_user_info = {
|
||
"id": user_info.get("user") or user_info.get("name") or user_info.get("username", ""),
|
||
"username": user_info.get("user") or user_info.get("name") or user_info.get("username", ""),
|
||
"email": user_info.get("email", ""),
|
||
"avatar": user_info.get("user_image", ""),
|
||
"first_name": user_info.get("first_name", ""),
|
||
"last_name": user_info.get("last_name", ""),
|
||
"user_type": user_info.get("user_type", "System User")
|
||
}
|
||
|
||
return {
|
||
"success": True,
|
||
"user_info": formatted_user_info
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"message": f"获取用户信息失败 (HTTP {response.status_code})"
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
"success": False,
|
||
"message": f"获取用户信息异常: {str(e)}"
|
||
}
|