dependencies:

-python=3.6

-numpy

-scipy

-pillow

-tqdm

-requests

/Volumes/GoogleDrive/My Drive/root/work/github_private/webpage-monitor/gmail_app_pswd

fromscipy.ndimageimportgaussian_filter

parser=argparse.ArgumentParser(description="Webpage monitor.")

parser=argparse.ArgumentParser(description='Webpage monitor.')

parser.add_argument('--roster_json',

type=str,

default='./roster.json',

help="path to the roster")

help='path to the roster')

parser.add_argument('--check_every',

type=int,

default=43200,

help="check every N seconds")

help='check every N seconds')

parser.add_argument('--exit_after',

type=int,

default=None,

help="quit after N seconds")

parser.add_argument('--tmp_dir',

help='quit after N seconds')

parser.add_argument('--snapshot_dir',

type=str,

default='/tmp/webpage-monitor',

help="directory to dump screenshots for comparison")

default='./snapshots',

help='directory to dump screenshots for comparison')

parser.add_argument('--clear_cached',

action='store_true',

help="whether to clear the screenshots on disk")

help='whether to clear the screenshots on disk')

defmain(args):

ifargs.exit_afterisNone:

exit_after=np.inf

else:

exit_after=args.exit_after

exit_after=np.infifargs.exit_afterisNoneelseargs.exit_after

roster=load_roster(args.roster_json)

withopen(args.roster_json,'rb')asfile_handle:

roster=json.load(file_handle)

start_t=time()

ifargs.clear_cachedandexists(args.tmp_dir):

rmtree(args.tmp_dir)

ifargs.clear_cachedandexists(args.snapshot_dir):

rmtree(args.snapshot_dir)

whileTrue:

iftime()-last_check_t>args.check_every:

changed,deltas= [], []

forurl,optintqdm(roster.items(),desc="Checking URLs"):

out_dir=join(args.tmp_dir,

replace_special_char(url)).rstrip('/')

screenshot(url,out_dir,opt)

pngs=sorted(glob(join(out_dir,'*.png')))

iflen(pngs)>1:

delta=diff_screenshots(*pngs[-2:],delta_png)

ifdeltaisnotNone:

forurl,optintqdm(roster.items(),desc='Checking URLs'):

out_dir=join(args.snapshot_dir,

util.folder_name_from_url(url))

snapshot(url,out_dir,opt)

snapshot_paths=sorted(

glob(join(out_dir,'????_??_??_??_??_??.html')))

iflen(snapshot_paths)>1:

delta=diff_snapshots(snapshot_paths[-2],

snapshot_paths[-1],out_dir,opt)

ifdelta!='':

changed.append(url)

deltas.append(delta)

iflen(pngs)>2:

forfinpngs[:-2]:

remove(f)

iflen(snapshot_paths)>2:

forxinsnapshot_paths[:-2]:

remove(x)

last_check_t=time()

ifchanged:

forurl,deltainzip(changed,deltas):

msg+="file://{delta}\n{url}\n\n".format(delta=delta,

url=url)

util.email_myself(msg,subject="Webpage Monitor")

util.format_print("Change detected; email sent",'header')

iftime()-start_t>exit_after:

defdiff_screenshots(old_png,

new_png,

delta_png,

pix_diff_thres=0.1,

n_diff_thres=16,

unchanged_alpha=0.2,

diff_blur_sigma=4):

old=util.imread_arr(old_png)

new=util.imread_arr(new_png)

ifold.shape!=new.shape:

util.imwrite_arr(new,delta_png)

pixel_is_diff=np.abs(old-new)>=pix_diff_thres# (H, W, 3)

pixel_is_diff=np.sum(pixel_is_diff,axis=2)>0

ifnp.sum(pixel_is_diff)<=n_diff_thres:

alpha=unchanged_alpha*np.ones_like(new)

alpha[np.dstack([pixel_is_diff]*3)]=1

alpha=gaussian_filter(alpha,diff_blur_sigma)

delta=alpha*new+ (1-alpha)*np.zeros_like(new)

util.imwrite_arr(delta,delta_png)

defscreenshot(url,out_dir,opt,width=512,delay=3):

util.email_myself(msg,subject='Webpage Monitor')

