From 013fa0c3fd37597c2843e0c051e8bd30e49b03b0 Mon Sep 17 00:00:00 2001 From: mxr612 Date: Tue, 27 May 2025 17:49:03 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E6=94=B9=E8=BF=9B=E6=97=A5?= =?UTF-8?q?=E5=BF=97=E8=AE=B0=E5=BD=95=E5=92=8C=E9=94=99=E8=AF=AF=E5=A4=84?= =?UTF-8?q?=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 将print语句替换为logging.info以统一日志输出 - 优化错误日志格式，添加方括号提高可读性 - 在split_level函数中添加异常处理，返回默认值(0,0) - 改进空职级的错误处理逻辑 - 修复relativedelta计算工龄时的错误参数 - 为历史记录添加入职时间校验 - 统一使用logging.warning替代部分logging.error --- main.py | 73 +++++++++++++++++++++++++++++++++------------------------ 1 file changed, 43 insertions(+), 30 deletions(-) diff --git a/main.py b/main.py index 44e4670..8e8ee86 100644 --- a/main.py +++ b/main.py @@ -33,6 +33,8 @@ for index, row in Promote.iterrows(): for col in ["任职时间","工资执行时间"]: Promote.at[index, col] = custom_date_parser(row[col]) +logging.info("人员信息加载完成") + Rule_Role = [] col = 2 while True: # 职位规则 @@ -73,7 +75,7 @@ while True: # 名称变化 except: break -print("读取完成") +logging.info("规则加载完成") Rule_Role = sorted(Rule_Role, 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() def split_level(level:str): - parts = level.split('-') - return (int(parts[0]), int(parts[1])) + try: + parts = level.split('-') + return (int(parts[0]), int(parts[1])) + except: + logging.warning(f"职级[{level}]格式错误") + return (0, 0) def role_salary(role:str, time): for rule in Rule_Role: @@ -92,7 +98,10 @@ def role_salary(role:str, time): tmp = rule["rule"][rule["rule"]["role"] == role].iloc[0] return tmp["salary"] except: - logging.error(f"职位{role}在{time}时不存在工资规则") + if role == "": + logging.error("空职级") + else: + logging.warning(f"职位[{role}]在[{time}]时不存在工资规则") return 0 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] return tmp["salary"] except: - logging.error(f"职级{level}在{time}时不存在工资规则") + logging.warning(f"职级[{level}]在[{time}]时不存在工资规则") return 0 def role_limit(role:str): @@ -110,7 +119,7 @@ def role_limit(role:str): tmp = Level_Limit[Level_Limit["role"] == role].iloc[0] return tmp["limit"] except: - logging.error(f"职位{role}不存在职级上限规则") + logging.warning(f"职位[{role}]不存在职级上限规则") return -1 max_promote = 0 @@ -142,13 +151,13 @@ def fill_basic_info(ws, row):# 填充基本信息 def fill_prompt_info(ws, promote):# 填充晋升信息 for index, prow in promote.iterrows(): 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]) break try: ws.cell(row=P_START+index, column=1, value=prow["任职时间"].strftime("%Y.%m")) 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=3, value="任"+prow["新职务"]) @@ -159,7 +168,7 @@ def fill_history_info(ws, History_pd):# 填充历史记录 try: ws.cell(row=H_START+index, column=1, value=hrow["时间"].strftime("%Y.%m")) 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=3, value=hrow["工资额1"]) ws.cell(row=H_START+index, column=4, value=hrow["级别档次"]) @@ -173,13 +182,13 @@ BaseData["Latest_Prom"] = None for index, row in BaseData.iterrows(): try: - logging.info(f"台账：第{index+1}共{BaseData.shape[0]}现在是{row['身份证号码']}") - print(f"台账：第{index+1}共{BaseData.shape[0]}现在是{row['身份证号码']}") + logging.info(f"台账：第[{index+1}]共[{BaseData.shape[0]}]现在是[{row['身份证号码']}]") BaseData.at[index, "Latest_Role"] = row["初始职务"] BaseData.at[index, "Latest_Prom"] = row["入职时间"].strftime("%Y.%m") wb = load_workbook("个人台账.xlsx") ws = wb.active fill_basic_info(ws, row)# 填充基本信息 + # 查找晋升信息 promote = Promote[Promote["身份证号"] == row["身份证号码"]] if not promote.empty: @@ -199,23 +208,28 @@ for index, row in BaseData.iterrows(): History_pd.loc[len(History_pd)] = [row["入职时间"], row["初始职务"], "", row["入职时的初始级别"], "", "", "套改/定级"] for index, prow in promote.iterrows(): # 添加晋升记录 History_pd.loc[len(History_pd)] = [prow["工资执行时间"], prow["新职务"], "", "", "", "", "晋升"] - calctime=row["二档起始"] + relativedelta(minute=1) - while True: # 添加二级记录 - calctime += relativedelta(years=2) - if calctime > nowtime: - break - History_pd.loc[len(History_pd)] = [calctime, "", "", "", "", "", "两年晋档"] - calctime=row["五档起始"] - while True: # 添加五级记录 - calctime += relativedelta(years=5) - if calctime > nowtime: - break - History_pd.loc[len(History_pd)] = [calctime, "", "", "", "", "", "五年晋级"] + try: + calctime=row["二档起始"] + relativedelta(minute=1) + while True: # 添加二级记录 + calctime += relativedelta(years=2) + if calctime > nowtime: + break + History_pd.loc[len(History_pd)] = [calctime, "", "", "", "", "", "两年晋档"] + calctime=row["五档起始"] + while True: # 添加五级记录 + calctime += relativedelta(years=5) + if calctime > nowtime: + break + History_pd.loc[len(History_pd)] = [calctime, "", "", "", "", "", "五年晋级"] + except: + raise Exception(f"二、五档起始时间格式错误：{row['二档起始']}或{row['五档起始']}") for rule in Rule_Level: # 工资调标 if row["入职时间"] < rule["start"]: History_pd.loc[len(History_pd)] = [rule["start"], "", "", "", "", "", "工资调标"] 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(): # 数据计算 # 调整职务职级 if index > 0 and hrow["职务"] == "": @@ -247,14 +261,13 @@ for index, row in BaseData.iterrows(): fill_history_info(ws, History_pd)# 填充历史记录 wb.save(f"./output/{row['姓名']}_{row['身份证号码']}.xlsx") except Exception as e: - logging.error(e) + logging.error(f"{row['身份证号码']}:{e}") wb = load_workbook("./汇总.xlsx") ws = wb.active for index, row in BaseData.iterrows(): # 汇总 try: - logging.info(f"汇总：第{index+1}共{BaseData.shape[0]}现在是{row['身份证号码']}") - print(f"汇总：第{index+1}共{BaseData.shape[0]}现在是{row['身份证号码']}") + logging.info(f"汇总：第[{index+1}]共[{BaseData.shape[0]}]现在是[{row['身份证号码']}]") for col in range(1,16): 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) @@ -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=5, 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=9, 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=14, value=row["日期2"]) except Exception as e: - logging.error(e) + logging.error(f"{row['身份证号码']}:{e}") wb.save("./名册.xlsx") # 保存汇总 if max_promote > 0: - logging.error(f"最多有{max_promote}条晋升信息，需要调整模板。记得同时调整薪资历史的起始行和个人评价结果。") + logging.warning(f"最多有[{max_promote}]条晋升信息，需要调整模板。记得同时调整薪资历史的起始行和个人评价结果。") if max_history > 0: - logging.error(f"最多有{max_history}条薪资历史，需要调整模板。") \ No newline at end of file + logging.warning(f"最多有[{max_history}]条薪资历史，需要调整模板。") \ No newline at end of file