diffido/diffido.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import re
import io
import json
import shutil
import urllib
import smtplib
from email.mime.text import MIMEText
import logging
import datetime
import requests
import subprocess
import multiprocessing
from lxml import etree
from xml.etree import ElementTree

from tornado.ioloop import IOLoop
from apscheduler.triggers.cron import CronTrigger
from apscheduler.schedulers.tornado import TornadoScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

import tornado.httpserver
import tornado.ioloop
import tornado.options
from tornado.options import define, options
import tornado.web
from tornado import gen, escape

JOBS_STORE = 'sqlite:///conf/jobs.db'
API_VERSION = '1.0'
SCHEDULES_FILE = 'conf/schedules.json'
DEFAULT_CONF = 'conf/diffido.conf'
EMAIL_FROM = 'diffido@localhost'
GIT_CMD = 'git'

re_commit = re.compile(r'^(?P<id>[0-9a-f]{40}) (?P<message>.*)\n(?: .* '
                       '(?P<insertions>\d+) insertion.* (?P<deletions>\d+) deletion.*$)?', re.M)
re_insertion = re.compile(r'(\d+) insertion')
re_deletion = re.compile(r'(\d+) deletion')

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def read_schedules():
    if not os.path.isfile(SCHEDULES_FILE):
        return {'schedules': {}}
    try:
        with open(SCHEDULES_FILE, 'r') as fd:
            return json.loads(fd.read())
    except Exception as e:
        logger.error('unable to read %s: %s' % (SCHEDULES_FILE, e))
        return {'schedules': {}}


def write_schedules(schedules):
    with open(SCHEDULES_FILE, 'w') as fd:
        fd.write(json.dumps(schedules, indent=2))


def next_id(schedules):
    ids = schedules.get('schedules', {}).keys()
    if not ids:
        return '1'
    return str(max([int(i) for i in ids]) + 1)


def get_schedule(id_, add_id=True):
    schedules = read_schedules()
    data = schedules.get('schedules', {}).get(id_, {})
    if add_id:
        data['id'] = str(id_)
    return data


def select_xpath(content, xpath):
    fd = io.StringIO(content)
    tree = etree.parse(fd)
    elems = tree.xpath(xpath)
    if not elems:
        return content
    selected_content = []
    for elem in elems:
        selected_content.append(''.join([elem.text] + [ElementTree.tostring(e).decode('utf-8', 'replace')
                                                    for e in elem.getchildren()]))
    content = ''.join(selected_content)
    return content


def run_job(id_=None, *args, **kwargs):
    schedule = get_schedule(id_, add_id=False)
    url = schedule.get('url')
    if not url:
        return
    logger.debug('Running job id:%s title:%s url: %s' % (id_, schedule.get('title', ''), url))
    req = requests.get(url, allow_redirects=True, timeout=(30.10, 240))
    content = req.text
    xpath = schedule.get('xpath')
    if xpath:
        content = select_xpath(content, xpath)
    req_path = urllib.parse.urlparse(req.url).path
    base_name = os.path.basename(req_path) or 'index.html'
    def _commit(id_, filename, content, queue):
        os.chdir('storage/%s' % id_)
        current_lines = 0
        if os.path.isfile(filename):
            with open(filename, 'r') as fd:
                for line in fd:
                    current_lines += 1
        with open(filename, 'w') as fd:
            fd.write(content)
        p = subprocess.Popen([GIT_CMD, 'add', filename])
        p.communicate()
        p = subprocess.Popen([GIT_CMD, 'commit', '-m', '%s' % datetime.datetime.utcnow(), '--allow-empty'],
                             stdout=subprocess.PIPE)
        stdout, _ = p.communicate()
        stdout = stdout.decode('utf-8')
        insert = re_insertion.findall(stdout)
        if insert:
            insert = int(insert[0])
        else:
            insert = 0
        delete = re_deletion.findall(stdout)
        if delete:
            delete = int(delete[0])
        else:
            delete = 0
        queue.put({'insertions': insert, 'deletions': delete, 'previous_lines': current_lines,
                   'changes': max(insert, delete)})
    queue = multiprocessing.Queue()
    p = multiprocessing.Process(target=_commit, args=(id_, base_name, content, queue))
    p.start()
    res = queue.get()
    p.join()
    email = schedule.get('email')
    if not email:
        return
    changes = res.get('changes')
    if not changes:
        return
    min_change = schedule.get('minimum_change')
    previous_lines = res.get('previous_lines')
    if min_change and previous_lines:
        min_change = float(min_change)
        change_fraction = res.get('changes') / previous_lines
        if change_fraction < min_change:
            return
    # send notification
    diff = get_diff(id_).get('diff')
    if not diff:
        return
    send_email(to=email, subject='%s page changed' % schedule.get('title'),
               body='changes:\n\n%s' % diff)


