#!/usr/bin/env python
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

#
# SvnWcSub - Subscribe to a SvnPubSub stream, and keep a set of working copy
# paths in sync
#
# Example:
#  svnwcsub.py svnwcsub.conf
#
# On startup svnwcsub checks the working copy's path, runs a single svn update
# and then watches for changes to that path.
#
# See svnwcsub.conf for more information on its contents.
#

import subprocess
import threading
import sys
import os
import re
import ConfigParser
import time
import logging.handlers
import Queue
import optparse

from twisted.internet import reactor, task, threads
from twisted.internet.utils import getProcessOutput
from twisted.application import internet
from twisted.web.client import HTTPClientFactory, HTTPPageDownloader
from urlparse import urlparse
from xml.sax import handler, make_parser
from twisted.internet import protocol


# check_output() is only available in Python 2.7. Allow us to run with
# earlier versions
try:
    check_output = subprocess.check_output
except AttributeError:
    def check_output(args):  # note: we don't use anything beyond args
        pipe = subprocess.Popen(args, stdout=subprocess.PIPE)
        output, _ = pipe.communicate()
        if pipe.returncode:
            raise subprocess.CalledProcessError(pipe.returncode, args)
        return output


### note: this runs synchronously. within the current Twisted environment,
### it is called from ._get_match() which is run on a thread so it won't
### block the Twisted main loop.
def svn_info(svnbin, path):
    "Run 'svn info' on the target path, returning a dict of info data."
    args = [svnbin, "info", "--non-interactive", "--", path]
    output = check_output(args).strip()
    info = { }
    for line in output.split('\n'):
        idx = line.index(':')
        info[line[:idx]] = line[idx+1:].strip()
    return info


class WorkingCopy(object):
    def __init__(self, bdec, path, url):
        self.bdec = bdec
        self.path = path
        self.url = url
        self.repos = None
        self.match = None
        d = threads.deferToThread(self._get_match)
        d.addCallback(self._set_match)

    def _set_match(self, value):
        self.match = str(value[0])
        self.url = value[1]
        self.repos = value[2]
        self.uuid = value[3]
        self.bdec.wc_ready(self)

    def update_applies(self, uuid, path):
        if self.uuid != uuid:
            return False

        path = str(path)
        if path == self.match:
            #print "ua: Simple match"
            # easy case. woo.
            return True
        if len(path) < len(self.match):
            # path is potentially a parent directory of match?
            #print "ua: parent check"
            if self.match[0:len(path)] == path:
                return True
        if len(path) > len(self.match):
            # path is potentially a sub directory of match
            #print "ua: sub dir check"
            if path[0:len(self.match)] == self.match:
                return True
        return False

    def _get_match(self):
        ### quick little hack to auto-checkout missing working copies
        if not os.path.isdir(self.path):
            logging.info("autopopulate %s from %s" % (self.path, self.url))
            subprocess.check_call([self.bdec.svnbin, 'co', '-q',
                                   '--non-interactive',
                                   '--config-dir',
                                   '/home/svnwc/.subversion',
                                   '--', self.url, self.path])

        # Fetch the info for matching dirs_changed against this WC
        info = svn_info(self.bdec.svnbin, self.path)
        url = info['URL']
        repos = info['Repository Root']
        uuid = info['Repository UUID']
        relpath = url[len(repos):]  # also has leading '/'
        return [relpath, url, repos, uuid]


class HTTPStream(HTTPClientFactory):
    protocol = HTTPPageDownloader

    def __init__(self, url):
        self.url = url
        HTTPClientFactory.__init__(self, url, method="GET", agent="SvnWcSub/0.1.0")

    def pageStart(self, partial):
        pass

    def pagePart(self, data):
        pass

    def pageEnd(self):
        pass

class Revision:
    def __init__(self, repos, rev):
        self.repos = repos
        self.rev = rev
        self.dirs_changed = []

