解决钉钉导出考勤表统计（加班时长、补贴）支持自定义格式输出，生成每个人的考勤统计基于pandas和xlwings对表格进行读写
import pandas as pdimport sysimport osimport numpy as npimport xlwings as xwimport timeimport datetimeimport globsys.path.append(os.path.dirname(os.path.abspath(__file__)))class CheckIn:def __init__(self):passdef check_table(self, url):df = pd.read_excel(url, header=None, index_col=None)df = df[2:]df = df[[0, 1, 5, 6, 7, 8, 9]]df = df.drop(3, axis=0)df.columns = df.iloc[0].valuesdf = df.drop(2, axis=0)# 只处理含‘正常’班和休息的班次df = df[df['上班1打卡结果'].isin(['正常']) | df['班次'].isin(['休息'])]# 必须包含上下班打卡df = df[~df['上班1打卡时间'].isin([np.NaN]) & ~df['下班1打卡时间'].isin([np.NaN])]df['累计加班'] = 0df['加班次数'] = 0df['餐补标准'] = 25df['报销金额'] = 0for _, row in df.iterrows():# 计算上下班时间row['上班1打卡时间'] = str(row['上班1打卡时间'])[0: 5]row['下班1打卡时间'] = str(row['下班1打卡时间'])[0: 5]start_time = row['上班1打卡时间']over_time = row['下班1打卡时间']add_min = 0if '次日' in row['下班1打卡时间']: # 最晚有效下班时间为24：00over_time = '23:59'add_min = 60# print('start_time', str(start_time))start_time = datetime.datetime.strptime(start_time, "%H:%M")over_time = datetime.datetime.strptime(over_time, "%H:%M")work_time = max(round(((over_time - start_time).seconds + 1 + add_min) / 3600), 0)# 餐补统计if ('正常' in row['上班1打卡结果']) and (row['下班1打卡时间'] >= '19:00'):df.loc[row.name, '加班次数'] = 1elif '休息' in row['班次']:if work_time >= 4:df.loc[row.name, '加班次数'] = 1if work_time >= 7:df.loc[row.name, '加班次数'] = 2df.loc[row.name, '报销金额'] = df.loc[row.name, '加班次数'] * df.loc[row.name, '餐补标准']# 加班统计work_plus_time = 0if '正常' in row['上班1打卡结果']:if start_time.hour >= 7: # 有效打卡早7点以后if start_time.hour <= 9: # 九点后累计上班时长start_time = datetime.datetime.strptime('09:00', "%H:%M")work_plus_time = max(round(((over_time - start_time).seconds + 1 + add_min) / 3600 - 9.5),0) # 超过9.5计算加班时长,半小时一个粒度elif '休息' in row['班次']:if work_time >= 4:work_plus_time = work_timedf.loc[row.name, '累计加班'] = work_plus_timeword_plus = url.replace('base', 'new').replace('考勤表', '加班餐补-加班时长明细表')df.to_excel(word_plus, index=False)df['加班日期'] = df['日期'].map(lambda x: x.split(" ")[0] + ' ') + df['下班1打卡时间'].map(str)df.loc[df['班次'] == '休息', '加班日期'] = df['日期'].map(str) + df['上班1打卡时间'].map(str)df_output = df[['姓名', '加班日期', '加班次数', '餐补标准', '报销金额', '上班1打卡结果', '班次', '累计加班']]return df_output[['姓名', '加班日期', '加班次数', '餐补标准', '报销金额', '累计加班', '班次']]def money_table(self, url, data, output_url):data = data[data['加班次数'] > 0 | data['班次'].isin(['休息'])]data = data.drop(['班次'], axis=1)name_list = data['姓名'].unique()wb = xw.Book(url)sht_sheet1 = wb.sheets['Sheet1'] # 要复制的sheetsfor i, v in enumerate(name_list):sht_sheet1.api.Copy(Before=sht_sheet1.api)sht = wb.sheets['Sheet1 (2)']sht.api.Name = vstruct_time = time.localtime(time.time()) # 得到结构化时间格式now_time = time.strftime("%Y/%m/%d", struct_time)sht.range('H4').value = now_timesht.range('B4').value = output_url.split('/')[-1].split('\\')[-1].split('20')[0]item = data[data['姓名'] == v]new_data_rows = item.values.shape[0]item.insert(0, '序号', [i+1 for i in range(new_data_rows)])start_row = 7# last_row = sht.range('A7').expand().last_cell.rowlast_row = 38work_rows = last_row - start_rowsub_rows = abs(new_data_rows - work_rows)# print(f'last{last_row},word{work_rows}')if new_data_rows > work_rows:for _ in range(sub_rows):sht.api.rows(start_row + 1).insertelif new_data_rows < work_rows:for _ in range(sub_rows):sht.api.rows(start_row + 1).delete# print(f'last{sht.range("A7").expand().last_cell.row},new{new_data_rows},word{work_rows}')sht.range('A7').value = item.valueswb.save(output_url)wb.close()def total(self, df, output_url):total_df = df.groupby(by='姓名').sum()[['加班次数', '报销金额', '累计加班']]total_df = total_df.sort_values(by='加班次数', ascending=False)total_df['备注'] = ''total_df = total_df.reset_index()total_df = total_df.append([{'姓名': '合计', '加班次数': total_df['加班次数'].sum(), '报销金额': total_df['报销金额'].sum(),'累计加班': total_df['累计加班'].sum(), '备注':''}], ignore_index=True)total_df.to_excel(output_url, index=None)if __name__ == '__main__':t = time.time()year = ''month = '5'root = '../../data/check'standard_url = f'{root}/研发加班餐补08_standard.xlsx'for _, m in enumerate(range(10, 11, 1)):month = mprint(f'生成月份:{month}')for check_url in glob.glob(f"{root}/{year}-{month}/base/*.xlsx"):if '$' in check_url:continuecheck_in = CheckIn()check_df = check_in.check_table(url=check_url)new_url = check_url.replace('base', 'new')total_output_url = new_url.replace('考勤表', '总计表')money_output_url = new_url.replace('考勤表', '加班餐补表')check_in.total(df=check_df, output_url=total_output_url)check_df = check_df.drop('累计加班', axis=1)check_in.money_table(url=standard_url, data=check_df, output_url=money_output_url)print(time.time()-t)# print(df_output)# df_output.to_excel(url.replace('.xlsx', '_清洗.xlsx'), index=None)
