1. 开篇企业级脚本的“四根支柱”做脚本开发这行当久了你会发现一个特别有意思的现象很多人的脚本能跑但只是“能跑”。放在自己电脑上数据一正常、网络一顺畅、参数一凑巧脚本“唰”一下就过了一切岁月静好。可一旦扔到生产环境、放到定时任务里、或者交到别人手里去用问题就像雨后春笋一样往外冒好好的数据怎么突然不对了报错信息打印了一半就停了昨晚凌晨三点跑的脚本到底成功没成功没人知道。真正拉开普通脚本和企业级脚本差距的不是用了什么高深的框架也不是代码写得多么花哨而是四个基础得不能再基础的组件断言、日志、异常处理、重试机制。这四个东西单独拎出来任何一个写过几年代码的人都认识但真正把它们组合在一起、按照企业级标准落到脚本里的并不多。这篇我就结合自己多年的实战经验把这四块掰开揉碎了讲清楚它们各自解决什么问题、怎么设计才算合格、组合在一起会有什么化学反应以及我踩过哪些坑。这篇内容适合谁如果你正在写自动化测试脚本、数据处理脚本、定时任务、部署发布脚本或者任何“跑起来就不太想管”的后台脚本这篇文章就是给你准备的。你可以不用读完立刻重构所有代码但至少下次再写脚本时会下意识地问自己一句这里挂了我能知道吗这里错了日志里有线索吗这次失败值得再试一次吗2. 断言让脚本自己检查结果而不是靠人肉盯输出2.1 断言的本质是什么很多人对断言的理解停留在“assert 关键字”这个层面甚至有些写脚本的人会说“断言还要单独学吗不就是判断一下结果对不对”话是没错但断言的意义远不止“判断一下”。断言是脚本给自己设置的一道路障是自动化世界里最基础的“自我验收”机制。想象一个场景你写了个脚本去抓取某个网页的数据然后把数据写入数据库。脚本顺利跑完了打印了一句“抓取成功”。听起来挺好吧但你真的“成功”了吗如果网页改版了、字段名变了、返回的是个404页面你的脚本是不是还能“顺利跑完”并且打印“成功”如果答案是肯定的那这个脚本就处于最危险的“假成功”状态——它没有断言它不关心结果它只是机械地执行完流程然后把责任全部推给事后看日志的人。断言的本质是把“验证结果”这件事从人肉环节里剥离出来交给脚本自己去完成。一个合格的断言会明确回答三个问题我期待得到什么我实际得到了什么两者一致吗只有这三个问题都得到肯定回答脚本才有资格说“我这一步成功了”。2.2 断言的两条铁律我写过、review过很多带断言的脚本总结下来最值得记住的铁律有两条。第一条铁律断言的优先级很高但它要断在“关键路径”上。初学者最容易犯的毛病是到处加断言恨不得每一行都验证一下结果断言本身比业务代码还多维护成本陡然上升。我个人的习惯是只对“这一步错了后面就全废了”的节点做断言。比如登录成功后断言 token 字段存在下载完文件后断言文件大小大于0且文件头是预期的格式解析完 JSON 后断言关键字段的 key 存在。至于某个中间变量的取值是否符合预期那是单元测试的事不该塞在业务脚本里。第二条铁律断言失败必须是“硬失败”不能是“软提醒”。很多人在脚本里写“if result ! expected: print(结果不对)”然后脚本继续往下跑。这就把断言降级成了日志失去了它最核心的价值——拦截。断言一旦失败脚本就应该当场抛异常、中止执行、返回非零退出码让调度系统或上层调用者立刻感知到问题。在 Python 里用assert或者主动raise AssertionError在 Shell 里用set -e配合grep -q或test条件判断实现“不满足就退出”。顺便多说一句断言失败时输出的信息一定要带着“现场数据”别只写一句“结果校验失败”。我见过太多断言报错长这样AssertionError: data is wrong。等你顺着代码找过去根本不知道到底是哪个 data、哪一步的 data。合格的断言信息应该长这样AssertionError: login response missing token, status_code500, bodyhtmlbodyInternal Server Error/body/html。这才叫可排查的断言。2.3 shell / JMeter 场景里的断言长什么样既然热搜词里频繁出现了jmeter断言怎么写和shell脚本入门我就顺便把这两个场景的断言写法也梳理一下因为它们和通用的代码断言略有差异。JMeter 里做断言很多人第一反应是用“响应断言”检查响应文本里是否包含某个关键字。这个做法没错但要注意断言要跟取样器的层级正确对应而且断言可以叠加多种策略。我在实际项目中会同时配三层断言第一层是 HTTP 状态码必须是 200第二层是响应文本必须包含核心业务字段比如订单号的前缀、用户 ID 等第三层是响应时间超过阈值就判定为失败。三层全过才算一次成功的请求。这样设计的好处是接口返回 200 但业务逻辑报错比如返回了 JSON 里的 error_code 非 0也能被第二层或精算断言抓出来接口返回正常但性能劣化第三层会兜底。Shell 脚本里做断言核心思路是利用退出码和条件表达式。我常用的一个模式是#!/usr/bin/env bash # 下载文件并校验非空 curl -fsSL https://example.com/data.json -o /tmp/data.json if [ ! -s /tmp/data.json ]; then echo ERROR: download failed or file is empty 2 exit 1 fi # 校验核心字段 grep -q status: ok /tmp/data.json if [ $? -ne 0 ]; then echo ERROR: unexpected status field in /tmp/data.json 2 exit 1 fi echo SUCCESS: all assertions passed这里-f让 curl 在 HTTP 错误时返回非零退出码-s静默下载-S让错误信息仍然输出。[ ! -s file ]断言文件存在且非空grep -q断言内容包含预期字段。每一步失败都会带着明确错误信息退出而不是假装成功。这套写法我用了很多年稳定、直观、零依赖。3. 日志脚本唯一的“事后现场”3.1 日志的第一原则如果说断言是脚本在运行时的“自我检查”那日志就是脚本留给世界的“案发现场”。断言负责发现问题日志负责还原问题。一个脚本可以没有 UI、没有文档、没有注释但绝对不能没有日志。日志的第一原则是什么很多人会说是“详细”但我认为恰恰相反第一原则应该是**“可检索”**。日志写得再多出了问题的时候你 grep 不到关键信息那和没有日志没有区别。所以我在团队里一直强调一个标准每一行日志都必须是“可 grep、可追溯、可定位”的。什么叫可 grep就是出了问题你在日志文件里搜一个关键词比如订单号、任务 ID、IP 地址相关的日志行一搜就能拉出来前后上下文一拼事情经过就能还原个七八成。有一次我们一个数据同步脚本在凌晨三点半静默失败什么都查不到。后来翻了半天发现日志里压根没打任何标记性的业务关键词全是“INFO: processing”这种通用输出每条都长一个样。这就是典型的不合格日志。3.2 生产级日志的四件套我每次写脚本不管用什么语言日志输出都至少包含四件事时间、级别、上下文、消息体。四个元素缺一不可。时间精确到秒是起步生产环境建议到毫秒。没有时间的日志排查问题时连先后顺序都确定不了尤其是异步任务并行跑的时候没有时间戳的日志根本没法看。级别DEBUG / INFO / WARNING / ERROR 四档虽然像是老生常谈但执行得严格的项目真不多。我见过太多脚本把什么都打成 INFO结果想过滤错误日志都无从下手。定一条硬规则只有“现场数据详细信息”才用 DEBUG正常的进度节点用 INFO可能有问题但脚本还能继续跑的用 WARNING会导致当前步骤失败或数据不可用的必须用 ERROR。上下文这是最容易被忽略的一个。所谓上下文就是“这条日志是属于哪一次执行、哪个任务、哪个请求的”。多线程脚本、定时任务、多次循环里没有上下文的日志全都是噪音。我的做法是在脚本开始时就生成一个run_id用时间戳加随机串即可然后日志的每一行都带上run_id这样即使多个任务并行写同一个日志文件你grep run_id也能把某一次任务的全部日志完整提取出来。消息体日志的真正内容必须包含“发生了什么 关键变量/数据”。举个对比ERROR: request failed和ERROR: request to https://api.example.com/order/create failed after 3 retries, status502, resp_body{code:502,msg:bad gateway}差距有多大不用我多说。Python 里用 logging 模块的话我推荐直接用logging.basicConfig配上自定义 Formatter把上面四件套固化下来import logging logging.basicConfig( levellogging.INFO, format%(asctime)s | %(levelname)s | run_id%(run_id)s | %(message)s, handlers[ logging.StreamHandler(), logging.FileHandler(app.log, encodingutf-8) ] ) logger logging.getLogger(myscript) logger logging.LoggerAdapter(logger, {run_id: 20250101_001}) logger.info(task started, order_id%s, A10086)这里的LoggerAdapter是给所有日志自动带上 run_id 的方便做法省得每条都手写。FileHandler 会把日志同时写文件StreamHandler 会打到控制台调试和留痕两不误。3.3 日志轮转一个你迟早会踩的坑日志还有一个容易忽略但必须面对的问题日志文件会无限增长。开发环境跑几个脚本无所谓生产环境的定时任务如果一天跑一次每次打几 MB 日志几个月不清理就是 GB 级别的垃圾文件甚至可能把磁盘占满。热搜词里“binlog日志可以删除吗”“redis日志”这类问题本质上都是日志管理没做好。解决日志增长的标准方案是log rotation日志轮转。Python 的 logging 自带RotatingFileHandler按大小切分超了自动归档from logging.handlers import RotatingFileHandler handler RotatingFileHandler( app.log, maxBytes10 * 1024 * 1024, # 单个日志文件 10MB 封顶 backupCount5 # 最多保留 5 个备份文件 )Shell 脚本里处理起来更直接用 logrotate 这个系统工具在/etc/logrotate.d/下写一个配置/path/to/your/script.log { daily rotate 7 compress missingok notifempty copytruncate }这里的几个参数解释一下daily表示每天轮转一次rotate 7保留最近 7 个归档compress归档时压缩成 gzipcopytruncate是很多脚本类日志必须加的一个参数——它先把原文件拷贝走再把原文件截断这样即使脚本一直持有文件句柄也不会写乱。不加copytruncate的话进程不重启日志就可能写丢了。3.4 从单机日志到集中日志说完本地日志顺带提一下日志的后端演进。热搜词里有loki日志系统和windows安全日志这正好对应了两个方向生产环境服务的日志通常需要集中收集而系统安全日志则需要审计留存。如果你的脚本是跑在多台服务器上的那么单机日志的排查效率会断崖式下降。这时候就要引入集中日志系统。这个领域的常见方案是 Loki 配合 Promtail也可以是 ELKElasticsearch Logstash Kibana甚至简单一点直接通过 rsyslog 把日志转发到一台中心服务器也行。选型的核心判断标准只有一个你能不能在 5 分钟内根据一个任务 ID 把所有机器上的相关日志全部捞出来。Loki 这类方案比 ELK 轻量很多索引成本低特别适合“脚本日志”这种流量中等、结构简单的场景。我记得最早带团队的时候排查跨服务器的数据不一致问题经常要 ssh 登录三四台机器挨个 grep 日志文件遇到日志过大还得配合tail、awk慢慢翻折腾一两个小时是常有的事。后来统一接入了集中日志同样的排查一条 LogQL 查询语句{appsync_script} | order_idA10086几秒钟直接出来所有关联日志。这个效率提升是几何级别的。4. 异常允许失败但要优雅地失败4.1 异常三问异常处理可能是这四个主题里代码量最少、但含金量最高的一个。很多脚本“看起来没问题”但一到异常场景就崩溃得一塌糊涂根本原因是对异常缺少体系化的思考。我总结了三个问题每次写异常处理前先问自己一遍第一问这个操作可能会怎么失败网络请求超时、磁盘空间不足、数据库连接被拒、下游接口返回了非 JSON 内容、文件被其他进程锁住……先穷举失败模式再针对性地写处理逻辑而不是一把梭try...except Exception完事。第二问失败之后脚本应该怎么办有些异常比如网络抖动是可以等一下重试的有些异常比如认证失败重试一万次也没用应该立刻停止还有一些异常比如某个非核心数据解析失败可以降级处理——跳过这条数据记录 warning让脚本继续跑。这三种处理方式对应着三种截然不同的代码路径。第三问用户或者调用方需要知道什么脚本被谁调用人就可能被谁监控。如果是定时任务任务调度系统需要非零退出码来感知失败如果有上层 Web 接口那就需要把异常转换成合适的 HTTP 状态码返回如果只是开发者本地调试那异常信息就要尽可能详细最好能直接定位到代码行和关键变量。4.2 try-except 的分层处理异常处理最常见的错误是只在最外层写一个大大的try...except Exception: logger.error(something wrong)然后把所有异常一视同仁地吞掉。这种写法的问题在于你只知道“出了错”却不知道“错在哪一层、错在什么环节”排查起来照样两眼一抹黑。更合理的做法是分层处理 逐层转换。底层函数捕获异常后加上“当前上下文信息”再向上抛出顶层调用处用不同的 except 分支处理不同类型的异常。举个例子我常用的一个带重试的 HTTP 请求函数异常处理是这样设计的import requests import time def fetch_with_retry(url, timeout5, retries3): for attempt in range(1, retries 1): try: resp requests.get(url, timeouttimeout) resp.raise_for_status() return resp.json() except requests.exceptions.Timeout as e: # 超时属于临时性故障值得重试 logger.warning(attempt %s: request timeout to %s, error%s, attempt, url, e) if attempt retries: raise RuntimeError(ffetch {url} failed: timeout after {retries} attempts) from e time.sleep(2 ** attempt) except requests.exceptions.HTTPError as e: # 4xx 一般是不可恢复的5xx 可以再试一次 status_code e.response.status_code if e.response is not None else -1 if status_code 500 and attempt retries: logger.warning(attempt %s: server error %s from %s, retrying..., attempt, status_code, url) time.sleep(2 ** attempt) continue raise RuntimeError(ffetch {url} failed: http_status{status_code}) from e # 理论上走不到这里但加个兜底 raise RuntimeError(ffetch {url} failed: unknown error after {retries} attempts)这个函数的关键点在于超时和 5xx 被识别为临时性故障会触发重试4xx 和最终的重试耗尽会被转换成带有完整上下文的 RuntimeError由上层统一处理。raise ... from e会保留原始异常链排查时不会丢失最底层的 cause。4.3 前置校验最好的异常处理是让异常不发生讲完 try-except再讲一个很多人忽略的异常处理思路——前置校验。热搜词里有一批典型的“启动类异常”报错比如“终端进程启动失败: 启动期间发生本机异常”“无法将‘npm’项识别为 cmdlet、函数、脚本文件或可运行程序的名称”“由于 Windows 无法加载这个设备所需的驱动程序导致这个设备工作异常 (代码 31)”这些本质上都不是运行期异常而是前置条件不满足导致的失败。脚本启动时最不值钱的异常是“环境不满足”。与其等到脚本跑到第 30 步才报找不到命令、连不上数据库不如在入口处就把所有必要条件检查一遍#!/usr/bin/env bash # 前置校验命令存在性检查 for cmd in curl jq mysql; do if ! command -v $cmd /dev/null 21; then echo ERROR: required command $cmd not found in PATH 2 exit 1 fi done # 前置校验配置文件存在且非空 if [ ! -s config.json ]; then echo ERROR: config.json missing or empty 2 exit 1 fi # 前置校验关键目录可写 if [ ! -w /var/log/myscript ]; then echo ERROR: log directory /var/log/myscript is not writable 2 exit 1 fi echo INFO: all pre-checks passed, starting main logic... # 主逻辑...Python 里可以在启动时用一个validate_environment()函数把所有依赖的包、环境变量、目录权限、数据库连通性全部检查一遍任何一项不满足就带着明确的中文错误信息退出。这些前置校验在脚本生命周期里只浪费几秒钟但能免除无数次“跑到一半才发现环境不对”的痛苦。4.4 静默失败脚本稳定性最大的敌人异常处理里最隐蔽的坑不是异常抛得太多而是异常被悄悄吞掉。except Exception: pass这种写法是脚本事故的万恶之源。为什么很多人会写pass因为当时写的时候觉得“这个异常不重要跳过算了”但生产环境里这些被静默吞掉的异常往往意味着数据丢失、任务中断、流程错乱。我印象最深的一次事故一个定时统计脚本里有一段except Exception: pass原本是用来忽略“某个子模块解析失败”的。结果数据源格式换版后这段代码每天都会静默跳过主统计逻辑脚本每天“成功退出”但产出的报表数据是空的。这个问题持续了整整两周直到业务方发现数据对不上才排查出来。从那以后我们定了一条死规矩任何 except 分支都必须至少写一条日志记录异常类型、异常信息和现场上下文。你可以决定不处理某个异常但你不能决定不留下任何痕迹。这一条建议所有人都刻进骨髓里。5. 重试向“临时性故障”开战5.1 重试三要素间隔、次数、条件重试机制是脚本稳定性里最“性感”的部分因为它不是被动地防御而是主动地修复。不过重试不是简单地写个 for 循环把失败的操作再调一遍。没有策略的重试是自杀式攻击尤其是对下游服务来说无脑重试可能引发“重试风暴”反而把已经过载的服务彻底打挂。合格的重试机制至少要回答三个问题间隔多久重试、最多重试几次、哪些错误值得重试。间隔不能是固定的。固定间隔重试的典型问题如果下游服务已经过载你每秒重试一次等于持续对服务施压服务永远恢复不过来。行业通用做法是指数退避exponential backoff第一次失败后等 1 秒第二次等 2 秒第三次等 4 秒第四次等 8 秒……每次等待时间翻倍。这样既给了下游恢复的时间也避免了自己疯狂空转。实际工程中通常还会在退避时间上加一个“抖动jitter”防止多个客户端同时重试造成同步共振。Python 的tenacity库和 Go 的backoff库都内置了这两种能力不用自己手搓。重试次数要有限制。没有上限的重试等同于死循环脚本会卡在一个失败的请求上无限等待。我常用的默认值是 3 到 5 次也就是“初始 1 次 重试 3~4 次”。这个数字是根据多数线上故障的恢复时间尺度和脚本整体执行时间的预算综合得来的。退了 3~4 次之后如果还是失败说明这不是一个“等一下就能好”的临时问题及时止损、抛出异常才对。哪些错误值得重试这个判断最关键。前面已经提到了按状态码区分5xx、超时、连接重置属于临时性故障值得重试4xx 通常是请求本身有问题参数错了、没权限、资源不存在重试多少次都不会成功反而浪费资源和时间。对于数据库操作死锁重试是有意义的但字段长度超限、主键冲突这种重试毫无意义。5.2 一个现成的重试实现Python tenacity 库如果你用 Python我不建议自己写重试装饰器直接用tenacity这个库功能完整API 也直观。下面是我在一个数据采集脚本里实际用过的配置from tenacity import ( retry, stop_after_attempt, wait_exponential, retry_if_exception_type, before_sleep_log ) import logging import requests logger logging.getLogger(__name__) retry( stopstop_after_attempt(5), # 最多尝试 5 次 waitwait_exponential(multiplier1, min1, max16), # 指数退避 1s, 2s, 4s, 8s, 16s retryretry_if_exception_type(( requests.exceptions.Timeout, requests.exceptions.ConnectionError, )), before_sleepbefore_sleep_log(logger, logging.WARNING), reraiseTrue, ) def fetch_data(url): resp requests.get(url, timeout5) resp.raise_for_status() return resp.json()这个写法的好处是重试策略和业务逻辑彻底分离装饰器一加函数就自动拥有了“临时性失败自动重试 最终失败向上抛”的能力。before_sleep会在每次准备重试前打一条 warning 日志这样你事后看日志能清楚地知道“这个操作失败了三次、分别在哪些时间点重试的”。reraiseTrue表示重试耗尽后抛出最后一次异常保留原始异常链。5.3 重试的隐藏队友超时和幂等讲重试就必须讲它的两个“隐藏队友”——超时和幂等。没有超时的重试是灾难没有幂等的重试是双倍灾难。先说超时。一个 HTTP 请求如果设置了 60 秒超时遇到下游服务假死时一次重试周期可能白白消耗几分钟。我见过最离谱的例子某个脚本调第三方接口没设超时底层 socket 默认超时是 120 秒重试 5 次光这一个请求就能卡十几分钟。所以任何会阻塞的 I/O 操作都必须显式设置超时HTTP 请求设连接超时和读超时数据库操作设 statement timeout文件锁获取设等待超时。重试结合超时才有意义——一次失败的尝试必须在可预期的时间内结束重试调度才有节奏感。再说幂等。什么叫幂等就是同一个操作执行一次和执行十次产生的效果是一样的。重试的本质是“同一件事多做几遍”如果这件事不幂等重试就会导致重复下单、重复扣款、重复插入数据。对脚本来说重试接口请求时最好带上一个全局唯一的请求 ID幂等键让下游服务能够识别并去重重试数据库写入时可以用唯一索引做兜底重复插入会遇到主键冲突而不会产生脏数据重试文件操作时可以采用“先写临时文件再原子重命名”的策略避免重复执行导致文件内容覆盖错乱。没有幂等保障的重试宁可不要重试。5.4 重试之后的“最后防线”最后要强调一点重试不是万能的必须有明确的“放弃策略”和“失败上报”。重试全部耗尽后脚本必须做两件事一是给出清晰的中文错误报告说明尝试了多少次、每次的失败原因、最后一次的异常详情二是以非零退出码退出让上层调度器或监控系统能够感知失败。我见过不少脚本把重试写成“无限重试 不报错”的死循环形态理由往往是“我要确保任务一定成功”。但在分布式系统环境下这是最危险的思维。有些故障比如下游服务宕机可能要持续半小时以上你重试一次耗时数十秒无限重试下去调度队列会被卡死其他任务全部积压。正确的企业级做法是有限次重试 快速失败 失败进入待处理队列或触发告警。把问题暴露出来交给更上层的机制去决策才是稳定的王道。6. 实战组合一个企业级脚本块长什么样前面把断言、日志、异常、重试拆开讲了但实际生产环境里这四个组件永远是组合使用的它们互相配合才能形成一道完整的防线。我直接给一个比较完整的 Python 脚本骨架注释里标明了每个环节的设计意图。这个骨架来自我之前写的一个电商订单同步脚本的简化版经过了线上业务验证参考价值比较高。#!/usr/bin/env python3 # -*- coding: utf-8 -*- 订单增量同步脚本企业级骨架示例 功能从内部订单 API 拉取增量订单写入 MySQL 业务库 设计要点断言 结构化日志 分级异常 有限重试 import hashlib import json import logging import sys import time import uuid from datetime import datetime import mysql.connector import requests from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type, before_sleep_log # ------------------------------------------------------------ # 日志初始化控制台 文件双写带 run_id 上下文 # ------------------------------------------------------------ RUN_ID datetime.now().strftime(%Y%m%d_%H%M%S) _ uuid.uuid4().hex[:8] class ContextFilter(logging.Filter): def filter(self, record): record.run_id RUN_ID return True logger logging.getLogger(order_sync) logger.setLevel(logging.INFO) fmt logging.Formatter(%(asctime)s | %(levelname)s | run_id%(run_id)s | %(message)s) console logging.StreamHandler() console.setFormatter(fmt) logger.addHandler(console) file_handler logging.handlers.RotatingFileHandler( /var/log/order_sync/app.log, maxBytes10*1024*1024, backupCount5, encodingutf-8 ) file_handler.setFormatter(fmt) logger.addHandler(file_handler) logger.addFilter(ContextFilter()) # ------------------------------------------------------------ # 配置区实际项目建议从环境变量读取 # ------------------------------------------------------------ API_URL https://api.internal.example.com/v1/orders/sync API_TOKEN YOUR_TOKEN DB_CONFIG { host: 127.0.0.1, port: 3306, user: app, password: YOUR_PASS, database: order_db, } # ------------------------------------------------------------ # 带重试的 HTTP 拉取只重试临时性故障 # ------------------------------------------------------------ retry( stopstop_after_attempt(5), waitwait_exponential(multiplier1, min1, max16), retryretry_if_exception_type(( requests.exceptions.Timeout, requests.exceptions.ConnectionError, )), before_sleepbefore_sleep_log(logger, logging.WARNING), reraiseTrue, ) def fetch_orders(begin_time: str, end_time: str) - list: 拉取指定时间窗口的增量订单。5xx 也会被重试。 resp requests.get( API_URL, params{begin_time: begin_time, end_time: end_time}, headers{Authorization: fBearer {API_TOKEN}}, timeout(3, 10), # 连接超时 3s读超时 10s ) resp.raise_for_status() data resp.json() # 业务级断言接口返回结构必须符合预期 assert isinstance(data, dict), funexpected response type: {type(data)} assert orders in data, fresponse missing orders field: {json.dumps(data, ensure_asciiFalse)[:200]} assert isinstance(data[orders], list), forders must be list, got {type(data[orders])} return data[orders] def write_orders_to_db(orders: list) - int: 写入数据库使用 INSERT IGNORE 保证幂等配合唯一索引。 if not orders: logger.info(no orders to write, skip DB insert) return 0 conn mysql.connector.connect(**DB_CONFIG) cursor conn.cursor() written 0 try: sql INSERT IGNORE INTO t_order_sync (order_id, order_data, sync_time) VALUES (%s, %s, NOW()) rows [(o[order_id], json.dumps(o, ensure_asciiFalse)) for o in orders] cursor.executemany(sql, rows) conn.commit() written cursor.rowcount logger.info(DB write done, input%s, inserted%s, len(orders), written) except mysql.connector.Error as e: conn.rollback() logger.error(DB write failed, error_code%s, msg%s, orders%s, e.errno, e.msg, json.dumps([o.get(order_id) for o in orders], ensure_asciiFalse)[:500]) raise # 显式上抛由外层决定退出码 finally: cursor.close() conn.close() return written def main(): # 前置校验日志目录可写 if not sys.warnoptions: pass begin_time datetime.now().strftime(%Y-%m-%d %H:%M:%S) end_time begin_time # 简化演示实际操作中按业务窗口推进 logger.info(order sync task started, begin%s, end%s, begin_time, end_time) try: orders fetch_orders(begin_time, end_time) # 关键路径断言拉取到的订单数据必须符合预期 for o in orders[:5]: assert order_id in o, forder missing order_id field: {json.dumps(o, ensure_asciiFalse)} assert amount in o, forder missing amount field: {json.dumps(o, ensure_asciiFalse)} written write_orders_to_db(orders) logger.info(order sync finished, fetched%s, written%s, len(orders), written) except AssertionError as e: logger.error(assertion failed, detail%s, e) sys.exit(2) # 自定义退出码 2 表示数据校验失败 except requests.exceptions.HTTPError as e: logger.error(API HTTP error, status%s, url%s, e.response.status_code, e.request.url) sys.exit(3) except mysql.connector.Error as e: logger.error(DB error, errno%s, msg%s, e.errno, e.msg) sys.exit(4) except Exception as e: logger.error(unexpected exception, type%s, msg%s, type(e).__name__, e) sys.exit(1) if __name__ __main__: main()这个骨架至少有几点值得借鉴退出码有语义。1 表示未知异常2 表示断言失败数据不对3 表示 API 错误4 表示数据库错误。上层调度平台在感知失败时可以依赖退出码快速分类甚至可以针对不同退出码配置不同的告警级别。幂等写入兜底。INSERT IGNORE配合数据库里的唯一索引order_id 唯一即使重试导致重复执行也不会插入重复订单。重试只覆盖网络层。fetch_orders 上挂了 tenacity 装饰器只有超时和连接错误会触发重试数据库写入没有重试因为数据库故障通常不是几秒能恢复的而且executemany这类操作在大批次下重试成本很高。日志贯穿始终。每一步的关键节点都有日志每次重试失败也都会被记录配合 RUN_ID 可以在排查时把一次完整任务的所有日志一次性 grep 出来。7. 常见问题与排查技巧实录7.1 典型问题快览我整理了日常工作中最常见的几类脚本稳定性问题每一类都给出现象、原因层级和排查顺序方便你直接对照使用。这张表建议收藏踩坑的时候翻一翻比翻文档效率高得多。现象可能原因排查顺序脚本“成功退出”但数据没写入缺少关键路径断言异常被吞掉先查日志有没有 warning再查退出码是否为 0再核对数据源状态日志里有 ERROR 但找不到上下文日志缺少 run_id / 时间戳 / 关键变量检查日志格式配置确认是否有统一的 Formatter 和 ContextFilter重试了很多次但最终仍然失败重试的目标错误不可恢复重试间隔太短没有指数退避先看是哪类错误触发的重试4xx 还是 5xx再调整 retry 条件多次执行脚本产生了重复数据缺少幂等设计检查目标表是否有唯一索引写入语句是否用了 INSERT IGNORE / UPSERT日志文件无限增长磁盘被写满没有配置 log rotation检查 RotatingFileHandler 或 logrotate 配置设置大小/周期上限脚本在凌晨跑了但没人知道成功了没缺少“成功/失败通知”机制在脚本尾部增加“成功上报”逻辑或在入口加 try-except 统一发送失败告警环境变量缺失导致脚本跑到一半才报错缺少前置校验在 main 入口处统一检查依赖命令、环境变量、配置文件、目录权限7.2 一个真实的排查过程说得再具体一点。有一次线上一个广告数据回传脚本频繁失败日志显示“API request failed”。我第一反应是看重试日志如果有 warning 级联的“[retry]”输出说明是临时性故障导致重试且重试次数不够如果直接 ERROR说明第一次请求就报不可恢复的错误。结果日志里只有一条孤零零的 ERROR没有任何重试记录说明异常类型没有被 retry 条件捕获。查代码发现这个脚本用的旧版 HTTP 库网络异常的类型是URLError而 retry 里只配了requests.exceptions.Timeout和ConnectionError两者对不上所以重试机制完全没生效。对症下药把异常类型修正并重新部署问题立刻消失。这个案例的典型性在于重试机制写好了不等于生效了一定要用日志验证“异常路径真的会触发重试”。我现在的习惯是每个带重试的函数在写完之后都会手动模拟一次失败比如把 URL 指向一个本地不存在的端口然后确认日志里出现了预期次数的 warning 和退避记录再算这个功能真的完成的。7.3 实时触达让日志变成告警最后讲一个能大幅提升脚本稳定性的技巧把日志和即时通讯通知绑定起来。脚本稳定性的最终目标是“无人值守”但很多企业还达不到完全无人值守的程度所以至少要让异常日志在发生时立刻触达责任人而不是等人主动去翻日志。实现方式不复杂在脚本的顶层异常处理分支里调用一个 Webhook 地址钉钉、企业微信、飞书都支持把异常摘要发送到群里。示例如下import requests def notify_ops(title: str, content: str): webhook_url https://oapi.dingtalk.com/robot/send?access_tokenYOUR_TOKEN payload { msgtype: text, text: {content: f【脚本告警】{title}\n{content}} } try: requests.post(webhook_url, jsonpayload, timeout5) except Exception: logger.exception(failed to send ops notification)只需在except分支里加一行notify_ops(order_sync 同步失败, e)凌晨三点的故障就可以从“第二天早上发现”变成“三秒内群里弹出告警”。这也是我在团队里推行的最低投资高回报方案配合集中日志系统一起用稳定性治理才算真正成型。8. 从本地脚本到企业级工具链到这四个核心组件已经全部讲完了。最后再往外拉一个视角聊聊“本地脚本”和“企业级”之间的差距到底在哪。本地脚本不考虑这么多因为运行环境是你自己的数据规模是你自己的出错了你就在旁边随时能按 CtrlC。但企业级脚本的运行环境是无人值守的数据规模可能比本地大两个数量级出错的时候人大概率不在旁边。这就意味着你必须在“人不在场”的情况下让脚本能够自己发现问题、自己尝试修复问题、记录完整现场、并在无力回天时以可识别的方式宣告失败。断言、日志、异常、重试对应的正是这四件事。所以“企业级优化”这四个字从来不是指用了多贵的框架、多大的集群而是指脚本在被“放养”的情况下依然能够自我约束、自我留痕、自我修复、自我宣告。你可以从今天开始在下一个提交里先加一个断言、补一条结构化日志、把裸except改成带日志的异常分支、给最容易抖动的请求加上带退避的重试。四步走完你的脚本就已经比市面上大多数“能跑”的脚本稳健一大截了。回头说说我自己的习惯。我现在写任何脚本不管多小哪怕只是一个几十行的数据清洗脚本也会强制带上四件套入口处打一条 INFO 日志做什么、参数是什么、关键路径加断言拿到数据先校验结构、所有 except 都记日志绝不 pass、有外部 I/O 就一定设超时和重试。这个习惯帮我省了无数的排障时间也让很多脚本在转交给同事之后依然能够平稳运行很久。这套方法论不是银弹但它是目前我找到的最稳的底线——先把下限托住再谈优化上限。