<commit_before>#!/usr/bin/env python
# -*- coding: utf-8 -*-

import time, random, json, bson
from urllib import urlencode
import subprocess
from txjsonrpc import jsonrpclib
from txjsonrpc.jsonrpc import Introspection
from txjsonrpc.web import jsonrpc
from twisted.web import server
from twisted.python import log as logger
from twisted.internet import reactor, defer
from twisted.internet.task import LoopingCall
from twisted.internet.threads import deferToThread
from twisted.internet.defer import inlineCallbacks, returnValue as returnD
from twisted.internet.endpoints import TCP4ClientEndpoint
from twisted.internet.error import DNSLookupError, ConnectionRefusedError
from twisted.application.internet import TCPServer
from twisted.application.service import Application
from twisted.web.http_headers import Headers
from twisted.web.client import getPage, Agent, ProxyAgent, HTTPClientFactory, _HTTP11ClientFactory
HTTPClientFactory.noisy = False
_HTTP11ClientFactory.noisy = False
from hyphe_backend import processor
from hyphe_backend.lib import config_hci, urllru, gexf, user_agents
from hyphe_backend.lib.utils import *
from hyphe_backend.lib.mongo import MongoDB, sortdesc
from hyphe_backend.lib.corpus import CorpusFactory
from hyphe_backend.memorystructure import MemoryStructure as ms, constants as ms_const


# MAIN CORE API

class Core(jsonrpc.JSONRPC):

    addSlash = True

    def __init__(self):
        jsonrpc.JSONRPC.__init__(self)
        self.db = MongoDB(config['mongo-scrapy'])
        self.msclients = CorpusFactory(config['memoryStructure']['thrift.host'],
          port_range = config['memoryStructure']['thrift.portrange'],
          max_ram = config['memoryStructure']['thrift.max_ram'],
          loglevel = config['memoryStructure']['log.level'])
        self.corpora = {}
        self.crawler = Crawler(self)
        self.store = Memory_Structure(self)
        reactor.addSystemEventTrigger('before', 'shutdown', self.close)
        self.keepalive_default = None

    @inlineCallbacks
    def activate_monocorpus(self):
        yield self.start_corpus(create_if_missing=True)
        self.keepalive_default = LoopingCall(self.jsonrpc_ping, DEFAULT_CORPUS)
        self.keepalive_default.start(300, False)

    @inlineCallbacks
    def close(self):
        if not config["MULTICORPUS"] and self.keepalive_default and self.keepalive_default.running:
            self.keepalive_default.stop()
        for corpus in self.corpora.keys():
            yield self.stop_corpus(corpus, quiet=True)
        yield self.db.close()
        self.msclients.stop()

  # OVERWRITE JSONRPC REQUEST HANDLING

    def render(self, request):
        request.setHeader("Access-Control-Allow-Origin", "*")
        from_ip = ""
        if request.getHeader("x-forwarded-for"):
            from_ip = " from %s" % request.getHeader("x-forwarded-for")
        if config['DEBUG']:
            logger.msg(request.content.read(), system="DEBUG - QUERY%s" % from_ip)
