Skip to content

Instantly share code, notes, and snippets.

@etataurov
Created September 24, 2013 08:44
Show Gist options
  • Save etataurov/6682028 to your computer and use it in GitHub Desktop.
Save etataurov/6682028 to your computer and use it in GitHub Desktop.
Проверятор состояния посылки, отправленной ShopFans Express. Каждые N минут ходит по указанной ссылке, в случае нового события шлет письмо
#!/usr/bin/env python3
from datetime import timedelta
import time
import json
import logging
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import tornado.ioloop
import requests
from bs4 import BeautifulSoup
log = logging.getLogger(__name__)
logging.basicConfig(format=u'%(asctime)s %(levelname)s %(message)s', level=logging.DEBUG)
TIMEOUT = 60*10 # once in 10 minutes
LOGIN = ''
PASSWORD = ''
FROM_EMAIL = ""
TO_EMAIL = ""
SMTP_SERVER = ''
URL = ''
FILENAME = 'changes.json'
def set_timer(func):
ioloop = tornado.ioloop.IOLoop.current()
ioloop.add_timeout(timedelta(seconds=TIMEOUT), func)
class Checker:
def __init__(self):
self.changes = {}
def make_message(self, changes):
message = []
for date in sorted(self.changes.keys(), key=lambda x: time.strptime(x, '%Y-%m-%d %H:%M:%S')):
message.append("<p>{}: {}</p>".format(date, self.changes[date]))
for new_date, new_event in changes:
message.append("<p><b>{}</b>: {}</p>".format(new_date, new_event))
return '\n'.join(message)
def send_new_changes(self, changes):
msg = MIMEMultipart('alternative')
msg['Subject'] = 'New parcel event!'
msg['From'] = FROM_EMAIL
msg['To'] = TO_EMAIL
text = MIMEText(self.make_message(changes), 'html')
msg.attach(text)
s = smtplib.SMTP_SSL(SMTP_SERVER)
s.login(LOGIN, PASSWORD)
try:
log.debug('sending new mail to {}'.format(TO_EMAIL))
s.sendmail(FROM_EMAIL, [TO_EMAIL], msg.as_string())
finally:
s.quit()
def parse_content(self, content):
parsed = BeautifulSoup(content)
new_changes = []
for tr in parsed.table.find_all('tr')[:0:-1]: # newer events are higher
date, event = [x.text for x in tr.find_all('td')]
if date not in self.changes:
new_changes.append((date, event))
log.info('new event [{}]: {}'.format(date, event))
if new_changes:
self.send_new_changes(new_changes)
self.save_changes(new_changes)
self.save_changes_to_file()
else:
log.info('no new events')
def check_status(self):
log.debug('checking parcel status')
response = requests.get(URL)
self.parse_content(response.content)
set_timer(self.check_status)
def start(self):
self.load_changes_from_file()
self.check_status()
def save_changes(self, changes):
for date, event in changes:
self.changes[date] = event
def save_changes_to_file(self):
log.debug('saving to file')
with open(FILENAME, 'w') as f:
json.dump(self.changes, f)
def load_changes_from_file(self):
log.debug('loading changes from file')
try:
with open(FILENAME, 'r') as f:
self.changes = json.load(f)
except FileNotFoundError:
log.debug('file not found, so there are no parcel changes yet')
self.changes = {}
def main():
checker = Checker()
checker.start()
tornado.ioloop.IOLoop.current().start()
if __name__ == '__main__':
main()
@etataurov
Copy link
Author

В виде сервиса тут: https://github.com/etataurov/SFX-tracker
работает на http://sfxtracker.herokuapp.com/

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment