企业官网建设流程全解析

第 19 课 | 定时任务与自动化：让系统 7x24 小时运行

前 18 课我们手动运行每一个脚本。这节课，我们让整个系统自动运转——每天定时爬取、清洗、分析、生成报告，全部无人值守。

一、业务价值：从"工具"到"系统"

1.1 手动 vs 自动

核心价值：不会忘记、不会偷懒、不会出错。你休假时系统照常运转。

1.2 自动化流水线全景

二、Python schedule 库：轻量级任务调度

2.1 安装

uv pip install schedule

2.2 基础用法

importscheduleimporttimedefjob():print("执行任务...")# 每天 9:00 执行schedule.every().day.at("09:00").do(job)# 每小时执行schedule.every().hour.do(job)# 每 10 分钟执行schedule.every(10).minutes.do(job)# 每周一执行schedule.every().monday.do(job)whileTrue:schedule.run_pending()time.sleep(60)# 每分钟检查一次

schedule 的特点：

• 纯 Python，零依赖，几 KB 大小
• 语法直观，像自然语言
• 适合单机定时任务，不需要 Redis 等外部依赖
• 不支持秒级精度，但分钟级足够

2.3 为什么不用 APScheduler 或 Celery

三、完整流水线串联

3.1 流水线编排

importscheduleimporttimeimportsubprocessfromdatetimeimportdatetimefrompathlibimportPath PROJECT_ROOT=Path(__file__).parent.parent SCRIPTS={"crawl":PROJECT_ROOT/"code/lesson-13-web-crawler/ithome_crawler.py","clean":PROJECT_ROOT/"code/lesson-15-data-cleaning/data_cleaner.py","insight":PROJECT_ROOT/"code/lesson-17-competitor-insight/insight_generator.py","report":PROJECT_ROOT/"code/lesson-18-docx-report/docx_report_generator.py",}defrun_step(name:str,script:Path)->bool:"""运行一个步骤，返回是否成功"""print(f"[{datetime.now():%H:%M:%S}] 开始:{name}")try:result=subprocess.run(["python",str(script)],capture_output=True,text=True,timeout=300)ifresult.returncode==0:print(f" ✅{name}完成")returnTrueelse:print(f" ❌{name}失败:{result.stderr[:200]}")returnFalseexceptExceptionase:print(f" ❌{name}异常:{e}")returnFalsedefrun_pipeline():"""运行完整流水线"""print(f"\n{'='*50}")print(f"[{datetime.now():%Y-%m-%d%H:%M:%S}] 开始执行竞品监控流水线")print(f"{'='*50}")steps=[("爬取新闻",SCRIPTS["crawl"]),("数据清洗",SCRIPTS["clean"]),("生成洞察",SCRIPTS["insight"]),("生成报告",SCRIPTS["report"]),]forname,scriptinsteps:ifnotrun_step(name,script):print(f" ⚠ 流水线在「{name}」中断，跳过后续步骤")returnprint(f"[{datetime.now():%H:%M:%S}] 流水线完成！")

3.2 带重试的流水线

defrun_step_with_retry(name:str,script:Path,max_retries=3)->bool:"""带重试的步骤执行"""forattemptinrange(1,max_retries+1):ifattempt>1:wait=2**attempt# 指数退避: 2s, 4s, 8sprint(f" 重试{attempt}/{max_retries}，等待{wait}s...")time.sleep(wait)ifrun_step(name,script):returnTrueprint(f" ❌{name}重试{max_retries}次后仍失败")returnFalse

四、日志系统：无人值守的"眼睛"

4.1 为什么日志重要

自动化系统运行在后台，你看不到它的状态。日志就是你的"眼睛"：

• 今天爬了多少条新闻？→ 看日志
• 为什么报告没生成？→ 看日志
• 哪个步骤耗时最长？→ 看日志

4.2 结构化日志

importloggingfromlogging.handlersimportRotatingFileHandlerdefsetup_logger(log_dir:Path):"""配置日志系统"""log_dir.mkdir(exist_ok=True)logger=logging.getLogger("pipeline")logger.setLevel(logging.INFO)# 文件日志（自动轮转，每个文件最多 10MB，保留 5 个）file_handler=RotatingFileHandler(log_dir/"pipeline.log",maxBytes=10*1024*1024,backupCount=5,encoding="utf-8",)file_handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))# 控制台日志console_handler=logging.StreamHandler()console_handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))logger.addHandler(file_handler)logger.addHandler(console_handler)returnlogger

4.3 日志输出示例

2026-08-16 09:00:01 [INFO] ========== 流水线开始 ========== 2026-08-16 09:00:15 [INFO] ✅ 爬取新闻: 35 条 2026-08-16 09:00:32 [INFO] ✅ 数据清洗: 35 → 32 条 2026-08-16 09:01:05 [INFO] ✅ 生成洞察: 7 条 2026-08-16 09:01:18 [INFO] ✅ 生成报告: 竞品监控周报_2026-08-16.docx 2026-08-16 09:01:18 [INFO] ========== 流水线完成 ==========

五、Windows 任务计划程序集成

5.1 为什么需要系统级调度

schedule 库运行在 Python 进程里，进程退出就停了。Windows 任务计划程序是系统级调度：

• 开机自启动
• 进程崩溃后自动重启
• 支持多种触发器（每天、每小时、系统启动时）

5.2 创建计划任务

用 PowerShell 创建：

$action=New-ScheduledTaskAction-Execute"python"`-Argument"C:\ai-agent-tutorial\code\lesson-19-automation\pipeline.py"`-WorkingDirectory"C:\ai-agent-tutorial"$trigger=New-ScheduledTaskTrigger-Daily-At 9:00AM$principal=New-ScheduledTaskPrincipal-UserId"SYSTEM"`-LogonType ServiceAccount-RunLevel HighestRegister-ScheduledTask-TaskName"AI竞品监控"`-Action$action-Trigger$trigger-Principal$principal`-Description"每天自动运行竞品监控流水线"

5.3 两种方案对比

建议：开发阶段用 schedule 调试，生产环境用 Windows 任务计划。

六、监控与告警

6.1 流水线健康检查

defhealth_check():"""检查流水线是否正常运行"""# 检查今天是否已生成报告today=datetime.now().strftime("%Y-%m-%d")report_path=OUTPUT_DIR/f"竞品监控周报_{today}.docx"ifreport_path.exists():logger.info(f"✅ 今日报告已生成:{report_path}")else:logger.warning(f"⚠ 今日报告尚未生成！")

6.2 失败通知

defsend_alert(message:str):"""发送告警（可以扩展为邮件/钉钉/飞书）"""# 当前版本：写入告警日志alert_path=LOG_DIR/"alerts.log"withopen(alert_path,"a",encoding="utf-8")asf:f.write(f"[{datetime.now()}]{message}\n")# 扩展：发送邮件# send_email(to="admin@company.com", subject="竞品监控异常", body=message)# 扩展：飞书机器人通知# requests.post(webhook_url, json={"msg_type": "text", "content": {"text": message}})

七、总结

至此，场景一「竞品监控」全流程自动化完成。下一课，我们将对这一场景进行完整交付——效果验证、价值复盘、ROI 计算。

本课代码：code/lesson-19-automation/pipeline.py