最近工作中有这样一个场景:
某个文件夹(例如D:\Downloads)每间隔一段时间, 就应该收到一些新文件. 超出一定时间, 如果还没有新文件传过来, 一定是哪儿出问题了, 必须尽早发现, 尽早处理.
当然, 我不可能时刻盯着屏幕, 必须交给计算机自动监测, 超时自动给出警报. 代码如下, 写的比较长, 主要原因是用了tk做了个简单的界面, 这又牵扯到进程调度的问题. 我这种面向百度的业余程序员, 花了整整一晚上才写出来. 不过想到以后上班又可以快乐的摸鱼了, 还是很开心的.
from datetime import datetime,timedelta
import time
import os
import winsound
from apscheduler.schedulers.background import BackgroundScheduler
import tkinter as tk
import threading
# 功能描述, 监测路径path(可以是多个路径组成的列表)下是否有新文件, 如果超一个小时没有新文件,则报警.
# 利用python自带的tk, 写出简单的界面展示
def check(paths,threshhold,text):
# paths 存储所有要监控的路径
# threshhold 超出threshhold分钟,则报警
scan_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
minute = int(datetime.now().strftime('%M'))
if minute < threshhold:
minute = minute + 60
text.insert(tk.END,'--------------'+ scan_time + ' 开始检测----------------'+'\n' )
text.see(tk.END)
text.update()
well = True
for filePath in paths:
lists = os.listdir(filePath)
if 'xxx' in filePath:#带有xxx的文件,看创建日期
lists.sort(key=lambda fn:os.path.getctime(filePath + "\\"+fn))
new_file = os.path.join(filePath,lists[-1])
stamp = os.path.getctime(new_file)
else:#其它文件,看修改日期
lists.sort(key=lambda fn: os.path.getmtime(filePath + "\\" + fn))
new_file = os.path.join(filePath, lists[-1])
stamp = os.path.getmtime(new_file)
now = time.time()
# 间隔分钟
delta = int((now - stamp)/60)
info = filePath + ': ' + datetime.fromtimestamp(stamp).strftime('%Y-%m-%d %H:%M:%S')
text.insert(tk.END, info + '\n')
text.see(tk.END)
text.update()
if delta> minute:
text.insert(tk.END,filePath + ' %d分钟 没有收到文件!!!'%delta + '\n')
# 修改背景颜色, 这样报警更显著
text['bg'] = '#FF7F50'
text.see(tk.END)
text.update()
# 同时窗口置顶, 报警更显著
window.wm_attributes('-topmost',1)
# 开启一个线程,用来播放警报音
threading._start_new_thread(alarm,('alarm.wav',))
well = False
if(well):
# 恢复正常态
text['bg'] = 'white'
text.insert(tk.END,'\n')
text.update()
else:
text['bg'] = '#FF7F50'
text.insert(tk.END,'\n')
text.update()
# 报警:播放wavfile
def alarm(wavfile):
winsound.PlaySound(wavfile, winsound.SND_FILENAME)
window = tk.Tk()
window.title('文件监控')
window.geometry('460x265')
text = tk.Text(window,width=800,height=300)
text.pack(fill=tk.X,side=tk.BOTTOM)
paths = ['D:\\Downloads',]
threshhold = 30
# 利用scheduler调度检测进程
#
# 1. get current minute
# 2. if minute >= threshhold, check files every hour
# else minute < threshhold, compare to previous hour and then after (threshhold - minute) do 1
#
scheduler = BackgroundScheduler()
minute = int(datetime.now().strftime('%M'))
if minute < threshhold:
check(paths,threshhold,text)
next_run_time = datetime.now() + timedelta(seconds=60*(threshhold - minute))
scheduler.add_job(check,next_run_time=next_run_time,args=(paths,threshhold,text))
else:
check(paths,threshhold,text)
scheduler.add_job(check,'interval',minutes=30,args=(paths,threshhold,text))
scheduler.start()
# 开启UI
window.mainloop()
运行结果如下:
发现异常则窗口变色置顶,同时播放警示音