def safe_run_job(id_=None, *args, **kwargs):
    try:
        run_job(id_, *args, **kwargs)
    except Exception as e:
        send_email('error executing job %s: %s' % (id_, e))


def send_email(to, subject='diffido', body='', from_=None):
    msg = MIMEText(body)
    msg['Subject'] = subject
    msg['From'] = from_ or EMAIL_FROM
    msg['To'] = to
    s = smtplib.SMTP('localhost')
    s.send_message(msg)
    s.quit()


def get_history(id_):
    def _history(id_, queue):
        os.chdir('storage/%s' % id_)
        p = subprocess.Popen([GIT_CMD, 'log', '--pretty=oneline', '--shortstat'], stdout=subprocess.PIPE)
        stdout, _ = p.communicate()
        queue.put(stdout)
    queue = multiprocessing.Queue()
    p = multiprocessing.Process(target=_history, args=(id_, queue))
    p.start()
    res = queue.get().decode('utf-8')
    p.join()
    history = []
    for match in re_commit.finditer(res):
        info = match.groupdict()
        info['insertions'] = int(info['insertions'] or 0)
        info['deletions'] = int(info['deletions'] or 0)
        info['changes'] = max(info['insertions'], info['deletions'])
        history.append(info)
    lastid = None
    if history and 'id' in history[0]:
        lastid = history[0]['id']
    for idx, item in enumerate(history):
        item['seq'] = idx + 1
    schedule = get_schedule(id_)
    return {'history': history, 'lastid': lastid, 'schedule': schedule}


def get_diff(id_, commit_id='HEAD', old_commit_id=None):
    def _history(id_, commit_id, old_commit_id, queue):
        os.chdir('storage/%s' % id_)
        p = subprocess.Popen([GIT_CMD, 'diff', old_commit_id or '%s~' % commit_id, commit_id],
                             stdout=subprocess.PIPE)
        stdout, _ = p.communicate()
        queue.put(stdout)
    queue = multiprocessing.Queue()
    p = multiprocessing.Process(target=_history, args=(id_, commit_id, old_commit_id, queue))
    p.start()
    res = queue.get().decode('utf-8')
    p.join()
    schedule = get_schedule(id_)
    return {'diff': res, 'schedule': schedule}


def scheduler_update(scheduler, id_):
    schedule = get_schedule(id_, add_id=False)
    if not schedule:
        return
    trigger = schedule.get('trigger')
    if trigger not in ('interval', 'cron'):
        return
    args = {}
    if trigger == 'interval':
        args['trigger'] = 'interval'
        for unit in 'weeks', 'days', 'hours', 'minutes', 'seconds':
            if 'interval_%s' % unit not in schedule:
                continue
            args[unit] = int(schedule['interval_%s' % unit])
    elif trigger == 'cron':
        cron_trigger = CronTrigger.from_crontab(schedule.get('cron_crontab'))
        args['trigger'] = cron_trigger
    git_create_repo(id_)
    scheduler.add_job(safe_run_job, id=id_, replace_existing=True, kwargs={'id_': id_}, **args)


def scheduler_delete(scheduler, id_):
    scheduler.remove_job(job_id=id_)
    git_delete_repo(id_)


def reset_from_schedules(scheduler):
    scheduler.remove_all_jobs()
    for key in read_schedules().get('schedules', {}).keys():
        scheduler_update(scheduler, id_=key)


