taiqiuxuanzhi_spider/spider.py

import os
from threading import current_thread
import requests
from bs4 import BeautifulSoup
from processor import WuBaProcessor
import time
import random
from verify.yidun import YiDun
from threading import Thread
from queue import Queue
from log import PPLogger
from setting import CONCURRENT_REQUESTS, USER_AGENT, DOWNLOADER_IP_PROXY, START_URL
from concurrent.futures import ThreadPoolExecutor, as_completed


class WuBaSpider:
    def __init__(self):
        self.ip_proxy = DOWNLOADER_IP_PROXY  # 代理ip请求url
        self.ua = USER_AGENT
        self.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",
            "accept-encoding": "gzip, deflate, br",
            "accept-language": "zh-CN,zh;q=0.9",
            "user-agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.5060.114 Safari/537.36"
        }
        self.yidun = YiDun()  # 易盾检测
        self.q = Queue()
        self.logger = PPLogger(name='58spider')
        self.logger.setup_logger()
        self.psr = WuBaProcessor()  # 数据处理
        self.ip_pool = []  # 代理ip池
        # self.lock = Lock()

    def new_session(self, session_=None):
        """
        重置session，添加代理
        :param session_: 旧session
        :return:
        """
        if session_:
            try:
                self.ip_pool.remove(session_.proxies)
            except Exception as e:
                pass
            self.logger.debug(f'ip_pool remove {session_.proxies}')
        if not self.ip_pool:
            res_ip = requests.get(self.ip_proxy).json()
            self.logger.debug(res_ip)
            if res_ip['code'] != 200:
                os._exit(0)  # 代理ip请求失败，结束程序
            for i in res_ip['data']:
                self.ip_pool.append({
                    'http': f'http://{i["ip"]}:{i["port"]}',
                    'https': f'http://{i["ip"]}:{i["port"]}'
                })
        session = requests.Session()
        session.proxies = random.choice(self.ip_pool)
        return session

    def get_page_urls(self, url):
        """
        获取所有商铺列表页
        :param url: 初始url
        :return:商铺列表页url列表
        """
        session = self.new_session()
        # url = 'https://qd.58.com/licang/shangpucz/pn1/?area=100_300&huansuanyue=0_10000'
        try:
            res = session.get(url, headers=self.headers, allow_redirects=True, timeout=10)
        except OSError as e:
            self.logger.error(e)
            session = self.new_session(session)
            res = session.get(url, headers=self.headers, allow_redirects=True, timeout=10)
        # print(res.text)
        # 易盾检测
        if 'verifycode' in res.url:
            session = self.yidun.verify(session, res.url)
            try:
                res = session.get(url, headers=self.headers, allow_redirects=True, timeout=10)
            except OSError as e:
                self.logger.error(e)
                return None
        soup = BeautifulSoup(res.text, 'lxml')
        end_page = int(soup.find('div', class_='pager').find_all('span')[-2].text)
        # end_page = 8
        all_pages = [url.replace('pn1', f'pn{i + 1}') for i in range(end_page)]
        # if not all_pages:
        #     self.logger.info('no pages data')
        #     os._exit(0)
        # session = requests.Session()
        self.logger.info(f'all_pages {len(all_pages)}')
        return all_pages

    def get_detail_urls(self, page_url):
        """
        获取列表页上所有商铺详情页链接
        :param page_url: 商铺列表页url
        """
        session = self.new_session()
        try:
            res_page = session.get(page_url, headers=self.headers, allow_redirects=True, timeout=10)
        except OSError as e:
            self.logger.error(e)
            session = self.new_session(session)
            res_page = session.get(page_url, headers=self.headers, allow_redirects=True, timeout=10)
        # 易盾检测
        if 'verifycode' in res_page.url:
            session = self.yidun.verify(session, res_page.url)
            try:
                res_page = session.get(page_url, headers=self.headers, allow_redirects=True, timeout=10)
            except Exception as e:
                self.logger.error(e)
                session = self.new_session(session)
                res_page = session.get(page_url, headers=self.headers, allow_redirects=True, timeout=10)

        soup_page = BeautifulSoup(res_page.text, 'lxml')
        if soup_page.find('div', class_='content-side-left'):
            detail_urls = [li.find('a').attrs['href'] for li in soup_page.find('div', class_='content-side-left').find_all('li')]

            for detail_url in detail_urls:
                self.q.put_nowait(detail_url)  # 加入队列
                self.logger.debug(f'detail_url {self.q.qsize()}')

    def get_content(self):
        """
        获取商铺详情页数据
        """
        session = self.new_session()
        while True:
            self.logger.debug(f'剩余 {self.q.qsize()}')
            if self.q.empty():
                time.sleep(10)  # 无数据等待10秒再退出
                if self.q.empty():
                    self.logger.info(current_thread().name+' quit')
                    break
            detail_url = self.q.get()
            self.headers['user-agent'] = random.choice(self.ua)
            try:
                res_detail = session.get(detail_url, headers=self.headers, timeout=10)
            except OSError as e:
                session = self.new_session(session)
                self.logger.error(e)
                self.q.put_nowait(detail_url)  # 请求失败后，url放回队列
                continue
            else:
                self.logger.debug(res_detail.url)
                # 极验检测
                if 'geetest' in res_detail.text:
                    self.q.put_nowait(detail_url)
                    continue
                # 易盾检测
                elif 'verifycode' in res_detail.url:
                    session = self.yidun.verify(session, res_detail.url)
                    try:
                        res_detail = session.get(detail_url, headers=self.headers, allow_redirects=True, timeout=10)
                    except OSError as e:
                        self.logger.error(e)
                        self.q.put_nowait(detail_url)
                        continue
                # IP检测
                elif 'verifylogin' in res_detail.url or 'deny' in res_detail.url:
                    session = self.new_session(session)
                    self.q.put_nowait(detail_url)
                    continue
                # with self.lock:  # 文件写入锁
                try:
                    self.psr.processor(res_detail.text, res_detail.url)
                except Exception as e:
                    self.logger.error(str(e)+res_detail.url)
                    # self.q.put_nowait(detail_url)

    def start(self):
        page_urls = self.get_page_urls(START_URL)
        # for page_url in page_urls:
        #     self.get_detail_urls(page_url)
        with ThreadPoolExecutor() as pool:
            results1 = [pool.submit(self.get_detail_urls, page_url) for page_url in page_urls]
        for _ in range(CONCURRENT_REQUESTS):
            Thread(target=self.get_content).start()



# WuBaSpider().start()