企业官网建设流程全解析

1. 项目缘起：DeviantArt 每日精选的增量采集到底难在哪

先说结论：DeviantArt（以下简称 DA）是练手爬虫增量采集思路的好靶场，但想把它做成一个能长期稳定运行的采集流水线，坑比想象中多。

我最开始有这个需求，其实是为了给一个艺术灵感库做素材积累。DA 的 Daily Deviations（每日精选）栏目天天都会更新一批高质量作品，人眼逛着看当然爽，但想把这些作品信息结构化存下来、按日回溯、按作者聚合，没有程序化的采集管道根本做不到。于是就有了这个项目：用 Python 构建一条从 DA 每日精选页面/订阅源出发，能自动识别"有哪些是新内容、哪些之前已经抓过"并只增量入库的流水线。

先给没接触过 DA 的朋友说一下它的内容更新规律。DA 的每日精选并不是一整天才更新一次，而是编辑团队在一天内陆续挑选作品上线，所以同一个"每日精选"页面在 24 小时内其实会出现多轮内容变化。这意味着，如果你只是每天定时抓一次，你很可能漏掉当天的部分更新；如果你动不动就全量重抓，又会对服务器造成无谓压力，还会产生大量重复数据。增量采集的思路就在这里派上用场——只抓上次记录之后出现的新内容，并且把"上次记录到哪了"这个状态持久化下来。

这个项目适合谁来参考？两类人：一是刚学完 Requests、BeautifulSoup、SQLAlchemy 基础，想完整走一遍"采集-解析-存储-调度"全链路的爬虫学习者；二是确实需要定期从 DA 获取灵感素材，又不希望每次手动导出、手动整理的内容运营者。下文所有代码和思路都基于我实际跑通的版本，环境是 Python 3.10 + SQLite/MySQL + APScheduler，框架只依赖 Requests 和 BeautifulSoup。

很多人一提到 DA 就以为必须处理复杂的网页解析，实际上 DA 提供了一个非常良心的官方 RSS 订阅源（Daily Deviations 对应的 RSS 地址是https://www.deviantart.com/daily-deviations/rss），这算是官方给的"半合法顺手数据通道"。虽然它是 RSS 而不是标准 JSON API，但信息密度已经足够：标题、链接、发布日期、作者名、图片地址都能拿到。至于网页端的 HTML 解析，我只把它作为 RSS 字段缺失时的补充手段，而不是主通道。这个取舍在后面会详细讲原因。

2. 增量采集的核心设计：先想清楚三个问题再动手

增量采集不是简单加个 WHERE 条件判断就完事，它本质上是在问三个问题：以什么字段作为"增量游标"？游标存在哪里？游标丢失后怎么恢复？这三个问题想不清楚，后面写代码一定会来回返工。

2.1 为什么用日期作为增量游标，而不是用作品 ID

DA 每日精选的作品链接长这样：https://www.deviantart.com/someartist/art/Work-Title-123456789，末尾那串数字是作品 ID。用 ID 判断增量看起来很自然——大于上次最大 ID 的就是新作品——但实际会翻车。因为 DA 的精选栏目同一作品可能被编辑重新推荐，RSS 里同一个 ID 会再次出现；另外 RSS 条目的 ID 字段格式在不同地区域名下并不统一，有的带前缀（https://www.deviantart.com/...），有的直接给纯数字，解析时容易出岔子。

我最终采用的游标是RSS 条目的发布时间（pubDate）加上作品链接的联合去重。理由很简单：每日精选的内容是按编辑推荐时间排序的，我们认为"今天更新的就是新内容"，所以只需要记录上次成功抓取到的最新发布时间，下一次抓取时把大于这个时间的条目全部视为增量。由于 DA 的 RSS 用的是标准 RFC 822 时间格式（库跑得快，dateutil解析一下就是datetime），配合作品链接做唯一键，基本能覆盖"同作品重复推荐""同作品在不同精选分类出现"这两种常见场景。

2.2 游标持久化：文件、数据库还是环境变量

增量游标的存储我试过三种方式，真实感受如下：

• 存本地 JSON 文件：最简单，读出来改回去一分钟搞定。缺点是如果脚本被多个进程并发执行，写文件会有竞争风险，而且部署到服务器上文件路径容易搞乱。适合本地小规模实验。
• 存在数据库表里：我最终选了这个方案。专门建一张crawl_state表，字段就两三个：source（标识哪个采集源）、last_cursor（上次游标值）、updated_at。好处是和业务数据在同一个事务环境里，写入失败可以回滚，天然支持并发锁。
• 用环境变量或者 Redis：环境变量只适合无状态任务（比如云函数一次一跑），Redis 适合分布式多实例场景。我的项目规模用不上 Redis，就没引入额外依赖。