def git_create_repo(id_):
    repo_dir = 'storage/%s' % id_
    if os.path.isdir(repo_dir):
        return
    p = subprocess.Popen([GIT_CMD, 'init', repo_dir])
    p.communicate()


def git_delete_repo(id_):
    repo_dir = 'storage/%s' % id_
    if not os.path.isdir(repo_dir):
        return
    shutil.rmtree(repo_dir)


class DiffidoBaseException(Exception):
    """Base class for diffido custom exceptions.

    :param message: text message
    :type message: str
    :param status: numeric http status code
    :type status: int"""
    def __init__(self, message, status=400):
        super(DiffidoBaseException, self).__init__(message)
        self.message = message
        self.status = status


class BaseHandler(tornado.web.RequestHandler):
    """Base class for request handlers."""
    # Cache currently connected users.
    _users_cache = {}

    # set of documents we're managing (a collection in MongoDB or a table in a SQL database)
    document = None
    collection = None

    # A property to access the first value of each argument.
    arguments = property(lambda self: dict([(k, v[0].decode('utf-8'))
                                            for k, v in self.request.arguments.items()]))

    @property
    def clean_body(self):
        """Return a clean dictionary from a JSON body, suitable for a query on MongoDB.

        :returns: a clean copy of the body arguments
        :rtype: dict"""
        return escape.json_decode(self.request.body or '{}')

    def write_error(self, status_code, **kwargs):
        """Default error handler."""
        if isinstance(kwargs.get('exc_info', (None, None))[1], DiffidoBaseException):
            exc = kwargs['exc_info'][1]
            status_code = exc.status
            message = exc.message
        else:
            message = 'internal error'
        self.build_error(message, status=status_code)

    def is_api(self):
        """Return True if the path is from an API call."""
        return self.request.path.startswith('/v%s' % API_VERSION)

    def initialize(self, **kwargs):
        """Add every passed (key, value) as attributes of the instance."""
        for key, value in kwargs.items():
            setattr(self, key, value)

    def build_error(self, message='', status=400):
        """Build and write an error message.

        :param message: textual message
        :type message: str
        :param status: HTTP status code
        :type status: int
        """
        self.set_status(status)
        self.write({'error': True, 'message': message})

    def build_success(self, message='', status=200):
        """Build and write a success message.

        :param message: textual message
        :type message: str
        :param status: HTTP status code
        :type status: int
        """
        self.set_status(status)
        self.write({'error': False, 'message': message})


class SchedulesHandler(BaseHandler):
    @gen.coroutine
    def get(self, id_=None, *args, **kwargs):
        if id_ is not None:
            self.write({'schedule': get_schedule(id_)})
            return
        schedules = read_schedules()
        self.write(schedules)

    @gen.coroutine
    def put(self, id_=None, *args, **kwargs):
        if id_ is None:
            return self.build_error(message='update action requires an ID')
        data = self.clean_body
        schedules = read_schedules()
        if id_ not in schedules.get('schedules', {}):
            return self.build_error(message='schedule %s not found' % id_)
        schedules['schedules'][id_] = data
        write_schedules(schedules)
        scheduler_update(scheduler=self.scheduler, id_=id_)
        self.write(get_schedule(id_=id_))

    @gen.coroutine
    def post(self, *args, **kwargs):
        data = self.clean_body
        schedules = read_schedules()
        id_ = next_id(schedules)
        schedules['schedules'][id_] = data
        write_schedules(schedules)
        scheduler_update(scheduler=self.scheduler, id_=id_)
        self.write(get_schedule(id_=id_))

    @gen.coroutine
    def delete(self, id_=None, *args, **kwargs):
        if id_ is None:
            return self.build_error(message='an ID must be specified')
        schedules = read_schedules()
        if id_ in schedules.get('schedules', {}):
            del schedules['schedules'][id_]
            write_schedules(schedules)
        scheduler_delete(scheduler=self.scheduler, id_=id_)
        self.build_success(message='removed schedule %s' % id_)


class ResetSchedulesHandler(BaseHandler):
    @gen.coroutine
    def post(self, *args, **kwargs):
        reset_from_schedules(self.scheduler)


