Python 网络爬虫技术基础与实战

Python 网络爬虫技术基础与实战 | 极客日志

from urllib.request import urlopen
url = 'http://www.baidu.com'
response = urlopen(url)
print(response.read().decode('utf-8'))

pip install requests

import requests
url = "http://www.baidu.com"
response = requests.get(url)
response.encoding = "utf-8"
print(response.text)

import requests
content = input("请输入要搜索的内容：")
url = f"http://www.baidu.com/s?wd={content}"
# 请求头，模拟浏览器访问
headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"}
response = requests.get(url, headers=headers)
print(response.text)

import requests
url = "https://fanyi.baidu.com/sug"
data = {"kw": input("请输入要翻译的单词：")}
response = requests.post(url, data=data)
print(response.json()) # 返回的是 json 格式数据，所以使用 json() 方法

import requests
url = "https://movie.douban.com/j/chart/top_list"
data = {"type": 5, "interval_id": "100:90", "action": "", "start": 0, "limit": 20}
headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/101.0.4951.64 Safari/537.36"}
response = requests.get(url, params=data, headers=headers)
print(response.json())

import requests
session = requests.Session()
url = "http://www.baidu.com"
data = {"username": "username", "password": "password"}
response = session.post(url, data=data)
print(response.text)

import requests
url = "http://www.baidu.com"
proxy = {"http": "http://127.0.0.1:10809", "https": "http://127.0.0.1:10809"}
r = requests.get(url, proxies=proxy)
print(r.text)

符号	含义
.	匹配除换行符以外的任意字符
\w	字母、数字、下划线
\s	空白符
\d	数字
\n	换行符
\t	制表符
^	字符串开头
$	字符串结尾
\W	非字母、非数字、非下划线
\D	非数字
\S	非空白符
a	b
()	分组，匹配括号内表达式
[...]	匹配字符串中的字符，如 a-z
[^...]	匹配除字符串中字符的所有字符

符号	含义
*	重复 0 次或更多次数
+	重复 1 次或更多次数
?	重复 0 或 1 次
{n}	重复 n 次
{n,}	重复 n 次或更多次数
{n,m}	重复 n 到 m 次
.*	贪婪匹配，尽可能多的匹配
.*?	惰性匹配，尽可能少的匹配

import re
text = "我的电话号码是 1234567890 和 1234567891，请不要告诉别人。"
pattern = r"\d{10}"
result = re.findall(pattern, text)
print(result)

import re
text = "我的电话号码是 1234567890 和 1234567891，请不要告诉别人。"
pattern = r"\d{10}"
result = re.finditer(pattern, text)
for match in result:
    print(match.group()) # 获取匹配的字符串

import re
text = "我的电话号码是 1234567890 和 1234567891，请不要告诉别人。"
pattern = r"\d{10}"
result = re.search(pattern, text)
print(result.group()) # 获取匹配的字符串

import re
text = "我的电话号码是 1234567890 和 1234567891，请不要告诉别人。"
pattern = r"\d{10}"
result = re.match(pattern, text)
print(result) # 获取匹配的字符串

import re
pattern_obj = re.compile(r"\d{10}")
text = "我的电话号码是 1234567890 和 1234567891，请不要告诉别人。"
result1 = pattern_obj.findall(text)
result2 = pattern_obj.finditer(text)
result3 = pattern_obj.search(text)
print(result1)
print(list(result2))
print(result3.group())

import re
pattern_obj = re.compile(r'<div class="(?P<class>.*?)">(?P<content>.*?)</div>')
s = """ <div>这是一个标题</div>
<div>这是一个内容</div> """
result = pattern_obj.finditer(s)
for item in result:
    class_name = item.group("class")
    content = item.group("content")
    print(f"类名：{class_name}，内容：{content}")

from bs4 import BeautifulSoup
html = """ <div>这是一个标题</div>
<div>这是一个内容</div> """
page = BeautifulSoup(html, "html.parser")
div = page.find("div", class_="title")
print(div.text)
divs = page.find_all("div")
classes = page.find_all("div", class_=True)
print(classes)
for div in divs:
    print(div.text)
    print(div.get("class"))

from lxml import etree
xml = """ <books>
<id>1</id>
<author>
<nickname>Neo</nickname>
<nickname>小王子</nickname>
<nickname>Python 爬虫基础</nickname>
</author>
<price>100</price>
<name>Python 爬虫基础</name>
</books> """
tree = etree.XML(xml)
et = tree.xpath('//nickname/text()')
result = tree.xpath('//nickname') # 获取下级所有节点
for item in result:
    print(item.text)
one = tree.xpath('//nickname[@class="1"]/text()')[0]

from pyquery import PyQuery as pq
html = """ <div>
<ul>
<li>first item</li>
<li><a href="link2.html">second item</a></li>
<li><a href="link3.html"><span>third item</span></a></li>
<li><a href="link4.html">fourth item</a></li>
<li><a href="link5.html">fifth item</a></li>
</ul>
</div> """
doc = pq(html)
a = doc("li")
b = doc("li a")
c = doc("li a span")
class_names = doc("li").attr("class")
item_text = doc("li").text()
print(class_names)
print(item_text)
print(a)
print(b)
print(c)