这里还有一个细节很多人会忽略：游标更新的时机。必须等一批增量数据全部解析、入库成功之后，再更新游标；如果中途抛异常，游标保持不变。否则就会出现"数据写失败了，但游标已经前移"的坑，重启后直接跳过那批内容。

2.3 增量去重表：宁可多加一个唯一索引，也不信巧合

即便有了日期游标，我仍然在作品信息表里加了(link)唯一索引。这属于双保险：万一某次 RSS 返回的时间异常、排序错乱，重复的链接也不会真的插进数据库，而是触发IntegrityError，我在异常处理里捕获后跳过即可。真出了重复，说明去重已经挡不住了，这时该去查源头而不是改 SQL。

3. 请求层解析：从 RSS 入口拿到每日精选的完整路径

3.1 为什么主通道选 RSS 而不是直接解析 HTML

DA 网页版的每日精选页面是动态渲染的，图片列表藏在deviantart.com的 JS 脚本里，直接用 Requests 抓到的 HTML 里有一大堆__INITIAL_STATE__、_react_root之类的占位符，真正的作品数据要等浏览器执行 JS 后才出现。没有渲染环境，BeautifulSoup 基本拆不出东西。

相比之下，RSS 订阅源是服务端渲染的 XML，Requests 直接 GET 就能拿到干净数据，每条 item 的结构非常规整，包含<title>、<link>、<pubDate>、<dc:creator>、<content:encoded>等字段。解析成本低，稳定性高，也天然适合增量采集。

有朋友可能会问：为什么不用官方 API？因为 DA 的个人 Developer API 申请流程麻烦，而且 Rate Limit 很敏感，一不小心就被限流好几天。RSS 是公共订阅源，头部带一个合理的 User-Agent 就能正常访问，对个人项目来说已经足够。

3.2 解析函数的关键代码

import requests import feedparser from dateutil import parser as dt_parser REQ_HEADERS = { "User-Agent": "Mozilla/5.0 (compatible; DailyCollector/1.0; +https://example.local/bot)", "Accept": "application/rss+xml, application/xml;q=0.9, */*;q=0.8", } def fetch_rss_entries(rss_url: str) -> list[dict]: resp = requests.get(rss_url, headers=REQ_HEADERS, timeout=20) resp.raise_for_status() feed = feedparser.parse(resp.content) entries = [] for e in feed.entries: entries.append({ "title": e.get("title", "").strip(), "link": e.get("link", "").strip(), "published_at": dt_parser.parse(e.get("published", e.get("updated", ""))), "author": e.get("dc_creator", e.get("author", "")).strip(), "summary": (e.get("summary", "") or "")[:500], "image_url": extract_image_url(e), }) return entries

说明一下，feedparser在 Python 社区是老牌库了，它能把 XML 解析成字典结构，省去自己写 XPath 的麻烦。而extract_image_url这个函数是从<content:encoded>里用正则抽第一张<img src>，因为 DA 的 RSS 里不直接给enclosure字段，图片地址藏在富文本正文里。

import re def extract_image_url(entry) -> str: content = entry.get("content", []) if content: html = content[0].get("value", "") m = re.search(r'<img[^>]+src="([^"]+)"', html) if m: return m.group(1) media = entry.get("media_content") or entry.get("media_thumbnail") or [] if media: for m in media: url = m.get("url") if url: return url return ""

3.3 增量判断与时间解析的边界条件

published_at解析出来的时间，默认会带时区信息。这里有个大坑：DA 的 RSS 返回时间一般是 GMT，跟你本地时区常常差 8 个小时。如果不做时区归一化，你的增量游标就会"越跑越偏"。

我的做法是：所有时间字段一律用datetime.utcnow()基准值比较，解析出带时区的published_at后，先.astimezone(timezone.utc)再存库。存进数据库的时候，如果 SQLAlchemy 配的是 SQLite，DateTime类型不会自动带时区，我干脆统一存 UTC 的datetime，读取时再转成本地时间展示。这样可以避免"早上 8 点跑批时发现昨天下午的增量一直没抓到"这种诡异问题。

另外feedparser在极少数情况下解析pubDate会直接报错，比如缺时区、格式非标准。所以我在dt_parser.parse外面套了一个try/except，解析失败的条目单独记日志，不阻断整批流程。宁可丢一条解析异常的数据，也不能让整个流水线崩掉，这是采集系统设计的一个基本原则。