class StreamHandler(handler.ContentHandler):
    def __init__(self, stream, bdec):
        handler.ContentHandler.__init__(self)
        self.stream = stream
        self.bdec =  bdec
        self.rev = None
        self.text_value = None

    def startElement(self, name, attrs):
        #print "start element: %s" % (name)
        """
        <commit revision="7">
                        <dirs_changed><path>/</path></dirs_changed>
                      </commit>
        """
        if name == "commit":
            self.rev = Revision(attrs['repository'], int(attrs['revision']))
        elif name == "stillalive":
            self.bdec.stillalive(self.stream)
    def characters(self, data):
        if self.text_value is not None:
            self.text_value = self.text_value + data
        else:
            self.text_value = data

    def endElement(self, name):
        #print "end   element: %s" % (name)
        if name == "commit":
            self.bdec.commit(self.stream, self.rev)
            self.rev = None
        if name == "path" and self.text_value is not None and self.rev is not None:
            self.rev.dirs_changed.append(self.text_value.strip())
        self.text_value = None


class XMLHTTPStream(HTTPStream):
    def __init__(self, url, bdec):
        HTTPStream.__init__(self, url)
        self.alive = 0
        self.bdec =  bdec
        self.parser = make_parser(['xml.sax.expatreader'])
        self.handler = StreamHandler(self, bdec)
        self.parser.setContentHandler(self.handler)

    def pageStart(self, parital):
        self.bdec.pageStart(self)

    def pagePart(self, data):
        self.parser.feed(data)

    def pageEnd(self):
        self.bdec.pageEnd(self)

def connectTo(url, bdec):
    u = urlparse(url)
    port = u.port
    if not port:
        port = 80
    s = XMLHTTPStream(url, bdec)
    if bdec.service:
      conn = internet.TCPClient(u.hostname, u.port, s)
      conn.setServiceParent(bdec.service)
    else:
      conn = reactor.connectTCP(u.hostname, u.port, s)
    return [s, conn]


CHECKBEAT_TIME = 60
PRODUCTION_RE_FILTER = re.compile("/websites/production/[^/]+/")

class BigDoEverythingClasss(object):
    def __init__(self, config, service = None):
        self.urls = [s.strip() for s in config.get_value('streams').split()]
        self.svnbin = config.get_value('svnbin')
        self.env = config.get_env()
        self.worker = BackgroundWorker(self.svnbin, self.env)
        self.service = service
        self.failures = 0
        self.alive = time.time()
        self.checker = task.LoopingCall(self._checkalive)
        self.transports = {}
        self.streams = {}
        for u in self.urls:
          self._restartStream(u)
        self.watch = []
        for path, url in config.get_track().items():
            # working copies auto-register with the BDEC when they are ready.
            WorkingCopy(self, path, url)
        self.checker.start(CHECKBEAT_TIME)

    def pageStart(self, stream):
        logging.info("Stream %s Connection Established" % (stream.url))
        self.failures = 0

    def pageEnd(self, stream):
        logging.info("Stream %s Connection Dead" % (stream.url))
        self.streamDead(stream.url)

    def _restartStream(self, url):
        (self.streams[url], self.transports[url]) = connectTo(url, self)
        self.streams[url].deferred.addBoth(self.streamDead, url)
        self.streams[url].alive = time.time()

    def _checkalive(self):
        n = time.time()
        for k in self.streams.keys():
          s = self.streams[k]
          if n - s.alive > CHECKBEAT_TIME:
            logging.info("Stream %s is dead, reconnecting" % (s.url))
            #self.transports[s.url].disconnect()
            self.streamDead(self, s.url)

