187 lines
6.9 KiB
Python
187 lines
6.9 KiB
Python
"""
|
|
动态分类数据 API
|
|
"""
|
|
import uuid
|
|
import json
|
|
from datetime import datetime
|
|
from flask import Blueprint, request, jsonify, Response
|
|
from config import DATA_DIR
|
|
from utils import load_data, save_data, parse_date_to_timestamp
|
|
|
|
items_bp = Blueprint('api_items', __name__)
|
|
|
|
|
|
def is_review_required():
|
|
try:
|
|
import app as app_module
|
|
return getattr(app_module, 'REQUIRE_REVIEW', False)
|
|
except:
|
|
return False
|
|
|
|
|
|
@items_bp.route('/api/items/<category_id>')
|
|
def api_items(category_id):
|
|
items_file = DATA_DIR / f'items_{category_id}.json'
|
|
items = load_data(items_file)
|
|
hide_hidden = request.args.get('all', '0') == '0'
|
|
if hide_hidden:
|
|
items = [i for i in items if i.get('visible', True)]
|
|
sort_by = request.args.get('sort', 'default')
|
|
reverse = request.args.get('order', 'desc') == 'desc'
|
|
def safe_key(x, key):
|
|
val = x.get(key)
|
|
if val is None:
|
|
return 0 if key in ['price', 'views', 'year'] else ''
|
|
return val
|
|
if sort_by == 'default':
|
|
items = sorted(items, key=lambda x: (
|
|
not x.get('is_pinned', False),
|
|
-(parse_date_to_timestamp(x.get('publish_date', '')) or
|
|
parse_date_to_timestamp(x.get('created_at', '')) or 0)
|
|
))
|
|
elif sort_by in ['name', 'price', 'year', 'created_at', 'publish_date', 'views', 'updated_at']:
|
|
items = sorted(items, key=lambda x: safe_key(x, sort_by), reverse=reverse)
|
|
return jsonify(items)
|
|
|
|
|
|
@items_bp.route('/api/items/<category_id>/<item_id>')
|
|
def api_item_detail(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
|
|
return jsonify(item)
|
|
|
|
|
|
@items_bp.route('/api/items/<category_id>', methods=['POST'])
|
|
def api_create_item(category_id):
|
|
data = request.get_json()
|
|
|
|
# 审核模式
|
|
if is_review_required():
|
|
from modules.routes.api_reviews import submit_for_review
|
|
from modules.routes.api_notifications import create_notification
|
|
|
|
review = submit_for_review(category_id, data, source='web')
|
|
product_name = data.get('name', '未知')
|
|
create_notification(
|
|
title='新产品待审核',
|
|
message=f'有新的产品"{product_name}"待审核',
|
|
level='warning',
|
|
category='review',
|
|
data={'review_id': review['id'], 'category': category_id}
|
|
)
|
|
return jsonify({'success': True, 'message': '已提交审核，请等待管理员确认', 'review_id': review['id']})
|
|
|
|
items_file = DATA_DIR / f'items_{category_id}.json'
|
|
items = load_data(items_file)
|
|
data['id'] = uuid.uuid4().hex[:12]
|
|
data['category_id'] = category_id
|
|
data['created_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
data['visible'] = data.get('visible', True)
|
|
data['publish_date'] = data.get('publish_date', '')
|
|
data['views'] = data.get('views', 0)
|
|
data['is_pinned'] = data.get('is_pinned', False)
|
|
items.append(data)
|
|
save_data(items_file, items)
|
|
return jsonify(data)
|
|
|
|
|
|
@items_bp.route('/api/items/<category_id>/<item_id>', methods=['PUT'])
|
|
def api_update_item(category_id, item_id):
|
|
data = request.get_json()
|
|
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
|
|
item.update(data)
|
|
item['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
save_data(items_file, items)
|
|
return jsonify(item)
|
|
|
|
|
|
@items_bp.route('/api/items/<category_id>/<item_id>', methods=['DELETE'])
|
|
def api_delete_item(category_id, item_id):
|
|
items_file = DATA_DIR / f'items_{category_id}.json'
|
|
items = load_data(items_file)
|
|
items = [i for i in items if i['id'] != item_id]
|
|
save_data(items_file, items)
|
|
return jsonify({'success': True})
|
|
|
|
|
|
@items_bp.route('/api/items/<category_id>/<item_id>/visible', methods=['POST'])
|
|
def api_toggle_item_visible(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
|
|
item['visible'] = not item.get('visible', True)
|
|
save_data(items_file, items)
|
|
return jsonify({'success': True, 'visible': item['visible']})
|
|
|
|
|
|
# ─── 导出导入 ───────────────────────────────────────────────────────
|
|
|
|
@items_bp.route('/api/items/<category_id>/export', methods=['GET'])
|
|
def api_export_items(category_id):
|
|
"""导出指定分类的所有数据"""
|
|
try:
|
|
items_file = DATA_DIR / f'items_{category_id}.json'
|
|
items = load_data(items_file)
|
|
export_data = {
|
|
'type': 'items',
|
|
'category_id': category_id,
|
|
'items': items,
|
|
'count': len(items),
|
|
'export_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
|
'version': '1.0'
|
|
}
|
|
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
|
|
|
|
|
|
@items_bp.route('/api/items/<category_id>/import', methods=['GET', 'POST'])
|
|
def api_import_items(category_id):
|
|
"""导入指定分类的数据"""
|
|
try:
|
|
if request.method == 'GET':
|
|
return jsonify({'endpoint': f'/api/items/{category_id}/import', 'method': 'POST', 'params': {'mode': 'merge 或 replace'}})
|
|
|
|
import_data = request.get_json()
|
|
if not import_data or 'items' not in import_data:
|
|
return jsonify({'error': '缺少 items 字段'}), 400
|
|
|
|
mode = request.args.get('mode', 'merge')
|
|
items_file = DATA_DIR / f'items_{category_id}.json'
|
|
items = load_data(items_file)
|
|
imported_items = import_data['items']
|
|
|
|
result = {'success': True, 'imported': 0, 'updated': 0, 'skipped': []}
|
|
|
|
for item in imported_items:
|
|
existing = next((i for i in items if i['id'] == item['id']), None)
|
|
if existing:
|
|
if mode == 'replace':
|
|
existing.update(item)
|
|
result['updated'] += 1
|
|
else:
|
|
result['skipped'].append(item['id'])
|
|
else:
|
|
items.append(item)
|
|
result['imported'] += 1
|
|
|
|
save_data(items_file, items)
|
|
return jsonify(result)
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|