"""
"""
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)