""" 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} ) if response.status_code == 200 and response.json().get('success'): return True return False except Exception as e: print(f"登录失败: {str(e)}") return False def _ensure_login(self): """确保已登录；session 失效时自动重新登录""" if not self.session: return self.login() return True def _request(self, method, url, **kwargs): """发送请求，401 时自动重新登录并重试一次""" self._ensure_login() try: response = self.session.request(method, url, **kwargs) # 401 未登录：重新登录后重试一次 if response.status_code == 401: print(f"登录态失效，重新登录后重试: {method} {url}") if self.login(): response = self.session.request(method, url, **kwargs) return response except Exception: # 连接失败时也尝试重新登录一次（服务可能重启过） print(f"请求失败，尝试重新登录: {method} {url}") try: if self.login(): return self.session.request(method, url, **kwargs) except Exception: pass raise def get_categories(self): """获取所有分类""" try: response = self._request('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: response = self._request('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: # 根据分类类型选择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._request('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: response = self._request('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: response = self._request('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 get_review_status(self, review_id): """ 获取审核状态 Returns: { 'status': 'pending'/'approved'/'rejected', 'reject_reason': str (被拒时的理由), 'review': dict } or None """ try: response = self._request('GET', f'{self.base_url}/api/reviews/{review_id}') if response.status_code == 200: review = response.json() return { 'status': review.get('status', 'pending'), 'reject_reason': review.get('reject_reason', ''), 'review': review } return None except Exception as e: print(f"获取审核状态失败: {str(e)}") return None def send_notification(self, message): """发送通知到后台管理""" try: # 使用通知API发送通知 response = self._request('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 def check_product_exists(self, product_name): """ 检查产品是否已存在于系统中（已发布或待审核） Args: product_name: 产品名称 Returns: { 'exists': bool, 'status': str ('published'/'pending'/None), 'message': str } """ try: # 1. 检查已发布的产品（通过搜索API） response = self._request('GET', f'{self.base_url}/api/search', params={'q': product_name}) if response.status_code == 200: result = response.json() # 检查是否匹配（精确匹配或包含匹配） product_name_lower = product_name.lower().strip() # 检查已发布的模型 for model in result.get('models', []): name = model.get('name', '').lower().strip() if product_name_lower == name or product_name_lower in name or name in product_name_lower: return { 'exists': True, 'status': 'published', 'message': f'产品 "{product_name}" 已在已发布的模型中', 'category': 'ai-models', 'data': model } # 检查已发布的GPU for gpu in result.get('gpus', []): name = gpu.get('name', '').lower().strip() if product_name_lower == name or product_name_lower in name or name in product_name_lower: return { 'exists': True, 'status': 'published', 'message': f'产品 "{product_name}" 已在已发布的GPU中', 'category': 'gpus', 'data': gpu } # 检查已发布的CPU for cpu in result.get('cpus', []): name = cpu.get('name', '').lower().strip() if product_name_lower == name or product_name_lower in name or name in product_name_lower: return { 'exists': True, 'status': 'published', 'message': f'产品 "{product_name}" 已在已发布的CPU中', 'category': 'cpus', 'data': cpu } # 2. 检查待审核列表 response = self._request('GET', f'{self.base_url}/api/reviews', params={'status': 'pending'}) if response.status_code == 200: reviews = response.json() product_name_lower = product_name.lower().strip() for review in reviews: review_data = review.get('data', {}) review_name = review_data.get('name', '').lower().strip() if product_name_lower == review_name or product_name_lower in review_name or review_name in product_name_lower: return { 'exists': True, 'status': 'pending', 'message': f'产品 "{product_name}" 已在待审核列表中', 'review_id': review.get('id'), 'category': review.get('category_id'), 'data': review } # 产品不存在 return { 'exists': False, 'status': None, 'message': f'产品 "{product_name}" 未在系统中找到' } except Exception as e: print(f"检查产品是否存在失败: {str(e)}") # 出错时返回不存在，允许后续处理 return { 'exists': False, 'status': None, 'message': f'检查失败: {str(e)}' } # 全局客户端实例 paramhub_client = ParamHubClient()