topic_spider/find_source/crawler/services/excavate.py

import re
import time
from concurrent.futures import ThreadPoolExecutor, wait

from common.databases import int2long
from common.log import logger
from crawler.Task import Task
from crawler.analysis import TimeExtractor
from crawler.services.basics import BasicService
from crawler.utils import (
    extract_host,
    extract_page_title,
    extract_domain,
    err_details,
    is_url,
    html2element,
    iter_node,
    check_page_by_words,
    predict_bidding_model,
    is_contains
)
from settings import Dzr

TLDS = ['com', 'cn', 'net', 'org']
URL_SUFFIX = ['pdf', 'xls', 'xlsx', 'docx', 'doc', 'rar', 'zip']
URL_SUFFIX_PATTERN = '.*(' + '|'.join(URL_SUFFIX) + ')$'


def predict_data(html, task: Task):
    data_json = {'html': html, 'predict': 0}
    if html is None:
        return data_json
    data = {'contenthtml': html}
    data_json = predict_bidding_model(data)
    Dzr.insert_one({
        'site': task['origin'],
        # 'html': compress_str(html),
        'url': task['url'],
        'predict': data_json['predict'],
        'comeintime': int2long(int(time.time()))
    })
    return data_json


class DataExcavate(BasicService):
    """数据挖掘服务"""

    def __init__(self, **kwargs):
        self._workers = (kwargs.pop('workers', None) or 1)
        self._max_depth = (kwargs.pop('excavate_depth', None) or 1)
        self._interval = (kwargs.pop('excavate_interval', None) or 60)
        super(DataExcavate, self).__init__(**kwargs)
        self._default_depth = 1

    def _init_depth(self, task: Task):
        if task.get('depth') is None:
            task.update({'depth': self._default_depth})

    def interceptor(self, task: Task):
        """
        拦截器：拦截需要丢弃的任务

        :param task: 请求任务
        :return:
        """
        url = task['url']
        if not is_url(url):
            logger.debug(f'<{self.thread_name}> - 垃圾网址 - {url}')
            return True

        curr_depth = task['depth']
        domain = extract_domain(url)
        is_attachment_url = re.match(URL_SUFFIX_PATTERN, url) is not None
        is_login_url = is_contains(url, 'login')
        if any([
            curr_depth > self._max_depth,  # 检查任务层级
            is_attachment_url,  # 判断是否附件下载地址
            is_login_url,  # 判断是否登录地址
            self.validator.data(url),  # 垃圾池 - 判重任务请求网址
            self.validator.data(domain),  # 垃圾池 - 过滤域名
            curr_depth > 1 and self.collector.data(domain),  # 收录池 - 判重域名（一级）[种子不与已收录判重]
        ]):
            logger.debug(f'<{self.thread_name}> - 无效任务 - {curr_depth} - {url}')
            return True

        return False

    def same_origin_strategy(self, source, task: Task):
        """同源策略"""
        # 排查时间文本
        hit_date_total = 0
        date_nodes = []
        element = html2element(source)
        for node, _ in iter_node(element):
            pt = TimeExtractor().extractor(node)
            if pt and not node.getchildren() and node not in date_nodes:
                date_nodes.append(node)
                hit_date_total += 1

        # 全文排查检索词
        hit_text_total = 0
        all_text = ["".join(text.split()) for text in element.itertext()]
        all_text = [text for text in all_text if len(text) > 0]
        for text in all_text:
            # print(text)
            if check_page_by_words(text):
                hit_text_total += 1

        # 寻源结果
        if all([3 < hit_date_total < 50, hit_text_total > 3]):
            if hit_text_total < 5:
                # 关键词计数为3-5网站 标记为2
                task["hit5"] = 2
            self.push_domain(task)
        elif hit_text_total > 5:
            # 关键词计数为5以上网站   标记为1
            task["hit5"] = 1
            self.push_domain(task)
        else:
            self.push_remove(task)

        # 抽取当前页面同源链接（使用同源模式:mode=1）,不遗漏挖掘层级,同时权重优先级+1,优先处理
        lst = []
        history = []
        sub_depth = task['depth'] + 1
        sub_weight = task['weight'] + 1
        items = self.parser.turls(source, task['url'], mode=1)
        for item in items:
            name, url = item['title'], item['href']
            if not self.validator.data(url) and url not in history:
                lst.append(self.make_task(
                    url=url,
                    name=name,
                    depth=sub_depth,
                    origin=task['origin'],
                    groups=task['groups'],
                    classify=self.visit_classify,
                    weight=sub_weight
                ))
                history.append(url)
        self.scheduler.add_excavate(lst, level=sub_weight)

    def non_origin_strategy(self, source, task: Task):
        """非同源策略"""
        lst = []
        history = []
        # 抽取当前页面非同源链接（使用非同源模式:mode=2）,添加挖掘队列,一个站点挖掘结束,开始处理新站点
        non_origin_urls = self.parser.urls(source, task['url'], mode=2)
        for url in non_origin_urls:
            if not self.validator.data(url) and url not in history:
                lst.append(self.make_task(
                    url=url,
                    origin=task['origin'],
                    groups=task['groups'],
                    classify=self.visit_classify,
                    weight=self.url_weight
                ))
                history.append(url)
        self.scheduler.add_excavate(lst, level=self.url_weight)

    def fetch_page(self, task: Task):
        url = task['url']
        response = self.downloader.get(url, timeout=5)
        task['status_code'] = status_code = response.status_code
        html = response.text
        reason = response.reason
        if status_code != 200 or html in ['', None]:
            task['err_reason'] = reason
            msg = '<{thread_name}> - {url} - {status_code} - {reason}'.format(
                thread_name=self.thread_name,
                url=url,
                status_code=status_code,
                reason=reason
            )
            logger.error(msg)
            return None
        return html

    def process(self, task: Task):
        logger.info(f'<{self.thread_name}> - 请求 - {task["url"]}')
        status = False
        # 预处理域名[domain]与基准地址[base_url]
        task['domain'] = extract_domain(task['url'])
        task['base_url'] = extract_host(task['url'])
        html = self.fetch_page(task)  # 发起请求
        if html is not None:
            predict_res = predict_data(html, task)  # 招投标预测结果
            if predict_res['predict']:
                task['name'] = extract_page_title(html)
                self.same_origin_strategy(html, task)
                self.non_origin_strategy(html, task)
                status = True
        # 处理过的任务加入过滤器，防止重复处理
        self.push_remove(task)
        return status

    def excavate(self):
        logger.info(f'开启线程 - <{self.thread_name}>')
        while True:
            tasks = self.scheduler.get_excavate_task()
            if len(tasks) == 0:
                self.loops_interval(self._interval, enable_debug_log=True)
                continue

            task_key, task = tasks
            self._init_depth(task)  # 初始化网站层级
            if not self.interceptor(task):
                try:
                    self.process(task)
                except Exception as e:
                    logger.exception(f'<{self.thread_name}> {e}')
            # '''挖掘记录'''
            # self.push_records(task)

    def start(self):
        with ThreadPoolExecutor(self._workers, 'dataExcavate') as executor:
            futures = []
            for _ in range(1, self._workers + 1):
                f = executor.submit(self.excavate)
                f.add_done_callback(err_details)
                futures.append(f)
            wait(futures)