修复停止功能无法终止旧会话的问题

- 修改 stop_session 方法，支持停止数据库中的旧会话
- 服务重启后 active_sessions 为空，导致无法停止旧会话
- 现在会检查数据库并直接更新会话状态
- 修复 chatglm-3 等长时间运行会话无法停止的问题

README.md

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

config/agent_submit_template.txt

@@ -0,0 +1,58 @@
+## 任务背景
+
+### 产品基本信息
+- **产品名称：** {{product_name}}
+- **产品类别：** {{category}}
+- **子类别：** {{subcategory}}
+
+### 待提交的产品数据
+上一步已通过格式检查，生成了以下产品数据：
+```json
+{{product_data}}
+```
+
+## 任务要求
+
+请将上述产品数据提交到 ParamHub 审核系统。
+
+### 1. 确定提交接口
+根据产品类别（{{category}}）选择对应的 API 接口：
+- AI模型 → `POST /api/models`
+- GPU → `POST /api/gpus`
+- CPU → `POST /api/cpus`
+- 其他动态分类 → `POST /api/items/{category_id}`
+
+### 2. 提交数据
+使用以下命令提交数据：
+
+```bash
+# 先登录获取cookie
+curl -c /tmp/paramhub_cookie.txt -X POST "http://localhost:16041/login" \
+  -H "Content-Type: application/json" \
+  -d '{"password": "admin123"}'
+
+# 提交产品数据（根据类别选择对应的API）
+curl -b /tmp/paramhub_cookie.txt -X POST "http://localhost:16041/api/{对应类别API}" \
+  -H "Content-Type: application/json" \
+  -d '{{product_data}}'
+```
+
+### 3. 输出要求
+请以JSON格式输出提交结果：
+```json
+{
+  "success": true,
+  "review_id": "审核ID",
+  "message": "提交说明",
+  "submitted_data": {
+    "name": "产品名称",
+    "field1": "值1",
+    "field2": "值2"
+  }
+}
+```
+
+**注意：**
+- 确保提交的数据格式正确
+- 记录返回的 review_id
+- 如果提交失败，说明错误原因

models/database.py

