58 lines
2.1 KiB
Python
58 lines
2.1 KiB
Python
"""Attachment validation, storage and retention without request parsing."""
|
|
|
|
import os
|
|
import secrets
|
|
import time
|
|
from datetime import datetime
|
|
|
|
from .config import get_setting
|
|
|
|
|
|
|
|
def cleanup_uploads():
|
|
"""删除 uploads/ 里 7 天前的图片"""
|
|
try:
|
|
udir = get_setting('UPLOAD_DIR')
|
|
if not os.path.isdir(udir):
|
|
return
|
|
cutoff = time.time() - 7 * 86400
|
|
for f in os.listdir(udir):
|
|
fp = os.path.join(udir, f)
|
|
if os.path.isfile(fp) and os.path.getmtime(fp) < cutoff:
|
|
os.remove(fp)
|
|
except Exception as e:
|
|
print(f'[清理上传文件失败] {e}')
|
|
|
|
|
|
def upload_file(files):
|
|
"""上传附件：保存到 uploads 目录，返回文件名与访问地址"""
|
|
if 'file' not in files:
|
|
return ({'ok': False, 'error': '没有文件'}, 400)
|
|
f = files['file']
|
|
if not f or not f.filename:
|
|
return ({'ok': False, 'error': '文件名为空'}, 400)
|
|
ext = os.path.splitext(f.filename)[1].lower()
|
|
if ext not in get_setting('ALLOWED_EXT'):
|
|
return ({'ok': False, 'error': f'不支持的格式 {ext}，仅支持图片/PDF/TXT/CSV'}, 400)
|
|
f.seek(0, os.SEEK_END)
|
|
size = f.tell()
|
|
f.seek(0)
|
|
if size > get_setting('MAX_UPLOAD_MB') * 1024 * 1024:
|
|
return ({'ok': False, 'error': f"文件超过 {get_setting('MAX_UPLOAD_MB')}MB"}, 400)
|
|
os.makedirs(get_setting('UPLOAD_DIR'), exist_ok=True)
|
|
safe_name = secrets.token_hex(16) + ext
|
|
f.save(os.path.join(get_setting('UPLOAD_DIR'), safe_name))
|
|
from .db import connect, now_str
|
|
from .visitor_service import visitor_id
|
|
owner = visitor_id()
|
|
if owner:
|
|
connection = connect()
|
|
try:
|
|
with connection:
|
|
connection.execute('INSERT INTO visitor_uploads(filename,visitor_id,ctime) VALUES(?,?,?)',
|
|
(safe_name, owner, now_str()))
|
|
finally:
|
|
connection.close()
|
|
is_image = ext in ('.png', '.jpg', '.jpeg', '.gif', '.webp')
|
|
return {'ok': True, 'name': f.filename, 'url': f'/uploads/{safe_name}', 'is_image': is_image}
|