hz4th_coder/param-hub-python
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

简化分类导出导入，仅处理分类本身不含数据项

Browse Source
This commit is contained in:
hz4th_coder
2026-07-11 00:14:08 +08:00
parent bc2720257d
commit f83bf11669
3 changed files with 38 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/routes/__pycache__/api_categories.cpython-310.pyc
BIN
View File
Binary file not shown.
modules/routes/api_categories.py
+34 -148
View File
@@ -5,8 +5,8 @@ import uuid
import json
from datetime import datetime
from flask import Blueprint, request, jsonify, Response
from config import CATEGORIES_FILE, DATA_DIR
from utils import load_data, save_data, load_config
from config import CATEGORIES_FILE
from utils import load_data, save_data
cat_bp = Blueprint('api_categories', __name__)
@@ -73,54 +73,27 @@ def api_toggle_category_visible(category_id):
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'
'Content-Disposition': f'attachment; filename=param-hub-categories-{datetime.now().strftime("%Y%m%d%H%M%S")}.json'
}
)
return response
@@ -130,7 +103,7 @@ def api_export_categories():
@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)
@@ -143,32 +116,13 @@ def api_export_single_category(category_id):
'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'
'Content-Disposition': f'attachment; filename={category_id}-{datetime.now().strftime("%Y%m%d%H%M%S")}.json'
}
)
return response
@@ -176,33 +130,42 @@ def api_export_single_category(category_id):
return jsonify({'error': str(e)}), 500
# ─── 导入分类相关数据 ───────────────────────────────────────────────────────
# ─── 导入分类数据 ───────────────────────────────────────────────────────
@cat_bp.route('/api/categories/import', methods=['POST'])
@cat_bp.route('/api/categories/import', methods=['GET', 'POST'])
def api_import_categories():
"""导入分类及其相关数据"""
"""导入分类配置（仅分类本身，不含数据项）"""
try:
if request.method == 'GET':
# 返回导入接口说明
return jsonify({
'endpoint': '/api/categories/import',
'method': 'POST',
'params': {
'mode': 'merge（合并）或 replace（替换）'
},
'body': {
'categories': '分类数组'
}
})
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': []
'imported': 0,
'updated': 0,
'skipped': []
}
# 处理分类
existing_categories = load_data(CATEGORIES_FILE)
imported_categories = import_data['categories']
@@ -210,68 +173,16 @@ def api_import_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'])
result['updated'] += 1
else:
# 合并模式：跳过已存在的分类
result['skipped'].append(cat['id'])
else:
# 新分类：添加
existing_categories.append(cat)
result['imported_categories'] += 1
result['imported'] += 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
@@ -279,7 +190,7 @@ def api_import_categories():
@cat_bp.route('/api/categories/import/<category_id>', methods=['POST'])
def api_import_single_category(category_id):
"""导入单个分类数据"""
"""导入单个分类配置"""
try:
import_data = request.get_json()
@@ -293,11 +204,10 @@ def api_import_single_category(category_id):
result = {
'success': True,
'imported_items': 0,
'updated_items': 0
'imported': False,
'updated': False
}
# 处理分类
categories = load_data(CATEGORIES_FILE)
imported_cat = import_data['category']
@@ -305,37 +215,13 @@ def api_import_single_category(category_id):
if existing:
if mode == 'replace':
existing.update(imported_cat)
result['updated'] = True
else:
categories.append(imported_cat)
result['imported'] = True
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
return jsonify({'error': str(e)}), 500