python爬取微信公众号的方法

发布时间:2022-02-19 16:10:03 作者:iii
来源:亿速云 阅读:177

本文小编为大家详细介绍“python爬取微信公众号的方法”,内容详细,步骤清晰,细节处理妥当,希望这篇“python爬取微信公众号的方法”文章能帮助大家解决疑惑,下面跟着小编的思路慢慢深入,一起来学习新知识吧。

pywinauto简介

pywinauto是一个python的工具,可以用于控制Windows的GUI程序。详细的文档可以参考这里。

WechatAutomator类

init_window

这个方法完成类的初始化,它的代码为:

    def init_window(self, exe_path=r"C:Program Files (x86)TencentWeChatWeChat.exe",
                    turn_page_interval=3,
                    click_url_interval=1,
                    win_width=1000,
                    win_height=600):
        app = Application(backend="uia").connect(path=exe_path)
        self.main_win = app.window(title=u"微信", class_name="WeChatMainWndForPC")
        self.main_win.set_focus()
        self.app = app
        self.visible_top = 70
        self.turn_page_interval = turn_page_interval
        self.click_url_interval = click_url_interval
        self.browser = None
        self.win_width = win_width
        self.win_height = win_height
        # 为了让移动窗口,同时使用非uia的backend,这是pywinauto的uia的一个bug
        self.app2 = Application().connect(path=exe_path)
        self.move_window()

我们首先来看函数的参数:

这个函数的主要功能是构建Application对象从而通过pywinauto实现控制,这里使用的是uia的backend,然后设置窗口的大小并且把窗口移到最左上角。因为根据so文章,pywinauto 0.6.8存在bug,只能通过win32的backend来移到窗口,所以构造了self.app2然后调用move_window()函数把窗口移到最左上角。

crawl_gongzhonghao

这个函数实现了某个公众号的文章抓取。它的基本控制逻辑如下:

第一个是通过locate_user函数实现,后面会介绍。第二个是通过process_page函数实现,后面也会介绍。判断是否继续抓取的逻辑为:

所以我们通常会在第一次抓取的时候把max_pages设置的很大(比如100),然后通过latest_date来抓到指定的日期。而之后的抓取则设置max_pages为较小的值(比如默认的6),这样只要爬虫在两次抓取之间公众号的更新不超过6页,那么就不会漏掉文章。具体的逻辑可以参考main.py,它会把抓取的文章通过http请求发给Server,并且每次抓取的时候从Server查询抓取过的文章存放到states这个list里states[i][“url”]就存储了第i篇文章的url。

    def crawl_gongzhonghao(self, account_name, articles, states, detail,
                           max_pages=6, latest_date=None, no_item_retry=3):
        logger.debug(account_name)
        if not self.locate_user(account_name):
            return False
        last_visited_titles = set()
        visited_urls = set()
        self.turn_page_up(min(20, max_pages * 2))

        pagedown_retry = 0
        last_visited_titles = []
        for page in range(0, max_pages):
            items = []
            last_visited_titles = self.process_page(account_name, items, last_visited_titles, states, visited_urls, detail)
            articles.extend(items)

            if len(items) == 0:
                pagedown_retry += 1
                if pagedown_retry >= no_item_retry:
                    s = "break because of retry {}".format(pagedown_retry)
                    logger.debug(s)
                    WechatAutomator.add_to_detail(s, detail)
                    break
            else:
                pagedown_retry = 0

            if len(items) > 0 and latest_date is not None:
                html = items[-1][-1]
                pub_date = WechatAutomator.get_pubdate(html)
                if pub_date and pub_date < latest_date:
                    s = "stop because {} < {}".format(pub_date, latest_date)
                    logger.debug(s)
                    WechatAutomator.add_to_detail(s, detail)
                    break

            url_exist = False
            for item in items:
                if WechatAutomator.url_in_states(item[0], states):
                    s = "stop because url exist {}".format(item[0])
                    logger.debug(s)
                    WechatAutomator.add_to_detail(s, detail)
                    url_exist = True
                    break
            if url_exist:
                break

            self.click_right()
            self.main_win.type_keys("{PGDN}")
            time.sleep(self.turn_page_interval)

        self.turn_page_up(page * 2)

        return True

locate_user

locate_user函数的控制流程为:

    def locate_user(self, user, retry=5):
        if not self.main_win:
            raise RuntimeError("you should call init_window first")

        search_btn = self.main_win.child_window(title="搜索", control_type="Edit")
        self.click_center(search_btn)

        self.main_win.type_keys("^a")
        self.main_win.type_keys("{BACKSPACE}")
        self.main_win.type_keys(user)
        for i in range(retry):
            time.sleep(1)
            try:
                search_list = self.main_win.child_window(title="搜索结果")
                match_result = search_list.child_window(title=user, control_type="ListItem")
                self.click_center(match_result)
                return True
            except:
                pass

        return False

