py-hutun-robot/huitun/__init__.py

"""

"""
import logging
from huitun.captcha_ident import CaptchaIdent

from playwright.sync_api import sync_playwright, Page, Playwright

import api
from browser import BaseBrowser
from util.lock_util import LockManager
from util.playwright_util import is_element_present
from urllib.parse import urlparse, parse_qs

HUITUN_URL = 'https://xhs.huitun.com/'
lock_manager = LockManager()
password_dict = {}


class HuiTunBrowser(BaseBrowser):
    def __get_name__(self):
        return 'huitun'

    def login(self, password: str):
        """
        登录抖音，一个登录之后，全部的页面都有了登录状态
        :return: 2- 需要验证码 1-登录成功
        """
        self.__init_browser__()
        self.page.goto(HUITUN_URL)
        password_dict[self.phone] = password
        self.login_if_need()
        self.page.wait_for_timeout(30_000)
        self.close()

    def login_if_need(self):
        """
        登录灰豚
        """
        login_info_expired = self.page.query_selector('.ant-btn-primary:has-text("知道了")')
        if login_info_expired is not None:
            login_info_expired.click()
        if is_element_present(self.page, '.ant-modal-body'):
            logging.info('灰豚需要重新登录')
            if not is_element_present(self.page, 'text=密码登录'):
                pwd_login = self.page.query_selector('.b9dOaTo9gfF3wLAi7jlXTg\=\=')
                if pwd_login is not None:
                    pwd_login.click()
            self.page.get_by_placeholder('请输入手机号').type(self.phone)
            self.page.get_by_placeholder('6-15位数字与字母组合').type(password_dict.get(self.phone))
            self.page.get_by_text('登 录', exact=True).click()
            # 验证码登录
            captcha_frame = self.page.frames[1]
            if captcha_frame is not None:
                captcha_tool = CaptchaIdent(self.page)
                captcha_tool.start()

    def search_note(self, tag_name: str, size: int):
        lock_manager.acquire_lock(self.phone)
        try:
            self.__init_browser__()
            self.list_result = []
            self.has_more = True
            api.assert_not_none(tag_name, "标签不能为空")
            self.page.goto('https://xhs.huitun.com/#/note/note_search')
            self.page.wait_for_timeout(2000)
            self.login_if_need()
            # 展开全部标签
            self.page.query_selector('.zgInWFcVVDjRN6BUMm3N0g\=\=').click()
            last_tag = self.page.query_selector('.fyBvQcyA81sogVJY0YVnhg\=\=')
            if last_tag is not None:
                last_tag.click()
            tag_ele = self.page.query_selector(f'.IRk6XOEYweiS9APLHrOp-w\=\=:has-text("{tag_name}")')
            if tag_ele is not None:
                tag_ele.click()
            self.page.get_by_text('图文笔记', exact=True).click()
            self.page.wait_for_timeout(500)
            self.page.on('response', self.search_note_handler)
            self.page.get_by_text('近3天', exact=True).click()
            # 限定一个上限
            page_num = int(2 * size / 10)
            for i in range(page_num):
                if size is not None and len(self.list_result) >= size:
                    break
                logging.info('继续搜索灰豚')
                self.page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
                self.page.wait_for_timeout(2000)
                logging.info('搜索灰豚结果数：%s', len(self.list_result))
                if not self.has_more:
                    break
            return self.list_result
        finally:
            lock_manager.release_lock(self.phone)
            self.close()

    def search_note_handler(self, response):
        """
        处理用户主页搜索图文请求响应
        :param response:
        :return:
        """
        if response is not None and '/note/search' in response.url:
            response_body = response.json()
            if response_body.get('status') == 0:
                note_list = response_body.get('extData').get('list')
                self.has_more = len(note_list) > 0
                if len(self.list_result) == 0:
                    self.list_result = note_list
                else:
                    self.list_result.extend(note_list)
            else:
                self.has_more = False

    def search_note_by_hot_tag(self, size: int):
        """抓取热词搜索文章"""
        lock_manager.acquire_lock(self.phone)
        try:
            self.__init_browser__()
            self.list_result = []
            self.has_more = True
            self.page.on('response', self.hot_tag_handler)
            self.page.goto('https://xhs.huitun.com/#/hot/topic_list')
            self.login_if_need()
            self.page.wait_for_timeout(3000)
            api.assert_not_empty(self.list_result,"获取标签失败")
            self.list_result = self.list_result[:size]
            topic_map = {}
            for tag in self.list_result:
                logging.info(f'搜索标签：{tag.get("topicName")}')
                topic_id = tag.get("topicId")
                topic_map[topic_id] = tag
                self.page.goto(f'https://xhs.huitun.com/#/anchor/topic_detail?id={topic_id}')
                self.page.reload()
                self.page.get_by_text('笔记分析',exact=True).click()
                self.page.on('response', self.hot_tag_note_handler)
                self.page.get_by_text('近7天', exact=True).click()
                page_num = 1
                # 目前版本最多只能翻 50页
                while self.map_result.get(topic_id) is not None and len(self.map_result.get(topic_id)) < 30 and page_num <= 50:
                    logging.info(f'继续搜索热标签图文,当前文章数量：{len(self.map_result.get(topic_id))}, 页数:{page_num}')
                    self.page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
                    self.page.wait_for_timeout(1000)
                    self.page.query_selector('.ant-pagination-next').click()
                    self.page.wait_for_timeout(1000)
                    page_num += 1
            return {
                'tagList': self.list_result,
                'tagNotes': self.map_result
            }
        finally:
            lock_manager.release_lock(self.phone)
            self.close()

    def hot_tag_handler(self, response):
        """处理热词搜索请求响应"""
        if response is not None and ('/topic/search' in response.url or '/rank/topic/add' in response.url):
            response_body = response.json()
            if response_body.get('status') == 0:
                tag_list = response_body.get('extData').get('list')
                self.list_result = tag_list[0:19]

    def hot_tag_note_handler(self, response):
        """处理热词搜索文章请求响应"""
        if response is not None and '/topic/detail/notes/' in response.url:
            response_body = response.json()
            if response_body.get('status') == 0:
                parsed_url = urlparse( response.url)
                query_params = parse_qs(parsed_url.query)
                topic_id = query_params.get('topicId', [None])[0]
                note_list = response_body.get('extData').get('list')
                # 只筛选图文笔记
                note_list = [note for note in note_list if note.get('type') == 'normal']
                exist_note_list = self.map_result.get(topic_id)
                if exist_note_list is None:
                    self.map_result[topic_id] = note_list
                else:
                    exist_note_list.extend(note_list)