72 lines
2.7 KiB
Python
72 lines
2.7 KiB
Python
"""Attachment validation, storage and retention without request parsing."""
|
|
|
|
import os
|
|
import secrets
|
|
import re
|
|
import time
|
|
from datetime import datetime
|
|
from urllib.parse import unquote
|
|
|
|
from .config import get_setting
|
|
|
|
|
|
|
|
def cleanup_uploads():
|
|
"""Keep attachments referenced by chat history; expire unused uploads after 7 days."""
|
|
try:
|
|
udir = get_setting('UPLOAD_DIR')
|
|
if not os.path.isdir(udir):
|
|
return
|
|
from .db import connect
|
|
connection = connect()
|
|
try:
|
|
referenced = set()
|
|
rows = connection.execute("SELECT content FROM messages WHERE content LIKE '%/uploads/%' "
|
|
"UNION ALL SELECT content FROM human_messages WHERE content LIKE '%/uploads/%'")
|
|
for (content,) in rows:
|
|
referenced.update(unquote(name) for name in re.findall(r'/uploads/([A-Za-z0-9_.%-]+)', content or ''))
|
|
finally:
|
|
connection.close()
|
|
cutoff = time.time() - 7 * 86400
|
|
for f in os.listdir(udir):
|
|
if f in referenced:
|
|
continue
|
|
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}
|