企业官网建设流程全解析

前言

一提「爬虫管理平台」，很多人第一反应是上框架：装 Airflow、接 Celery、配 Redis。这些方案当然成熟，但把它们跑起来所需的时间，往往比爬虫本身还多。如果你只是想让几十上百个地址「登记一下、排队跑、失败了能重来」，完全不必上这么重的东西。

一个「管理平台」真正要解决的其实只有四件事：任务登记（要抓哪些地址）、调度（按什么顺序、什么频率跑）、限速与合规（别把对方打挂、先看 robots）、状态与重试（跑过没有、失败了怎么办）。这四件事用标准库的sqlite3、argparse、time就能覆盖。

本文用一个零依赖的命令行骨架把这条链路搭起来，先能跑起来，再按瓶颈替换组件。涉及第三方调度框架时，本文只讲定位，具体用法以各框架官方文档为准。示例默认适用于 Python 3.8 及以上。

一、平台要解决的四个问题

先想清楚能力边界，再写代码：

1. 
   
2. 任务登记：把一个待抓地址写进持久化存储，重复登记要幂等。
3. 
   
4. 调度：每次运行取出若干「未完成」的任务，跑完更新状态。
5. 
   
6. 限速与合规：对每个域名先查 robots.txt；请求之间留间隔，避免把对端打满。
7. 
   
8. 状态与重试：记录每个任务的状态和尝试次数，失败的可再次进入待处理。
9. 
   

二、用 sqlite3 存任务

一张表就够。url上建唯一约束，登记时用INSERT OR IGNORE保证幂等；status记录状态；attempts记录尝试次数。

# 适用于 Python 3.8+
import sqlite3

DB = "crawler_tasks.db"


def init_db(conn):
conn.execute(
"CREATE TABLE IF NOT EXISTS tasks ("
" id INTEGER PRIMARY KEY AUTOINCREMENT,"
" url TEXT UNIQUE NOT NULL,"
" status TEXT NOT NULL DEFAULT 'pending',"
" attempts INTEGER NOT NULL DEFAULT 0,"
" last_error TEXT"
")"
)
conn.commit()


def add_task(conn, url):
conn.execute("INSERT OR IGNORE INTO tasks (url) VALUES (?)", (url,))
conn.commit()


def pending_tasks(conn, limit=50):
cur = conn.execute(
"SELECT id, url FROM tasks WHERE status = 'pending' ORDER BY id LIMIT ?",
(limit,),
)
return cur.fetchall()

状态字段建议只保留几个值：pending（待处理）、done（成功）、error（失败）、blocked（robots 不允许）。状态越少，后续查询越简单。

三、调度循环与限速

调度循环的骨架是「取一批 → 逐个处理 → 更新状态 → 间歇」。这里的间歇就是限速，不要省。

# 适用于 Python 3.8+
import time
import urllib.error
import urllib.request
import urllib.robotparser
from urllib.parse import urlsplit

USER_AGENT = "MiniCrawler/0.1 (contact: me@example.com)"
_ROBOTS_CACHE = {}


def robots_allows(url, user_agent):
parts = urlsplit(url)
base = parts.scheme + "://" + parts.netloc
if base not in _ROBOTS_CACHE:
rp = urllib.robotparser.RobotFileParser()
rp.set_url(base + "/robots.txt")
try:
rp.read()
except urllib.error.URLError:
_ROBOTS_CACHE[base] = None # 拿不到就按禁止处理
else:
_ROBOTS_CACHE[base] = rp
cached = _ROBOTS_CACHE[base]
if cached is None:
return False
return cached.can_fetch(user_agent, url)

把 robots 文件按站点缓存起来，避免每个地址都重新拉一次。

四、把网络操作与数据库写入分开

一个必须提前想清楚的点：sqlite3的连接默认不允许跨线程使用。如果打算用线程池并发抓取，不要让子线程去写同一个连接。正确分工是：子线程只做「网络请求 + 解析」，把结果返回给主线程，由主线程统一更新数据库。这样既拿到并发，又避开连接争用。

# 适用于 Python 3.8+
def fetch_one(url, timeout=10):
"""在子线程执行：只做网络请求与解析，不碰数据库。"""
if not robots_allows(url, USER_AGENT):
return "blocked", "robots.txt 不允许"
req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
raw = resp.read()
charset = resp.headers.get_content_charset() or "utf-8"
text = raw.decode(charset, errors="replace")
except urllib.error.HTTPError as e:
return "error", "HTTP " + str(e.code)
except urllib.error.URLError as e:
return "error", str(e.reason)
return "done", "长度 " + str(len(text))

实战：完整的命令行骨架

把上面的片段拼起来，配上argparse做成两个子命令：add登记地址，run执行一轮。

# 适用于 Python 3.8+
import argparse
import sqlite3
import time
import urllib.error
import urllib.request

DB = "crawler_tasks.db"
USER_AGENT = "MiniCrawler/0.1 (contact: me@example.com)"


