當前位置：首頁 > 编程语言 > python >内容正文

python

Python 大规模异步新闻爬虫、google翻译、百度翻译、有道翻译、百度指数

發布時間：2024/7/23 python 46 豆豆

生活随笔收集整理的這篇文章主要介紹了 Python 大规模异步新闻爬虫、google翻译、百度翻译、有道翻译、百度指数小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

參考：https://www.yuanrenxue.com/crawler/news-crawler-urlpool.html

url_pool.py

# -*- coding: utf-8 -*- # @Author : 佛祖保佑, 永無 bug # @Date : # @File : url_pool.py # @Software: PyCharm # @description : XXXimport time import redis import pickle import urllib.parse as urlparseclass UrlDB(object):"""使用 redis 來存儲 URL"""status_failure = b'0'status_success = b'1'def __init__(self, db_name):# self.name = db_name + '.urldb'# self.db = leveldb.LevelDB(self.name)self.name = db_name if db_name else 'redis_hashmap'self.db = redis.StrictRedis()def set_success(self, url=None):if isinstance(url, str):url = url.encode('utf8')try:self.db.hset(self.name, url, self.status_success)status = Trueexcept BaseException as be:status = Falsereturn statusdef set_failure(self, url):if isinstance(url, str):url = url.encode('utf8')try:self.db.hset(self.name, url, self.status_failure)status = Trueexcept BaseException as be:status = Falsereturn statusdef has(self, url):if isinstance(url, str):url = url.encode('utf8')try:attr = self.db.hget(self.name, url)return attrexcept BaseException as be:passreturn Falseclass UrlPool(object):""" 使用 UrlPool 來抓取和管理URLs"""def __init__(self, pool_name):self.name = pool_nameself.db = UrlDB(pool_name)self.waiting = dict() # {host: set([urls]), } 按host分組，記錄等待下載的URLself.pending = dict() # {url: pended_time, } 記錄已被取出（self.pop()）但還未被更新狀態（正在下載）的URLself.failure = dict() # {url: times,} 記錄失敗的URL的次數self.failure_threshold = 3self.pending_threshold = 10 # pending的最大時間，過期要重新下載self.waiting_count = 0 # self.waiting 字典里面的url的個數self.max_hosts = ['', 0] # [host: url_count] 目前pool中url最多的host及其url數量self.hub_pool = dict() # {url: last_query_time, } 存放hub urlself.hub_refresh_span = 0self.load_cache()passdef __del__(self):self.dump_cache()def load_cache(self,):path = self.name + '.pkl'try:with open(path, 'rb') as f:self.waiting = pickle.load(f)cc = [len(v) for k, v in self.waiting.items()]print('saved pool loaded! urls:', sum(cc))except BaseException as be:passdef dump_cache(self):path = self.name + '.pkl'try:with open(path, 'wb') as f:pickle.dump(self.waiting, f)print('self.waiting saved!')except BaseException as be:passdef set_hubs(self, urls, hub_refresh_span):self.hub_refresh_span = hub_refresh_spanself.hub_pool = dict()for url in urls:self.hub_pool[url] = 0def set_status(self, url, status_code):if url in self.pending:self.pending.pop(url)if status_code == 200:self.db.set_success(url)returnif status_code == 404:self.db.set_failure(url)returnif url in self.failure:self.failure[url] += 1if self.failure[url] > self.failure_threshold:self.db.set_failure(url)self.failure.pop(url)else:self.add(url)else:self.failure[url] = 1self.add(url)def push_to_pool(self, url=None):host = urlparse.urlparse(url).netlocif not host or '.' not in host:print('try to push_to_pool with bad url:', url, ', len of ur:', len(url))return Falseif host in self.waiting:if url in self.waiting[host]:return Trueself.waiting[host].add(url)if len(self.waiting[host]) > self.max_hosts[1]:self.max_hosts[1] = len(self.waiting[host])self.max_hosts[0] = hostelse:self.waiting[host] = set([url])self.waiting_count += 1return Truedef add(self, url=None, always=False):if always:return self.push_to_pool(url)pended_time = self.pending.get(url, 0)if time.time() - pended_time < self.pending_threshold:print('being downloading:', url)returnif self.db.has(url):returnif pended_time:self.pending.pop(url)return self.push_to_pool(url)def add_many(self, url_list=None, always=False):if isinstance(url_list, str):print('urls is a str !!!!', url_list)self.add(url_list, always)else:for url in url_list:self.add(url, always)def pop(self, count=None, hub_percent=50):print('\n\tmax of host:', self.max_hosts)# 取出的url有兩種類型：hub=1, 普通=0url_attr_url = 0url_attr_hub = 1# 1. 首先取出hub，保證獲取hub里面的最新url.hubs = dict()hub_count = count * hub_percent // 100for hub in self.hub_pool:span = time.time() - self.hub_pool[hub]if span < self.hub_refresh_span:continuehubs[hub] = url_attr_hub # 1 means hub-urlself.hub_pool[hub] = time.time()if len(hubs) >= hub_count:break# 2. 再取出普通urlleft_count = count - len(hubs)urls = dict()for host in self.waiting:if not self.waiting[host]:continueurl = self.waiting[host].pop()urls[url] = url_attr_urlself.pending[url] = time.time()if self.max_hosts[0] == host:self.max_hosts[1] -= 1if len(urls) >= left_count:breakself.waiting_count -= len(urls)print('To pop:%s, hubs: %s, urls: %s, hosts:%s' % (count, len(hubs), len(urls), len(self.waiting)))urls.update(hubs)return urlsdef size(self,):return self.waiting_countdef empty(self,):return self.waiting_count == 0def test():pool = UrlPool('crawl_url_pool')urls = ['http://1.a.cn/xyz','http://2.a.cn/xyz','http://3.a.cn/xyz','http://1.b.cn/xyz-1','http://1.b.cn/xyz-2','http://1.b.cn/xyz-3','http://1.b.cn/xyz-4',]pool.add_many(urls)# del pool# pool = UrlPool('crawl_url_pool')urls = pool.pop(5)urls = list(urls.keys())print('pop:', urls)print('pending:', pool.pending)pool.set_status(urls[0], 200)print('pending:', pool.pending)pool.set_status(urls[1], 404)print('pending:', pool.pending)if __name__ == '__main__':test()

