python爬取微信公众号文章的方法 |
您所在的位置：网站首页 › 微信公众号爬取 › python爬取微信公众号文章的方法 |
python爬取微信公众号文章的方法
|
最近在学习Python3网络爬虫开发实践（崔庆才 著）刚好也学习到他使用代理爬取公众号文章这里，但是照着他的代码写，出现了一些问题。在这里我用到了这本书的前面讲的一些内容进行了完善。（作者写这个代码已经是半年前的事了，但腾讯的网站在这半年前进行了更新） 下面我直接上代码： TIMEOUT = 20 from requests import Request, Session, PreparedRequest import requests from selenium import webdriver from selenium.common.exceptions import NoSuchElementException from bs4 import BeautifulSoup as bs import pymysql # 要爬取的内容 keyword = '美女图片' options = webdriver.ChromeOptions() # 设置中文 options.add_argument('lang=zh_CN.UTF-8') # 更换头部 options.add_argument( 'user-agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/69.0.3497.100 Safari/537.36"') browser = webdriver.Chrome(chrome_options=options) REDIS_HOST = '192.168.1.248' REDIS_PORT = 6379 REDIS_PASSWORD = '*****' REDIS_KEY = 'requests' PROXY_POOL_URL = 'http://127.0.0.1:8080/random' MAX_FAILED_TIME = 5 MYSQL_HOST = 'localhost' MYSQL_PORT = 3306 MYSQL_USER = 'moxiao' MYSQL_PASSWORD = '******' class mysqlConn(): def __init__(self, host=MYSQL_HOST, username=MYSQL_USER, password=MYSQL_PASSWORD, port=MYSQL_PORT): """ mysql 初始化 :param host: :param username: :param password: :param port: """ try: self.db = pymysql.Connection(host=host, user=username, password=password, database='weixin_data', port=port) self.cursor = self.db.cursor() except pymysql.MySQLError as e: print(e.args) def insert(self, table, data): keys = ', '.join(data.keys()) values = ', '.join(['%s'] * len(data)) sql = 'insert into %s (%s) values (%s)' % (table, keys, values) try: self.cursor.execute(sql, tuple(data.values())) self.db.commit() except pymysql.MySQLError as e: print(e.args) self.db.rollback() class WeixinRequest(Request): def __init__(self, url, callback, method="GET", headers=None, need_proxy=False, fail_time=0, timeout=TIMEOUT): super(WeixinRequest, self).__init__(url=url, method=method, headers=headers) self.callback = callback self.need_proxy = need_proxy self.fail_time = fail_time self.timeout = timeout def prepare(self): p = PreparedRequest() p.prepare( method=self.method, url=self.url, headers=self.headers, ) return p class WeixinResponse(): def __init__(self, text): self.text = text def set_status_code(self, status_code): self.status_code = status_code import pickle from redis import StrictRedis class RedisQueue(): def __init__(self): """ 初始化redis """ self.db = StrictRedis(host=REDIS_HOST, port=REDIS_PORT, password=REDIS_PASSWORD, db=3) def add(self, request): """ 向队列添加序列化后的Request :param request:请求对象 :return:添加结果 """ if isinstance(request, WeixinRequest): return self.db.rpush(REDIS_KEY, pickle.dumps(request)) return False def pop(self): """ 取出下一个request并反序列化 :return: Request 或者 None """ if self.db.llen(REDIS_KEY): return pickle.loads(self.db.lpop(REDIS_KEY)) return False def empty(self): return self.db.llen(REDIS_KEY) == 0 def del_all(self): return self.db.delete(REDIS_KEY) def get_proxy(self): """ 从代理池获取代理IP :return: """ try: response = requests.get(PROXY_POOL_URL) if response.status_code == 200: print('get Proxy', response.text) return response.text except requests.ConnectionError: return None from urllib.parse import urlencode from requests import ReadTimeout, ConnectionError from pyquery import PyQuery as pq VALD_STATUES = [200] class Spider(): base_url = 'http://weixin.sogou.com/weixin?' # 这里的page可以修改，即第几页，我本来想获取所有的个数再除以10 这样就能爬完了，但是我只是测试所以这里并没有做 # 但如果需要做可以加到schedule方法的while循环内的最下面 即self.params['page']+=1 params = {'type': 2, 's_from': 'input', 'query': keyword, 'page': 1, 'ie': 'utf8', '_sug_': 'n', '_sug_type_': ''} headers = {'Host': 'weixin.sogou.com', 'Connection': 'keep-alive', 'Cache-Control': 'max-age=0', 'Upgrade-Insecure-Requests': '1', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/69.0.3497.100 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8', 'Accept-Encoding': 'gzip, deflate', 'Accept-Language': 'zh-CN,zh;q=0.9', 'Referer': 'http: // weixin.sogou.com /', 'Cookie': '你的cookie'} # TODO 不可能把我的给你撒 session = Session() queue = RedisQueue() queue.del_all() mysql = mysqlConn() def start(self): """ 初始化工作 :return: """ # 全局更新headers # 如果你试过用这个方法修改headers，那么就知道这个在这里好像没什么用，我在这里浪费了至少两个小时 self.session.headers.update(self.headers) start_url = self.base_url + urlencode(self.params) # 这里我将need_proxy=False设为了False 即并没有使用代理 ps：我也就是测试一下 # 真正修改了headers是在这里 weixin_request = WeixinRequest(url=start_url, callback=self.parse_index, headers=self.headers, need_proxy=False) # 调度第一个请求 self.queue.add(weixin_request) def schedule(self): """ 调度请求 :return: """ while not self.queue.empty(): weixin_request = self.queue.pop() callback = weixin_request.callback print('Schedule', weixin_request.url) response = self.request(weixin_request) if response and response.status_code in VALD_STATUES: results = list(callback(response)) if results: for result in results: print('New Result', result) if isinstance(result, WeixinRequest): # 将新的文章详情的url也加入队列 self.queue.add(result) if isinstance(result, dict): # 储存到mysql self.mysql.insert('articles', result) else: self.error(weixin_request) else: self.error(weixin_request) def request(self, weixin_request): """ 执行请求 :param weixin_request:请求 :return: 响应 """ if not 'http://mp.weixin.qq.com/s?src' in weixin_request.url: try: if weixin_request.need_proxy: proxy = self.queue.get_proxy() if proxy: proxies = { 'http': 'http://' + proxy, 'https': 'https://' + proxy } return self.session.send(weixin_request.prepare(), timeout=weixin_request.timeout, allow_redirects=False, proxies=proxies) return self.session.send(weixin_request.prepare(), timeout=weixin_request.timeout, allow_redirects=False) except (ConnectionError, ReadTimeout) as e: print(e.args) return False else: print('-' * 20) browser.get(weixin_request.url) try: browser.find_element_by_class_name('rich_media_area_primary_inner') wr = WeixinResponse(browser.page_source) wr.set_status_code(200) return wr except NoSuchElementException: wr = WeixinResponse('') wr.set_status_code(403) return wr def parse_index(self, response): """ 解析索引页 :param response: 响应 :return: 新的响应 """ doc = pq(response.text) items = doc('.news-box .news-list li .txt-box h3 a').items() for item in items: url = item.attr('href') weixin_request = WeixinRequest(url=url, callback=self.parse_detail) yield weixin_request def parse_detail(self, response): """ 解析详情页 :param response: 响应 :return: 微信公众号文章 """ doc = pq(response.text) profile_inner = doc('.profile_inner') data = { 'title': doc('.rich_media_title').text(), 'content': doc('.rich_media_content').text(), 'date': doc('#publish_time').text(), # 'nickname':doc('#js_profile_qrcode > div > strong').text(), 'nickname': profile_inner.find('.profile_nickname').text(), 'wechat': [ns for ns in profile_inner.find('.profile_meta').find('.profile_meta_value').items()][ 0].text() } # 储存图片 print('#' * 30) soup = bs(response.text) wn = soup.find_all('img') for img in wn: if img.has_attr('_width') and img.has_attr('data-src'): print(img.attrs['data-src']) yield data def error(self, weixin_request): """ 错误处理 :param weixin_request:请求 :return: """ weixin_request.fail_time = weixin_request.fail_time + 1 print('Request Failed', weixin_request.fail_time, 'Times', weixin_request.url) if weixin_request.fail_time < MAX_FAILED_TIME: self.queue.add(weixin_request) def run(self): self.start() self.schedule() if __name__ == '__main__': spider = Spider() spider.run()2018-10-6更新： 今天测试之后使用了cookie并不能登录这个网站了，也许是腾讯使用了新的安全验证，具体也无从得知，但使用浏览器访问没有问题
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持脚本之家。 您可能感兴趣的文章:python 爬取百度文库并下载(免费文章限定)Python 微信公众号文章爬取的示例代码Python selenium爬取微信公众号文章代码详解Python如何爬取微信公众号文章和评论(基于 Fiddler 抓包分析)python爬取指定微信公众号文章Python爬取csnd文章并转为PDF文件 |
【本文地址】
公司简介
联系我们
今日新闻 |
点击排行 |
|
推荐新闻 |
图片新闻 |
|
专题文章 |