# TODO   catch corpus arg here and return corpus_error if needed
        return jsonrpc.JSONRPC.render(self, request)

    def _cbRender(self, result, request, id, version):
        if config['DEBUG'] == 2:
            txt = jsonrpclib.dumps(result, id=id, version=2.0)
            logger.msg("%s%s" % (txt[:1000], " ... [%d cars truncated]" % (len(txt)-1000) if len(txt) > 1000 else ''), system="DEBUG - ANSWER")
        return jsonrpc.JSONRPC._cbRender(self, result, request, id, version)

  # CORPUS HANDLING

    def jsonrpc_test_corpus(self, corpus=DEFAULT_CORPUS, msg=None):
        res = {
          "corpus_id": corpus,
          "ready": False,
          "status": self.msclients.status_corpus(corpus, simplify=True),
        }
        if res["status"] == "ready":
            res["ready"] = True
        elif res["status"] == "error":
            res["message"] = self.msclients.corpora[corpus].error
        elif res["status"] == "stopped":
            res["message"] = "Corpus is not started"
        if res["status"] == "starting":
            res["message"] = "Corpus is starting, please retry in a bit"
        elif msg:
            res["message"] = msg
        return format_result(res)

    @inlineCallbacks
    def jsonrpc_list_corpus(self):
        res = {}
        corpora = yield self.db.list_corpus()
        for corpus in corpora:
            corpus["password"] = (corpus["password"] != "")
            corpus.update(self.jsonrpc_test_corpus(corpus.pop('_id'))["result"])
            res[corpus["corpus_id"]] = corpus
        returnD(format_result(res))

    def corpus_ready(self, corpus):
        return corpus in self.corpora and self.msclients.test_corpus(corpus)

    def corpus_error(self, corpus=None):
        if not corpus:
            return format_error("Too many instances running already, please try again later")
        reactor.callLater(0, self.stop_corpus, corpus, quiet=True)
        return format_error(self.jsonrpc_test_corpus(corpus)["result"])

    def factory_full(self):
        return self.msclients.ram_free < 256 or not self.msclients.ports_free

    def jsonrpc_create_corpus(self, name=DEFAULT_CORPUS, password="", options=None):
        return self.create_corpus(name, password, options=options)

    @inlineCallbacks
    def create_corpus(self, name=DEFAULT_CORPUS, password="", options=None, noloop=False, quiet=False):
        if self.factory_full():
            returnD(self.corpus_error())

        corpus = clean_corpus_id(name)
        corpus_idx = 1
        existing = yield self.db.get_corpus(corpus)
        while existing:
            corpus = "%s-%s" % (clean_corpus_id(name), corpus_idx)
            corpus_idx += 1
            existing = yield self.db.get_corpus(corpus)

        self.corpora[corpus] = {"name": name}

        yield self.db.add_corpus(corpus, name, password, options)
        try:
            res = yield self.crawler.deploy_spider(corpus)
        except:
            returnD(format_error("Could not deploy corpus' scrapyd spider"))
        if not res:
            returnD(res)
        res = self.store.ensureDefaultCreationRuleExists(corpus, quiet=quiet)
        if not res:
            returnD(res)
        res = yield self.start_corpus(corpus, password=password, noloop=noloop, quiet=quiet)
        returnD(handle_standard_results(res))

    def jsonrpc_start_corpus(self, corpus=DEFAULT_CORPUS, password=""):
        return self.start_corpus(corpus, password)

    @inlineCallbacks
    def start_corpus(self, corpus=DEFAULT_CORPUS, password="", noloop=False, quiet=False, create_if_missing=False):
        if self.corpus_ready(corpus) or self.msclients.status_corpus(corpus, simplify=True) == "starting":
            returnD(self.jsonrpc_test_corpus(corpus, msg="Corpus already ready"))

        corpus_conf = yield self.db.get_corpus(corpus)
        if not corpus_conf:
            if create_if_missing:
                res = yield self.create_corpus(corpus, password, noloop=noloop, quiet=quiet)
                returnD(res)
            returnD(format_error("No corpus existing with ID %s, please create it first!" % corpus))
        if corpus_conf['password'] and corpus_conf['password'] != salt(password):
            returnD(format_error("Wrong auth for password-protected corpus %s" % corpus))
        if self.factory_full():
            returnD(self.corpus_error())

        res = self.msclients.start_corpus(corpus, quiet, ram=corpus_conf['ram'])
        if not res:
            returnD(format_error(self.jsonrpc_test_corpus(corpus)["result"]))
        self.corpora[corpus] = {
          "name": corpus_conf["name"],
          "ram": corpus_conf["ram"],
          "total_webentities": corpus_conf['total_webentities'],
          "crawls": corpus_conf['total_crawls'],
          "crawls_running": 0,
          "crawls_pending": 0,
          "pages_found": corpus_conf['total_pages'],
          "pages_crawled": corpus_conf['total_pages_crawled'],
          "pages_queued": 0,
          "links_found": 0,
          "last_index_loop": corpus_conf['last_index_loop'],
          "last_links_loop": corpus_conf['last_links_loop'],
          "index_loop": LoopingCall(self.store.index_batch_loop, corpus),
          "jobs_loop": LoopingCall(self.refresh_jobs, corpus)
        }
        if not noloop:
            reactor.callLater(5, self.corpora[corpus]['jobs_loop'].start, 1, False)
        yield self.store._start_loop(corpus, noloop=noloop)
        yield self.update_corpus(corpus)
        returnD(self.jsonrpc_test_corpus(corpus))

    @inlineCallbacks
    def update_corpus(self, corpus=DEFAULT_CORPUS):
        yield self.db.update_corpus(corpus, {
          "ram": self.msclients.corpora[corpus].ram,
          "total_webentities": self.corpora[corpus]['total_webentities'],
          "total_crawls": self.corpora[corpus]['crawls'],
          "total_pages": self.corpora[corpus]['pages_found'],
          "total_pages_crawled": self.corpora[corpus]['pages_crawled'],
          "last_index_loop": self.corpora[corpus]['last_index_loop'],
          "last_links_loop": self.corpora[corpus]['last_links_loop'],
          "last_activity": time.time()
        })

    def jsonrpc_stop_corpus(self, corpus=DEFAULT_CORPUS):
        return self.stop_corpus(corpus)

    @inlineCallbacks
    def stop_corpus(self, corpus=DEFAULT_CORPUS, quiet=False):
        if corpus in self.corpora:
            for f in ["jobs_loop", "index_loop"]:
                if f in self.corpora[corpus] and self.corpora[corpus][f].running:
                    self.corpora[corpus][f].stop()
            yield self.update_corpus(corpus)
            self.msclients.stop_corpus(corpus, quiet)
            for f in ["tags", "webentities", "webentities_links", "precision_exceptions"]:
                if f in self.corpora[corpus]:
                    del(self.corpora[corpus][f])
        res = self.jsonrpc_test_corpus(corpus)
        if "message" in res["result"]:
            res["result"]["message"] = "Corpus stopped"
        returnD(res)

    def jsonrpc_ping(self, corpus=None, timeout=3):
        if not corpus:
            return format_result('pong')
        if not self.corpus_ready(corpus) and self.msclients.status_corpus(corpus, simplify=True) != "starting":
            return self.corpus_error(corpus)

        st = time.time()
        res = self.msclients.sync.ping(corpus=corpus)
        while is_error(res) and time.time() < st + timeout:
            time.sleep(0.1)
            res = self.msclients.sync.ping(corpus=corpus)
        if is_error(res):
            return res
        return format_result('pong')

    def jsonrpc_reinitialize(self, corpus=DEFAULT_CORPUS):
        return self.reinitialize(corpus)

    @inlineCallbacks
    def reinitialize(self, corpus=DEFAULT_CORPUS, noloop=False, quiet=False):
        """Reinitializes both crawl jobs and memory structure."""
        res = yield self.crawler.reinitialize(corpus, recreate=(not noloop), quiet=quiet)
        if is_error(res):
            returnD(res)
        res = yield self.store.reinitialize(corpus, noloop=noloop, quiet=quiet)
        if is_error(res):
            returnD(res)
        returnD(format_result('Memory structure and crawling database contents emptied.'))

    def jsonrpc_destroy_corpus(self, corpus=DEFAULT_CORPUS):
        return self.destroy_corpus(corpus)

    @inlineCallbacks
    def destroy_corpus(self, corpus=DEFAULT_CORPUS, quiet=False):
        res = yield self.reinitialize(corpus, noloop=True, quiet=quiet)
        if is_error(res):
            returnD(res)
        res = yield self.stop_corpus(corpus, quiet)
        if is_error(res):
            returnD(res)
        yield self.db.delete_corpus(corpus)
        returnD(format_result("Corpus %s destroyed successfully" % corpus))

    @inlineCallbacks
    def jsonrpc_reset_all(self):
        corpora = yield self.db.list_corpus(fields=['_id', 'password'])
        for corpus in corpora:
            res = yield self.start_corpus(corpus['_id'], password=corpus['password'], noloop=True, quiet=not config['DEBUG'])
            if is_error(res):
                returnD(res)
            res = self.jsonrpc_ping(corpus['_id'], timeout=10)
            if is_error(res):
                returnD(res)
            res = yield self.destroy_corpus(corpus['_id'], quiet=not config['DEBUG'])
            if is_error(res):
                returnD(res)
        returnD(format_result("All corpora and databases cleaned up"))

  # CORE & CORPUS STATUS

    def jsonrpc_get_status(self, corpus=DEFAULT_CORPUS):
        status = {
          'hyphe': {
            'corpus_running': self.msclients.total_running(),
            'crawls_running': sum([c['crawls_running'] for c in self.corpora.values()]),
            'crawls_pending': sum([c['crawls_pending'] for c in self.corpora.values()]),
            'ports_left': len(self.msclients.ports_free),
            'ram_left': self.msclients.ram_free
          },
          'corpus': {
          }
        }
        status['corpus'].update(self.jsonrpc_test_corpus(corpus)["result"])
        if not self.corpus_ready(corpus):
            return format_result(status)
        WEs_total = self.corpora[corpus]['total_webentities']
        WEs_IN = len([1 for w in self.corpora[corpus]['webentities'] if ms.WebEntityStatus._NAMES_TO_VALUES[w.status] == ms.WebEntityStatus.IN])
        WEs_OUT = len([1 for w in self.corpora[corpus]['webentities'] if ms.WebEntityStatus._NAMES_TO_VALUES[w.status] == ms.WebEntityStatus.OUT])
        WEs_UND = len([1 for w in self.corpora[corpus]['webentities'] if ms.WebEntityStatus._NAMES_TO_VALUES[w.status] == ms.WebEntityStatus.UNDECIDED])
        WEs_DISC = WEs_total - WEs_IN - WEs_OUT - WEs_UND
        corpus_status = {
          'name': self.corpora[corpus]['name'],
          'ram': self.msclients.corpora[corpus].ram,
          'crawler': {
            'jobs_finished': self.corpora[corpus]['crawls'] - self.corpora[corpus]['crawls_pending'] - self.corpora[corpus]['crawls_running'],
            'jobs_pending': self.corpora[corpus]['crawls_pending'],
            'jobs_running': self.corpora[corpus]['crawls_running'],
            'pages_crawled': self.corpora[corpus]['pages_crawled'],
            'pages_found': self.corpora[corpus]['pages_found'],
            'links_found': self.corpora[corpus]['links_found']
          },
          'memory_structure': {
            'job_running': self.corpora[corpus]['loop_running'],
            'job_running_since': self.corpora[corpus]['loop_running_since']*1000 if self.corpora[corpus]['loop_running'] else 0,
            'last_index': self.corpora[corpus]['last_index_loop']*1000,
            'last_links_generation': self.corpora[corpus]['last_links_loop']*1000,
            'pages_to_index': self.corpora[corpus]['pages_queued'],
            'webentities': {
              'total': WEs_total,
              'IN': WEs_IN,
              'OUT': WEs_OUT,
              'UNDECIDED': WEs_UND,
              'DISCOVERED': WEs_DISC
            }
          }
        }
        status['corpus'].update(corpus_status)
        return format_result(status)

  # CRAWL JOBS MONITORING

    @inlineCallbacks
    def jsonrpc_listjobs(self, list_ids=None, corpus=DEFAULT_CORPUS):
        if not self.corpus_ready(corpus):
            returnD(self.corpus_error(corpus))
        query = {}
        if list_ids:
            query = {'_id': {'$in': list_ids}}
        jobs = yield self.db.list_jobs(corpus, query)
        returnD(format_result(list(jobs)))

    @inlineCallbacks
    def refresh_jobs(self, corpus=DEFAULT_CORPUS):
        """Runs a monitoring task on the list of jobs in the database to update their status from scrapy API and indexing tasks."""
        scrapyjobs = yield self.crawler.jsonrpc_list(corpus)
        if is_error(scrapyjobs):
            if not (type(scrapyjobs["message"]) is dict and "status" in scrapyjobs["message"]):
                logger.msg("Problem dialoguing with scrapyd server: %s" % scrapyjobs, system="WARNING - %s" % corpus)
            returnD(None)
        scrapyjobs = scrapyjobs['result']
        self.corpora[corpus]['crawls_pending'] = len(scrapyjobs['pending'])
        self.corpora[corpus]['crawls_running'] = len(scrapyjobs['running'])
        self.corpora[corpus]['pages_queued'] = yield self.db.queue(corpus).count()
        self.corpora[corpus]['pages_crawled'] = yield self.db.pages(corpus).count()
        jobs = yield self.db.list_jobs(corpus, fields=['nb_pages', 'nb_links'])
        self.corpora[corpus]['crawls'] = len(jobs)
        self.corpora[corpus]['pages_found'] = sum([j['nb_pages'] for j in jobs])
        self.corpora[corpus]['links_found'] = sum([j['nb_links'] for j in jobs])
        yield self.update_corpus(corpus)
        # clean lost jobs
        if len(scrapyjobs['running']) + len(scrapyjobs['pending']) == 0:
            yield self.db.update_jobs(corpus, {'crawling_status': {'$in': [crawling_statuses.PENDING, crawling_statuses.RUNNING]}}, {'crawling_status': crawling_statuses.FINISHED})
        # clean canceled jobs
        yield self.db.update_jobs(corpus, {'crawling_status': crawling_statuses.CANCELED}, {'indexing_status': indexing_statuses.CANCELED})
        # update jobs crawling status accordingly to crawler's statuses
        running_ids = [job['id'] for job in scrapyjobs['running']]
        for job_id in running_ids:
            crawled_pages = yield self.db.count_pages(corpus, job_id)
            yield self.db.update_jobs(corpus, job_id, {'nb_crawled_pages': crawled_pages})
        res = yield self.db.list_jobs(corpus, {'_id': {'$in': running_ids}, 'crawling_status': crawling_statuses.PENDING}, fields=['_id'])
        update_ids = [job['_id'] for job in res]
        if len(update_ids):
            yield self.db.update_jobs(corpus, update_ids, {'crawling_status': crawling_statuses.RUNNING})
            yield self.db.add_log(corpus, update_ids, "CRAWL_"+crawling_statuses.RUNNING)
        # update crawling status for finished jobs
        finished_ids = [job['id'] for job in scrapyjobs['finished']]
        res = yield self.db.list_jobs(corpus, {'_id': {'$in': finished_ids}, 'crawling_status': {'$nin': [crawling_statuses.RETRIED, crawling_statuses.CANCELED, crawling_statuses.FINISHED]}}, fields=['_id'])
        update_ids = [job['_id'] for job in res]
        if len(update_ids):
            yield self.db.update_jobs(corpus, update_ids, {'crawling_status': crawling_statuses.FINISHED})
            yield self.db.add_log(corpus, update_ids, "CRAWL_"+crawling_statuses.FINISHED)
        # collect list of crawling jobs whose outputs is not fully indexed yet
        jobs_in_queue = yield self.db.queue(corpus).distinct('_job')
        # set index finished for jobs with crawling finished and no page left in queue
        res = yield self.db.list_jobs(corpus, {'crawling_status': crawling_statuses.FINISHED})
        finished_ids = set([job['_id'] for job in res] + finished_ids)
        res = yield self.db.list_jobs(corpus, {'_id': {'$in': list(finished_ids-set(jobs_in_queue))}, 'crawling_status': crawling_statuses.FINISHED, 'indexing_status': {'$nin': [indexing_statuses.BATCH_RUNNING, indexing_statuses.FINISHED]}}, fields=['_id'])
        update_ids = [job['_id'] for job in res]
        if len(update_ids):
            yield self.db.update_jobs(corpus, update_ids, {'indexing_status': indexing_statuses.FINISHED})
            yield self.db.add_log(corpus, update_ids, "INDEX_"+indexing_statuses.FINISHED)
            # Try to restart in phantom mode all regular crawls that seem to have failed (less than 3 pages found for a depth of at least 1)
            res = yield self.db.list_jobs(corpus, {'_id': {'$in': update_ids}, 'nb_crawled_pages': {'$lt': 3}, 'crawl_arguments.phantom': False, 'crawl_arguments.maxdepth': {'$gt': 0}})
            for job in res:
                logger.msg("Crawl job %s seems to have failed, trying to restart it in phantom mode" % job['_id'], system="INFO - %s" % corpus)
                yield self.jsonrpc_crawl_webentity(job['webentity_id'], job['crawl_arguments']['maxdepth'], True, corpus=corpus)
                yield self.db.add_log(corpus, job['_id'], "CRAWL_RETRIED_AS_PHANTOM")
                yield self.db.update_jobs(corpus, job['_id'], {'crawling_status': crawling_statuses.RETRIED})
        res = yield self.jsonrpc_listjobs(corpus=corpus)
        returnD(res)

  # BASIC PAGE DECLARATION (AND WEBENTITY CREATION)

    @inlineCallbacks
    def jsonrpc_declare_page(self, url, corpus=DEFAULT_CORPUS):
        if not self.corpus_ready(corpus):
            returnD(self.corpus_error(corpus))
        res = yield self.store.declare_page(url, corpus=corpus)
        returnD(handle_standard_results(res))

    @inlineCallbacks
    def jsonrpc_declare_pages(self, list_urls, corpus=DEFAULT_CORPUS):
        res = []
        errors = []
        for url in list_urls:
            WE = yield self.jsonrpc_declare_page(url, corpus=corpus)
            if is_error(WE):
                errors.append({'url': url, 'error': WE['message']})
            else:
                res.append(WE['result'])
        if len(errors):
            returnD({'code': 'fail', 'message': '%d urls failed, see details in "errors" field and successes in "results" field.' % len(errors), 'errors': errors, 'results': res})
        returnD(format_result(res))

  # BASIC CRAWL METHODS

    @inlineCallbacks
    def jsonrpc_crawl_webentity(self, webentity_id, maxdepth=None, phantom_crawl=False, status=ms.WebEntityStatus._VALUES_TO_NAMES[ms.WebEntityStatus.IN], startpages="default", corpus=DEFAULT_CORPUS):
        """Tells scrapy to run crawl on a WebEntity defined by its id from memory structure."""
        if not self.corpus_ready(corpus):
            returnD(self.corpus_error(corpus))
        try:
            maxdepth = int(maxdepth)
        except:
            maxdepth = config['mongo-scrapy']['maxdepth']
        if maxdepth > config['mongo-scrapy']['maxdepth']:
            returnD(format_error('No crawl with a bigger depth than %d is allowed on this Hyphe instance.' % config['mongo-scrapy']['maxdepth']))
        if startpages not in ["default", "startpages", "pages", "prefixes"]:
            returnD(format_error('ERROR: startpages argument must be one of "startpages", "pages" or "prefixes"'))
        WE = yield self.store.get_webentity_with_pages_and_subWEs(webentity_id, startpages, corpus=corpus)
        if is_error(WE):
            returnD(WE)
        statusval = -1
        for s in ms.WebEntityStatus._NAMES_TO_VALUES:
            if status.lower() == s.lower():
                statusval = s
                break
        if statusval == -1:
            returnD(format_error("ERROR: status argument must be one of '%s'" % "','".join([s.lower() for s in ms.WebEntityStatus._NAMES_TO_VALUES])))
        yield self.store.jsonrpc_set_webentity_status(webentity_id, statusval, corpus=corpus)
        yield self.store.jsonrpc_rm_webentity_tag_value(webentity_id, "CORE", "recrawl_needed", "true", corpus=corpus)
        res = yield self.crawler.jsonrpc_start(webentity_id, WE['starts'], WE['lrus'], WE['subWEs'], config['discoverPrefixes'], maxdepth, phantom_crawl, corpus=corpus)
        returnD(res)

    @inlineCallbacks
    def jsonrpc_get_webentity_logs(self, webentity_id, corpus=DEFAULT_CORPUS):
        if not self.corpus_ready(corpus):
            returnD(self.corpus_error(corpus))
        jobs = yield self.db.list_jobs(corpus, {'webentity_id': webentity_id}, fields=['_id'])
        if not jobs:
            returnD(format_error('No job found for WebEntity %s.' % webentity_id))
        res = yield self.db.list_logs(corpus, [a['_id'] for a in list(jobs)])
        returnD(format_result([{'timestamp': log['timestamp'], 'job': log['_job'], 'log': log['log']} for log in list(res)]))

  # HTTP LOOKUP METHODS

    @inlineCallbacks
    def jsonrpc_lookup_httpstatus(self, url, timeout=5, tryout=0, noproxy=False, corpus=None):
        res = format_result(0)
        timeout = int(timeout)
        use_proxy = config['proxy']['host'] and not noproxy
        try:
            url = urllru.url_clean(str(url))
            if use_proxy:
                agent = ProxyAgent(TCP4ClientEndpoint(reactor, config['proxy']['host'], config['proxy']['port'], timeout=timeout))
            else:
                agent = Agent(reactor, connectTimeout=timeout)
            method = "HEAD"
            if tryout > 2:
                method = "GET"
            headers = {'Accept': ['*/*'],
                      'User-Agent': [user_agents.agents[random.randint(0, len(user_agents.agents) -1)]]}
            response = yield agent.request(method, url, Headers(headers), None)
        except DNSLookupError as e:
            res['message'] = "DNS not found for url %s : %s" % (url, e)
        except Exception as e:
            res['result'] = -1
            res['message'] = "Cannot process url %s : %s." % (url, e)
        if 'message' in res:
            returnD(res)
        try:
            assert(response.code == 200 or url in " ".join(response.headers._rawHeaders['location']))
            res['result'] = 200
        except:
            try:
                assert(url.startswith("http:") and tryout == 4 and response.code == 403 and "IIS" in response.headers._rawHeaders['server'][0])
                res['result'] = 301
            except:
                if use_proxy or response.code in [403, 405, 500, 501, 503]:
                    if tryout == 5 and use_proxy:
                        noproxy = True
                        tryout = 3
                    if tryout < 5:
                        if config['DEBUG']:
                            logger.msg("Retry lookup %s %s %s %s" % (method, url, tryout, response.__dict__), system="DEBUG - %s" % corpus)
                        res = yield self.jsonrpc_lookup_httpstatus(url, timeout=timeout+2, tryout=tryout+1, noproxy=noproxy)
                        returnD(res)
                res['result'] = response.code
        returnD(res)

    @inlineCallbacks
    def jsonrpc_lookup(self, url, timeout=5, noproxy=False, corpus=None):
        res = yield self.jsonrpc_lookup_httpstatus(url, timeout=timeout, noproxy=noproxy, corpus=corpus)
        if res['code'] == 'success' and (res['result'] == 200 or 300 < res['result'] < 400):<commit_msg>handle proxy not responding as separate lookup code<commit_after>#!/usr/bin/env python
