145 lines
5.0 KiB
Python
145 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import posixpath
|
||
import secrets
|
||
import time
|
||
|
||
import database
|
||
from app_config import get_config
|
||
from app_logger import get_logger
|
||
from app_security import is_safe_path, sanitize_filename
|
||
from flask import current_app, jsonify, request, url_for
|
||
from routes.admin_api import admin_api_bp
|
||
from routes.decorators import admin_required
|
||
|
||
logger = get_logger("app")
|
||
config = get_config()
|
||
|
||
|
||
def _get_upload_dir():
|
||
rel_dir = getattr(config, "ANNOUNCEMENT_IMAGE_DIR", "static/announcements")
|
||
if not is_safe_path(current_app.root_path, rel_dir):
|
||
rel_dir = "static/announcements"
|
||
abs_dir = os.path.join(current_app.root_path, rel_dir)
|
||
os.makedirs(abs_dir, exist_ok=True)
|
||
return abs_dir, rel_dir
|
||
|
||
|
||
def _get_file_size(file_storage):
|
||
try:
|
||
file_storage.stream.seek(0, os.SEEK_END)
|
||
size = file_storage.stream.tell()
|
||
file_storage.stream.seek(0)
|
||
return size
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
# ==================== 公告管理API（管理员） ====================
|
||
|
||
|
||
@admin_api_bp.route("/announcements/upload_image", methods=["POST"])
|
||
@admin_required
|
||
def admin_upload_announcement_image():
|
||
"""上传公告图片（返回可访问URL）"""
|
||
file = request.files.get("file")
|
||
if not file or not file.filename:
|
||
return jsonify({"error": "请选择图片"}), 400
|
||
|
||
filename = sanitize_filename(file.filename)
|
||
ext = os.path.splitext(filename)[1].lower()
|
||
allowed_exts = getattr(config, "ALLOWED_ANNOUNCEMENT_IMAGE_EXTENSIONS", {".png", ".jpg", ".jpeg"})
|
||
if not ext or ext not in allowed_exts:
|
||
return jsonify({"error": "不支持的图片格式"}), 400
|
||
if file.mimetype and not str(file.mimetype).startswith("image/"):
|
||
return jsonify({"error": "文件类型无效"}), 400
|
||
|
||
size = _get_file_size(file)
|
||
max_size = int(getattr(config, "MAX_ANNOUNCEMENT_IMAGE_SIZE", 5 * 1024 * 1024))
|
||
if size is not None and size > max_size:
|
||
max_mb = max_size // 1024 // 1024
|
||
return jsonify({"error": f"图片大小不能超过{max_mb}MB"}), 400
|
||
|
||
abs_dir, rel_dir = _get_upload_dir()
|
||
token = secrets.token_hex(6)
|
||
name = f"announcement_{int(time.time())}_{token}{ext}"
|
||
save_path = os.path.join(abs_dir, name)
|
||
file.save(save_path)
|
||
|
||
static_root = os.path.join(current_app.root_path, "static")
|
||
rel_to_static = os.path.relpath(abs_dir, static_root)
|
||
if rel_to_static.startswith(".."):
|
||
rel_to_static = "announcements"
|
||
url_path = posixpath.join(rel_to_static.replace(os.sep, "/"), name)
|
||
return jsonify({"success": True, "url": url_for("serve_static", filename=url_path)})
|
||
|
||
|
||
@admin_api_bp.route("/announcements", methods=["GET"])
|
||
@admin_required
|
||
def admin_get_announcements():
|
||
"""获取公告列表"""
|
||
try:
|
||
limit = int(request.args.get("limit", 50))
|
||
offset = int(request.args.get("offset", 0))
|
||
except (TypeError, ValueError):
|
||
limit, offset = 50, 0
|
||
|
||
limit = max(1, min(200, limit))
|
||
offset = max(0, offset)
|
||
return jsonify(database.get_announcements(limit=limit, offset=offset))
|
||
|
||
|
||
@admin_api_bp.route("/announcements", methods=["POST"])
|
||
@admin_required
|
||
def admin_create_announcement():
|
||
"""创建公告（默认启用并替换旧公告）"""
|
||
data = request.json or {}
|
||
title = (data.get("title") or "").strip()
|
||
content = (data.get("content") or "").strip()
|
||
image_url = (data.get("image_url") or "").strip()
|
||
is_active = bool(data.get("is_active", True))
|
||
|
||
if image_url and len(image_url) > 1000:
|
||
return jsonify({"error": "图片地址过长"}), 400
|
||
|
||
announcement_id = database.create_announcement(title, content, image_url=image_url, is_active=is_active)
|
||
if not announcement_id:
|
||
return jsonify({"error": "标题和内容不能为空"}), 400
|
||
|
||
return jsonify({"success": True, "id": announcement_id})
|
||
|
||
|
||
@admin_api_bp.route("/announcements/<int:announcement_id>/activate", methods=["POST"])
|
||
@admin_required
|
||
def admin_activate_announcement(announcement_id):
|
||
"""启用公告（会自动停用其他公告）"""
|
||
if not database.get_announcement_by_id(announcement_id):
|
||
return jsonify({"error": "公告不存在"}), 404
|
||
ok = database.set_announcement_active(announcement_id, True)
|
||
return jsonify({"success": ok})
|
||
|
||
|
||
@admin_api_bp.route("/announcements/<int:announcement_id>/deactivate", methods=["POST"])
|
||
@admin_required
|
||
def admin_deactivate_announcement(announcement_id):
|
||
"""停用公告"""
|
||
if not database.get_announcement_by_id(announcement_id):
|
||
return jsonify({"error": "公告不存在"}), 404
|
||
ok = database.set_announcement_active(announcement_id, False)
|
||
return jsonify({"success": ok})
|
||
|
||
|
||
@admin_api_bp.route("/announcements/<int:announcement_id>", methods=["DELETE"])
|
||
@admin_required
|
||
def admin_delete_announcement(announcement_id):
|
||
"""删除公告"""
|
||
if not database.get_announcement_by_id(announcement_id):
|
||
return jsonify({"error": "公告不存在"}), 404
|
||
ok = database.delete_announcement(announcement_id)
|
||
return jsonify({"success": ok})
|
||
|
||
|