ybw/ybw_details.py

# -*- coding: utf-8 -*-
"""
Created on 2024-06-17
---------
@summary: 元博网 详情页采集
---------
@author: Lzz
"""
import random
import re
import time

import requests.exceptions
from lxml.html import fromstring, HtmlElement, tostring
from lxml.html.clean import Cleaner
from pymongo import MongoClient

import setting
import utils.tools as tool
from dbs.RedisDB import RedisFilter
from log import logger
from utils.check_utils import CheckText, CheckTask
from utils.clean_html import cleaner
from utils.login import User, load_login_cookies, login, login_check

_proxies = setting.PROXIES


def iter_node(element: HtmlElement):
    yield element
    for sub_element in element:
        if isinstance(sub_element, HtmlElement):
            yield from iter_node(sub_element)


def pre_parse(element: HtmlElement):
    """对 HTML 进行预处理可能会破坏 HTML 原有的结构，导致根据原始 HTML 编写的 XPath 不可用"""
    pre_remove = {
        'log_col2', 'log_col1', 'cz', 'iconfont closei', 'p2 p1', 'cnxh_b',
        'msg_error', 'r_gg TB-focus', 'april', 'cont2', 'to_login', 'regtxt',
        'shouchang an_n sc', 'april_title red', 'cn_lt', 'dayin an_n',
        'dl_zc vip_t free_member', 'rmbq', 'login-form cl', 'dian_g fr',
        'di_n', 'd_fx', 'd_tub', 'd_dy', 'anniu1', 'cnxh_list', 'btns cl',
        'active', 'close', 'd_an fr', 'avatar', 'toolbar', 'deng_l',
        'cen_right fr', 'log_col5', 'agreement', 'log_col3',
        'shouchang_af an_n sc_after', 'fast_box', 'di_nr fl', 'xgfj', 'dianh',
        'cnxh_list tab_b2 city_list', 'contract cl', 'zb_cen_r fr', 'd_zsms',
        'sc_after active', 'dl_k', 'ewm_b', 'fl', 'wypj', 'rukou', 'p1',
        'dl_zc', 'success', 'daoh h_30', 'bd', 'april_content', 'print',
        'foot', 'cnxh zbgg', 'april_first', 'fastlog', 'tx_mc user_name',
        'tab_h2', 'fanding an_n', 'toux', 'log_col4 cl', 'hangy rem_1', 'red',
        'regshadow', 'bottom', 'dl_zc vip_t fee_member', 'xszn fl', 'no-print',
        'cnxh_b zbgg_b', 'rem rem_1', 'logshadowz', 'd_pj fl', 'tjgjc',
        'spdujaiwlohh', 'di_ewm fr', 'dian_h fl',
        'tab_h2 zbgg_b_gray', 'fanshou an_n fs', 'login-btn', 'fl gjc',
        'agreeshadow', 'guang_db', 'footer_1', 'log_p', 'cnxh_list tab_b2',
        'd_sw', 'april_close', 'd_sc', 'erweima no-print', 'qgzx', 'p2', 'sc',
        'hd', 'log_col6', 'dh_b', 'dian_guang', 'zhu_c', 'ck cai_k', 'april_box',
        'display:none'
    }
    for node in iter_node(element):
        id_attr = node.attrib.get('id')
        class_attr = node.attrib.get('class')
        style_attr = node.attrib.get('style')
        if any([id_attr in pre_remove,
                class_attr in pre_remove,
                style_attr in pre_remove]):
            node.drop_tree()
    return element


def page_source(element: HtmlElement):
    clear = Cleaner(
        forms=False,
        style=True
    )
    return clear.clean_html(tostring(element, encoding="utf-8").decode())