# -*- coding: utf-8 -*-

import time, random, json, bson
from urllib import urlencode
import subprocess
from txjsonrpc import jsonrpclib
from txjsonrpc.jsonrpc import Introspection
from txjsonrpc.web import jsonrpc
from twisted.web import server
from twisted.python import log as logger
from twisted.internet import reactor, defer
from twisted.internet.task import LoopingCall
from twisted.internet.threads import deferToThread
from twisted.internet.defer import inlineCallbacks, returnValue as returnD
from twisted.internet.endpoints import TCP4ClientEndpoint
from twisted.internet.error import DNSLookupError, ConnectionRefusedError
from twisted.application.internet import TCPServer
from twisted.application.service import Application
from twisted.web.http_headers import Headers
from twisted.web.client import getPage, Agent, ProxyAgent, HTTPClientFactory, _HTTP11ClientFactory
HTTPClientFactory.noisy = False
_HTTP11ClientFactory.noisy = False
from hyphe_backend import processor
from hyphe_backend.lib import config_hci, urllru, gexf, user_agents
from hyphe_backend.lib.utils import *
from hyphe_backend.lib.mongo import MongoDB, sortdesc
from hyphe_backend.lib.corpus import CorpusFactory
from hyphe_backend.memorystructure import MemoryStructure as ms, constants as ms_const


