412 lines
15 KiB
Python
412 lines
15 KiB
Python
|
"""
|
||
|
|
数据处理服务 - 核心处理逻辑
|
||
|
|
"""
|
||
|
|
import json
|
||
|
|
import re
|
||
|
|
from datetime import datetime
|
||
|
|
from config import Config
|
||
|
|
from models.database import db
|
||
|
|
from services.search_service import search_service
|
||
|
|
from services.paramhub_client import paramhub_client
|
||
|
|
|
||
|
|
class DataProcessService:
|
||
|
|
def __init__(self):
|
||
|
|
self.config = Config
|
||
|
|
|
||
|
|
def process_product(self, product_info):
|
||
|
|
"""
|
||
|
|
处理单个产品的完整流程
|
||
|
|
|
||
|
|
Args:
|
||
|
|
product_info: {
|
||
|
|
'product_name': str,
|
||
|
|
'category': str (可选),
|
||
|
|
'subcategory': str (可选)
|
||
|
|
}
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
{
|
||
|
|
'success': bool,
|
||
|
|
'message': str,
|
||
|
|
'review_id': str (如果成功提交),
|
||
|
|
'new_products': list (发现的新产品)
|
||
|
|
}
|
||
|
|
"""
|
||
|
|
product_name = product_info['product_name']
|
||
|
|
category = product_info.get('category')
|
||
|
|
subcategory = product_info.get('subcategory')
|
||
|
|
|
||
|
|
result = {
|
||
|
|
'success': False,
|
||
|
|
'message': '',
|
||
|
|
'review_id': None,
|
||
|
|
'new_products': []
|
||
|
|
}
|
||
|
|
|
||
|
|
try:
|
||
|
|
# 1. 从内容库和互联网搜索原始数据
|
||
|
|
print(f"[处理] 开始处理产品: {product_name}")
|
||
|
|
search_results = search_service.search_all(
|
||
|
|
keyword=product_name,
|
||
|
|
category=category,
|
||
|
|
include_internet=True
|
||
|
|
)
|
||
|
|
|
||
|
|
if search_results['total'] == 0:
|
||
|
|
# 没有找到数据，发送通知
|
||
|
|
paramhub_client.send_notification(f"未找到产品 '{product_name}' 的相关数据")
|
||
|
|
result['message'] = '未找到相关数据'
|
||
|
|
return result
|
||
|
|
|
||
|
|
# 2. 提取对应产品的具体内容（排除无关产品）
|
||
|
|
extracted_data = self.extract_product_data(
|
||
|
|
product_name,
|
||
|
|
search_results,
|
||
|
|
category,
|
||
|
|
subcategory
|
||
|
|
)
|
||
|
|
|
||
|
|
if not extracted_data:
|
||
|
|
result['message'] = '无法提取有效数据'
|
||
|
|
return result
|
||
|
|
|
||
|
|
# 3. 按照类别和子类别字段填充内容
|
||
|
|
filled_data = self.fill_product_fields(
|
||
|
|
extracted_data,
|
||
|
|
category,
|
||
|
|
subcategory
|
||
|
|
)
|
||
|
|
|
||
|
|
if not filled_data:
|
||
|
|
result['message'] = '填充数据失败'
|
||
|
|
return result
|
||
|
|
|
||
|
|
# 4. 提交到待审核区
|
||
|
|
category_type = self.get_category_type(category)
|
||
|
|
success, review_id_or_error = paramhub_client.submit_for_review(
|
||
|
|
category_type,
|
||
|
|
filled_data,
|
||
|
|
subcategory
|
||
|
|
)
|
||
|
|
|
||
|
|
if success:
|
||
|
|
# 记录处理历史
|
||
|
|
db.add_process_history(
|
||
|
|
product_name=product_name,
|
||
|
|
category=category,
|
||
|
|
subcategory=subcategory,
|
||
|
|
status='submitted',
|
||
|
|
review_id=review_id_or_error,
|
||
|
|
details={'data': filled_data}
|
||
|
|
)
|
||
|
|
|
||
|
|
result['success'] = True
|
||
|
|
result['message'] = f'已提交审核，review_id: {review_id_or_error}'
|
||
|
|
result['review_id'] = review_id_or_error
|
||
|
|
|
||
|
|
print(f"[处理] 产品 {product_name} 提交成功")
|
||
|
|
else:
|
||
|
|
result['message'] = f'提交失败: {review_id_or_error}'
|
||
|
|
db.add_process_history(
|
||
|
|
product_name=product_name,
|
||
|
|
category=category,
|
||
|
|
subcategory=subcategory,
|
||
|
|
status='failed',
|
||
|
|
details={'error': review_id_or_error}
|
||
|
|
)
|
||
|
|
|
||
|
|
# 5. 检查是否发现新的未处理产品
|
||
|
|
new_products = self.discover_new_products(
|
||
|
|
product_name,
|
||
|
|
search_results,
|
||
|
|
category
|
||
|
|
)
|
||
|
|
result['new_products'] = new_products
|
||
|
|
|
||
|
|
return result
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
error_msg = f"处理失败: {str(e)}"
|
||
|
|
print(f"[错误] {error_msg}")
|
||
|
|
result['message'] = error_msg
|
||
|
|
db.add_process_history(
|
||
|
|
product_name=product_name,
|
||
|
|
category=category,
|
||
|
|
subcategory=subcategory,
|
||
|
|
status='error',
|
||
|
|
details={'error': str(e)}
|
||
|
|
)
|
||
|
|
return result
|
||
|
|
|
||
|
|
def extract_product_data(self, target_product, search_results, category, subcategory):
|
||
|
|
"""
|
||
|
|
从搜索结果中提取目标产品的具体内容
|
||
|
|
排除无关产品
|
||
|
|
"""
|
||
|
|
extracted = {
|
||
|
|
'name': target_product,
|
||
|
|
'category': category,
|
||
|
|
'subcategory': subcategory,
|
||
|
|
'raw_data': []
|
||
|
|
}
|
||
|
|
|
||
|
|
# 收集所有相关内容
|
||
|
|
all_content = []
|
||
|
|
|
||
|
|
# 从内容库结果中提取
|
||
|
|
for article in search_results.get('articles', []):
|
||
|
|
product_names = json.loads(article.get('product_names', '[]'))
|
||
|
|
|
||
|
|
# 检查是否包含目标产品
|
||
|
|
if self.is_product_match(target_product, product_names):
|
||
|
|
all_content.append({
|
||
|
|
'source': article.get('source'),
|
||
|
|
'url': article.get('url'),
|
||
|
|
'summary': article.get('summary'),
|
||
|
|
'content': article.get('content'),
|
||
|
|
'keywords': json.loads(article.get('keywords', '[]'))
|
||
|
|
})
|
||
|
|
|
||
|
|
# 从互联网搜索结果中提取
|
||
|
|
for item in search_results.get('internet', []):
|
||
|
|
if self.is_product_match(target_product, [item.get('title', '')]):
|
||
|
|
all_content.append({
|
||
|
|
'source': 'internet',
|
||
|
|
'url': item.get('url'),
|
||
|
|
'content': item.get('content')
|
||
|
|
})
|
||
|
|
|
||
|
|
if not all_content:
|
||
|
|
return None
|
||
|
|
|
||
|
|
extracted['raw_data'] = all_content
|
||
|
|
return extracted
|
||
|
|
|
||
|
|
def fill_product_fields(self, extracted_data, category, subcategory):
|
||
|
|
"""
|
||
|
|
根据分类字段配置，填充产品数据
|
||
|
|
严格按照来源数据，不创造内容
|
||
|
|
"""
|
||
|
|
if not extracted_data:
|
||
|
|
return None
|
||
|
|
|
||
|
|
# 获取分类字段配置
|
||
|
|
category_info = paramhub_client.get_category_fields(subcategory) if subcategory else None
|
||
|
|
|
||
|
|
# 基础字段
|
||
|
|
filled_data = {
|
||
|
|
'name': extracted_data['name'],
|
||
|
|
'visible': True,
|
||
|
|
'is_pinned': False
|
||
|
|
}
|
||
|
|
|
||
|
|
# 根据分类类型填充字段
|
||
|
|
category_type = self.get_category_type(category)
|
||
|
|
|
||
|
|
if category_type == 'model':
|
||
|
|
filled_data.update(self.extract_model_fields(extracted_data['raw_data']))
|
||
|
|
elif category_type == 'gpu':
|
||
|
|
filled_data.update(self.extract_gpu_fields(extracted_data['raw_data']))
|
||
|
|
elif category_type == 'cpu':
|
||
|
|
filled_data.update(self.extract_cpu_fields(extracted_data['raw_data']))
|
||
|
|
else:
|
||
|
|
# 动态分类，使用分类字段配置
|
||
|
|
if category_info and 'fields' in category_info:
|
||
|
|
for field in category_info['fields']:
|
||
|
|
value = self.extract_field_from_data(field, extracted_data['raw_data'])
|
||
|
|
if value:
|
||
|
|
filled_data[field] = value
|
||
|
|
|
||
|
|
# 添加数据来源
|
||
|
|
filled_data['_source'] = 'auto_manager'
|
||
|
|
filled_data['_extracted_at'] = datetime.now().isoformat()
|
||
|
|
|
||
|
|
return filled_data
|
||
|
|
|
||
|
|
def extract_model_fields(self, raw_data):
|
||
|
|
"""提取AI模型字段"""
|
||
|
|
fields = {}
|
||
|
|
|
||
|
|
for data in raw_data:
|
||
|
|
content = data.get('content', '') or ''
|
||
|
|
summary = data.get('summary', '') or ''
|
||
|
|
text = f"{summary}\n{content}"
|
||
|
|
|
||
|
|
# 提取参数量
|
||
|
|
if 'parameters' not in fields:
|
||
|
|
params_match = re.search(r'(\d+(?:\.\d+)?)\s*[Bb]', text)
|
||
|
|
if params_match:
|
||
|
|
fields['parameters'] = f"{params_match.group(1)}B"
|
||
|
|
|
||
|
|
# 提取上下文长度
|
||
|
|
if 'context_length' not in fields:
|
||
|
|
ctx_match = re.search(r'context[:\s]+(\d+)', text, re.I)
|
||
|
|
if ctx_match:
|
||
|
|
fields['context_length'] = int(ctx_match.group(1))
|
||
|
|
|
||
|
|
# 提取组织
|
||
|
|
if 'organization' not in fields:
|
||
|
|
org_match = re.search(r'(?:by\s+|from\s+|developed\s+by\s+)([\w\s]+?)(?:\s|,|\.|$)', text, re.I)
|
||
|
|
if org_match:
|
||
|
|
fields['organization'] = org_match.group(1).strip()
|
||
|
|
|
||
|
|
# 提取发布日期
|
||
|
|
if 'publish_date' not in fields:
|
||
|
|
date_match = re.search(r'(\d{4}[-/]\d{1,2}[-/]\d{1,2})', text)
|
||
|
|
if date_match:
|
||
|
|
fields['publish_date'] = date_match.group(1).replace('/', '-')
|
||
|
|
|
||
|
|
return fields
|
||
|
|
|
||
|
|
def extract_gpu_fields(self, raw_data):
|
||
|
|
"""提取GPU字段"""
|
||
|
|
fields = {}
|
||
|
|
|
||
|
|
for data in raw_data:
|
||
|
|
content = data.get('content', '') or ''
|
||
|
|
summary = data.get('summary', '') or ''
|
||
|
|
text = f"{summary}\n{content}"
|
||
|
|
|
||
|
|
# 提取显存
|
||
|
|
if 'memory_gb' not in fields:
|
||
|
|
mem_match = re.search(r'(\d+)\s*GB', text)
|
||
|
|
if mem_match:
|
||
|
|
fields['memory_gb'] = int(mem_match.group(1))
|
||
|
|
|
||
|
|
# 提取CUDA核心
|
||
|
|
if 'cuda_cores' not in fields:
|
||
|
|
cuda_match = re.search(r'(\d+)\s*(?:CUDA|cuda)\s*(?:cores|core)', text, re.I)
|
||
|
|
if cuda_match:
|
||
|
|
fields['cuda_cores'] = int(cuda_match.group(1))
|
||
|
|
|
||
|
|
# 提取价格
|
||
|
|
if 'price_usd' not in fields:
|
||
|
|
price_match = re.search(r'\$(\d+(?:,\d+)*)', text)
|
||
|
|
if price_match:
|
||
|
|
fields['price_usd'] = int(price_match.group(1).replace(',', ''))
|
||
|
|
|
||
|
|
return fields
|
||
|
|
|
||
|
|
def extract_cpu_fields(self, raw_data):
|
||
|
|
"""提取CPU字段"""
|
||
|
|
fields = {}
|
||
|
|
|
||
|
|
for data in raw_data:
|
||
|
|
content = data.get('content', '') or ''
|
||
|
|
summary = data.get('summary', '') or ''
|
||
|
|
text = f"{summary}\n{content}"
|
||
|
|
|
||
|
|
# 提取核心数
|
||
|
|
if 'cores' not in fields:
|
||
|
|
cores_match = re.search(r'(\d+)\s*(?:cores?|Cores?)', text, re.I)
|
||
|
|
if cores_match:
|
||
|
|
fields['cores'] = int(cores_match.group(1))
|
||
|
|
|
||
|
|
# 提取线程数
|
||
|
|
if 'threads' not in fields:
|
||
|
|
threads_match = re.search(r'(\d+)\s*(?:threads?|Threads?)', text, re.I)
|
||
|
|
if threads_match:
|
||
|
|
fields['threads'] = int(threads_match.group(1))
|
||
|
|
|
||
|
|
# 提取频率
|
||
|
|
if 'base_clock' not in fields:
|
||
|
|
clock_match = re.search(r'(\d+(?:\.\d+)?)\s*GHz', text)
|
||
|
|
if clock_match:
|
||
|
|
fields['base_clock'] = float(clock_match.group(1))
|
||
|
|
|
||
|
|
return fields
|
||
|
|
|
||
|
|
def extract_field_from_data(self, field_name, raw_data):
|
||
|
|
"""从数据中提取指定字段"""
|
||
|
|
# 通用字段提取逻辑
|
||
|
|
for data in raw_data:
|
||
|
|
content = data.get('content', '') or ''
|
||
|
|
summary = data.get('summary', '') or ''
|
||
|
|
text = f"{summary}\n{content}"
|
||
|
|
|
||
|
|
# 尝试直接匹配字段名
|
||
|
|
pattern = rf'{field_name}[:\s]+([^\n]+)'
|
||
|
|
match = re.search(pattern, text, re.I)
|
||
|
|
if match:
|
||
|
|
return match.group(1).strip()
|
||
|
|
|
||
|
|
return None
|
||
|
|
|
||
|
|
def get_category_type(self, category):
|
||
|
|
"""获取分类类型"""
|
||
|
|
if not category:
|
||
|
|
return 'dynamic'
|
||
|
|
|
||
|
|
category_lower = category.lower()
|
||
|
|
if 'model' in category_lower or 'ai' in category_lower or 'llm' in category_lower:
|
||
|
|
return 'model'
|
||
|
|
elif 'gpu' in category_lower:
|
||
|
|
return 'gpu'
|
||
|
|
elif 'cpu' in category_lower:
|
||
|
|
return 'cpu'
|
||
|
|
else:
|
||
|
|
return 'dynamic'
|
||
|
|
|
||
|
|
def is_product_match(self, target_product, product_names):
|
||
|
|
"""检查产品名称是否匹配"""
|
||
|
|
target = target_product.lower().strip()
|
||
|
|
|
||
|
|
for name in product_names:
|
||
|
|
name_lower = name.lower().strip()
|
||
|
|
|
||
|
|
# 完全匹配
|
||
|
|
if target == name_lower:
|
||
|
|
return True
|
||
|
|
|
||
|
|
# 包含匹配
|
||
|
|
if target in name_lower or name_lower in target:
|
||
|
|
return True
|
||
|
|
|
||
|
|
# 关键词匹配（去掉型号后缀）
|
||
|
|
target_base = re.sub(r'[-\d]+$', '', target)
|
||
|
|
name_base = re.sub(r'[-\d]+$', '', name_lower)
|
||
|
|
if target_base == name_base:
|
||
|
|
return True
|
||
|
|
|
||
|
|
return False
|
||
|
|
|
||
|
|
def discover_new_products(self, current_product, search_results, category):
|
||
|
|
"""
|
||
|
|
从搜索结果中发现新的未处理产品
|
||
|
|
"""
|
||
|
|
new_products = []
|
||
|
|
|
||
|
|
# 收集所有产品名称
|
||
|
|
all_product_names = set()
|
||
|
|
|
||
|
|
for article in search_results.get('articles', []):
|
||
|
|
product_names = json.loads(article.get('product_names', '[]'))
|
||
|
|
all_product_names.update(product_names)
|
||
|
|
|
||
|
|
# 检查是否在待处理列表中
|
||
|
|
for product_name in all_product_names:
|
||
|
|
# 排除当前产品
|
||
|
|
if product_name.lower() == current_product.lower():
|
||
|
|
continue
|
||
|
|
|
||
|
|
# 检查是否已在待处理列表或已处理
|
||
|
|
pending_count = db.get_pending_count()
|
||
|
|
pending_products = [p['product_name'] for p in db.get_pending_products(limit=1000)]
|
||
|
|
|
||
|
|
if product_name not in pending_products:
|
||
|
|
# 检查是否已处理过
|
||
|
|
history = db.get_history_by_product(product_name)
|
||
|
|
if not history:
|
||
|
|
# 添加到待处理列表
|
||
|
|
db.add_pending_product(
|
||
|
|
product_name=product_name,
|
||
|
|
category=category,
|
||
|
|
source='discovered',
|
||
|
|
priority=1 # 发现的产品优先级较低
|
||
|
|
)
|
||
|
|
new_products.append(product_name)
|
||
|
|
|
||
|
|
return new_products
|
||
|
|
|
||
|
|
# 全局处理服务实例
|
||
|
|
process_service = DataProcessService()
|