Compare commits

..
2 Commits
Author SHA1 Message Date
hz4th_coder 11edee581b 新增产品状态检查功能
- 处理前检查产品是否已存在(已发布/待审核)
- 避免重复处理已有产品
- 新增 check_product_exists API 方法
- 更新 README 文档
2026-07-16 23:35:39 +08:00
hz4th_coder a367887ace fix: 修复步骤3抓取网页问题
- 去掉[:5]限制,现在抓取所有互联网搜索结果
- 创建background_tasks记录,在/search页面显示抓取进度
- 每抓取一个URL更新进度和当前项
- 完成/失败/停止时更新任务状态
2026-07-16 01:01:08 +08:00
4 changed files with 196 additions and 3 deletions
+17 -2
View File
@@ -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
- 抓取任务在后台独立运行,不受页面刷新影响
+108
View File
@@ -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()
+40 -1
View File
@@ -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: 提取产品数据(调用智能体执行)
+31
View File
@@ -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(