@@ -204,9 +204,30 @@ class Database:
                 )
             ''')
             
+            # 异常产品表
+            cursor.execute('''
+                CREATE TABLE IF NOT EXISTS abnormal_products (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    product_name TEXT NOT NULL UNIQUE,
+                    category TEXT,
+                    subcategory TEXT,
+                    abnormal_type TEXT DEFAULT 'no_search_results',
+                    abnormal_reason TEXT,
+                    search_results TEXT,
+                    retry_count INTEGER DEFAULT 0,
+                    status TEXT DEFAULT 'pending',
+                    resolution TEXT,
+                    resolved_at DATETIME,
+                    resolved_by TEXT,
+                    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                    last_retry_at DATETIME
+                )
+            ''')
+            
             # 创建索引
             cursor.execute('CREATE INDEX IF NOT EXISTS idx_process_steps_session ON process_steps(process_id)')
             cursor.execute('CREATE INDEX IF NOT EXISTS idx_process_sessions_status ON process_sessions(status)')
+            cursor.execute('CREATE INDEX IF NOT EXISTS idx_abnormal_products_status ON abnormal_products(status)')
             
             conn.commit()
 
@@ -830,5 +851,99 @@ class Database:
             intervention_data=intervention_data
         )
 
+# ========== 异常产品操作 ==========
+    def add_abnormal_product(self, product_name, category=None, subcategory=None, 
+                            abnormal_type='no_search_results', abnormal_reason=None, 
+                            search_results=None):
+        """添加异常产品"""
+        with self.get_connection() as conn:
+            cursor = conn.cursor()
+            try:
+                cursor.execute('''
+                    INSERT INTO abnormal_products 
+                    (product_name, category, subcategory, abnormal_type, abnormal_reason, search_results)
+                    VALUES (?, ?, ?, ?, ?, ?)
+                ''', (product_name, category, subcategory, abnormal_type, abnormal_reason,
+                      json.dumps(search_results, ensure_ascii=False) if search_results else None))
+                conn.commit()
+                return cursor.lastrowid
+            except sqlite3.IntegrityError:
+                # 产品已存在，更新重试次数
+                cursor.execute('''
+                    UPDATE abnormal_products 
+                    SET retry_count = retry_count + 1, 
+                        last_retry_at = CURRENT_TIMESTAMP,
+                        abnormal_reason = ?,
+                        search_results = ?
+                    WHERE product_name = ?
+                ''', (abnormal_reason, 
+                      json.dumps(search_results, ensure_ascii=False) if search_results else None,
+                      product_name))
+                conn.commit()
+                return None
+    
+    def get_abnormal_products(self, limit=100, status='pending'):
+        """获取异常产品列表"""
+        with self.get_connection() as conn:
+            cursor = conn.cursor()
+            if status == 'all':
+                cursor.execute('''
+                    SELECT * FROM abnormal_products 
+                    ORDER BY created_at DESC 
+                    LIMIT ?
+                ''', (limit,))
+            else:
+                cursor.execute('''
+                    SELECT * FROM abnormal_products 
+                    WHERE status = ?
+                    ORDER BY created_at DESC 
+                    LIMIT ?
+                ''', (status, limit))
+            return [dict(row) for row in cursor.fetchall()]
+    
+    def get_abnormal_count(self, status='pending'):
+        """获取异常产品数量"""
+        with self.get_connection() as conn:
+            cursor = conn.cursor()
+            if status == 'all':
+                cursor.execute('SELECT COUNT(*) FROM abnormal_products')
+            else:
+                cursor.execute('SELECT COUNT(*) FROM abnormal_products WHERE status = ?', (status,))
+            return cursor.fetchone()[0]
+    
+    def resolve_abnormal_product(self, product_name, resolution, resolved_by='manual'):
+        """标记异常产品为已解决"""
+        with self.get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute('''
+                UPDATE abnormal_products 
+                SET status = 'resolved', 
+                    resolution = ?, 
+                    resolved_by = ?,
+                    resolved_at = CURRENT_TIMESTAMP
+                WHERE product_name = ?
+            ''', (resolution, resolved_by, product_name))
+            conn.commit()
+            return cursor.rowcount > 0
+    
+    def delete_abnormal_product(self, product_name):
+        """删除异常产品记录"""
+        with self.get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute('DELETE FROM abnormal_products WHERE product_name = ?', (product_name,))
+            conn.commit()
+            return cursor.rowcount > 0
+    
+    def get_abnormal_product(self, product_name):
+        """获取异常产品详情"""
+        with self.get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute('SELECT * FROM abnormal_products WHERE product_name = ?', (product_name,))
+            row = cursor.fetchone()
+            result = dict(row) if row else None
+            if result and result.get('search_results'):
+                result['search_results'] = json.loads(result['search_results'])
+            return result
+
 # 全局数据库实例
 db = Database()

routes/process_monitor.py

@@ -338,3 +338,84 @@ def preview_fill_fields_template():
     except Exception as e:
         logger.error(f"预览填充字段模板失败: {e}")
         return jsonify({'success': False, 'error': str(e)}), 500
+
+
+# ===== 步骤6提交审核模板 API =====
+
+SUBMIT_TEMPLATE_FILE = os.path.join(TEMPLATE_DIR, 'agent_submit_template.txt')
+
+
+@bp.route('/submit-template', methods=['GET'])
+def get_submit_template():
+    """获取步骤6提交审核任务文本模板"""
+    try:
+        if os.path.exists(SUBMIT_TEMPLATE_FILE):
+            with open(SUBMIT_TEMPLATE_FILE, 'r', encoding='utf-8') as f:
+                template = f.read()
+            return jsonify({
+                'success': True,
+                'template': template
+            })
+        else:
+            return jsonify({
+                'success': False,
+                'error': '模板文件不存在'
+            }), 404
+    except Exception as e:
+        logger.error(f"获取提交模板失败: {e}")
+        return jsonify({'success': False, 'error': str(e)}), 500
+
+
+@bp.route('/submit-template', methods=['POST'])
+def save_submit_template():
+    """保存步骤6提交审核任务文本模板"""
+    try:
+        data = request.get_json()
+        template = data.get('template', '')
+        
+        if not template:
+            return jsonify({'success': False, 'error': '模板内容不能为空'}), 400
+        
+        os.makedirs(TEMPLATE_DIR, exist_ok=True)
+        with open(SUBMIT_TEMPLATE_FILE, 'w', encoding='utf-8') as f:
+            f.write(template)
+        
+        return jsonify({
+            'success': True,
+            'message': '模板已保存'
+        })
+    except Exception as e:
+        logger.error(f"保存提交模板失败: {e}")
+        return jsonify({'success': False, 'error': str(e)}), 500
+
+
+@bp.route('/submit-template/preview', methods=['POST'])
+def preview_submit_template():
+    """预览填充后的步骤6任务文本"""
+    try:
+        data = request.get_json()
+        product_name = data.get('product_name', '示例产品')
+        category = data.get('category', '示例类别')
+        subcategory = data.get('subcategory', '示例子类别')
+        product_data = data.get('product_data', '{"name": "示例产品", "visible": true}')
+        
+        # 读取模板
+        if os.path.exists(SUBMIT_TEMPLATE_FILE):
+            with open(SUBMIT_TEMPLATE_FILE, 'r', encoding='utf-8') as f:
+                template = f.read()
+        else:
+            return jsonify({'success': False, 'error': '模板文件不存在'}), 404
+        
+        # 填充示例数据
+        filled = template.replace('{{product_name}}', product_name)
+        filled = filled.replace('{{category}}', category)
+        filled = filled.replace('{{subcategory}}', subcategory or '无')
+        filled = filled.replace('{{product_data}}', product_data)
+        
+        return jsonify({
+            'success': True,
+            'preview': filled
+        })
+    except Exception as e:
+        logger.error(f"预览提交模板失败: {e}")
+        return jsonify({'success': False, 'error': str(e)}), 500

routes/products.py

@@ -213,4 +213,122 @@ def process_batch():
         'success': True,
         'processed': len(results),
         'results': results
-    })
+    })
+
+# ========== 异常产品 API ==========
+
+@bp.route('/abnormal', methods=['GET'])
+def list_abnormal():
+    """获取异常产品列表"""
+    limit = request.args.get('limit', 100, type=int)
+    status = request.args.get('status', 'pending')
+    
+    products = db.get_abnormal_products(limit=limit, status=status)
+    count = db.get_abnormal_count(status=status)
+    
+    # 解析 JSON 字段
+    for item in products:
+        if item.get('search_results'):
+            try:
+                item['search_results'] = __import__('json').loads(item['search_results'])
+            except:
+                pass
+    
+    return jsonify({
+        'success': True,
+        'products': products,
+        'count': count
+    })
+
+
+@bp.route('/abnormal/<product_name>', methods=['GET'])
+def get_abnormal(product_name):
+    """获取异常产品详情"""
+    product = db.get_abnormal_product(product_name)
+    
+    if not product:
+        return jsonify({'error': '异常产品不存在'}), 404
+    
+    return jsonify({
+        'success': True,
+        'product': product
+    })
+
+
+@bp.route('/abnormal/<product_name>/resolve', methods=['POST'])
+def resolve_abnormal(product_name):
+    """解决异常产品"""
+    data = request.get_json()
+    resolution = data.get('resolution', '人工处理完成')
+    resolved_by = data.get('resolved_by', 'manual')
+    
+    success = db.resolve_abnormal_product(product_name, resolution, resolved_by)
+    
+    if success:
+        return jsonify({
+            'success': True,
+            'message': '异常产品已标记为已解决'
+        })
+    else:
+        return jsonify({'error': '异常产品不存在'}), 404
+
+
+@bp.route('/abnormal/<product_name>', methods=['DELETE'])
+def delete_abnormal(product_name):
+    """删除异常产品记录"""
+    success = db.delete_abnormal_product(product_name)
+    
+    if success:
+        return jsonify({
+            'success': True,
+            'message': '异常产品记录已删除'
+        })
+    else:
+        return jsonify({'error': '异常产品不存在'}), 404
+
+
+@bp.route('/abnormal/<product_name>/retry', methods=['POST'])
+def retry_abnormal(product_name):
+    """重试处理异常产品"""
+    # 获取异常产品详情
+    abnormal = db.get_abnormal_product(product_name)
+    
+    if not abnormal:
+        return jsonify({'error': '异常产品不存在'}), 404
+    
+    # 检查是否正在处理
+    processing = db.get_processing_products()
+    if any(p['product_name'] == product_name for p in processing):
+        return jsonify({'error': '该产品正在处理中'}), 400
+    
+    # 添加到处理中列表
+    db.start_processing(
+        product_name=product_name,
+        category=abnormal.get('category'),
+        subcategory=abnormal.get('subcategory')
+    )
+    
+    try:
+        # 执行处理
+        result = process_service.process_product({
+            'product_name': product_name,
+            'category': abnormal.get('category'),
+            'subcategory': abnormal.get('subcategory')
+        })
+        
+        # 如果处理成功，从异常库移除
+        if result['success']:
+            db.resolve_abnormal_product(
+                product_name, 
+                f"重试处理成功: {result['message']}",
+                'auto_retry'
+            )
+        
+        return jsonify({
+            'success': result['success'],
+            'message': result['message'],
+            'review_id': result.get('review_id')
+        })
+    finally:
+        # 完成处理，从处理中列表移除
+        db.finish_processing(product_name)

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

@@ -22,7 +22,7 @@ PROCESS_STEPS = [
     {'num': 3, 'name': '抓取网页内容', 'description': '抓取搜索结果网页的详细内容'},
     {'num': 4, 'name': '提取产品数据(智能体)', 'description': '调用hz4th_editor智能体提取产品相关内容'},
     {'num': 5, 'name': '填充字段(智能体)', 'description': '调用智能体生成产品数据并检查格式'},
-    {'num': 6, 'name': '提交审核', 'description': '提交到ParamHub待审核区'},
+    {'num': 6, 'name': '提交审核(智能体)', 'description': '调用智能体将产品数据提交到ParamHub审核系统'},
 ]
 
 class ProcessMonitor:
@@ -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: 提取产品数据（调用智能体执行）
@@ -276,38 +315,59 @@ class ProcessMonitor:
                 except Exception as e:
                     self._fail_step(session_id, 5, str(e))
             
-            # 步骤6: 提交审核
+            # 步骤6: 提交审核（调用智能体执行）
             if not self._check_pause(session_id) and all_data['filled_data']:
-                self._start_step(session_id, product_name, 6, '提交审核')
+                self._start_step(session_id, product_name, 6, '提交审核(智能体)')
                 try:
-                    category_type = self._get_category_type(category)
-                    success, review_id_or_error = paramhub_client.submit_for_review(
-                        category_type,
-                        all_data['filled_data'],
-                        subcategory
+                    # 构建任务文本
+                    submit_task_text = self._build_submit_task(
+                        product_name, category, subcategory, all_data['filled_data']
                     )
                     
-                    if success:
-                        self._complete_step(session_id, 6, {'review_id': review_id_or_error})
-                        result['success'] = True
-                        result['review_id'] = review_id_or_error
+                    # 调用智能体
+                    submit_agent_result = self._call_agent(submit_task_text)
+                    
+                    if submit_agent_result.get('success'):
+                        submit_parsed = self._parse_submit_agent_response(submit_agent_result.get('output', ''))
                         
-                        db.update_session_status(session_id, 'completed',
-                            review_id=review_id_or_error,
-                            result=json.dumps(result, ensure_ascii=False))
-                        
-                        db.add_process_history(
-                            product_name=product_name,
-                            category=category,
-                            subcategory=subcategory,
-                            status='submitted',
-                            review_id=review_id_or_error,
-                            details=all_data
-                        )
-                        logger.info(f"[{session_id}] 步骤6完成: 提交成功")
+                        if submit_parsed and submit_parsed.get('success'):
+                            review_id = submit_parsed.get('review_id')
+                            
+                            if review_id:
+                                self._complete_step(session_id, 6, {
+                                    'submitted': True,
+                                    'agent': 'hz4th_editor',
+                                    'task_text': submit_task_text,
+                                    'review_id': review_id,
+                                    'agent_output': submit_agent_result.get('output', '')[:2000]
+                                })
+                                
+                                result['success'] = True
+                                result['review_id'] = review_id
+                                
+                                db.update_session_status(session_id, 'completed',
+                                    review_id=review_id,
+                                    result=json.dumps(result, ensure_ascii=False))
+                                
+                                db.add_process_history(
+                                    product_name=product_name,
+                                    category=category,
+                                    subcategory=subcategory,
+                                    status='submitted',
+                                    review_id=review_id,
+                                    details=all_data
+                                )
+                                logger.info(f"[{session_id}] 步骤6完成: 智能体提交成功, review_id={review_id}")
+                            else:
+                                self._fail_step(session_id, 6, '智能体未返回review_id')
+                                result['message'] = '智能体提交成功但未获取到review_id'
+                        else:
+                            error_msg = submit_parsed.get('message', '未知错误') if submit_parsed else '解析失败'
+                            self._fail_step(session_id, 6, f"智能体提交失败: {error_msg}")
+                            result['message'] = f'智能体提交失败: {error_msg}'
                     else:
-                        self._fail_step(session_id, 6, review_id_or_error)
-                        db.update_session_status(session_id, 'failed')
+                        self._fail_step(session_id, 6, f"智能体调用失败: {submit_agent_result.get('error', '未知错误')}")
+                        result['message'] = f'智能体调用失败: {submit_agent_result.get("error")}'
                 except Exception as e:
                     self._fail_step(session_id, 6, str(e))
             
@@ -320,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):
@@ -380,11 +443,27 @@ class ProcessMonitor:
     
     def stop_session(self, session_id):
         """停止会话"""
+        # 先尝试从内存中停止
         if session_id in self.active_sessions:
             self.active_sessions[session_id]['stop'] = True
             self.active_sessions[session_id]['paused'] = False
             db.update_session_status(session_id, 'stopped')
+            logger.info(f"停止会话(内存): {session_id}")
             return True
+        
+        # 如果不在内存中，检查数据库并直接更新状态
+        session = db.get_process_session(session_id)
+        if session:
+            # 只有运行中或暂停状态的会话才能停止
+            if session.get('status') in ('running', 'paused', 'pending'):
+                db.update_session_status(session_id, 'stopped')
+                logger.info(f"停止会话(数据库): {session_id}")
+                return True
+            else:
+                logger.warning(f"会话状态为 {session.get('status')}，无法停止")
+                return False
+        
+        logger.warning(f"会话不存在: {session_id}")
         return False
     
     def get_session_status(self, session_id):
@@ -445,36 +524,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,
-                text=True,
-                timeout=300  # 5分钟超时
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                preexec_fn=os.setsid  # 创建新进程组，方便杀死所有子进程
             )
             
-            if result.returncode == 0:
-                output = result.stdout.strip()
-                logger.info(f"智能体返回: {output[:500]}...")
-                return {'success': True, 'output': output}
-            else:
-                error = result.stderr.strip() or result.stdout.strip()
-                logger.error(f"智能体调用失败: {error}")
-                return {'success': False, 'error': error}
+            try:
+                stdout, stderr = proc.communicate(timeout=180)  # 3分钟超时
+                raw_output = stdout.decode('utf-8', errors='replace').strip()
+                
+                if proc.returncode == 0:
+                    # 解析JSON输出
+                    try:
+                        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):
@@ -685,6 +793,80 @@ class ProcessMonitor:
             'warnings': warnings
         }
     
+    def _build_submit_task(self, product_name, category, subcategory, product_data):
+        """构建步骤6提交审核的智能体任务文本"""
+        # 读取模板
+        template_file = os.path.join(
+            os.path.dirname(os.path.dirname(__file__)),
+            'config', 'agent_submit_template.txt'
+        )
+        
+        if os.path.exists(template_file):
+            with open(template_file, 'r', encoding='utf-8') as f:
+                template = f.read()
+        else:
+            # 默认模板
+            template = (
+                "请将以下产品数据提交到ParamHub审核系统。\n"
+                "产品名称: {{product_name}}\n"
+                "类别: {{category}} / {{subcategory}}\n\n"
+                "产品数据:\n{{product_data}}\n\n"
+                "使用curl命令提交，并记录返回的review_id。"
+            )
+        
+        # 填充模板
+        task = template.replace('{{product_name}}', product_name or '未知')
+        task = task.replace('{{category}}', category or '未分类')
+        task = task.replace('{{subcategory}}', subcategory or '无')
+        task = task.replace('{{product_data}}', json.dumps(product_data, ensure_ascii=False, indent=2))
+        
+        return task
+    
+    def _parse_submit_agent_response(self, output):
+        """解析步骤6智能体返回的结果"""
+        if not output:
+            return None
+        
+        import re
+        
+        parsed_data = None
+        
+        # 查找JSON块
+        json_match = re.search(r'```(?:json)?\s*(\{.*?\})\s*```', output, re.DOTALL)
+        if json_match:
+            try:
+                parsed_data = json.loads(json_match.group(1))
+            except json.JSONDecodeError:
+                pass
+        
+        # 尝试直接解析整个输出为JSON
+        if not parsed_data:
+            try:
+                parsed_data = json.loads(output)
+            except json.JSONDecodeError:
+                pass
+        
+        if parsed_data:
+            return {
+                'success': parsed_data.get('success', False),
+                'review_id': parsed_data.get('review_id'),
+                'message': parsed_data.get('message', ''),
+                'submitted_data': parsed_data.get('submitted_data', {}),
+                'raw_output': output
+            }
+        
+        # 尝试从文本中提取review_id
+        review_match = re.search(r'review[_-]?id[\s:]*([\w-]+)', output, re.I)
+        if review_match:
+            return {
+                'success': True,
+                'review_id': review_match.group(1),
+                'message': '从输出中提取到review_id',
+                'raw_output': output
+            }
+        
+        return None
+    
     def _extract_data(self, product_name, all_data):
         """提取产品数据（备用，已被智能体替代）"""
         all_content = []

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(
@@ -53,9 +84,38 @@ class DataProcessService:
             )
             
             if search_results['total'] == 0:
-                # 没有找到数据，发送通知
-                paramhub_client.send_notification(f"未找到产品 '{product_name}' 的相关数据")
-                result['message'] = '未找到相关数据'
+                # 没有找到数据，存入异常产品库
+                print(f"[异常] 未找到产品 '{product_name}' 的相关数据，存入异常库")
+                
+                # 添加到异常产品库
+                db.add_abnormal_product(
+                    product_name=product_name,
+                    category=category,
+                    subcategory=subcategory,
+                    abnormal_type='no_search_results',
+                    abnormal_reason='内容库和互联网均未搜索到相关数据',
+                    search_results=search_results
+                )
+                
+                # 发送通知
+                paramhub_client.send_notification(
+                    f"⚠️ 产品 '{product_name}' 未找到相关数据\n"
+                    f"已存入异常产品库，请人工排查"
+                )
+                
+                # 记录处理历史
+                db.add_process_history(
+                    product_name=product_name,
+                    category=category,
+                    subcategory=subcategory,
+                    status='abnormal',
+                    details={
+                        'reason': 'no_search_results',
+                        'message': '内容库和互联网均未搜索到相关数据'
+                    }
+                )
+                
+                result['message'] = f"产品 '{product_name}' 未找到相关数据，已存入异常库"
                 return result
             
             # 2. 提取对应产品的具体内容（排除无关产品）

static/js/process.js

@@ -11,6 +11,7 @@ document.addEventListener('DOMContentLoaded', () => {
     loadHistory();
     loadAgentTemplate();
     loadFillFieldsTemplate();
+    loadSubmitTemplate();
     
     // 启动自动刷新（每2秒）
     startAutoRefresh();
@@ -588,3 +589,90 @@ async function doFillFieldsPreview() {
         document.getElementById('template-preview-content').textContent = '预览失败: ' + error.message;
     }
 }
+
+// ===== 步骤6提交审核模板 =====
+
+// 加载步骤6模板
+async function loadSubmitTemplate() {
+    try {
+        const response = await fetch(`${API_BASE}/api/process/submit-template`);
+        const data = await response.json();
+        
+        if (data.success) {
+            document.getElementById('submit-template-editor').value = data.template;
+        } else {
+            document.getElementById('submit-template-editor').value = '// 模板加载失败: ' + (data.error || '未知错误');
+        }
+    } catch (error) {
+        console.error('加载提交模板失败:', error);
+        document.getElementById('submit-template-editor').value = '// 加载模板失败: ' + error.message;
+    }
+}
+
+// 保存步骤6模板
+async function saveSubmitTemplate() {
+    const template = document.getElementById('submit-template-editor').value;
+    
+    if (!template.trim()) {
+        showToast('模板内容不能为空', 'error');
+        return;
+    }
+    
+    try {
+        const response = await fetch(`${API_BASE}/api/process/submit-template`, {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify({ template })
+        });
+        const data = await response.json();
+        
+        if (data.success) {
+            showToast('步骤6模板已保存 ✓', 'success');
+        } else {
+            showToast('保存失败: ' + data.error, 'error');
+        }
+    } catch (error) {
+        showToast('保存失败: ' + error.message, 'error');
+    }
+}
+
+// 预览步骤6模板
+function previewSubmitTemplate() {
+    document.getElementById('template-preview-modal').classList.add('active');
+    doSubmitPreview();
+}
+
+// 执行步骤6预览
+async function doSubmitPreview() {
+    const product = document.getElementById('preview-product').value || '示例产品';
+    const category = document.getElementById('preview-category').value || 'AI模型';
+    
+    try {
+        const response = await fetch(`${API_BASE}/api/process/submit-template/preview`, {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify({
+                product_name: product,
+                category: category,
+                subcategory: '',
+                product_data: JSON.stringify({
+                    "name": product,
+                    "organization": "示例组织",
+                    "parameters": "70B",
+                    "context_length": 4096,
+                    "visible": true,
+                    "is_pinned": false
+                }, null, 2)
+            })
+        });
+        const data = await response.json();
+        
+        if (data.success) {
+            document.getElementById('template-preview-content').textContent = data.preview;
+        } else {
+            document.getElementById('template-preview-content').textContent = '预览失败: ' + data.error;
+        }
+    } catch (error) {
+        document.getElementById('template-preview-content').textContent = '预览失败: ' + error.message;
+    }
+}

templates/process.html

@@ -81,7 +81,7 @@
         <!-- 步骤5填充字段模板区域 -->
         <div class="panel template-section">
             <div class="panel-header">
-                <h2><i class="ri-edit-box-line"></i> 步骤5：填充字段并提交 - 智能体任务模板</h2>
+                <h2><i class="ri-edit-box-line"></i> 步骤5：填充字段 - 智能体任务模板</h2>
                 <div class="template-actions">
                     <button onclick="previewFillFieldsTemplate()" class="btn btn-secondary btn-sm">
                         <i class="ri-eye-line"></i> 预览
@@ -93,19 +93,47 @@
             </div>
             <div class="panel-body">
                 <div class="template-info">
-                    <p><strong>说明：</strong>此模板用于步骤5「填充字段并提交审核」中调用智能体 <code>hz4th_editor</code> 的任务文本。</p>
+                    <p><strong>说明：</strong>此模板用于步骤5「填充字段」中调用智能体 <code>hz4th_editor</code> 的任务文本。</p>
                     <p><strong>可用变量：</strong>
                         <code>{{product_name}}</code> 产品名称、
                         <code>{{category}}</code> 类别、
                         <code>{{subcategory}}</code> 子类别、
                         <code>{{relevant_content_ids}}</code> 上一步筛选的相关内容数据ID
                     </p>
-                    <p><strong>任务目标：</strong>智能体根据API文档获取字段定义，整理产品参数，并通过API提交到ParamHub审核系统。</p>
+                    <p><strong>任务目标：</strong>智能体根据API文档获取字段定义，整理产品参数，并进行格式检查。</p>
                 </div>
                 <textarea id="fill-fields-template-editor" class="template-editor" rows="15" placeholder="加载模板中..."></textarea>
             </div>
         </div>
 
+        <!-- 步骤6提交审核模板区域 -->
+        <div class="panel template-section">
+            <div class="panel-header">
+                <h2><i class="ri-upload-cloud-line"></i> 步骤6：提交审核 - 智能体任务模板</h2>
+                <div class="template-actions">
+                    <button onclick="previewSubmitTemplate()" class="btn btn-secondary btn-sm">
+                        <i class="ri-eye-line"></i> 预览
+                    </button>
+                    <button onclick="saveSubmitTemplate()" class="btn btn-primary btn-sm">
+                        <i class="ri-save-line"></i> 保存模板
+                    </button>
+                </div>
+            </div>
+            <div class="panel-body">
+                <div class="template-info">
+                    <p><strong>说明：</strong>此模板用于步骤6「提交审核」中调用智能体 <code>hz4th_editor</code> 的任务文本。</p>
+                    <p><strong>可用变量：</strong>
+                        <code>{{product_name}}</code> 产品名称、
+                        <code>{{category}}</code> 类别、
+                        <code>{{subcategory}}</code> 子类别、
+                        <code>{{product_data}}</code> 上一步生成的产品数据（JSON格式）
+                    </p>
+                    <p><strong>任务目标：</strong>智能体将产品数据提交到ParamHub审核系统，获取review_id。</p>
+                </div>
+                <textarea id="submit-template-editor" class="template-editor" rows="15" placeholder="加载模板中..."></textarea>
+            </div>
+        </div>
+
         <!-- 预览模态框 -->
         <div id="template-preview-modal" class="modal">
             <div class="modal-content large">