""" @author: luojunhui """ import time import schedule from datetime import datetime from functions import MatchRate, RateDetail from feishu import Feishu def job(): """ 定时任务方法 :return: """ MR = MatchRate() RD = RateDetail() F = Feishu(document_token="QF4YsYNJHhSOy3t6OM7cYfrBnq3") today_str = datetime.today().strftime("%Y%m%d") e_time = MR.generate_today_stamp(today_str) s_time = MR.generate_yesterday_stamp(today_str) result_list = MR.match_rate(start_time_stamp=s_time, end_time_stamp=e_time) result_obj = RD.rate_and_error_list(result_list) rate_list = [ datetime.utcfromtimestamp(e_time / 1000).strftime("%Y%m%d"), result_obj['total_count'], result_obj['success_count'], result_obj['success_count'] / result_obj['total_count'] if result_obj['total_count'] else None, result_obj['fail_count'], result_obj['fail_count'] / result_obj['total_count'] if result_obj['total_count'] else None, result_obj['processing_count'], result_obj['processing_count'] / result_obj['total_count'] if result_obj['total_count'] else None, ] error_list = result_obj['error_list'] # insert rate_list rate_sheet_id = "c65def" F.prepend_value(sheet_id=rate_sheet_id, values=[["******"]], ranges="A2:A2") F.insert_value(sheet_id=rate_sheet_id, values=[rate_list], ranges="A2:H2") # insert error_list error_sheet_id = "67wu2O" for item in error_list: F.prepend_value(sheet_id=error_sheet_id, values=[["*****"]], ranges="A2:A2") F.insert_value(sheet_id=error_sheet_id, values=[[today_str] + item], ranges="A2:D2") if __name__ == '__main__': job() # schedule.every().day.at("01:00").do(job) # while True: # schedule.run_pending() # time.sleep(1)