4. 流水线架构：调度器、解析器、存储模块这样分工最干净

我一直很反感把爬虫代码写成"一个几百行的 main 函数从头跑到尾"。增量化之后，这种写法几乎必然出问题——因为你要处理网络异常、数据库异常、游标回滚、重试策略，混在一起根本没法调。

这个项目最终拆成了四个模块，各干各的事：

• Scanner（扫描器）：只负责请求 RSS、解析 entries、按游标过滤出增量，不关心数据存哪里。
• Parser（解析器）：对增量条目做字段清洗、补全图片 URL、截断摘要，这一步产出干净的 dict 列表。
• Writer（存储模块）：用 SQLAlchemy 把 dict 写入数据库，处理重复键和事务回滚。
• Scheduler（调度器）：控制整个流水线多久跑一次、并发度多大、失败重试几次。

这样做的好处是，你可以随便替换任何一层。比如今天用了 SQLite，明天想迁到 MySQL，只需要改 Writer 的连接配置，Scanner 和 Parser 完全不用动；今天只用 RSS，明天想加一个网页解析通道，也是加个新的 Scanner 实现而已。

核心调度代码用queue.Queue加两个后台线程实现——一个线程负责扫描和产出增量任务，另一个线程负责消费任务并写入数据库。线程间的数据量并不大（每日精选大概几十条），所以用最简单的队列就够，不需要引入 Celery。

import queue import threading import time def run_pipeline(crawl_state_service, writer_service): task_queue = queue.Queue() stop_event = threading.Event() def scanner_worker(): while not stop_event.is_set(): try: last_cursor = crawl_state_service.get_cursor("daily_deviations") entries = fetch_rss_entries(RSS_URL) new_entries = [ e for e in entries if e["published_at"] > last_cursor ] new_entries.sort(key=lambda x: x["published_at"]) for entry in new_entries: task_queue.put(entry) # 等队列全部消费完成后，再把游标前移 task_queue.join() if new_entries: cursor_value = new_entries[-1]["published_at"] crawl_state_service.update_cursor("daily_deviations", cursor_value) except Exception as exc: logger.exception("scanner failed: %s", exc) stop_event.wait(REFRESH_INTERVAL) def writer_worker(): while not stop_event.is_set(): try: entry = task_queue.get(timeout=3) writer_service.save_entry(entry) task_queue.task_done() except queue.Empty: continue except Exception as exc: logger.exception("writer failed: %s", exc) task_queue.task_done() t1 = threading.Thread(target=scanner_worker, daemon=True) t2 = threading.Thread(target=writer_worker, daemon=True) t1.start() t2.start() return t1, t2

有朋友看到这里可能会问：为什么更新游标不放在每处理完一条之后？因为我希望一批增量要么全部成功，要么全部保持游标不变，这样重跑成本低。如果每一条都更新游标，某条数据入库失败、游标却前进了，下次就不会再补抓这条，就形成永久缺失。

5. SQLAlchemy 存储层：让爬下来的数据能查能用

5.1 建表模型：作品表加作者表，关联关系别搞复杂

增量采集的目的是积累数据，那存储结构就很重要。我不建议把所有东西塞进一张大宽表。DA 每日精选的核心实体是作品（Post），作品天然关联作者（Author），而且作者可能多次入选。所以两张表加一个关联关系字段就够了。

# models.py from sqlalchemy import Column, Integer, String, DateTime, Text, ForeignKey, UniqueConstraint from sqlalchemy.orm import declarative_base, relationship Base = declarative_base() class Author(Base): __tablename__ = "authors" id = Column(Integer, primary_key=True, autoincrement=True) username = Column(String(128), unique=True, nullable=False) profile_url = Column(String(512)) first_seen_at = Column(DateTime, default=datetime.utcnow) class Post(Base): __tablename__ = "posts" id = Column(Integer, primary_key=True, autoincrement=True) title = Column(String(512), nullable=False) link = Column(String(1024), nullable=False) image_url = Column(String(2048)) summary = Column(Text) published_at = Column(DateTime, nullable=False, index=True) author_id = Column(Integer, ForeignKey("authors.id")) author = relationship("Author", backref="posts") created_at = Column(DateTime, default=datetime.utcnow) __table_args__ = ( UniqueConstraint("link", name="uq_posts_link"), )

