基于Python進(jìn)行定時(shí)任務(wù)管理封裝
更新時(shí)間：2024年12月24日 09:49:41 作者：qh0526wy
這篇文章主要為大家詳細(xì)介紹了如何基于Python進(jìn)行定時(shí)任務(wù)管理封裝,文中的示例代碼講解詳細(xì),具有一定的借鑒價(jià)值,有需要的小伙伴可以參考下
效果圖
主邏輯代碼
# -*- coding: utf-8 -*-
# import apscheduler
import pandas as pd
from datetime import datetime
# 導(dǎo)入調(diào)度器，此處使用BackgroundScheduler阻塞調(diào)度器
from apscheduler.schedulers.background import BackgroundScheduler
# 導(dǎo)入觸發(fā)器，此處使用IntervalTrigger特定時(shí)間間隔觸發(fā)
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.date import DateTrigger
from apscheduler.executors.pool import ThreadPoolExecutor,ProcessPoolExecutor
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.util import undefined
import os,time,sys,random,copy,json
from QhTestJob import Ui_Form
from PyQt5.QtWidgets import QApplication
from PyQt5 import QtWidgets
from PyQt5.QtCore import Qt
# cron定時(shí)調(diào)度（某一定時(shí)時(shí)刻執(zhí)行）
# (int|str) 表示參數(shù)既可以是int類型，也可以是str類型
# (datetime | str) 表示參數(shù)既可以是datetime類型，也可以是str類型
# year (int|str) – 4-digit year -（表示四位數(shù)的年份，如2008年）
# month (int|str) – month (1-12) -（表示取值范圍為1-12月）
# day (int|str) – day of the (1-31) -（表示取值范圍為1-31日）
# week (int|str) – ISO week (1-53) -（格里歷2006年12月31日可以寫成2006年-W52-7（擴(kuò)展形式）或2006W527（緊湊形式））
# day_of_week (int|str) – number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun) - （表示一周中的第幾天，既可以用0-6表示也可以用其英語(yǔ)縮寫表示）
# hour (int|str) – hour (0-23) - （表示取值范圍為0-23時(shí)）
# minute (int|str) – minute (0-59) - （表示取值范圍為0-59分）
# second (int|str) – second (0-59) - （表示取值范圍為0-59秒）
# start_date (datetime|str) – earliest possible date/time to trigger on (inclusive) - （表示開(kāi)始時(shí)間）
# end_date (datetime|str) – latest possible date/time to trigger on (inclusive) - （表示結(jié)束時(shí)間）
# timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations (defaults to scheduler timezone) -（表示時(shí)區(qū)取值）
# interval 間隔調(diào)度（每隔多久執(zhí)行）
# weeks (int) – number of weeks to wait
# days (int) – number of days to wait
# hours (int) – number of hours to wait
# minutes (int) – number of minutes to wait
# seconds (int) – number of seconds to wait
# start_date (datetime|str) – starting point for the interval calculation
# end_date (datetime|str) – latest possible date/time to trigger on
# timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations
# date 定時(shí)調(diào)度（作業(yè)只會(huì)執(zhí)行一次）
# run_date (datetime|str) – the date/time to run the job at -（任務(wù)開(kāi)始的時(shí)間）
# timezone (datetime.tzinfo|str) – time zone for run_date if it doesn't have one already
def QhRR(QhVALUE,QhVALUE1):
print("{}--{}執(zhí)行定時(shí)任務(wù)...".format(QhVALUE,QhVALUE1))
class QhApscheduler():
QhFiled = {
"QhJabId": "None", # 任務(wù)ID
"QhJabName": "None", # 任務(wù)名稱
"QhJabFuncName": "None", # 任務(wù)名稱 執(zhí)行的程序名
"QhTimesType": "None", # 任務(wù)類型 重復(fù) 間隔 一次
"QhYear": "None", # 年
"QhMonth": "None", # 月
"QhDay": "None", # 日
"QhWeek": "None", # 周
"QhDayOfWeek": "None", # 星期幾
"QhHour": "None", # 小時(shí)
"QhMinute": "None", # 分鐘
"QhSecond": "None", # 秒鐘
"QhJabArgs": "None", # 任務(wù)參數(shù) 函數(shù)參數(shù)
"QhJabKwargs": "None", # 任務(wù)參數(shù) 函數(shù)參數(shù)
"QhJabStartDate": "None", # 開(kāi)始時(shí)間
"QhJabNextRunTime": "None", # 下次運(yùn)行時(shí)間
"QhJabLastRunTime": "None", # 上次運(yùn)行時(shí)間
"QhJabStatus": "None" # 任務(wù)狀態(tài)
}
def __init__(self,
*args, **kwargs):
self.QhCheduPath = os.path.dirname(os.path.abspath(__file__)) #絕對(duì)路徑 用于獲取定時(shí)模塊的根目錄
print(self.QhCheduPath)
self.QhCheduPoolJobDf = self._QhinitJobPoolCsv()
QhExecutors = {
'default':ThreadPoolExecutor(20),
'processpool':ProcessPoolExecutor(10)
}
self.QhJobstores = {
'default': SQLAlchemyJobStore(url='sqlite:///QhJobPoolDb/QhJabSqlite.sqlite')
}
self.Qhscheduler = BackgroundScheduler(jobstores=self.QhJobstores,executors=QhExecutors,misfire_grace_time=3,coalescing=True)
self.Qhscheduler.start()
self._QhInitAddJobFor()
def _QhInitAddJobFor(self,QhIsFor = True):
if self.QhCheduPoolJobDf.shape[0] == 0: return
for QhInx,QhRow in self.QhCheduPoolJobDf.iterrows():
QhJabId = QhRow["QhJabId"]
QhJabName = QhRow["QhJabName"]
QhJabFuncName = QhRow["QhJabFuncName"]
QhTimesType = QhRow["QhTimesType"]
QhYear = self.QhGeShiZH(QhRow["QhYear"])
QhMonth = self.QhGeShiZH(QhRow["QhMonth"])
QhDay = self.QhGeShiZH(QhRow["QhDay"])
QhWeek = self.QhGeShiZH(QhRow["QhWeek"])
QhDayOfWeek = self.QhGeShiZH(QhRow["QhDayOfWeek"])
QhHour = self.QhGeShiZH(QhRow["QhHour"])
QhMinute = self.QhGeShiZH(QhRow["QhMinute"])
QhSecond = self.QhGeShiZH(QhRow["QhSecond"])
QhJabArgs = self.QhGeShiZH(QhRow["QhJabArgs"])
QhJabKwargs = self.QhGeShiZH(QhRow["QhJabKwargs"])
QhJabStartDate = QhRow["QhJabStartDate"]
QhJabNextRunTime = QhRow["QhJabNextRunTime"]
# QhJabLastRunTime = QhRow["QhJabLastRunTime"]
# QhJabEndDate = QhRow["QhJabEndDate"]
QhJabStatus = self.QhGeShiZH(QhRow["QhJabStatus"])
print(QhYear,QhMonth,QhDay,QhWeek,QhDayOfWeek,QhHour,QhMinute,QhSecond)
print(type(QhYear),type(QhMonth),type(QhDay),type(QhWeek),type(QhDayOfWeek),type(QhHour),type(QhMinute),type(QhSecond))
self.QhAddJob(QhJabId,
QhTimesType=QhTimesType,
QhJabFuncName = QhJabFuncName,
QhJabArgs=QhJabArgs,
QhJabKwargs=QhJabKwargs,
QhName = QhJabName,
QhYear = QhYear,
QhMonth = QhMonth,
QhDay = QhDay,
QhWeek = QhWeek,
QhDayOfWeek = QhDayOfWeek,
QhHour = QhHour,
QhMinute = QhMinute,
QhSecond = QhSecond,
QhIsFor = QhIsFor,
QhJabStatus=QhJabStatus)
# QhIsFor當(dāng)為True時(shí)，保存到csv文件，批量增加時(shí)的場(chǎng)景
if QhIsFor == True:self.QhCheduPoolJobDf.to_csv(self.QhCheduPoolJobCsv, index=False,encoding='gbk')
def _QhinitJobPoolCsv(self):
# 初始化定時(shí)任務(wù)數(shù)據(jù)庫(kù)文件夾
# 作者：闕輝
QhCheduPathpd = "{}\QhJobPoolDb".format(self.QhCheduPath)
if os.path.exists(QhCheduPathpd):
print("{} is exist!".format(QhCheduPathpd))
else:
os.mkdir(QhCheduPathpd)
print("{} is not exist, create it!".format(QhCheduPathpd))
self.QhCheduPoolJobCsv = "{}\QhJobPoolCsv.csv".format(QhCheduPathpd)
if not os.path.exists(self.QhCheduPoolJobCsv):
QhFiled = list(QhApscheduler.QhFiled.keys())
self.QhCheduPoolJobDf = pd.DataFrame(columns=QhFiled)
self.QhCheduPoolJobDf.to_csv(self.QhCheduPoolJobCsv, index=False)
else:
self.QhCheduPoolJobDf = pd.read_csv(self.QhCheduPoolJobCsv, encoding='gbk')
self.QhCheduPoolJobDf.fillna("None", inplace=True)
return self.QhCheduPoolJobDf
def QhAddJob(self,
QhJabId,
QhTimesType, # 重復(fù) 間隔 一次
QhJabFuncName,
QhJabArgs=None,
QhJabKwargs=None,
QhName=None,
QhYear = None,
QhMonth = None,
QhDay = None,
QhWeek = None,
QhDayOfWeek = None,
QhHour = None,
QhMinute = None,
QhSecond = None,
QhStartDate = None,
QhEndDate = None,
QhTimezone=None,
QhJitter=None,
misfire_grace_time=undefined,
next_run_time=undefined,
jobstore='default',
executor='default',
coalesce=undefined,
max_instances=undefined,
replace_existing=False,
QhIsFor = False,
QhJabStatus = None,
QhJobIsOpen = False,
):
if self.Qhscheduler.get_job(QhJabId):
print("{} is exist!".format(QhJabId))
return # 如果任務(wù)已經(jīng)存在，則不添加
if QhJabStatus == "已關(guān)閉" and (not QhJobIsOpen):
print("{} 任務(wù)已關(guān)閉，不用新建!".format(QhJabId))
return
QhTiggers = self.QhTiggers(QhTimesType =QhTimesType,
QhYear = QhYear,
QhMonth = QhMonth,
QhDay = QhDay,
QhWeek = QhWeek,
QhDayOfWeek = QhDayOfWeek,
QhHour = QhHour,
QhMinute = QhMinute,
QhSecond = QhSecond,
QhStartDate = QhStartDate,
QhEndDate = QhEndDate,
QhTimezone=QhTimezone,
QhJitter=QhJitter)
if QhJabArgs!=None:
QhJabArgs = tuple(QhJabArgs.split("+"))
if QhJabKwargs != None:
QhJabKwargs = QhJabKwargs.replace("+",',')
QhJabKwargs = QhJabKwargs.replace("'",'"')
QhJabKwargs = json.loads(QhJabKwargs)
self.Qhscheduler.add_job(func=globals()[QhJabFuncName],
trigger=QhTiggers,
args=QhJabArgs,
kwargs=QhJabKwargs,
id=QhJabId,
name=QhName,
misfire_grace_time=misfire_grace_time,
next_run_time=next_run_time,
jobstore=jobstore,
executor=executor,
coalesce=coalesce,
max_instances=max_instances,
replace_existing=replace_existing,)
# 函數(shù)參數(shù)還原輸入格式處理 闕輝
# print(QhJabArgs,type(QhJabArgs))
if isinstance(QhJabArgs, tuple):
QhJabArgs = str(QhJabArgs).replace("'",'').replace("(",'').\
replace(")",'').replace(",","+").replace("+ ","+").replace("+ ","+")
# print(QhJabArgs,type(QhJabArgs))
# print(QhJabKwargs,type(QhJabKwargs))
if isinstance(QhJabKwargs, dict):
QhJabKwargs = str(QhJabKwargs).replace("'",'"').replace(",","+")
# print(QhJabKwargs,type(QhJabKwargs))
QhAddJoblDic = copy.deepcopy(QhApscheduler.QhFiled)
QhAddJoblDic["QhJabId"] = QhJabId
QhAddJoblDic["QhJabName"] = QhName
QhAddJoblDic["QhJabFuncName"] = "None" if QhJabFuncName==None else QhJabFuncName
QhAddJoblDic["QhTimesType"] = QhTimesType
QhAddJoblDic["QhYear"] = "None" if QhYear==None else QhYear
QhAddJoblDic["QhMonth"] = "None" if QhMonth==None else QhMonth
QhAddJoblDic["QhDay"] = "None" if QhDay==None else QhDay
QhAddJoblDic["QhWeek"] = "None" if QhWeek==None else QhWeek
QhAddJoblDic["QhDayOfWeek"] = "None" if QhDayOfWeek==None else QhDayOfWeek
QhAddJoblDic["QhHour"] = "None" if QhHour==None else QhHour
QhAddJoblDic["QhMinute"] = "None" if QhMinute==None else QhMinute
QhAddJoblDic["QhSecond"] = "None" if QhSecond==None else QhSecond
QhAddJoblDic["QhJabArgs"] = "None" if QhJabArgs==None else QhJabArgs
QhAddJoblDic["QhJabKwargs"] = "None" if QhJabKwargs==None else QhJabKwargs
# QhAddJoblDic["QhJabStartDate"] = "None"
# QhAddJoblDic["QhJabNextRunTime"] = "None"
# QhAddJoblDic["QhJabLastRunTime"] = "None"
QhAddJoblDic["QhJabStatus"] = "已運(yùn)行"
try: # 任務(wù)id存在，則更新任務(wù)，不存在則新增
self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf["QhJabId"]==QhJabId].index[0]
for QhKey,QhValue in QhAddJoblDic.items():
if QhKey == "QhJabId":continue
self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf["QhJabId"]==QhJabId,QhKey] = QhValue
except:
QhCheduPoolJobDfRow = pd.DataFrame([QhAddJoblDic])
print(QhCheduPoolJobDfRow)
try:
self.QhCheduPoolJobDf = self.QhCheduPoolJobDf._append(QhCheduPoolJobDfRow)
except:
self.QhCheduPoolJobDf = self.QhCheduPoolJobDf.append(QhCheduPoolJobDfRow)
print(self.QhCheduPoolJobDf)
# QhIsFor當(dāng)為False時(shí)，保存到csv文件，單個(gè)增加時(shí)的場(chǎng)景
if (not QhIsFor):self.QhCheduPoolJobDf.to_csv(self.QhCheduPoolJobCsv, index=False,encoding='gbk')
def QhTiggers(self,QhTimesType,
QhYear = None,
QhMonth = None,
QhDay = None,
QhWeek = None,
QhDayOfWeek = None,
QhHour = None,
QhMinute = None,
QhSecond = None,
QhStartDate = None,
QhEndDate = None,
QhTimezone=None,
QhJitter=None
):
if QhTimesType == "重復(fù)":
QhTiggers = CronTrigger(year=QhYear,
month=QhMonth,
day=QhDay,
week=QhWeek,
day_of_week=QhDayOfWeek,
hour=QhHour,
minute=QhMinute,
second=QhSecond,
start_date=QhStartDate,
end_date=QhEndDate,
timezone=QhTimezone,
jitter=QhJitter)
elif QhTimesType == "間隔":
QhTiggers = IntervalTrigger(weeks= int(0) if QhWeek == None else QhWeek,
days= 0 if QhDay == None else QhDay,
hours= 0 if QhHour == None else QhHour,
minutes= 0 if QhMinute == None else QhMinute,
seconds= 0 if QhSecond == None else QhSecond,
start_date=QhStartDate,
end_date=QhEndDate,
timezone=QhTimezone,
jitter=QhJitter)
elif QhTimesType == "一次":
QhRunDate = datetime(0 if QhYear == None else QhYear,
0 if QhMonth == None else QhMonth,
0 if QhDay == None else QhDay,
0 if QhHour == None else QhHour,
0 if QhMinute == None else QhMinute,
0 if QhSecond == None else QhSecond)
QhTiggers = DateTrigger(run_date=QhRunDate,
timezone=QhTimezone)
return QhTiggers
def QhPauseJob(self,QhJabId,QhIsFor=False):
# 暫停任務(wù)
# QhIsFor當(dāng)為False時(shí)，保存到csv文件，單個(gè)增加時(shí)的場(chǎng)景
# 闕輝
try:
if QhJabId == "":
print("任務(wù)ID不能空，請(qǐng)輸入任務(wù)ID")
return
if self.Qhscheduler.get_job(QhJabId):
self.Qhscheduler.pause_job(QhJabId)
print("暫停任務(wù)",QhJabId)
self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabStatus'] = '已暫停'
if (not QhIsFor):self.QhCheduPoolJobDf.to_csv(self.QhCheduPoolJobCsv, index=False, encoding='gbk')
# AA = self.Qhscheduler.get_job(QhJabId)
# print(AA.state)
else:
print("暫停失敗,請(qǐng)檢查任務(wù)ID,任務(wù)可能不存在")
except:
print("暫停失敗,請(qǐng)檢查任務(wù)ID,任務(wù)可能不存在")
def QhPauseJobFor(self,QhIsFor=True):
# 暫停所有任務(wù)
for QhInx,QhRow in self.QhCheduPoolJobDf.iterrows():
self.QhPauseJob(QhRow['QhJabId'],QhIsFor)
def QhGetJobsState(self):
for job in self.Qhscheduler.get_jobs():
print(f"Job ID: {job.id}, 任務(wù)的下一次運(yùn)行時(shí)間: {job.next_run_time}")
print(f"Job ID: {job.id}, 任務(wù)是否有待執(zhí)行: {job.pending}")
# print(f"Job ID: {job.id}, 任務(wù)是否有待執(zhí)行: {job.job_state}")
# print(f"Job ID: {job.id}, 任務(wù)是否正在運(yùn)行: {job.running}")
def QhResumeJob(self,QhJabId,QhIsFor=False):
# 恢復(fù)任務(wù)
# QhIsFor當(dāng)為False時(shí)，保存到csv文件，單個(gè)增加時(shí)的場(chǎng)景
# 闕輝
try:
if QhJabId == "":
print("任務(wù)ID不能空，請(qǐng)輸入任務(wù)ID")
return
if self.Qhscheduler.get_job(QhJabId):
self.Qhscheduler.resume_job(QhJabId)
print("恢復(fù)任務(wù)",QhJabId)
self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabStatus'] = '已運(yùn)行'
if (not QhIsFor):self.QhCheduPoolJobDf.to_csv(self.QhCheduPoolJobCsv, index=False, encoding='gbk')
else:
print("恢復(fù)失敗,請(qǐng)檢查任務(wù)ID,任務(wù)可能不存在")
except:
print("恢復(fù)失敗,請(qǐng)檢查任務(wù)ID,任務(wù)可能不存在")
def QhResumeJobFor(self,QhIsFor=True):
# 恢復(fù)所有任務(wù)
# QhIsFor當(dāng)為False時(shí)，保存到csv文件，單個(gè)增加時(shí)的場(chǎng)景
# 闕輝
for QhInx,QhRow in self.QhCheduPoolJobDf.iterrows():
self.QhResumeJob(QhRow['QhJabId'],QhIsFor)
def QhRemoveJob(self,QhJabId,QhIsFor=False):
# 刪除任務(wù)
# QhIsFor當(dāng)為False時(shí)，保存到csv文件，單個(gè)增加時(shí)的場(chǎng)景
# 闕輝
try:
if QhJabId == "":
print("任務(wù)ID不能空，請(qǐng)輸入任務(wù)ID")
return
if self.Qhscheduler.get_job(QhJabId):
self.Qhscheduler.remove_job(QhJabId)
print("刪除任務(wù)",QhJabId)
self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabStatus'] = '已關(guān)閉'
if (not QhIsFor):self.QhCheduPoolJobDf.to_csv(self.QhCheduPoolJobCsv, index=False, encoding='gbk')
else:
print("刪除失敗,請(qǐng)檢查任務(wù)ID,任務(wù)可能不存在")
except:
print("刪除失敗,請(qǐng)檢查任務(wù)ID,任務(wù)可能不存在")
def QhRemoveJobFor(self,QhIsFor=True):
# 刪除所有任務(wù)
# QhIsFor當(dāng)為False時(shí)，保存到csv文件，單個(gè)增加時(shí)的場(chǎng)景
# 闕輝
for QhInx,QhRow in self.QhCheduPoolJobDf.iterrows():
self.QhRemoveJob(QhRow['QhJabId'],QhIsFor)
def QhReopenJob(self,QhJabId,QhIsFor=False):
# 重新打開(kāi)任務(wù)
# QhIsFor當(dāng)為False時(shí)，保存到csv文件，單個(gè)增加時(shí)的場(chǎng)景
# 闕輝
# try:
if QhJabId == "":
print("任務(wù)ID不能空，請(qǐng)輸入任務(wù)ID")
return
QhJabStatus = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabStatus'].values[0])
if not self.Qhscheduler.get_job(QhJabId):
if QhJabStatus == "已關(guān)閉":
# QhJabId = QhRow["QhJabId"]
QhJabName = self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabName'].values[0]
QhJabFuncName = self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabFuncName'].values[0]
QhTimesType = self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhTimesType'].values[0]
QhYear = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhYear'].values[0])
QhMonth = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhMonth'].values[0])
QhDay = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhDay'].values[0])
QhWeek = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhWeek'].values[0])
QhDayOfWeek = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhDayOfWeek'].values[0])
QhHour = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhHour'].values[0])
QhMinute = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhMinute'].values[0])
QhSecond = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhSecond'].values[0])
QhJabArgs = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabArgs'].values[0])
QhJabKwargs = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabKwargs'].values[0])
QhJabStartDate = self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabStartDate'].values[0]
QhJabNextRunTime = self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabNextRunTime'].values[0]
# QhJabLastRunTime = QhRow["QhJabLastRunTime"]
# QhJabEndDate = QhRow["QhJabEndDate"]
# QhJabStatus = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabStatus'].values[0])
print(QhYear,QhMonth,QhDay,QhWeek,QhDayOfWeek,QhHour,QhMinute,QhSecond)
print(type(QhYear),type(QhMonth),type(QhDay),type(QhWeek),type(QhDayOfWeek),type(QhHour),type(QhMinute),type(QhSecond))
self.QhAddJob(QhJabId,
QhTimesType=QhTimesType,
QhJabFuncName = QhJabFuncName,
QhJabArgs=QhJabArgs,
QhJabKwargs=QhJabKwargs,
QhName = QhJabName,
QhYear = QhYear,
QhMonth = QhMonth,
QhDay = QhDay,
QhWeek = QhWeek,
QhDayOfWeek = QhDayOfWeek,
QhHour = QhHour,
QhMinute = QhMinute,
QhSecond = QhSecond,
QhJabStatus=QhJabStatus,
QhIsFor = QhIsFor,
QhJobIsOpen = True) # 狀態(tài)已在此函數(shù)更新
# except:
# print("打開(kāi)任務(wù)失敗,請(qǐng)檢查任務(wù)ID,任務(wù)可能不存在")
def QhReopenJobFor(self,QhIsFor=True):
# 重新打開(kāi)所有任務(wù)
# QhIsFor當(dāng)為False時(shí)，保存到csv文件，單個(gè)增加時(shí)的場(chǎng)景
# 闕輝
for QhInx,QhRow in self.QhCheduPoolJobDf.iterrows():
self.QhReopenJob(QhRow['QhJabId'],QhIsFor)
def QhShouDongRunJob(self,QhJabId):
# 手動(dòng)運(yùn)行任務(wù)
# 闕輝
try:
if QhJabId == "":
print("任務(wù)ID不能空，請(qǐng)輸入任務(wù)ID")
return
# if self.Qhscheduler.get_job(QhJabId):
QhJabFuncName = self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabFuncName'].values[0]
QhJabArgs = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabArgs'].values[0])
QhJabKwargs = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabKwargs'].values[0])
QhJabFunc = globals().get(QhJabFuncName)
if QhJabFunc:
if QhJabArgs!= None:
QhJabArgs = tuple(QhJabArgs.split("+"))
QhJabFunc(*QhJabArgs)
elif QhJabKwargs != None:
QhJabKwargs = QhJabKwargs.replace("+",',')
QhJabKwargs = QhJabKwargs.replace("'",'"')
QhJabKwargs = json.loads(QhJabKwargs)
QhJabFunc(**QhJabKwargs)
else:
print(f"Function '{QhJabFuncName}' not found.")
# self.Qhscheduler.run_job(QhJabId)
print("手動(dòng)執(zhí)行任務(wù)",QhJabId)
except Exception as e:
print(f"An error occurred while executing the job: {e}")
print("2手動(dòng)執(zhí)行任務(wù)失敗,請(qǐng)檢查任務(wù)ID,任務(wù)可能不存在")
def QhShouDongRunJobFor(self):
# 手動(dòng)運(yùn)行所有任務(wù)
# 闕輝
for QhInx,QhRow in self.QhCheduPoolJobDf.iterrows():
self.QhShouDongRunJob(QhRow['QhJabId'])
def QhXiuGaiJob(self,QhJabId,
QhIsFor=True):
# 修改任務(wù)方案：先刪除，修改后重新添加
pass
# self.QhCheduPoolJobDf = self._QhinitJobPoolCsv() # 重新加載任務(wù)池?cái)?shù)據(jù)
# QhJabName = self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabName'].values[0]
# QhJabFuncName = self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabFuncName'].values[0]
# QhTimesType = self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhTimesType'].values[0]
# QhYear = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhYear'].values[0])
# QhMonth = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhMonth'].values[0])
# QhDay = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhDay'].values[0])
# QhWeek = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhWeek'].values[0])
# QhDayOfWeek = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhDayOfWeek'].values[0])
# QhHour = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhHour'].values[0])
# QhMinute = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhMinute'].values[0])
# QhSecond = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhSecond'].values[0])
# QhJabArgs = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabArgs'].values[0])
# QhJabKwargs = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabKwargs'].values[0])
# QhJabStartDate = self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabStartDate'].values[0]
# QhJabNextRunTime = self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabNextRunTime'].values[0]
# # QhJabLastRunTime = QhRow["QhJabLastRunTime"]
# # QhJabEndDate = QhRow["QhJabEndDate"]
# # QhJabStatus = self.QhGeShiZH(self.QhCheduPoolJobDf.loc[self.QhCheduPoolJobDf['QhJabId'] == QhJabId, 'QhJabStatus'].values[0])
# print(QhYear,QhMonth,QhDay,QhWeek,QhDayOfWeek,QhHour,QhMinute,QhSecond)
# print(type(QhYear),type(QhMonth),type(QhDay),type(QhWeek),type(QhDayOfWeek),type(QhHour),type(QhMinute),type(QhSecond))
# self.Qhscheduler.modify_job()
def QhGeShiZH(self,QhValue):
# 時(shí)間格式轉(zhuǎn)化 后期需要優(yōu)化
# 作者：闕輝
try:
if isinstance(QhValue, str):
if QhValue in ["None","","0"]:
# print("zhuan01")
return None
else:
try:
# print("zhuan02")
return int(QhValue)
except:
# print("zhuan03")
return QhValue
else:
# print("zhuan04")
return int(QhValue)
except:
# print("zhuan05")
return None
class QhTestApschedulerGui(QtWidgets.QWidget,Ui_Form):
def __init__(self, parent=None):
super(QhTestApschedulerGui, self).__init__(parent)
self.setupUi(self)
self.aa=QhApscheduler()
self.pushButton_2.clicked.connect(self.QhPauseJob)
self.pushButton_18.clicked.connect(self.QhResumeJob)
self.pushButton_19.clicked.connect(self.aa.QhGetJobsState)
self.pushButton.clicked.connect(self.QhAddJob)
self.pushButton_3.clicked.connect(self.QhRemoveJob)
self.pushButton_20.clicked.connect(self.aa.QhPauseJobFor)
self.pushButton_21.clicked.connect(self.aa.QhResumeJobFor)
self.pushButton_22.clicked.connect(self.aa.QhRemoveJobFor)
self.pushButton_23.clicked.connect(self.aa.QhReopenJobFor)
self.pushButton_6.clicked.connect(self.QhReopenJob)
self.pushButton_4.clicked.connect(self.QhShouDongRunJob)
self.pushButton_24.clicked.connect(self.aa.QhShouDongRunJobFor)
def QhAddJob(self):
QhJobId = self.lineEdit_5.text()
if QhJobId == "":
print("任務(wù)ID不能空，請(qǐng)輸入任務(wù)ID")
return
QhJobName = "{}-Name".format(QhJobId)
QhScoends = random.randint(8, 20)
QhTimesType = "間隔"
QhJabFuncName = "QhRR"
self.aa.QhAddJob(QhJobId,
QhTimesType=QhTimesType,
QhJabFuncName = QhJabFuncName,
QhName = QhJobName,
QhSecond = QhScoends,
QhJabArgs = "{QhJobId}-測(cè)試順序+QUEHUI".format(QhJobId=QhJobId),
)
print("添加任務(wù){(diào)}".format(QhJobId))
def QhPauseJob(self):
QhJobId = self.lineEdit.text()
print(QhJobId)
self.aa.QhPauseJob(QhJobId)
def QhResumeJob(self):
QhJobId = self.lineEdit_18.text()
print(QhJobId)
self.aa.QhResumeJob(QhJobId)
def QhRemoveJob(self):
QhJobId = self.lineEdit_2.text()
print(QhJobId)
self.aa.QhRemoveJob(QhJobId)
def QhReopenJob(self):
QhJobId = self.lineEdit_6.text()
print(QhJobId)
self.aa.QhReopenJob(QhJobId)
def QhShouDongRunJob(self):
QhJobId = self.lineEdit_3.text()
print(QhJobId)
self.aa.QhShouDongRunJob(QhJobId)
if __name__ == '__main__':
# aa=QhApscheduler()
# aa._QhInitAddJobFor()
# # 保持主線程運(yùn)行
# try:
# while True:
# time.sleep(2)
# except (KeyboardInterrupt, SystemExit):
# aa.Qhscheduler.shutdown()
QApplication.setHighDpiScaleFactorRoundingPolicy(
Qt.HighDpiScaleFactorRoundingPolicy.PassThrough)
QApplication.setAttribute(Qt.AA_EnableHighDpiScaling)
QApplication.setAttribute(Qt.AA_UseHighDpiPixmaps)
qh_app = QApplication(sys.argv) # 創(chuàng)建應(yīng)用實(shí)例
qh_MyWindows = QhTestApschedulerGui() # 創(chuàng)建窗口實(shí)例
qh_MyWindows.show() # 顯示窗口
qh_n = qh_app.exec() # 執(zhí)行exec()方法，進(jìn)入事件循環(huán)，如果遇到窗口退出命令，返回整數(shù)qh_n
print(qh_n) # 輸出輸出關(guān)閉事件返回的整數(shù)
try: # 捕獲程序退出事件
sys.exit(qh_n) # 通知python系統(tǒng)，結(jié)束程序運(yùn)行
except SystemExit:
print("請(qǐng)?jiān)诖俗鲆恍┢渌ぷ鳌?) # python解釋器停止執(zhí)行前的工作UI代碼
# -*- coding: utf-8 -*-
# Form implementation generated from reading ui file 'QhTestJob.ui'
#
# Created by: PyQt5 UI code generator 5.15.7
#
# WARNING: Any manual changes made to this file will be lost when pyuic5 is
# run again. Do not edit this file unless you know what you are doing.
from PyQt5 import QtCore, QtGui, QtWidgets
class Ui_Form(object):
def setupUi(self, Form):
Form.setObjectName("Form")
Form.resize(632, 610)
self.gridLayout = QtWidgets.QGridLayout(Form)
self.gridLayout.setObjectName("gridLayout")
self.verticalLayout = QtWidgets.QVBoxLayout()
self.verticalLayout.setObjectName("verticalLayout")
self.horizontalLayout_5 = QtWidgets.QHBoxLayout()
self.horizontalLayout_5.setObjectName("horizontalLayout_5")
self.pushButton = QtWidgets.QPushButton(Form)
self.pushButton.setObjectName("pushButton")
self.horizontalLayout_5.addWidget(self.pushButton)
self.lineEdit_5 = QtWidgets.QLineEdit(Form)
self.lineEdit_5.setMinimumSize(QtCore.QSize(500, 28))
self.lineEdit_5.setMaximumSize(QtCore.QSize(500, 16777215))
self.lineEdit_5.setObjectName("lineEdit_5")
self.horizontalLayout_5.addWidget(self.lineEdit_5)
self.verticalLayout.addLayout(self.horizontalLayout_5)
self.horizontalLayout_4 = QtWidgets.QHBoxLayout()
self.horizontalLayout_4.setObjectName("horizontalLayout_4")
self.pushButton_2 = QtWidgets.QPushButton(Form)
self.pushButton_2.setObjectName("pushButton_2")
self.horizontalLayout_4.addWidget(self.pushButton_2)
self.lineEdit = QtWidgets.QLineEdit(Form)
self.lineEdit.setMinimumSize(QtCore.QSize(230, 28))
self.lineEdit.setMaximumSize(QtCore.QSize(500, 16777215))
self.lineEdit.setObjectName("lineEdit")
self.horizontalLayout_4.addWidget(self.lineEdit)
self.verticalLayout.addLayout(self.horizontalLayout_4)
self.horizontalLayout_18 = QtWidgets.QHBoxLayout()
self.horizontalLayout_18.setObjectName("horizontalLayout_18")
self.pushButton_18 = QtWidgets.QPushButton(Form)
self.pushButton_18.setObjectName("pushButton_18")
self.horizontalLayout_18.addWidget(self.pushButton_18)
self.lineEdit_18 = QtWidgets.QLineEdit(Form)
self.lineEdit_18.setMinimumSize(QtCore.QSize(230, 28))
self.lineEdit_18.setMaximumSize(QtCore.QSize(500, 16777215))
self.lineEdit_18.setObjectName("lineEdit_18")
self.horizontalLayout_18.addWidget(self.lineEdit_18)
self.verticalLayout.addLayout(self.horizontalLayout_18)
self.horizontalLayout_3 = QtWidgets.QHBoxLayout()
self.horizontalLayout_3.setObjectName("horizontalLayout_3")
self.pushButton_3 = QtWidgets.QPushButton(Form)
self.pushButton_3.setObjectName("pushButton_3")
self.horizontalLayout_3.addWidget(self.pushButton_3)
self.lineEdit_2 = QtWidgets.QLineEdit(Form)
self.lineEdit_2.setMinimumSize(QtCore.QSize(230, 28))
self.lineEdit_2.setMaximumSize(QtCore.QSize(500, 16777215))
self.lineEdit_2.setObjectName("lineEdit_2")
self.horizontalLayout_3.addWidget(self.lineEdit_2)
self.verticalLayout.addLayout(self.horizontalLayout_3)
self.horizontalLayout_6 = QtWidgets.QHBoxLayout()
self.horizontalLayout_6.setObjectName("horizontalLayout_6")
self.pushButton_6 = QtWidgets.QPushButton(Form)
self.pushButton_6.setObjectName("pushButton_6")
self.horizontalLayout_6.addWidget(self.pushButton_6)
self.lineEdit_6 = QtWidgets.QLineEdit(Form)
self.lineEdit_6.setMinimumSize(QtCore.QSize(230, 28))
self.lineEdit_6.setMaximumSize(QtCore.QSize(500, 16777215))
self.lineEdit_6.setObjectName("lineEdit_6")
self.horizontalLayout_6.addWidget(self.lineEdit_6)
self.verticalLayout.addLayout(self.horizontalLayout_6)
self.horizontalLayout_2 = QtWidgets.QHBoxLayout()
self.horizontalLayout_2.setObjectName("horizontalLayout_2")
self.pushButton_4 = QtWidgets.QPushButton(Form)
self.pushButton_4.setObjectName("pushButton_4")
self.horizontalLayout_2.addWidget(self.pushButton_4)
self.lineEdit_3 = QtWidgets.QLineEdit(Form)
self.lineEdit_3.setMinimumSize(QtCore.QSize(230, 28))
self.lineEdit_3.setMaximumSize(QtCore.QSize(500, 16777215))
self.lineEdit_3.setObjectName("lineEdit_3")
self.horizontalLayout_2.addWidget(self.lineEdit_3)
self.verticalLayout.addLayout(self.horizontalLayout_2)
self.horizontalLayout = QtWidgets.QHBoxLayout()
self.horizontalLayout.setObjectName("horizontalLayout")
self.pushButton_5 = QtWidgets.QPushButton(Form)
self.pushButton_5.setObjectName("pushButton_5")
self.horizontalLayout.addWidget(self.pushButton_5)
self.lineEdit_4 = QtWidgets.QLineEdit(Form)
self.lineEdit_4.setMinimumSize(QtCore.QSize(230, 28))
self.lineEdit_4.setMaximumSize(QtCore.QSize(500, 16777215))
self.lineEdit_4.setObjectName("lineEdit_4")
self.horizontalLayout.addWidget(self.lineEdit_4)
self.verticalLayout.addLayout(self.horizontalLayout)
self.pushButton_19 = QtWidgets.QPushButton(Form)
self.pushButton_19.setObjectName("pushButton_19")
self.verticalLayout.addWidget(self.pushButton_19)
self.pushButton_24 = QtWidgets.QPushButton(Form)
self.pushButton_24.setObjectName("pushButton_24")
self.verticalLayout.addWidget(self.pushButton_24)
self.pushButton_20 = QtWidgets.QPushButton(Form)
self.pushButton_20.setObjectName("pushButton_20")
self.verticalLayout.addWidget(self.pushButton_20)
self.pushButton_21 = QtWidgets.QPushButton(Form)
self.pushButton_21.setObjectName("pushButton_21")
self.verticalLayout.addWidget(self.pushButton_21)
self.pushButton_22 = QtWidgets.QPushButton(Form)
self.pushButton_22.setObjectName("pushButton_22")
self.verticalLayout.addWidget(self.pushButton_22)
self.pushButton_23 = QtWidgets.QPushButton(Form)
self.pushButton_23.setObjectName("pushButton_23")
self.verticalLayout.addWidget(self.pushButton_23)
self.gridLayout.addLayout(self.verticalLayout, 0, 0, 1, 1)
self.retranslateUi(Form)
QtCore.QMetaObject.connectSlotsByName(Form)
def retranslateUi(self, Form):
_translate = QtCore.QCoreApplication.translate
Form.setWindowTitle(_translate("Form", "Form"))
self.pushButton.setText(_translate("Form", "添加任務(wù)"))
self.pushButton_2.setText(_translate("Form", "暫停任務(wù)"))
self.pushButton_18.setText(_translate("Form", "啟動(dòng)任務(wù)"))
self.pushButton_3.setText(_translate("Form", "刪除任務(wù)"))
self.pushButton_6.setText(_translate("Form", "打開(kāi)任務(wù)"))
self.pushButton_4.setText(_translate("Form", "手動(dòng)執(zhí)行"))
self.pushButton_5.setText(_translate("Form", "獲取任務(wù)狀態(tài)"))
self.pushButton_19.setText(_translate("Form", "獲取所有任務(wù)的狀態(tài)"))
self.pushButton_24.setText(_translate("Form", "手動(dòng)執(zhí)行所有任務(wù)"))
self.pushButton_20.setText(_translate("Form", "暫停所有任務(wù)"))
self.pushButton_21.setText(_translate("Form", "恢復(fù)所有任務(wù)"))
self.pushButton_22.setText(_translate("Form", "關(guān)閉所有任務(wù)"))
self.pushButton_23.setText(_translate("Form", "打開(kāi)所有任務(wù)"))到此這篇關(guān)于基于Python進(jìn)行定時(shí)任務(wù)管理封裝的文章就介紹到這了,更多相關(guān)Python定時(shí)任務(wù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
這篇文章主要介紹了Python讀取表格類型文件代碼實(shí)例,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-02-02
這篇文章主要介紹了二十種Python代碼游戲源代碼分享,本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2021-04-04
python遞歸調(diào)用中的坑:打印有值, 返回卻None
這篇文章主要介紹了python遞歸調(diào)用中的坑:打印有值, 返回卻None,本文通過(guò)問(wèn)題分析給出解決方法，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2020-03-03
在性能和資源占用方面,venv?通常更適合輕量級(jí)項(xiàng)目、純Python應(yīng)用以及對(duì)資源敏感的場(chǎng)景,其簡(jiǎn)單快速的特點(diǎn)使得環(huán)境創(chuàng)建和管理更為高效,這篇文章主要介紹了Python虛擬環(huán)境conda的安裝使用教程,需要的朋友可以參考下2024-03-03
python pygame實(shí)現(xiàn)2048游戲
這篇文章主要為大家詳細(xì)介紹了python pygame實(shí)現(xiàn)2048游戲，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2018-11-11
這篇文章主要介紹了django解決訂單并發(fā)問(wèn)題，本文給大家介紹的非常詳細(xì)，具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2019-07-07
淺談django url請(qǐng)求與數(shù)據(jù)庫(kù)連接池的共享問(wèn)題
今天小編就為大家分享一篇淺談django url請(qǐng)求與數(shù)據(jù)庫(kù)連接池的共享問(wèn)題，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2019-08-08
Django 如何使用日期時(shí)間選擇器規(guī)范用戶的時(shí)間輸入示例代碼詳解
這篇文章主要介紹了 Django 如何使用日期時(shí)間選擇器規(guī)范用戶的時(shí)間輸入，本文通過(guò)實(shí)例代碼給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2020-05-05
這篇文章主要介紹了Python中的整除和取模實(shí)例，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2020-06-06
pytorch 如何自定義卷積核權(quán)值參數(shù)
這篇文章主要介紹了pytorch 自定義卷積核權(quán)值參數(shù)的操作，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2021-05-05