Compare commits

...
2 Commits
Author SHA1 Message Date
hz4th_coder 19c7592e0e 新增异常产品管理功能
- 新增 abnormal_products 数据库表
- 内容库和互联网均无搜索结果时存入异常库
- 新增异常产品 API:
  - GET /api/products/abnormal - 获取异常产品列表
  - GET /api/products/abnormal/<product_name> - 获取异常产品详情
  - POST /api/products/abnormal/<product_name>/resolve - 解决异常产品
  - DELETE /api/products/abnormal/<product_name> - 删除异常产品记录
  - POST /api/products/abnormal/<product_name>/retry - 重试处理
- 更新 README 文档
2026-07-17 00:24:11 +08:00
hz4th_coder 11edee581b 新增产品状态检查功能
- 处理前检查产品是否已存在(已发布/待审核)
- 避免重复处理已有产品
- 新增 check_product_exists API 方法
- 更新 README 文档
2026-07-16 23:35:39 +08:00
5 changed files with 432 additions and 6 deletions
+27 -2
View File
@@ -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
- 抓取任务在后台独立运行,不受页面刷新影响
+115
View File
@@ -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()
+119 -1
View File
@@ -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)
+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()
+63 -3
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(
@@ -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. 提取对应产品的具体内容(排除无关产品)