forked from jiaoly/financial_system
62 lines
2.8 KiB
Python
62 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import shutil
|
|
from pathlib import Path
|
|
from typing import BinaryIO
|
|
from uuid import uuid4
|
|
|
|
from ..core.logger import AppLogger
|
|
from ..core.settings import AppSettings
|
|
|
|
logger = AppLogger.get_logger(__name__)
|
|
|
|
ALLOWED_AVATAR_SUFFIXES = {".jpg", ".jpeg", ".png", ".webp", ".gif"}
|
|
|
|
|
|
class FileStorage:
|
|
"""统一管理上传文件和导出文件路径，避免路由层直接操作磁盘。"""
|
|
|
|
def __init__(self, settings: AppSettings):
|
|
self.settings = settings
|
|
|
|
def save_upload(self, filename: str, stream: BinaryIO) -> Path:
|
|
if not filename.lower().endswith(".xlsx"):
|
|
logger.error("上传文件格式错误 filename=%s", filename)
|
|
raise ValueError("请上传 .xlsx 格式的月度汇总 Excel")
|
|
self.settings.upload_dir.mkdir(parents=True, exist_ok=True)
|
|
path = self.settings.upload_dir / f"attendance_{uuid4().hex}{Path(filename).suffix}"
|
|
with path.open("wb") as buffer:
|
|
shutil.copyfileobj(stream, buffer)
|
|
logger.info("上传文件保存完成 filename=%s path=%s", filename, path)
|
|
return path
|
|
|
|
def payroll_output_path(self, source_type: str) -> Path:
|
|
self.settings.output_dir.mkdir(parents=True, exist_ok=True)
|
|
path = self.settings.output_dir / f"payroll_{source_type}_{uuid4().hex}.xlsx"
|
|
logger.info("工资结果输出路径已生成 source_type=%s path=%s", source_type, path)
|
|
return path
|
|
|
|
def resolve_output_file(self, filename: str) -> Path:
|
|
"""只允许下载输出目录内文件，避免路径穿越。"""
|
|
path = (self.settings.output_dir / Path(filename).name).resolve()
|
|
output_dir = self.settings.output_dir.resolve()
|
|
if output_dir not in path.parents or not path.is_file():
|
|
logger.error("下载文件不存在或越权 filename=%s resolved_path=%s", filename, path)
|
|
raise FileNotFoundError("文件不存在")
|
|
logger.info("下载文件解析完成 filename=%s path=%s", filename, path)
|
|
return path
|
|
|
|
def save_avatar(self, filename: str, stream: BinaryIO) -> str:
|
|
"""保存用户头像并返回可供前端访问的静态资源路径。"""
|
|
suffix = Path(filename).suffix.lower()
|
|
if suffix not in ALLOWED_AVATAR_SUFFIXES:
|
|
logger.error("头像文件格式错误 filename=%s", filename)
|
|
raise ValueError("请上传 jpg、jpeg、png、webp 或 gif 格式头像")
|
|
|
|
self.settings.avatar_dir.mkdir(parents=True, exist_ok=True)
|
|
path = self.settings.avatar_dir / f"avatar_{uuid4().hex}{suffix}"
|
|
with path.open("wb") as buffer:
|
|
shutil.copyfileobj(stream, buffer)
|
|
logger.info("头像文件保存完成 filename=%s path=%s", filename, path)
|
|
return f"/static/uploads/avatars/{path.name}"
|