ezpymysql.py

：大規模異步新聞爬蟲：讓MySQL 數據庫操作更方便 - 猿人學

# file: ezpymysql.py # Author: veelion"""A lightweight wrapper around PyMySQL. only for python3"""import time import logging import traceback import pymysql import pymysql.cursorsversion = "0.7" version_info = (0, 7, 0, 0)class Connection(object):"""A lightweight wrapper around PyMySQL."""def __init__(self, host, database, user=None, password=None,port=0, max_idle_time=7 * 3600, connect_timeout=10,time_zone="+0:00", charset="utf8mb4", sql_mode="TRADITIONAL"):self.host = hostself.database = databaseself.max_idle_time = float(max_idle_time)args = dict(use_unicode=True, charset=charset, database=database,init_command=('SET time_zone = "%s"' % time_zone),cursorclass=pymysql.cursors.DictCursor,connect_timeout=connect_timeout, sql_mode=sql_mode)if user is not None:args["user"] = userif password is not None:args["passwd"] = password# We accept a path to a MySQL socket file or a host(:port) stringif "/" in host:args["unix_socket"] = hostelse:self.socket = Nonepair = host.split(":")if len(pair) == 2:args["host"] = pair[0]args["port"] = int(pair[1])else:args["host"] = hostargs["port"] = 3306if port:args['port'] = portself._db = Noneself._db_args = argsself._last_use_time = time.time()try:self.reconnect()except BaseException as be:logging.error("Cannot connect to MySQL on %s", self.host, exc_info=True)def _ensure_connected(self):# Mysql by default closes client connections that are idle for# 8 hours, but the client library does not report this fact until# you try to perform a query and it fails. Protect against this# case by preemptively closing and reopening the connection# if it has been idle for too long (7 hours by default).if self._db is None or (time.time() - self._last_use_time > self.max_idle_time):self.reconnect()self._last_use_time = time.time()def _cursor(self):self._ensure_connected()return self._db.cursor()def __del__(self):self.close()def close(self):"""Closes this database connection."""if getattr(self, "_db", None) is not None:self._db.close()self._db = Nonedef reconnect(self):"""Closes the existing database connection and re-opens it."""self.close()self._db = pymysql.connect(**self._db_args)self._db.autocommit(True)def query(self, query, *parameters, **kwparameters):"""Returns a row list for the given query and parameters."""cursor = self._cursor()try:cursor.execute(query, kwparameters or parameters)result = cursor.fetchall()return resultfinally:cursor.close()def get(self, query, *parameters, **kwparameters):"""Returns the (singular) row returned by the given query."""cursor = self._cursor()try:cursor.execute(query, kwparameters or parameters)return cursor.fetchone()finally:cursor.close()def execute(self, query, *parameters, **kwparameters):"""Executes the given query, returning the lastrowid from the query."""cursor = self._cursor()try:cursor.execute(query, kwparameters or parameters)return cursor.lastrowidexcept Exception as e:if e.args[0] == 1062:passelse:traceback.print_exc()raise efinally:cursor.close()insert = execute# =============== high level method for table ===================def table_has(self, table_name, field, value):if isinstance(value, str):value = value.encode('utf8')sql_str = f'SELECT {field} FROM {table_name} WHERE {field}="{value}"'d = self.get(sql_str)return ddef table_insert(self, table_name, item):"""item is a dict : key is mysql table field"""fields = list(item.keys())values = list(item.values())field_str = ','.join(fields)val_str = ','.join(['%s'] * len(item))for i in range(len(values)):if isinstance(values[i], str):values[i] = values[i].encode('utf8')sql_str = f'INSERT INTO {table_name} ({field_str}) VALUES({val_str})'try:last_id = self.execute(sql_str, *values)return last_idexcept Exception as e:if e.args[0] == 1062:# just skip duplicated itempasselse:traceback.print_exc()print('sql:', sql)print('item:')for i in range(len(fields)):vs = str(values[i])if len(vs) > 300:print(fields[i], ' : ', len(vs), type(values[i]))else:print(fields[i], ' : ', vs, type(values[i]))raise edef table_update(self, table_name, updates, field_where, value_where):"""updates is a dict of {field_update:value_update}"""upsets = []values = []for k, v in updates.items():s = '%s=%%s' % kupsets.append(s)values.append(v)upsets = ','.join(upsets)sql_str = f'UPDATE {table_name} SET {upsets} WHERE {field_where}="{value_where}"'self.execute(sql_str, *values)if __name__ == '__main__':db = Connection('localhost','db_name','user','password')# 獲取一條記錄sql = 'select * from test_table where id=%s'data = db.get(sql, 2)# 獲取多天記錄sql = 'select * from test_table where id>%s'data = db.query(sql, 2)# 插入一條數據sql = 'insert into test_table(title, url) values(%s, %s)'last_id = db.execute(sql, 'test', 'http://a.com/')# 或者last_id = db.insert(sql, 'test', 'http://a.com/')# 使用更高級的方法插入一條數據item = {'title': 'test','url': 'http://a.com/',}last_id = db.table_insert('test_table', item)

