platform-spiders/a_gtcgpt_cgjg/Gtcgpt.py

# -*- coding: utf-8 -*-
"""
Created on 2024-09-19
---------
@summary: 国铁采购平台
---------
@author: lzz
"""
from collections import namedtuple

import feapder
import feapder.utils.tools as tools
from items.spider_item import BidingListItem
from untils.tools import get_proxy

from fingerprint import fetch_alteon_pcgmh, get_fingerprint, check_fingerprint


class Spider(feapder.BiddingListSpider):

    def start_callback(self):
        Menu = namedtuple('Menu', ['channel', 'code', 'noticeType', 'tid', 'crawl_page'])
        self.site = "国铁采购平台"

        self.menus = [
            Menu('采购结果', 'a_gtcgpt_cgjg', '001', 'queryProcurementResultsList', 20),
        ]

        self.headers = {
            "Accept": "application/json, text/javascript, */*; q=0.01",
            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
            "Cache-Control": "no-cache",
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36",
            "X-Requested-With": "XMLHttpRequest",
        }

        self.alteon_pcgmh = None
        self.cookies = None
        self.proxy = get_proxy()
        self.fp = get_fingerprint()
        self.dt = tools.get_current_date("%Y-%m-%d")  # 搜索查询时间

    def start_requests(self):
        for menu in self.menus:
            referer = "https://cg.95306.cn/baseinfor/notice/procurementNotice"
            params = {
                "bidType": "",
                "noticeType": f"{menu.noticeType}",
                "transactionType": "01",
                "wzType": "",
                "title": "",
                "bidding": "",
                "navigation": ""
            }
            self.headers["Referer"] = tools.joint_url(referer, params)
            url = f"https://cg.95306.cn/proxy/portal/elasticSearch/{menu.tid}"
            for page in range(1, menu.crawl_page + 1):
                yield feapder.Request(url, proxies=False, item=menu._asdict(), page=page)

    def download_midware(self, request):
        if self.alteon_pcgmh is None:
            self.alteon_pcgmh = fetch_alteon_pcgmh(self.proxy)

        if self.cookies is None:
            self.cookies = {
                'AlteonPcgmh': self.alteon_pcgmh,
                'mhId': self.fp,
            }

        if request.page > 1:
            data = {
                'mhId': self.fp,
                'projBidType': '01',
                'bidType': '',
                'noticeType': f"{request.item['noticeType']}",
                'title': '',
                'inforCode': '',
                'startDate': self.dt,
                'endDate': self.dt,
                'pageNum': request.page,
                'projType': '',
                'professionalCode': '',
                'createPeopUnit': '',
            }
        else:
            data = {
                'mhId': self.fp,
                'projBidType': '01',
                'bidType': '',
                'noticeType': f"{request.item['noticeType']}",
                'wzType': '',
                'title': '',
            }

        request.data = data
        request.headers = self.headers
        request.cookies = self.cookies
        request.proxies = self.proxy

    def validate(self, request, response):
        if response.json['code'] == '0-0203':
            referer = self.headers["Referer"]
            check_fingerprint(self.fp, self.cookies, referer, self.proxy)
            request.callback = self.request_retry
            return True
        elif response.json.get('data'):
            return True
        else:
            raise ValueError('数据不能为空!')

    def request_retry(self, request, response):
        request.callback = self.parse
        yield request

    def parse(self, request, response):
        menu = request.item
        info_list = response.json.get('data')
        for info in info_list.get('resultData').get('result'):
            href_id = info.get('id')
            href = f"https://cg.95306.cn/baseinfor/notice/informationShow?id={href_id}"
            title = info.get('notTitle').strip()
            create_time = info.get('checkTime')

            area = "全国"  # 省份
            city = ""  # 城市

            list_item = BidingListItem()  # 存储数据的管道
            list_item.href = href  # 标书链接
            list_item.channel = menu.get("channel")  # 最上方定义的抓取栏目 （编辑器定的）
            list_item.spidercode = menu.get("code")  # 最上方定义的爬虫code（编辑器定的）
            list_item.title = title  # 标题
            list_item.publishtime = create_time  # 标书发布时间
            list_item.site = self.site
            list_item.area = area  # 城市默认:全国
            list_item.city = city  # 城市 默认为空

            list_item.unique_key = ("href",)
            list_item.parse = "self.detail_get"
            list_item.proxies = False
            params_d = {
                "noticeId": f"{href_id}",
                "mhId": self.fp,
            }
            list_item.request_params = {"params": params_d}
            list_item.parse_url = "https://cg.95306.cn/proxy/portal/elasticSearch/indexView"
            yield list_item

    def exception_request(self, request, response):
        self.alteon_pcgmh = None
        self.cookies = None
        self.proxy = get_proxy()
        self.fp = get_fingerprint()
        yield request


if __name__ == "__main__":
    Spider(redis_key="lzz:Gtcgpt").start()