refactor(历史记录计算): 重构历史记录计算逻辑,优化字段顺序和继承关系

调整历史记录字段顺序,将复杂计算字段移到后面
重构数据继承逻辑,简化职务和级别档次的处理
优化晋升校验规则和工资计算流程
This commit is contained in:
Miu Li 2025-06-14 04:21:37 +08:00
parent d5ef9972a1
commit ced4de4d4b

59
main.py
View File

@ -348,8 +348,8 @@ def main():
"身份证号码", "姓名", # 统一填入 "身份证号码", "姓名", # 统一填入
"变动后时间", "变动后职务", "变动原因", "晋升备注", # 直接填入 "变动后时间", "变动后职务", "变动原因", "晋升备注", # 直接填入
"工龄", "五年1级年份", "两年1档年份", # 简单计算更新 "工龄", "五年1级年份", "两年1档年份", # 简单计算更新
"变动后级别档次", "变动后职务工资", "变动后级别工资", "变动后津贴工资", "变动后工资合计", # 复杂计算更新 "变动前时间", "变动前职务", "变动前级别档次", "变动前职务工资", "变动前级别工资", "变动前津贴工资", "变动前工资合计", # 排序后更新
"变动前时间", "变动前职务", "变动前级别档次", "变动前职务工资", "变动前级别工资", "变动前津贴工资", "变动前工资合计",]) # 排序后(最后)更新 "变动后级别档次", "变动后职务工资", "变动后级别工资", "变动后津贴工资", "变动后工资合计"]) # 复杂计算更新
# 添加入职记录 # 添加入职记录
History_pd.loc[len(History_pd), ["变动后时间","变动后职务","变动原因","变动后级别档次"]] = [ History_pd.loc[len(History_pd), ["变动后时间","变动后职务","变动原因","变动后级别档次"]] = [
row["入职时间"],row["初始职务"],"套改/定级",row["入职时的初始级别"]] row["入职时间"],row["初始职务"],"套改/定级",row["入职时的初始级别"]]
@ -385,27 +385,34 @@ def main():
if History_pd.at[0,"时间"] != row["入职时间"]: if History_pd.at[0,"时间"] != row["入职时间"]:
raise Exception(f"入职时间晚于其他时间:{row['入职时间']} < {History_pd.at[0,'时间']} ({History_pd.at[0,'变动原因']})") raise Exception(f"入职时间晚于其他时间:{row['入职时间']} < {History_pd.at[0,'时间']} ({History_pd.at[0,'变动原因']})")
for index, hrow in History_pd.iterrows(): # 数据计算
# 调整职务职级 # 复杂数据计算
if index > 0 and hrow["职务"] == "": for index, hrow in History_pd.iterrows():
History_pd.at[index, "职务"] = History_pd.iloc[index-1]["职务"] # 继承上一条复杂计算数据
for rule in Rule_RoleName: # 名称变化 if index > 0:
if rule["start"] <= hrow["时间"] <= rule["end"]: History_pd.loc[index,["变动前时间", "变动前职务", "变动前级别档次", "变动前职务工资", "变动前级别工资", "变动前津贴工资", "变动前工资合计"]] = History_pd.loc[index - 1,["变动前时间", "变动前职务", "变动前级别档次", "变动前职务工资", "变动前级别工资", "变动前津贴工资", "变动前工资合计"]]
if History_pd.iloc[index]["职务"] in rule["rule"]["原名称"].values: # 继承名称
History_pd.at[index, "职务"] = rule["rule"][rule["rule"]["原名称"] == History_pd.iloc[index]["职务"]]["现名称"].values[0] if hrow["变动后职务"] == "":
if index > 0 and hrow["级别档次"] == "": History_pd.at[index,"变动后职务"] = History_pd.at[index,"变动前职务"]
jb, dc = split_level(History_pd.iloc[index-1]["级别档次"]) # 名称变化
for rule in Rule_RoleName:
if rule["start"] <= hrow["变动后时间"] <= rule["end"]:
if History_pd.at[index,"变动后职务"] in rule["rule"]["原名称"].values:
History_pd.at[index,"变动后职务"] = rule["rule"][rule["rule"]["原名称"] == History_pd.at[index,"变动后职务"]]["现名称"].values[0]
# 级别档次
if index > 0:
jb, dc = split_level(History_pd.at[index,"变动前级别档次"])
if hrow["变动原因"] == "两年晋档": if hrow["变动原因"] == "两年晋档":
History_pd.at[index, "级别档次"] = f"{jb}-{dc+1}" History_pd.at[index, "变动后级别档次"] = f"{jb}-{dc+1}"
elif hrow["变动原因"] == "五年晋级": elif hrow["变动原因"] == "五年晋级":
if jb-1 < 1 or jb-1 < role_limit(History_pd.iloc[index]["职务"]): if jb-1 < 1 or jb-1 < role_limit(History_pd.iloc[index]["职务"]):
History_pd.at[index, "级别档次"] = f"{jb}-{dc+1}" History_pd.at[index, "变动后级别档次"] = f"{jb}-{dc+1}"
else: else:
History_pd.at[index, "级别档次"] = f"{jb-1}-{dc-1}" History_pd.at[index, "变动后级别档次"] = f"{jb-1}-{dc-1}"
elif hrow["变动原因"] == "工资调标": elif hrow["变动原因"] == "工资调标":
History_pd.at[index, "级别档次"] = f"{jb}-{dc}" History_pd.at[index, "变动后级别档次"] = f"{jb}-{dc}"
elif hrow["变动原因"] == "晋升": elif hrow["变动原因"] == "晋升":
role = History_pd.iloc[index]["职务"] role = History_pd.iloc[index]["变动后职务"]
if role in Promote_Level.keys(): if role in Promote_Level.keys():
new_jb = jb + Promote_Level[role][0] new_jb = jb + Promote_Level[role][0]
new_dc = dc + Promote_Level[role][1] new_dc = dc + Promote_Level[role][1]
@ -415,22 +422,24 @@ def main():
else: else:
new_jb = int(new_jb) new_jb = int(new_jb)
new_dc = int(new_dc) new_dc = int(new_dc)
if (History_pd.iloc[index-1]["职务"] in Promote_verify.iloc[:,0].values and if (History_pd.at[index,"变动后职务"] in Promote_verify.iloc[:,0].values and
role in Promote_verify.iloc[:,1].values): role in Promote_verify.iloc[:,1].values):
logging.info(f"[{row['身份证号码']}]命中晋升校验规则[{History_pd.iloc[index-1]['职务']}]->[{role}]") logging.info(f"[{row['身份证号码']}]命中晋升校验规则[{History_pd.at[index,'变动前职务']}]->[{role}]")
History_pd.at[index, "级别档次"] = f"{jb}-{dc}" History_pd.at[index, "变动后级别档次"] = f"{jb}-{dc}"
elif new_jb < role_limit(role): elif new_jb < role_limit(role):
History_pd.at[index, "级别档次"] = f"{jb}-{dc+1}" History_pd.at[index, "变动后级别档次"] = f"{jb}-{dc+1}"
elif new_jb < 1 or new_dc < 1: elif new_jb < 1 or new_dc < 1:
raise Exception(f"级别档次小于0[{new_jb}]-[{new_dc}]") raise Exception(f"级别档次小于0[{new_jb}]-[{new_dc}]")
else: else:
History_pd.at[index, "级别档次"] = f"{new_jb}-{new_dc}" History_pd.at[index, "变动后级别档次"] = f"{new_jb}-{new_dc}"
else: else:
logging.warning(f"职位[{role}]不存在职级上限规则") logging.warning(f"职位[{role}]不存在职级上限规则")
else:
History_pd.at[index, "变动后级别档次"] = History_pd.at[index, "变动前级别档次"]
# 计算工资 # 计算工资
History_pd.at[index, "职务工资"] = role_salary(History_pd.iloc[index]["职务"], hrow["时间"]) History_pd.at[index, "变动后职务工资"] = role_salary(History_pd.iloc[index]["职务"], hrow["时间"])
History_pd.at[index, "级别工资"] = level_salary(History_pd.iloc[index]["级别档次"], hrow["时间"]) History_pd.at[index, "变动后级别工资"] = level_salary(History_pd.iloc[index]["级别档次"], hrow["时间"])
History_pd.at[index, "工资合计"] = to_int(History_pd.iloc[index]["职务工资"]) + to_int(History_pd.iloc[index]["级别工资"]) History_pd.at[index, "变动后工资合计"] = to_int(History_pd.iloc[index]["职务工资"]) + to_int(History_pd.iloc[index]["级别工资"])
fill_history_info(ws, History_pd)# 填充历史记录 fill_history_info(ws, History_pd)# 填充历史记录