functions.py

# -*- coding: utf-8 -*- # @Author : 佛祖保佑, 永無 bug # @Date : # @File : functions.py # @Software: PyCharm # @description : XXXimport re import requests import cchardet import traceback import urllib.parse as urlparseasync def fetch(session=None, url=None, headers=None, timeout=9, binary=False):_headers = {'User-Agent': ('Mozilla/5.0 (compatible; MSIE 9.0; ''Windows NT 6.1; Win64; x64; Trident/5.0)'),}_headers = headers if headers else _headerstry:async with session.get(url, headers=_headers, timeout=timeout) as response:status_code = response.statushtml_bin_or_text = ''html_content = await response.read()if not binary:encoding = cchardet.detect(html_content)['encoding']html_bin_or_text = html_content.decode(encoding, errors='ignore')request_url = str(response.url)except Exception as e:msg = 'Failed download: {} | exception: {}, {}'.format(url, str(type(e)), str(e))print(msg)html_bin_or_text = ''status_code = -1request_url = urlreturn status_code, html_bin_or_text, request_urldef downloader(url=None, timeout=10, headers=None, debug=False, binary=False):_headers = {'User-Agent': ('Mozilla/5.0 (compatible; MSIE 9.0; ''Windows NT 6.1; Win64; x64; Trident/5.0)'),}_headers = headers if headers else _headersrequest_url = urltry:r = requests.get(url, headers=_headers, timeout=timeout)if binary:html_bin_or_text = r.contentelse:encoding = cchardet.detect(r.content)['encoding']html_bin_or_text = r.content.decode(encoding, errors='ignore')status_code = r.status_coderequest_url = r.urlexcept BaseException as be:if debug:traceback.print_exc()msg = 'failed download: {}'.format(url)print(msg)html_bin_or_text = b'' if binary else ''status_code = -1return status_code, html_bin_or_text, request_urlg_bin_postfix = {'exe', 'doc', 'docx', 'xls', 'xlsx', 'ppt', 'pptx', 'pdf','jpg', 'png', 'bmp', 'jpeg', 'gif', 'zip', 'rar', 'tar','bz2', '7z', 'gz', 'flv', 'mp4', 'avi', 'wmv', 'mkv', 'apk' } g_news_postfix = ['.html?', '.htm?', '.shtml?', '.shtm?']def clean_url(url=None):# 1. 是否為合法的http urlif not url.startswith('http'):return ''# 2. 去掉靜態化url后面的參數for np in g_news_postfix:p = url.find(np)if p > -1:p = url.find('?')url = url[:p]return url# 3. 不下載二進制類內容的鏈接up = urlparse.urlparse(url)path = up.pathif not path:path = '/'postfix = path.split('.')[-1].lower()if postfix in g_bin_postfix:return ''# 4. 去掉標識流量來源的參數# badquery = ['spm', 'utm_source', 'utm_source', 'utm_medium', 'utm_campaign']good_queries = []for query in up.query.split('&'):qv = query.split('=')if qv[0].startswith('spm') or qv[0].startswith('utm_'):continueif len(qv) == 1:continuegood_queries.append(query)query = '&'.join(good_queries)url = urlparse.urlunparse((up.scheme,up.netloc,path,up.params,query,'' # crawler do not care fragment))return urlg_pattern_tag_a = re.compile(r'<a[^>]*?href=[\'"]?([^> \'"]+)[^>]*?>(.*?)</a>', re.I | re.S | re.M)def extract_links_re(url=None, html=None):"""use re module to extract links from html"""news_links = set()tag_a_list = g_pattern_tag_a.findall(html)for tag_a in tag_a_list:link = tag_a[0].strip()if not link:continuelink = urlparse.urljoin(url, link)link = clean_url(link)if not link:continuenews_links.add(link)return news_linksdef init_file_logger(f_name=None):# config loggingimport loggingfrom logging.handlers import TimedRotatingFileHandlerch = TimedRotatingFileHandler(f_name, when="midnight")ch.setLevel(logging.INFO)# create formatterfmt = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'formatter = logging.Formatter(fmt)# add formatter to chch.setFormatter(formatter)logger = logging.getLogger(f_name)# add ch to loggerlogger.addHandler(ch)return loggerif __name__ == '__main__':temp_url = 'http://news.baidu.com/'t_status_code, t_html, t_url = downloader(url=temp_url)print(f'[{t_status_code}, {t_url}]:{len(t_html)}')

config.py

db_host = 'localhost' db_db = 'crawler' db_user = 'your-user' db_password = 'your-password'

新聞爬蟲 (?同步 )

