223 lines
7.0 KiB
Python
223 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from datetime import datetime
|
|
from typing import Iterator
|
|
|
|
import database
|
|
from app_config import get_config
|
|
from app_security import is_safe_path
|
|
from flask import Blueprint, jsonify, send_from_directory
|
|
from flask_login import current_user, login_required
|
|
from PIL import Image, ImageOps
|
|
from services.client_log import log_to_client
|
|
from services.time_utils import BEIJING_TZ
|
|
|
|
config = get_config()
|
|
SCREENSHOTS_DIR = config.SCREENSHOTS_DIR
|
|
_IMAGE_EXTENSIONS = (".png", ".jpg", ".jpeg")
|
|
_THUMBNAIL_DIR = os.path.join(SCREENSHOTS_DIR, ".thumbs")
|
|
_THUMBNAIL_MAX_SIZE = (480, 270)
|
|
_THUMBNAIL_QUALITY = 80
|
|
|
|
try:
|
|
_RESAMPLE_FILTER = Image.Resampling.LANCZOS
|
|
except AttributeError: # Pillow<9 fallback
|
|
_RESAMPLE_FILTER = Image.LANCZOS
|
|
|
|
api_screenshots_bp = Blueprint("api_screenshots", __name__)
|
|
|
|
|
|
def _get_user_prefix(user_id: int) -> str:
|
|
user_info = database.get_user_by_id(user_id)
|
|
return user_info["username"] if user_info else f"user{user_id}"
|
|
|
|
|
|
def _is_user_screenshot(filename: str, username_prefix: str) -> bool:
|
|
return filename.startswith(username_prefix + "_") and filename.lower().endswith(_IMAGE_EXTENSIONS)
|
|
|
|
|
|
def _iter_user_screenshot_entries(username_prefix: str) -> Iterator[os.DirEntry]:
|
|
if not os.path.exists(SCREENSHOTS_DIR):
|
|
return
|
|
|
|
with os.scandir(SCREENSHOTS_DIR) as entries:
|
|
for entry in entries:
|
|
if (not entry.is_file()) or (not _is_user_screenshot(entry.name, username_prefix)):
|
|
continue
|
|
yield entry
|
|
|
|
|
|
def _build_display_name(filename: str) -> str:
|
|
base_name, ext = filename.rsplit(".", 1)
|
|
parts = base_name.split("_", 1)
|
|
if len(parts) > 1:
|
|
return f"{parts[1]}.{ext}"
|
|
return filename
|
|
|
|
|
|
def _thumbnail_name(filename: str) -> str:
|
|
stem, _ = os.path.splitext(filename)
|
|
return f"{stem}.thumb.jpg"
|
|
|
|
|
|
def _thumbnail_path(filename: str) -> str:
|
|
return os.path.join(_THUMBNAIL_DIR, _thumbnail_name(filename))
|
|
|
|
|
|
def _ensure_thumbnail(source_path: str, thumb_path: str) -> bool:
|
|
if not os.path.exists(source_path):
|
|
return False
|
|
|
|
source_mtime = os.path.getmtime(source_path)
|
|
if os.path.exists(thumb_path) and os.path.getmtime(thumb_path) >= source_mtime:
|
|
return True
|
|
|
|
os.makedirs(_THUMBNAIL_DIR, exist_ok=True)
|
|
|
|
with Image.open(source_path) as image:
|
|
image = ImageOps.exif_transpose(image)
|
|
if image.mode != "RGB":
|
|
image = image.convert("RGB")
|
|
image.thumbnail(_THUMBNAIL_MAX_SIZE, _RESAMPLE_FILTER)
|
|
image.save(
|
|
thumb_path,
|
|
format="JPEG",
|
|
quality=_THUMBNAIL_QUALITY,
|
|
optimize=True,
|
|
progressive=True,
|
|
)
|
|
|
|
os.utime(thumb_path, (source_mtime, source_mtime))
|
|
return True
|
|
|
|
|
|
def _remove_thumbnail(filename: str) -> None:
|
|
thumb_path = _thumbnail_path(filename)
|
|
if os.path.exists(thumb_path):
|
|
os.remove(thumb_path)
|
|
|
|
|
|
@api_screenshots_bp.route("/api/screenshots", methods=["GET"])
|
|
@login_required
|
|
def get_screenshots():
|
|
"""获取当前用户的截图列表"""
|
|
user_id = current_user.id
|
|
username_prefix = _get_user_prefix(user_id)
|
|
|
|
try:
|
|
screenshots = []
|
|
for entry in _iter_user_screenshot_entries(username_prefix):
|
|
filename = entry.name
|
|
stat = entry.stat()
|
|
created_time = datetime.fromtimestamp(stat.st_mtime, tz=BEIJING_TZ)
|
|
|
|
screenshots.append(
|
|
{
|
|
"filename": filename,
|
|
"display_name": _build_display_name(filename),
|
|
"size": stat.st_size,
|
|
"created": created_time.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"_created_ts": stat.st_mtime,
|
|
}
|
|
)
|
|
|
|
screenshots.sort(key=lambda item: item.get("_created_ts", 0), reverse=True)
|
|
for item in screenshots:
|
|
item.pop("_created_ts", None)
|
|
|
|
return jsonify(screenshots)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@api_screenshots_bp.route("/screenshots/<filename>")
|
|
@login_required
|
|
def serve_screenshot(filename):
|
|
"""提供原图文件访问"""
|
|
user_id = current_user.id
|
|
username_prefix = _get_user_prefix(user_id)
|
|
|
|
if not _is_user_screenshot(filename, username_prefix):
|
|
return jsonify({"error": "无权访问"}), 403
|
|
|
|
if not is_safe_path(SCREENSHOTS_DIR, filename):
|
|
return jsonify({"error": "非法路径"}), 403
|
|
|
|
return send_from_directory(SCREENSHOTS_DIR, filename)
|
|
|
|
|
|
@api_screenshots_bp.route("/screenshots/thumb/<filename>")
|
|
@login_required
|
|
def serve_screenshot_thumbnail(filename):
|
|
"""提供缩略图访问（失败时自动回退原图）"""
|
|
user_id = current_user.id
|
|
username_prefix = _get_user_prefix(user_id)
|
|
|
|
if not _is_user_screenshot(filename, username_prefix):
|
|
return jsonify({"error": "无权访问"}), 403
|
|
|
|
if not is_safe_path(SCREENSHOTS_DIR, filename):
|
|
return jsonify({"error": "非法路径"}), 403
|
|
|
|
source_path = os.path.join(SCREENSHOTS_DIR, filename)
|
|
if not os.path.exists(source_path):
|
|
return jsonify({"error": "文件不存在"}), 404
|
|
|
|
thumb_path = _thumbnail_path(filename)
|
|
|
|
try:
|
|
if _ensure_thumbnail(source_path, thumb_path) and os.path.exists(thumb_path):
|
|
return send_from_directory(_THUMBNAIL_DIR, os.path.basename(thumb_path), max_age=86400, conditional=True)
|
|
except Exception:
|
|
pass
|
|
|
|
return send_from_directory(SCREENSHOTS_DIR, filename, max_age=3600, conditional=True)
|
|
|
|
|
|
@api_screenshots_bp.route("/api/screenshots/<filename>", methods=["DELETE"])
|
|
@login_required
|
|
def delete_screenshot(filename):
|
|
"""删除指定截图"""
|
|
user_id = current_user.id
|
|
username_prefix = _get_user_prefix(user_id)
|
|
|
|
if not _is_user_screenshot(filename, username_prefix):
|
|
return jsonify({"error": "无权删除"}), 403
|
|
|
|
if not is_safe_path(SCREENSHOTS_DIR, filename):
|
|
return jsonify({"error": "非法路径"}), 403
|
|
|
|
try:
|
|
filepath = os.path.join(SCREENSHOTS_DIR, filename)
|
|
if os.path.exists(filepath):
|
|
os.remove(filepath)
|
|
_remove_thumbnail(filename)
|
|
log_to_client(f"删除截图: {filename}", user_id)
|
|
return jsonify({"success": True})
|
|
return jsonify({"error": "文件不存在"}), 404
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@api_screenshots_bp.route("/api/screenshots/clear", methods=["POST"])
|
|
@login_required
|
|
def clear_all_screenshots():
|
|
"""清空当前用户的所有截图"""
|
|
user_id = current_user.id
|
|
username_prefix = _get_user_prefix(user_id)
|
|
|
|
try:
|
|
deleted_count = 0
|
|
for entry in _iter_user_screenshot_entries(username_prefix):
|
|
os.remove(entry.path)
|
|
_remove_thumbnail(entry.name)
|
|
deleted_count += 1
|
|
|
|
log_to_client(f"清理了 {deleted_count} 个截图文件", user_id)
|
|
return jsonify({"success": True, "deleted": deleted_count})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|