Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 hz4th_coder
|
11edee581b | ||
|
hz4th_coder
|
a367887ace | ||
|
hz4th_coder
|
42c2a53623 |
@@ -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）
|
||||
- 抓取任务在后台独立运行，不受页面刷新影响
|
||||
|
||||
@@ -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()
|
||||
+88
-17
@@ -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,
|
||||
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):
|
||||
|
||||
@@ -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(
|
||||
|
||||
Reference in New Issue
Block a user