ktianc
Gitee
commit
3054471123
@ -11,7 +11,7 @@ from fastapi.security import OAuth2PasswordBearer
|
||||
"""
|
||||
系统版本
|
||||
"""
|
||||
VERSION = "3.6.5"
|
||||
VERSION = "3.7.0"
|
||||
|
||||
"""安全警告: 不要在生产中打开调试运行!"""
|
||||
DEBUG = False
|
||||
@ -67,13 +67,9 @@ ACCESS_TOKEN_CACHE_MINUTES = 30
|
||||
|
||||
"""
|
||||
挂载临时文件目录，并添加路由访问，此路由不会在接口文档中显示
|
||||
TEMP_ENABLE：是否启用临时文件目录访问
|
||||
TEMP_URL：路由访问
|
||||
TEMP_DIR：临时文件目录绝对路径
|
||||
官方文档：https://fastapi.tiangolo.com/tutorial/static-files/
|
||||
"""
|
||||
TEMP_ENABLE = True
|
||||
TEMP_URL = "/temp"
|
||||
TEMP_DIR = os.path.join(BASE_DIR, "temp")
|
||||
|
||||
"""
|
||||
|
||||
@ -37,7 +37,6 @@ from datetime import datetime
|
||||
|
||||
|
||||
class UserDal(DalBase):
|
||||
|
||||
import_headers = [
|
||||
{"label": "姓名", "field": "name", "required": True},
|
||||
{"label": "昵称", "field": "nickname", "required": False},
|
||||
@ -205,11 +204,15 @@ class UserDal(DalBase):
|
||||
:param params:
|
||||
: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))
|
||||
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:
|
||||
data = []
|
||||
for item in header:
|
||||
@ -221,16 +224,35 @@ class UserDal(DalBase):
|
||||
elif field == "is_staff":
|
||||
value = "是" if value else "否"
|
||||
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 ""
|
||||
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)
|
||||
rows.append(data)
|
||||
em = ExcelManage()
|
||||
em.create_excel("用户列表")
|
||||
em.write_list(rows, row)
|
||||
file_url = em.save_excel()
|
||||
remote_file_url = em.save_excel().get("remote_path")
|
||||
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:
|
||||
"""
|
||||
@ -238,17 +260,17 @@ class UserDal(DalBase):
|
||||
:return:
|
||||
"""
|
||||
# 角色选择项
|
||||
roles = await RoleDal(self.db).get_datas(limit=0, v_return_objs=True, disabled=False, is_admin=False)
|
||||
role_options = self.import_headers[4]
|
||||
assert isinstance(role_options, dict)
|
||||
role_options["options"] = [{"label": role.name, "value": role.id} for role in roles]
|
||||
role_options = await RoleDal(self.db).get_datas(limit=0, v_return_objs=True, disabled=False, is_admin=False)
|
||||
role_item = self.import_headers[4]
|
||||
assert isinstance(role_item, dict)
|
||||
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 = self.import_headers[3]
|
||||
assert isinstance(gender_options, dict)
|
||||
sys_vadmin_gender = dict_types.get("sys_vadmin_gender")
|
||||
gender_options["options"] = [{"label": item["label"], "value": item["value"]} for item in sys_vadmin_gender]
|
||||
gender_options = await vadmin_system_crud.DictTypeDal(self.db).get_dicts_details(["sys_vadmin_gender"])
|
||||
gender_item = self.import_headers[3]
|
||||
assert isinstance(gender_item, dict)
|
||||
sys_vadmin_gender = gender_options.get("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:
|
||||
"""
|
||||
|
||||
@ -60,8 +60,6 @@ def create_app():
|
||||
# 挂在静态目录
|
||||
if settings.STATIC_ENABLE:
|
||||
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:
|
||||
app.include_router(url["ApiRouter"], prefix=url["prefix"], tags=url["tags"])
|
||||
|
||||
@ -9,12 +9,13 @@
|
||||
import datetime
|
||||
import os
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from openpyxl.utils import get_column_letter
|
||||
from openpyxl import load_workbook, Workbook
|
||||
from application.settings import TEMP_DIR, TEMP_URL
|
||||
import hashlib
|
||||
import random
|
||||
from application.settings import STATIC_ROOT, STATIC_URL
|
||||
from openpyxl.styles import Alignment, Font, PatternFill, Border, Side
|
||||
from utils.file.file_base import FileBase
|
||||
from .excel_schema import AlignmentModel, FontModel, PatternFillModel
|
||||
|
||||
|
||||
@ -150,26 +151,19 @@ class ExcelManage:
|
||||
self.__auto_width()
|
||||
self.__set_row_border()
|
||||
|
||||
def save_excel(self, filename: str = None):
|
||||
def save_excel(self, path: str = "excel_manage"):
|
||||
"""
|
||||
保存 excel 文件到本地
|
||||
默认保存到临时目录中
|
||||
:param filename: 保存的文件名称
|
||||
保存 excel 文件到本地 static 目录
|
||||
:param path: static 指定目录类别
|
||||
:return:
|
||||
"""
|
||||
date = datetime.datetime.strftime(datetime.datetime.now(), "%Y%m%d")
|
||||
file_dir = os.path.join(TEMP_DIR, date)
|
||||
if not os.path.exists(file_dir):
|
||||
os.mkdir(file_dir)
|
||||
if not filename:
|
||||
name = hashlib.md5(str(random.random()).encode()).hexdigest() + ".xlsx"
|
||||
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}"
|
||||
file_path = FileBase.generate_static_file_path(path=path, suffix="xlsx")
|
||||
Path(file_path).parent.mkdir(parents=True, exist_ok=True)
|
||||
self.wb.save(file_path)
|
||||
return {
|
||||
"local_path": file_path,
|
||||
"remote_path": file_path.replace(STATIC_ROOT, STATIC_URL)
|
||||
}
|
||||
|
||||
def __set_row_style(
|
||||
self,
|
||||
@ -249,7 +243,3 @@ class ExcelManage:
|
||||
:return:
|
||||
"""
|
||||
self.wb.close()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
print([chr(a) for a in range(ord('A'), ord('Z') + 1)])
|
||||
|
||||
@ -55,8 +55,8 @@ class FontModel(BaseModel):
|
||||
|
||||
|
||||
class PatternFillModel(BaseModel):
|
||||
start_color: str = Field(None, description="起始颜色（RGB 值或颜色名称）")
|
||||
end_color: str = Field(None, description="结束颜色（RGB 值或颜色名称）")
|
||||
start_color: str = Field("FFFFFF", description="起始颜色（RGB 值或颜色名称）")
|
||||
end_color: str = Field("FFFFFF", description="结束颜色（RGB 值或颜色名称）")
|
||||
fill_type: str = Field("solid", description="填充类型（'none'、'solid'、'darkDown' 等）")
|
||||
|
||||
class Config:
|
||||
|
||||
@ -50,7 +50,7 @@ class FileBase:
|
||||
1. filename 参数或者 suffix 参数必须填写一个
|
||||
2. filename 参数和 suffix 参数都存在则优先取 suffix 参数为后缀
|
||||
:param path: static 指定目录类别
|
||||
:param filename: 文件名称
|
||||
:param filename: 文件名称，只用户获取后缀，不做真实文件名称，避免文件重复问题
|
||||
:param suffix: 文件后缀
|
||||
:return:
|
||||
"""
|
||||
@ -63,7 +63,8 @@ class FileBase:
|
||||
path = path[1:]
|
||||
if 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
|
||||
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):
|
||||
raise CustomException("目标目录不存在！")
|
||||
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