From a0b870ee984408483c0417fb79802bba40df2167 Mon Sep 17 00:00:00 2001 From: hz4th_coder Date: Thu, 13 Aug 2026 18:25:22 +0800 Subject: [PATCH] =?UTF-8?q?v2.0.1=20=E4=BF=AE=E5=A4=8D=E6=90=9C=E7=B4=A2?= =?UTF-8?q?=E5=B9=B6=E5=8F=91=E5=86=B2=E7=AA=81=E4=B8=8E=E6=8F=90=E4=BA=A4?= =?UTF-8?q?=E5=A4=B1=E8=B4=A5=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 搜索服务改用 namespace 隔离浏览器实例（参考 webtest-agent 方案）， 解决定时任务与手动处理并发调用 agent-browser 互相踢掉导致搜索结果为0的问题 - agent-browser 调用加瞬时错误自动重试，使用 /tmp/xdg-rt 目录 - 步骤4 提示词放宽：品牌/系列相关页面也纳入提取，无精确型号时兜底选最相关内容 - 步骤4 大模型返回空时自动兜底，不再直接跳过导致会话卡死 - 会话收尾修复：失败/无数据时会话状态正确标记，并自动移除待处理产品 - 处理入口统一：定时任务/单产品/批量处理均改为 process_monitor 大模型流程（防重） - paramhub_client 增加登录态自动恢复：401 或连接失败时自动重新登录重试 - 全流程实测通过：deepseek-v4-flash-0731 → review_id 19ed2daaabfe --- config/agent_fill_fields_template.txt | 10 +++ config/agent_task_template.txt | 14 ++-- models/database.py | 8 ++ routes/process_monitor.py | 10 ++- routes/products.py | 82 ++++++++------------ services/paramhub_client.py | 77 +++++++++---------- services/process_monitor.py | 98 +++++++++++++++++++++--- services/search_service.py | 106 ++++++++++++++++++-------- utils/scheduler.py | 35 +++------ 9 files changed, 280 insertions(+), 160 deletions(-) diff --git a/config/agent_fill_fields_template.txt b/config/agent_fill_fields_template.txt index 89b3abf..37fc981 100644 --- a/config/agent_fill_fields_template.txt +++ b/config/agent_fill_fields_template.txt @@ -28,6 +28,16 @@ ### 3. 整理产品参数 根据获取到的内容，提取并整理产品的各项参数，严格按照API文档中定义的字段格式填充。 +**提取原则：** +- 优先从内容中提取精确型号对应的参数（参数量、上下文长度、性能指标等） +- 如果找不到型号级参数，**品牌/系列级信息也可用于填充基础字段**： + - `organization`：厂商/组织（如 DeepSeek） + - `name`：产品名称（保留原始名称） + - `publish_date`：如内容提及系列发布时间可提取 + - `series`：所属系列（如 V4 系列） +- 不要编造或推测任何参数，只使用内容中实际存在的信息 +- 找不到的字段留空即可 + ### 4. 格式检查 对生成的数据进行以下检查： - 必填字段是否齐全（name必须有值） diff --git a/config/agent_task_template.txt b/config/agent_task_template.txt index b307b8f..62796d6 100644 --- a/config/agent_task_template.txt +++ b/config/agent_task_template.txt @@ -15,16 +15,19 @@ ## 任务要求 -请分析上述内容库ID对应的数据，判断哪些与产品「{{product_name}}」**直接相关**且**对提取产品参数有用**。 +请分析上述内容库ID对应的数据，判断哪些与产品「{{product_name}}」**相关**且**对提取产品参数有用**。 -### 判断标准： -1. **直接相关性**：内容必须明确提及该产品名称或其主要型号，排除仅提及相似产品或竞品的内容 -2. **参数提取价值**：内容应包含可用于填充产品字段的信息，如： +### 判断标准（宽松模式）： +1. **直接相关**：内容明确提及该产品名称或其主要型号 → 最高优先级 +2. **品牌/系列相关**：内容提及该产品的品牌或所属系列（例如产品是某品牌的某个版本，则品牌官网、系列介绍页、开放平台页均算相关）→ 应纳入 +3. **信息价值**：内容应包含可用于填充产品字段的信息，如： - 产品规格参数（尺寸、重量、容量等） - 技术规格（性能指标、接口、兼容性等） - 功能特性 - 发布信息（发布日期、价格等） + - 组织/厂商信息 - 其他结构化产品数据 +4. **注意**：如果找不到精确型号页面，**品牌官网、系列页面、开放平台等也纳入**，便于提取品牌、组织、系列、发布时间等基础字段；只有当内容与产品完全无关（竞品、其他产品）时才排除 ### 输出要求： 请以JSON格式输出分析结果，包含以下字段： @@ -46,8 +49,9 @@ ``` **注意：** -- `relevant_ids`：与产品直接相关且对参数提取有用的数据ID列表 +- `relevant_ids`：与产品相关且对参数提取有用的数据ID列表（品牌/系列相关也应纳入） - `analysis`：每个相关ID的简要分析说明 - `excluded_ids`：被排除的ID列表（可选） - `exclusion_reasons`：排除原因说明（可选） - `confidence`：整体判断的置信度 +- **至少返回1个最相关的ID**，除非所有内容都与该产品完全无关 diff --git a/models/database.py b/models/database.py index b26eef9..c0c4c0e 100644 --- a/models/database.py +++ b/models/database.py @@ -772,6 +772,14 @@ class Database: ''') return [dict(row) for row in cursor.fetchall()] + def get_session_status(self, session_id): + """获取单个会话状态""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute('SELECT * FROM process_sessions WHERE session_id = ?', (session_id,)) + row = cursor.fetchone() + return dict(row) if row else None + def get_recent_sessions(self, limit=20): """获取最近的会话""" with self.get_connection() as conn: diff --git a/routes/process_monitor.py b/routes/process_monitor.py index 3f67fc9..9a23c85 100644 --- a/routes/process_monitor.py +++ b/routes/process_monitor.py @@ -48,7 +48,15 @@ def start_process(): subcategory = data.get('subcategory') # 启动处理流程 - session_id = process_monitor.start_process(product_name, category, subcategory) + session_id, started = process_monitor.start_process(product_name, category, subcategory) + + if not started: + return jsonify({ + 'success': True, + 'session_id': session_id, + 'message': f'产品 {product_name} 已有处理会话在进行中，已复用现有会话', + 'duplicated': True + }) return jsonify({ 'success': True, diff --git a/routes/products.py b/routes/products.py index f7687bd..d18308e 100644 --- a/routes/products.py +++ b/routes/products.py @@ -114,7 +114,9 @@ def get_product_history(product_name): @bp.route('/process', methods=['POST']) def process_single(): - """处理单个产品""" + """处理单个产品（大模型驱动）""" + from services.process_monitor import process_monitor + data = request.get_json() if 'product_name' not in data: @@ -126,40 +128,32 @@ def process_single(): 'subcategory': data.get('subcategory') } - # 检查是否正在处理 - processing = db.get_processing_products() - if any(p['product_name'] == product_info['product_name'] for p in processing): - return jsonify({'error': '该产品正在处理中'}), 400 - - # 添加到处理中列表 - db.start_processing( + # 启动大模型处理会话（内部有防重） + session_id, started = process_monitor.start_process( product_name=product_info['product_name'], category=product_info['category'], subcategory=product_info['subcategory'] ) - try: - # 执行处理 - result = process_service.process_product(product_info) - - # 从待处理列表移除 - db.remove_pending_product(product_info['product_name']) - - # 如果发现新产品，已在process_service中添加到待处理列表 - + if not started: return jsonify({ - 'success': result['success'], - 'message': result['message'], - 'review_id': result.get('review_id'), - 'new_products': result.get('new_products', []) + 'success': True, + 'message': f'产品 {product_info["product_name"]} 已有处理会话在进行中', + 'session_id': session_id, + 'duplicated': True }) - finally: - # 完成处理，从处理中列表移除 - db.finish_processing(product_info['product_name']) + + return jsonify({ + 'success': True, + 'message': f'处理流程已启动: {product_info["product_name"]}', + 'session_id': session_id + }) @bp.route('/process/batch', methods=['POST']) def process_batch(): - """批量处理产品""" + """批量处理产品（大模型驱动，异步启动会话）""" + from services.process_monitor import process_monitor + data = request.get_json() limit = data.get('limit', 5) @@ -175,39 +169,27 @@ def process_batch(): results = [] for product in products: - # 检查是否正在处理 - processing = db.get_processing_products() - if any(p['product_name'] == product['product_name'] for p in processing): - results.append({ - 'product_name': product['product_name'], - 'success': False, - 'message': '正在处理中' - }) - continue - - # 添加到处理中列表 - db.start_processing( + # 启动大模型处理会话（内部有防重） + session_id, started = process_monitor.start_process( product_name=product['product_name'], category=product.get('category'), subcategory=product.get('subcategory') ) - try: - # 执行处理 - result = process_service.process_product(product) - - # 从待处理列表移除 - db.remove_pending_product(product['product_name']) - + if started: results.append({ 'product_name': product['product_name'], - 'success': result['success'], - 'message': result['message'], - 'review_id': result.get('review_id') + 'success': True, + 'message': '处理会话已启动', + 'session_id': session_id + }) + else: + results.append({ + 'product_name': product['product_name'], + 'success': False, + 'message': '已有处理会话在进行中', + 'session_id': session_id }) - finally: - # 完成处理 - db.finish_processing(product['product_name']) return jsonify({ 'success': True, diff --git a/services/paramhub_client.py b/services/paramhub_client.py index 640e0a6..ffb6909 100644 --- a/services/paramhub_client.py +++ b/services/paramhub_client.py @@ -18,18 +18,44 @@ class ParamHubClient: f'{self.base_url}/login', json={'password': self.password} ) - return response.json().get('success', False) + 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: - if not self.session: - self.login() - - response = self.session.get(f'{self.base_url}/api/categories?all=1') + response = self._request('GET', f'{self.base_url}/api/categories?all=1') return response.json() except Exception as e: print(f"获取分类失败: {str(e)}") @@ -38,10 +64,7 @@ class ParamHubClient: 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}') + response = self._request('GET', f'{self.base_url}/api/categories/{category_id}') return response.json() except Exception as e: print(f"获取分类字段失败: {str(e)}") @@ -60,9 +83,6 @@ class ParamHubClient: (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' @@ -78,7 +98,7 @@ class ParamHubClient: # 添加审核模式需要的字段 data['status'] = 'pending' - response = self.session.post(endpoint, json=data) + response = self._request('POST', endpoint, json=data) result = response.json() if response.status_code == 200 or response.status_code == 201: @@ -91,10 +111,7 @@ class ParamHubClient: 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}') + response = self._request('GET', f'{self.base_url}/api/reviews?status={status}') return response.json() except Exception as e: print(f"获取审核列表失败: {str(e)}") @@ -103,10 +120,7 @@ class ParamHubClient: def get_review_count(self): """获取待审核数量""" try: - if not self.session: - self.login() - - response = self.session.get(f'{self.base_url}/api/reviews/count') + 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)}") @@ -115,14 +129,8 @@ class ParamHubClient: 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} - ) + 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)}") @@ -143,14 +151,8 @@ class ParamHubClient: } """ try: - if not self.session: - self.login() - # 1. 检查已发布的产品（通过搜索API） - response = self.session.get( - f'{self.base_url}/api/search', - params={'q': product_name} - ) + response = self._request('GET', f'{self.base_url}/api/search', params={'q': product_name}) if response.status_code == 200: result = response.json() @@ -196,10 +198,7 @@ class ParamHubClient: # 2. 检查待审核列表 - response = self.session.get( - f'{self.base_url}/api/reviews', - params={'status': 'pending'} - ) + response = self._request('GET', f'{self.base_url}/api/reviews', params={'status': 'pending'}) if response.status_code == 200: reviews = response.json() diff --git a/services/process_monitor.py b/services/process_monitor.py index 13bbc52..193c4b0 100644 --- a/services/process_monitor.py +++ b/services/process_monitor.py @@ -37,7 +37,14 @@ class ProcessMonitor: return f"proc_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}" def start_process(self, product_name, category=None, subcategory=None): - """启动产品处理流程""" + """启动产品处理流程（同产品防重）""" + # 防重：检查该产品是否已有活跃会话 + active = db.get_active_sessions() + for s in active: + if s['product_name'] == product_name and s['status'] in ('pending', 'running', 'paused'): + logger.warning(f"产品 {product_name} 已有活跃会话 {s['session_id']}，跳过重复启动") + return s['session_id'], False + session_id = self.create_session_id() # 创建会话记录 @@ -59,7 +66,7 @@ class ProcessMonitor: thread.start() logger.info(f"启动处理会话: {session_id}, 产品: {product_name}") - return session_id + return session_id, True def _run_process(self, session_id, product_name, category, subcategory): """执行处理流程""" @@ -253,15 +260,57 @@ class ProcessMonitor: }) logger.info(f"[{session_id}] 步骤4完成: 大模型返回 {len(parsed['relevant_ids'])} 个相关ID") else: + # 大模型未返回相关ID：兜底处理——选取搜索内容中前1条作为候选 all_data['extracted_data'] = None - self._complete_step(session_id, 4, { - 'has_data': False, - 'agent': '大模型', - 'model': self._get_active_model_name(), - 'task_text': task_text, - 'agent_output': agent_result.get('output', '')[:2000] - }, status='skipped') - result['message'] = '大模型未找到相关数据ID' + fallback_ids = [] + if all_data.get('fetched_contents'): + fallback_ids = [all_data['fetched_contents'][0]['id']] + elif all_data.get('library_results'): + fallback_ids = [all_data['library_results'][0]['id']] + + fallback_contents = [] + for aid in fallback_ids: + article = db.get_article_by_id(aid) + if article: + fallback_contents.append({ + 'id': aid, + 'title': article.get('search_title', ''), + 'url': article.get('url', ''), + 'content': article.get('content', ''), + 'summary': article.get('summary', ''), + 'analysis': '兜底候选：大模型未筛选出精确相关数据，选用该条作为参考' + }) + + if fallback_contents: + all_data['extracted_data'] = { + 'name': product_name, + 'relevant_ids': fallback_ids, + 'relevant_contents': fallback_contents, + 'confidence': 'low', + 'raw_output': agent_result.get('output', ''), + 'fallback': True + } + self._complete_step(session_id, 4, { + 'has_data': True, + 'agent': '大模型', + 'model': self._get_active_model_name(), + 'task_text': task_text, + 'relevant_ids': fallback_ids, + 'relevant_count': len(fallback_contents), + 'confidence': 'low', + 'fallback': True, + 'agent_output': agent_result.get('output', '')[:2000] + }) + logger.info(f"[{session_id}] 步骤4兜底: 大模型未筛选出相关ID，选用 {fallback_ids} 作为候选") + else: + self._complete_step(session_id, 4, { + 'has_data': False, + 'agent': '大模型', + 'model': self._get_active_model_name(), + 'task_text': task_text, + 'agent_output': agent_result.get('output', '')[:2000] + }, status='skipped') + result['message'] = '大模型未找到相关数据且无兜底内容' else: self._fail_step(session_id, 4, f"大模型调用失败: {agent_result.get('error', '未知错误')}") result['message'] = f'大模型调用失败: {agent_result.get("error")}' @@ -361,6 +410,35 @@ class ProcessMonitor: except Exception as e: self._fail_step(session_id, 6, str(e)) + # 兜底收尾：未成功提交的会话标记为 failed + if not result.get('success'): + if not result.get('message'): + result['message'] = '处理未完成（无有效数据或中途停止）' + # 会话已在步骤失败时标记 failed；这里确保未走到提交分支的会话也收尾 + cur_status = db.get_session_status(session_id) + if cur_status and cur_status.get('status') in ('running', 'pending'): + db.update_session_status(session_id, 'failed') + # 记录处理历史（异常/失败情况） + db.add_process_history( + product_name=product_name, + category=category, + subcategory=subcategory, + status='failed', + details={'message': result['message'], 'all_data': { + 'library_count': len(all_data.get('library_results', [])), + 'internet_count': len(all_data.get('internet_results', [])), + 'fetched_count': len(all_data.get('fetched_contents', [])), + 'extracted': bool(all_data.get('extracted_data')), + 'filled': bool(all_data.get('filled_data')) + }} + ) + + # 处理结束（无论成功失败），从待处理列表移除 + try: + db.remove_pending_product(product_name) + except Exception as e: + logger.warning(f"[{session_id}] 移除待处理产品失败: {e}") + # 清理 if session_id in self.active_sessions: del self.active_sessions[session_id] diff --git a/services/search_service.py b/services/search_service.py index c046305..cd8ae39 100644 --- a/services/search_service.py +++ b/services/search_service.py @@ -7,6 +7,8 @@ import json import subprocess import os import re +import time +import threading import urllib.parse from datetime import datetime from config import Config @@ -16,22 +18,60 @@ class SearchService: def __init__(self): self.timeout = Config.SEARCH_TIMEOUT self.max_results = Config.SEARCH_MAX_RESULTS + self._ns_counter = 0 + self._ns_lock = threading.Lock() - def _run_browser(self, *args, timeout=30000): - """运行 agent-browser 命令""" + def _new_namespace(self): + """生成独立浏览器 namespace，避免与其他流程/服务并发冲突""" + with self._ns_lock: + self._ns_counter += 1 + return f"search_{int(time.time())}_{self._ns_counter}" + + def _run_browser(self, *args, timeout=30000, namespace=None, retries=2): + """运行 agent-browser 命令（带 namespace 隔离 + 自动重试）""" env = os.environ.copy() - env['XDG_RUNTIME_DIR'] = '/tmp/agent-browser-runtime' + env['XDG_RUNTIME_DIR'] = '/tmp/xdg-rt' os.makedirs(env['XDG_RUNTIME_DIR'], exist_ok=True) + try: + os.chmod(env['XDG_RUNTIME_DIR'], 0o700) + except OSError: + pass - cmd = ['agent-browser'] + list(args) - result = subprocess.run( - cmd, - capture_output=True, - text=True, - env=env, - timeout=timeout // 1000 + 5 - ) - return result.stdout, result.stderr, result.returncode + cmd = ['agent-browser'] + if namespace: + cmd += ['--namespace', namespace] + cmd += list(args) + + last_err = None + for attempt in range(retries + 1): + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + env=env, + timeout=timeout // 1000 + 5 + ) + if result.returncode == 0: + return result.stdout, result.stderr, result.returncode + last_err = result.stderr or result.stdout + # 瞬时错误自动重试 + if attempt < retries and any(t in str(last_err) for t in ( + 'ERR_EMPTY_RESPONSE', 'ERR_CONNECTION_REFUSED', 'ERR_CONNECTION_RESET', + 'ERR_CONNECTION_CLOSED', 'ERR_TIMED_OUT', 'ERR_NAME_NOT_RESOLVED', + 'ERR_SOCKET_NOT_CONNECTED', 'ERR_ADDRESS_UNREACHABLE', 'ERR_NETWORK_CHANGED', + 'ERR_INTERNET_DISCONNECTED', 'session already exists' + )): + time.sleep(2 * (attempt + 1)) + continue + return result.stdout, result.stderr, result.returncode + except subprocess.TimeoutExpired: + last_err = '命令超时' + if attempt < retries: + time.sleep(2 * (attempt + 1)) + continue + return '', f'超时: {" ".join(args)}', 1 + return '', str(last_err), 1 def search_internet(self, keyword, max_results=None, engine='bing_cn', use_cache=True, cache_days=7): """ @@ -68,18 +108,21 @@ class SearchService: search_url = search_urls.get(engine, search_urls['bing_cn']) + # 本次搜索使用独立 namespace，避免并发冲突 + ns = self._new_namespace() + try: # 1. 打开搜索引擎 - stdout, stderr, code = self._run_browser('open', search_url, '--timeout', '20000') + stdout, stderr, code = self._run_browser('open', search_url, '--timeout', '20000', namespace=ns) if code != 0: print(f"打开搜索页面失败: {stderr}") return results # 等待页面加载 - stdout, stderr, code = self._run_browser('wait', '5000') + stdout, stderr, code = self._run_browser('wait', '5000', namespace=ns) # 2. 获取搜索结果页面结构 (JSON 格式) - stdout, stderr, code = self._run_browser('snapshot', '--json', '--timeout', '30000') + stdout, stderr, code = self._run_browser('snapshot', '--json', '--timeout', '30000', namespace=ns) if code != 0: print(f"获取页面结构失败: {stderr}") return results @@ -95,10 +138,10 @@ class SearchService: if engine == 'baidu': results = self._parse_baidu_results(data, max_results) else: - results = self._parse_bing_results(data, max_results) + results = self._parse_bing_results(data, max_results, ns) # 5. 关闭浏览器 - self._run_browser('close') + self._run_browser('close', namespace=ns) # 6. 保存到缓存 if results and use_cache: @@ -110,13 +153,13 @@ class SearchService: print(f"搜索出错: {str(e)}") # 尝试关闭浏览器 try: - self._run_browser('close') + self._run_browser('close', namespace=ns) except: pass return results - def _parse_bing_results(self, snapshot_data, max_results=10): + def _parse_bing_results(self, snapshot_data, max_results=10, ns=None): """ 从 Bing 搜索结果的 snapshot 中解析出标题和链接 @@ -168,7 +211,7 @@ class SearchService: for title, ref in refs[:max_results + 5]: if len(results) >= max_results: break - url = self._get_link_url(ref) + url = self._get_link_url(ref, ns) if url and 'bing.com/search' not in url: # 过滤搜索结果页本身的链接 results.append({ 'title': title, @@ -179,7 +222,7 @@ class SearchService: return results - def _parse_baidu_results(self, snapshot_data, max_results=10): + def _parse_baidu_results(self, snapshot_data, max_results=10, ns=None): """从百度搜索结果中解析标题和链接""" results = [] @@ -206,7 +249,7 @@ class SearchService: # 获取每个结果的 URL for title, ref in refs[:max_results]: - url = self._get_link_url(ref) + url = self._get_link_url(ref, ns) if url and 'baidu.com' not in url: results.append({ 'title': title, @@ -217,10 +260,10 @@ class SearchService: return results - def _get_link_url(self, ref): + def _get_link_url(self, ref, ns=None): """通过 agent-browser 获取链接的 URL""" try: - stdout, stderr, code = self._run_browser('get', 'attr', f'@{ref}', 'href', '--json', '--timeout', '5000') + stdout, stderr, code = self._run_browser('get', 'attr', f'@{ref}', 'href', '--json', '--timeout', '5000', namespace=ns) if code == 0 and stdout: data = json.loads(stdout) return data.get('data', {}).get('value', '') @@ -231,9 +274,10 @@ class SearchService: def fetch_url_content(self, url): """抓取网页内容（使用 agent-browser 浏览器方式，绕过反爬虫）""" error_message = None + ns = self._new_namespace() try: # 使用浏览器方式抓取，增加超时时间到60秒 - stdout, stderr, code = self._run_browser('open', url, '--timeout', '60000') + stdout, stderr, code = self._run_browser('open', url, '--timeout', '60000', namespace=ns) if code != 0: error_message = stderr.strip() if stderr else '浏览器打开页面失败' print(f"打开页面失败: {stderr}") @@ -244,14 +288,14 @@ class SearchService: return {'success': False, 'error': error_message} # 等待页面加载（增加到10秒） - self._run_browser('wait', '10000') + self._run_browser('wait', '10000', namespace=ns) # 获取页面标题 - stdout, stderr, code = self._run_browser('get', 'title', '--timeout', '5000') + stdout, stderr, code = self._run_browser('get', 'title', '--timeout', '5000', namespace=ns) title = stdout.strip().replace('[agent-browser] ', '').strip() if code == 0 else '' # 获取页面内容（通过 snapshot 获取 accessibility tree） - stdout, stderr, code = self._run_browser('snapshot', '--json', '--timeout', '15000') + stdout, stderr, code = self._run_browser('snapshot', '--json', '--timeout', '15000', namespace=ns) text = '' if code == 0 and stdout: try: @@ -263,11 +307,11 @@ class SearchService: pass # 获取 URL（可能被重定向） - stdout, stderr, code = self._run_browser('get', 'url', '--timeout', '5000') + stdout, stderr, code = self._run_browser('get', 'url', '--timeout', '5000', namespace=ns) actual_url = stdout.strip() if code == 0 else url # 关闭浏览器 - self._run_browser('close') + self._run_browser('close', namespace=ns) # 提取描述（从页面内容的前200字符） description = text[:200].strip() if text else '' @@ -285,7 +329,7 @@ class SearchService: print(f"抓取URL失败: {url}, 错误: {error_message}") # 尝试关闭浏览器 try: - self._run_browser('close') + self._run_browser('close', namespace=ns) except: pass return {'success': False, 'error': error_message} diff --git a/utils/scheduler.py b/utils/scheduler.py index 2abf89a..8d4041f 100644 --- a/utils/scheduler.py +++ b/utils/scheduler.py @@ -5,7 +5,6 @@ from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.interval import IntervalTrigger from datetime import datetime from models.database import db -from services.process_service import process_service from config import Config import logging @@ -63,8 +62,11 @@ task_scheduler = TaskScheduler() def auto_process_task(): """ - 自动处理产品的定时任务 + 自动处理产品的定时任务（大模型驱动） + 通过 process_monitor 启动异步处理会话，避免与手动处理并发冲突 """ + from services.process_monitor import process_monitor + try: # 检查是否启用自动处理 enabled = db.get_system_config('auto_process_enabled', 'true') @@ -86,33 +88,18 @@ def auto_process_task(): for product in products: try: - # 检查是否正在处理 - processing = db.get_processing_products() - if any(p['product_name'] == product['product_name'] for p in processing): - logger.warning(f"产品 {product['product_name']} 正在处理中，跳过") - continue - - # 添加到处理中列表 - db.start_processing( + # 启动大模型处理会话（内部有防重，已有活跃会话会自动跳过） + session_id, started = process_monitor.start_process( product_name=product['product_name'], category=product.get('category'), subcategory=product.get('subcategory') ) - - # 执行处理 - result = process_service.process_product(product) - - # 从待处理列表移除 - db.remove_pending_product(product['product_name']) - - logger.info(f"产品 {product['product_name']} 处理完成: {result['message']}") - + if started: + logger.info(f"产品 {product['product_name']} 处理会话已启动: {session_id}") + else: + logger.info(f"产品 {product['product_name']} 已有处理会话: {session_id}") except Exception as e: - logger.error(f"处理产品 {product['product_name']} 时出错: {str(e)}") - db.finish_processing(product['product_name']) - finally: - # 确保从处理中列表移除 - db.finish_processing(product['product_name']) + logger.error(f"启动产品 {product['product_name']} 处理会话时出错: {str(e)}") except Exception as e: logger.error(f"自动处理任务执行失败: {str(e)}")