添加索引的目的是让"按时间范围查询""按作者聚合"这类操作不至于全表扫描。SQLite 存几百条数据还好，但如果跑上一年积累了上万条，没有索引会明显变慢。

5.2 写入逻辑：先查后插还是 try-except

具体写入时用"先按作者名查出或创建作者记录，再插入作品记录"，这样能保证同一作者多次入选不会在 authors 表产生重复行。

# writer.py from sqlalchemy.orm import sessionmaker from models import Author, Post def save_entry(session, entry: dict) -> bool: author = session.query(Author).filter_by(username=entry["author"]).one_or_none() if not author: author = Author(username=entry["author"]) session.add(author) session.flush() # 先拿到自增 id post = Post( title=entry["title"], link=entry["link"], image_url=entry["image_url"], summary=entry["summary"], published_at=entry["published_at"], author_id=author.id, ) session.add(post) try: session.commit() return True except IntegrityError: session.rollback() logger.info("duplicated post skipped: %s", entry["link"]) return False

这里flush的作用是把 Author 先落地拿到数据库自增 ID，这样 Post 的外键就有值了。如果你不 flush，ORM 也会自动处理，但在save_entry这种独立事务里显式写更安全，不会出现"作者还没入库就插入作品"的顺序问题。

5.3 保留原始字段：别为了"干净"丢掉排查线索

我在 Post 表里特意加了link和image_url这两个原始字段，哪怕它们对后续分析可能没用。因为在跑批过程中如果某条数据解析异常，你至少可以从 link 回溯到 DA 原页面，手动确认是解析规则问题还是数据本身问题。这个习惯帮我排查过好几次 RSS 字段变更造成的诡异 Bug。

6. 定时调度与断点续跑：命令行、后台线程和 APScheduler 三套方案

增量采集流水线的最后一个拼图是定时运行。我实际开发过程中经历了三个阶段，分别对应三种调度方式。

6.1 阶段一：手动命令行运行

刚开始，就是python run.py，每次手动跑一遍。这在调试阶段完全够用，也方便观察输出日志。但问题是，如果你忘了跑，今天的精选就漏了。增量游标的意义在这种场景下不明显，因为手动跑大概率会隔一两天，一跑就是两条增量，容易怀疑"是不是没增量了"。

6.2 阶段二：APScheduler 定时任务

APScheduler 是 Python 的定时任务库，支持 cron 表达式。我在调度器模块里增加了一个start_scheduler()函数：

from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger def start_scheduler(): scheduler = BackgroundScheduler() scheduler.add_job( run_pipeline, CronTrigger(hour="8,20", minute="15"), id="daily_deviations_job", replace_existing=True, max_instances=1, coalesce=True, ) scheduler.start() return scheduler

这里有两个参数值得解释。max_instances=1保证上一次任务没跑完时，下一次触发不会开新实例，避免并发写库。coalesce=True表示如果错过了多次触发时间，只补跑最近一次，防止服务器关机两天后一开机就连跑几十个任务。这两个参数是做定时爬虫必调的，不调早晚出事故。

6.3 阶段三：要不要做分布式

实话说，我这个项目完全不需要分布式。DA 每日精选一天几十条增量，单线程跑几秒就完了。但很多人在网上看分布式爬虫案例看多了，总觉得要上 Scrapy + Redis + 分布式节点才叫爬虫系统。我个人的看法是：选型跟着数据量走，你的任务规模还没到一台服务器扛不住的程度，就老老实实用单进程 + 队列。分布式会引入的不只是部署复杂度，还有任务分发一致性、去重一致性、游标共享一致性问题，这些比采集本身难处理得多。硬上分布式，等于给自己造坑。

6.4 断点续跑的设计细节

既然有增量游标，那断点续跑就是天然支持的。具体来说：

• 手动跑python run.py --full可以做一次全量初始化，把游标设置成一个非常早的时间点（比如 2000 年），这样第一次跑会把 RSS 里所有历史条目都抓进来。
• 正常运行python run.py则按照数据库里的游标增量抓取。
• 如果某天 RSS 地址改了或者解析逻辑报错，你不用慌，因为游标还停留在上次成功位置，修好代码再跑一次就能续上。

启动时脚本会先检查数据库里有没有游标记录，没有的话就默认 2000-01-01，让冷启动变成一次"全量抓取"。这个设计让系统既适合第一天上线的初始化，也适合日后的日常运转。

7. 反爬与合规：不让自己成为"不速之客"的底线操作

7.1 身份标识与请求频率

DA 算是对爬虫相对友好的网站，但这不是你乱来的理由。我的流水线严格遵守三点：

