新增产品状态检查功能

- 处理前检查产品是否已存在（已发布/待审核）
- 避免重复处理已有产品
- 新增 check_product_exists API 方法
- 更新 README 文档

README.md

@@ -15,6 +15,10 @@
   - 可随时查看任务进度和状态
   - 支持手动停止正在运行的任务
   - 自动记录成功/失败/保存数量
+- 🛡️ **产品状态检查**：处理前自动检查产品是否已存在
+  - 检查已发布产品（模型/GPU/CPU）
+  - 检查待审核产品
+  - 已存在则自动跳过，避免重复处理
 
 ## 系统架构
 
@@ -262,13 +266,18 @@ GET /api/system/health
 ## 处理流程
 
 1. **添加待处理产品**：手动添加或系统自动发现新产品
-2. **自动/手动触发处理**：
+2. **产品状态检查**：
+   - 检查产品是否已在 ParamHub 系统中
+   - 检查已发布产品（模型/GPU/CPU）
+   - 检查待审核列表
+   - 如已存在则跳过后续处理
+3. **自动/手动触发处理**：
    - 从内容库搜索相关文章
    - 从互联网搜索最新数据
    - 提取产品具体内容
    - 根据类别字段填充数据
    - 提交到ParamHub待审核区
-3. **发现新产品**：处理过程中自动发现并添加相关产品
+4. **发现新产品**：处理过程中自动发现并添加相关产品
 
 ## 数据库表结构
 
@@ -313,6 +322,12 @@ BATCH_SIZE = 5                 # 批量处理数量
 
 ## 版本历史
 
+- v1.2.0 (2026-07-16): 产品状态检查
+  - 新增产品状态检查功能
+  - 处理前检查产品是否已存在（已发布/待审核）
+  - 避免重复处理已有产品
+  - 新增 check_product_exists API 方法
+
 - v1.1.0 (2026-07-14): 后台任务系统
   - 新增后台任务API（/api/tasks）
   - 抓取任务在后台独立运行，不受页面刷新影响

services/paramhub_client.py

@@ -127,6 +127,114 @@ class ParamHubClient:
         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:
+            if not self.session:
+                self.login()
+            
+            # 1. 检查已发布的产品（通过搜索API）
+            response = self.session.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.session.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()

services/process_monitor.py

@@ -105,12 +105,36 @@ class ProcessMonitor:
                 try:
                     fetched = []
                     failed_count = 0
-                    urls_to_fetch = [r['url'] for r in all_data['internet_results'][:5]]
+                    urls_to_fetch = [r['url'] for r in all_data['internet_results']]
+                    total_urls = len(urls_to_fetch)
+                    
+                    # 创建后台任务记录，这样 /search 页面能看到进度
+                    bg_task_id = f"fetch_{session_id}"
+                    db.create_task(bg_task_id, 'fetch_urls', {
+                        'total': total_urls,
+                        'auto_save': True,
+                        'category': category,
+                        'source': 'process_monitor',
+                        'product_name': product_name
+                    })
+                    db.update_task_status(bg_task_id, 'running', total=total_urls)
                     
                     for i, url in enumerate(urls_to_fetch):
                         if self._check_pause(session_id):
+                            db.update_task_status(bg_task_id, 'stopped', progress=i)
                             break
                         
+                        # 获取当前URL对应的标题
+                        result_item = next((r for r in all_data['internet_results'] if r.get('url') == url), {})
+                        current_title = result_item.get('title', url[:50])
+                        
+                        # 更新后台任务进度
+                        db.update_task_status(
+                            bg_task_id, 'running',
+                            progress=i,
+                            current_item=current_title
+                        )
+                        
                         fetch_result = search_service.fetch_url_content(url)
                         if fetch_result.get('success'):
                             title = fetch_result.get('title', '')
@@ -158,10 +182,25 @@ class ProcessMonitor:
                         
                         time.sleep(0.3)
                     
+                    # 更新后台任务状态为完成
+                    db.update_task_status(
+                        bg_task_id, 'completed',
+                        progress=total_urls,
+                        result={
+                            'total': total_urls,
+                            'success': len(fetched),
+                            'failed': failed_count,
+                            'saved': len(fetched)
+                        }
+                    )
+                    
                     all_data['fetched_contents'] = fetched
                     self._complete_step(session_id, 3, {'count': len(fetched), 'failed': failed_count})
                     logger.info(f"[{session_id}] 步骤3完成: 抓取 {len(fetched)} 个网页, 失败 {failed_count} 个")
                 except Exception as e:
+                    # 更新后台任务状态为失败
+                    if 'bg_task_id' in locals():
+                        db.update_task_status(bg_task_id, 'failed', error_message=str(e))
                     self._fail_step(session_id, 3, str(e))
             
             # 步骤4: 提取产品数据（调用智能体执行）