news_sync.py

# -*- coding: utf-8 -*- # @Author : 佛祖保佑, 永無 bug # @Date : # @File : news_sync.py # @Software: PyCharm # @description : XXXimport urllib.parse as urlparse import lzma import farmhash import tracebackfrom ezpymysql import Connection from url_pool import UrlPool import functions as fn import configclass NewsCrawlerSync:def __init__(self, name):self.db = Connection(config.db_host,config.db_db,config.db_user,config.db_password)self.logger = fn.init_file_logger(name + '.log')self.url_pool = UrlPool(name)self.hub_hosts = Noneself.load_hubs()def load_hubs(self,):sql = 'select url from crawler_hub'data = self.db.query(sql)self.hub_hosts = set()hubs = []for d in data:host = urlparse.urlparse(d['url']).netlocself.hub_hosts.add(host)hubs.append(d['url'])self.url_pool.set_hubs(hubs, 300)def save_to_db(self, url, html):url_hash = farmhash.hash64(url)sql = f'select url from crawler_html where urlhash={url_hash}'d = self.db.get(sql, url_hash)if d:if d['url'] != url:msg = 'farm_hash collision: %s <=> %s' % (url, d['url'])self.logger.error(msg)return Trueif isinstance(html, str):html = html.encode('utf8')html_lzma = lzma.compress(html)sql = 'insert into crawler_html(urlhash, url, html_lzma) values(%s, %s, %s)'good = Falsetry:self.db.execute(sql, url_hash, url, html_lzma)good = Trueexcept Exception as e:if e.args[0] == 1062:# Duplicate entrygood = Truepasselse:traceback.print_exc()raise ereturn gooddef filter_good(self, urls):good_links = []for url in urls:host = urlparse.urlparse(url).netlocif host in self.hub_hosts:good_links.append(url)return good_linksdef process(self, url, is_hub):status, html, redirected_url = fn.downloader(url)self.url_pool.set_status(url, status)if redirected_url != url:self.url_pool.set_status(redirected_url, status)# 提取hub網頁中的鏈接, 新聞網頁中也有“相關新聞”的鏈接，按需提取if status != 200:returnif is_hub:new_links = fn.extract_links_re(redirected_url, html)good_links = self.filter_good(new_links)print(f"{len(good_links)} / {len(new_links)}, good_links/new_links")self.url_pool.add_many(good_links)else:self.save_to_db(redirected_url, html)def run(self,):while 1:urls = self.url_pool.pop(5)for url, is_hub in urls.items():self.process(url, is_hub)if __name__ == '__main__':crawler = NewsCrawlerSync('sync_spider')crawler.run()

新聞爬蟲 (?異步 )

news_async.py

# -*- coding: utf-8 -*- # @Author : 佛祖保佑, 永無 bug # @Date : # @File : news_async.py # @Software: PyCharm # @description : XXXimport traceback import time import asyncio import aiohttp import urllib.parse as urlparse import farmhash import lzma# import uvloop # asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())import sanicdbfrom url_pool import UrlPool import functions as fn import configclass NewsCrawlerAsync:def __init__(self, name):self.hub_hosts = set()self._workers = 0self._workers_max = 30self.logger = fn.init_file_logger(name + '.log')self.url_pool = UrlPool(name)self.loop = asyncio.get_event_loop()self.session = aiohttp.ClientSession(loop=self.loop)self.db = sanicdb.SanicDB(config.db_host,config.db_db,config.db_user,config.db_password,loop=self.loop)async def load_hubs(self, ):sql = 'select url from crawler_hub'data = await self.db.query(sql)hubs = []for d in data:host = urlparse.urlparse(d['url']).netlocself.hub_hosts.add(host)hubs.append(d['url'])self.url_pool.set_hubs(hubs, 300)async def save_to_db(self, url, html):url_hash = farmhash.hash64(url)sql = 'select url from crawler_html where urlhash=%s'd = await self.db.get(sql, url_hash)if d:if d['url'] != url:msg = 'farmhash collision: %s <=> %s' % (url, d['url'])self.logger.error(msg)return Trueif isinstance(html, str):html = html.encode('utf8')html_lzma = lzma.compress(html)sql = 'insert into crawler_html(urlhash, url, html_lzma) values(%s, %s, %s)'good = Falsetry:await self.db.execute(sql, url_hash, url, html_lzma)good = Trueexcept Exception as e:if e.args[0] == 1062:# Duplicate entrygood = Truepasselse:traceback.print_exc()raise ereturn gooddef filter_good(self, urls):good_links = []for url in urls:host = urlparse.urlparse(url).netlocif host in self.hub_hosts:good_links.append(url)return good_linksasync def process(self, url, is_hub):status, html, redirected_url = await fn.fetch(self.session, url)self.url_pool.set_status(url, status)if redirected_url != url:self.url_pool.set_status(redirected_url, status)# 提取hub網頁中的鏈接, 新聞網頁中也有“相關新聞”的鏈接，按需提取if status != 200:self._workers -= 1returnif is_hub:new_links = fn.extract_links_re(redirected_url, html)good_links = self.filter_good(new_links)print(f"{len(good_links)} / {len(new_links)}, good_links / new_links")self.url_pool.add_many(good_links)else:await self.save_to_db(redirected_url, html)self._workers -= 1async def loop_crawl(self):await self.load_hubs()last_rating_time = time.time()counter = 0while 1:to_pop = self._workers_max - self._workerstasks = self.url_pool.pop(to_pop)if not tasks:print('no url to crawl, sleep')await asyncio.sleep(3)continuefor url, is_hub in tasks.items():self._workers += 1counter += 1print('crawl:', url)asyncio.ensure_future(self.process(url, is_hub))gap = time.time() - last_rating_timeif gap > 5:rate = counter / gapprint(f'\tloop_crawl() rate:{round(rate, 2)}, counter: {counter}, workers: {self._workers}')last_rating_time = time.time()counter = 0if self._workers > self._workers_max:print('====== got workers_max, sleep 3 sec to next worker =====')await asyncio.sleep(3)def run(self):try:self.loop.run_until_complete(self.loop_crawl())except KeyboardInterrupt:print('stopped by yourself!')del self.url_poolpassif __name__ == '__main__':nc = NewsCrawlerAsync('async_spider')nc.run()