from pyquery import PyQuery as pq
html = """ <div>这是一个标题</div>
<div>这是一个内容</div> """
doc = pq(html)
after = doc(".title").after("<div>这是一个 after 标签</div>")
before = doc(".title").before("<div>这是一个 before 标签</div>")
print(doc)
remove = doc(".title").remove()
remove_attr = doc(".content").remove_attr("class")
print(doc)
add_class = doc(".after").add_class("new-class")
add_attr = doc(".after").attr("data-id", "123")
print(doc)

from threading import Thread

def task():
    for i in range(5):
        print(f"线程任务执行第 {i+1} 次")

thread = Thread(target=task) # 创建一个线程对象，target 参数指定线程要执行的函数
thread.start() # 启动线程，线程开始执行任务

for i in range(5):
    print(f"主线程执行第 {i+1} 次")

from multiprocessing import Process

def task():
    for i in range(5):
        print(f"子进程任务执行第 {i+1} 次")

pro = Process(target=task) # 创建一个进程对象，target 参数指定进程要执行的函数
pro.start() # 启动进程，进程开始执行任务

for i in range(5):
    print(f"主进程执行第 {i+1} 次")

from concurrent.futures import ThreadPoolExecutor

def task(n):
    print(f"线程任务执行第 {n} 次")

with ThreadPoolExecutor(max_workers=5) as executor: # 创建一个线程池，max_workers 参数指定线程池中线程的最大数量
    for i in range(40):
        executor.submit(task, i+1) # 提交任务到线程池，参数通过 submit 方法传递
        print(f"主线程提交了第 {i+1} 个任务")

import asyncio

async def task(n):
    for i in range(10):
        print(f"异步任务执行第 {n} 次，第 {i+1} 次")
    await asyncio.sleep(1) # 模拟异步任务的耗时操作，可以使用 asyncio.sleep() 来实现非阻塞的睡眠

async def task2(n):
    for i in range(10):
        print(f"异步任务 2 执行第 {n} 次，第 {i+1} 次")
    await asyncio.sleep(1)

async def task3(n):
    for i in range(10):
        print(f"异步任务 3 执行第 {n} 次，第 {i+1} 次")
    await asyncio.sleep(1)

await asyncio.gather(task(1), task2(2), task3(3)) # 使用 asyncio.gather() 来并行执行多个异步任务

import asyncio

async def task(n):
    for i in range(10):
        print(f"异步任务执行第 {n} 次，第 {i+1} 次")
    await asyncio.sleep(1)

async def task2(n):
    for i in range(10):
        print(f"异步任务 2 执行第 {n} 次，第 {i+1} 次")
    await asyncio.sleep(1)

async def task3(n):
    for i in range(10):
        print(f"异步任务 3 执行第 {n} 次，第 {i+1} 次")
    await asyncio.sleep(1)

async def main():
    tasks = [asyncio.create_task(task(1)), asyncio.create_task(task2(2)), asyncio.create_task(task3(3))] # 创建一个包含多个异步任务的列表
    await asyncio.wait(tasks) # 等待所有异步任务完成

if __name__ == "__main__":
    asyncio.run(main()) # 运行主函数，启动事件循环，执行异步任务

import asyncio
import aiohttp

async def fetch(url):
    async with aiohttp.ClientSession() as session: # 创建一个异步 HTTP 会话
        async with session.get(url) as response: # 发送 GET 请求，获取响应对象
            return await response.text() # 获取响应内容，使用 await 等待异步操作完成

async def main():
    url = "http://www.baidu.com"
    content = await fetch(url) # 调用 fetch 函数，获取网页内容，使用 await 等待异步操作完成
    print(content)

if __name__ == "__main__":
    asyncio.run(main())

Python 网络爬虫技术基础与实战

一、什么是爬虫

二、基础

（一）获取网页源代码

（二）网页加载方式

三、requests 模块

（一）安装

（二）使用

（三）变量访问与伪装

（四）post 请求获取翻译信息

（五）get 请求获取信息

（六）cookie 处理

（七）防盗链

（八）代理使用

四、数据处理

（一）正则表达式

1. 元字符

2. 量词

3. 使用：re 库

（二）使用 BeautifulSoup 处理

（三）xpath 处理 (lxml)

（四）pyquery

五、提高效率

（一）多线程

（二）多进程

（三）线程池

（四）协程

aiohttp

更多推荐文章

相关免费在线工具

Python 网络爬虫技术基础与实战

一、什么是爬虫

二、基础

（一）获取网页源代码

（二）网页加载方式

三、requests 模块

（一）安装

（二）使用

（三）变量访问与伪装

（四）post 请求获取翻译信息

（五）get 请求获取信息

（六）cookie 处理

（七）防盗链

（八）代理使用

四、数据处理

（一）正则表达式

1. 元字符

2. 量词

3. 使用：re 库

（二）使用 BeautifulSoup 处理

（三）xpath 处理 (lxml)

（四）pyquery

五、提高效率

（一）多线程

（二）多进程

（三）线程池

（四）协程

aiohttp

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具