<commit_before>from twisted.internet import defer

from scrapy.utils.misc import mustbe_deferred, defer_result
from scrapy.core import log
from scrapy.core.engine import scrapyengine
from scrapy.core.exceptions import DropItem, NotConfigured
from scrapy.http import Request
from scrapy.spider import spiders
from scrapy.conf import settings


class DomainInfo(object):
    def __init__(self, domain):
        self.domain = domain
        self.spider = spiders.fromdomain(domain)
        self.downloading = {}
        self.downloaded = {}
        self.waiting = {}
        self.extra = {}


class MediaPipeline(object):
    def __init__(self):
        self.cache = {}

    def open_domain(self, domain):
        self.cache[domain] = DomainInfo(domain)

    def close_domain(self, domain):
        del self.cache[domain]

    def process_item(self, domain, response, item):
        info = self.cache[domain]
        urls = self.get_urls_from_item(item, info)
        assert urls is None or hasattr(urls, '__iter__'), \
                'get_urls_from_item should return None or iterable'

        def _bugtrap(_failure, request):
            log.msg('MediaPipeline Unhandled ERROR in %s: %s' % (request, _failure), log.ERROR, domain=domain)

        lst = []
        for url in urls or ():
            request = url if isinstance(url, Request) else Request(url=url)
            dfd = self._enqueue(request, info)
            dfd.addCallbacks(
                    callback=self.new_item_media,
                    callbackArgs=(item, request, info),
                    errback=self.failed_item_media,
                    errbackArgs=(item, request, info),
                    )
            dfd.addErrback(_bugtrap, request)
            lst.append(dfd)

        dlst = defer.DeferredList(lst, consumeErrors=False)
        dlst.addBoth(lambda _: self.item_completed(item, info))
        return dlst

    def _enqueue(self, request, info):
        result = self.media_to_download(request, info)
        if result is not None:
            return defer_result(result)

        fp = request.fingerprint()
        if fp in info.downloaded:
            return defer_result(info.downloaded[fp])

        if fp not in info.downloading:
            self._download(request, info, fp)

        wad = defer.Deferred()
        waiting = info.waiting.setdefault(fp, []).append(wad)
        return wad

    def _download(self, request, info, fp):
        def _bugtrap(_failure, request):
            log.msg('MediaPipeline Unhandled ERROR in %s: %s' % (request, _failure), log.ERROR, domain=info.domain)

        dwld = mustbe_deferred(self.download, request, info)
        dwld.addCallbacks(
                callback=self.media_downloaded,
                callbackArgs=(request, info),
                errback=self.media_failure,
                errbackArgs=(request, info),
                )
        dwld.addErrback(_bugtrap, request)
        dwld.addBoth(self._downloaded, info, fp)
        info.downloading[fp] = (request, dwld)


    def _downloaded(self, result, info, fp):
        info.downloaded[fp] = result # cache result
        waiting = info.waiting[fp] # client list
        del info.waiting[fp]
        del info.downloading[fp]
        for wad in waiting:
            defer_result(result).chainDeferred(wad)


    ### Overradiable Interface
    def download(self, request, info):
        """ Defines how to request the download of media

        Default gives high priority to media requests and use scheduler,
        shouldn't be necessary to override.

        This methods is called only if result for request isn't cached,
        request fingerprint is used as cache key.

        """
        return scrapyengine.schedule(request, info.spider, priority=0)
<commit_msg>move bugtraps to avoid silence of bugs in media_downloaded or media_failed<commit_after>from twisted.internet import defer

from scrapy.utils.misc import mustbe_deferred, defer_result
from scrapy.core import log
from scrapy.core.engine import scrapyengine
from scrapy.core.exceptions import DropItem, NotConfigured
from scrapy.http import Request
from scrapy.spider import spiders
from scrapy.conf import settings


class DomainInfo(object):
    def __init__(self, domain):
        self.domain = domain
        self.spider = spiders.fromdomain(domain)
        self.downloading = {}
        self.downloaded = {}
        self.waiting = {}
        self.extra = {}


class MediaPipeline(object):
    def __init__(self):
        self.cache = {}

    def open_domain(self, domain):
        self.cache[domain] = DomainInfo(domain)

    def close_domain(self, domain):
        del self.cache[domain]

    def process_item(self, domain, response, item):
        info = self.cache[domain]
        urls = self.get_urls_from_item(item, info)
        assert urls is None or hasattr(urls, '__iter__'), \
                'get_urls_from_item should return None or iterable'

        def _bugtrap(_failure, request):
            log.msg('Unhandled ERROR in MediaPipeline.{new,failed}_item_media for %s: %s' % (request, _failure), log.ERROR, domain=domain)

        lst = []
        for url in urls or ():
            request = url if isinstance(url, Request) else Request(url=url)
            dfd = self._enqueue(request, info)
            dfd.addCallbacks(
                    callback=self.new_item_media,
                    callbackArgs=(item, request, info),
                    errback=self.failed_item_media,
                    errbackArgs=(item, request, info),
                    )
            dfd.addErrback(_bugtrap, request)
            lst.append(dfd)

        dlst = defer.DeferredList(lst, consumeErrors=False)
        dlst.addBoth(lambda _: self.item_completed(item, info))
        return dlst

    def _enqueue(self, request, info):
        result = self.media_to_download(request, info)
        if result is not None:
            return defer_result(result)

        fp = request.fingerprint()
        if fp in info.downloaded:
            return defer_result(info.downloaded[fp])

        if fp not in info.downloading:
            self._download(request, info, fp)

        wad = defer.Deferred()
        waiting = info.waiting.setdefault(fp, []).append(wad)
        return wad

    def _download(self, request, info, fp):
        dwld = mustbe_deferred(self.download, request, info)
        dwld.addCallbacks(
                callback=self.media_downloaded,
                callbackArgs=(request, info),
                errback=self.media_failure,
                errbackArgs=(request, info),
                )
        dwld.addBoth(self._downloaded, info, fp)
        info.downloading[fp] = (request, dwld)

        def _bugtrap(_failure):
            log.msg('Unhandled ERROR in MediaPipeline._downloaded: %s' % (_failure), log.ERROR, domain=info.domain)
        dwld.errback(_bugtrap)


    def _downloaded(self, result, info, fp):
        info.downloaded[fp] = result # cache result
        waiting = info.waiting[fp] # client list
        del info.waiting[fp]
        del info.downloading[fp]
        for wad in waiting:
            defer_result(result).chainDeferred(wad)


    ### Overradiable Interface
    def download(self, request, info):
        """ Defines how to request the download of media

        Default gives high priority to media requests and use scheduler,
        shouldn't be necessary to override.

        This methods is called only if result for request isn't cached,
        request fingerprint is used as cache key.

        """
        return scrapyengine.schedule(request, info.spider, priority=0)

{'fraction_non_alphanumeric': 0.09137523008151459, 'fraction_numerical': 0.00026295030239284776, 'mean_word_length': 2.351101321585903, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 36, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2003201', 'n_tokens_mistral': 2215, 'n_tokens_neox': 2118, 'n_words': 569}