refactor: 改进日志记录和错误处理
- 将print语句替换为logging.info以统一日志输出 - 优化错误日志格式，添加方括号提高可读性 - 在split_level函数中添加异常处理，返回默认值(0,0) - 改进空职级的错误处理逻辑 - 修复relativedelta计算工龄时的错误参数 - 为历史记录添加入职时间校验 - 统一使用logging.warning替代部分logging.error
This commit is contained in:
parent
c34d04c3d8
commit
013fa0c3fd
73
main.py
73
main.py
@ -33,6 +33,8 @@ for index, row in Promote.iterrows():
|
|||||||
for col in ["任职时间","工资执行时间"]:
|
for col in ["任职时间","工资执行时间"]:
|
||||||
Promote.at[index, col] = custom_date_parser(row[col])
|
Promote.at[index, col] = custom_date_parser(row[col])
|
||||||
|
|
||||||
|
logging.info("人员信息加载完成")
|
||||||
|
|
||||||
Rule_Role = []
|
Rule_Role = []
|
||||||
col = 2
|
col = 2
|
||||||
while True: # 职位规则
|
while True: # 职位规则
|
||||||
@ -73,7 +75,7 @@ while True: # 名称变化
|
|||||||
except:
|
except:
|
||||||
break
|
break
|
||||||
|
|
||||||
print("读取完成")
|
logging.info("规则加载完成")
|
||||||
|
|
||||||
Rule_Role = sorted(Rule_Role, key=lambda x: x['start'])
|
Rule_Role = sorted(Rule_Role, key=lambda x: x['start'])
|
||||||
Rule_Level = sorted(Rule_Level, key=lambda x: x['start'])
|
Rule_Level = sorted(Rule_Level, key=lambda x: x['start'])
|
||||||
@ -82,8 +84,12 @@ Rule_RoleName = sorted(Rule_RoleName, key=lambda x: x['start'])
|
|||||||
nowtime = datetime.now()
|
nowtime = datetime.now()
|
||||||
|
|
||||||
def split_level(level:str):
|
def split_level(level:str):
|
||||||
parts = level.split('-')
|
try:
|
||||||
return (int(parts[0]), int(parts[1]))
|
parts = level.split('-')
|
||||||
|
return (int(parts[0]), int(parts[1]))
|
||||||
|
except:
|
||||||
|
logging.warning(f"职级[{level}]格式错误")
|
||||||
|
return (0, 0)
|
||||||
|
|
||||||
def role_salary(role:str, time):
|
def role_salary(role:str, time):
|
||||||
for rule in Rule_Role:
|
for rule in Rule_Role:
|
||||||
@ -92,7 +98,10 @@ def role_salary(role:str, time):
|
|||||||
tmp = rule["rule"][rule["rule"]["role"] == role].iloc[0]
|
tmp = rule["rule"][rule["rule"]["role"] == role].iloc[0]
|
||||||
return tmp["salary"]
|
return tmp["salary"]
|
||||||
except:
|
except:
|
||||||
logging.error(f"职位{role}在{time}时不存在工资规则")
|
if role == "":
|
||||||
|
logging.error("空职级")
|
||||||
|
else:
|
||||||
|
logging.warning(f"职位[{role}]在[{time}]时不存在工资规则")
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
def level_salary(level:str, time):
|
def level_salary(level:str, time):
|
||||||
@ -102,7 +111,7 @@ def level_salary(level:str, time):
|
|||||||
tmp = rule["rule"][rule["rule"]["level"] == level].iloc[0]
|
tmp = rule["rule"][rule["rule"]["level"] == level].iloc[0]
|
||||||
return tmp["salary"]
|
return tmp["salary"]
|
||||||
except:
|
except:
|
||||||
logging.error(f"职级{level}在{time}时不存在工资规则")
|
logging.warning(f"职级[{level}]在[{time}]时不存在工资规则")
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
def role_limit(role:str):
|
def role_limit(role:str):
|
||||||
@ -110,7 +119,7 @@ def role_limit(role:str):
|
|||||||
tmp = Level_Limit[Level_Limit["role"] == role].iloc[0]
|
tmp = Level_Limit[Level_Limit["role"] == role].iloc[0]
|
||||||
return tmp["limit"]
|
return tmp["limit"]
|
||||||
except:
|
except:
|
||||||
logging.error(f"职位{role}不存在职级上限规则")
|
logging.warning(f"职位[{role}]不存在职级上限规则")
|
||||||
return -1
|
return -1
|
||||||
|
|
||||||
max_promote = 0
|
max_promote = 0
|
||||||
@ -142,13 +151,13 @@ def fill_basic_info(ws, row):# 填充基本信息
|
|||||||
def fill_prompt_info(ws, promote):# 填充晋升信息
|
def fill_prompt_info(ws, promote):# 填充晋升信息
|
||||||
for index, prow in promote.iterrows():
|
for index, prow in promote.iterrows():
|
||||||
if index > P_LIMIT-1:
|
if index > P_LIMIT-1:
|
||||||
logging.error(f"超过{P_LIMIT}条晋升信息，共{promote.shape[0]}条。")
|
logging.warning(f"超过[{P_LIMIT}]条晋升信息，共[{promote.shape[0]}]条。")
|
||||||
max_promote = max(max_promote, promote.shape[0])
|
max_promote = max(max_promote, promote.shape[0])
|
||||||
break
|
break
|
||||||
try:
|
try:
|
||||||
ws.cell(row=P_START+index, column=1, value=prow["任职时间"].strftime("%Y.%m"))
|
ws.cell(row=P_START+index, column=1, value=prow["任职时间"].strftime("%Y.%m"))
|
||||||
except:
|
except:
|
||||||
logging.error(f"晋升时间格式错误：{prow['任职时间']}")
|
logging.warning(f"晋升时间格式错误：{prow['任职时间']}")
|
||||||
ws.cell(row=P_START+index, column=2, value=prow["变动批注"])
|
ws.cell(row=P_START+index, column=2, value=prow["变动批注"])
|
||||||
ws.cell(row=P_START+index, column=3, value="任"+prow["新职务"])
|
ws.cell(row=P_START+index, column=3, value="任"+prow["新职务"])
|
||||||
|
|
||||||
@ -159,7 +168,7 @@ def fill_history_info(ws, History_pd):# 填充历史记录
|
|||||||
try:
|
try:
|
||||||
ws.cell(row=H_START+index, column=1, value=hrow["时间"].strftime("%Y.%m"))
|
ws.cell(row=H_START+index, column=1, value=hrow["时间"].strftime("%Y.%m"))
|
||||||
except:
|
except:
|
||||||
logging.error(f"历史时间格式错误：{hrow['时间']}")
|
logging.warning(f"历史时间格式错误：{hrow['时间']}")
|
||||||
ws.cell(row=H_START+index, column=2, value=hrow["职务"])
|
ws.cell(row=H_START+index, column=2, value=hrow["职务"])
|
||||||
ws.cell(row=H_START+index, column=3, value=hrow["工资额1"])
|
ws.cell(row=H_START+index, column=3, value=hrow["工资额1"])
|
||||||
ws.cell(row=H_START+index, column=4, value=hrow["级别档次"])
|
ws.cell(row=H_START+index, column=4, value=hrow["级别档次"])
|
||||||
@ -173,13 +182,13 @@ BaseData["Latest_Prom"] = None
|
|||||||
|
|
||||||
for index, row in BaseData.iterrows():
|
for index, row in BaseData.iterrows():
|
||||||
try:
|
try:
|
||||||
logging.info(f"台账：第{index+1}共{BaseData.shape[0]}现在是{row['身份证号码']}")
|
logging.info(f"台账：第[{index+1}]共[{BaseData.shape[0]}]现在是[{row['身份证号码']}]")
|
||||||
print(f"台账：第{index+1}共{BaseData.shape[0]}现在是{row['身份证号码']}")
|
|
||||||
BaseData.at[index, "Latest_Role"] = row["初始职务"]
|
BaseData.at[index, "Latest_Role"] = row["初始职务"]
|
||||||
BaseData.at[index, "Latest_Prom"] = row["入职时间"].strftime("%Y.%m")
|
BaseData.at[index, "Latest_Prom"] = row["入职时间"].strftime("%Y.%m")
|
||||||
wb = load_workbook("个人台账.xlsx")
|
wb = load_workbook("个人台账.xlsx")
|
||||||
ws = wb.active
|
ws = wb.active
|
||||||
fill_basic_info(ws, row)# 填充基本信息
|
fill_basic_info(ws, row)# 填充基本信息
|
||||||
|
|
||||||
# 查找晋升信息
|
# 查找晋升信息
|
||||||
promote = Promote[Promote["身份证号"] == row["身份证号码"]]
|
promote = Promote[Promote["身份证号"] == row["身份证号码"]]
|
||||||
if not promote.empty:
|
if not promote.empty:
|
||||||
@ -199,23 +208,28 @@ for index, row in BaseData.iterrows():
|
|||||||
History_pd.loc[len(History_pd)] = [row["入职时间"], row["初始职务"], "", row["入职时的初始级别"], "", "", "套改/定级"]
|
History_pd.loc[len(History_pd)] = [row["入职时间"], row["初始职务"], "", row["入职时的初始级别"], "", "", "套改/定级"]
|
||||||
for index, prow in promote.iterrows(): # 添加晋升记录
|
for index, prow in promote.iterrows(): # 添加晋升记录
|
||||||
History_pd.loc[len(History_pd)] = [prow["工资执行时间"], prow["新职务"], "", "", "", "", "晋升"]
|
History_pd.loc[len(History_pd)] = [prow["工资执行时间"], prow["新职务"], "", "", "", "", "晋升"]
|
||||||
calctime=row["二档起始"] + relativedelta(minute=1)
|
try:
|
||||||
while True: # 添加二级记录
|
calctime=row["二档起始"] + relativedelta(minute=1)
|
||||||
calctime += relativedelta(years=2)
|
while True: # 添加二级记录
|
||||||
if calctime > nowtime:
|
calctime += relativedelta(years=2)
|
||||||
break
|
if calctime > nowtime:
|
||||||
History_pd.loc[len(History_pd)] = [calctime, "", "", "", "", "", "两年晋档"]
|
break
|
||||||
calctime=row["五档起始"]
|
History_pd.loc[len(History_pd)] = [calctime, "", "", "", "", "", "两年晋档"]
|
||||||
while True: # 添加五级记录
|
calctime=row["五档起始"]
|
||||||
calctime += relativedelta(years=5)
|
while True: # 添加五级记录
|
||||||
if calctime > nowtime:
|
calctime += relativedelta(years=5)
|
||||||
break
|
if calctime > nowtime:
|
||||||
History_pd.loc[len(History_pd)] = [calctime, "", "", "", "", "", "五年晋级"]
|
break
|
||||||
|
History_pd.loc[len(History_pd)] = [calctime, "", "", "", "", "", "五年晋级"]
|
||||||
|
except:
|
||||||
|
raise Exception(f"二、五档起始时间格式错误：{row['二档起始']}或{row['五档起始']}")
|
||||||
for rule in Rule_Level: # 工资调标
|
for rule in Rule_Level: # 工资调标
|
||||||
if row["入职时间"] < rule["start"]:
|
if row["入职时间"] < rule["start"]:
|
||||||
History_pd.loc[len(History_pd)] = [rule["start"], "", "", "", "", "", "工资调标"]
|
History_pd.loc[len(History_pd)] = [rule["start"], "", "", "", "", "", "工资调标"]
|
||||||
History_pd = History_pd.sort_values(by="时间").reset_index(drop=True)
|
History_pd = History_pd.sort_values(by="时间").reset_index(drop=True)
|
||||||
|
|
||||||
|
if History_pd.at[0,"时间"] != row["入职时间"]:
|
||||||
|
raise Exception(f"入职时间晚于其他计算的时间:{row['入职时间']} < {History_pd.at[0,'时间']} ({History_pd.at[0,'变动原因']})")
|
||||||
for index, hrow in History_pd.iterrows(): # 数据计算
|
for index, hrow in History_pd.iterrows(): # 数据计算
|
||||||
# 调整职务职级
|
# 调整职务职级
|
||||||
if index > 0 and hrow["职务"] == "":
|
if index > 0 and hrow["职务"] == "":
|
||||||
@ -247,14 +261,13 @@ for index, row in BaseData.iterrows():
|
|||||||
fill_history_info(ws, History_pd)# 填充历史记录
|
fill_history_info(ws, History_pd)# 填充历史记录
|
||||||
wb.save(f"./output/{row['姓名']}_{row['身份证号码']}.xlsx")
|
wb.save(f"./output/{row['姓名']}_{row['身份证号码']}.xlsx")
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logging.error(e)
|
logging.error(f"{row['身份证号码']}:{e}")
|
||||||
|
|
||||||
wb = load_workbook("./汇总.xlsx")
|
wb = load_workbook("./汇总.xlsx")
|
||||||
ws = wb.active
|
ws = wb.active
|
||||||
for index, row in BaseData.iterrows(): # 汇总
|
for index, row in BaseData.iterrows(): # 汇总
|
||||||
try:
|
try:
|
||||||
logging.info(f"汇总：第{index+1}共{BaseData.shape[0]}现在是{row['身份证号码']}")
|
logging.info(f"汇总：第[{index+1}]共[{BaseData.shape[0]}]现在是[{row['身份证号码']}]")
|
||||||
print(f"汇总：第{index+1}共{BaseData.shape[0]}现在是{row['身份证号码']}")
|
|
||||||
for col in range(1,16):
|
for col in range(1,16):
|
||||||
ws.cell(row=3+index, column=col)._style = ws.cell(row=3, column=col)._style
|
ws.cell(row=3+index, column=col)._style = ws.cell(row=3, column=col)._style
|
||||||
ws.cell(row=6+index, column=1, value=index+1)
|
ws.cell(row=6+index, column=1, value=index+1)
|
||||||
@ -263,7 +276,7 @@ for index, row in BaseData.iterrows(): # 汇总
|
|||||||
ws.cell(row=6+index, column=4, value=row["出生年月"])
|
ws.cell(row=6+index, column=4, value=row["出生年月"])
|
||||||
ws.cell(row=6+index, column=5, value=row["参加工作时间"])
|
ws.cell(row=6+index, column=5, value=row["参加工作时间"])
|
||||||
ws.cell(row=6+index, column=6, value=row["学历"])
|
ws.cell(row=6+index, column=6, value=row["学历"])
|
||||||
ws.cell(row=6+index, column=7, value=relativedelta(nowtime.year(), row["入职时间"]).years+row["工龄调增"]-row["工龄调减"]+1)
|
ws.cell(row=6+index, column=7, value=relativedelta(nowtime, row["入职时间"]).years+row["工龄调增"]-row["工龄调减"]+1)
|
||||||
ws.cell(row=6+index, column=8, value=relativedelta(nowtime, row["入职时间"]).years)
|
ws.cell(row=6+index, column=8, value=relativedelta(nowtime, row["入职时间"]).years)
|
||||||
ws.cell(row=6+index, column=9, value=row["工龄调增"])
|
ws.cell(row=6+index, column=9, value=row["工龄调增"])
|
||||||
ws.cell(row=6+index, column=10, value=row["工龄调减"])
|
ws.cell(row=6+index, column=10, value=row["工龄调减"])
|
||||||
@ -272,10 +285,10 @@ for index, row in BaseData.iterrows(): # 汇总
|
|||||||
ws.cell(row=6+index, column=13, value=row["职务2"])
|
ws.cell(row=6+index, column=13, value=row["职务2"])
|
||||||
ws.cell(row=6+index, column=14, value=row["日期2"])
|
ws.cell(row=6+index, column=14, value=row["日期2"])
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logging.error(e)
|
logging.error(f"{row['身份证号码']}:{e}")
|
||||||
wb.save("./名册.xlsx") # 保存汇总
|
wb.save("./名册.xlsx") # 保存汇总
|
||||||
|
|
||||||
if max_promote > 0:
|
if max_promote > 0:
|
||||||
logging.error(f"最多有{max_promote}条晋升信息，需要调整模板。记得同时调整薪资历史的起始行和个人评价结果。")
|
logging.warning(f"最多有[{max_promote}]条晋升信息，需要调整模板。记得同时调整薪资历史的起始行和个人评价结果。")
|
||||||
if max_history > 0:
|
if max_history > 0:
|
||||||
logging.error(f"最多有{max_history}条薪资历史，需要调整模板。")
|
logging.warning(f"最多有[{max_history}]条薪资历史，需要调整模板。")
|
||||||
Loading…
x
Reference in New Issue
Block a user