main.py

@@ -9,91 +9,34 @@ from datetime import datetime
 # 配置日志记录
 logging.basicConfig(filename='log.txt', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 
-# 常量
-
 P_LIMIT = 6 # 最大晋升次数
 
 P_START = 10 # 晋升记录开始行
 H_START = 15 + P_LIMIT # 历史记录开始行
 
-NOWTIME = datetime.now()
+# 自定义日期解析函数
-
-# 全局变量
-
-## 通过函数读取
-BaseData = pd.DataFrame()
-Promote = pd.DataFrame()
-Rule_Role = []
-Rule_Level = []
-Rule_RoleName = []
-Level_Limit = pd.DataFrame()
-Promote_Level = pd.DataFrame()
-Promote_verify = pd.DataFrame()
-
-## 统计量
-max_promote = 0
-max_history = 0
-
-# 工具函数
-
 def custom_date_parser(x):
     try:
         return datetime.strptime(x, '%Y-%m-%d')
     except:
         return x
 
-def format_time(dt,info):
-    try:
-        return dt.strftime("%Y.%m")
-    except:
-        logging.warning(f"[{info}]时间格式错误：{dt}")
-        return dt
-
-def to_int(x):
-    try:
-        return int(x)
-    except:
-        return 0
-
-def fallback(x):
-    for i in x:
-        if pd.notna(i) and i != '':
-            return i
-    return ''
-
-def split_level(level:str):
-    try:
-        parts = level.split('-')
-        return (int(parts[0]), int(parts[1]))
-    except:
-        raise Exception(f"职级[{level}]格式错误")
-
-# 读取信息
-
-def read_base_data(): # 读取员工数据
-    global BaseData
 BaseData = pd.read_excel("原数据.xlsx", sheet_name="入职信息")
+Promote = pd.read_excel("原数据.xlsx", sheet_name="职务变动") # 
+
+for index, row in BaseData.iterrows():
     for col in ["出生年月","任职年月","原职时间","参加工作时间","入职时间", "晋档起始", "晋级起始", "日期2"]:
-        BaseData[col] = BaseData[col].apply(custom_date_parser)
+        BaseData.at[index, col] = custom_date_parser(row[col])
-    for col in ["晋档起始", "晋级起始"]:
-        BaseData[col] = BaseData[col].apply(lambda x: datetime(x.year, 1, 1) if isinstance(x, datetime) else x)
-    BaseData["Latest_Role"] = None
-    BaseData["Latest_Prom"] = None
-    BaseData["工龄调增"] = BaseData["工龄调增"].apply(to_int)
-    BaseData["工龄调减"] = BaseData["工龄调减"].apply(to_int)
-    BaseData["学龄"] = BaseData["学龄"].apply(to_int)
-    BaseData["工龄"] = BaseData.apply(lambda row: NOWTIME.year-row["参加工作时间"].year+row["工龄调增"]-row["工龄调减"]+1, axis=1)
 
-def read_promote(): # 读取晋升记录
+for index, row in Promote.iterrows():
-    global Promote
-    Promote = pd.read_excel("原数据.xlsx", sheet_name="职务变动")
     for col in ["任职时间","工资执行时间"]:
-        Promote[col] = Promote[col].apply(custom_date_parser)
+        Promote.at[index, col] = custom_date_parser(row[col])
 
-def read_rule_role(): # 读取职位规则
+logging.info("人员信息加载完成")
-    global Rule_Role
+
+Rule_Role = []
 col = 4
-    while True: 
+while True: # 职位规则
     try:
         rule = pd.read_excel("原数据.xlsx", sheet_name="职位规则",usecols=f"{get_column_letter(col)}:{get_column_letter(col+1)}", header=None)
         Rule_Role.append({
@@ -104,10 +47,7 @@ def read_rule_role(): # 读取职位规则
         col += 2
     except:
         break
-    Rule_Role = sorted(Rule_Role, key=lambda x: x['start'])
+Rule_Level = []
-
-def read_rule_level(): # 读取职级规则
-    global Rule_Level
 col = 1
 while True: # 职级规则
     try:
@@ -120,10 +60,7 @@ def read_rule_level(): # 读取职级规则
         col += 2
     except:
         break
-    Rule_Level = sorted(Rule_Level, key=lambda x: x['start'])
+Rule_RoleName = []
-
-def read_rule_role_name(): # 读取名称变化规则
-    global Rule_RoleName
 col = 1
 while True: # 名称变化
     try:
@@ -136,10 +73,8 @@ def read_rule_role_name(): # 读取名称变化规则
         col += 2
     except:
         break
-    Rule_RoleName = sorted(Rule_RoleName, key=lambda x: x['start'])
 
-def read_level_limit(): # 读取职位对应的级别限制
+# 读取职位对应的级别限制
-    global Level_Limit, Promote_Level
 Level_Limit_tmp = pd.read_excel("原数据.xlsx", sheet_name="职位规则", usecols="A:A", skiprows=2, names=["limit"])
 Promote_Level_tmp = pd.read_excel("原数据.xlsx", sheet_name="职位规则", usecols="B:C", skiprows=2, names=["级别","档次"])
 Level_Limit = {}
@@ -149,24 +84,24 @@ def read_level_limit(): # 读取职位对应的级别限制
         Level_Limit[row["role"]] = Level_Limit_tmp.iloc[index]["limit"]
         Promote_Level[row["role"]] = (Promote_Level_tmp.iloc[index]["级别"], Promote_Level_tmp.iloc[index]["档次"])
 
-def read_promote_verify(): # 读取晋升校验
+# 晋升校验
-    global Promote_verify
+Promote_verify = Level_Limit_tmp = pd.read_excel("原数据.xlsx", sheet_name="晋升校验", usecols="A:B")
-    Promote_verify = pd.read_excel("原数据.xlsx", sheet_name="晋升校验", usecols="A:B")
 
-def load_people():
-    read_base_data()
-    read_promote()
-    logging.info("人员信息加载完成")
-
-def load_rule():
-    read_rule_role()
-    read_rule_level()
-    read_rule_role_name()
-    read_level_limit()
-    read_promote_verify()
 logging.info("规则加载完成")
 
-# 获取配置类函数
+Rule_Role = sorted(Rule_Role, key=lambda x: x['start'])
+Rule_Level = sorted(Rule_Level, key=lambda x: x['start'])
+Rule_RoleName = sorted(Rule_RoleName, key=lambda x: x['start'])
+
+nowtime = datetime.now()
+
+def split_level(level:str):
+    try:
+        parts = level.split('-')
+        return (int(parts[0]), int(parts[1]))
+    except:
+        raise Exception(f"职级[{level}]格式错误")
+        return (0, 0)
 
 def role_salary(role:str, time):
     for rule in Rule_Role:
@@ -198,7 +133,30 @@ def role_limit(role:str):
         logging.warning(f"职位[{role}]不存在职级上限规则")
         return -1
 
-# 填充类辅助函数
+def format_time(dt,info):
+    try:
+        return dt.strftime("%Y.%m")
+    except:
+        logging.warning(f"[{info}]时间格式错误：{dt}")
+        return dt
+
+def to_int(x):
+    try:
+        return int(x)
+    except:
+        return 0
+BaseData["工龄调增"] = BaseData["工龄调增"].apply(to_int)
+BaseData["工龄调减"] = BaseData["工龄调减"].apply(to_int)
+BaseData["学龄"] = BaseData["学龄"].apply(to_int)
+
+def fallback(x):
+    for i in x:
+        if pd.notna(i) and i != '':
+            return i
+    return ''
+
+max_promote = 0
+max_history = 0
 
 def fill_basic_info(ws, row):# 填充基本信息
     ws.cell(row=2, column=1, value=f"部门：{row['部门']}    职务：{row['职务']}")
@@ -248,13 +206,11 @@ def fill_history_info(ws, History_pd):# 填充历史记录
         ws.cell(row=H_START+index, column=7, value=hrow["变动原因"])
         # ws.cell(row=H_START+index, column=8, value=index) # Debug
 
-def main():
+BaseData["Latest_Role"] = None
-
+BaseData["Latest_Prom"] = None
-    load_people()
-    load_rule()
 
 # 创建一个空的DataFrame来存储所有历史记录
-    all_history = pd.DataFrame(columns=["身份证号码", "姓名", "时间", "职务", "职务工资", "级别档次", "级别工资", "工资合计", "变动原因", "晋升备注"])
+all_history = pd.DataFrame(columns=["身份证号码", "姓名", "时间", "职务", "职务工资", "级别档次", "级别工资", "工资合计", "变动原因"])
 
 for index, row in BaseData.iterrows():
     try:
@@ -268,7 +224,7 @@ def main():
         # 查找晋升信息
         promote = Promote[Promote["身份证号"] == row["身份证号码"]]
         if not promote.empty:
-                promote = promote.sort_values(by=["工资执行时间", "任职时间"], ascending=[False, False]).reset_index(drop=True)
+            promote = promote.sort_values(by="任职时间", ascending=False).reset_index(drop=True)
             BaseData.at[index, "Latest_Role"] = promote.iloc[0]["新职务"]
             BaseData.at[index, "Latest_Prom"] = promote.iloc[0]["任职时间"]
             # 把原职务取出来
@@ -279,33 +235,33 @@ def main():
                 BaseData.at[index, "职务2"] = row["初始职务"]
                 BaseData.at[index, "日期2"] = row["入职时间"]
 
-                promote = promote.sort_values(by=["工资执行时间", "任职时间"]).reset_index(drop=True)
+            promote = promote.sort_values(by="任职时间").reset_index(drop=True)
             fill_prompt_info(ws, promote)# 填充晋升信息
             
         # 根据规则匹配职级薪资
-            History_pd = pd.DataFrame(columns=["时间", "职务", "职务工资", "级别档次", "级别工资", "工资合计", "变动原因", "晋升备注"])
+        History_pd = pd.DataFrame(columns=["时间", "职务", "职务工资", "级别档次", "级别工资", "工资合计", "变动原因"])
         # 添加入职记录
-            History_pd.loc[len(History_pd)] = [row["入职时间"], row["初始职务"], "", row["入职时的初始级别"], "", "", "套改/定级", ""]
+        History_pd.loc[len(History_pd)] = [row["入职时间"], row["初始职务"], "", row["入职时的初始级别"], "", "", "套改/定级"]
         for index, prow in promote.iterrows(): # 添加晋升记录
-                History_pd.loc[len(History_pd)] = [prow["工资执行时间"]+relativedelta(hours=prow["任职时间"].month,minutes=prow["任职时间"].day), prow["新职务"], "", "", "", "", "晋升", f"任{prow['新职务']} {prow['变动批注'] if pd.notna(prow['变动批注']) else ''}"]
+            History_pd.loc[len(History_pd)] = [prow["工资执行时间"], prow["新职务"], "", "", "", "", "晋升"]
         try:
             calctime=row["晋档起始"] + relativedelta(minute=1)
             while True: # 添加晋档记录
                 calctime += relativedelta(years=row["晋档间隔"]) 
-                    if calctime > NOWTIME:
+                if calctime > nowtime:
                     break
-                    History_pd.loc[len(History_pd)] = [calctime, "", "", "", "", "", "两年晋档", ""]
+                History_pd.loc[len(History_pd)] = [calctime, "", "", "", "", "", "两年晋档"]
             calctime=row["晋级起始"]
             while True: # 添加晋级记录
                 calctime += relativedelta(years=row["晋级间隔"])
-                    if calctime > NOWTIME:
+                if calctime > nowtime:
                     break
-                    History_pd.loc[len(History_pd)] = [calctime, "", "", "", "", "", "五年晋级", ""]
+                History_pd.loc[len(History_pd)] = [calctime, "", "", "", "", "", "五年晋级"]
         except:
-                raise Exception(f"晋级、档起始或间隔时间格式错误：{row['晋级起始']}-{row['晋档起始']}-{row['晋级间隔']}-{row['晋档间隔']}")
+            raise Exception(f"晋级、档起始时间格式错误：{row['晋档起始']}或{row['晋级起始']}")
         for rule in Rule_Level: # 工资调标
             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)
 
         if History_pd.at[0,"时间"] != row["入职时间"]:
@@ -323,9 +279,6 @@ def main():
                 if hrow["变动原因"] == "两年晋档":
                     History_pd.at[index, "级别档次"] = f"{jb}-{dc+1}"
                 elif hrow["变动原因"] == "五年晋级":
-                        if jb-1 < 1 or jb-1 < role_limit(History_pd.iloc[index]["职务"]):
-                            History_pd.at[index, "级别档次"] = f"{jb}-{dc+1}"
-                        else:
                     History_pd.at[index, "级别档次"] = f"{jb-1}-{dc-1}"
                 elif hrow["变动原因"] == "工资调标":
                     History_pd.at[index, "级别档次"] = f"{jb}-{dc}"
@@ -386,8 +339,8 @@ def main():
         ws.cell(row=6+index, column=4, value=format_time(row["出生年月"], "出生年月"))
         ws.cell(row=6+index, column=5, value=format_time(row["参加工作时间"], "参加工作时间"))
         ws.cell(row=6+index, column=6, value=fallback([row["现学历"],row["学历"]]))
-            ws.cell(row=6+index, column=7, value=row['工龄']+row["学龄"])
+        ws.cell(row=6+index, column=7, value=nowtime.year-row["参加工作时间"].year+row["工龄调增"]-row["工龄调减"]+1+row["学龄"])
-            ws.cell(row=6+index, column=8, value=row['工龄'])
+        ws.cell(row=6+index, column=8, value=nowtime.year-row["参加工作时间"].year+row["工龄调增"]-row["工龄调减"]+1)
         ws.cell(row=6+index, column=9, value=row["学龄"])
         ws.cell(row=6+index, column=10, value=row["工龄调减"])
         ws.cell(row=6+index, column=11, value=row["Latest_Role"])
@@ -402,6 +355,3 @@ def main():
     logging.warning(f"最多有[{max_promote}]条晋升信息，需要调整模板。记得同时调整薪资历史的起始行和个人评价结果。")
 if max_history > 0:
     logging.warning(f"最多有[{max_history}]条薪资历史，需要调整模板。")
-
-if __name__ == "__main__":
-    main()