playstreamproxy2 (WSGI)

ltc2

daemonize.py

@@ -0,0 +1,287 @@
+#!/bin/env python
+'''
+***
+Modified generic daemon class
+***
+
+Author:
+                https://web.archive.org/web/20160305151936/http://www.jejik.com/articles/2007/02/a_simple_unix_linux_daemon_in_python/
+
+Modified by ivars777@gmail.com
+
+License:        http://creativecommons.org/licenses/by-sa/3.0/
+
+'''
+
+# Core modules
+from __future__ import print_function
+import atexit
+import errno
+import os
+import sys
+import time
+import signal
+
+
+class Daemon(object):
+    """
+    A generic daemon class.
+
+    Usage: subclass the Daemon class and override the run() method
+    """
+    def __init__(self, app, pidfile, stdin=os.devnull,
+                 stdout=os.devnull, stderr=os.devnull,
+                 home_dir='.', umask=0o22, verbose=1,
+                 use_gevent=False, use_eventlet=False):
+        self.app = app
+        self.stdin = stdin
+        self.stdout = stdout
+        self.stderr = stderr
+        self.pidfile = pidfile
+        self.home_dir = home_dir
+        self.verbose = verbose
+        self.umask = umask
+        self.daemon_alive = True
+        self.use_gevent = use_gevent
+        self.use_eventlet = use_eventlet
+
+    def log(self, *args):
+        if self.verbose >= 1:
+            print(*args)
+
+    def daemonize(self):
+        """
+        Do the UNIX double-fork magic, see Stevens' "Advanced
+        Programming in the UNIX Environment" for details (ISBN 0201563177)
+        http://www.erlenstar.demon.co.uk/unix/faq_2.html#SEC16
+        """
+        if self.use_eventlet:
+            import eventlet.tpool
+            eventlet.tpool.killall()
+        try:
+            pid = os.fork()
+            if pid > 0:
+                # Exit first parent
+                sys.exit(0)
+        except OSError as e:
+            sys.stderr.write(
+                "fork #1 failed: %d (%s)\n" % (e.errno, e.strerror))
+            sys.exit(1)
+
+        # Decouple from parent environment
+        os.chdir(self.home_dir)
+        os.setsid()
+        os.umask(self.umask)
+
+        # Do second fork
+        try:
+            pid = os.fork()
+            if pid > 0:
+                # Exit from second parent
+                sys.exit(0)
+        except OSError as e:
+            sys.stderr.write(
+                "fork #2 failed: %d (%s)\n" % (e.errno, e.strerror))
+            sys.exit(1)
+
+        if sys.platform != 'darwin':  # This block breaks on OS X
+            # Redirect standard file descriptors
+            sys.stdout.flush()
+            sys.stderr.flush()
+            si = open(self.stdin, 'r')
+            so = open(self.stdout, 'a+')
+            if self.stderr:
+                try:
+                    se = open(self.stderr, 'a+', 0)
+                except ValueError:
+                    # Python 3 can't have unbuffered text I/O
+                    se = open(self.stderr, 'a+', 1)
+            else:
+                se = so
+            os.dup2(si.fileno(), sys.stdin.fileno())
+            os.dup2(so.fileno(), sys.stdout.fileno())
+            os.dup2(se.fileno(), sys.stderr.fileno())
+
+        def sigtermhandler(signum, frame):
+            self.daemon_alive = False
+            sys.exit()
+
+        if self.use_gevent:
+            import gevent
+            gevent.reinit()
+            gevent.signal(signal.SIGTERM, sigtermhandler, signal.SIGTERM, None)
+            gevent.signal(signal.SIGINT, sigtermhandler, signal.SIGINT, None)
+        else:
+            signal.signal(signal.SIGTERM, sigtermhandler)
+            signal.signal(signal.SIGINT, sigtermhandler)
+
+        self.log("Started")
+
+        # Write pidfile
+        atexit.register(
+            self.delpid)  # Make sure pid file is removed if we quit
+        pid = str(os.getpid())
+        open(self.pidfile, 'w+').write("%s\n" % pid)
+
+    def delpid(self):
+        try:
+            # the process may fork itself again
+            pid = int(open(self.pidfile, 'r').read().strip())
+            if pid == os.getpid():
+                os.remove(self.pidfile)
+        except OSError as e:
+            if e.errno == errno.ENOENT:
+                pass
+            else:
+                raise
+
+    def start(self, *args, **kwargs):
+        """
+        Start the daemon
+        """
+
+        self.log("Starting...")
+
+        # Check for a pidfile to see if the daemon already runs
+        try:
+            pf = open(self.pidfile, 'r')
+            pid = int(pf.read().strip())
+            pf.close()
+        except IOError:
+            pid = None
+        except SystemExit:
+            pid = None
+
+        if pid:
+            message = "pidfile %s already exists. Is it already running?\n"
+            sys.stderr.write(message % self.pidfile)
+            sys.exit(1)
+
+        # Start the daemon
+        self.daemonize()
+        self.run(*args, **kwargs)
+
+    def stop(self):
+        """
+        Stop the daemon
+        """
+
+        if self.verbose >= 1:
+            self.log("Stopping...")
+
+        # Get the pid from the pidfile
+        pid = self.get_pid()
+
+        if not pid:
+            message = "pidfile %s does not exist. Not running?\n"
+            sys.stderr.write(message % self.pidfile)
+
+            # Just to be sure. A ValueError might occur if the PID file is
+            # empty but does actually exist
+            if os.path.exists(self.pidfile):
+                os.remove(self.pidfile)
+
+            return  # Not an error in a restart
+
+        # Try killing the daemon process
+        try:
+            i = 0
+            while 1:
+                os.kill(pid, signal.SIGTERM)
+                time.sleep(0.1)
+                i = i + 1
+                if i % 10 == 0:
+                    os.kill(pid, signal.SIGHUP)
+        except OSError as err:
+            if err.errno == errno.ESRCH:
+                if os.path.exists(self.pidfile):
+                    os.remove(self.pidfile)
+            else:
+                print(str(err))
+                sys.exit(1)
+
+        self.log("Stopped")
+
+    def restart(self):
+        """
+        Restart the daemon
+        """
+        self.stop()
+        self.start()
+
+    def get_pid(self):
+        try:
+            pf = open(self.pidfile, 'r')
+            pid = int(pf.read().strip())
+            pf.close()
+        except IOError:
+            pid = None
+        except SystemExit:
+            pid = None
+        return pid
+
+    def is_running(self):
+        pid = self.get_pid()
+
+        if pid is None:
+            self.log('Process is stopped')
+            return False
+        elif os.path.exists('/proc/%d' % pid):
+            self.log('Process (pid %d) is running...' % pid)
+            return True
+        else:
+            self.log('Process (pid %d) is killed' % pid)
+            return False
+
+    def run(self, *args, **kwargs):
+        """
+        Running app
+        """
+        self.app(*args, **kwargs)
+        #self.log("Starting foreground...")
+
+
+def main(cmd):
+    from datetime import datetime
+    from time import sleep
+    import subprocess
+    print("Starting cmd: ")
+    print(str(cmd))
+    subprocess.call(cmd)
+    #while True:
+    #    print(str(datetime.now()))
+    #    sleep(5)
+
+#def main2(cmd):
+#    from bottle import route, run
+
+#    @route('/')
+#    def index():
+#        return '<b>Hello </b>!'
+
+#    run(host='localhost', port=8080)
+
+
+if __name__ == "__main__":
+    if len(sys.argv) > 1:
+        cmd=sys.argv[2:]
+        app = cmd[0] if cmd else "daemonize2"
+        pid = "/var/run/%s.pid"%app
+        daemon = Daemon(main,pid)
+        if "start" == sys.argv[1]:
+            daemon.start(cmd)
+        elif "stop" == sys.argv[1]:
+            daemon.stop()
+        elif "restart" == sys.argv[1]:
+            daemon.restart()
+        elif "manualstart" == sys.argv[1]:
+            daemon.run(cmd)
+        elif "status" == sys.argv[1]:
+            daemon.is_running()
+        else:
+            print("Unknown command")
+            sys.exit(2)
+        sys.exit(0)
+    else:
+        print("usage: %s start|stop|restart|manualstart" % sys.argv[0])
+        sys.exit(2)