#        d=filter(lambda x:x not in self.streams.keys(), self.urls)
#        for u in d:
#          self._restartStream(u)

    def stillalive(self, stream):
        stream.alive = time.time()

    def streamDead(self, url, result=None):
        s = self.streams.get(url)
        if not s:
          logging.info("Stream %s is messed up" % (url))
          return
        BACKOFF_SECS = 5
        BACKOFF_MAX = 60
        #self.checker.stop()

        self.streams[url] = None
        self.transports[url] = None
        self.failures += 1
        backoff = min(self.failures * BACKOFF_SECS, BACKOFF_MAX)
        logging.info("Stream disconnected, trying again in %d seconds.... %s" % (backoff, s.url))
        reactor.callLater(backoff, self._restartStream, url)

    def wc_ready(self, wc):
        # called when a working copy object has its basic info/url,
        # Add it to our watchers, and trigger an svn update.
        logging.info("Watching WC at %s <-> %s" % (wc.path, wc.url))
        self.watch.append(wc)
        self.worker.add_work(OP_UPDATE, wc)

    def _normalize_path(self, path):
        if path[0] != '/':
            return "/" + path
        return os.path.abspath(path)

    def commit(self, stream, rev):
        logging.info("COMMIT r%d (%d paths) via %s" % (rev.rev, len(rev.dirs_changed), stream.url))
        paths = map(self._normalize_path, rev.dirs_changed)
        if len(paths):
            pre = os.path.commonprefix(paths)
            if pre == "/websites/":
                # special case for svnmucc "dynamic content" buildbot commits
                # just take the first production path to avoid updating all cms working copies
                for p in paths:
                    m = PRODUCTION_RE_FILTER.match(p)
                    if m:
                        pre = m.group(0)
                        break

            #print "Common Prefix: %s" % (pre)
            wcs = [wc for wc in self.watch if wc.update_applies(rev.repos, pre)]
            logging.info("Updating %d WC for r%d" % (len(wcs), rev.rev))
            for wc in wcs:
                self.worker.add_work(OP_UPDATE, wc)


# Start logging warnings if the work backlog reaches this many items
BACKLOG_TOO_HIGH = 20
OP_UPDATE = 'update'
OP_CLEANUP = 'cleanup'

class BackgroundWorker(threading.Thread):
    def __init__(self, svnbin, env):
        threading.Thread.__init__(self)

        # The main thread/process should not wait for this thread to exit.
        ### compat with Python 2.5
        self.setDaemon(True)

        self.svnbin = svnbin
        self.env = env
        self.q = Queue.Queue()

        self.has_started = False

    def run(self):
        while True:
            if self.q.qsize() > BACKLOG_TOO_HIGH:
                logging.warn('worker backlog is at %d', self.q.qsize())

            # This will block until something arrives
            operation, wc = self.q.get()
            try:
                if operation == OP_UPDATE:
                    self._update(wc)
                elif operation == OP_CLEANUP:
                    self._cleanup(wc)
                else:
                    logging.critical('unknown operation: %s', operation)
            except:
                logging.exception('exception in worker')

            # In case we ever want to .join() against the work queue
            self.q.task_done()

    def add_work(self, operation, wc):
        # Start the thread when work first arrives. Thread-start needs to
        # be delayed in case the process forks itself to become a daemon.
        if not self.has_started:
            self.start()
            self.has_started = True

        self.q.put((operation, wc))

    def _update(self, wc):
        "Update the specified working copy."

        # For giggles, let's clean up the working copy in case something
        # happened earlier.
        self._cleanup(wc)

        logging.info("updating: %s", wc.path)

        ### we need to move some of these args into the config. these are
        ### still specific to the ASF setup.
        args = [self.svnbin, 'update',
                '--quiet',
                '--config-dir', '/home/svnwc/.subversion',
                '--non-interactive',
                '--trust-server-cert',
                '--ignore-externals',
                wc.path]
        subprocess.check_call(args, env=self.env)

        ### check the loglevel before running 'svn info'?
        info = svn_info(self.svnbin, wc.path)
        logging.info("updated: %s now at r%s", wc.path, info['Revision'])

    def _cleanup(self, wc):
        "Run a cleanup on the specified working copy."

        ### we need to move some of these args into the config. these are
        ### still specific to the ASF setup.
        args = [self.svnbin, 'cleanup',
                '--config-dir', '/home/svnwc/.subversion',
                '--non-interactive',
                '--trust-server-cert',
                wc.path]
        subprocess.check_call(args, env=self.env)