class HistoryHandler(BaseHandler):
    @gen.coroutine
    def get(self, id_, *args, **kwargs):
        self.write(get_history(id_))


class DiffHandler(BaseHandler):
    @gen.coroutine
    def get(self, id_, commit_id, old_commit_id=None, *args, **kwargs):
        self.write(get_diff(id_, commit_id, old_commit_id))


class TemplateHandler(BaseHandler):
    """Handler for the / path."""
    app_path = os.path.join(os.path.dirname(__file__), "dist")

    @gen.coroutine
    def get(self, *args, **kwargs):
        page = 'index.html'
        if args and args[0]:
            page = args[0].strip('/')
        arguments = self.arguments
        self.render(page, **arguments)


def serve():
    global EMAIL_FROM
    jobstores = {'default': SQLAlchemyJobStore(url=JOBS_STORE)}
    scheduler = TornadoScheduler(jobstores=jobstores)
    scheduler.start()

    define('port', default=3210, help='run on the given port', type=int)
    define('address', default='', help='bind the server at the given address', type=str)
    define('ssl_cert', default=os.path.join(os.path.dirname(__file__), 'ssl', 'diffido_cert.pem'),
            help='specify the SSL certificate to use for secure connections')
    define('ssl_key', default=os.path.join(os.path.dirname(__file__), 'ssl', 'diffido_key.pem'),
            help='specify the SSL private key to use for secure connections')
    define('admin-email', default='', help='email address of the site administrator', type=str)
    define('debug', default=False, help='run in debug mode')
    define('config', help='read configuration file',
            callback=lambda path: tornado.options.parse_config_file(path, final=False))
    if not options.config and os.path.isfile(DEFAULT_CONF):
        tornado.options.parse_config_file(DEFAULT_CONF, final=False)
    tornado.options.parse_command_line()
    if options.admin_email:
        EMAIL_FROM = options.admin_email

    if options.debug:
        logger.setLevel(logging.DEBUG)

    ssl_options = {}
    if os.path.isfile(options.ssl_key) and os.path.isfile(options.ssl_cert):
        ssl_options = dict(certfile=options.ssl_cert, keyfile=options.ssl_key)

    init_params = dict(listen_port=options.port, logger=logger, ssl_options=ssl_options,
                       scheduler=scheduler)

    _reset_schedules_path = r'schedules/reset'
    _schedules_path = r'schedules/?(?P<id_>\d+)?'
    _history_path = r'history/?(?P<id_>\d+)'
    _diff_path = r'diff/(?P<id_>\d+)/(?P<commit_id>[0-9a-f]+)/?(?P<old_commit_id>[0-9a-f]+)?/?'
    application = tornado.web.Application([
            (r'/api/%s' % _reset_schedules_path, ResetSchedulesHandler, init_params),
            (r'/api/v%s/%s' % (API_VERSION, _reset_schedules_path), ResetSchedulesHandler, init_params),
            (r'/api/%s' % _schedules_path, SchedulesHandler, init_params),
            (r'/api/v%s/%s' % (API_VERSION, _schedules_path), SchedulesHandler, init_params),
            (r'/api/%s' % _history_path, HistoryHandler, init_params),
            (r'/api/v%s/%s' % (API_VERSION, _history_path), HistoryHandler, init_params),
            (r'/api/%s' % _diff_path, DiffHandler, init_params),
            (r'/api/v%s/%s' % (API_VERSION, _diff_path), DiffHandler, init_params),
            (r'/?(.*)', TemplateHandler, init_params),
        ],
        static_path=os.path.join(os.path.dirname(__file__), 'dist/static'),
        template_path=os.path.join(os.path.dirname(__file__), 'dist/'),
        debug=options.debug)
    http_server = tornado.httpserver.HTTPServer(application, ssl_options=ssl_options or None)
    logger.info('Start serving on %s://%s:%d', 'https' if ssl_options else 'http',
                                                 options.address if options.address else '127.0.0.1',
                                                 options.port)
    http_server.listen(options.port, options.address)

    try:
        IOLoop.instance().start()
    except (KeyboardInterrupt, SystemExit):
        pass


if __name__ == '__main__':
    serve()