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_code=500, body=<html><body>Internal 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, status=502, resp_body={"code":502,"msg":"bad gateway"}，差距有多大，不用我多说。
Python 里用 logging 模块的话，我推荐直接用logging.basicConfig配上自定义 Formatter，把上面四件套固化下来：
import logging logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | run_id=%(run_id)s | %(message)s", handlers=[ logging.StreamHandler(), logging.FileHandler("app.log", encoding="utf-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", maxBytes=10 * 1024 * 1024, # 单个日志文件 10MB 封顶 backupCount=5 # 最多保留 5 个备份文件 )Shell 脚本里处理起来更直接，用 logrotate 这个系统工具，在/etc/logrotate.d/下写一个配置：
/path/to/your/script.log { daily rotate 7 compress missingok notifempty copytruncate }这里的几个参数解释一下：daily表示每天轮转一次，rotate 7保留最近 7 个归档，compress归档时压缩成 gzip，copytruncate是很多脚本类日志必须加的一个参数——它先把原文件拷贝走再把原文件截断，这样即使脚本一直持有文件句柄也不会写乱。不加copytruncate的话，进程不重启日志就可能写丢了。
3.4 从单机日志到集中日志
说完本地日志，顺带提一下日志的后端演进。热搜词里有loki日志系统和windows安全日志，这正好对应了两个方向：生产环境服务的日志通常需要集中收集，而系统安全日志则需要审计留存。
如果你的脚本是跑在多台服务器上的，那么单机日志的排查效率会断崖式下降。这时候就要引入集中日志系统。这个领域的常见方案是 Loki 配合 Promtail，也可以是 ELK（Elasticsearch + Logstash + Kibana），甚至简单一点直接通过 rsyslog 把日志转发到一台中心服务器也行。选型的核心判断标准只有一个：你能不能在 5 分钟内根据一个任务 ID 把所有机器上的相关日志全部捞出来。Loki 这类方案比 ELK 轻量很多，索引成本低，特别适合“脚本日志”这种流量中等、结构简单的场景。
我记得最早带团队的时候，排查跨服务器的数据不一致问题，经常要 ssh 登录三四台机器，挨个 grep 日志文件，遇到日志过大还得配合tail、awk慢慢翻，折腾一两个小时是常有的事。后来统一接入了集中日志，同样的排查，一条 LogQL 查询语句{app="sync_script"} |= "order_id=A10086"，几秒钟直接出来所有关联日志。这个效率提升是几何级别的。
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, timeout=5, retries=3): for attempt in range(1, retries + 1): try: resp = requests.get(url, timeout=timeout) 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(f"fetch {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(f"fetch {url} failed: http_status={status_code}") from e # 理论上走不到这里，但加个兜底 raise RuntimeError(f"fetch {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 2>&1; 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( stop=stop_after_attempt(5), # 最多尝试 5 次 wait=wait_exponential(multiplier=1, min=1, max=16), # 指数退避 1s, 2s, 4s, 8s, 16s retry=retry_if_exception_type(( requests.exceptions.Timeout, requests.exceptions.ConnectionError, )), before_sleep=before_sleep_log(logger, logging.WARNING), reraise=True, ) def fetch_data(url): resp = requests.get(url, timeout=5) resp.raise_for_status() return resp.json()这个写法的好处是，重试策略和业务逻辑彻底分离，装饰器一加，函数就自动拥有了“临时性失败自动重试 + 最终失败向上抛”的能力。before_sleep会在每次准备重试前打一条 warning 日志，这样你事后看日志能清楚地知道“这个操作失败了三次、分别在哪些时间点重试的”。reraise=True表示重试耗尽后抛出最后一次异常，保留原始异常链。
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", maxBytes=10*1024*1024, backupCount=5, encoding="utf-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( stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=1, max=16), retry=retry_if_exception_type(( requests.exceptions.Timeout, requests.exceptions.ConnectionError, )), before_sleep=before_sleep_log(logger, logging.WARNING), reraise=True, ) 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": f"Bearer {API_TOKEN}"}, timeout=(3, 10), # 连接超时 3s，读超时 10s ) resp.raise_for_status() data = resp.json() # 业务级断言：接口返回结构必须符合预期 assert isinstance(data, dict), f"unexpected response type: {type(data)}" assert "orders" in data, f"response missing 'orders' field: {json.dumps(data, ensure_ascii=False)[:200]}" assert isinstance(data["orders"], list), f"'orders' 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_ascii=False)) 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_ascii=False)[: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, f"order missing order_id field: {json.dumps(o, ensure_ascii=False)}" assert "amount" in o, f"order missing amount field: {json.dumps(o, ensure_ascii=False)}" 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_token=YOUR_TOKEN" payload = { "msgtype": "text", "text": {"content": f"【脚本告警】{title}\n{content}"} } try: requests.post(webhook_url, json=payload, timeout=5) except Exception: logger.exception("failed to send ops notification")只需在except分支里加一行notify_ops("order_sync 同步失败", e)，凌晨三点的故障就可以从“第二天早上发现”变成“三秒内群里弹出告警”。这也是我在团队里推行的最低投资高回报方案，配合集中日志系统一起用，稳定性治理才算真正成型。
8. 从本地脚本到企业级工具链
到这，四个核心组件已经全部讲完了。最后再往外拉一个视角，聊聊“本地脚本”和“企业级”之间的差距到底在哪。
本地脚本不考虑这么多，因为运行环境是你自己的，数据规模是你自己的，出错了你就在旁边，随时能按 Ctrl+C。但企业级脚本的运行环境是无人值守的，数据规模可能比本地大两个数量级，出错的时候人大概率不在旁边。这就意味着：你必须在“人不在场”的情况下，让脚本能够自己发现问题、自己尝试修复问题、记录完整现场、并在无力回天时以可识别的方式宣告失败。断言、日志、异常、重试，对应的正是这四件事。
所以“企业级优化”这四个字，从来不是指用了多贵的框架、多大的集群，而是指脚本在被“放养”的情况下，依然能够自我约束、自我留痕、自我修复、自我宣告。你可以从今天开始，在下一个提交里，先加一个断言、补一条结构化日志、把裸except改成带日志的异常分支、给最容易抖动的请求加上带退避的重试。四步走完，你的脚本就已经比市面上大多数“能跑”的脚本稳健一大截了。
回头说说我自己的习惯。我现在写任何脚本，不管多小，哪怕只是一个几十行的数据清洗脚本，也会强制带上四件套：入口处打一条 INFO 日志（做什么、参数是什么）、关键路径加断言（拿到数据先校验结构）、所有 except 都记日志（绝不 pass）、有外部 I/O 就一定设超时和重试。这个习惯帮我省了无数的排障时间，也让很多脚本在转交给同事之后依然能够平稳运行很久。这套方法论不是银弹，但它是目前我找到的最稳的底线——先把下限托住，再谈优化上限。
