torrent.py

bittorrent source by python. please enjoy

# The contents of this file are subject to the BitTorrent Open Source License
# Version 1.1 (the License).  You may not copy or use this file, in either
# source code or executable form, except in compliance with the License.  You
# may obtain a copy of the License at http://www.bittorrent.com/license/.
#
# Software distributed under the License is distributed on an AS IS basis,
# WITHOUT WARRANTY OF ANY KIND, either express or implied.  See the License
# for the specific language governing rights and limitations under the
# License.

# Written by Bram Cohen and Uoti Urpala

from __future__ import division
from __future__ import generators

import os
import gc
import sys
import time
import errno
import shutil
import random
import socket
import cPickle
import logging
import urlparse
import itertools
import traceback
from cStringIO import StringIO
from BitTorrent.hash import sha
from BitTorrent.translation import _
from BitTorrent.NamedMutex import NamedMutex

import BitTorrent.stackthreading as threading

from BitTorrent.platform import bttime, is_path_too_long
from BitTorrent.platform import decode_from_filesystem, get_filesystem_encoding
from BitTorrent.ConnectionManager import ConnectionManager
from BitTorrent import PeerID
from BitTorrent.defer import Deferred, ThreadedDeferred
from BitTorrent.yielddefer import launch_coroutine, _wrap_task
from BitTorrent.TorrentStats import TorrentStats
from BitTorrent.RateMeasure import RateMeasure
from BitTorrent.PiecePicker import PiecePicker
from BitTorrent.Rerequester import Rerequester, DHTRerequester
from BitTorrent.NewRateLimiter import MultiRateLimiter as RateLimiter
from BitTorrent.CurrentRateMeasure import Measure
from BitTorrent.Storage import Storage, FilePool
from BitTorrent.HTTPConnector import URLage
from BitTorrent.StorageWrapper import StorageWrapper
from BitTorrent.Upload import Upload
from BitTorrent.MultiDownload import MultiDownload
from BitTorrent import version
from BitTorrent import BTFailure, UserFailure
from BitTorrent.prefs import Preferences
#from BitTorrent.bencode import bencode


from khashmir import const


class Feedback(object):
    """Inidivual torrents (Torrent) perform callbacks regarding
       changes of state to the rest of the program via a Feedback
       object."""

    def finished(self, torrent):
        pass

    def failed(self, torrent):
        pass

    def error(self, torrent, level, text):
        pass

    def exception(self, torrent, text):
        self.error(torrent, logging.CRITICAL, text)

    def started(self, torrent):
        pass


class FeedbackMultiplier(object):
    def __init__(self, *a):
        self.chain = list(a)
    def __getattr__(self, attr):
        def multiply_calls(*a, **kw):
            exc_info = None
            for x in self.chain:
                try:
                    getattr(x, attr)(*a, **kw)
                except:
                    exc_info = sys.exc_info()
            if exc_info:
                raise exc_info[0], exc_info[1], exc_info[2]
        return multiply_calls