diskcache/__init__.py

@@ -0,0 +1,36 @@
+"DiskCache: disk and file backed cache."
+
+from .core import Cache, Disk, UnknownFileWarning, EmptyDirWarning, Timeout
+from .core import DEFAULT_SETTINGS, ENOVAL, EVICTION_POLICY, UNKNOWN
+from .fanout import FanoutCache
+from .persistent import Deque, Index
+
+__all__ = [
+    'Cache',
+    'Disk',
+    'UnknownFileWarning',
+    'EmptyDirWarning',
+    'Timeout',
+    'DEFAULT_SETTINGS',
+    'ENOVAL',
+    'EVICTION_POLICY',
+    'UNKNOWN',
+    'FanoutCache',
+    'Deque',
+    'Index',
+]
+
+try:
+    from .djangocache import DjangoCache  # pylint: disable=wrong-import-position
+    __all__.append('DjangoCache')
+except Exception:  # pylint: disable=broad-except
+    # Django not installed or not setup so ignore.
+    pass
+
+
+__title__ = 'diskcache'
+__version__ = '3.1.1'
+__build__ = 0x030101
+__author__ = 'Grant Jenks'
+__license__ = 'Apache 2.0'
+__copyright__ = 'Copyright 2016-2018 Grant Jenks'

diskcache/cli.py

@@ -0,0 +1 @@
+"Command line interface to disk cache."

diskcache/fanout.py

