hz4th_coder
b7b09251f8
功能: - 文章内容库管理 - 待处理产品列表管理 - 自动处理流程 - 智能搜索和数据提取 - ParamHub API集成 - 定时任务调度 部署端口: 16043
132 lines
4.3 KiB
Python
132 lines
4.3 KiB
Python
"""
|
|
ParamHub API 客户端服务
|
|
"""
|
|
import requests
|
|
from config import Config
|
|
|
|
class ParamHubClient:
|
|
def __init__(self):
|
|
self.base_url = Config.PARAMHUB_BASE_URL
|
|
self.password = Config.PARAMHUB_PASSWORD
|
|
self.session = None
|
|
|
|
def login(self):
|
|
"""登录获取session"""
|
|
try:
|
|
self.session = requests.Session()
|
|
response = self.session.post(
|
|
f'{self.base_url}/login',
|
|
json={'password': self.password}
|
|
)
|
|
return response.json().get('success', False)
|
|
except Exception as e:
|
|
print(f"登录失败: {str(e)}")
|
|
return False
|
|
|
|
def get_categories(self):
|
|
"""获取所有分类"""
|
|
try:
|
|
if not self.session:
|
|
self.login()
|
|
|
|
response = self.session.get(f'{self.base_url}/api/categories?all=1')
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"获取分类失败: {str(e)}")
|
|
return []
|
|
|
|
def get_category_fields(self, category_id):
|
|
"""获取分类的字段配置"""
|
|
try:
|
|
if not self.session:
|
|
self.login()
|
|
|
|
response = self.session.get(f'{self.base_url}/api/categories/{category_id}')
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"获取分类字段失败: {str(e)}")
|
|
return None
|
|
|
|
def submit_for_review(self, category_type, data, category_id=None):
|
|
"""
|
|
提交数据到待审核区
|
|
|
|
Args:
|
|
category_type: 分类类型 (model/gpu/cpu/dynamic)
|
|
data: 产品数据
|
|
category_id: 分类ID（用于动态分类）
|
|
|
|
Returns:
|
|
(success, review_id or error_message)
|
|
"""
|
|
try:
|
|
if not self.session:
|
|
self.login()
|
|
|
|
# 根据分类类型选择API端点
|
|
if category_type == 'model':
|
|
endpoint = f'{self.base_url}/api/models'
|
|
elif category_type == 'gpu':
|
|
endpoint = f'{self.base_url}/api/gpus'
|
|
elif category_type == 'cpu':
|
|
endpoint = f'{self.base_url}/api/cpus'
|
|
elif category_type == 'dynamic' and category_id:
|
|
endpoint = f'{self.base_url}/api/items/{category_id}'
|
|
else:
|
|
return False, "无效的分类类型或缺少分类ID"
|
|
|
|
# 添加审核模式需要的字段
|
|
data['status'] = 'pending'
|
|
|
|
response = self.session.post(endpoint, json=data)
|
|
result = response.json()
|
|
|
|
if response.status_code == 200 or response.status_code == 201:
|
|
return True, result.get('review_id', 'submitted')
|
|
else:
|
|
return False, result.get('error', '提交失败')
|
|
except Exception as e:
|
|
return False, str(e)
|
|
|
|
def get_reviews(self, status='pending'):
|
|
"""获取待审核列表"""
|
|
try:
|
|
if not self.session:
|
|
self.login()
|
|
|
|
response = self.session.get(f'{self.base_url}/api/reviews?status={status}')
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"获取审核列表失败: {str(e)}")
|
|
return []
|
|
|
|
def get_review_count(self):
|
|
"""获取待审核数量"""
|
|
try:
|
|
if not self.session:
|
|
self.login()
|
|
|
|
response = self.session.get(f'{self.base_url}/api/reviews/count')
|
|
return response.json().get('count', 0)
|
|
except Exception as e:
|
|
print(f"获取审核数量失败: {str(e)}")
|
|
return 0
|
|
|
|
def send_notification(self, message):
|
|
"""发送通知到后台管理"""
|
|
try:
|
|
if not self.session:
|
|
self.login()
|
|
|
|
# 使用通知API发送通知
|
|
response = self.session.post(
|
|
f'{self.base_url}/api/notifications',
|
|
json={'message': message}
|
|
)
|
|
return response.status_code == 200 or response.status_code == 201
|
|
except Exception as e:
|
|
print(f"发送通知失败: {str(e)}")
|
|
return False
|
|
|
|
# 全局客户端实例
|
|
paramhub_client = ParamHubClient() |