class Torrent(object):
    """Represents a single file transfer or transfer for a batch of files
       in the case of a batch torrent.  During the course of a single
       transfer, a Torrent may have many different connections to peers."""

    STATES = ["created", "initializing", "initialized", "running",
              "finishing", "failed"]
    POLICIES = ["stop", "start", "auto"]
    PRIORITIES = ["low", "normal", "high"]

    def __init__(self, metainfo, working_path, destination_path, config,
                 data_dir, rawserver, choker,
                 singleport_listener, ratelimiter, total_downmeasure,
                 filepool, dht, feedback, log_root,
                 hidden=False, is_auto_update=False):
        # The passed working path and destination_path should be filesystem
        # encoded or should be unicode if the filesystem supports unicode.
        fs_encoding = get_filesystem_encoding()
        assert fs_encoding == None and isinstance(working_path,unicode) \
               or isinstance(working_path,str)
        assert fs_encoding == None and isinstance(destination_path,unicode) \
               or isinstance(destination_path,str)

        self.state = "created"
        self.data_dir = data_dir
        self.feedback = FeedbackMultiplier(feedback)
        self.finished_this_session = False
        self._rawserver = rawserver
        self._singleport_listener = singleport_listener
        self._ratelimiter = ratelimiter
        self._filepool = filepool
        self._dht = dht
        self._picker = None
        self._choker = choker
        self._storage = None
        self._storagewrapper = None
        self._ratemeasure = None
        self._upmeasure = None
        self._downmeasure = None
        self._total_downmeasure = total_downmeasure
        self._connection_manager = None
        self._rerequest = None
        self._statuscollector = None
        self._announced = False
        self._listening = False
        self.reserved_ports = []
        self.reported_port = None
        self._myfiles = None
        self._last_myfiles = None
        self.total_bytes = None
        self._doneflag = threading.Event()
        self.finflag = threading.Event()
        self._contfunc = None
        self._activity = (_("Initial startup"), 0)
        self._pending_file_priorities = []
        self._mutex = None
        self.time_created = bttime()
        self.time_started = None

        self.metainfo = metainfo
        self.infohash = metainfo.infohash

        self.log_root = log_root
        self.logger = logging.getLogger(log_root + '.' + repr(self.infohash))
        self.logger.setLevel(logging.DEBUG)

        self.total_bytes = metainfo.total_bytes
        if not metainfo.reported_errors:
            metainfo.show_encoding_errors(self._error)

        self.config = Preferences(config)#, persist_callback=self._dump_torrent_config)
        self.working_path = working_path #sets in config. See _set_working_path

        self.destination_path = destination_path # sets in config.
        self.priority = "normal"
        self.policy = "auto"

        self.hidden = hidden #sets in config
        self.is_auto_update = is_auto_update #sets in config

        self._completed = False
        self.config['finishtime'] = 0
        self.uptotal = 0
        self.uptotal_old = 0
        self.downtotal = 0
        self.downtotal_old = 0
        self.context_valid = True

    def update_config(self, config):
        self.config.update(config)

        d = self.config.get('file_priorities', {})
        for k, v in d.iteritems():
            self.set_file_priority(k, v)

        if self.policy not in self.POLICIES:
            self.policy = "auto"

        if self.priority not in self.PRIORITIES:
            self.priority = "normal"

    def _set_state(self, value):
        assert value in self.STATES
        self._state = value
    def _get_state(self):
        return self._state
    state = property(_get_state, _set_state)

    def _set_policy(self, value):
        assert value in self.POLICIES
        self.config['policy'] = value
    def _get_policy(self):
        return self.config['policy']
    policy = property(_get_policy, _set_policy)

    def _set_priority(self, value):
        assert value in self.PRIORITIES
        self.config['priority'] = value
    def _get_priority(self):
        return self.config['priority']
    priority = property(_get_priority, _set_priority)

    def _set_hidden(self, value):
        self.config['hidden'] = value
    def _get_hidden(self):
        return self.config['hidden']
    hidden = property(_get_hidden, _set_hidden)

    def _set_is_auto_update(self, value):
        self.config['is_auto_update'] = value
    def _get_is_auto_update(self):
        return self.config['is_auto_update']
    is_auto_update = property(_get_is_auto_update, _set_is_auto_update)

    def _set_completed(self, val):
        self._completed = val
        self.config['finishtime'] = bttime()
    def _get_completed(self):
        return self._completed
    completed = property(_get_completed, _set_completed)

    def _get_finishtime(self):
        return self.config['finishtime']
    finishtime = property(_get_finishtime)

    def _set_destination_path(self, value):
        # The following assertion will not always work.  Consider
        # Torrent.py: self.working_path = self.destination_path
        # This assignment retrieves a unicode path from
        # config['destination_path'].
        #assert isinstance(value,str) # assume filesystem encoding.
        #
        # The following if statement is not necessary because config here
        # is not really a config file, but rather state that is pickled when
        # the Torrent shuts down.
        #if isinstance(value, str):
        #    value = decode_from_filesystem(value)
        self.config['destination_path'] = value
    def _get_destination_path(self):
        return self.config['destination_path']
    destination_path = property(_get_destination_path, _set_destination_path)

    def _set_working_path(self, value):
        # See comments for _set_destination_path.
        self.config['working_path'] = value
    def _get_working_path(self):
        return self.config['working_path']
    working_path = property(_get_working_path, _set_working_path)

    def __cmp__(self, other):
        return cmp(self.metainfo.infohash, other.metainfo.infohash)

    def is_initialized(self):
        return self.state not in ["created", "initializing", "failed"]

    def is_running(self):
        return self.state == "running"

    def is_context_valid(self):
        return self.context_valid

    def _context_wrap(self, _f, *a, **kw):
        # this filters out calls
        # to an invalid torrent
        # sloppy technique
        if not self.context_valid:
            return
        try:
            _f(*a, **kw)
        except KeyboardInterrupt:
            raise
        except:
            self.got_exception(*sys.exc_info())

    # these wrappers add _context_wrap to the chain, so that calls on a dying
    # object are filtered, and errors on a valid call are logged.
    def add_task(self, delay, func, *a, **kw):
        self._rawserver.add_task(delay, self._context_wrap,
                                 func, *a, **kw)
    def external_add_task(self, delay, func, *a, **kw):
        self._rawserver.external_add_task(delay, self._context_wrap,
                                          func, *a, **kw)

    def _register_files(self):
        if self.metainfo.is_batch:
            myfiles = [os.path.join(self.destination_path, f) for f in
                       self.metainfo.files_fs]
        else:
            myfiles = [self.destination_path, ]

        for filename in myfiles:
            if is_path_too_long(filename):
                raise BTFailure("Filename path exceeds platform limit: %s" % filename)

        # if the destination path contains any of the files in the torrent
        # then use the destination path instead of the working path.
        if len([x for x in myfiles if os.path.exists(x)]) > 0:
            self.working_path = self.destination_path
        else:
            if self.metainfo.is_batch:
                myfiles = [os.path.join(self.working_path, f) for f in
                           self.metainfo.files_fs]
            else:
                myfiles = [self.working_path, ]

        assert self._myfiles == None, '_myfiles should be None!'
        self._filepool.add_files(myfiles, self)
        self._myfiles = myfiles

    def _build_url_mapping(self):
        # TODO: support non [-1] == '/' urls
        url_suffixes = []

        if self.metainfo.is_batch:
            for filename in self.metainfo.orig_files:
                path = '%s/%s' % (self.metainfo.name, filename)
                # am I right that orig_files could have windows paths?
                path = path.replace('\\', '/')
                url_suffixes.append(path)
        else:
            url_suffixes = [self.metainfo.name, ]
        self._url_suffixes = url_suffixes

        total = 0
        piece_size = self.metainfo.piece_length
        self._urls = zip(self._url_suffixes, self.metainfo.sizes)

    def _unregister_files(self):
        if self._myfiles is not None:
            self._filepool.remove_files(self._myfiles)
            self._last_myfiles = self._myfiles
            self._myfiles = None

    def initialize(self):
        self.context_valid = True
        assert self.state in ["created", "failed", "finishing"]
        self.state = "initializing"
        df = launch_coroutine(_wrap_task(self.add_task), self._initialize)
        df.addErrback(lambda e : self.got_exception(*e))