• User-Agent 明确标识自己：写上项目名和联系方式（或项目主页）。这样对方服务器能知道这是正常的个人采集项目，而不是匿名攻击脚本。
• 请求间隔最少 10 秒：即使 RSS 一次能拉全，我也只在每次调度时请求一次，绝不循环重试。本来爬虫的意义是"用合理频率获取公开信息"，不是"以最大速率击穿对方服务"。
• 不碰图片原图下载：图片是 CDN 资源，每条作品的 image_url 背后用的是 DA 的图片服务。我项目里只采集元数据，不自动下载图片，既节省存储也避免对图片 CDN 造成压力。如果你确实需要图片做本地素材库，建议手动挑着下载，不要全线批量拉。

7.2 robots.txt 与内容使用边界

DA 的 robots.txt 有规定部分路径禁止爬取，我遵守了。这个项目的采集路径只涉及 RSS 订阅源，本身就是 DA 主动向公众分发内容的渠道，可以理解为"网站希望你消费信息的方式"。这也解释了为什么我选择 RSS 作为主通道而不是解析 HTML——RSS 是官方推荐的阅读渠道，合规性要强得多。

7.3 Retry 策略与 429 处理

即便再小心，偶尔也会遇到 429（请求过多）。我对 HTTP 状态码的处理策略很简单：

• 500/502/503：睡眠 30 秒后重试一次，还是失败就放弃本轮，等下次调度再跑。
• 404：说明 RSS 地址或页面已变，发告警日志，人工介入。
• 429：说明访问太频繁，把REFRESH_INTERVAL临时拉长到 1 小时，并持续观察，避免继续硬怼。

def robust_get(url, headers, max_retries=2): for attempt in range(max_retries + 1): try: resp = requests.get(url, headers=headers, timeout=20) if resp.status_code == 429: time.sleep(30) continue resp.raise_for_status() return resp except requests.RequestException as exc: if attempt == max_retries: raise time.sleep(5 * (attempt + 1)) return None

有一个特别重要的守则：不要做绕过限制的事。如果对方要求登录才能访问，那就说明这部分内容不适合爬；如果对方返回了验证码页面，就停止。采集的边界应该由网站的通告和技术手段来划，而不是靠你自己的欲望。

8. 踩坑清单：这些细节不处理，流水线跑一周就会挂

这个项目从跑通到真正稳定运行，中间经历了不少"看起来是小问题、实际能引发连锁故障"的坑。整理几个印象最深的，希望帮你避开。

8.1 时区问题：游标越跑越偏，最后漏抓一整天

这是我踩的第一个大坑。最初的版本里published_at解析后直接存库，没有转 UTC，游标比较也用本地时间。DA 的 RSS 时间是 GMT，本地时间是 GMT+8，结果就是每次增量判断都会把游标往前多推 8 小时，跑到第三天，昨天的数据全部漏掉。大概从那天起，我的代码里所有时间字段的处理都强制走 UTC，显示层才转本地。

8.2 feedparser 对日期解析的迷之宽容与严格

feedparser在解析大多数标准时间时很好用，但碰到个别不标准的格式（比如缺少时区偏移），它会返回None或者抛异常。我的fetch_rss_entries里对每个 entry 都做了异常捕获，把解析失败的时间统一设置为datetime.utcnow()再做兜底，然后日志里打警告。这虽然不完美，但至少不会因为一条脏数据导致整批任务失败。另外，极端情况下feedparser解析失败后会把published字段变成None，所以从头到尾我都用e.get("published", e.get("updated", ""))这个后备逻辑，保证至少有一个字段能拿到。

8.3 IntegrityError 的误判：重复链接不代表程序出错

一开始看到IntegrityError我以为是自己代码 bug，后来才知道这是去重的正常信号。只要唯一索引存在，重复插入就一定会报这个错。正确的处理方式就是我在 Writer 里做的：捕获、回滚、记日志、返回 False。千万别往上抛异常，否则会因为一条重复数据中断整轮增量任务，进而影响游标推进。

8.4 SQLite 并发写入：后台线程和主线程同时碰库

SQLite 对多线程写入并不友好，如果你用默认连接方式，sqlalchemy.exc.OperationalError: database is locked是家常便饭。我给 SQLAlchemy engine 配置了connect_args={"check_same_thread": False, "timeout": 30}，并且把连接池大小设为 1，保证同一时间只有一个线程在写。如果你的业务量比这个项目大，建议直接上 MySQL 或 PostgreSQL，不要再跟 SQLite 的并发死磕。