@@ -0,0 +1,595 @@
+"Fanout cache automatically shards keys and values."
+
+import itertools as it
+import os.path as op
+import sqlite3
+import time
+
+from .core import ENOVAL, DEFAULT_SETTINGS, Cache, Disk, Timeout
+from .memo import memoize
+from .persistent import Deque, Index
+
+
+class FanoutCache(object):
+    "Cache that shards keys and values."
+    def __init__(self, directory, shards=8, timeout=0.010, disk=Disk,
+                 **settings):
+        """Initialize cache instance.
+
+        :param str directory: cache directory
+        :param int shards: number of shards to distribute writes
+        :param float timeout: SQLite connection timeout
+        :param disk: `Disk` instance for serialization
+        :param settings: any of `DEFAULT_SETTINGS`
+
+        """
+        self._directory = directory
+        self._count = shards
+        default_size_limit = DEFAULT_SETTINGS['size_limit']
+        size_limit = settings.pop('size_limit', default_size_limit) / shards
+        self._shards = tuple(
+            Cache(
+                op.join(directory, '%03d' % num),
+                timeout=timeout,
+                disk=disk,
+                size_limit=size_limit,
+                **settings
+            )
+            for num in range(shards)
+        )
+        self._hash = self._shards[0].disk.hash
+        self._deques = {}
+        self._indexes = {}
+
+
+    @property
+    def directory(self):
+        """Cache directory."""
+        return self._directory
+
+
+    def __getattr__(self, name):
+        return getattr(self._shards[0], name)
+
+
+    def set(self, key, value, expire=None, read=False, tag=None, retry=False):
+        """Set `key` and `value` item in cache.
+
+        When `read` is `True`, `value` should be a file-like object opened
+        for reading in binary mode.
+
+        If database timeout occurs then fails silently unless `retry` is set to
+        `True` (default `False`).
+
+        :param key: key for item
+        :param value: value for item
+        :param float expire: seconds until the key expires
+            (default None, no expiry)
+        :param bool read: read value as raw bytes from file (default False)
+        :param str tag: text to associate with key (default None)
+        :param bool retry: retry if database timeout expires (default False)
+        :return: True if item was set
+
+        """
+        index = self._hash(key) % self._count
+        set_func = self._shards[index].set
+
+        while True:
+            try:
+                return set_func(key, value, expire, read, tag)
+            except Timeout:
+                if retry:
+                    continue
+                else:
+                    return False
+
+
+    def __setitem__(self, key, value):
+        """Set `key` and `value` item in cache.
+
+        Calls :func:`FanoutCache.set` internally with `retry` set to `True`.
+
+        :param key: key for item
+        :param value: value for item
+
+        """
+        self.set(key, value, retry=True)
+
+
+    def add(self, key, value, expire=None, read=False, tag=None, retry=False):
+        """Add `key` and `value` item to cache.
+
+        Similar to `set`, but only add to cache if key not present.
+
+        This operation is atomic. Only one concurrent add operation for given
+        key from separate threads or processes will succeed.
+
+        When `read` is `True`, `value` should be a file-like object opened
+        for reading in binary mode.
+
+        :param key: key for item
+        :param value: value for item
+        :param float expire: seconds until the key expires
+            (default None, no expiry)
+        :param bool read: read value as bytes from file (default False)
+        :param str tag: text to associate with key (default None)
+        :param bool retry: retry if database timeout expires (default False)
+        :return: True if item was added
+
+        """
+        index = self._hash(key) % self._count
+        add_func = self._shards[index].add
+
+        while True:
+            try:
+                return add_func(key, value, expire, read, tag)
+            except Timeout:
+                if retry:
+                    continue
+                else:
+                    return False
+
+
+    def incr(self, key, delta=1, default=0, retry=False):
+        """Increment value by delta for item with key.
+
+        If key is missing and default is None then raise KeyError. Else if key
+        is missing and default is not None then use default for value.
+
+        Operation is atomic. All concurrent increment operations will be
+        counted individually.
+
+        Assumes value may be stored in a SQLite column. Most builds that target
+        machines with 64-bit pointer widths will support 64-bit signed
+        integers.
+
+        :param key: key for item
+        :param int delta: amount to increment (default 1)
+        :param int default: value if key is missing (default 0)
+        :param bool retry: retry if database timeout expires (default False)
+        :return: new value for item on success else None
+        :raises KeyError: if key is not found and default is None
+
+        """
+        index = self._hash(key) % self._count
+        incr_func = self._shards[index].incr
+
+        while True:
+            try:
+                return incr_func(key, delta, default)
+            except Timeout:
+                if retry:
+                    continue
+                else:
+                    return None
+
+
+    def decr(self, key, delta=1, default=0, retry=False):
+        """Decrement value by delta for item with key.
+
+        If key is missing and default is None then raise KeyError. Else if key
+        is missing and default is not None then use default for value.
+
+        Operation is atomic. All concurrent decrement operations will be
+        counted individually.
+
+        Unlike Memcached, negative values are supported. Value may be
+        decremented below zero.
+
+        Assumes value may be stored in a SQLite column. Most builds that target
+        machines with 64-bit pointer widths will support 64-bit signed
+        integers.
+
+        :param key: key for item
+        :param int delta: amount to decrement (default 1)
+        :param int default: value if key is missing (default 0)
+        :param bool retry: retry if database timeout expires (default False)
+        :return: new value for item on success else None
+        :raises KeyError: if key is not found and default is None
+
+        """
+        return self.incr(key, -delta, default, retry)
+
+
+    def get(self, key, default=None, read=False, expire_time=False, tag=False,
+            retry=False):
+        """Retrieve value from cache. If `key` is missing, return `default`.
+
+        If database timeout occurs then returns `default` unless `retry` is set
+        to `True` (default `False`).
+
+        :param key: key for item
+        :param default: return value if key is missing (default None)
+        :param bool read: if True, return file handle to value
+            (default False)
+        :param float expire_time: if True, return expire_time in tuple
+            (default False)
+        :param tag: if True, return tag in tuple (default False)
+        :param bool retry: retry if database timeout expires (default False)
+        :return: value for item if key is found else default
+
+        """
+        index = self._hash(key) % self._count
+        get_func = self._shards[index].get
+
+        while True:
+            try:
+                return get_func(
+                    key, default=default, read=read, expire_time=expire_time,
+                    tag=tag,
+                )
+            except (Timeout, sqlite3.OperationalError):
+                if retry:
+                    continue
+                else:
+                    return default
+
+
+    def __getitem__(self, key):
+        """Return corresponding value for `key` from cache.
+
+        Calls :func:`FanoutCache.get` internally with `retry` set to `True`.
+
+        :param key: key for item
+        :return: value for item
+        :raises KeyError: if key is not found
+
+        """
+        value = self.get(key, default=ENOVAL, retry=True)
+
+        if value is ENOVAL:
+            raise KeyError(key)
+
+        return value
+
+
+    def read(self, key):
+        """Return file handle corresponding to `key` from cache.
+
+        :param key: key for item
+        :return: file open for reading in binary mode
+        :raises KeyError: if key is not found
+
+        """
+        handle = self.get(key, default=ENOVAL, read=True, retry=True)
+        if handle is ENOVAL:
+            raise KeyError(key)
+        return handle
+
+
+    def __contains__(self, key):
+        """Return `True` if `key` matching item is found in cache.
+
+        :param key: key for item
+        :return: True if key is found
+
+        """
+        index = self._hash(key) % self._count
+        return key in self._shards[index]
+
+
+    def pop(self, key, default=None, expire_time=False, tag=False,
+            retry=False):
+        """Remove corresponding item for `key` from cache and return value.
+
+        If `key` is missing, return `default`.
+
+        Operation is atomic. Concurrent operations will be serialized.
+
+        :param key: key for item
+        :param default: return value if key is missing (default None)
+        :param float expire_time: if True, return expire_time in tuple
+            (default False)
+        :param tag: if True, return tag in tuple (default False)
+        :param bool retry: retry if database timeout expires (default False)
+        :return: value for item if key is found else default
+
+        """
+        index = self._hash(key) % self._count
+        pop_func = self._shards[index].pop
+
+        while True:
+            try:
+                return pop_func(
+                    key, default=default, expire_time=expire_time, tag=tag,
+                )
+            except Timeout:
+                if retry:
+                    continue
+                else:
+                    return default
+
+
+    def delete(self, key, retry=False):
+        """Delete corresponding item for `key` from cache.
+
+        Missing keys are ignored.
+
+        If database timeout occurs then fails silently unless `retry` is set to
+        `True` (default `False`).
+
+        :param key: key for item
+        :param bool retry: retry if database timeout expires (default False)
+        :return: True if item was deleted
+
+        """
+        index = self._hash(key) % self._count
+        del_func = self._shards[index].__delitem__
+
+        while True:
+            try:
+                return del_func(key)
+            except Timeout:
+                if retry:
+                    continue
+                else:
+                    return False
+            except KeyError:
+                return False
+
+
+    def __delitem__(self, key):
+        """Delete corresponding item for `key` from cache.
+
+        Calls :func:`FanoutCache.delete` internally with `retry` set to `True`.
+
+        :param key: key for item
+        :raises KeyError: if key is not found
+
+        """
+        deleted = self.delete(key, retry=True)
+
+        if not deleted:
+            raise KeyError(key)
+
+
+    memoize = memoize
+
+
+    def check(self, fix=False):
+        """Check database and file system consistency.
+
+        Intended for use in testing and post-mortem error analysis.
+
+        While checking the cache table for consistency, a writer lock is held
+        on the database. The lock blocks other cache clients from writing to
+        the database. For caches with many file references, the lock may be
+        held for a long time. For example, local benchmarking shows that a
+        cache with 1,000 file references takes ~60ms to check.
+
+        :param bool fix: correct inconsistencies
+        :return: list of warnings
+        :raises Timeout: if database timeout expires
+
+        """
+        return sum((shard.check(fix=fix) for shard in self._shards), [])
+
+
+    def expire(self):
+        """Remove expired items from cache.
+
+        :return: count of items removed
+
+        """
+        return self._remove('expire', args=(time.time(),))
+
+
+    def create_tag_index(self):
+        """Create tag index on cache database.
+
+        It is better to initialize cache with `tag_index=True` than use this.
+
+        :raises Timeout: if database timeout expires
+
+        """
+        for shard in self._shards:
+            shard.create_tag_index()
+
+
+    def drop_tag_index(self):
+        """Drop tag index on cache database.
+
+        :raises Timeout: if database timeout expires
+
+        """
+        for shard in self._shards:
+            shard.drop_tag_index()
+
+
+    def evict(self, tag):
+        """Remove items with matching `tag` from cache.
+
+        :param str tag: tag identifying items
+        :return: count of items removed
+
+        """
+        return self._remove('evict', args=(tag,))
+
+
+    def cull(self):
+        """Cull items from cache until volume is less than size limit.
+
+        :return: count of items removed
+
+        """
+        return self._remove('cull')
+
+
+    def clear(self):
+        """Remove all items from cache.
+
+        :return: count of items removed
+
+        """
+        return self._remove('clear')
+
+
+    def _remove(self, name, args=()):
+        total = 0
+        for shard in self._shards:
+            method = getattr(shard, name)
+            while True:
+                try:
+                    count = method(*args)
+                    total += count
+                except Timeout as timeout:
+                    total += timeout.args[0]
+                else:
+                    break
+        return total
+
+
+    def stats(self, enable=True, reset=False):
+        """Return cache statistics hits and misses.
+
+        :param bool enable: enable collecting statistics (default True)
+        :param bool reset: reset hits and misses to 0 (default False)
+        :return: (hits, misses)
+
+        """
+        results = [shard.stats(enable, reset) for shard in self._shards]
+        return (sum(result[0] for result in results),
+                sum(result[1] for result in results))
+
+
+    def volume(self):
+        """Return estimated total size of cache on disk.
+
+        :return: size in bytes
+
+        """
+        return sum(shard.volume() for shard in self._shards)
+
+
+    def close(self):
+        "Close database connection."
+        for shard in self._shards:
+            shard.close()
+        self._deques.clear()
+        self._indexes.clear()
+
+
+    def __enter__(self):
+        return self
+
+
+    def __exit__(self, *exception):
+        self.close()
+
+
+    def __getstate__(self):
+        return (self._directory, self._count, self.timeout, type(self.disk))
+
+
+    def __setstate__(self, state):
+        self.__init__(*state)
+
+
+    def __iter__(self):
+        "Iterate keys in cache including expired items."
+        iterators = [iter(shard) for shard in self._shards]
+        return it.chain.from_iterable(iterators)
+
+
+    def __reversed__(self):
+        "Reverse iterate keys in cache including expired items."
+        iterators = [reversed(shard) for shard in self._shards]
+        return it.chain.from_iterable(reversed(iterators))
+
+
+    def __len__(self):
+        "Count of items in cache including expired items."
+        return sum(len(shard) for shard in self._shards)
+
+
+    def reset(self, key, value=ENOVAL):
+        """Reset `key` and `value` item from Settings table.
+
+        If `value` is not given, it is reloaded from the Settings
+        table. Otherwise, the Settings table is updated.
+
+        Settings attributes on cache objects are lazy-loaded and
+        read-only. Use `reset` to update the value.
+
+        Settings with the ``sqlite_`` prefix correspond to SQLite
+        pragmas. Updating the value will execute the corresponding PRAGMA
+        statement.
+
+        :param str key: Settings key for item
+        :param value: value for item (optional)
+        :return: updated value for item
+        :raises Timeout: if database timeout expires
+
+        """
+        for shard in self._shards:
+            while True:
+                try:
+                    result = shard.reset(key, value)
+                except Timeout:
+                    pass
+                else:
+                    break
+        return result
+
+
+    def deque(self, name):
+        """Return Deque with given `name` in subdirectory.
+
+        >>> cache = FanoutCache('/tmp/diskcache/fanoutcache')
+        >>> deque = cache.deque('test')
+        >>> deque.clear()
+        >>> deque.extend('abc')
+        >>> deque.popleft()
+        'a'
+        >>> deque.pop()
+        'c'
+        >>> len(deque)
+        1
+
+        :param str name: subdirectory name for Deque
+        :return: Deque with given name
+
+        """
+        _deques = self._deques
+
+        try:
+            return _deques[name]
+        except KeyError:
+            parts = name.split('/')
+            directory = op.join(self._directory, 'deque', *parts)
+            temp = Deque(directory=directory)
+            _deques[name] = temp
+            return temp
+
+
+    def index(self, name):
+        """Return Index with given `name` in subdirectory.
+
+        >>> cache = FanoutCache('/tmp/diskcache/fanoutcache')
+        >>> index = cache.index('test')
+        >>> index.clear()
+        >>> index['abc'] = 123
+        >>> index['def'] = 456
+        >>> index['ghi'] = 789
+        >>> index.popitem()
+        ('ghi', 789)
+        >>> del index['abc']
+        >>> len(index)
+        1
+        >>> index['def']
+        456
+
+        :param str name: subdirectory name for Index
+        :return: Index with given name
+
+        """
+        _indexes = self._indexes
+
+        try:
+            return _indexes[name]
+        except KeyError:
+            parts = name.split('/')
+            directory = op.join(self._directory, 'index', *parts)
+            temp = Index(directory)
+            _indexes[name] = temp
+            return temp

