添加分类管理导出导入功能
This commit is contained in:
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -2,10 +2,11 @@
|
||||
分类管理 API
|
||||
"""
|
||||
import uuid
|
||||
import json
|
||||
from datetime import datetime
|
||||
from flask import Blueprint, request, jsonify
|
||||
from config import CATEGORIES_FILE
|
||||
from utils import load_data, save_data
|
||||
from flask import Blueprint, request, jsonify, Response
|
||||
from config import CATEGORIES_FILE, DATA_DIR
|
||||
from utils import load_data, save_data, load_config
|
||||
|
||||
cat_bp = Blueprint('api_categories', __name__)
|
||||
|
||||
@@ -70,3 +71,271 @@ def api_toggle_category_visible(category_id):
|
||||
category['visible'] = not category.get('visible', True)
|
||||
save_data(CATEGORIES_FILE, categories)
|
||||
return jsonify({'success': True, 'visible': category['visible']})
|
||||
|
||||
|
||||
# ─── 导出分类相关数据 ───────────────────────────────────────────────────────
|
||||
|
||||
@cat_bp.route('/api/categories/export', methods=['GET'])
|
||||
def api_export_categories():
|
||||
"""导出所有分类及其相关数据（子类别、参数字段、关联数据项）"""
|
||||
try:
|
||||
# 获取所有分类
|
||||
categories = load_data(CATEGORIES_FILE)
|
||||
|
||||
# 获取各类别下的数据
|
||||
export_data = {
|
||||
'categories': categories,
|
||||
'export_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
||||
'version': '1.0'
|
||||
}
|
||||
|
||||
# 加载各类别数据文件
|
||||
# 内置分类：AI模型、GPU、CPU
|
||||
models_file = DATA_DIR / 'models.json'
|
||||
gpus_file = DATA_DIR / 'gpus.json'
|
||||
cpus_file = DATA_DIR / 'cpus.json'
|
||||
|
||||
if models_file.exists():
|
||||
export_data['models'] = load_data(models_file)
|
||||
if gpus_file.exists():
|
||||
export_data['gpus'] = load_data(gpus_file)
|
||||
if cpus_file.exists():
|
||||
export_data['cpus'] = load_data(cpus_file)
|
||||
|
||||
# 动态分类数据（每个自定义分类有独立的数据文件）
|
||||
for cat in categories:
|
||||
cat_id = cat['id']
|
||||
# 排除内置分类
|
||||
if cat_id in ['ai-models', 'gpus', 'cpus']:
|
||||
continue
|
||||
# 查找该分类的数据文件
|
||||
cat_data_file = DATA_DIR / f'{cat_id}.json'
|
||||
if cat_data_file.exists():
|
||||
export_data[cat_id] = load_data(cat_data_file)
|
||||
|
||||
# 导出为 JSON 文件
|
||||
json_str = json.dumps(export_data, ensure_ascii=False, indent=2)
|
||||
|
||||
response = Response(
|
||||
json_str,
|
||||
mimetype='application/json',
|
||||
headers={
|
||||
'Content-Disposition': f'attachment; filename=param-hub-categories-export-{datetime.now().strftime("%Y%m%d%H%M%S")}.json'
|
||||
}
|
||||
)
|
||||
return response
|
||||
except Exception as e:
|
||||
return jsonify({'error': str(e)}), 500
|
||||
|
||||
|
||||
@cat_bp.route('/api/categories/export/<category_id>', methods=['GET'])
|
||||
def api_export_single_category(category_id):
|
||||
"""导出单个分类及其相关数据"""
|
||||
try:
|
||||
categories = load_data(CATEGORIES_FILE)
|
||||
category = next((c for c in categories if c['id'] == category_id), None)
|
||||
if not category:
|
||||
return jsonify({'error': 'Category not found'}), 404
|
||||
|
||||
export_data = {
|
||||
'category': category,
|
||||
'export_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
||||
'version': '1.0'
|
||||
}
|
||||
|
||||
# 加载该分类的数据
|
||||
if category_id == 'ai-models':
|
||||
models_file = DATA_DIR / 'models.json'
|
||||
if models_file.exists():
|
||||
export_data['items'] = load_data(models_file)
|
||||
elif category_id == 'gpus':
|
||||
gpus_file = DATA_DIR / 'gpus.json'
|
||||
if gpus_file.exists():
|
||||
export_data['items'] = load_data(gpus_file)
|
||||
elif category_id == 'cpus':
|
||||
cpus_file = DATA_DIR / 'cpus.json'
|
||||
if cpus_file.exists():
|
||||
export_data['items'] = load_data(cpus_file)
|
||||
else:
|
||||
# 动态分类
|
||||
cat_data_file = DATA_DIR / f'{category_id}.json'
|
||||
if cat_data_file.exists():
|
||||
export_data['items'] = load_data(cat_data_file)
|
||||
|
||||
json_str = json.dumps(export_data, ensure_ascii=False, indent=2)
|
||||
|
||||
response = Response(
|
||||
json_str,
|
||||
mimetype='application/json',
|
||||
headers={
|
||||
'Content-Disposition': f'attachment; filename={category_id}-export-{datetime.now().strftime("%Y%m%d%H%M%S")}.json'
|
||||
}
|
||||
)
|
||||
return response
|
||||
except Exception as e:
|
||||
return jsonify({'error': str(e)}), 500
|
||||
|
||||
|
||||
# ─── 导入分类相关数据 ───────────────────────────────────────────────────────
|
||||
|
||||
@cat_bp.route('/api/categories/import', methods=['POST'])
|
||||
def api_import_categories():
|
||||
"""导入分类及其相关数据"""
|
||||
try:
|
||||
import_data = request.get_json()
|
||||
|
||||
if not import_data:
|
||||
return jsonify({'error': '无导入数据'}), 400
|
||||
|
||||
# 验证数据结构
|
||||
if 'categories' not in import_data:
|
||||
return jsonify({'error': '缺少 categories 字段'}), 400
|
||||
|
||||
# 导入模式：merge（合并）或 replace（替换）
|
||||
mode = request.args.get('mode', 'merge')
|
||||
|
||||
result = {
|
||||
'success': True,
|
||||
'imported_categories': 0,
|
||||
'imported_items': 0,
|
||||
'skipped': [],
|
||||
'updated': []
|
||||
}
|
||||
|
||||
# 处理分类
|
||||
existing_categories = load_data(CATEGORIES_FILE)
|
||||
imported_categories = import_data['categories']
|
||||
|
||||
for cat in imported_categories:
|
||||
existing = next((c for c in existing_categories if c['id'] == cat['id']), None)
|
||||
if existing:
|
||||
if mode == 'replace':
|
||||
# 替换模式：更新现有分类
|
||||
existing.update(cat)
|
||||
result['updated'].append(cat['id'])
|
||||
else:
|
||||
# 合并模式：跳过已存在的分类
|
||||
result['skipped'].append(cat['id'])
|
||||
else:
|
||||
# 新分类：添加
|
||||
existing_categories.append(cat)
|
||||
result['imported_categories'] += 1
|
||||
|
||||
save_data(CATEGORIES_FILE, existing_categories)
|
||||
|
||||
# 导入各类别数据
|
||||
data_mappings = {
|
||||
'models': 'models.json',
|
||||
'gpus': 'gpus.json',
|
||||
'cpus': 'cpus.json'
|
||||
}
|
||||
|
||||
for key, filename in data_mappings.items():
|
||||
if key in import_data:
|
||||
data_file = DATA_DIR / filename
|
||||
existing_items = load_data(data_file) if data_file.exists() else []
|
||||
imported_items = import_data[key]
|
||||
|
||||
for item in imported_items:
|
||||
existing = next((i for i in existing_items if i['id'] == item['id']), None)
|
||||
if existing:
|
||||
if mode == 'replace':
|
||||
existing.update(item)
|
||||
else:
|
||||
continue # 合并模式跳过
|
||||
else:
|
||||
existing_items.append(item)
|
||||
result['imported_items'] += 1
|
||||
|
||||
save_data(data_file, existing_items)
|
||||
|
||||
# 导入动态分类数据
|
||||
for cat in imported_categories:
|
||||
cat_id = cat['id']
|
||||
if cat_id in ['ai-models', 'gpus', 'cpus']:
|
||||
continue
|
||||
if cat_id in import_data:
|
||||
data_file = DATA_DIR / f'{cat_id}.json'
|
||||
existing_items = load_data(data_file) if data_file.exists() else []
|
||||
imported_items = import_data[cat_id]
|
||||
|
||||
for item in imported_items:
|
||||
existing = next((i for i in existing_items if i['id'] == item['id']), None)
|
||||
if existing:
|
||||
if mode == 'replace':
|
||||
existing.update(item)
|
||||
else:
|
||||
continue
|
||||
else:
|
||||
existing_items.append(item)
|
||||
result['imported_items'] += 1
|
||||
|
||||
save_data(data_file, existing_items)
|
||||
|
||||
return jsonify(result)
|
||||
except Exception as e:
|
||||
return jsonify({'error': str(e)}), 500
|
||||
|
||||
|
||||
@cat_bp.route('/api/categories/import/<category_id>', methods=['POST'])
|
||||
def api_import_single_category(category_id):
|
||||
"""导入单个分类数据"""
|
||||
try:
|
||||
import_data = request.get_json()
|
||||
|
||||
if not import_data:
|
||||
return jsonify({'error': '无导入数据'}), 400
|
||||
|
||||
if 'category' not in import_data:
|
||||
return jsonify({'error': '缺少 category 字段'}), 400
|
||||
|
||||
mode = request.args.get('mode', 'merge')
|
||||
|
||||
result = {
|
||||
'success': True,
|
||||
'imported_items': 0,
|
||||
'updated_items': 0
|
||||
}
|
||||
|
||||
# 处理分类
|
||||
categories = load_data(CATEGORIES_FILE)
|
||||
imported_cat = import_data['category']
|
||||
|
||||
existing = next((c for c in categories if c['id'] == category_id), None)
|
||||
if existing:
|
||||
if mode == 'replace':
|
||||
existing.update(imported_cat)
|
||||
else:
|
||||
categories.append(imported_cat)
|
||||
|
||||
save_data(CATEGORIES_FILE, categories)
|
||||
|
||||
# 处理数据项
|
||||
if 'items' in import_data:
|
||||
if category_id == 'ai-models':
|
||||
data_file = DATA_DIR / 'models.json'
|
||||
elif category_id == 'gpus':
|
||||
data_file = DATA_DIR / 'gpus.json'
|
||||
elif category_id == 'cpus':
|
||||
data_file = DATA_DIR / 'cpus.json'
|
||||
else:
|
||||
data_file = DATA_DIR / f'{category_id}.json'
|
||||
|
||||
existing_items = load_data(data_file) if data_file.exists() else []
|
||||
imported_items = import_data['items']
|
||||
|
||||
for item in imported_items:
|
||||
existing = next((i for i in existing_items if i['id'] == item['id']), None)
|
||||
if existing:
|
||||
if mode == 'replace':
|
||||
existing.update(item)
|
||||
result['updated_items'] += 1
|
||||
else:
|
||||
existing_items.append(item)
|
||||
result['imported_items'] += 1
|
||||
|
||||
save_data(data_file, existing_items)
|
||||
|
||||
return jsonify(result)
|
||||
except Exception as e:
|
||||
return jsonify({'error': str(e)}), 500
|
||||
Reference in New Issue
Block a user