@@ -341,6 +380,9 @@ class ProcessMonitor:
         except Exception as e:
             logger.error(f"处理会话异常: {session_id} - {e}")
             db.update_session_status(session_id, 'failed')
+            # 确保清理
+            if session_id in self.active_sessions:
+                del self.active_sessions[session_id]
             return {'success': False, 'message': str(e)}
     
     def _start_step(self, session_id, product_name, step_num, step_name):
@@ -466,36 +508,65 @@ class ProcessMonitor:
     
     def _call_agent(self, task_text):
         """调用智能体执行任务"""
+        import signal
+        
         try:
             cmd = [
                 'openclaw', 'agent',
                 '--agent', 'hz4th_editor',
-                '--message', task_text
+                '--message', task_text,
+                '--json'  # 输出JSON格式以便解析
             ]
             
-            logger.info(f"调用智能体命令: openclaw agent --agent hz4th_editor --message '[任务文本 {len(task_text)} 字符]'")
+            logger.info(f"调用智能体命令: openclaw agent --agent hz4th_editor --message '[任务文本 {len(task_text)} 字符]' --json")
             
-            result = subprocess.run(
+            # 使用Popen以便更好地控制超时和进程杀死
+            proc = subprocess.Popen(
                 cmd,
-                capture_output=True,
+                stdout=subprocess.PIPE,
-                text=True,
+                stderr=subprocess.PIPE,
-                timeout=300  # 5分钟超时
+                preexec_fn=os.setsid  # 创建新进程组，方便杀死所有子进程
             )
             
-            if result.returncode == 0:
+            try:
-                output = result.stdout.strip()
+                stdout, stderr = proc.communicate(timeout=180)  # 3分钟超时
-                logger.info(f"智能体返回: {output[:500]}...")
+                raw_output = stdout.decode('utf-8', errors='replace').strip()
-                return {'success': True, 'output': output}
+                
-            else:
+                if proc.returncode == 0:
-                error = result.stderr.strip() or result.stdout.strip()
+                    # 解析JSON输出
-                logger.error(f"智能体调用失败: {error}")
+                    try:
-                return {'success': False, 'error': error}
+                        data = json.loads(raw_output)
+                        # 提取实际回复文本: result.payloads[0].text
+                        payloads = data.get('result', {}).get('payloads', [])
+                        if payloads and isinstance(payloads[0], dict):
+                            output = payloads[0].get('text', '')
+                        else:
+                            output = raw_output
+                        
+                        logger.info(f"智能体返回: {output[:500]}...")
+                        return {'success': True, 'output': output}
+                    except json.JSONDecodeError as e:
+                        logger.warning(f"JSON解析失败，使用原始输出: {e}")
+                        return {'success': True, 'output': raw_output}
+                else:
+                    error = stderr.decode('utf-8', errors='replace').strip() or raw_output
+                    logger.error(f"智能体调用失败(returncode={proc.returncode}): {error}")
+                    return {'success': False, 'error': error}
+            
+            except subprocess.TimeoutExpired:
+                # 超时，杀死整个进程组
+                logger.error(f"智能体执行超时(>3分钟)，杀死进程组")
+                try:
+                    os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
+                except Exception:
+                    proc.kill()
+                proc.wait()
+                return {'success': False, 'error': '智能体执行超时(>3分钟)'}
         
-        except subprocess.TimeoutExpired:
-            return {'success': False, 'error': '智能体执行超时(>5分钟)'}
         except FileNotFoundError:
             return {'success': False, 'error': 'openclaw命令未找到'}
         except Exception as e:
+            logger.error(f"智能体调用异常: {e}")
             return {'success': False, 'error': str(e)}
     
     def _parse_agent_response(self, output):

services/process_service.py

@@ -44,6 +44,37 @@ class DataProcessService:
         }
         
         try:
+            # 0. 【新增】检查产品是否已存在于系统中（已发布或待审核）
+            print(f"[检查] 检查产品是否已存在: {product_name}")
+            exists_check = paramhub_client.check_product_exists(product_name)
+            
+            if exists_check['exists']:
+                # 产品已存在，不进行后续处理
+                status_text = {
+                    'published': '已发布',
+                    'pending': '待审核'
+                }.get(exists_check['status'], '未知状态')
+                
+                result['message'] = f"产品 '{product_name}' 已在系统中{status_text}，跳过处理"
+                print(f"[跳过] {result['message']}")
+                
+                # 记录处理历史
+                db.add_process_history(
+                    product_name=product_name,
+                    category=category,
+                    subcategory=subcategory,
+                    status='skipped',
+                    details={
+                        'reason': 'product_exists',
+                        'existing_status': exists_check['status'],
+                        'existing_data': exists_check.get('data')
+                    }
+                )
+                
+                return result
+            
+            print(f"[检查] 产品未找到，继续处理流程")
+            
             # 1. 从内容库和互联网搜索原始数据
             print(f"[处理] 开始处理产品: {product_name}")
             search_results = search_service.search_all(