util.format_print('Change detected; email sent','header')

iftime()-start_t>exit_after:

defdiff_snapshots(html0_path,html1_path,out_dir,opt):

html0_content=util.read_file(html0_path)

html1_content=util.read_file(html1_path)

delta=difflib.ndiff(html0_content.split('\n'),html1_content.split('\n'))

delta='\n'.join(xforxindelta

ifx.startswith('+ ')orx.startswith('- '))

delta_path=join(out_dir,'delta.html')

util.write_file(delta,delta_path)

defsnapshot(url,out_dir,opt):

request=requests.get(url)

print(url)

html_src=request.content.decode()

ifnotexists(out_dir):

makedirs(out_dir)

cmd= ('webkit2png --fullsize --no-images --ignore-ssl-check --width={w} '

'--delay={delay} --dir={dir_} --filename={t} {url}').format(

w=width,delay=delay,dir_=out_dir,t=time(),url=url)

util.call(cmd,silence_stdout=True)

defload_roster(roster_json):

withopen(roster_json,'r')ash:

roster=json.load(h)

defreplace_special_char(url):

returnurl.replace('/','_').replace('?',

'_').replace('&',

'_').replace(':','_')

timestamp=datetime.datetime.now().strftime('%Y_%m_%d_%H_%M_%S')

html_path=join(out_dir,timestamp+'.html')

util.write_file(html_src,html_path)

if__name__=='__main__':

fromsubprocessimportPopen,PIPE

fromemail.mime.textimportMIMEText

defimread_arr(path):

im=Image.open(path)

arr=np.array(im)[:, :, :3]

arr=arr.astype(float)/np.iinfo(arr.dtype).max

defread_file(path):

withopen(path,'rb')asfile_handle:

content=file_handle.read()

returncontent.decode()

defimwrite_arr(arr,path,dtype='uint8'):

arr= (arr*np.iinfo(dtype).max).astype(dtype)

im=Image.fromarray(arr)

im.save(path)

defwrite_file(str_,path):

withopen(path,'wb')asfile_handle:

file_handle.write(str_.encode())

defcall(cmd,cwd=None,silence_stdout=False):

process=Popen(cmd,stdout=PIPE,stderr=PIPE,cwd=cwd,shell=True)

deffolder_name_from_url(url):

folder_name=url.rstrip('/')

folder_name=folder_name.replace('http://','').replace('https://','')

folder_name=folder_name.replace('/','_')

folder_name=folder_name.replace('?','_')

folder_name=folder_name.replace('&','_')

folder_name=folder_name.replace(':','_')

stdout,stderr=process.communicate()# waits for completion

stdout,stderr=stdout.decode(),stderr.decode()

ifnotsilence_stdout:

ifstdout!='':

format_print(stdout,'okblue')

ifstderr!='':

format_print(cmd,'warn')

format_print(stderr,'fail')

defemail_myself(

msg,subject="Untitled",email='xiuming6zhang@gmail.com',

pswd_path='./gmail_app_pswd'):

withopen(pswd_path,'r')ash:

pswd=h.readlines()

pswd=pswd[0].strip()

defemail_myself(msg,

subject='Untitled',

email='xiuming6zhang@gmail.com',

pswd_path='./gmail_app_pswd'):

pswd=read_file(pswd_path)

pswd=pswd.strip()

to_emails= [email]

defformat_print(msg,fmt):

fmt_strs= {

'header':'\033[95m',

'warn':'\033[93m',

'fail':'\033[91m',

'bold':'\033[1m',

'underline':'\033[4m'}

'underline':'\033[4m'

}

iffmtinfmt_strs.keys():

start_str=fmt_strs[fmt]

eliflen(fmt)==1:

start_str="\n<"+"".join([fmt]*78)+'\n\n'# as per PEP8

end_str='\n'+start_str[2:-2]+">\n"

start_str='\n<'+''.join([fmt]*78)+'\n\n'# as per PEP8

end_str='\n'+start_str[2:-2]+'>\n'

else:

raiseValueError((

"but input is'%s'")% (list(fmt_strs.keys()),fmt))

raiseValueError(

'but input is %s')% (list(fmt_strs.keys()),fmt))

print(start_str+msg+end_str)