leguadmin/prs_server
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

1.优化函数名

Browse Source
This commit is contained in:
李伟 2022-08-09 17:29:16 +08:00
parent b0e65dc77d
commit 876f0bdb1a
5 changed files with 174 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

146
api/api_v1/endpoints/interview.py
View File

@ -172,6 +172,8 @@ async def interview_file_insert(
) -> schemas.Msg:
""" interview面试数据导入 """
path_data = os.getcwd() + '/jianli/' # 当前文件所在的目录
if not os.path.exists(path_data):
os.makedirs(path_data)
contents = await file.read()
filename = file.filename
try:
@ -337,7 +339,7 @@ async def get_job(
job_list['principal'] = i['principal'] # 招聘负责人
job_list['job_sector'] = i['job_sector'] # 部门
job_list['job_num'] = i['job_num'] # 目标招聘人数
job_list['now_job_num'] = num[0]['nu'] # 对应职位的入职人数
job_list['now_job_num'] = num[0]['nu'] # 对应职位的入职人数
job_list['hou_num'] = hou_num[0]['nu'] # 候选人总数
job.append(job_list)
data = {
@ -405,7 +407,7 @@ async def find_remark(
@router.post("/find_criterion")
async def find_criterion(
request: Request,
data_in: schemas.get_uid,
data_in: schemas.get_uids,
ckdb: CKDrive = Depends(get_ck_db)
) -> schemas.Msg:
"""获取标准版简历"""
@ -603,6 +605,7 @@ async def name(
data = []
for i in res:
data_dict = {}
data_dict['_id'] = i.get('_id')
data_dict['name'] = i.get('name')
data_dict['user_id'] = i.get('user_id')
data_dict['email'] = i.get('email')
@ -654,10 +657,10 @@ async def delete(
@router.post("/get_dashboard")
async def get_dashboard(request: Request,
data_in: schemas.ReadDashboard,
db: AsyncIOMotorDatabase = Depends(get_database),
current_user: schemas.UserDB = Depends(deps.get_current_user)
):
data_in: schemas.ReadDashboard,
db: AsyncIOMotorDatabase = Depends(get_database),
current_user: schemas.UserDB = Depends(deps.get_current_user)
):
"""获取一个看板"""
res = await crud.dashboard.get(db, id=data_in.id)
reports = {item['report_id']: item for item in res['reports']}
@ -903,10 +906,10 @@ async def reset_password(request: Request,
@router.post("/reset_my_password")
async def reset_my_password(request: Request,
data_in: schemas.UserRestMyPassword,
db: AsyncIOMotorDatabase = Depends(get_database),
current_user: schemas.User = Depends(deps.get_current_user)
) -> Any:
data_in: schemas.UserRestMyPassword,
db: AsyncIOMotorDatabase = Depends(get_database),
current_user: schemas.User = Depends(deps.get_current_user)
) -> Any:
"""
修改自己的密码
"""
@ -991,6 +994,127 @@ async def get_section(
# current_user: schemas.UserDB = Depends(deps.get_current_user)
) -> schemas.Msg:
"""获取对应部门及职位"""
res=await crud.section.get_all(db)
res = await crud.section.get_all(db)
return schemas.Msg(code=200, msg='ok', data=res)
@router.post("/ins_section")
async def get_section(
request: Request,
data_in: schemas.Ins_section,
db: AsyncIOMotorDatabase = Depends(get_database),
# current_user: schemas.UserDB = Depends(deps.get_current_user)
) -> schemas.Msg:
"""新增对应部门及职位"""
await crud.section.ins_section(db, data_in)
return schemas.Msg(code=200, msg='ok', data='')
@router.get("/api_list")
async def api_list(
request: Request,
db: AsyncIOMotorDatabase = Depends(get_database),
# current_user: schemas.UserDB = Depends(deps.get_current_user)
) -> schemas.Msg:
"""
所有的api
"""
re = await crud.api_module.get_api_module(db)
res = []
for i in re:
if i['path_name'] != 'root':
i['_id'] = str(i['_id'])
res.append(i)
return schemas.Msg(code=0, msg='ok', data=res)
@router.post("/del_api_module")
async def add_policy(
request: Request,
data_in: schemas.Add_module,
db: AsyncIOMotorDatabase = Depends(get_database),
# current_user: schemas.UserDB = Depends(deps.get_current_user)
):
"""修改api_module权限状态"""
res = await crud.api_module.get_one_module(db, data_in)
for i in range(len(res['state'])):
if data_in.url == res['api_list'][i]:
res['state'][i] = False
await crud.api_module.update_one_module(db, res)
return schemas.Msg(code=0, msg='修改成功', data='')
@router.post("/add_api_module")
async def add_policy(
request: Request,
data_in: schemas.Add_module,
db: AsyncIOMotorDatabase = Depends(get_database),
# current_user: schemas.UserDB = Depends(deps.get_current_user)
):
"""修改api_module权限状态"""
res = await crud.api_module.get_one_module(db, data_in)
for i in range(len(res['state'])):
if data_in.url == res['api_list'][i]:
res['state'][i] = True
await crud.api_module.update_one_module(db, res)
return schemas.Msg(code=0, msg='修改成功', data='')
@router.get("/api_module")
async def domain_list(
request: Request,
db: AsyncIOMotorDatabase = Depends(get_database),
current_user: schemas.UserDB = Depends(deps.get_current_user)
) -> schemas.Msg:
"""
角色管理创建角色时显示的各个模块
"""
res = await crud.api_module.get_api_module(db)
api_module=[]
for i in res:
if i['path_name'] !='root':
data=[]
data.append(i['auth_id'])
data.append(i['path_name'])
api_module.append(data)
return schemas.Msg(code=0, msg='ok', data=api_module)
@router.post("/add_api")
async def add_api(
request: Request,
data_in: schemas.AddApi,
db: AsyncIOMotorDatabase = Depends(get_database),
current_user: schemas.UserDB = Depends(deps.get_current_user)
) -> schemas.Msg:
"""
添加api
"""
res = await crud.api_module.get_api_module(db)
for i in res:
if data_in.path in i['api_list']:
return schemas.Msg(code=0, msg='该路由已存在', data='')
path_list = []
for i in res:
path_list.append(i['path_name'])
if data_in.name in path_list:
for i in res:
if data_in.name == i['path_name']:
i['api_list'].append(data_in.path)
i['api_name'].append(data_in.desc)
i['state'].append(True)
await crud.api_module.updata_quanxian_module(db, schemas.Url_module(auth_id=i['auth_id'],
path_name=data_in.name,
api_list=i['api_list'],
api_name=i['api_name'],
state=i['state']))
return schemas.Msg(code=0, msg='ok', data='路由添加成功！')
else:
auth_list = []
for i in res:
auth_list.append(i['auth_id'])
auth_id = max(auth_list)
auth_id = 'abc' + str(int(auth_id.split('c')[-1]) + 1)
await crud.api_module.insert_quanxian(db, schemas.Url_module(auth_id=auth_id, path_name=data_in.name,
api_list=[data_in.path],
api_name=[data_in.desc], state=[True]))
return schemas.Msg(code=0, msg='ok', data='路由添加成功！')

2
crud/crud_jobs.py
View File

@ -23,7 +23,7 @@ class CRUDJobs(CRUDBase):
where = {}
data_dict = data_in.dict()
for k, v in data_dict.items():
if v != None:
if v != None and v != '' and v != []:
where[k] = v
return await self.find_many(db, where, {'_id': 0})

17
crud/crud_section.py
View File

@ -13,18 +13,17 @@ class Section(CRUDBase):
# async def get_url(self, db: AsyncIOMotorDatabase, data_in: schemas.Url_list):
# return await self.find_many(db, {'name': data_in.name})
#
# # 插入单条对应级别权限的路由和状态
# async def insert_url(self, db: AsyncIOMotorDatabase, data_in: schemas.Url_list):
# return await self.insert_one(db, data_in.dict())
#
# 插入单条部门及职位
async def ins_section(self, db: AsyncIOMotorDatabase, data_in: schemas.Ins_section):
return await self.insert_one(db, data_in.dict())
# async def insert_urls(self, db: AsyncIOMotorDatabase, data_in: schemas.Url_lists):
# return await self.insert_one(db, data_in.dict())
#
# # 更新单条对应级别权限的路由和状态
# async def update_url_url(self, db: AsyncIOMotorDatabase, res):
# return await self.update_one(db, {'_id': res['_id']}, {
# '$set': {'state': res['state']}})
#
# 更新单条部门及职位信息
async def up_section(self, db: AsyncIOMotorDatabase,data_in: schemas.Ins_section):
return await self.update_one(db, {'name': data_in.name}, {
'$set': {'position': data_in.position}})
# async def find_one_url(self, db: AsyncIOMotorDatabase, data_in: schemas.Datalist):
# return await self.find_one(db, {'auth_id': data_in.role_id, 'path_name': data_in.path_name})
#

17
liwei_接口文档.md
View File

@ -305,5 +305,20 @@ api:/api/v1/itr/owner_edit
职能类别function_type，职位级别job_rank，学历要求education，部门sector
#获取对应部门及职位
api:/api/v1/itr/owner_list
api:/api/v1/itr/get_section
请求方式：get
#获取对应部门及职位
api:/api/v1/itr/ins_section
请求方式：post
参数：
name: str # 部门名
position: List[position] # 职位名及职位id
position
job_id: str= str(int(time.time())) # 职位的唯一id # 职位id
job_name: str # 职位名
#所有的api
api:/api/v1/itr/api_list
请求方式：get

15
schemas/interview_plan.py
View File

@ -1,8 +1,11 @@
import time
from datetime import datetime
from typing import List, Union, Dict
from pydantic import BaseModel
from typing import Optional
from utils import get_uid
class Interview(BaseModel):
date: dict # 要查询的条件
@ -28,7 +31,7 @@ class Post_uid(BaseModel):
resume_affix_id: List[str] # 附件id
class get_uid(BaseModel):
class get_uids(BaseModel):
uid: str # 面试者简历在CK中的唯一标识
@ -105,3 +108,13 @@ class Createuser(BaseModel):
class Get_userid(BaseModel):
user_id: str # 钉钉的唯一标识
type: int # 0为解禁1为禁止
class position(BaseModel):
job_id: str= str(get_uid()) # 职位的唯一id # 职位id
job_name: str # 职位名
class Ins_section(BaseModel):
name: str # 部门名
position: List[position] # 职位名及职位id