python做文件监控

最近工作中有这样一个场景:

某个文件夹(例如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()

运行结果如下:

发现异常则窗口变色置顶,同时播放警示音