分布式爬蟲（ CS 模型?）

server.py

# -*- coding: utf-8 -*- # @Author : 佛祖保佑, 永無 bug # @Date : # @File : server.py # @Software: PyCharm # @description : XXXfrom sanic import Sanic from sanic import responsefrom my_url_pool import UrlPoolurl_pool = UrlPool(__file__)# 初始化 url_pool，根據你的需要進行修改 hub_urls = [] url_pool.set_hubs(hub_urls, 300) url_pool.add('https://news.sina.com.cn/')# init main_app = Sanic(__name__)@main_app.listener('after_server_stop') async def cache_url_pool(app=None, loop=None):global url_poolprint('caching url_pool after_server_stop')del url_poolprint('bye!')@main_app.route('/task') async def task_get(request=None):count = request.args.get('count', 10)try:count = int(count)except BaseException as be:count = 10urls = url_pool.pop(count)return response.json(urls)@main_app.route('/task', methods=['POST', ]) async def task_post(request=None):result = request.jsonurl_pool.set_status(result['url'], result['status'])if result['url_real'] != result['url']:url_pool.set_status(result['url_real'], result['status'])if result['new_urls']:print('receive URLs:', len(result['new_urls']))for url in result['new_urls']:url_pool.add(url)return response.text('ok')if __name__ == '__main__':main_app.run(host='0.0.0.0', port=8080, debug=False, access_log=False, workers=1)pass

client.py

# -*- coding: utf-8 -*- # @Author : 佛祖保佑, 永無 bug # @Date : # @File : client.py # @Software: PyCharm # @description : XXXimport re import cchardet import traceback import time import json import asyncio import urllib.parse as urlparse import aiohttp# import uvloop # asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())p_tag_a = re.compile(r'<a[^>]*?href=[\'"]?([^> \'"]+)[^>]*?>(.*?)</a>', re.I | re.S | re.M)def extract_links_re(url, html):new_links = set()aa = p_tag_a.findall(html)for a in aa:link = a[0].strip()if not link:continuelink = urlparse.urljoin(url, link)if not link.startswith('http'):continuenew_links.add(link)return new_linksclass CrawlerClient:def __init__(self, ):self._workers = 0self.workers_max = 20self.server_host = 'localhost'self.server_port = 8080self.headers = {'User-Agent': ('Mozilla/5.0 (compatible; MSIE 9.0; ''Windows NT 6.1; Win64; x64; Trident/5.0)')}self.loop = asyncio.get_event_loop()self.queue = asyncio.Queue(loop=self.loop)self.session = aiohttp.ClientSession(loop=self.loop)async def download(self, url, timeout=25):status_code = 900html = ''url_now = urltry:async with self.session.get(url_now, headers=self.headers, timeout=timeout) as response:status_code = response.statushtml = await response.read()encoding = cchardet.detect(html)['encoding']html = html.decode(encoding, errors='ignore')url_now = str(response.url)except BaseException as be:# traceback.print_exc()print('=== exception: ', be, type(be), str(be))msg = 'Failed download: {} | exception: {}, {}'.format(url, str(type(be)), str(be))print(msg)return status_code, html, url_nowasync def get_urls(self, ):count = self.workers_max - self.queue.qsize()if count <= 0:print('no need to get urls this time')return Noneurl = f'http://{self.server_host}:{self.server_port}/task?count={count}'try:async with self.session.get(url, timeout=3) as response:if response.status not in [200, 201]:returnjsn = await response.text()urls = json.loads(jsn)msg = f'get_urls() to get [{count}] but got[{len(urls)}], @{time.strftime("%Y-%m-%d %H:%M:%S")}'print(msg)for kv in urls.items():await self.queue.put(kv)print('queue size:', self.queue.qsize(), ', _workers:', self._workers)except BaseException as be:traceback.print_exc()returnasync def send_result(self, result):url = f'http://{self.server_host}:{self.server_port}/task'try:async with self.session.post(url, json=result, timeout=3) as response:return response.statusexcept BaseException as be:traceback.print_exc()pass@staticmethoddef save_html(url, html):print('saved:', url, len(html))@staticmethoddef filter_good(urls):"""根據抓取目的過濾提取的URLs，只要你想要的"""good = []for url in urls:if url.startswith('http'):good.append(url)return goodasync def process(self, url, is_hub):status, html, url_now = await self.download(url)self._workers -= 1print('downloaded:', url, ', html:', len(html))if html:new_urls = extract_links_re(url, html)new_urls = self.filter_good(new_urls)self.save_html(url, html)else:new_urls = []result = {'url': url,'url_real': url_now,'status': status,'new_urls': new_urls,}await self.send_result(result)async def loop_get_urls(self, ):print('loop_get_urls() start')while 1:await self.get_urls()await asyncio.sleep(1)async def loop_crawl(self, ):print('loop_crawl() start')asyncio.ensure_future(self.loop_get_urls())counter = 0while 1:item = await self.queue.get()url, url_level = itemself._workers += 1counter += 1asyncio.ensure_future(self.process(url, url_level))if self._workers > self.workers_max:print('====== got workers_max, sleep 3 sec to next worker =====')await asyncio.sleep(3)def start(self):try:self.loop.run_until_complete(self.loop_crawl())except KeyboardInterrupt:print('stopped by yourself!')passdef run():ant = CrawlerClient()ant.start()if __name__ == '__main__':run()