8.5 RSS 内容字段变化：解析器要能容忍"少字段"

某天 RSS 突然少了一个<content:encoded>标签，导致我的extract_image_url抽不到图片地址。这个不是网站改版，只是该条内容格式特殊。我加了"找不到就跳过图片字段"的兜底逻辑，把 image_url 置空字符串，也允许入库。宁可图片为空，也不能因为没有图片就丢一条作品元数据，因为后续还能通过 link 手动补。

8.6 调度器重复注册：APScheduler 最常见的隐形 bug

add_job如果在一个反复调用的函数里执行，会把同一个 job 注册很多次，导致同一时间触发多个流水线。解决方法是设置id参数加上replace_existing=True，并且在启动脚本入口处用一个全局变量确保调度器只启动一次。我在这个坑上耗了大半天，起因是 Flask 的开发服务器 reload 模式下会执行两次启动代码，两个调度器同时跑，数据库里立刻冒出一堆重复链接。

8.7 日志与监控：羞答答的 print 在跑批时代不够用

早期版本我用print打印进度，跑起来之后发现一个问题：一旦有任务在夜深人静的时候悄悄失败，你根本不知道，直到第二天发现游标没有推进、数据没有更新。后来我引入logging+ 写入日志文件，并且跑批结束后在控制台打印一行摘要（共抓到多少条、多少条重复、耗时多少秒），这才有了最基本的"事后复盘"能力。如果部署在服务器上，建议再把异常推送到钉钉/企业微信机器人之类的通知渠道，但这就属于扩展范围了。

9. 增量采集之外的两种延伸思路

如果这个项目你已经跑通了，想让它更实用，可以考虑两个小方向。

9.1 按作者聚合的周报

Post 表里存了 author_id，你可以写一条联表查询：统计最近 7 天哪些作者入选次数最多、哪些作品的 summary 包含某个关键词等。这本质上是把采集数据变成内容运营的素材。比如你是一个设计团队的灵感管理员，每周一发一份"DA 本周精选 Top10 作者榜"给团队，用 SQLAlchemy 查一下就行，全程不用打开浏览器。

9.2 多分类/多栏目的统一采集

DA 除了每日精选，还有专题栏目、每周精选、艺术类别的 Top 排行等。它们的 RSS 地址格式类似，只是路径不同。你完全可以用相同的 Scanner 和 Writer，只需要把crawl_state表里的source字段区分开（比如daily_deviation、weekly_top_fanart），并在每个 source 下面记录各自的游标。这就是把"单一流水线"升级成"多流水线系统"的最干净路径，不需要改动任何表结构。

不过这里提醒一点：每个采集源都要独立设置合理的抓取频率和间隔，不要为了省事把所有 source 都塞进同一个 Scan 循环。不同栏目的更新频率完全不同，统一调度会造成严重的资源浪费，也容易触发频率限制。

我的实盘经验是，每日精选这种栏目一天跑 2 次（早上 8 点和晚上 8 点各一次）已经能保证非常低的延迟，毕竟 RSS 更新本身也有延迟，多跑多次纯属浪费。而像每周精选这种栏目，一周跑一次就够了，放在周一早上执行，正好覆盖周末的更新。调度频率的设计，原则上是"覆盖更新的最小必要次数"，而不是"越频繁越好"。

10. 最后再分享一个让我少走弯路的小技巧

在你把整个流水线写成代码之前，先花一天时间手动观察目标数据源的原始响应格式。我在项目启动的第一天，空手写了一堆解析逻辑，结果发现真实数据结构和网上教程里显示的完全不同——版本差异、字段名变化、时区格式不统一，全是文档里没有的细节。

后来我养成一个习惯：新建爬虫项目的第一件事，把目标 URL 的原始响应保存成.xml或.html文件放在项目的samples/目录下，之后再写解析代码时，直接对着样本文件调试，不需要反复请求线上服务。这样既快又省事，还避免了在开发期就被网站反爬封掉的风险。数据格式一变，对比新旧样本文件就能快速定位问题。

这条流水线我实际跑了三个多月，除了 RSS 字段部分变化导致极少量数据需要手动补，日常增量采集基本达到了零维护状态。我觉得对于 DA 这种更新频率不高、内容结构相对稳定的目标，用 Requests 加 SQLAlchemy 这种轻量组合，配合明确的游标和持久化状态，确实是最省心也最容易理解方案。如果你正打算做类似目标的增量采集，照着这个思路搭一条，应该能少踩一半的坑。