def init_db(conn):
conn.execute(
"CREATE TABLE IF NOT EXISTS tasks ("
" id INTEGER PRIMARY KEY AUTOINCREMENT,"
" url TEXT UNIQUE NOT NULL,"
" status TEXT NOT NULL DEFAULT 'pending',"
" attempts INTEGER NOT NULL DEFAULT 0,"
" last_error TEXT"
")"
)
conn.commit()


def add_task(conn, url):
conn.execute("INSERT OR IGNORE INTO tasks (url) VALUES (?)", (url,))
conn.commit()


def pending_tasks(conn, limit=50):
cur = conn.execute(
"SELECT id, url FROM tasks WHERE status = 'pending' ORDER BY id LIMIT ?",
(limit,),
)
return cur.fetchall()


def fetch_one(url, timeout=10):
# 简化处理：此处略去 robots 检查，实际项目应先调用 robots_allows 并缓存结果
req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
raw = resp.read()
charset = resp.headers.get_content_charset() or "utf-8"
text = raw.decode(charset, errors="replace")
except urllib.error.HTTPError as e:
return "error", "HTTP " + str(e.code)
except urllib.error.URLError as e:
return "error", str(e.reason)
return "done", "长度 " + str(len(text))


def run_once(conn, delay=1.0, limit=50):
rows = pending_tasks(conn, limit)
if not rows:
print("没有待处理任务")
return 0
for task_id, url in rows:
status, info = fetch_one(url)
conn.execute(
"UPDATE tasks SET status = ?, attempts = attempts + 1, "
"last_error = ? WHERE id = ?",
(status, info, task_id),
)
conn.commit()
print("任务", task_id, status, info, url)
time.sleep(delay) # 限速：每个请求之间等待
return len(rows)


def main():
ap = argparse.ArgumentParser(description="最小爬虫管理平台")
sub = ap.add_subparsers(dest="cmd", required=True)
p_add = sub.add_parser("add", help="登记一个地址")
p_add.add_argument("url")
p_run = sub.add_parser("run", help="执行一轮")
p_run.add_argument("--delay", type=float, default=1.0, help="请求间隔秒数")
args = ap.parse_args()

conn = sqlite3.connect(DB)
init_db(conn)
if args.cmd == "add":
add_task(conn, args.url)
print("已登记:", args.url)
elif args.cmd == "run":
n = run_once(conn, delay=args.delay)
print("本轮处理:", n, "个任务")
conn.close()


if __name__ == "__main__":
main()

用法就是在命令行里执行python crawler.py add 某个地址登记，再用python crawler.py run --delay 2跑一轮。需要并发时，用concurrent.futures.ThreadPoolExecutor把fetch_one派到子线程，主线程收集结果后再统一写库——切记别让子线程直接写连接。

常见坑点

1. 
   
2. 子线程里用同一个 sqlite3 连接
3. 
   

❌ 在线程池的 worker 里调用conn.execute(...)—— 默认check_same_thread=True，会抛ProgrammingError。 ✅ 子线程只返回结果，主线程统一写库；或每个线程各建一个连接。

1. 
   
2. 登记任务不做幂等
3. 
   

❌INSERT INTO tasks (url) VALUES (?)—— 重复登记同一地址会触发唯一约束错误。 ✅ 用INSERT OR IGNORE，并给url建唯一约束。

1. 
   
2. 限速写在错误的位置
3. 
   

❌ 只在开始前sleep一次，循环里连续发请求。 ✅ 每次请求之间都要time.sleep(delay)；delay设成可配置参数。

1. 
   
2. robots.txt 每个地址都重新拉一次
3. 
   

❌ 每次请求前都rp.read()—— 既慢又给对方增加无谓请求。 ✅ 按站点（scheme + host）缓存解析结果，模块级字典即可。

1. 
   
2. 把网络异常和解析异常混在一起吞掉
3. 
   

❌except Exception: pass—— 任务永远停在pending，看不出原因。 ✅ 分别处理HTTPError（读e.code）与URLError（读e.reason），把信息写进last_error。

1. 
   
2. 状态机没有终态，任务无限重跑
3. 
   

❌ 失败后不区分状态，下一轮又把同一个坏地址取出来。 ✅ 记录attempts，超过阈值置为error并停止重试。

1. 
   
2. 以为argparse子命令可以省略
3. 
   

❌ap.add_subparsers()后直接args.cmd—— 用户不传子命令时属性缺失。 ✅ 传required=True（Python 3.7+ 支持），并给每个子命令加help。

1. 
   
2. 搬 Python 2 的老写法
3. 
   

❌print "已登记"、for k, v in d.iteritems():—— Python 2 已于 2020 年 1 月 1 日停止维护，这些写法在 3 里都不成立。 ✅ 用print(...)函数调用，字典遍历用d.items()。

总结

搭平台的正确定序是「先让单机骨架跑通并稳定，再按瓶颈换组件」。一个能持久化、能限速、能记录失败原因的 sqlite 脚本，比一个配置半天还跑不起来的大框架更有用。