diskcache/memo.py

@@ -0,0 +1,105 @@
+"""Memoization utilities.
+
+"""
+
+from functools import wraps
+
+from .core import ENOVAL
+
+def memoize(cache, name=None, typed=False, expire=None, tag=None):
+    """Memoizing cache decorator.
+
+    Decorator to wrap callable with memoizing function using cache. Repeated
+    calls with the same arguments will lookup result in cache and avoid
+    function evaluation.
+
+    If name is set to None (default), the callable name will be determined
+    automatically.
+
+    If typed is set to True, function arguments of different types will be
+    cached separately. For example, f(3) and f(3.0) will be treated as distinct
+    calls with distinct results.
+
+    The original underlying function is accessible through the __wrapped__
+    attribute. This is useful for introspection, for bypassing the cache, or
+    for rewrapping the function with a different cache.
+
+    >>> from diskcache import FanoutCache
+    >>> cache = FanoutCache('/tmp/diskcache/fanoutcache')
+    >>> @cache.memoize(typed=True, expire=1, tag='fib')
+    ... def fibonacci(number):
+    ...     if number == 0:
+    ...         return 0
+    ...     elif number == 1:
+    ...         return 1
+    ...     else:
+    ...         return fibonacci(number - 1) + fibonacci(number - 2)
+    >>> print(sum(fibonacci(number=value) for value in range(100)))
+    573147844013817084100
+
+    Remember to call memoize when decorating a callable. If you forget, then a
+    TypeError will occur. Note the lack of parenthenses after memoize below:
+
+    >>> @cache.memoize
+    ... def test():
+    ...     pass
+    Traceback (most recent call last):
+        ...
+    TypeError: name cannot be callable
+
+    :param cache: cache to store callable arguments and return values
+    :param str name: name given for callable (default None, automatic)
+    :param bool typed: cache different types separately (default False)
+    :param float expire: seconds until arguments expire
+        (default None, no expiry)
+    :param str tag: text to associate with arguments (default None)
+    :return: callable decorator
+
+    """
+    if callable(name):
+        raise TypeError('name cannot be callable')
+
+    def decorator(function):
+        "Decorator created by memoize call for callable."
+        if name is None:
+            try:
+                reference = function.__qualname__
+            except AttributeError:
+                reference = function.__name__
+
+            reference = function.__module__ + reference
+        else:
+            reference = name
+
+        reference = (reference,)
+
+        @wraps(function)
+        def wrapper(*args, **kwargs):
+            "Wrapper for callable to cache arguments and return values."
+
+            key = reference + args
+
+            if kwargs:
+                key += (ENOVAL,)
+                sorted_items = sorted(kwargs.items())
+
+                for item in sorted_items:
+                    key += item
+
+            if typed:
+                key += tuple(type(arg) for arg in args)
+
+                if kwargs:
+                    key += tuple(type(value) for _, value in sorted_items)
+
+            result = cache.get(key, default=ENOVAL, retry=True)
+
+            if result is ENOVAL:
+                result = function(*args, **kwargs)
+                cache.set(key, result, expire=expire, tag=tag, retry=True)
+
+            return result
+
+        return wrapper
+
+    return decorator

