from starlette.formparsers import MultiPartParser as StartletteMultiPartParser
from fastapi import APIRouter, Request, Response, Depends
from fastapi.responses import FileResponse
from key import v1 as key_v1
from hub import methods, Global
import importlib
router = APIRouter()
config_dict = {
'v6': {
# --- 用户管理 ---
1001: 'v6.code1000.code1001', # 新增用户
1002: 'v6.code1000.code1002', # 查询用户列表（分页）
1003: 'v6.code1000.code1003', # 删除指定用户
1004: 'v6.code1000.code1004', # 禁用指定用户
1005: 'v6.code1000.code1005', # 启用指定用户
1006: 'v6.code1000.code1006', # 获取指定用户详情
1007: 'v6.code1000.code1007', # 修改指定用户信息
# --- 车辆管理 ---
2001: 'v6.code2000.code2001', # 新增车辆
2002: 'v6.code2000.code2002', # 查询车辆列表（分页）
2003: 'v6.code2000.code2003', # 修改指定车辆信息
2004: 'v6.code2000.code2004', # 禁止指定车辆远程操作
2005: 'v6.code2000.code2005', # 允许指定车辆远程操作
2006: 'v6.code2000.code2006', # 删除指定作业车辆
2007: 'v6.code2000.code2007', # 获取指定作业车辆详情
# --- 关于日志 ---
3001: 'v6.code3000.code3001', # 查询驾驶人员操作记录列表
3002: 'v6.code3000.code3002', # 下载指定驾驶人员操作日志
}
}
methods_dict = {} # {<tag>: {<method>: <path>}}
def _get_method_by_code(code, tag):
"""
通过code获取method（不调用不加载）
"""
# --- check ---
if tag not in methods_dict:
methods_dict[tag] = {}
# --- check ---
if code in methods_dict.get(tag):
return methods_dict.get(tag).get(code)
# --- get method ---
method_path = config_dict.get(tag).get(code)
tag, file_name, method_name = method_path.split('.')
script = importlib.import_module(f"api.{tag}.{file_name}")
method = getattr(script, method_name)
# --- set method ---
methods_dict[tag][code] = method
return method
@router.post('/{tag}/api')
async def post(tag: str, request: Request, response: Response, user: dict = Depends(key_v1.login_required)):
""""""
# --- check ---
if not user.get('skip_is'):
response.headers['authorization'] = key_v1.get_token_by_user(user)
# --- get sources ---
sources = await request.json()
code = int(sources.get('code'))
page = sources.get('page', 1)
size = sources.get('size', 10)
ban_keys = ['code']
# --- set global args ---
sources = {key: val for key, val in sources.items() if key not in ban_keys}
sources['g_user_id'] = user.get('uid')
sources['g_user_name'] = user.get('username')
sources['g_user_pass'] = user.get('password')
sources['g_role_id'] = user.get('role_id')
# --- call action ---
try:
run_at = methods.now_ts()
method = _get_method_by_code(code=code, tag=tag)
result = await method(**sources)
methods.debug_log(f"api.post.95", f"#{tag}/{code}: use time {round(methods.now_ts() - run_at, 2)}s")
# --- check ---
if code in [
# 8201, # 访客记录，列表
] and result.__class__.__name__ == 'dict':
return dict(
code=result.get('code'),
data=result.get('data', [])[(page - 1) * size: page * size],
page=page,
size=size,
total=len(result.get('data', [])),
)
elif result.__class__.__name__ in ['FileResponse', 'dict']:
return result
else:
return dict(data=result, type=result.__class__.__name__)
except Exception as exception:
methods.debug_log('api.post.115', f"#tag: {tag}, #code: {code}")
methods.debug_log('api.post.115', f"#exception: {exception}")
methods.debug_log('api.post.115', f"#traceback: {methods.trace_log()}")
return dict(code=-1, data=[], message=f"something is wrong. [{exception.__class__.__name__}]",
details=f"{methods.trace_log()}")
@router.put('/actions')
async def upload(request: Request, user: dict = Depends(key_v1.login_required)):
try:
# --- get sources ---
sources = dict()
parser = StartletteMultiPartParser(request.headers, request.stream())
params = await parser.parse()
# sources['params'] = params
# --- set sources --- fromdata列表处理
for key, value in params.multi_items():
# methods.debug_log('api.upload.137', f"#key: {key}, #value: {value}")
# --- check list ---
if key[-4:] == 'list' and key not in sources:
sources[key] = list()
if key[-4:] == 'list':
sources[key].append(value)
# --- make list ---
else:
if key in sources:
sources[key] = [sources[key]]
sources[key].append(value)
else:
sources[key] = value
# methods.debug_log('actions.put', f"m-201: sources: {sources}")
# --- clean ---
tag = sources.get('tag', 'v1')
code = int(sources.get('code'))
del sources['tag']
del sources['code']
# --- set sources ---
sources['g_user_id'] = user.get('uid')
sources['g_user_name'] = user.get('username')
sources['g_user_pass'] = user.get('password')
sources['g_role_id'] = user.get('role_id')
# --- set files ---
files = dict()
for key, value in sources.items():
if value.__class__.__name__ == 'UploadFile':
files[key] = value
for key, value in files.items():
del sources[key]
sources['files'] = files
# --- call action ---
return await _get_method_by_code(code=code, tag=tag)(**sources)
except Exception as exception:
methods.debug_log('api.upload.176', f"#exception: {exception}")
methods.debug_log('api.upload.176', f"#traceback: {methods.trace_log()}")
return dict(code=-1, message=f"something is wrong. [{exception.__class__.__name__}]",
details=f"{methods.trace_log()}")
@router.get('/{tag}/api')
async def download(request: Request, response: Response, tag: str):
try:
# --- get ---
params = request.query_params
# methods.debug_log('api.download.236', f"#params: {params}")
method = _get_method_by_code(code=params.get('code'), tag=tag)
result = await method(**params)
# --- 是否弹框 ---
if True:
return FileResponse(result.get('file_path'),
media_type="application/octet-stream", filename=result.get('file_name'))
else:
return FileResponse(result.get('file_path'))
except Exception as exception:
methods.debug_log('api.download.249', f"#exception: {exception}")
methods.debug_log('api.download.249', f"#traceback: {methods.trace_log()}")
return dict(code=-1, message=f"something is wrong. [{exception.__class__.__name__}]",
details=f"{methods.trace_log()}")