Files
ai-app-database/app/utils/file_handler.py
T
huty f103148ebf feat: 初始化个人资料库 Web 应用
基于 Flask + MySQL + Bootstrap 5 的全栈个人资料库管理系统。

主要功能:
- 管理员/普通用户双角色权限体系,全站登录保护
- 资源管理:文本、图片、音频、视频四类资源
- 三种添加方式:本地上传(拖拽)、URL 后台下载、磁力下载(aria2c)
- 在线预览:文本、图片、HTML5 音视频播放器
- 安全:bcrypt 加盐密码哈希、CSRF 防护、SQLAlchemy ORM 防注入

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-23 00:16:59 +09:00

108 lines
3.5 KiB
Python

import os
import uuid
import mimetypes
from flask import current_app
from werkzeug.utils import secure_filename
# MIME 类型到资源类型的映射
MIME_TYPE_MAP = {
# 文本
'text/plain': 'text',
'text/markdown': 'text',
'text/csv': 'text',
'text/html': 'text',
'text/xml': 'text',
'application/json': 'text',
'application/xml': 'text',
# 图片
'image/jpeg': 'image',
'image/png': 'image',
'image/gif': 'image',
'image/webp': 'image',
'image/bmp': 'image',
'image/svg+xml': 'image',
'image/x-icon': 'image',
# 音频
'audio/mpeg': 'audio',
'audio/wav': 'audio',
'audio/ogg': 'audio',
'audio/flac': 'audio',
'audio/mp4': 'audio',
'audio/aac': 'audio',
'audio/x-ms-wma': 'audio',
# 视频
'video/mp4': 'video',
'video/webm': 'video',
'video/x-msvideo': 'video',
'video/x-matroska': 'video',
'video/quicktime': 'video',
'video/x-ms-wmv': 'video',
'video/x-flv': 'video',
'video/mp2t': 'video',
}
EXT_TYPE_MAP = {
'txt': 'text', 'md': 'text', 'csv': 'text', 'json': 'text',
'xml': 'text', 'log': 'text', 'html': 'text', 'htm': 'text',
'jpg': 'image', 'jpeg': 'image', 'png': 'image', 'gif': 'image',
'webp': 'image', 'bmp': 'image', 'svg': 'image', 'ico': 'image',
'mp3': 'audio', 'wav': 'audio', 'ogg': 'audio', 'flac': 'audio',
'm4a': 'audio', 'aac': 'audio', 'wma': 'audio',
'mp4': 'video', 'webm': 'video', 'avi': 'video', 'mkv': 'video',
'mov': 'video', 'wmv': 'video', 'flv': 'video', 'm4v': 'video',
}
def guess_resource_type(filename, mime=None):
"""根据 MIME 或扩展名猜测资源类型"""
if mime and mime in MIME_TYPE_MAP:
return MIME_TYPE_MAP[mime]
ext = filename.rsplit('.', 1)[-1].lower() if '.' in filename else ''
return EXT_TYPE_MAP.get(ext)
def allowed_file(filename):
"""判断文件扩展名是否在允许列表中"""
if '.' not in filename:
return False
ext = filename.rsplit('.', 1)[1].lower()
return ext in current_app.config.get('ALLOWED_TEXT_EXT', set()) | \
current_app.config.get('ALLOWED_IMAGE_EXT', set()) | \
current_app.config.get('ALLOWED_AUDIO_EXT', set()) | \
current_app.config.get('ALLOWED_VIDEO_EXT', set())
def save_uploaded_file(file_obj, resource_type):
"""
保存上传的文件到 uploads/<resource_type>/ 目录。
返回 (filename_on_disk, relative_path, file_size, mime_type)
"""
original_name = secure_filename(file_obj.filename)
ext = original_name.rsplit('.', 1)[-1].lower() if '.' in original_name else ''
unique_name = f"{uuid.uuid4().hex}.{ext}" if ext else uuid.uuid4().hex
save_dir = os.path.join(current_app.config['UPLOAD_FOLDER'], resource_type)
os.makedirs(save_dir, exist_ok=True)
save_path = os.path.join(save_dir, unique_name)
file_obj.save(save_path)
file_size = os.path.getsize(save_path)
mime_type = mimetypes.guess_type(original_name)[0] or 'application/octet-stream'
# 相对于 static 的路径,供 url_for('static', filename=...) 使用
rel_path = os.path.join('uploads', resource_type, unique_name).replace('\\', '/')
return unique_name, rel_path, file_size, mime_type, original_name
def delete_resource_file(file_path):
"""删除资源文件,file_path 是相对于 static 的路径"""
if not file_path:
return
abs_path = os.path.join(
current_app.root_path, 'static', file_path.lstrip('/')
)
if os.path.exists(abs_path):
os.remove(abs_path)