diskcache/stampede.py

@@ -0,0 +1,78 @@
+"Stampede barrier implementation."
+
+import functools as ft
+import math
+import random
+import tempfile
+import time
+
+from .core import Cache, ENOVAL
+
+
+class StampedeBarrier(object):
+    """Stampede barrier mitigates cache stampedes.
+
+    Cache stampedes are also known as dog-piling, cache miss storm, cache
+    choking, or the thundering herd problem.
+
+    Based on research by Vattani, A.; Chierichetti, F.; Lowenstein, K. (2015),
+    Optimal Probabilistic Cache Stampede Prevention,
+    VLDB, pp. 886?897, ISSN 2150-8097
+
+    Example:
+
+    ```python
+    stampede_barrier = StampedeBarrier('/tmp/user_data', expire=3)
+
+    @stampede_barrier
+    def load_user_info(user_id):
+        return database.lookup_user_info_by_id(user_id)
+    ```
+
+    """
+    # pylint: disable=too-few-public-methods
+    def __init__(self, cache=None, expire=None):
+        if isinstance(cache, Cache):
+            pass
+        elif cache is None:
+            cache = Cache(tempfile.mkdtemp())
+        else:
+            cache = Cache(cache)
+
+        self._cache = cache
+        self._expire = expire
+
+    def __call__(self, func):
+        cache = self._cache
+        expire = self._expire
+
+        @ft.wraps(func)
+        def wrapper(*args, **kwargs):
+            "Wrapper function to cache function result."
+            key = (args, kwargs)
+
+            try:
+                result, expire_time, delta = cache.get(
+                    key, default=ENOVAL, expire_time=True, tag=True
+                )
+
+                if result is ENOVAL:
+                    raise KeyError
+
+                now = time.time()
+                ttl = expire_time - now
+
+                if (-delta * math.log(random.random())) < ttl:
+                    return result
+
+            except KeyError:
+                pass
+
+            now = time.time()
+            result = func(*args, **kwargs)
+            delta = time.time() - now
+            cache.set(key, result, expire=expire, tag=delta)
+
+            return result
+
+        return wrapper