这里主要就是通过child_window函数进行定位,关于它的用法这里不介绍。关于怎么定位元素的方法可以使用Inspect.exe或者print_control_identifiers函数,具体参考这里。

process_page

这个函数是最主要的抓取代码,它处理当前一页的内容,它的控制流程如下:

逻辑比较简单,但是有一些很trick的地方:

与此类似的是右上角的黑色头部(不能滚到并且会遮挡)也有一定空间

因为这个框可能很窄(bottom-top很小)并且可能在很靠上或者靠下的位置。所以有如下代码:

    # 计算可见的高度
    visible_height = min(rect.bottom, win_rect.bottom) - max(rect.top, win_rect.top+self.visible_top)
    # 太窄的不点击,希望下次翻页后能显示更多像素从而可以点击,
    # 但是如果微信的某个文章的框的高度小于10个像素,那么这篇文章就无法被点击
    # 不过作者目前为发现这么窄的文章
    if visible_height < 10:
        continue
    
    # 如果某个文章的框太大,则抛出异常,目前为止为发现这样的问题。
    if rect.bottom - rect.top >= win_rect.bottom - self.visible_top:
        raise RuntimeError("{}-{}>={}-{}".format(rect.bottom, rect.top,
                                                 win_rect.bottom, self.visible_top))
    # 如果下部部分可见,那么点击上方是比较”安全“的
    if rect.bottom >= win_rect.bottom:
        click_up = True
    # 如果下部完全可见,则点击下方是”安全“的
    else:
        click_up = False

完整代码如下:

    def process_page(self, account_name, items, lastpage_clicked_titles, states, visited_urls, detail):
        clicked_titles = set()
        text = self.main_win.child_window(title=account_name, control_type="Text", found_index=0)
        parent = text
        while parent:
            parent = parent.parent()
            if '会话列表' == parent.element_info.name:
                break
        paths = [0, 2, 0, 0, 0, 1, 0]
        for idx in paths:
            parent = parent.children()[idx]

        elems = []
        self.recursive_get(parent, elems)
        win_rect = self.main_win.rectangle()
        for elem in elems:
            rect = elem.rectangle()

            if elem.element_info.name in lastpage_clicked_titles:
                continue

            if rect.top >= win_rect.bottom or rect.bottom <= self.visible_top:
                continue

            visible_height = min(rect.bottom, win_rect.bottom) - max(rect.top, win_rect.top+self.visible_top)
            if visible_height < 10:
                continue

            if rect.bottom - rect.top >= win_rect.bottom - self.visible_top:
                raise RuntimeError("{}-{}>={}-{}".format(rect.bottom, rect.top,
                                                         win_rect.bottom, self.visible_top))
            if rect.bottom >= win_rect.bottom:
                click_up = True
            else:
                click_up = False
            if self.is_bad_elem(elem):
                s = "not good elem {}".format(elem.element_info.name[0:10])
                logger.debug(s)
                WechatAutomator.add_to_detail(s, detail)
                continue

            try:
                self.click_url(rect, win_rect, click_up)
                copy_btn = self.browser.child_window(title="复制链接地址")
                self.click_center(copy_btn, click_main=False)
                url = clipboard.GetData()
                if elem.element_info.name != '图片':
                    clicked_titles.add(elem.element_info.name)
                if url and not url in visited_urls:
                    visited_urls.add(url)
                    html = None
                    try:
                        html = requests.get(url).text
                    except:
                        s = "fail get {}".format(url)
                        logger.debug(s)
                        WechatAutomator.add_to_detail(s, detail)

                    items.append((url, rect, elem.element_info.name, html))

            except:
                traceback.print_exc()
                pass
            finally:
                if self.browser:
                    try:
                        self.browser.close()
                    except:
                        pass
                    self.browser = None

            time.sleep(self.click_url_interval)

        return clicked_titles

读到这里,这篇“python爬取微信公众号的方法”文章已经介绍完毕,想要掌握这篇文章的知识点还需要大家自己动手实践使用过才能领会,如果想了解更多相关内容的文章,欢迎关注亿速云行业资讯频道。

推荐阅读:
  1. Python Scrapy图片爬取的方法
  2. Python爬取微信公众号文章和评论的案例

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

python

上一篇:python如何定义变量和给变量赋值

下一篇:python如何连接数据库

相关阅读

您好,登录后才能下订单哦!

密码登录
登录注册
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》