v1.8.0: 模块化重构 + 后台登录认证
This commit is contained in:
184
modules/routes/api_smart.py
Normal file
184
modules/routes/api_smart.py
Normal file
@@ -0,0 +1,184 @@
|
||||
"""
|
||||
智能解析/添加/补充 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)
|
||||
Reference in New Issue
Block a user