第 19 课 | 定时任务与自动化让系统 7x24 小时运行前 18 课我们手动运行每一个脚本。这节课我们让整个系统自动运转——每天定时爬取、清洗、分析、生成报告全部无人值守。一、业务价值从工具到系统1.1 手动 vs 自动步骤手动方式自动化方式爬取新闻python crawler.py每天 9:00 自动执行数据清洗python cleaner.py爬取完成后自动触发生成洞察python insight.py清洗完成后自动触发生成报告python report.py洞察完成后自动触发发送通知手动发邮件报告完成后自动发送总计每天 15 分钟0 秒核心价值不会忘记、不会偷懒、不会出错。你休假时系统照常运转。1.2 自动化流水线全景失败仍失败失败失败⏰ 定时触发每天 9:00爬虫采集新闻数据清洗pandas LLM竞品洞察LLM 分析报告生成python-docx通知日志/邮件重试 3 次指数退避告警日志二、Python schedule 库轻量级任务调度2.1 安装uv pip install schedule2.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方案适用场景复杂度我们的选择schedule单机、简单定时低✅ 当前使用APScheduler需要持久化、多种触发器中进阶可选Celery分布式、大规模任务高未到需要时Windows 任务计划系统级、开机自启低生产环境配合三、完整流水线串联3.1 流水线编排importscheduleimporttimeimportsubprocessfromdatetimeimportdatetimefrompathlibimportPath PROJECT_ROOTPath(__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:resultsubprocess.run([python,str(script)],capture_outputTrue,textTrue,timeout300)ifresult.returncode0: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_retries3)-bool:带重试的步骤执行forattemptinrange(1,max_retries1):ifattempt1:wait2**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_okTrue)loggerlogging.getLogger(pipeline)logger.setLevel(logging.INFO)# 文件日志自动轮转每个文件最多 10MB保留 5 个file_handlerRotatingFileHandler(log_dir/pipeline.log,maxBytes10*1024*1024,backupCount5,encodingutf-8,)file_handler.setFormatter(logging.Formatter(%(asctime)s [%(levelname)s] %(message)s))# 控制台日志console_handlerlogging.StreamHandler()console_handler.setFormatter(logging.Formatter(%(asctime)s [%(levelname)s] %(message)s))logger.addHandler(file_handler)logger.addHandler(console_handler)returnlogger4.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 创建$actionNew-ScheduledTaskAction-Executepython-ArgumentC:\ai-agent-tutorial\code\lesson-19-automation\pipeline.py-WorkingDirectoryC:\ai-agent-tutorial$triggerNew-ScheduledTaskTrigger-Daily-At 9:00AM$principalNew-ScheduledTaskPrincipal-UserIdSYSTEM-LogonType ServiceAccount-RunLevel HighestRegister-ScheduledTask-TaskNameAI竞品监控-Action$action-Trigger$trigger-Principal$principal-Description每天自动运行竞品监控流水线5.3 两种方案对比维度schedule 库Windows 任务计划部署复杂度低pip install中PowerShell 配置开机自启需额外配置✅ 原生支持崩溃恢复需额外处理✅ 自动重启跨平台✅❌ 仅 Windows适用阶段开发/测试生产环境建议开发阶段用 schedule 调试生产环境用 Windows 任务计划。六、监控与告警6.1 流水线健康检查defhealth_check():检查流水线是否正常运行# 检查今天是否已生成报告todaydatetime.now().strftime(%Y-%m-%d)report_pathOUTPUT_DIR/f竞品监控周报_{today}.docxifreport_path.exists():logger.info(f✅ 今日报告已生成:{report_path})else:logger.warning(f⚠ 今日报告尚未生成)6.2 失败通知defsend_alert(message:str):发送告警可以扩展为邮件/钉钉/飞书# 当前版本写入告警日志alert_pathLOG_DIR/alerts.logwithopen(alert_path,a,encodingutf-8)asf:f.write(f[{datetime.now()}]{message}\n)# 扩展发送邮件# send_email(toadmincompany.com, subject竞品监控异常, bodymessage)# 扩展飞书机器人通知# requests.post(webhook_url, json{msg_type: text, content: {text: message}})七、总结能力工具效果定时调度schedule 库每天 9:00 自动触发流水线编排链式调用 重试4 步串联失败自动重试日志记录RotatingFileHandler完整追踪自动轮转系统级调度Windows 任务计划开机自启崩溃恢复健康检查文件检测确认报告已生成至此场景一「竞品监控」全流程自动化完成。下一课我们将对这一场景进行完整交付——效果验证、价值复盘、ROI 计算。本课代码code/lesson-19-automation/pipeline.py 「AI Agent 全栈开发 50 讲」——从本地模型部署到多智能体系统，一年省 87 万 第 19 课 | 定时任务与自动化：让系统 7x24 小时运行
第 19 课 | 定时任务与自动化让系统 7x24 小时运行前 18 课我们手动运行每一个脚本。这节课我们让整个系统自动运转——每天定时爬取、清洗、分析、生成报告全部无人值守。一、业务价值从工具到系统1.1 手动 vs 自动步骤手动方式自动化方式爬取新闻python crawler.py每天 9:00 自动执行数据清洗python cleaner.py爬取完成后自动触发生成洞察python insight.py清洗完成后自动触发生成报告python report.py洞察完成后自动触发发送通知手动发邮件报告完成后自动发送总计每天 15 分钟0 秒核心价值不会忘记、不会偷懒、不会出错。你休假时系统照常运转。1.2 自动化流水线全景失败仍失败失败失败⏰ 定时触发每天 9:00爬虫采集新闻数据清洗pandas LLM竞品洞察LLM 分析报告生成python-docx通知日志/邮件重试 3 次指数退避告警日志二、Python schedule 库轻量级任务调度2.1 安装uv pip install schedule2.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方案适用场景复杂度我们的选择schedule单机、简单定时低✅ 当前使用APScheduler需要持久化、多种触发器中进阶可选Celery分布式、大规模任务高未到需要时Windows 任务计划系统级、开机自启低生产环境配合三、完整流水线串联3.1 流水线编排importscheduleimporttimeimportsubprocessfromdatetimeimportdatetimefrompathlibimportPath PROJECT_ROOTPath(__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:resultsubprocess.run([python,str(script)],capture_outputTrue,textTrue,timeout300)ifresult.returncode0: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_retries3)-bool:带重试的步骤执行forattemptinrange(1,max_retries1):ifattempt1:wait2**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_okTrue)loggerlogging.getLogger(pipeline)logger.setLevel(logging.INFO)# 文件日志自动轮转每个文件最多 10MB保留 5 个file_handlerRotatingFileHandler(log_dir/pipeline.log,maxBytes10*1024*1024,backupCount5,encodingutf-8,)file_handler.setFormatter(logging.Formatter(%(asctime)s [%(levelname)s] %(message)s))# 控制台日志console_handlerlogging.StreamHandler()console_handler.setFormatter(logging.Formatter(%(asctime)s [%(levelname)s] %(message)s))logger.addHandler(file_handler)logger.addHandler(console_handler)returnlogger4.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 创建$actionNew-ScheduledTaskAction-Executepython-ArgumentC:\ai-agent-tutorial\code\lesson-19-automation\pipeline.py-WorkingDirectoryC:\ai-agent-tutorial$triggerNew-ScheduledTaskTrigger-Daily-At 9:00AM$principalNew-ScheduledTaskPrincipal-UserIdSYSTEM-LogonType ServiceAccount-RunLevel HighestRegister-ScheduledTask-TaskNameAI竞品监控-Action$action-Trigger$trigger-Principal$principal-Description每天自动运行竞品监控流水线5.3 两种方案对比维度schedule 库Windows 任务计划部署复杂度低pip install中PowerShell 配置开机自启需额外配置✅ 原生支持崩溃恢复需额外处理✅ 自动重启跨平台✅❌ 仅 Windows适用阶段开发/测试生产环境建议开发阶段用 schedule 调试生产环境用 Windows 任务计划。六、监控与告警6.1 流水线健康检查defhealth_check():检查流水线是否正常运行# 检查今天是否已生成报告todaydatetime.now().strftime(%Y-%m-%d)report_pathOUTPUT_DIR/f竞品监控周报_{today}.docxifreport_path.exists():logger.info(f✅ 今日报告已生成:{report_path})else:logger.warning(f⚠ 今日报告尚未生成)6.2 失败通知defsend_alert(message:str):发送告警可以扩展为邮件/钉钉/飞书# 当前版本写入告警日志alert_pathLOG_DIR/alerts.logwithopen(alert_path,a,encodingutf-8)asf:f.write(f[{datetime.now()}]{message}\n)# 扩展发送邮件# send_email(toadmincompany.com, subject竞品监控异常, bodymessage)# 扩展飞书机器人通知# requests.post(webhook_url, json{msg_type: text, content: {text: message}})七、总结能力工具效果定时调度schedule 库每天 9:00 自动触发流水线编排链式调用 重试4 步串联失败自动重试日志记录RotatingFileHandler完整追踪自动轮转系统级调度Windows 任务计划开机自启崩溃恢复健康检查文件检测确认报告已生成至此场景一「竞品监控」全流程自动化完成。下一课我们将对这一场景进行完整交付——效果验证、价值复盘、ROI 计算。本课代码code/lesson-19-automation/pipeline.py