class DetailSpider:

    def __init__(self):
        _mgo = MongoClient(setting.MONGO_IP, setting.MONGO_PORT)
        self.ybw_list = _mgo[setting.MONGO_DB]["ybw_list"]
        self.ybw_info = _mgo[setting.MONGO_DB]["ybw_info"]
        self.save_tab = _mgo[setting.MONGO_DB]["data_bak"]

        self.dedup = RedisFilter()

        self.user = User(phone=setting.ACCOUNT, passwd=setting.PASSWORD)
        self.login_times = 0

    def json_request(self, fid, request_params):
        headers = {
            "Accept": "application/json, text/plain, */*",
            "Accept-Language": "zh-CN,zh;q=0.9",
            "Connection": "keep-alive",
            "Referer": "https://www.chinabidding.cn/public/bidagency/index.html",
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
        }
        url = "https://www.chinabidding.cn/agency.info.Detail/show"
        params = {
            "fid": f"{fid}"
        }
        return requests.get(url, headers=headers, params=params, **request_params)

    def crawl_request(self, item: dict):
        url = item["competehref"]
        headers = {
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
            'Host': 'www.chinabidding.cn',
            'Upgrade-Insecure-Requests': '1',
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4692.71 Safari/537.36',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
        }

        request_params = {}
        request_params.setdefault('headers', headers)
        request_params.setdefault('timeout', 30)
        request_params.setdefault('proxies', _proxies)

        # 登录代理
        proxy_params = dict(proxies=_proxies, timeout=180)

        retries = 0
        retries_502, max_retries_502 = 0, 3
        while retries < 3:
            if retries_502 > max_retries_502:
                # 网站已移除该数据
                self.ybw_list.update_one({'_id': item["_id"]}, {'$set': {"crawl_status": "remove"}})
                break

            login_cookies = load_login_cookies(self.user.phone)
            if login_cookies is None:
                login(*self.user, **proxy_params)
                self.login_times += 1
                self.update_account_login_times()
                continue

            if 'cookies' not in request_params:
                request_params.setdefault('cookies', login_cookies)
            else:
                request_params.update({'cookies': login_cookies})

            fid = "".join(re.findall('\?fid=(.*)', url)).split('&')[0]
            if fid:
                try:
                    request_params.pop('headers', None)
                    r = self.json_request(fid, request_params)
                    # 账号登录状态检查
                    retry_login = login_check(self.user.phone, url, False, **proxy_params)
                    if retry_login:
                        logger.info(f"[重新登录]{self.user.phone}")
                        _, code = login(*self.user, **proxy_params)
                        self.login_times += 1
                        retries += 1
                        if code != 200:
                            time.sleep(600)

                        continue

                    logger.info(f'[采集正文] fid_{fid}')
                    return r
                except:
                    retries += 1
                finally:
                    self.update_account_login_times()
            else:
                try:
                    r = requests.get(url, **request_params)
                    # 账号登录状态检查
                    retry_login = login_check(self.user.phone, url, False, **proxy_params)
                    if retry_login:
                        logger.info(f"[重新登录]{self.user.phone}")
                        _, code = login(*self.user, **proxy_params)
                        self.login_times += 1
                        retries += 1
                        if code != 200:
                            time.sleep(1800)

                        continue

                    element = fromstring(r.content.decode())
                    nodes = element.xpath('//*[@id="main_dom"]/div[1]')
                    if len(nodes) != 1:
                        retries_502 += 1
                        logger.debug(f'"main_dom"属性匹配个数:{len(nodes)}, {r.status_code} - {url}')
                    else:
                        node = nodes[0]  # list index out of range
                        logger.info(f'[采集正文] id={node.attrib.get("id")}')
                        return r
                except requests.RequestException:
                    retries += 1
                finally:
                    self.update_account_login_times()

        return None

    def crawl_response(self, response, item):
        if re.match('^\{', response.text):
            html = response.json().get('c_info').get('content')
        else:
            element: HtmlElement = fromstring(response.text)
            node = element.xpath('//*[@id="infoDescription"]')[0]
            node = pre_parse(node)
            features = {
                './div[@class="ckgys_cont"]',
                './/div[@class="detail-title ng-scope"]',
                './/table[@class="detail_Table"]',
            }
            for feature in features:
                extract_node = node.xpath(feature)
                if len(extract_node) > 0:
                    valid_node = extract_node[0]
                    break
            else:
                valid_node = node

            html = page_source(valid_node)

        '''检查原始页面内容'''
        CheckText(html)
        item["contenthtml"] = html
        special = {
            '若附件无法下载，你可以尝试使用360极速浏览器进行下载！': '',
            # 'DD000E;|EE000F;|FF000E;': '',
            '[(]?[)]?[A-Z]{2}000[A-Z]{1};[(]?[\d{1,4}]*[;]?[)]?[;]?': '',
        }
        item["detail"] = cleaner(html, special)
        item["comeintime"] = tool.int2long(int(time.time()))

        '''检查清洗之后的详情'''
        CheckText(item["detail"])
        insert = {}
        for key, val in item.items():
            if key not in ['crawl_status', 'crawl', 'count', '_id']:
                insert[key] = val

        self.save_tab.insert_one(insert)
        logger.info('[采集成功]{}-{}'.format(item['title'], item['publishtime']))

    def update_account_login_times(self):
        self.ybw_info.update_one(
            {"account": self.user.phone},
            {"$set": {
                "login_times": self.login_times,
                "update_time": tool.get_current_date()
            }}
        )

    def crawl_spider(self, account, item):
        _id = item["_id"]
        err = "unknown error"

        try:
            CheckTask(item)  # 检查请求采集任务
            response = self.crawl_request(item)
            if response is not None:
                self.crawl_response(response, item)
                self.ybw_list.update_one({"_id": _id}, {"$set": {"crawl_status": "finished"}})
                self.ybw_info.update_one(
                    {"account": self.user.phone},
                    {"$set": {
                        "count": account["count"] + 1,
                        "update_time": tool.get_current_date(),
                    }}
                )
                return True
        except Exception as e:
            err = e
            logger.error(f"请求错误：{err}")

        self.ybw_list.update_one({'_id': _id}, {'$set': {'crawl_status': f'{err}'}})
        return False

    def start(self):
        logger.info(" *** start ***")

        query = {"crawl_status": {"$exists": False}, "es_count": 0}
        sort = [('publishtime', -1)]
        limit = 100
        with self.ybw_list.find(query, sort=sort).limit(limit) as cursor:
            tasks = [doc for doc in cursor]

        download_count = 0
        for item in tasks:
            # 检查账号
            account = self.ybw_info.find_one({"account": self.user.phone})
            if account is None:
                logger.error(f"数据库无此账号信息|{self.user.phone}")
                return

            # 登录次数检查
            self.login_times = account["login_times"]
            if self.login_times >= 3:
                logger.warning(f"账号限制|{self.user.phone}")
                return

            # 数据发布时间延迟采集
            publish_ts = tool.date_to_timestamp(item["publishtime"])
            if publish_ts > int(time.time()) - 43200:
                logger.warning("未到采集时间")
                continue

            fp = "detail_" + item.get("competehref")
            if not self.dedup.get(fp):
                self.dedup.add(fp)

                download_count += 1
                rst = self.crawl_spider(account, item)
                if not rst:
                    self.dedup.delete(fp)

                if download_count >= account["total"]:
                    logger.warning("当日采集数量已达上限")
                    break

                time.sleep(random.randint(80, 180))

        logger.info(" *** end ***")


if __name__ == '__main__':
    DetailSpider().start()