4.3 数据处理与自动化(极简版)
这一节只做一件事：
把你从 API/爬虫拿到的数据，简单清洗一下，存成文件，再让脚本自动跑一次。
不追求复杂的飞书同步、分布式任务，只要有一个「小流水线」跑起来就够了。
第一步： 简单清洗 + 去重
假设你现在有一批文章数据，大致长这样：
articles = [
{"title": " Hello World ", "content": " 第一篇 文章 \n\n", "link": "https://example.com/1"},
{"title": "Hello World", "content": "内容重复", "link": "https://example.com/1"},
]你可以写一个很小的「清洗函数」:
def clean_text(text: str) -> str:
# 去掉多余空格和换行
text = " ".join(text.split())
return text.strip()
def clean_articles(raw_articles: list[dict]) -> list[dict]:
cleaned: list[dict] = []
seen_titles: set[str] = set()
for article in raw_articles:
title = clean_text(article.get("title", ""))
if not title or title in seen_titles:
continue # 标题为空或重复就跳过
seen_titles.add(title)
cleaned.append(
{
"title": title,
"content": clean_text(article.get("content", "")),
"link": article.get("link", ""),
}
)
return cleaned做到这里，你已经完成了「数据清洗」里最常见的两步： 去空 & 去重。
第二步： 存成 CSV 文件
把清洗后的数据存成一个简单的表格文件，方便用 Excel/Notion 打开：
import csv
from pathlib import Path
def save_to_csv(articles: list[dict], filename: str = "articles.csv") -> None:
Path(filename).parent.mkdir(parents=True, exist_ok=True)
with open(filename, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=["title", "content", "link"])
writer.writeheader()
writer.writerows(articles)然后你可以在主脚本里这样串起来：
if __name__ == "__main__":
# 这里可以替换成你的真实数据,比如从 API 或爬虫获取
raw_articles = articles
cleaned = clean_articles(raw_articles)
save_to_csv(cleaned, "output/articles.csv")
print(f"已保存 {len(cleaned)} 条数据到 output/articles.csv")运行一遍，用表格工具打开 output/articles.csv，确认格式没问题。
第三步： 让脚本每天自动跑一次(可选)
如果你希望它「每天自己跑一下」，可以先用最简单的方式——一个常驻脚本 + schedule:
import schedule
import time
def job():
print("开始执行抓取与处理...")
# TODO: 在这里调用你的爬虫/API + clean + save_to_csv
print("本次任务完成")
schedule.every().day.at("08:00").do(job)
while True:
schedule.run_pending()
time.sleep(60) # 每分钟检查一次这一部分属于「加分项」:
先确保手动执行一次流程是成功的，再考虑让它自动化；一开始不要追求完美的调度系统。
做到这里，你已经把「4.1 API → 4.2 爬虫 → 4.3 数据处理」串成了一条最简单的自动化小流水线。
后面如果想接飞书/Notion/数据库，完全可以把现有脚本丢给 AI，让它在这个基础上继续升级。
Last updated on