185 lines
7.5 KiB
Python
185 lines
7.5 KiB
Python
"""
|
|
智能解析/添加/补充 API
|
|
"""
|
|
import uuid
|
|
from datetime import datetime
|
|
from flask import Blueprint, request, jsonify
|
|
from config import MODELS_FILE, GPUS_FILE, CPUS_FILE, DATA_DIR
|
|
from utils import load_data, save_data
|
|
from llm import parse_with_llm, get_parse_prompt_template
|
|
|
|
smart_bp = Blueprint('api_smart', __name__)
|
|
|
|
|
|
def _smart_update_generic(entity_id, file_path, category_type, entity_name):
|
|
"""通用智能补充逻辑"""
|
|
data = request.get_json()
|
|
text = data.get('text', '')
|
|
images = data.get('images', [])
|
|
if not text and not images:
|
|
return jsonify({'error': '文本或图片不能都为空'}), 400
|
|
items = load_data(file_path)
|
|
item = next((i for i in items if i['id'] == entity_id), None)
|
|
if not item:
|
|
return jsonify({'error': f'{entity_name} not found'}), 404
|
|
parsed_list = parse_with_llm(text, category_type, images)
|
|
if not parsed_list:
|
|
return jsonify({'error': '解析失败'}), 500
|
|
parsed = parsed_list[0]
|
|
updated_fields = []
|
|
for key, value in parsed.items():
|
|
if value is not None and value != '' and value != 0:
|
|
existing = item.get(key)
|
|
if existing is None or existing == '' or existing == 0:
|
|
item[key] = value
|
|
updated_fields.append(key)
|
|
item['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
parse_source = {
|
|
'type': 'smart_update',
|
|
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
|
'images': images,
|
|
'text': text[:500] if text else '',
|
|
'updated_fields': updated_fields
|
|
}
|
|
if 'parse_sources' not in item:
|
|
item['parse_sources'] = []
|
|
item['parse_sources'].append(parse_source)
|
|
save_data(file_path, items)
|
|
return jsonify({'success': True, 'updated_fields': updated_fields, entity_name.lower(): item})
|
|
|
|
|
|
def _smart_add_generic(category_type, file_path, category_id=None):
|
|
"""通用智能添加逻辑"""
|
|
data = request.get_json()
|
|
text = data.get('text', '')
|
|
images = data.get('images', [])
|
|
subcategory_id = data.get('subcategory_id', '')
|
|
custom_prompt = data.get('custom_prompt', '')
|
|
if not text and not images:
|
|
return jsonify({'error': '文本或图片不能都为空'}), 400
|
|
parsed_list = parse_with_llm(text, category_type, images, category_id=category_id,
|
|
subcategory_id=subcategory_id, custom_prompt=custom_prompt)
|
|
results = []
|
|
items = load_data(file_path)
|
|
parse_source = {
|
|
'type': 'smart_add',
|
|
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
|
'images': images,
|
|
'text': text[:500] if text else ''
|
|
}
|
|
for parsed in parsed_list:
|
|
parsed['id'] = uuid.uuid4().hex[:12]
|
|
parsed['created_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
parsed['visible'] = True
|
|
parsed['subcategory_id'] = subcategory_id
|
|
parsed['publish_date'] = parsed.get('publish_date', '')
|
|
parsed['views'] = 0
|
|
parsed['is_pinned'] = False
|
|
parsed['product_images'] = []
|
|
parsed['parse_sources'] = [parse_source]
|
|
if category_id:
|
|
parsed['category_id'] = category_id
|
|
items.append(parsed)
|
|
results.append(parsed)
|
|
save_data(file_path, items)
|
|
return jsonify({'success': True, 'count': len(results), 'products': results})
|
|
|
|
|
|
# ---- Parse Prompt ----
|
|
@smart_bp.route('/api/parse-prompt', methods=['POST'])
|
|
def api_get_parse_prompt():
|
|
data = request.get_json()
|
|
category_type = data.get('category_type', 'dynamic')
|
|
category_id = data.get('category_id')
|
|
subcategory_id = data.get('subcategory_id')
|
|
template = get_parse_prompt_template(category_type, category_id, subcategory_id)
|
|
return jsonify({'success': True, 'template': template})
|
|
|
|
|
|
# ---- Image Parse ----
|
|
@smart_bp.route('/api/parse-images', methods=['POST'])
|
|
def api_parse_images():
|
|
data = request.get_json()
|
|
text = data.get('text', '')
|
|
images = data.get('images', [])
|
|
category_type = data.get('category_type', 'dynamic')
|
|
subcategory_id = data.get('subcategory_id', '')
|
|
custom_prompt = data.get('custom_prompt', '')
|
|
if not text and not images:
|
|
return jsonify({'error': '文本或图片不能都为空'}), 400
|
|
if not images:
|
|
return jsonify({'error': '请上传至少一张图片'}), 400
|
|
type_to_cat_id = {'model': 'ai-models', 'gpu': 'gpus', 'cpu': 'cpus', 'dynamic': None}
|
|
category_id = type_to_cat_id.get(category_type)
|
|
parsed_list = parse_with_llm(text, category_type, images, category_id=category_id,
|
|
subcategory_id=subcategory_id, custom_prompt=custom_prompt)
|
|
return jsonify({'success': True, 'count': len(parsed_list), 'products': parsed_list,
|
|
'raw_text': text, 'images': images})
|
|
|
|
|
|
# ---- Smart Update ----
|
|
@smart_bp.route('/api/models/<model_id>/smart-update', methods=['POST'])
|
|
def api_smart_update_model(model_id):
|
|
return _smart_update_generic(model_id, MODELS_FILE, 'model', 'Model')
|
|
|
|
@smart_bp.route('/api/gpus/<gpu_id>/smart-update', methods=['POST'])
|
|
def api_smart_update_gpu(gpu_id):
|
|
return _smart_update_generic(gpu_id, GPUS_FILE, 'gpu', 'GPU')
|
|
|
|
@smart_bp.route('/api/cpus/<cpu_id>/smart-update', methods=['POST'])
|
|
def api_smart_update_cpu(cpu_id):
|
|
return _smart_update_generic(cpu_id, CPUS_FILE, 'cpu', 'CPU')
|
|
|
|
@smart_bp.route('/api/items/<category_id>/<item_id>/smart-update', methods=['POST'])
|
|
def api_smart_update_item(category_id, item_id):
|
|
items_file = DATA_DIR / f'items_{category_id}.json'
|
|
items = load_data(items_file)
|
|
item = next((i for i in items if i['id'] == item_id), None)
|
|
if not item:
|
|
return jsonify({'error': 'Item not found'}), 404
|
|
data = request.get_json()
|
|
text = data.get('text', '')
|
|
images = data.get('images', [])
|
|
if not text and not images:
|
|
return jsonify({'error': '文本或图片不能都为空'}), 400
|
|
parsed_list = parse_with_llm(text, 'dynamic', images)
|
|
if not parsed_list:
|
|
return jsonify({'error': '解析失败'}), 500
|
|
parsed = parsed_list[0]
|
|
updated_fields = []
|
|
for key, value in parsed.items():
|
|
if value is not None and value != '' and value != 0:
|
|
existing = item.get(key)
|
|
if existing is None or existing == '' or existing == 0:
|
|
item[key] = value
|
|
updated_fields.append(key)
|
|
item['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
parse_source = {
|
|
'type': 'smart_update', 'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
|
'images': images, 'text': text[:500] if text else '', 'updated_fields': updated_fields
|
|
}
|
|
if 'parse_sources' not in item:
|
|
item['parse_sources'] = []
|
|
item['parse_sources'].append(parse_source)
|
|
save_data(items_file, items)
|
|
return jsonify({'success': True, 'updated_fields': updated_fields, 'item': item})
|
|
|
|
|
|
# ---- Smart Add ----
|
|
@smart_bp.route('/api/models/smart-add', methods=['POST'])
|
|
def api_smart_add_model():
|
|
return _smart_add_generic('model', MODELS_FILE, 'ai-models')
|
|
|
|
@smart_bp.route('/api/gpus/smart-add', methods=['POST'])
|
|
def api_smart_add_gpu():
|
|
return _smart_add_generic('gpu', GPUS_FILE, 'gpus')
|
|
|
|
@smart_bp.route('/api/cpus/smart-add', methods=['POST'])
|
|
def api_smart_add_cpu():
|
|
return _smart_add_generic('cpu', CPUS_FILE, 'cpus')
|
|
|
|
@smart_bp.route('/api/items/<category_id>/smart-add', methods=['POST'])
|
|
def api_smart_add_item(category_id):
|
|
items_file = DATA_DIR / f'items_{category_id}.json'
|
|
return _smart_add_generic('dynamic', items_file, category_id)
|