📚 Termux 无root学习系列
阶段一：环境搭建与终端基础 ✅ 已完成
01-05 Termux 安装 → 包管理 → 文件管理 → Vim/tmux
阶段二：编程开发环境 ✅ 已完成
06-09 Python → C/C++ → Node.js → Git
阶段三：网络与远程操作 ✅ 已完成
10-12 SSH → 网络工具 → 代理与安全通信
阶段四：Termux:API 与自动化 ✅ 已完成
13-14 硬件调用 → 定时任务自动化
阶段五：proot-distro 完整 Linux ✅ 已完成
15-17 proot-distro → Jupyter → 桌面环境 VNC
阶段六：AI 与 AIGC 实战 ✅ 已完成
18-20 LLM API → llama.cpp 本地推理 → RAG 知识库
阶段七：安全工具实战 ✅ 已完成
21-23 信息收集 → sqlmap/hydra → 哈希破解与取证
阶段八：自托管服务与综合实战（进行中）
✓ 24 Gitea 私有 Git 服务器与静态站托管
✓ 25 Telegram Bot 开发部署
26 Web 爬虫与数据采集管道（当前篇）
阶段九：综合项目与运维（共 2 篇，后续持续更新）
01 requests 基础与网页抓取
爬虫的第一步是「发请求、拿响应」。requests 是 Python 最流行的 HTTP 客户端库，语法简洁直观，用它来打开网页就像在浏览器地址栏输入网址一样自然。
1.1 安装 requests
在 Termux 原生 Python 环境中直接用 pip 安装。截至 2026 年 5 月，最新稳定版为 2.34.2：
Termux
pip install requests==2.34.2
# 验证安装
python3 -c "import requests; print(requests.__version__)"
1.2 发送 GET 请求
最基础的爬虫操作：用 GET 请求获取一个网页的 HTML 源码。我们以一个公开的测试站为例：
scraper01.py
import requests
url = "https://quotes.toscrape.com/"
response = requests.get(url)
print(f"状态码: {response.status_code}")
print(f"编码: {response.encoding}")
print(f"内容长度: {len(response.text)} 字符")
# 打印前 500 个字符
print(response.text[:500])
输出（前几行）
状态码: 200
编码: utf-8
内容长度: 11053 字符
Quotes to Scrape
状态码 200 表示请求成功。常见的状态码还有 404（页面不存在）、403（被拒绝访问）、500（服务器错误）。写爬虫时一定要先检查状态码，确认拿到的是有效页面。
💡 小贴士
quotes.toscrape.com 是 Scrapy 官方提供的免费爬虫练习站，专门设计用来练习爬取技术，不会触发反爬封禁。学习爬虫时建议优先使用这类公开测试站点，避免对正规网站造成压力。
1.3 带参数的请求
很多网站通过 URL 查询参数（?key=value）来控制页面内容。requests 支持用 params 字典传参，自动编码：
scraper02.py
import requests
# 搜索 "love" 标签的名言
params = {"tag": "love"}
url = "https://quotes.toscrape.com/tag/{}/".format(params["tag"])
response = requests.get(url, params=params)
print(f"最终URL: {response.url}")
print(f"状态码: {response.status_code}")
03 进阶爬取与反爬策略
单页面抓取只是开始。实际项目中往往需要翻页、登录、处理动态内容，还要应对网站的反爬机制。本节介绍进阶技巧和爬虫伦理。
3.1 分页爬取
大多数列表页都有多页。常见的翻页方式有两种：URL 带页码参数（如 ?page=2），以及页面底部有「下一页」链接。下面是一个自动翻页抓取所有名言的示例：
paged_scraper.py
import requests
from bs4 import BeautifulSoup
import time
base_url = "https://quotes.toscrape.com"
all_quotes = []
def parse_page(url):
response = requests.get(url)
soup = BeautifulSoup(response.text, "lxml")
for quote in soup.select("div.quote"):
all_quotes.append({
"text": quote.select_one("span.text").get_text(strip=True),
"author": quote.select_one("small.author").get_text(strip=True),
})
# 找下一页链接
next_btn = soup.select_one("li.next a")
if next_btn:
return base_url + next_btn["href"]
return None
# 主循环：逐页抓取
current_url = base_url + "/"
page = 1
while current_url:
print(f"正在抓取第 {page} 页...")
current_url = parse_page(current_url)
page += 1
time.sleep(1) # 礼貌延时，别把服务器压垮
print(f"\n抓取完成！共 {len(all_quotes)} 条名言")
3.2 设置 User-Agent 和请求头
默认情况下，requests 的 User-Agent 是 python-requests/2.x，一眼就能被识别为爬虫。设置一个真实浏览器的 UA 是基本操作：
带请求头的请求
import requests
headers = {
"User-Agent": "Mozilla/5.0 (Linux; Android 14; Pixel 8) \
AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 \
Mobile Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
}
response = requests.get(url, headers=headers)
3.3 爬虫伦理与常见反爬
📜 爬虫伦理：爬取公开数据是合法的，但要遵守 robots.txt 协议、控制访问频率、不爬取隐私数据、不用于商业牟利。礼貌延时（1-3秒）是基本素养。
🚫 403 Forbidden：服务器拒绝访问。通常是 UA 被识别为爬虫、IP 被封禁，或缺少必要的 Cookie/Referer。
🕐 频率限制：短时间内请求过多会被限流。解决方法：time.sleep() 加延时、使用代理 IP 池、降低并发数。
🔐 登录验证：需要登录才能访问的页面，用 requests.Session() 保持 Cookie 会话，先 POST 登录接口再 GET 受保护页面。
⚠️ 常见错误
不检查状态码直接解析 — 404 或 500 页面的 HTML 结构完全不同，会导致 select_one 返回 None 然后报 AttributeError
✓ 正确：先判断 response.status_code == 200，非 200 时打印错误信息并跳过，不要假设每次请求都成功
04 Scrapy 框架入门
requests + BeautifulSoup 适合小型爬虫，但当爬取目标扩展到几十上百个页面、需要去重、并发、中间件、管道化处理时，手写框架会越来越臃肿。Scrapy 是 Python 生态中最成熟的爬虫框架，自带异步调度、数据管道、中间件扩展，适合中大型采集项目。
💡 比喻：如果 requests 是手工螺丝刀，Scrapy 就是一整套电动工具箱——拧一颗螺丝时手工更快，但要组装一整台机器时，电动工具的效率优势会指数级放大。
4.1 Termux 安装 Scrapy
Scrapy 依赖 lxml、cryptography 等 C 扩展库，在 Termux 原生环境安装可能遇到编译问题。推荐两种方案：
方案 A：proot Ubuntu（推荐）：在 proot-distro 的 Ubuntu 环境中直接 pip install scrapy，所有依赖自动编译安装，兼容性最好。
方案 B：Termux 原生：需要先安装编译依赖 pkg install libxml2 libxslt libffi openssl python，再 pip 安装，部分版本可能需要手动指定编译参数。
# proot Ubuntu 中安装 Scrapy
pip install scrapy==2.19.0
# 验证安装
scrapy version
# 输出：Scrapy 2.19.0
4.2 创建第一个 Scrapy 项目
Scrapy 有固定的项目结构，使用 startproject 命令一键生成骨架，然后用 genspider 创建爬虫文件。
# 创建项目
scrapy startproject quotes_scraper
cd quotes_scraper
# 生成爬虫（爬虫名 + 域名）
scrapy genspider quotes quotes.toscrape.com
生成的项目结构如下，每个文件各司其职：
quotes_scraper/
├── scrapy.cfg # 项目配置文件
└── quotes_scraper/
├── __init__.py
├── items.py # 数据结构定义
├── middlewares.py # 中间件（请求/响应拦截）
├── pipelines.py # 数据管道（清洗/存储）
├── settings.py # 全局设置
└── spiders/ # 爬虫目录
├── __init__.py
└── quotes.py # 我们刚生成的爬虫
4.3 编写 Spider 爬取名言
打开 spiders/quotes.py，Scrapy 已经生成了基础模板。我们来改写它，实现与之前 requests+BS4 版本相同的功能——爬取 quotes.toscrape.com 的名言和作者。
# spiders/quotes.py
import scrapy
class QuotesSpider(scrapy.Spider):
name = "quotes"
allowed_domains = ["quotes.toscrape.com"]
start_urls = ["https://quotes.toscrape.com/"]
def parse(self, response):
# 提取当前页所有名言
for quote in response.css("div.quote"):
yield {
"text": quote.css("span.text::text").get(),
"author": quote.css("small.author::text").get(),
}
# 找下一页链接，继续爬
next_page = response.css("li.next a::attr(href)").get()
if next_page:
yield response.follow(next_page, self.parse)
核心变化：Scrapy 用 yield 交出数据和新请求，调度器自动管理队列和并发，不需要手写 while 循环和 time.sleep。
# 运行爬虫并输出到 JSON
scrapy crawl quotes -o quotes.json
# 也可以导出 CSV、XML、JSON Lines
scrapy crawl quotes -o quotes.csv
scrapy crawl quotes -o quotes.jl
4.4 Scrapy 核心架构
理解 Scrapy 的数据流，才能用好它的扩展能力。数据从 Spider 出发，经过引擎调度、下载器获取、回到 Spider 解析，再通过管道落地。
🕷️ Spider（爬虫）：你写的解析逻辑，负责生成初始请求、解析响应、yield 数据和新请求。
⚙️ Engine（引擎）：中枢调度器，协调各组件之间的数据流，你一般不需要直接操作它。
📥 Downloader（下载器）：负责发请求拿响应，支持重试、重定向、并发控制。
📦 Item Pipeline（管道）：Spider yield 的数据流经管道，可以做清洗、去重、存数据库等。
🔌 Scheduler（调度器）：维护待爬 URL 队列，负责去重和优先级排序。
4.5 Item Pipeline 数据管道
Scrapy 的 Pipeline 可以串联多个处理步骤，每个 Pipeline 只做一件事，按优先级顺序执行。下面写一个去重管道和一个 SQLite 存储管道。
# pipelines.py — 去重管道
from scrapy.exceptions import DropItem
class DuplicatesPipeline:
def __init__(self):
self.seen = set()
def process_item(self, item, spider):
key = item["text"]
if key in self.seen:
raise DropItem(f"重复条目: {key[:30]}...")
self.seen.add(key)
return item
在 settings.py 中启用管道并设置优先级（数字越小越先执行）：
# settings.py
ITEM_PIPELINES = {
"quotes_scraper.pipelines.DuplicatesPipeline": 300,
}
# 并发控制（礼貌爬取）
CONCURRENT_REQUESTS = 2
DOWNLOAD_DELAY = 1 # 每次请求间隔 1 秒
RANDOMIZE_DOWNLOAD_DELAY = True
💡 小贴士：Termux 上跑 Scrapy 建议把 CONCURRENT_REQUESTS 设为 2-4，DOWNLOAD_DELAY 设为 1-2 秒，避免手机 CPU 和网络负载过高导致系统杀进程。
4.6 什么时候该用 Scrapy
不是所有爬虫都需要上框架。选工具看场景：
动态 JS 渲染页面
❌ 不支持
⚠️ 需配合 Playwright
手机 Termux 资源占用
✅ 轻量
⚠️ 较重
05 数据清洗与管道化
爬下来的原始数据往往"不干净"——有多余空格、换行、特殊字符，还有格式不统一、缺字段等问题。数据清洗就是把"原材料"加工成能用的"成品"，这一步通常占整个爬虫项目 60% 以上的工作量。
5.1 常见脏数据类型
空格/换行：HTML 提取的文本常带 \n、\t、不间断空格 ，首尾多余空白。
格式不一致：日期有的写"2026-09-13"，有的写"9月13日"；价格有的带"¥"符号，有的是纯数字。
缺值/空值：部分条目缺少作者、日期等字段，直接存数据库会报错。
HTML 残留：提取内容时不小心带了 <br>、<em> 等标签碎片。
重复数据：分页重叠或 URL 去重不彻底导致同一条记录被爬多次。
5.2 文本清洗常用方法
Python 标准库 + 正则表达式就能处理大部分清洗场景。下面是一个实用的文本清洗函数模板：
# clean_utils.py
import re
import html
def clean_text(raw: str) -> str:
if not raw:
return ""
# 1. 解码 HTML 实体（如 &、nbsp; 等）
text = html.unescape(raw)
# 2. 去掉残留 HTML 标签
text = re.sub(r"<[^>]+>", " ", text)
# 3. 统一空白符：换行/制表符/多个空格 → 单个空格
text = re.sub(r"\s+", " ", text)
# 4. 去掉首尾空白
text = text.strip()
return text
def parse_price(price_str: str) -> float:
# 提取数字，支持 ¥99.9 / 99元 / $19.99 等格式
match = re.search(r"[\d.]+", price_str)
if match:
return float(match.group())
return 0.0
5.3 数据校验与去重
清洗完还要校验——数据是否完整、格式对不对、有没有重复。校验不通过的数据应该丢弃或单独记录，不要混进正常数据里。
# 数据校验示例
from datetime import datetime
def validate_quote(item: dict) -> tuple[bool, str]:
# 必填字段检查
required = ["text", "author"]
for field in required:
if not item.get(field):
return False, f"缺少字段: {field}"
# 内容长度合理性
if len(item["text"]) < 5:
return False, "内容过短"
return True, "ok"
💡 小贴士：去重策略分三层——URL 去重（爬之前就跳过已爬 URL）、内容哈希去重（同内容不同 URL 也能识别）、业务主键去重（如文章 ID、商品 SKU）。三层叠加能把重复率压到 0.1% 以下。
5.4 构建采集管道
把"爬取 → 清洗 → 存储 → 通知"串成一条流水线，就是一个完整的数据采集管道。用 Python 写一个管道函数，每个环节独立，出错可定位。
# pipeline.py — 数据采集管道
import json
import time
import sqlite3
from datetime import datetime
import requests
from bs4 import BeautifulSoup
from clean_utils import clean_text, validate_quote
class QuotePipeline:
def __init__(self, db_path="quotes.db"):
self.db_path = db_path
self.seen_urls = set()
self.stats = {"crawled": 0, "valid": 0, "failed": 0}
def fetch(self, url):
"""第1步：抓取"""
if url in self.seen_urls:
return None # URL 去重
self.seen_urls.add(url)
resp = requests.get(url, timeout=10)
resp.raise_for_status()
self.stats["crawled"] += 1
return resp.text
def parse(self, html):
"""第2步：解析 + 清洗"""
soup = BeautifulSoup(html, "lxml")
items = []
for q in soup.select("div.quote"):
item = {
"text": clean_text(q.select_one("span.text").get_text()),
"author": clean_text(q.select_one("small.author").get_text()),
"crawled_at": datetime.now().isoformat(),
}
ok, msg = validate_quote(item)
if ok:
items.append(item)
self.stats["valid"] += 1
else:
self.stats["failed"] += 1
return items
def save(self, items):
"""第3步：存储（下一节展开）"""
pass
def run(self, url):
html = self.fetch(url)
if not html:
return []
items = self.parse(html)
self.save(items)
return items
管道化的好处是每一步都可单独测试、出错可定位、未来要加新环节（比如加一个 AI 摘要步骤）只需插一个函数进去，不影响其他环节。
06 数据存储与导出
爬下来的数据不能一直放在内存里，得落地存储。Termux 环境下常用三种存储方式：JSON/CSV 文件（最简单）、SQLite 数据库（结构化查询）、共享存储（方便其他 App 读取）。
6.1 JSON / CSV 文件导出
数据量小（几千条以内）时，直接存 JSON 或 CSV 文件最省事，Python 标准库就能搞定，不需要额外依赖。
# 导出 JSON
import json
def save_json(items, path="quotes.json"):
with open(path, "w", encoding="utf-8") as f:
json.dump(items, f, ensure_ascii=False, indent=2)
# 导出 CSV
import csv
def save_csv(items, path="quotes.csv"):
if not items:
return
fieldnames = items[0].keys()
with open(path, "w", newline="", encoding="utf-8-sig") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(items)
⚠️ 注意：CSV 用 utf-8-sig 编码，Excel 打开中文才不会乱码；纯程序处理用 utf-8 即可。
6.2 SQLite 数据库存储
数据量超过万条或需要查询、统计时，SQLite 是最佳选择——零配置、单文件、Python 内置支持，手机上跑完全无压力。
# storage.py — SQLite 存储
import sqlite3
class QuoteDB:
def __init__(self, db_path="quotes.db"):
self.conn = sqlite3.connect(db_path)
self.conn.row_factory = sqlite3.Row
self._init_table()
def _init_table(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS quotes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
text TEXT NOT NULL,
author TEXT NOT NULL,
crawled_at TEXT,
UNIQUE(text, author) -- 联合唯一约束去重
)
""")
self.conn.commit()
def insert_many(self, items):
# 批量插入，冲突则忽略（去重）
rows = [(i["text"], i["author"], i.get("crawled_at")) for i in items]
self.conn.executemany(
"INSERT OR IGNORE INTO quotes (text, author, crawled_at) VALUES (?, ?, ?)",
rows
)
self.conn.commit()
def count_by_author(self):
# 按作者统计数量
cur = self.conn.execute(
"SELECT author, COUNT(*) as cnt FROM quotes GROUP BY author ORDER BY cnt DESC"
)
return [dict(row) for row in cur.fetchall()]
def close(self):
self.conn.close()
把上一节管道类的 save 方法替换成 SQLite 存储，一条完整的采集链路就打通了：
# 在 QuotePipeline 中接入数据库
def save(self, items):
db = QuoteDB(self.db_path)
db.insert_many(items)
db.close()
6.3 导出到手机共享存储
爬好的数据如果想在手机上用其他 App 打开（比如用 WPS 看 CSV、用相册看图），需要把文件复制到共享存储目录。Termux 通过 termux-setup-storage 建立软链接访问。
# 首次执行，授权后会在 ~/storage 下建立软链接
termux-setup-storage
# 共享目录一览
~/storage/shared/ # 内部存储根目录
~/storage/downloads/ # Download 目录
~/storage/dcim/ # 相册目录
# 把爬好的数据复制到下载目录
cp quotes.csv ~/storage/downloads/
cp quotes.db ~/storage/downloads/
💡 小贴士：数据文件小的时候直接用 cp，大文件（几百 MB 以上）建议用 rsync 增量同步，省流量也省时间。
6.4 定时采集与增量更新
一次性爬虫价值有限，真正有价值的是定时采集、增量更新。Termux 上可以用 termux-job-scheduler 实现定时任务，配合"已爬 URL 记录"做增量。
# 增量爬取：记录已爬 URL，下次只爬新内容
import os
def load_seen_urls(path="seen_urls.txt"):
if os.path.exists(path):
with open(path) as f:
return set(line.strip() for line in f if line.strip())
return set()
def save_seen_urls(seen, path="seen_urls.txt"):
with open(path, "w") as f:
f.write("\n".join(seen))
配合 termux-job-scheduler 每天定时执行一次，就能实现无人值守的持续采集：
# 每天早上 8 点执行爬虫脚本
termux-job-scheduler -s ~/cron/run_scraper.sh --period-ms 86400000
⚠️ 常见错误
每次爬完覆盖整个文件 — 增量采集时覆盖文件会丢失历史数据
✓ 正确：用 INSERT OR IGNORE 追加写入，已存在的数据自动跳过，新数据持续累加