diff --git "a/Day66-75/04.\345\271\266\345\217\221\344\270\213\350\275\275.md" "b/Day66-75/04.\345\271\266\345\217\221\344\270\213\350\275\275.md" index b15fa5d68f4058ef0dec4c9b5b1d17f001700a83..8553299b57764cea04057047be11b46aee35c792 100644 --- "a/Day66-75/04.\345\271\266\345\217\221\344\270\213\350\275\275.md" +++ "b/Day66-75/04.\345\271\266\345\217\221\344\270\213\350\275\275.md" @@ -1,2 +1,145 @@ ## 并发下载 +### 多线程和多进程回顾 + + + +### 实例 - 多线程下载“手机搜狐网”所有页面。 + +```Python + +from enum import Enum, unique +from queue import Queue +from random import random +from threading import Thread, current_thread +from time import sleep +from urllib.parse import urlparse + +import requests +from bs4 import BeautifulSoup + + +@unique +class SpiderStatus(Enum): + IDLE = 0 + WORKING = 1 + + +def decode_page(page_bytes, charsets=('utf-8',)): + page_html = None + for charset in charsets: + try: + page_html = page_bytes.decode(charset) + break + except UnicodeDecodeError: + pass + return page_html + + +class Retry(object): + + def __init__(self, *, retry_times=3, + wait_secs=5, errors=(Exception, )): + self.retry_times = retry_times + self.wait_secs = wait_secs + self.errors = errors + + def __call__(self, fn): + + def wrapper(*args, **kwargs): + for _ in range(self.retry_times): + try: + return fn(*args, **kwargs) + except self.errors as e: + print(e) + sleep((random() + 1) * self.wait_secs) + return None + + return wrapper + + +class Spider(object): + + def __init__(self): + self.status = SpiderStatus.IDLE + + @Retry() + def fetch(self, current_url, *, charsets=('utf-8', ), + user_agent=None, proxies=None): + thread_name = current_thread().name + print(f'[{thread_name}]: {current_url}') + headers = {'user-agent': user_agent} if user_agent else {} + resp = requests.get(current_url, + headers=headers, proxies=proxies) + return decode_page(resp.content, charsets) \ + if resp.status_code == 200 else None + + def parse(self, html_page, *, domain='m.sohu.com'): + soup = BeautifulSoup(html_page, 'lxml') + url_links = [] + for a_tag in soup.body.select('a[href]'): + parser = urlparse(a_tag.attrs['href']) + scheme = parser.scheme or 'http' + netloc = parser.netloc or domain + if scheme != 'javascript' and netloc == domain: + path = parser.path + query = '?' + parser.query if parser.query else '' + full_url = f'{scheme}://{netloc}{path}{query}' + if full_url not in visited_urls: + url_links.append(full_url) + return url_links + + def extract(self, html_page): + pass + + def store(self, data_dict): + pass + + +class SpiderThread(Thread): + + def __init__(self, name, spider, tasks_queue): + super().__init__(name=name, daemon=True) + self.spider = spider + self.tasks_queue = tasks_queue + + def run(self): + while True: + current_url = self.tasks_queue.get() + visited_urls.add(current_url) + self.spider.status = SpiderStatus.WORKING + html_page = self.spider.fetch(current_url) + if html_page not in [None, '']: + url_links = self.spider.parse(html_page) + for url_link in url_links: + self.tasks_queue.put(url_link) + self.spider.status = SpiderStatus.IDLE + + +def is_any_alive(spider_threads): + return any([spider_thread.spider.status == SpiderStatus.WORKING + for spider_thread in spider_threads]) + + +visited_urls = set() + + +def main(): + task_queue = Queue() + task_queue.put('http://m.sohu.com/') + spider_threads = [SpiderThread('thread-%d' % i, Spider(), task_queue) + for i in range(10)] + for spider_thread in spider_threads: + spider_thread.start() + + while not task_queue.empty() or is_any_alive(spider_threads): + sleep(5) + + print('Over!') + + +if __name__ == '__main__': + main() + +``` + diff --git "a/Day66-75/07.\346\225\260\346\215\256\346\270\205\346\264\227.md" "b/Day66-75/07.\346\225\260\346\215\256\346\270\205\346\264\227.md" deleted file mode 100644 index 0d449da3d8068feccab176edf08752604ffb440e..0000000000000000000000000000000000000000 --- "a/Day66-75/07.\346\225\260\346\215\256\346\270\205\346\264\227.md" +++ /dev/null @@ -1,2 +0,0 @@ -## 数据清洗 - diff --git "a/Day66-75/Scrapy\347\232\204\345\272\224\347\224\25001.md" "b/Day66-75/Scrapy\347\232\204\345\272\224\347\224\25001.md" index e69de29bb2d1d6434b8b29ae775ad8c2e48c5391..e66f5ff86215a96ae615a7badd2a8aa73fc52631 100644 --- "a/Day66-75/Scrapy\347\232\204\345\272\224\347\224\25001.md" +++ "b/Day66-75/Scrapy\347\232\204\345\272\224\347\224\25001.md" @@ -0,0 +1,76 @@ +## Scrapy的应用(01) + +### Scrapy概述 + +Scrapy是Python开发的一个非常流行的网络爬虫框架，可以用来抓取Web站点并从页面中提取结构化的数据，被广泛的用于数据挖掘、数据监测和自动化测试等领域。下图展示了Scrapy的基本架构，其中包含了主要组件和系统的数据处理流程（图中的绿色箭头）。 + +![](./res/scrapy-architecture.jpg) + +#### 组件 + +1. Scrapy引擎（Engine）：Scrapy引擎是用来控制整个系统的数据处理流程。 +2. 调度器（Scheduler）：调度器从Scrapy引擎接受请求并排序列入队列，并在Scrapy引擎发出请求后返还给它们。 +3. 下载器（Downloader）：下载器的主要职责是抓取网页并将网页内容返还给蜘蛛（Spiders）。 +4. 蜘蛛（Spiders）：蜘蛛是有Scrapy用户自定义的用来解析网页并抓取特定URL返回的内容的类，每个蜘蛛都能处理一个域名或一组域名，简单的说就是用来定义特定网站的抓取和解析规则。 +5. 条目管道（Item Pipeline）：条目管道的主要责任是负责处理有蜘蛛从网页中抽取的数据条目，它的主要任务是清理、验证和存储数据。当页面被蜘蛛解析后，将被发送到条目管道，并经过几个特定的次序处理数据。每个条目管道组件都是一个Python类，它们获取了数据条目并执行对数据条目进行处理的方法，同时还需要确定是否需要在条目管道中继续执行下一步或是直接丢弃掉不处理。条目管道通常执行的任务有：清理HTML数据、验证解析到的数据（检查条目是否包含必要的字段）、检查是不是重复数据（如果重复就丢弃）、将解析到的数据存储到数据库（关系型数据库或NoSQL数据库）中。 +6. 中间件（Middlewares）：中间件是介于Scrapy引擎和其他组件之间的一个钩子框架，主要是为了提供自定义的代码来拓展Scrapy的功能，包括下载器中间件和蜘蛛中间件。 + +#### 数据处理流程 + +Scrapy的整个数据处理流程由Scrapy引擎进行控制，通常的运转流程包括以下的步骤： + +1. 引擎询问蜘蛛需要处理哪个网站，并让蜘蛛将第一个需要处理的URL交给它。 +2. 引擎让调度器将需要处理的URL放在队列中。 +3. 引擎从调度那获取接下来进行爬取的页面。 +4. 调度将下一个爬取的URL返回给引擎，引擎将它通过下载中间件发送到下载器。 +5. 当网页被下载器下载完成以后，响应内容通过下载中间件被发送到引擎；如果下载失败了，引擎会通知调度器记录这个URL，待会再重新下载。 +6. 引擎收到下载器的响应并将它通过蜘蛛中间件发送到蜘蛛进行处理。 +7. 蜘蛛处理响应并返回爬取到的数据条目，此外还要将需要跟进的新的URL发送给引擎。 +8. 引擎将抓取到的数据条目送入条目管道，把新的URL发送给调度器放入队列中。 +9. 上述操作会一直重复直到调度器中没有需要请求的URL，爬虫停止工作。 + +### 安装和使用Scrapy + +可以先创建虚拟环境并在虚拟环境下使用pip安装scrapy。 + +```Shell + +$ +``` + +项目的目录结构如下图所示。 + +```Shell + +(venv) $ tree +. +|____ scrapy.cfg +|____ qianmu +| |____ spiders +| | |____ __init__.py +| | |____ __pycache__ +| |____ __init__.py +| |____ __pycache__ +| |____ middlewares.py +| |____ settings.py +| |____ items.py +| |____ pipelines.py +``` + +> 说明：Windows系统的命令行提示符下有tree命令，但是Linux和MacOS的终端是没有tree命令的，可以用下面给出的命令来定义tree命令，其实是对find命令进行了定制并别名为tree。 +> +> `alias tree="find . -print | sed -e 's;[^/]*/;|____;g;s;____|; |;g'"` +> +> Linux系统也可以通过yum或其他的包管理工具来安装tree。 +> +> `yum install tree` + +根据刚才描述的数据处理流程，基本上需要我们做的有以下几件事情： + +1. 在items.py文件中定义字段，这些字段用来保存数据，方便后续的操作。 +2. 在spiders文件夹中编写自己的爬虫。 +3. 在pipelines.py中完成对数据进行持久化的操作。 +4. 修改settings.py文件对项目进行配置。 + + + diff --git a/Day66-75/code/main.py b/Day66-75/code/main.py index 571a14dc7478363d715d4f67c940bbfe6fa6bd96..51defece9a90a8504bc7f03560f9481d77ab39b4 100644 --- a/Day66-75/code/main.py +++ b/Day66-75/code/main.py @@ -123,7 +123,7 @@ def main(): spider_thread.start() while not task_queue.empty() or is_any_alive(spider_threads): - pass + sleep(5) print('Over!') diff --git a/Day66-75/code/main_redis.py b/Day66-75/code/main_redis.py index b73a6bcc8473128dfa3e747bbafc9df828cbeade..85a262b1b315f047d373602613cf300462a656b9 100644 --- a/Day66-75/code/main_redis.py +++ b/Day66-75/code/main_redis.py @@ -124,7 +124,7 @@ def is_any_alive(spider_threads): for spider_thread in spider_threads]) -redis_client = redis.Redis(host='120.77.222.217', +redis_client = redis.Redis(host='1.2.3.4', port=6379, password='1qaz2wsx') mongo_client = pymongo.MongoClient(host='120.77.222.217', port=27017) db = mongo_client.msohu diff --git a/Day66-75/res/scrapy-architecture.jpg b/Day66-75/res/scrapy-architecture.jpg new file mode 100644 index 0000000000000000000000000000000000000000..403f963e3f06752b3df97a56a3830ccb116a1b80 Binary files /dev/null and b/Day66-75/res/scrapy-architecture.jpg differ