google 翻譯

google 翻譯：Google 翻譯

# -*- coding: utf-8 -*- # @Author : 佛祖保佑, 永無 bug # @Date : # @File : translate_google.py # @Software: PyCharm # @description : XXXimport requests import urllib3 urllib3.disable_warnings()def google_translate(kw=None):url = 'https://translate.google.cn/_/TranslateWebserverUi/data/batchexecute?rpcids=MkEWBc&hl=zh-CN'custom_headers = {'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8'}# kw = 'this is a test'payload = f"f.req=[[[\"MkEWBc\",\"[[\\\"{kw}\\\",\\\"auto\\\",\\\"zh-CN\\\",true],[null]]\",null,\"generic\"]]]&"resp = requests.post(url, data=payload, headers=custom_headers)print(resp.status_code)print(resp.text)passif __name__ == '__main__':kw_list = ['I love u', 'hello, baby', 'king','this is a test']for item in kw_list:google_translate(kw=item)pass

百度翻譯

百度翻譯：https://fanyi.baidu.com/

查看 basetrans 請求。

（多請求幾次，可以發現 sign 每次都不一樣，所以需要逆向 sign??）

使用 postman 來精簡請求參數，看那些參數是必須的，那些參數是可以直接刪除的。精簡后參數

請求URL：https://fanyi.baidu.com/basetrans
請求頭：Content-Type: application/x-www-form-urlencoded
Cookie: BAIDUID=AF87393A8DB7C8FED7859A909FF081A3:SL=0:NR=50:FG=1;?
User-Agent: Mozilla/5.0 (Linux; Android 8.0; Pixel 2 Build/OPD3.170816.012) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.82 Mobile Safari/537.36
請求 body：query=king&from=en&to=zh&token=94c55bca8b920035077b58d58ba32bea&sign=612765.899756

postman 生成的 Python 代碼：

import requestsurl = "https://fanyi.baidu.com/basetrans"payload = "query=king&from=en&to=zh&token=94c55bca8b920035077b58d58ba32bea&sign=612765.899756" headers = {'Content-Type': 'application/x-www-form-urlencoded','Cookie': 'BAIDUID=AF87393A8DB7C8FED7859A909FF081A3:SL=0:NR=50:FG=1;','User-Agent': 'Mozilla/5.0 (Linux; Android 8.0; Pixel 2 Build/OPD3.170816.012) ''AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.82 Mobile Safari/537.36' }response = requests.request("POST", url, headers=headers, data=payload)print(response.text)

JS 斷點調試

打斷點，然后通過調用堆棧追蹤 sign 是怎么生成的。。。

因為多個 ajax 請求都會走 b.send(e.data ? e.data:null)?這個函數，所以需要點擊好幾次 "跳轉到下一個斷點" 才能看到 sign 值。

如果想直接斷點就能看到 sign 值，則可以添加 "URL包含" 斷點。

這里使用 "跳轉到下一個斷點" 來追蹤?sign 值。

生成 sign 的函數

點擊 P(e) 函數，查看函數實現

方法 1：直接使用 Python 實現這個函數的邏輯

方法 2：把 js 代碼扣出來，直接 Python 執行

代碼太多，這里直接扣 js 代碼

執行結果：

也可以直接在 Chrome 上的 console 中執行 js

Python 直接調用 JS 代碼（ js_code = r"""js代碼"""? 中 r 不能少?）

// l = null !== i ? i : (i = o.common[d] || "") || "";? ? //? l 的值等于 gkt，通過調試可知，這是個固定值
l = "320305.131321201";? ? // 直接令 l =?"320305.131321201";