# MAIN CORE API

class Core(jsonrpc.JSONRPC):

    addSlash = True

    def __init__(self):
        jsonrpc.JSONRPC.__init__(self)
        self.db = MongoDB(config['mongo-scrapy'])
        self.msclients = CorpusFactory(config['memoryStructure']['thrift.host'],
          port_range = config['memoryStructure']['thrift.portrange'],
          max_ram = config['memoryStructure']['thrift.max_ram'],
          loglevel = config['memoryStructure']['log.level'])
        self.corpora = {}
        self.crawler = Crawler(self)
        self.store = Memory_Structure(self)
        reactor.addSystemEventTrigger('before', 'shutdown', self.close)
        self.keepalive_default = None

    @inlineCallbacks
    def activate_monocorpus(self):
        yield self.start_corpus(create_if_missing=True)
        self.keepalive_default = LoopingCall(self.jsonrpc_ping, DEFAULT_CORPUS)
        self.keepalive_default.start(300, False)

    @inlineCallbacks
    def close(self):
        if not config["MULTICORPUS"] and self.keepalive_default and self.keepalive_default.running:
            self.keepalive_default.stop()
        for corpus in self.corpora.keys():
            yield self.stop_corpus(corpus, quiet=True)
        yield self.db.close()
        self.msclients.stop()

  # OVERWRITE JSONRPC REQUEST HANDLING

    def render(self, request):
        request.setHeader("Access-Control-Allow-Origin", "*")
        from_ip = ""
        if request.getHeader("x-forwarded-for"):
            from_ip = " from %s" % request.getHeader("x-forwarded-for")
        if config['DEBUG']:
            logger.msg(request.content.read(), system="DEBUG - QUERY%s" % from_ip)