mtwsgi.py

@@ -0,0 +1,60 @@
+'''
+WSGI-compliant HTTP server.  Dispatches requests to a pool of threads.
+https://github.com/RonRothman/mtwsgi
+'''
+
+from wsgiref.simple_server import WSGIServer, WSGIRequestHandler
+import multiprocessing.pool
+
+__all__ = ['ThreadPoolWSGIServer', 'make_server']
+
+import bottle
+
+class ThreadPoolWSGIServer(WSGIServer):
+    '''WSGI-compliant HTTP server.  Dispatches requests to a pool of threads.'''
+
+    def __init__(self, thread_count=None, *args, **kwargs):
+        '''If 'thread_count' == None, we'll use multiprocessing.cpu_count() threads.'''
+        WSGIServer.__init__(self, *args, **kwargs)
+        self.thread_count = thread_count
+        self.pool = multiprocessing.pool.ThreadPool(self.thread_count)
+
+    # Inspired by SocketServer.ThreadingMixIn.
+    def process_request_thread(self, request, client_address):
+        try:
+            self.finish_request(request, client_address)
+            self.shutdown_request(request)
+        except:
+            self.handle_error(request, client_address)
+            self.shutdown_request(request)
+
+    def process_request(self, request, client_address):
+        self.pool.apply_async(self.process_request_thread, args=(request, client_address))
+
+
+def make_server(host, port, app, thread_count=None, handler_class=WSGIRequestHandler):
+    '''Create a new WSGI server listening on `host` and `port` for `app`'''
+    httpd = ThreadPoolWSGIServer(thread_count, (host, port), handler_class)
+    httpd.set_app(app)
+    return httpd
+
+
+class MTServer(bottle.ServerAdapter):
+    def run(self, handler):
+        thread_count = self.options.pop('thread_count', None)
+        server = make_server(self.host, self.port, handler, thread_count, **self.options)
+        try:
+            server.serve_forever()
+        except KeyboardInterrupt:
+            server.server_close()  # Prevent ResourceWarning: unclosed socket
+            raise
+
+if __name__ == '__main__':
+    from wsgiref.simple_server import demo_app
+    httpd = make_server('', 8000, demo_app)
+    sa = httpd.socket.getsockname()
+    print "Serving HTTP on", sa[0], "port", sa[1], "..."
+    import webbrowser
+    webbrowser.open('http://localhost:8000/xyz?abc')
+    httpd.serve_forever()
+

playstreamproxy.py