class ReloadableConfig(ConfigParser.SafeConfigParser):
    def __init__(self, fname):
        ConfigParser.SafeConfigParser.__init__(self)

        self.fname = fname
        self.read(fname)

        ### install a signal handler to set SHOULD_RELOAD. BDEC should
        ### poll this flag, and then adjust its internal structures after
        ### the reload.
        self.should_reload = False

    def reload(self):
        # Delete everything. Just re-reading would overlay, and would not
        # remove sections/options. Note that [DEFAULT] will not be removed.
        for section in self.sections():
            self.remove_section(section)

        # Now re-read the configuration file.
        self.read(fname)

    def get_value(self, which):
        return self.get(ConfigParser.DEFAULTSECT, which)

    def get_env(self):
        env = os.environ.copy()
        default_options = self.defaults().keys()
        for name, value in self.items('env'):
            if name not in default_options:
                env[name] = value
        return env

    def get_track(self):
        "Return the {PATH: URL} dictionary of working copies to track."
        track = dict(self.items('track'))
        for name in self.defaults().keys():
            del track[name]
        return track

    def optionxform(self, option):
        # Do not lowercase the option name.
        return str(option)


def prepare_logging(logfile):
    "Log to the specified file, or to stdout if None."

    if logfile:
        # Rotate logs daily, keeping 7 days worth.
        handler = logging.handlers.TimedRotatingFileHandler(
          logfile, when='midnight', backupCount=7,
          )
    else:
        handler = logging.StreamHandler(sys.stdout)

    # Add a timestamp to the log records
    formatter = logging.Formatter('%(asctime)s [%(levelname)s] %(message)s',
                                  '%Y-%m-%d %H:%M:%S')
    handler.setFormatter(formatter)

    # Apply the handler to the root logger
    root = logging.getLogger()
    root.addHandler(handler)
    
    ### use logging.INFO for now. switch to cmdline option or a config?
    root.setLevel(logging.INFO)


def handle_options(options):
    # Set up the logging, then process the rest of the options.
    prepare_logging(options.logfile)

    if options.pidfile:
        pid = os.getpid()
        open(options.pidfile, 'w').write('%s\n' % pid)
        logging.info('pid %d written to %s', pid, options.pidfile)

    if options.uid:
        try:
            uid = int(options.uid)
        except ValueError:
            import pwd
            uid = pwd.getpwnam(options.uid)[2]
        logging.info('setting uid %d', uid)
        os.setuid(uid)

    if options.gid:
        try:
            gid = int(options.gid)
        except ValueError:
            import grp
            gid = grp.getgrnam(options.gid)[2]
        logging.info('setting gid %d', gid)
        os.setgid(gid)

    if options.umask:
        umask = int(options.umask, 8)
        os.umask(umask)
        logging.info('umask set to %03o', umask)


def main(args):
    parser = optparse.OptionParser(
        description='An SvnPubSub client to keep working copies synchronized '
                    'with a repository.',
        usage='Usage: %prog [options] CONFIG_FILE',
        )
    parser.add_option('--logfile',
                      help='filename for logging')
    parser.add_option('--pidfile',
                      help="the process' PID will be written to this file")
    parser.add_option('--uid',
                      help='switch to this UID before running')
    parser.add_option('--gid',
                      help='switch to this GID before running')
    parser.add_option('--umask',
                      help='set this (octal) umask before running')

    options, extra = parser.parse_args(args)

    if len(extra) != 1:
        parser.error('CONFIG_FILE is required')
    config_file = extra[0]

    # Process any provided options.
    handle_options(options)

    c = ReloadableConfig(config_file)
    big = BigDoEverythingClasss(c)
    reactor.run()


if __name__ == "__main__":
    main(sys.argv[1:])