# TODO   catch corpus arg here and return corpus_error if needed
        return jsonrpc.JSONRPC.render(self, request)

    def _cbRender(self, result, request, id, version):
        if config['DEBUG'] == 2:
            txt = jsonrpclib.dumps(result, id=id, version=2.0)
            logger.msg("%s%s" % (txt[:1000], " ... [%d cars truncated]" % (len(txt)-1000) if len(txt) > 1000 else ''), system="DEBUG - ANSWER")
        return jsonrpc.JSONRPC._cbRender(self, result, request, id, version)

  # CORPUS HANDLING

    def jsonrpc_test_corpus(self, corpus=DEFAULT_CORPUS, msg=None):
        res = {
          "corpus_id": corpus,
          "ready": False,
          "status": self.msclients.status_corpus(corpus, simplify=True),
        }
        if res["status"] == "ready":
            res["ready"] = True
        elif res["status"] == "error":
            res["message"] = self.msclients.corpora[corpus].error
        elif res["status"] == "stopped":
            res["message"] = "Corpus is not started"
        if res["status"] == "starting":
            res["message"] = "Corpus is starting, please retry in a bit"
        elif msg:
            res["message"] = msg
        return format_result(res)

    @inlineCallbacks
    def jsonrpc_list_corpus(self):
        res = {}
        corpora = yield self.db.list_corpus()
        for corpus in corpora:
            corpus["password"] = (corpus["password"] != "")
            corpus.update(self.jsonrpc_test_corpus(corpus.pop('_id'))["result"])
            res[corpus["corpus_id"]] = corpus
        returnD(format_result(res))

    def corpus_ready(self, corpus):
        return corpus in self.corpora and self.msclients.test_corpus(corpus)

    def corpus_error(self, corpus=None):
        if not corpus:
            return format_error("Too many instances running already, please try again later")
        reactor.callLater(0, self.stop_corpus, corpus, quiet=True)
        return format_error(self.jsonrpc_test_corpus(corpus)["result"])

    def factory_full(self):
        return self.msclients.ram_free < 256 or not self.msclients.ports_free

    def jsonrpc_create_corpus(self, name=DEFAULT_CORPUS, password="", options=None):
        return self.create_corpus(name, password, options=options)

    @inlineCallbacks
    def create_corpus(self, name=DEFAULT_CORPUS, password="", options=None, noloop=False, quiet=False):
        if self.factory_full():
            returnD(self.corpus_error())

        corpus = clean_corpus_id(name)
        corpus_idx = 1
        existing = yield self.db.get_corpus(corpus)
        while existing:
            corpus = "%s-%s" % (clean_corpus_id(name), corpus_idx)
            corpus_idx += 1
            existing = yield self.db.get_corpus(corpus)

        self.corpora[corpus] = {"name": name}

        yield self.db.add_corpus(corpus, name, password, options)
        try:
            res = yield self.crawler.deploy_spider(corpus)
        except:
            returnD(format_error("Could not deploy corpus' scrapyd spider"))
        if not res:
            returnD(res)
        res = self.store.ensureDefaultCreationRuleExists(corpus, quiet=quiet)
        if not res:
            returnD(res)
        res = yield self.start_corpus(corpus, password=password, noloop=noloop, quiet=quiet)
        returnD(handle_standard_results(res))

    def jsonrpc_start_corpus(self, corpus=DEFAULT_CORPUS, password=""):
        return self.start_corpus(corpus, password)

    @inlineCallbacks
    def start_corpus(self, corpus=DEFAULT_CORPUS, password="", noloop=False, quiet=False, create_if_missing=False):
        if self.corpus_ready(corpus) or self.msclients.status_corpus(corpus, simplify=True) == "starting":
            returnD(self.jsonrpc_test_corpus(corpus, msg="Corpus already ready"))

        corpus_conf = yield self.db.get_corpus(corpus)
        if not corpus_conf:
            if create_if_missing:
                res = yield self.create_corpus(corpus, password, noloop=noloop, quiet=quiet)
                returnD(res)
            returnD(format_error("No corpus existing with ID %s, please create it first!" % corpus))
        if corpus_conf['password'] and corpus_conf['password'] != salt(password):
            returnD(format_error("Wrong auth for password-protected corpus %s" % corpus))
        if self.factory_full():
            returnD(self.corpus_error())

        res = self.msclients.start_corpus(corpus, quiet, ram=corpus_conf['ram'])
        if not res:
            returnD(format_error(self.jsonrpc_test_corpus(corpus)["result"]))
        self.corpora[corpus] = {
          "name": corpus_conf["name"],
          "ram": corpus_conf["ram"],
          "total_webentities": corpus_conf['total_webentities'],
          "crawls": corpus_conf['total_crawls'],
          "crawls_running": 0,
          "crawls_pending": 0,
          "pages_found": corpus_conf['total_pages'],
          "pages_crawled": corpus_conf['total_pages_crawled'],
          "pages_queued": 0,
          "links_found": 0,
          "last_index_loop": corpus_conf['last_index_loop'],
          "last_links_loop": corpus_conf['last_links_loop'],
          "index_loop": LoopingCall(self.store.index_batch_loop, corpus),
          "jobs_loop": LoopingCall(self.refresh_jobs, corpus)
        }
        if not noloop:
            reactor.callLater(5, self.corpora[corpus]['jobs_loop'].start, 1, False)
        yield self.store._start_loop(corpus, noloop=noloop)
        yield self.update_corpus(corpus)
        returnD(self.jsonrpc_test_corpus(corpus))

    @inlineCallbacks
    def update_corpus(self, corpus=DEFAULT_CORPUS):
        yield self.db.update_corpus(corpus, {
          "ram": self.msclients.corpora[corpus].ram,
          "total_webentities": self.corpora[corpus]['total_webentities'],
          "total_crawls": self.corpora[corpus]['crawls'],
          "total_pages": self.corpora[corpus]['pages_found'],
          "total_pages_crawled": self.corpora[corpus]['pages_crawled'],
          "last_index_loop": self.corpora[corpus]['last_index_loop'],
          "last_links_loop": self.corpora[corpus]['last_links_loop'],
          "last_activity": time.time()
        })

    def jsonrpc_stop_corpus(self, corpus=DEFAULT_CORPUS):
        return self.stop_corpus(corpus)

    @inlineCallbacks
    def stop_corpus(self, corpus=DEFAULT_CORPUS, quiet=False):
        if corpus in self.corpora:
            for f in ["jobs_loop", "index_loop"]:
                if f in self.corpora[corpus] and self.corpora[corpus][f].running:
                    self.corpora[corpus][f].stop()
            yield self.update_corpus(corpus)
            self.msclients.stop_corpus(corpus, quiet)
            for f in ["tags", "webentities", "webentities_links", "precision_exceptions"]:
                if f in self.corpora[corpus]:
                    del(self.corpora[corpus][f])
        res = self.jsonrpc_test_corpus(corpus)
        if "message" in res["result"]:
            res["result"]["message"] = "Corpus stopped"
        returnD(res)

    def jsonrpc_ping(self, corpus=None, timeout=3):
        if not corpus:
            return format_result('pong')
        if not self.corpus_ready(corpus) and self.msclients.status_corpus(corpus, simplify=True) != "starting":
            return self.corpus_error(corpus)

        st = time.time()
        res = self.msclients.sync.ping(corpus=corpus)
        while is_error(res) and time.time() < st + timeout:
            time.sleep(0.1)
            res = self.msclients.sync.ping(corpus=corpus)
        if is_error(res):
            return res
        return format_result('pong')

    def jsonrpc_reinitialize(self, corpus=DEFAULT_CORPUS):
        return self.reinitialize(corpus)

    @inlineCallbacks
    def reinitialize(self, corpus=DEFAULT_CORPUS, noloop=False, quiet=False):
        """Reinitializes both crawl jobs and memory structure."""
        res = yield self.crawler.reinitialize(corpus, recreate=(not noloop), quiet=quiet)
        if is_error(res):
            returnD(res)
        res = yield self.store.reinitialize(corpus, noloop=noloop, quiet=quiet)
        if is_error(res):
            returnD(res)
        returnD(format_result('Memory structure and crawling database contents emptied.'))

    def jsonrpc_destroy_corpus(self, corpus=DEFAULT_CORPUS):
        return self.destroy_corpus(corpus)

    @inlineCallbacks
    def destroy_corpus(self, corpus=DEFAULT_CORPUS, quiet=False):
        res = yield self.reinitialize(corpus, noloop=True, quiet=quiet)
        if is_error(res):
            returnD(res)
        res = yield self.stop_corpus(corpus, quiet)
        if is_error(res):
            returnD(res)
        yield self.db.delete_corpus(corpus)
        returnD(format_result("Corpus %s destroyed successfully" % corpus))

    @inlineCallbacks
    def jsonrpc_reset_all(self):
        corpora = yield self.db.list_corpus(fields=['_id', 'password'])
        for corpus in corpora:
            res = yield self.start_corpus(corpus['_id'], password=corpus['password'], noloop=True, quiet=not config['DEBUG'])
            if is_error(res):
                returnD(res)
            res = self.jsonrpc_ping(corpus['_id'], timeout=10)
            if is_error(res):
                returnD(res)
            res = yield self.destroy_corpus(corpus['_id'], quiet=not config['DEBUG'])
            if is_error(res):
                returnD(res)
        returnD(format_result("All corpora and databases cleaned up"))

  # CORE & CORPUS STATUS

    def jsonrpc_get_status(self, corpus=DEFAULT_CORPUS):
        status = {
          'hyphe': {
            'corpus_running': self.msclients.total_running(),
            'crawls_running': sum([c['crawls_running'] for c in self.corpora.values()]),
            'crawls_pending': sum([c['crawls_pending'] for c in self.corpora.values()]),
            'ports_left': len(self.msclients.ports_free),
            'ram_left': self.msclients.ram_free
          },
          'corpus': {
          }
        }
        status['corpus'].update(self.jsonrpc_test_corpus(corpus)["result"])
        if not self.corpus_ready(corpus):
            return format_result(status)
        WEs_total = self.corpora[corpus]['total_webentities']
        WEs_IN = len([1 for w in self.corpora[corpus]['webentities'] if ms.WebEntityStatus._NAMES_TO_VALUES[w.status] == ms.WebEntityStatus.IN])
        WEs_OUT = len([1 for w in self.corpora[corpus]['webentities'] if ms.WebEntityStatus._NAMES_TO_VALUES[w.status] == ms.WebEntityStatus.OUT])
        WEs_UND = len([1 for w in self.corpora[corpus]['webentities'] if ms.WebEntityStatus._NAMES_TO_VALUES[w.status] == ms.WebEntityStatus.UNDECIDED])
        WEs_DISC = WEs_total - WEs_IN - WEs_OUT - WEs_UND
        corpus_status = {
          'name': self.corpora[corpus]['name'],
          'ram': self.msclients.corpora[corpus].ram,
          'crawler': {
            'jobs_finished': self.corpora[corpus]['crawls'] - self.corpora[corpus]['crawls_pending'] - self.corpora[corpus]['crawls_running'],
            'jobs_pending': self.corpora[corpus]['crawls_pending'],
            'jobs_running': self.corpora[corpus]['crawls_running'],
            'pages_crawled': self.corpora[corpus]['pages_crawled'],
            'pages_found': self.corpora[corpus]['pages_found'],
            'links_found': self.corpora[corpus]['links_found']
          },
          'memory_structure': {
            'job_running': self.corpora[corpus]['loop_running'],
            'job_running_since': self.corpora[corpus]['loop_running_since']*1000 if self.corpora[corpus]['loop_running'] else 0,
            'last_index': self.corpora[corpus]['last_index_loop']*1000,
            'last_links_generation': self.corpora[corpus]['last_links_loop']*1000,
            'pages_to_index': self.corpora[corpus]['pages_queued'],
            'webentities': {
              'total': WEs_total,
              'IN': WEs_IN,
              'OUT': WEs_OUT,
              'UNDECIDED': WEs_UND,
              'DISCOVERED': WEs_DISC
            }
          }
        }
        status['corpus'].update(corpus_status)
        return format_result(status)

  # CRAWL JOBS MONITORING

    @inlineCallbacks
    def jsonrpc_listjobs(self, list_ids=None, corpus=DEFAULT_CORPUS):
        if not self.corpus_ready(corpus):
            returnD(self.corpus_error(corpus))
        query = {}
        if list_ids:
            query = {'_id': {'$in': list_ids}}
        jobs = yield self.db.list_jobs(corpus, query)
        returnD(format_result(list(jobs)))

    @inlineCallbacks
    def refresh_jobs(self, corpus=DEFAULT_CORPUS):
        """Runs a monitoring task on the list of jobs in the database to update their status from scrapy API and indexing tasks."""
        scrapyjobs = yield self.crawler.jsonrpc_list(corpus)
        if is_error(scrapyjobs):
            if not (type(scrapyjobs["message"]) is dict and "status" in scrapyjobs["message"]):
                logger.msg("Problem dialoguing with scrapyd server: %s" % scrapyjobs, system="WARNING - %s" % corpus)
            returnD(None)
        scrapyjobs = scrapyjobs['result']
        self.corpora[corpus]['crawls_pending'] = len(scrapyjobs['pending'])
        self.corpora[corpus]['crawls_running'] = len(scrapyjobs['running'])
        self.corpora[corpus]['pages_queued'] = yield self.db.queue(corpus).count()
        self.corpora[corpus]['pages_crawled'] = yield self.db.pages(corpus).count()
        jobs = yield self.db.list_jobs(corpus, fields=['nb_pages', 'nb_links'])
        self.corpora[corpus]['crawls'] = len(jobs)
        self.corpora[corpus]['pages_found'] = sum([j['nb_pages'] for j in jobs])
        self.corpora[corpus]['links_found'] = sum([j['nb_links'] for j in jobs])
        yield self.update_corpus(corpus)
        # clean lost jobs
        if len(scrapyjobs['running']) + len(scrapyjobs['pending']) == 0:
            yield self.db.update_jobs(corpus, {'crawling_status': {'$in': [crawling_statuses.PENDING, crawling_statuses.RUNNING]}}, {'crawling_status': crawling_statuses.FINISHED})
        # clean canceled jobs
        yield self.db.update_jobs(corpus, {'crawling_status': crawling_statuses.CANCELED}, {'indexing_status': indexing_statuses.CANCELED})
        # update jobs crawling status accordingly to crawler's statuses
        running_ids = [job['id'] for job in scrapyjobs['running']]
        for job_id in running_ids:
            crawled_pages = yield self.db.count_pages(corpus, job_id)
            yield self.db.update_jobs(corpus, job_id, {'nb_crawled_pages': crawled_pages})
        res = yield self.db.list_jobs(corpus, {'_id': {'$in': running_ids}, 'crawling_status': crawling_statuses.PENDING}, fields=['_id'])
        update_ids = [job['_id'] for job in res]
        if len(update_ids):
            yield self.db.update_jobs(corpus, update_ids, {'crawling_status': crawling_statuses.RUNNING})
            yield self.db.add_log(corpus, update_ids, "CRAWL_"+crawling_statuses.RUNNING)
        # update crawling status for finished jobs
        finished_ids = [job['id'] for job in scrapyjobs['finished']]
        res = yield self.db.list_jobs(corpus, {'_id': {'$in': finished_ids}, 'crawling_status': {'$nin': [crawling_statuses.RETRIED, crawling_statuses.CANCELED, crawling_statuses.FINISHED]}}, fields=['_id'])
        update_ids = [job['_id'] for job in res]
        if len(update_ids):
            yield self.db.update_jobs(corpus, update_ids, {'crawling_status': crawling_statuses.FINISHED})
            yield self.db.add_log(corpus, update_ids, "CRAWL_"+crawling_statuses.FINISHED)
        # collect list of crawling jobs whose outputs is not fully indexed yet
        jobs_in_queue = yield self.db.queue(corpus).distinct('_job')
        # set index finished for jobs with crawling finished and no page left in queue
        res = yield self.db.list_jobs(corpus, {'crawling_status': crawling_statuses.FINISHED})
        finished_ids = set([job['_id'] for job in res] + finished_ids)
        res = yield self.db.list_jobs(corpus, {'_id': {'$in': list(finished_ids-set(jobs_in_queue))}, 'crawling_status': crawling_statuses.FINISHED, 'indexing_status': {'$nin': [indexing_statuses.BATCH_RUNNING, indexing_statuses.FINISHED]}}, fields=['_id'])
        update_ids = [job['_id'] for job in res]
        if len(update_ids):
            yield self.db.update_jobs(corpus, update_ids, {'indexing_status': indexing_statuses.FINISHED})
            yield self.db.add_log(corpus, update_ids, "INDEX_"+indexing_statuses.FINISHED)
            # Try to restart in phantom mode all regular crawls that seem to have failed (less than 3 pages found for a depth of at least 1)
            res = yield self.db.list_jobs(corpus, {'_id': {'$in': update_ids}, 'nb_crawled_pages': {'$lt': 3}, 'crawl_arguments.phantom': False, 'crawl_arguments.maxdepth': {'$gt': 0}})
            for job in res:
                logger.msg("Crawl job %s seems to have failed, trying to restart it in phantom mode" % job['_id'], system="INFO - %s" % corpus)
                yield self.jsonrpc_crawl_webentity(job['webentity_id'], job['crawl_arguments']['maxdepth'], True, corpus=corpus)
                yield self.db.add_log(corpus, job['_id'], "CRAWL_RETRIED_AS_PHANTOM")
                yield self.db.update_jobs(corpus, job['_id'], {'crawling_status': crawling_statuses.RETRIED})
        res = yield self.jsonrpc_listjobs(corpus=corpus)
        returnD(res)

  # BASIC PAGE DECLARATION (AND WEBENTITY CREATION)

    @inlineCallbacks
    def jsonrpc_declare_page(self, url, corpus=DEFAULT_CORPUS):
        if not self.corpus_ready(corpus):
            returnD(self.corpus_error(corpus))
        res = yield self.store.declare_page(url, corpus=corpus)
        returnD(handle_standard_results(res))

    @inlineCallbacks
    def jsonrpc_declare_pages(self, list_urls, corpus=DEFAULT_CORPUS):
        res = []
        errors = []
        for url in list_urls:
            WE = yield self.jsonrpc_declare_page(url, corpus=corpus)
            if is_error(WE):
                errors.append({'url': url, 'error': WE['message']})
            else:
                res.append(WE['result'])
        if len(errors):
            returnD({'code': 'fail', 'message': '%d urls failed, see details in "errors" field and successes in "results" field.' % len(errors), 'errors': errors, 'results': res})
        returnD(format_result(res))

  # BASIC CRAWL METHODS

    @inlineCallbacks
    def jsonrpc_crawl_webentity(self, webentity_id, maxdepth=None, phantom_crawl=False, status=ms.WebEntityStatus._VALUES_TO_NAMES[ms.WebEntityStatus.IN], startpages="default", corpus=DEFAULT_CORPUS):
        """Tells scrapy to run crawl on a WebEntity defined by its id from memory structure."""
        if not self.corpus_ready(corpus):
            returnD(self.corpus_error(corpus))
        try:
            maxdepth = int(maxdepth)
        except:
            maxdepth = config['mongo-scrapy']['maxdepth']
        if maxdepth > config['mongo-scrapy']['maxdepth']:
            returnD(format_error('No crawl with a bigger depth than %d is allowed on this Hyphe instance.' % config['mongo-scrapy']['maxdepth']))
        if startpages not in ["default", "startpages", "pages", "prefixes"]:
            returnD(format_error('ERROR: startpages argument must be one of "startpages", "pages" or "prefixes"'))
        WE = yield self.store.get_webentity_with_pages_and_subWEs(webentity_id, startpages, corpus=corpus)
        if is_error(WE):
            returnD(WE)
        statusval = -1
        for s in ms.WebEntityStatus._NAMES_TO_VALUES:
            if status.lower() == s.lower():
                statusval = s
                break
        if statusval == -1:
            returnD(format_error("ERROR: status argument must be one of '%s'" % "','".join([s.lower() for s in ms.WebEntityStatus._NAMES_TO_VALUES])))
        yield self.store.jsonrpc_set_webentity_status(webentity_id, statusval, corpus=corpus)
        yield self.store.jsonrpc_rm_webentity_tag_value(webentity_id, "CORE", "recrawl_needed", "true", corpus=corpus)
        res = yield self.crawler.jsonrpc_start(webentity_id, WE['starts'], WE['lrus'], WE['subWEs'], config['discoverPrefixes'], maxdepth, phantom_crawl, corpus=corpus)
        returnD(res)

    @inlineCallbacks
    def jsonrpc_get_webentity_logs(self, webentity_id, corpus=DEFAULT_CORPUS):
        if not self.corpus_ready(corpus):
            returnD(self.corpus_error(corpus))
        jobs = yield self.db.list_jobs(corpus, {'webentity_id': webentity_id}, fields=['_id'])
        if not jobs:
            returnD(format_error('No job found for WebEntity %s.' % webentity_id))
        res = yield self.db.list_logs(corpus, [a['_id'] for a in list(jobs)])
        returnD(format_result([{'timestamp': log['timestamp'], 'job': log['_job'], 'log': log['log']} for log in list(res)]))

  # HTTP LOOKUP METHODS

    @inlineCallbacks
    def jsonrpc_lookup_httpstatus(self, url, timeout=5, tryout=0, noproxy=False, corpus=None):
        res = format_result(0)
        timeout = int(timeout)
        use_proxy = config['proxy']['host'] and not noproxy
        try:
            url = urllru.url_clean(str(url))
            if use_proxy:
                agent = ProxyAgent(TCP4ClientEndpoint(reactor, config['proxy']['host'], config['proxy']['port'], timeout=timeout))
            else:
                agent = Agent(reactor, connectTimeout=timeout)
            method = "HEAD"
            if tryout > 2:
                method = "GET"
            headers = {'Accept': ['*/*'],
                      'User-Agent': [user_agents.agents[random.randint(0, len(user_agents.agents) -1)]]}
            response = yield agent.request(method, url, Headers(headers), None)
        except DNSLookupError as e:
            if use_proxy and config['proxy']['host'] in str(e):
                res['message'] = "Proxy not responding"
                res['result'] = -2
            else:
                res['message'] = "DNS not found for url %s : %s" % (url, e)
        except Exception as e:
            res['result'] = -1
            res['message'] = "Cannot process url %s : %s." % (url, e)
        if 'message' in res:
            returnD(res)
        try:
            assert(response.code == 200 or url in " ".join(response.headers._rawHeaders['location']))
            res['result'] = 200
        except:
            try:
                assert(url.startswith("http:") and tryout == 4 and response.code == 403 and "IIS" in response.headers._rawHeaders['server'][0])
                res['result'] = 301
            except:
                if use_proxy or response.code in [403, 405, 500, 501, 503]:
                    if tryout == 5 and use_proxy:
                        noproxy = True
                        tryout = 3
                    if tryout < 5:
                        if config['DEBUG']:
                            logger.msg("Retry lookup %s %s %s %s" % (method, url, tryout, response.__dict__), system="DEBUG - %s" % corpus)
                        res = yield self.jsonrpc_lookup_httpstatus(url, timeout=timeout+2, tryout=tryout+1, noproxy=noproxy)
                        returnD(res)
                res['result'] = response.code
        returnD(res)

    @inlineCallbacks
    def jsonrpc_lookup(self, url, timeout=5, noproxy=False, corpus=None):
        res = yield self.jsonrpc_lookup_httpstatus(url, timeout=timeout, noproxy=noproxy, corpus=corpus)
        if res['code'] == 'success' and (res['result'] == 200 or 300 < res['result'] < 400):
{'fraction_non_alphanumeric': 0.12298146074878082, 'fraction_numerical': 0.0044213362682403844, 'mean_word_length': 2.9469742360695026, 'pattern_counts': {'":': 64, '<': 13, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '732988', 'n_tokens_mistral': 17060, 'n_tokens_neox': 16168, 'n_words': 3397}