# @Author : 佛祖保佑, 永無 bug # @Date : # @File : translate_baidu.py # @Software: PyCharm # @description : XXXimport execjs import requestsjs_code = r""" function n(r, o) {for (var t = 0; t < o.length - 2; t += 3) {var e = o.charAt(t + 2);e = e >= "a" ? e.charCodeAt(0) - 87 : Number(e),e = "+" === o.charAt(t + 1) ? r >>> e : r << e,r = "+" === o.charAt(t) ? r + e & 4294967295 : r ^ e}return r } function sign(r) {var t = r.match(/[\uD800-\uDBFF][\uDC00-\uDFFF]/g);if (null === t) {var a = r.length;a > 30 && (r = "" + r.substr(0, 10) + r.substr(Math.floor(a / 2) - 5, 10) + r.substr(-10, 10))} else {for (var C = r.split(/[\uD800-\uDBFF][\uDC00-\uDFFF]/), h = 0, f = C.length, u = []; f > h; h++)"" !== C[h] && u.push.apply(u, e(C[h].split(""))),h !== f - 1 && u.push(t[h]);var g = u.length;g > 30 && (r = u.slice(0, 10).join("") + u.slice(Math.floor(g / 2) - 5, Math.floor(g / 2) + 5).join("") + u.slice(-10).join(""))}var l = void 0, d = "" + String.fromCharCode(103) + String.fromCharCode(116) + String.fromCharCode(107);// l = null !== i ? i : (i = o.common[d] || "") || "";l = "320305.131321201";for (var m = l.split("."), S = Number(m[0]) || 0, s = Number(m[1]) || 0, c = [], v = 0, F = 0; F < r.length; F++) {var p = r.charCodeAt(F);128 > p ? c[v++] = p : (2048 > p ? c[v++] = p >> 6 | 192 : (55296 === (64512 & p) && F + 1 < r.length && 56320 === (64512 & r.charCodeAt(F + 1)) ? (p = 65536 + ((1023 & p) << 10) + (1023 & r.charCodeAt(++F)),c[v++] = p >> 18 | 240,c[v++] = p >> 12 & 63 | 128) : c[v++] = p >> 12 | 224,c[v++] = p >> 6 & 63 | 128),c[v++] = 63 & p | 128)}for (var w = S, A = "" + String.fromCharCode(43) + String.fromCharCode(45) + String.fromCharCode(97) + ("" + String.fromCharCode(94) + String.fromCharCode(43) + String.fromCharCode(54)), b = "" + String.fromCharCode(43) + String.fromCharCode(45) + String.fromCharCode(51) + ("" + String.fromCharCode(94) + String.fromCharCode(43) + String.fromCharCode(98)) + ("" + String.fromCharCode(43) + String.fromCharCode(45) + String.fromCharCode(102)), D = 0; D < c.length; D++)w += c[D], w = n(w, A);return w = n(w, b),w ^= s,0 > w && (w = (2147483647 & w) + 2147483648),w %= 1e6,w.toString() + "." + (w ^ S) } console.log(sign('king')) """js_func = execjs.compile(js_code)def bd_translate(kw=None):url = "https://fanyi.baidu.com/basetrans"sign = js_func.call('sign', kw)payload = f"query={kw}&from=en&to=zh&token=94c55bca8b920035077b58d58ba32bea&sign={sign}"headers = {'Content-Type': 'application/x-www-form-urlencoded','Cookie': 'BAIDUID=AF87393A8DB7C8FED7859A909FF081A3:SL=0:NR=50:FG=1;','User-Agent': 'Mozilla/5.0 (Linux; Android 8.0; Pixel 2 Build/OPD3.170816.012) ''AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.82 Mobile Safari/537.36'}response = requests.request("POST", url, headers=headers, data=payload)# print(response.text)print(response.json()['trans'][0]['dst'])if __name__ == '__main__':kw_list = ['hello baby', 'I love u','king']for item in kw_list:bd_translate(kw=item)pass

執行結果：

有道翻譯

參考：https://www.cnblogs.com/xuchunlin/p/10412505.html

JS逆向 ——? 百度翻譯參數（sign）爬蟲超級詳細：https://blog.csdn.net/qq_38534107/article/details/90440403

Python 執行 js 代碼

關于 Python 執行 js 代碼

1. 通過 python 模塊?execjs 來解析 js 文件（安裝：?pip install PyExecJS）
2.?python 調用 node.js（需要先安裝 node.js）
3. 使用 PyV8 在 Python 爬蟲中執行 js 代碼：https://www.baidu.com/s?wd=python%20pyv8
4. 使用瀏覽器（ selenium、PhantomJS）執行 js：https://www.cnblogs.com/chenhuabin/p/10946085.html

PyV8、PyExecJS、js2py 區別：https://www.jianshu.com/p/2da6f6ad01f0

PyV8 github 地址：https://github.com/emmetio/pyv8-binaries

Python 使用 execjs 示例：

python 的 execjs 只支持一些常規的 js 代碼 ：https://blog.csdn.net/weixin_42081389/article/details/99984352

import execjs js_str = ''' function add(x, y){return x + y; } ''' test = execjs.compile(js_str)# call 即調用js函數，add 為 js_str 中的函數名，1，2 為所需要的參數。 result = test.call('add', 1, 2) print(result)

有道詞典 js 加密參數 sign破解

分析：

然后轉到 Headers，查看請求體，可以看到有請求驗證字段，如：salt，sign 等?:

Ctrl + Shift? + F : 全局搜索 sign 這個關鍵字，搜索所有包含 sign 的 js 文件，發現只搜索出一個，雙擊這個 js 文件：

在 js 文件內容中搜索 sign ，分析 sign 怎么產生的：

提取 js 的加密函數（這里是 md5 ）；

通過 js 調試，可以找到 md5 函數：

把 md5 函數提取出來，放到一個單獨的 js 文件里面，然后，再通過 js調試把 md5 函數所依賴的 js 函數全部提取出來。

目錄結構：

