python---JSON与HTML解析入门；DrissionPage自动化操作；多线程与异步编程------讲解
·
今天将分享三个非常实用的 Python 技能点：JSON与HTML解析入门、轻量级自动化工具DrissionPage，以及多线程与异步编程。
一、JSON与HTML解析入门
1.JSON 常见解析
在Python中，字典的格式和json格式在显示上是一样的，但是字典是dict，而json是str
Python的json模块主要包含两大函数。
- json.dumps把一个Python对象编码转换成json字符串
- json.loads把json格式字符串解码转换成Python对象
1.1 json.dumps
dic1 = {'type':'dic1','username':'张三','age':25}
json_dic1 = json.dumps(dic1)
print(json_dic1,type(json_dic1))
json_dic2 = json.dumps(dic1,sort_keys=True,indent =4,separators=(',', ': '),
ensure_ascii=True )
print(json_dic2,type(json_dic2))
json_dic3 = json.dumps(dic1,ensure_ascii=False )
print(json_dic3,type(json_dic3))
#将python对象test转换json对象
test = [{'type':'dic1','username':'孙成全','age':25},(2,3),1]
print(type(test))
python_to_json = json.dumps(test,ensure_ascii=False)
print(python_to_json,type(python_to_json))
'''
输出
{"type": "dic1", "username": "\u5f20\u4e09", "age": 25} <class 'str'>
{
"age": 25,
"type": "dic1",
"username": "\u5f20\u4e09"
} <class 'str'>
{"type": "dic1", "username": "张三", "age": 25} <class 'str'>
<class 'list'>
[{"type": "dic1", "username": "孙成全", "age": 25}, [2, 3], 1] <class 'str'>
'''
json.dumps 参数解释：
ensure_ascii：默认值True，如果dict内含有non-ASCII的字符，则会类似\uXXXX的显示数据，设置成False后，就能正常显示。
indent：应该是一个非负的整型，如果是0，或者为空，则一行显示数据，否则会换行且按照indent的数量显示前面的空白，这样打印出来的json数据也叫pretty-printed
json。 sort_keys：将数据根据keys的值进行排序。 separators：分隔符，实际上是(item_separator,
dict_separator)的一个元组，默认的就是(‘,’,’:’)；这表示dictionary内keys之间用“,”隔开，而KEY和value之间用“：”隔开。
1.2 json.loads
# 将json对象转换成python对象
json_to_python = json.loads(python_to_json)
print(json_to_python,type(json_to_python))
print(json_to_python[0]['username'])
'''
输出
[{'type': 'dic1', 'username': '孙成全', 'age': 25}, [2, 3], 1] <class 'list'>
孙成全
'''
1.3 读取JSON文件
with open("data.json", "r", encoding="utf-8") as f:
data = json.load(f)
1.4 写入JSON文件
with open("output.json", "w", encoding="utf-8") as f:
json.dump(python_data, f, ensure_ascii=False, indent=4)
2.HTML解析
2.1 HTML基础概念
HTML（超文本标记语言）是网页的骨架，用标签组织内容。想象一下图书馆的书籍分类系统：每本书都有标签（tag）、分类号（class）、（ID）。
核心概念：标签(tag)是元素类型，类名(class)用于样式分组，ID是唯一标识符。
主流方案有BeautifulSoup4（BS4）和lxml。
2.2 安装
pip install beautifulsoup4
pip install lxml # 推荐使用 lxml 作为解析器（速度更快）
如果你没有 lxml，可以使用 Python 内置的 html.parser 作为解析器。
2.3. BeautifulSoup对Html 文件分析
在使用BeautiSoup对其进行解析，解析的时候要使用相应类型的解析器html.parser或者lxml
from bs4 import BeautifulSoup
import requests
# 使用 requests 获取网页内容
url = 'https://cn.bing.com/' # 抓取bing搜索引擎的网页内容
response = requests.get(url)
# 使用 BeautifulSoup 解析网页
soup = BeautifulSoup(response.text, 'lxml') # 使用 lxml 解析器
# 解析网页内容 html.parser 解析器
# soup = BeautifulSoup(response.text, 'html.parser')
具体的BeautifulSoup 属性与方法 可以参考–https://www.runoob.com/python3/python-spider-beautifulsoup.html
2.4 lxml对Html 文件分析----lxml.etree解析html
这里用到etree.HTML方法把html的文本内容解析成html对象
要打印html内容，可以用etree.tostring方法，encoding=”utf-8”参数可以正常输出html里面的中文内容。pretty_print=True是以标准格式输出
# coding:utf-8
from lxml import etree
htmldemo = '''
<meta charset="UTF-8"> <!-- for HTML5 -->
<meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
<html><head><title>yoyo ketang</title></head>
<body>
<b><!--Hey, this in comment!--></b>
<p class="title"><b>yoyoketang</b></p>
<p class="yoyo">这里是我的空间：LLLLLL
<a href="http://www.cnblogs.com/yoyoketang/tag/fiddler/" class="sister" id="link1">fiddler教程</a>,
<a href="http://www.cnblogs.com/yoyoketang/tag/python/" class="sister" id="link2">python笔记</a>,
<a href="http://www.cnblogs.com/yoyoketang/tag/selenium/" class="sister" id="link3">selenium文档</a>;
快来关注吧！</p>
<p class="story">...</p>
# etree.HTML解析html内容
demo = etree.HTML(htmldemo)
nodes = demo.xpath('//p[@class="yoyo"]')
# 获取文本
t = nodes[0].text
print(t)
运行结果：
这里是我的空间：LLLLLL
nodes是xpath定位获取到的一个list对象，会找出所有符合条件的元素对象。可以用for 循环查看详情
# coding:utf-8
from lxml import etree
htmldemo = '''
复制上面的html内容
'''
# etree.HTML解析html内容
demo = etree.HTML(htmldemo)
nodes = demo.xpath('//p[@class="yoyo"]')
print(nodes) # list对象
for i in nodes:
# 打印定位到的内容
print(etree.tostring(i, encoding="utf-8", pretty_print=True).decode("utf-8"))
# 元素属性，字典格式
print(i.attrib)
二、DrissionPage 常规操作
** 文档：https://drissionpage.cn/browser_control/intro
DrissionPage是一个混合驱动的网页自动化库，核心特点：
- 双引擎驱动：支持Selenium（浏览器渲染）和Requests（HTTP请求）无缝切换
- 简洁API：类似jQuery的选择器语法，操作更直观
- 智能等待：自动处理元素加载延迟
- 多浏览器支持：Chrome/Firefox/Edge等
适合场景：
- 网页数据采集
- 自动化测试
- 表单填写
- 文件上传/下载
- 任何需要模拟用户操作的场景
2.1 安装
pip install DrissionPage
2.2 连接浏览器与启动设置*
2.2.1 基础启动方式
from DrissionPage import ChromiumPage, ChromiumOptions
# ===== 方式1：简洁启动 =====
page = ChromiumPage() # 默认启动 Chrome，有界面模式
# ===== 方式2：自定义配置启动 =====
co = ChromiumOptions()
# 无头模式（后台运行）
co.headless(True)
# 设置启动端口（多开浏览器时必须指定不同端口）
co.set_local_port(9222)
# 设置用户数据目录（保持登录状态）
co.set_user_data_path(r'C:\Users\YourName\ChromeData')
#设置浏览器可执行文件路径
co.set_browser_path('你的浏览器路径')
# 设置 User-Agent（反检测）
co.set_user_agent('Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36')
# 添加启动参数
co.set_argument('--no-sandbox') # Linux 必需
co.set_argument('--disable-gpu') # 禁用 GPU 加速
co.set_argument('--window-size', '1920,1080')
# 使用配置创建浏览器
page = ChromiumPage(chromium_options=co,timeout=3)
2.2.2 浏览器对象与基础操作
**
（1） 访问网页
from DrissionPage import ChromiumPage
page = ChromiumPage()
# 基本访问
page.get('https://www.baidu.com')
# 设置超时时间（秒）
page.get('https://example.com', timeout=20)
# 设置重试次数
page.get('https://example.com', retry=3)
（2） 获取网页信息
# 页面基础信息
print(page.title) # 页面标题
print(page.url) # 当前 URL
print(page.html) # 页面源码
print(page.json) # 如果返回的是 JSON，自动解析
# 浏览器信息
print(page.tabs_count) # 标签页数量
print(page.tab_ids) # 所有标签页 ID 列表
# 截图
page.screenshot('full_page.png', full_page=True) # 整页截图
page.screenshot('viewport.png') # 视口截图
（3）页面操作
# 刷新
page.refresh()
# 前进/后退
page.forward()
page.back()
# 停止加载
page.stop_loading()
# 执行 JavaScript
result = page.run_js('return document.title')
page.run_js('window.scrollTo(0, document.body.scrollHeight)') # 滚动到底部
# 设置窗口大小
page.set.window.size(1920, 1080)
page.set.window.maximize()
page.set.window.minimize()
（4）查找元素
from DrissionPage import ChromiumPage
page = ChromiumPage()
page.get('https://example.com')
# ===== 查找单个元素 =====
# CSS 选择器
ele = page.ele('#username') # ID
ele = page.ele('.classname') # 类名
ele = page.ele('input[name="email"]') # 属性选择器
# XPath
ele = page.ele('//input[@id="username"]')
ele = page.ele('//div[@class="item"][2]') # 第二个匹配项
# ===== 查找多个元素 =====
eles = page.eles('.item') # 所有 class=item 的元素
eles = page.eles('tag:li') # 所有 li 标签
for ele in eles:
print(ele.text)
（5）获取元素信息
ele = page.ele('#example')
# 基础信息
print(ele.text) # 可见文本内容
print(ele.html) # 元素 HTML 源码
print(ele.tag) # 标签名
# 属性获取
print(ele.attr('href')) # 获取属性值
print(ele.attr('data-id'))
（6）监听网络数据
a.基础监听
from DrissionPage import ChromiumPage
page = ChromiumPage()
#用于启动监听器，启动同时可设置获取的目标特征
# ⚠️ 重要：必须先启动监听，再执行动作
page.listen.start('gitee.com/explore') # 监听包含指定文本的 URL
page.get('https://gitee.com/explore') # 这行产生的数据包不会被监听
# 点击下一页，触发请求
for _ in range(5):
page('@rel=next').click()
# 等待并获取一个数据包
res = page.listen.wait()
print(f'URL: {res.url}')
print(f'状态: {res.response.statusText}')
print(f'响应头: {res.response.headers}')
print(f'响应体: {res.response.body}') # JSON 或文本内容
b.监听多个接口配置
# 监听多个模式
page.listen.start(['api.example.com', 'graphql']) # 多个关键词
# 等待所有指定的请求完成
packets = page.listen.wait_silent()
# 停止监听
page.listen.stop()
三、多线程、线程池与异步
**
3.1 多线程（Threading）
文档 添加链接描述
什么是多线程？
Python 的 threading 模块基于操作系统原生线程
GIL 限制：同一时刻只有一个线程执行 Python 字节码，但 I/O 等待时会释放 GIL
适用场景：I/O 密集型任务（网络请求、文件读写、数据库操作）
基础功能：
- threading.Thread()：thread_obj = threading.Thread(target=函数名, args=(参数元组,)) 创建线程对象
- target: 指定线程启动后要执行的函数。
- args: 传递给 target 函数的参数，必须是元组类型。如果只有一个参数，需要写成
(参数,) 的形式。 - start()：启动线程（非阻塞）
- join()：等待线程结束（阻塞）
- 线程间共享内存，需要注意线程安全问题（可使用
threading.Lock）
import threading
import time
def task(name, delay):
"""模拟一个耗时任务"""
print(f"线程 {name} 开始执行")
time.sleep(delay)
print(f"线程 {name} 执行完毕")
# 创建线程
threads = []
for i in range(3):
t = threading.Thread(target=task, args=(f"Thread-{i}", i+1))
threads.append(t)
t.start() # 启动线程
# 等待所有线程完成
for t in threads:
t.join()
print("所有线程执行完毕")
3.1.1 线程同步
在多线程编程中，多个线程可能会同时访问共享资源，这可能导致数据不一致的问题。为了避免这种情况，可以使用线程同步机制，如锁（Lock）。
import threading
# 创建一个锁对象
lock = threading.Lock()
def my_function():
with lock:
print("线程开始执行")
# 在这里编写线程要执行的代码
print("线程执行结束")
# 创建线程实例
thread1 = threading.Thread(target=my_function)
thread2 = threading.Thread(target=my_function)
# 启动线程
thread1.start()
thread2.start()
# 等待线程执行完毕
thread1.join()
thread2.join()
print("主线程结束")
3.1.2 线程间通信
线程间通信可以通过队列（Queue）来实现。Queue 是线程安全的，可以在多个线程之间安全地传递数据。
import threading
import queue
def worker(q):
while not q.empty():
item = q.get()
print(f"处理项目: {item}")
q.task_done()
# 创建一个队列并填充数据
q = queue.Queue()
for i in range(10):
q.put(i)
# 创建线程实例
thread1 = threading.Thread(target=worker, args=(q,))
thread2 = threading.Thread(target=worker, args=(q,))
# 启动线程
thread1.start()
thread2.start()
# 等待队列中的所有项目被处理完毕
q.join()
print("所有项目处理完毕")
**3.2 线程池（ThreadPoolExecutor）**文档 添加链接描述
核心 API：concurrent.futures.ThreadPoolExecutor，支持map()（批量任务）、submit()（单个任务）。
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import os
# 定义单个任务：模拟文件处理
def process_file(file_name):
"""
功能：模拟文件读取和处理（I/O密集型任务）
:param file_name: 文件名
:return: 处理结果
"""
try:
start_time = time.time()
# 模拟文件读取（I/O操作）
time.sleep(0.5) # 代替实际的文件读写耗时
file_size = os.path.getsize(file_name) if os.path.exists(file_name) else 0
cost = round(time.time() - start_time, 2)
return {
"file": file_name,
"status": "成功",
"size": file_size,
"cost": cost
}
except Exception as e:
return {
"file": file_name,
"status": f"失败：{str(e)}",
"size": 0,
"cost": 0
}
if __name__ == "__main__":
# 模拟待处理的文件列表
file_list = [f"file_{i}.txt" for i in range(1, 11)] # 10个文件
# 记录总开始时间
total_start = time.time()
# 1. 创建线程池：max_workers指定最大并发数（建议设为CPU核心数*5~10，I/O密集型）
with ThreadPoolExecutor(max_workers=3) as executor:
# 方式1：submit提交单个任务（灵活，可获取结果）
task_list = []
for file in file_list:
# submit返回Future对象（代表未来的结果）
task = executor.submit(process_file, file)
task_list.append(task)
# 2. 遍历已完成的任务（按完成顺序，非提交顺序）
print("===== 任务执行结果（按完成顺序） =====")
for future in as_completed(task_list):
result = future.result() # 获取任务结果（阻塞直到任务完成）
print(f"文件：{result['file']} | 状态：{result['status']} | 大小：{result['size']}B | 耗时：{result['cost']}s")
# 总耗时
total_cost = round(time.time() - total_start, 2)
print(f"\n总耗时：{total_cost}s（并发数3，10个任务各耗时0.5s，串行需5s，并发约1.7s）")
- with ThreadPoolExecutor(…)：上下文管理器，自动管理线程池（无需手动关闭）。
- max_workers：最大并发数，I/O 密集型任务可设为5~20（CPU 密集型建议设为 CPU 核心数）。
- submit()：提交单个任务，返回Future对象，通过result()获取结果（会阻塞）。
- as_completed()：按任务完成顺序遍历Future对象，比map()更灵活（可处理异常、获取耗时）
**3.3 异步编程（Asyncio）**文档 添加链接描述
关键概念：
- async def：定义协程函数
- await：等待异步任务完成
- asyncio.run()：启动事件循环
- asyncio.gather()：并发执行多个协程
import asyncio
import aiohttp
import time
async def fetch(session, url, idx):
"""
异步获取 URL
功能：展示基本的 async/await 用法
"""
print(f"[{idx}] 🚀 开始请求: {url}")
start = time.time()
try:
# async with: 异步上下文管理器
async with session.get(url) as response:
# await: 挂起当前协程，等待 I/O 完成
content = await response.text()
elapsed = time.time() - start
print(f"[{idx}] ✅ 完成: {len(content)} bytes, 耗时 {elapsed:.2f}s")
return {
'idx': idx,
'url': url,
'status': response.status,
'length': len(content)
}
except Exception as e:
print(f"[{idx}] ❌ 失败: {e}")
return {'idx': idx, 'url': url, 'error': str(e)}
async def main_basic():
"""基础异步演示"""
urls = [
'https://www.python.org',
'https://www.github.com',
'https://www.gitlab.com',
'https://www.stackoverflow.com'
]
# 创建 HTTP 会话（连接池）
async with aiohttp.ClientSession() as session:
# 创建任务列表
tasks = [fetch(session, url, i+1) for i, url in enumerate(urls)]
# gather: 并发执行所有任务，等待全部完成
results = await asyncio.gather(*tasks, return_exceptions=True)
# 处理结果
success = [r for r in results if isinstance(r, dict) and 'error' not in r]
print(f"\n🎉 成功: {len(success)}/{len(urls)}")
return results
# 启动事件循环
if __name__ == '__main__':
start = time.time()
asyncio.run(main_basic())
print(f"总耗时: {time.time() - start:.2f}s")
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
12
0- 0
已为社区贡献1条内容
所有评论(0)