1. 取消TEMP静态目录挂载
2. 修复用户数据导出功能 3. 生成 static 静态目录将时间戳改为日期
This commit is contained in:
parent
3adfa91560
commit
cab33828dd
@ -11,7 +11,7 @@ from fastapi.security import OAuth2PasswordBearer
|
|||||||
"""
|
"""
|
||||||
系统版本
|
系统版本
|
||||||
"""
|
"""
|
||||||
VERSION = "3.6.5"
|
VERSION = "3.7.0"
|
||||||
|
|
||||||
"""安全警告: 不要在生产中打开调试运行!"""
|
"""安全警告: 不要在生产中打开调试运行!"""
|
||||||
DEBUG = False
|
DEBUG = False
|
||||||
@ -67,13 +67,9 @@ ACCESS_TOKEN_CACHE_MINUTES = 30
|
|||||||
|
|
||||||
"""
|
"""
|
||||||
挂载临时文件目录,并添加路由访问,此路由不会在接口文档中显示
|
挂载临时文件目录,并添加路由访问,此路由不会在接口文档中显示
|
||||||
TEMP_ENABLE:是否启用临时文件目录访问
|
|
||||||
TEMP_URL:路由访问
|
|
||||||
TEMP_DIR:临时文件目录绝对路径
|
TEMP_DIR:临时文件目录绝对路径
|
||||||
官方文档:https://fastapi.tiangolo.com/tutorial/static-files/
|
官方文档:https://fastapi.tiangolo.com/tutorial/static-files/
|
||||||
"""
|
"""
|
||||||
TEMP_ENABLE = True
|
|
||||||
TEMP_URL = "/temp"
|
|
||||||
TEMP_DIR = os.path.join(BASE_DIR, "temp")
|
TEMP_DIR = os.path.join(BASE_DIR, "temp")
|
||||||
|
|
||||||
"""
|
"""
|
||||||
|
@ -37,7 +37,6 @@ from datetime import datetime
|
|||||||
|
|
||||||
|
|
||||||
class UserDal(DalBase):
|
class UserDal(DalBase):
|
||||||
|
|
||||||
import_headers = [
|
import_headers = [
|
||||||
{"label": "姓名", "field": "name", "required": True},
|
{"label": "姓名", "field": "name", "required": True},
|
||||||
{"label": "昵称", "field": "nickname", "required": False},
|
{"label": "昵称", "field": "nickname", "required": False},
|
||||||
@ -205,11 +204,15 @@ class UserDal(DalBase):
|
|||||||
:param params:
|
:param params:
|
||||||
:return:
|
:return:
|
||||||
"""
|
"""
|
||||||
datas = await self.get_datas(**params.dict(), v_return_objs=True)
|
datas = await self.get_datas(
|
||||||
|
**params.dict(),
|
||||||
|
v_return_objs=True,
|
||||||
|
v_options=[joinedload(self.model.depts), joinedload(self.model.roles)]
|
||||||
|
)
|
||||||
# 获取表头
|
# 获取表头
|
||||||
row = list(map(lambda i: i.get("label"), header))
|
row = list(map(lambda i: i.get("label"), header))
|
||||||
rows = []
|
rows = []
|
||||||
options = await vadmin_system_crud.DictTypeDal(self.db).get_dicts_details(["sys_vadmin_gender"])
|
options = await self.get_export_headers_options()
|
||||||
for user in datas:
|
for user in datas:
|
||||||
data = []
|
data = []
|
||||||
for item in header:
|
for item in header:
|
||||||
@ -221,16 +224,35 @@ class UserDal(DalBase):
|
|||||||
elif field == "is_staff":
|
elif field == "is_staff":
|
||||||
value = "是" if value else "否"
|
value = "是" if value else "否"
|
||||||
elif field == "gender":
|
elif field == "gender":
|
||||||
result = list(filter(lambda i: i["value"] == value, options["sys_vadmin_gender"]))
|
result = list(filter(lambda i: i["value"] == value, options["gender_options"]))
|
||||||
value = result[0]["label"] if result else ""
|
value = result[0]["label"] if result else ""
|
||||||
|
elif field == "roles":
|
||||||
|
value = ",".join([i.name for i in value])
|
||||||
|
elif field == "depts":
|
||||||
|
value = ",".join([i.name for i in value])
|
||||||
data.append(value)
|
data.append(value)
|
||||||
rows.append(data)
|
rows.append(data)
|
||||||
em = ExcelManage()
|
em = ExcelManage()
|
||||||
em.create_excel("用户列表")
|
em.create_excel("用户列表")
|
||||||
em.write_list(rows, row)
|
em.write_list(rows, row)
|
||||||
file_url = em.save_excel()
|
remote_file_url = em.save_excel().get("remote_path")
|
||||||
em.close()
|
em.close()
|
||||||
return {"url": file_url, "filename": "用户列表.xlsx"}
|
return {"url": remote_file_url, "filename": "用户列表.xlsx"}
|
||||||
|
|
||||||
|
async def get_export_headers_options(self, include: list[str] = None) -> dict[str, list]:
|
||||||
|
"""
|
||||||
|
获取导出所需选择项
|
||||||
|
:param include: 包括的选择项
|
||||||
|
:return:
|
||||||
|
"""
|
||||||
|
options = {}
|
||||||
|
# 性别选择项
|
||||||
|
if include is None or 'gender' in include:
|
||||||
|
gender_objs = await vadmin_system_crud.DictTypeDal(self.db).get_dicts_details(["sys_vadmin_gender"])
|
||||||
|
sys_vadmin_gender = gender_objs.get("sys_vadmin_gender", [])
|
||||||
|
gender_options = [{"label": i["label"], "value": i["value"]} for i in sys_vadmin_gender]
|
||||||
|
options["gender_options"] = gender_options
|
||||||
|
return options
|
||||||
|
|
||||||
async def get_import_headers_options(self) -> None:
|
async def get_import_headers_options(self) -> None:
|
||||||
"""
|
"""
|
||||||
@ -238,17 +260,17 @@ class UserDal(DalBase):
|
|||||||
:return:
|
:return:
|
||||||
"""
|
"""
|
||||||
# 角色选择项
|
# 角色选择项
|
||||||
roles = await RoleDal(self.db).get_datas(limit=0, v_return_objs=True, disabled=False, is_admin=False)
|
role_options = await RoleDal(self.db).get_datas(limit=0, v_return_objs=True, disabled=False, is_admin=False)
|
||||||
role_options = self.import_headers[4]
|
role_item = self.import_headers[4]
|
||||||
assert isinstance(role_options, dict)
|
assert isinstance(role_item, dict)
|
||||||
role_options["options"] = [{"label": role.name, "value": role.id} for role in roles]
|
role_item["options"] = [{"label": role.name, "value": role.id} for role in role_options]
|
||||||
|
|
||||||
# 性别选择项
|
# 性别选择项
|
||||||
dict_types = await vadmin_system_crud.DictTypeDal(self.db).get_dicts_details(["sys_vadmin_gender"])
|
gender_options = await vadmin_system_crud.DictTypeDal(self.db).get_dicts_details(["sys_vadmin_gender"])
|
||||||
gender_options = self.import_headers[3]
|
gender_item = self.import_headers[3]
|
||||||
assert isinstance(gender_options, dict)
|
assert isinstance(gender_item, dict)
|
||||||
sys_vadmin_gender = dict_types.get("sys_vadmin_gender")
|
sys_vadmin_gender = gender_options.get("sys_vadmin_gender")
|
||||||
gender_options["options"] = [{"label": item["label"], "value": item["value"]} for item in sys_vadmin_gender]
|
gender_item["options"] = [{"label": item["label"], "value": item["value"]} for item in sys_vadmin_gender]
|
||||||
|
|
||||||
async def download_import_template(self) -> dict:
|
async def download_import_template(self) -> dict:
|
||||||
"""
|
"""
|
||||||
|
@ -60,8 +60,6 @@ def create_app():
|
|||||||
# 挂在静态目录
|
# 挂在静态目录
|
||||||
if settings.STATIC_ENABLE:
|
if settings.STATIC_ENABLE:
|
||||||
app.mount(settings.STATIC_URL, app=StaticFiles(directory=settings.STATIC_ROOT))
|
app.mount(settings.STATIC_URL, app=StaticFiles(directory=settings.STATIC_ROOT))
|
||||||
if settings.TEMP_ENABLE:
|
|
||||||
app.mount(settings.TEMP_URL, app=StaticFiles(directory=settings.TEMP_DIR))
|
|
||||||
# 引入应用中的路由
|
# 引入应用中的路由
|
||||||
for url in urls.urlpatterns:
|
for url in urls.urlpatterns:
|
||||||
app.include_router(url["ApiRouter"], prefix=url["prefix"], tags=url["tags"])
|
app.include_router(url["ApiRouter"], prefix=url["prefix"], tags=url["tags"])
|
||||||
|
@ -9,12 +9,13 @@
|
|||||||
import datetime
|
import datetime
|
||||||
import os
|
import os
|
||||||
import re
|
import re
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
from openpyxl.utils import get_column_letter
|
from openpyxl.utils import get_column_letter
|
||||||
from openpyxl import load_workbook, Workbook
|
from openpyxl import load_workbook, Workbook
|
||||||
from application.settings import TEMP_DIR, TEMP_URL
|
from application.settings import STATIC_ROOT, STATIC_URL
|
||||||
import hashlib
|
|
||||||
import random
|
|
||||||
from openpyxl.styles import Alignment, Font, PatternFill, Border, Side
|
from openpyxl.styles import Alignment, Font, PatternFill, Border, Side
|
||||||
|
from utils.file.file_base import FileBase
|
||||||
from .excel_schema import AlignmentModel, FontModel, PatternFillModel
|
from .excel_schema import AlignmentModel, FontModel, PatternFillModel
|
||||||
|
|
||||||
|
|
||||||
@ -150,26 +151,19 @@ class ExcelManage:
|
|||||||
self.__auto_width()
|
self.__auto_width()
|
||||||
self.__set_row_border()
|
self.__set_row_border()
|
||||||
|
|
||||||
def save_excel(self, filename: str = None):
|
def save_excel(self, path: str = "excel_manage"):
|
||||||
"""
|
"""
|
||||||
保存 excel 文件到本地
|
保存 excel 文件到本地 static 目录
|
||||||
默认保存到临时目录中
|
:param path: static 指定目录类别
|
||||||
:param filename: 保存的文件名称
|
|
||||||
:return:
|
:return:
|
||||||
"""
|
"""
|
||||||
date = datetime.datetime.strftime(datetime.datetime.now(), "%Y%m%d")
|
file_path = FileBase.generate_static_file_path(path=path, suffix="xlsx")
|
||||||
file_dir = os.path.join(TEMP_DIR, date)
|
Path(file_path).parent.mkdir(parents=True, exist_ok=True)
|
||||||
if not os.path.exists(file_dir):
|
self.wb.save(file_path)
|
||||||
os.mkdir(file_dir)
|
return {
|
||||||
if not filename:
|
"local_path": file_path,
|
||||||
name = hashlib.md5(str(random.random()).encode()).hexdigest() + ".xlsx"
|
"remote_path": file_path.replace(STATIC_ROOT, STATIC_URL)
|
||||||
filename = os.path.join(file_dir, name)
|
}
|
||||||
else:
|
|
||||||
name = filename
|
|
||||||
filename = os.path.join(file_dir, filename)
|
|
||||||
self.wb.save(filename)
|
|
||||||
# 返回访问路由
|
|
||||||
return f"{TEMP_URL}/{date}/{name}"
|
|
||||||
|
|
||||||
def __set_row_style(
|
def __set_row_style(
|
||||||
self,
|
self,
|
||||||
@ -249,7 +243,3 @@ class ExcelManage:
|
|||||||
:return:
|
:return:
|
||||||
"""
|
"""
|
||||||
self.wb.close()
|
self.wb.close()
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
|
||||||
print([chr(a) for a in range(ord('A'), ord('Z') + 1)])
|
|
||||||
|
@ -50,7 +50,7 @@ class FileBase:
|
|||||||
1. filename 参数或者 suffix 参数必须填写一个
|
1. filename 参数或者 suffix 参数必须填写一个
|
||||||
2. filename 参数和 suffix 参数都存在则优先取 suffix 参数为后缀
|
2. filename 参数和 suffix 参数都存在则优先取 suffix 参数为后缀
|
||||||
:param path: static 指定目录类别
|
:param path: static 指定目录类别
|
||||||
:param filename: 文件名称
|
:param filename: 文件名称,只用户获取后缀,不做真实文件名称,避免文件重复问题
|
||||||
:param suffix: 文件后缀
|
:param suffix: 文件后缀
|
||||||
:return:
|
:return:
|
||||||
"""
|
"""
|
||||||
@ -63,7 +63,8 @@ class FileBase:
|
|||||||
path = path[1:]
|
path = path[1:]
|
||||||
if path[-1] == "/":
|
if path[-1] == "/":
|
||||||
path = path[:-1]
|
path = path[:-1]
|
||||||
return f"{STATIC_ROOT}/{path}/{cls.get_today_timestamp()}/{cls.get_random_filename(suffix)}"
|
today = datetime.datetime.strftime(datetime.datetime.now(), "%Y%m%d")
|
||||||
|
return f"{STATIC_ROOT}/{path}/{today}/{cls.get_random_filename(suffix)}"
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def generate_temp_file_path(cls, filename: str = None, suffix: str = None) -> str:
|
def generate_temp_file_path(cls, filename: str = None, suffix: str = None) -> str:
|
||||||
|
@ -137,9 +137,3 @@ class FileManage(FileBase):
|
|||||||
if not os.path.exists(dst):
|
if not os.path.exists(dst):
|
||||||
raise CustomException("目标目录不存在!")
|
raise CustomException("目标目录不存在!")
|
||||||
await aioshutil.copytree(src, dst, dirs_exist_ok=dirs_exist_ok)
|
await aioshutil.copytree(src, dst, dirs_exist_ok=dirs_exist_ok)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
|
||||||
_src = r"E:\ktianc\linfeng\project\leadership\temp\20231212\3nCx1702349951\template"
|
|
||||||
_dst = r"E:\ktianc\linfeng\project\leadership\static\template"
|
|
||||||
asyncio.run(FileManage.async_copy_dir(_src, _dst))
|
|
||||||
|
Loading…
x
Reference in New Issue
Block a user