提取的? 所有 js 函數如下( yd_js.js)：

python 代碼（ yd.py ）：

import execjs import time import random import requests import json""" 通過在js文件中查找salt或者sign，可以找到 1.可以找到這個計算salt的公式 r = "" + ((new Date).getTime() + parseInt(10 * Math.random(), 10)) 2.sign：n.md5("fanyideskweb" + t + r + "p09@Bn{h02_BIEe]$P^nG"); md5 一共需要四個參數，第一個和第四個都是固定值得字符串，第三個是所謂的salt，第二個參數是輸入的需要翻譯的單詞 """def get_md5(v):# 讀取js文件with open('yd_js.js', encoding='utf-8') as f:js = f.read()# 通過compile命令轉成一個js對象js_obj = execjs.compile(js)res = js_obj.call('md5', v)return resdef get_sign(key, salt):sign = "fanyideskweb" + str(key) + str(salt) + "n%A-rKaT5fb[Gy?;N5@Tj"sign = get_md5(sign)return signdef you_dao(key):url = "http://fanyi.youdao.com/translate_o?smartresult=dict&smartresult=rule"ts = str(int((time.time() * 1000)))salt = str(ts) + str(random.randint(0, 10))data = {"i": key,"from": "AUTO","to": "AUTO","smartresult": "dict","client": "fanyideskweb","salt": str(salt),"sign": get_sign(key, salt),"ts": ts,"bv": "5872543b025b19167cde3785ecf1e925","doctype": "json","version": "2.1","keyfrom": "fanyi.web","action": "FY_BY_REALTIME","typoResult": "false",}headers = {"Host": "fanyi.youdao.com",# "Proxy-Connection":"keep-alive","Content-Length": str(len(data)),"Accept": "application/json, text/javascript, */*; q=0.01","Origin": "http://fanyi.youdao.com","X-Requested-With": "XMLHttpRequest","User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) ""Chrome/75.0.3770.90 Safari/537.36","Content-Type": "application/x-www-form-urlencoded; charset=UTF-8","Referer": "http://fanyi.youdao.com/","Accept-Encoding": "gzip, deflate","Accept-Language": "zh-CN,zh;q=0.9","Cookie": "OUTFOX_SEARCH_USER_ID=-803834638@61.149.7.46; ""JSESSIONID=aaa2KqSwX9shJdA5Mk9Ww; OUTFOX_SEARCH_USER_ID_NCOO=1481235354.231604; ""___rl__test__cookies=1564486753731",}r = requests.post(url=url, data=data, headers=headers)if r.status_code == 200:data = json.loads(r.text)src = data.get('translateResult')[0][0]['src']tgt = data.get('translateResult')[0][0]['tgt']print(f'翻譯前 :{src}')print(f'翻譯后 :{tgt}')else:print(f'請求失敗 status code {r.status_code}')if __name__ == '__main__':you_dao("hello, baby")you_dao("I love you")you_dao("thank you very much")you_dao("天王蓋地虎，寶塔鎮河妖")

運行結果截圖：

百度指數 js 破解

分析過程參考：https://blog.csdn.net/wang785994599/article/details/97135979

Python爬蟲 - 簡單抓取百度指數：https://zhuanlan.zhihu.com/p/78634149

Python 代碼實現（?直接從瀏覽器拿到登錄后的Cookie復制粘貼到代碼中 ）：

# -*- coding: utf-8 -*-import requests import execjs import urllib3# 禁用警告 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)js_string = ''' function decrypt(t, e) {for (var n = t.split(""), i = e.split(""), a = {}, r = [], o = 0; o < n.length / 2; o++)a[n[o]] = n[n.length / 2 + o];for (var s = 0; s < e.length; s++)r.push(a[i[s]]);return r.join("") } '''headers = {"Cookie": "直接從瀏覽器拿到登錄后的Cookie復制粘貼","User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) ""Chrome/75.0.3770.142 Safari/537.36" }data_url = 'https://index.baidu.com/api/SearchApi/index?word={}&area=0&days=7' uniq_id_url = 'https://index.baidu.com/Interface/ptbk?uniqid={}' keys = ["all", "pc", "wise"]class BDIndex(object):def __init__(self):self.session = self.get_session()pass@staticmethoddef get_session():"""初始化 session 會話:return:"""session = requests.session()session.headers = headerssession.verify = Falsereturn session@staticmethoddef decrypt(key, data):"""得到解密后的數據:param key: key:param data: key 對應的 value:return:"""js_handler = execjs.compile(js_string)return js_handler.call('decrypt', key, data)def get_bd_index(self, key_word):"""得到百度指數:param key_word::return:"""response = self.session.get(data_url.format(key_word)).json()uniq_id = self.session.get(uniq_id_url.format(response.get("data").get("uniqid"))).json().get("data")result = []data_dict = response.get("data").get("userIndexes")[0]for key in keys:decrypt_data = self.decrypt(uniq_id, data_dict.get(key).get("data"))result.append({key: decrypt_data})return resultif __name__ == '__main__':bd = BDIndex()d = bd.get_bd_index("楊冪")print(d)

運行結果：

總結

以上是生活随笔為你收集整理的Python 大规模异步新闻爬虫、google翻译、百度翻译、有道翻译、百度指数的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇：这个程序没毛病
下一篇： Python 使用 smtp ssl 模