Created
September 24, 2013 08:44
-
-
Save etataurov/6682028 to your computer and use it in GitHub Desktop.
Проверятор состояния посылки, отправленной ShopFans Express.
Каждые N минут ходит по указанной ссылке, в случае нового события шлет письмо
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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() |
Планирую сделать в виде сервиса по подписке на обновления, если у шопфанс еще не появились собственные уведомления
В виде сервиса тут: 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
tornado здесь на самом деле не нужен, просто хотелось его запихнуть 😄