@@ -4,21 +4,19 @@
 StreamProxy daemon (based on Livestream daemon)
 Provides API to ContetSources + stream serving to play via m3u8 playlists
 """
-import os
-import sys
-import time
+import os, sys, time, re, json
+import ConfigParser
 import atexit
-import re, json
-import binascii
-
 from signal import SIGTERM
 
 from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler
 from SocketServer import ThreadingMixIn
+
 from urllib import unquote, quote
 import urllib,urlparse
-#import cookielib,urllib2
 import requests
+
+#import cookielib,urllib2
 from ContentSources import ContentSources
 from sources.SourceBase import stream_type
 import util
@@ -32,8 +30,9 @@ except:
 
 HOST_NAME = ""
 PORT_NUMBER = 8880
-DEBUG = False
+DEBUG = True
 DEBUG2 = False
+REDIRECT = True
 
 SPLIT_CHAR = "~"
 SPLIT_CODE = "%7E"
@@ -59,6 +58,18 @@ sessions = {}
 cfg_file = "streams.cfg"
 sources = ContentSources("", cfg_file)
 
+config = ConfigParser.ConfigParser()
+proxy_cfg_file = os.path.join(cur_directory, "playstreamproxy.cfg")
+if not os.path.exists(proxy_cfg_file):
+    config.add_section("default")
+    config["default"]["port"] = 80
+    config["default"]["host"] = localhost
+    config["default"]["redirect"] = True
+    config.write(open.file(proxy_cfg_file, "w"))
+else:
+    config.read(proxy_cfg_file)
+
+
 class StreamHandler(BaseHTTPRequestHandler):
 
     def do_HEAD(self):
@@ -130,6 +141,20 @@ class StreamHandler(BaseHTTPRequestHandler):
         self.wfile.write(open("offline.mp4", "rb").read())
         #self.wfile.close()
 
+    def redirect_source(self, urlp):
+        cmd, data, headers, qs = streamproxy_decode3(urlp)
+        streams = sources.get_streams(data)
+        if not streams:
+            self.write_error(500)  # TODO
+            return
+        stream = streams[0]
+        url = stream["url"]
+        headers = stream["headers"] if "headers" in stream else headers0
+        self.send_response(307)
+        self.send_header("Location", url)
+        self.end_headers()
+
+
     def fetch_source(self, urlp):
         cmd, data, headers, qs = streamproxy_decode3(urlp)
         if DEBUG:
@@ -160,6 +185,14 @@ class StreamHandler(BaseHTTPRequestHandler):
             else:
                 url = urlp.replace(base_data, slinks[base_data]["base_url"])
                 if DEBUG: print "Existing new link", url
+        if REDIRECT:
+            print "-->redirect to: " + url
+            self.send_response(307)
+            self.send_header("Location", url)
+            self.end_headers()
+            #self.wfile.close()
+            return
+
         headers2 = headers if headers else self.headers.dict
         headers2 = del_headers(headers2, ["host"])
         r = self.get_page_ses(url,ses,True,headers = headers2)
@@ -240,7 +273,10 @@ class StreamHandler(BaseHTTPRequestHandler):
 class ThreadedHTTPServer(ThreadingMixIn, HTTPServer):
     """Handle requests in a separate thread."""
 
-def start(host = HOST_NAME, port = PORT_NUMBER):
+def start(host = HOST_NAME, port = PORT_NUMBER, redirect=None):
+    global REDIRECT
+    if redirect:
+        REDIRECT = redirect
     httpd = ThreadedHTTPServer((host, port), StreamHandler)
     print time.asctime(), "Server Starts - %s:%s" % (HOST_NAME, PORT_NUMBER)
     try:

playstreamproxy2.py

@@ -0,0 +1,152 @@
+#!/bin/env python
+# -*- coding: utf-8 -*-
+"""
+PlayStreamProxy server
+Provides API to ContetSources + stream serving to play via m3u8 playlists
+"""
+
+import os, sys, time
+import urllib,urlparse
+from urllib import unquote, quote
+#import cookielib,urllib2
+import requests
+import re, json
+from diskcache import Cache
+
+import bottle
+from bottle import Bottle, hook, response, route, request, run
+
+#if not sys.platform == 'win32':
+import daemonize
+
+from ContentSources import ContentSources
+from sources.SourceBase import stream_type
+import util
+from util import streamproxy_decode3, streamproxy_encode3
+
+try:
+    from requests.packages.urllib3.exceptions import InsecureRequestWarning
+    requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
+except:
+    pass
+
+# Default arguments
+SERVER = "wsgiref"
+PORT_NUMBER = 8880
+DEBUG = False
+
+cunicode = lambda s: s.decode("utf8") if isinstance(s, str) else s
+cstr = lambda s: s.encode("utf8") if isinstance(s, unicode) else s
+headers2dict = lambda  h: dict([l.strip().split(": ") for l in h.strip().splitlines()])
+
+headers0 = headers2dict("""
+User-Agent: GStreamer souphttpsrc libsoup/2.52.2
+icy-metadata: 1
+""")
+headers0_ = headers2dict("""
+Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8
+User-Agent: Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.102 Safari/537.36
+""")
+
+cur_directory = os.path.dirname(os.path.realpath(__file__))
+slinks = {}
+sessions = {}
+
+cfg_file = "streams.cfg"
+sources0 = ContentSources("", cfg_file)
+
+cache_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), "cache")
+s0 = Cache(cache_dir)
+s0["n"] = 0
+s0["sources"] = sources0
+
+app = Bottle()
+
+@app.hook('before_request')
+def set_globals():
+    global s, sources, cmd, data, headers, qs
+    cmd, data, headers, qs = streamproxy_decode3(request.url)
+    cache_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), "cache")
+    s = Cache(cache_dir)
+    sources = s["sources"]
+
+# http://localhost:8880/playstream/ltc%3A%3Acontent%2Flive-streams%2F101%3Finclude%3Dquality/
+#@app.route('/')
+@app.route('/playstream/<url:re:.*>')
+def fetch_source(url):
+    global s, sources, cmd, data, headers, qs
+    return 'cmd=%s, data=%s, qs=%s'% (cmd, data, str(qs))
+
+
+
+####################################################################
+# Run WSGI server
+def start(server,port):
+    options = {}
+    if server == "mtwsgi":
+        import mtwsgi
+        server = mtwsgi.MTServer
+        options = {"thread_count": 3,}
+
+    run(app=app,server=server, host='0.0.0.0',
+            port=port,
+            #reloader=True,
+            quiet=False,
+            plugins=None,
+            debug=True,
+            #thread_count=3,
+            config=None,
+            **options)
+
+def print_headers(headers):
+    for h in headers:
+        print "%s: %s"%(h,headers[h])
+
+def del_headers(headers0,tags):
+    headers = headers0.copy()
+    for t in tags:
+        if t in headers:
+            del headers[t]
+        if t.lower() in headers:
+            del headers[t.lower()]
+    return headers
+
+def hls_base(url):
+    base = url.split("?")[0]
+    base = "/".join(base.split("/")[0:3])+ "/"
+    rest = url.replace(base, "")
+    return base
+
+def hls_base2(url):
+    base = url.split("?")[0]
+    base = "/".join(base.split("/")[0:-1])+ "/"
+    rest = url.replace(base, "")
+    return base
+
+
+if __name__ == '__main__':
+    if len(sys.argv) > 1:
+        pid = "/var/run/playstreamproxy2.pid"
+        daemon = daemonize.Daemon(start, pid)
+        server = sys.argv[2] if len(sys.argv) > 2 else SERVER
+        port = sys.argv[3] if len(sys.argv) > 3 else PORT_NUMBER
+        if "start" == sys.argv[1]:
+            daemon.start(server,port)
+            daemon.is_running()
+        elif "stop" == sys.argv[1]:
+            daemon.stop()
+        elif "restart" == sys.argv[1]:
+            daemon.restart()
+            daemon.is_running()
+        elif "manualstart" == sys.argv[1]:
+            start(server,port)
+        elif "status" == sys.argv[1]:
+            daemon.is_running()
+        else:
+            print "Unknown command"
+            print "usage: %s start|stop|restart|manualstart" % sys.argv[0]
+            sys.exit(2)
+        sys.exit(0)
+    else:
+        print "usage: %s start|stop|restart|manualstart" % sys.argv[0]
+        sys.exit(2)

sources/streams.cfg

@@ -23,6 +23,7 @@ SerialGURU.ru|serialguru::home|http://serialguru.ru/images/xlogo_new.png.pagespe
 MoviePlace.lv|movieplace::home|movieplace.png|Movieplace.lv - filmas latviesu valodā
 Engima2|enigma2::home|enigma2.png|Get streams from engima2 sat receiver
 test|http://mfe1-iptv.baltcom.lv/dash/CH_7_50.ism/playlist.mpd||testa strīms
+Shortcut.lv (lattelecom.tv)|ltc2::home|shortcut.png|Shortcut.lv (lattelecom.tv) satura skatīšanās
 
 [my_tv]
 My Tv

test_url.py

@@ -0,0 +1,23 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+from util import streamproxy_encode3, streamproxy_decode3
+dt2ts = lambda dt: int(time.mktime(dt.timetuple()))
+ts2dt = lambda ts: datetime.datetime.fromtimestamp(ts)
+import time, datetime, pytz
+from datetime import datetime as dt
+
+# "20190302203000 +0200" panorāma
+print dt2ts(dt.now())
+print dt2ts(dt(2019, 3, 2, 20, 30))
+#print urlp
+# ltc::content/live-streams/101?include=quality
+print streamproxy_encode3("playstream", "ltc::content/catchup/101?start=1551551400")
+# http://localhost:8880/playstream/ltc%3A%3Acontent%2Fcatchup%2F101%3Fstart%3D1551551400/
+
+print streamproxy_encode3("playstream", "replay::tiesraide/ltv1")
+print streamproxy_encode3("playstream", "tvdom::tiesraides/ltv1/")
+print streamproxy_encode3("playstream", "tvplay::asset/10311641")
+print streamproxy_encode3("playstream", "xtv::rigatv24/video/Zw4pVPqr7OX-festivals_mainam_pasauli_sakam_ar_sevi")
+print streamproxy_encode3("playstream", "iplayer::episodes/b094f49s")
+#cmd, data, headers, qs = streamproxy_decode3(urlp)

util.py

@@ -260,11 +260,12 @@ def streamproxy_encode3(cmd, data, proxy_url=None, headers=None, qs=None):
 
 def streamproxy_decode3(urlp):
     m = re.search("http://[^/]+",urlp)
-    path = urlp.replace(m.group(0),"") if m else urlp
+    urlp = urlp.replace(m.group(0),"") if m else urlp
+    path = urlp.split("?")[0]
     cmd = path.split("/")[1]
-    data = path.split("/")[2].split("?")[0]
+    data = "/".join(path.split("/")[2:])
     data = urllib.unquote_plus(data)
-    qs = path.split("?")[1] if "?" in path else ""
+    qs = urlp.split("?")[1] if "?" in urlp else ""
     qs2 = {}
     headers = {}
     if qs: