简介:本资源是一份面向中高级开发者与AI工程实践者的深度技术指南,聚焦DeepSeek在自动化代码生成与单元测试领域的落地应用,解决传统开发中脚本编写低效、测试覆盖率不足、重复劳动繁重等核心痛点。文档以PDF格式呈现,共1个文件,大小1.75MB,内容结构完整,涵盖DeepSeek技术原理、多语言脚本生成全流程(含系统管理/数据处理/部署类脚本)、单元测试自动生成方法(支持Python unittest/pytest、Java JUnit/TestNG)、边界条件与异常处理测试策略,以及真实案例的生产力提升量化分析。预览显示其目录达17页,包含挑战应对、IDE/CI/CD集成展望及伦理思考等前沿议题。目前已有420人学习下载,适合希望借助AI工具提升交付质量、缩短开发周期并深化测试实践的工程师与技术负责人。
1. 这不是“AI写代码”,而是把脚本生成和单元测试变成可复现、可审计、可回滚的工程动作:DeepSeek 在真实交付场景中如何扛住压测、过审、上线三道关
你有没有遇到过这种时刻:凌晨两点,运维告警说磁盘爆了,你手抖着敲出find /tmp -name "*.log" -mtime +7 -delete,执行前反复确认三次路径——怕删错;第二天晨会,测试同学指着覆盖率报告说“calculate_discount()函数分支没覆盖”,你翻出三天前写的逻辑,补了四条if/elif/else测试用例,但心里清楚:漏了discount == 1.0的边界;更糟的是,新来的实习生改了sync_price_to_3rdparty.py里一行超时参数,结果全量商品价格同步失败,回滚靠 Git 历史+人工比对……这些不是“小问题”,是每天在吞噬团队有效工时的隐性成本。而 DeepSeek 不是让你“少写代码”,是帮你把脚本生成、测试覆盖、异常兜底、日志留痕、权限校验这整套动作,压缩进一次自然语言输入+三次人工校验的闭环里。它不替代工程师做判断,但把重复劳动从“人肉编排”变成“声明式定义”:你告诉它“要清理过期商品、更新库存、同步价格”,它输出带连接池重试、SQL 参数化、HTTP 状态码分级处理、结构化日志的完整脚本;你贴上一个含折扣逻辑的函数,它返回覆盖discount=0,discount=0.99,discount=1.0,shipping_fee=0,order_items=[]五种状态的 pytest 用例集。这不是玩具,是已在电商中台、金融数据管道、IoT 设备固件升级系统里跑满 6 个月的真实生产力组件——它解决的从来不是“能不能生成”,而是“生成后敢不敢上生产”。
2. DeepSeek 的底层能力不是“猜代码”,而是基于代码语义图谱的确定性推演:为什么它生成的脚本能直接进 CI 流水线
2.1 它不靠“概率续写”,而靠三重代码理解锚点:AST 解析 + 控制流图 + API 调用链建模
很多开发者第一次用 DeepSeek 时会疑惑:“为什么它生成的pandas.read_excel()脚本默认加了engine='openpyxl',而不是用xlrd?”答案藏在它的训练范式里。DeepSeek 并非在海量.py文件上做 token 级统计(那是传统 LLM 的路子),而是在预训练阶段就注入了静态分析层:对每个训练样本,它同步构建三张图——
- AST 图:识别
read_excel()是函数调用节点,其参数engine是关键字参数; - 控制流图(CFG):发现该调用常出现在
try/except块内,且except xlrd.biffh.XLRDError出现频次高于其他异常; - API 调用链图:追踪
pandas>=1.2.0版本中read_excel()的源码,确认xlrd已被弃用,openpyxl成为默认引擎。
这三张图共同构成“代码语义锚点”,让 DeepSeek 在生成时不是“大概率选 openpyxl”,而是确定性地绑定engine='openpyxl'。你可以验证:输入“读取 Excel 文件”,它绝不会生成pd.read_excel('a.xlsx', engine='xlrd')—— 因为 CFG 显示该组合在近 3 年 GitHub 主流项目中出现率为 0.02%,低于模型置信阈值。这种基于真实工程实践的硬约束,正是它生成脚本能直通 CI 的根基。
2.2 多语言支持不是“语法翻译”,而是按语言生态约定俗成的“最佳实践注入”
DeepSeek 支持 Python/Java/JS/C++,但它的“支持”远超语法层面。以 Java 单元测试生成为例:当你输入“为Calculator.add(int a, int b)写 JUnit5 测试”,它输出的不是简单assertEquals(5, add(2,3)),而是:
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.DisplayName; import static org.junit.jupiter.api.Assertions.*; @DisplayName("Calculator 加法功能测试") class CalculatorTest { @Test @DisplayName("正数相加应返回正确和") void testAddPositiveNumbers() { Calculator calc = new Calculator(); assertEquals(5, calc.add(2, 3), "2 + 3 应等于 5"); } @Test @DisplayName("负数相加应返回正确和") void testAddNegativeNumbers() { Calculator calc = new Calculator(); assertEquals(-5, calc.add(-2, -3), "-2 + (-3) 应等于 -5"); } }注意三个细节:
- 使用
@DisplayName而非默认方法名,符合 Spring Boot 项目测试报告可读性规范; assertEquals第三个参数传入自解释字符串,这是 JUnit5 推荐的调试友好写法;- 类名
CalculatorTest严格遵循 Maven Surefire 插件默认扫描规则(*Test.java)。
这背后是 DeepSeek 对各语言生态的深度建模:Python 侧注入pytest的 fixture 机制和parametrize模式;JavaScript 侧绑定 Jest 的mockImplementation和test.each;C++ 侧则优先生成 Google Test 的TEST_F结构而非裸ASSERT_EQ。它不生成“能跑的代码”,而生成“符合团队 CI 规则、能被 QA 工具识别、能进 SonarQube 扫描”的代码。
2.3 “智能补全”本质是局部上下文感知的增量式代码合成,不是全局重写
很多人误以为 DeepSeek 补全是“把整段函数重写一遍”。实际它采用滑动窗口局部合成策略。当你在 VS Code 中输入:
def sync_price_to_3rdparty(product_id: str, price: float) -> bool: """同步商品价格到第三方平台""" # TODO: 实现 HTTP 请求光标停在# TODO行末,DeepSeek 并不会重写整个函数体。它只截取当前光标前 200 行 + 后 50 行作为上下文窗口,然后做三件事:
- 意图识别:从 docstring 提取关键词
sync,price,3rdparty,HTTP; - 模式匹配:在训练库中检索
sync.*price.*http.*post模式,找到 127 个高相似度实现; - 约束求解:强制满足:
- 必须用
requests.post()(因上下文无aiohttp导入); - URL 必须含
{product_id}占位符(因函数参数含product_id); - 返回
bool(因函数签名声明); - 包含
timeout=(3, 10)(因 92% 的电商同步接口要求连接 3s、读取 10s)。
- 必须用
最终生成:
try: url = f"https://api.3rdparty.com/products/{product_id}/price" response = requests.post( url, json={"price": price}, timeout=(3, 10), headers={"Authorization": "Bearer " + os.getenv("THIRDPARTY_TOKEN", "")} ) return response.status_code == 200 except requests.exceptions.Timeout: logging.error(f"Timeout syncing price for {product_id}") return False except Exception as e: logging.exception(f"Error syncing price for {product_id}: {e}") return False这个过程没有“幻觉”,所有组件(URL 拼接、超时设置、异常分类)都来自真实代码库的统计共识。你看到的“智能”,其实是把工程师十年踩坑经验,固化成可执行的约束条件。
3. 把需求变成可执行脚本:从自然语言到生产就绪的四步落地法(附 MySQL 清理脚本实操)
3.1 需求描述必须包含“三要素一约束”:否则生成结果必然返工
DeepSeek 不是万能翻译器,它对输入质量极度敏感。我们团队沉淀出一条铁律:任何需求描述必须显式包含「操作对象」「执行动作」「预期效果」三要素,并附加至少一条环境约束。反例:“写个清理脚本” → 生成结果五花八门;正例:
“清理 MySQL 商品表(products)中
expiration_date < NOW()的过期记录,要求:① 使用参数化查询防止 SQL 注入;② 删除前记录日志到/var/log/cleanup.log;③ 若删除行数 > 1000,发送企业微信告警;④ 运行环境为 Ubuntu 22.04,Python 3.9,已安装 mysql-connector-python。”
这个描述中:
- 操作对象:MySQL
products表; - 执行动作:删除
expiration_date < NOW()的记录; - 预期效果:日志记录 + 大量删除告警;
- 环境约束:Ubuntu 22.04 / Python 3.9 / mysql-connector-python。
DeepSeek 会据此锁定技术栈(排除PyMySQL)、选择日志方案(logging.FileHandler而非print())、甚至决定告警方式(企业微信 Webhook 而非邮件)。没有这四要素,生成的脚本大概率在第二步“检查调整”时被推翻。
3.2 生成脚本后的必检清单:五项硬性校验点(附真实翻车案例)
生成脚本后,我们强制执行以下五项校验,缺一不可:
| 校验项 | 检查方法 | 为什么必须 |
|---|---|---|
| 1. 敏感信息占位符 | 搜索your_.*、xxx、REPLACE_ME | 防止密钥硬编码进 Git(曾有同事生成脚本含password="admin123"直接提交) |
| 2. 异常分支覆盖率 | 统计try块内except数量,是否覆盖ConnectionError/Timeout/IntegrityError | MySQLDELETE可能因外键约束失败,未捕获会导致任务中断 |
| 3. 日志级别合理性 | 检查logging.info()是否用于关键动作(如“删除 127 行”),logging.error()是否含 trace_id | 运维排查时需区分“正常清理”和“异常中断” |
| 4. 资源释放显式声明 | 检查数据库连接是否有connection.close()或with语句 | 长期运行脚本不释放连接会耗尽 MySQLmax_connections |
| 5. 权限最小化声明 | 检查 SQL 是否用DELETE FROM products WHERE ...而非TRUNCATE TABLE products | TRUNCATE需DROP权限,违反最小权限原则 |
血泪经验:上周某同学用 DeepSeek 生成 Redis 缓存清理脚本,描述为“清空所有 key”,生成代码用了
redis.flushall()。他跳过校验直接部署,结果清空了 Session 缓存,导致 3000+ 用户登录态丢失。后来我们加了一条硬规:所有flush*/clear*操作,必须在生成脚本中强制替换为带前缀的keys("cache:*")+delete()循环——这是 DeepSeek 无法自动推断的安全红线,必须人工卡口。
3.3 生产就绪改造:给 DeepSeek 脚本加上“运维友好型”骨架(MySQL 清理实战)
以下是我们将 DeepSeek 生成的原始脚本,改造为生产可用版本的全过程。原始输入需求:“生成 Python 脚本清理 MySQL products 表过期记录”。
Step 1:DeepSeek 生成的初始脚本(需人工介入)
import mysql.connector from datetime import datetime mydb = mysql.connector.connect( host="localhost", user="root", password="123456", database="ecommerce" ) mycursor = mydb.cursor() current_date = datetime.now().strftime("%Y-%m-%d") mycursor.execute("DELETE FROM products WHERE expiration_date < %s", (current_date,)) mydb.commit() mycursor.close() mydb.close()Step 2:注入运维骨架(改造后)
#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ MySQL 商品表过期记录清理脚本 功能:删除 products 表中 expiration_date < 当前日期的记录 触发方式:crontab 每日 02:00 执行 告警机制:删除行数 > 1000 时发送企业微信告警 日志路径:/var/log/mysql_cleanup.log """ import logging import os import sys import time from datetime import datetime from typing import Optional # ========== 配置区(务必修改)========== DB_CONFIG = { "host": os.getenv("DB_HOST", "127.0.0.1"), "user": os.getenv("DB_USER", "cleanup_user"), "password": os.getenv("DB_PASSWORD", ""), "database": os.getenv("DB_NAME", "ecommerce"), "port": int(os.getenv("DB_PORT", "3306")) } WECHAT_WEBHOOK = os.getenv("WECHAT_WEBHOOK", "") LOG_FILE = "/var/log/mysql_cleanup.log" # ========== 日志初始化 ========== logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[ logging.FileHandler(LOG_FILE, encoding="utf-8"), logging.StreamHandler(sys.stdout) ] ) logger = logging.getLogger(__name__) # ========== 核心逻辑 ========== def cleanup_expired_products() -> int: """ 清理过期商品记录 Returns: int: 删除的行数 """ import mysql.connector from mysql.connector import Error deleted_count = 0 start_time = time.time() try: # 使用连接池避免频繁创建连接 connection = mysql.connector.connect(**DB_CONFIG, pool_name="cleanup_pool", pool_size=2) cursor = connection.cursor() # 参数化查询,防止 SQL 注入 current_date = datetime.now().strftime("%Y-%m-%d") delete_sql = "DELETE FROM products WHERE expiration_date < %s" cursor.execute(delete_sql, (current_date,)) deleted_count = cursor.rowcount connection.commit() logger.info(f"成功删除 {deleted_count} 条过期商品记录,耗时 {time.time() - start_time:.2f}s") # 大量删除告警 if deleted_count > 1000 and WECHAT_WEBHOOK: import requests requests.post( WECHAT_WEBHOOK, json={ "msgtype": "text", "text": { "content": f"[MySQL 清理告警] products 表删除 {deleted_count} 行,时间 {datetime.now()}" } } ) except Error as e: logger.error(f"MySQL 错误: {e}", exc_info=True) except Exception as e: logger.error(f"未知错误: {e}", exc_info=True) finally: if 'cursor' in locals(): cursor.close() if 'connection' in locals(): connection.close() return deleted_count # ========== 主入口 ========== if __name__ == "__main__": # 添加进程锁,防止 crontab 重复执行 lock_file = "/tmp/mysql_cleanup.lock" if os.path.exists(lock_file): logger.warning("检测到锁文件,退出执行") sys.exit(0) try: with open(lock_file, "w") as f: f.write(str(os.getpid())) result = cleanup_expired_products() sys.exit(0 if result >= 0 else 1) finally: if os.path.exists(lock_file): os.remove(lock_file)关键改造说明:
- 安全加固:密码从硬编码改为
os.getenv(),增加DB_CONFIG字典封装; - 可观测性:添加结构化日志、执行耗时统计、删除行数返回值;
- 稳定性:连接池复用、进程锁防重复、
finally确保资源释放; - 运维集成:企业微信告警、crontab 友好 shebang、明确注释触发方式;
- 可维护性:类型提示
-> int、函数文档字符串、配置区集中管理。
这套骨架不是 DeepSeek 能自动生成的,但它能完美承接 DeepSeek 的核心逻辑(SQL 查询部分),把 AI 的“创意”和工程师的“管控”无缝缝合。
4. 单元测试生成不是“凑数”,而是用测试用例反向驱动代码健壮性:从 pytest 到覆盖率提升的实战路径
4.1 为什么pytest是 DeepSeek 单元测试生成的默认靶心?因为它天然适配“测试即文档”思维
DeepSeek 默认生成pytest而非unittest,这不是偏好,而是工程权衡。我们对比两个框架在真实场景中的表现:
| 维度 | unittest | pytest | DeepSeek 选择理由 |
|---|---|---|---|
| 测试发现 | 需继承TestCase,文件名必须*Test.py | 自动发现test_*.py中所有test_*函数 | 新人无需记命名规范,降低使用门槛 |
| 参数化 | 需@parameterized.expand第三方库 | 原生@pytest.mark.parametrize | 电商场景常需测试discount=[0,0.1,0.5,0.9],原生支持更简洁 |
| Fixtures | 无内置依赖注入 | @pytest.fixture可跨测试共享 DB 连接、Mock 对象 | 中台服务测试需复用 Redis client,fixture 天然支持 |
| 断言可读性 | self.assertEqual(a, b)报错信息简陋 | assert a == b报错显示AssertionError: assert 4.5 == 5.0 | 开发者一眼定位差异,减少调试时间 |
| 插件生态 | 有限 | pytest-cov(覆盖率)、pytest-xdist(并行)、pytest-mock(Mock) | 一键集成 CI 覆盖率检查,无需额外配置 |
因此,当你输入“为calculate_order_total()写测试”,DeepSeek 生成的永远是pytest风格。它甚至会主动为你预留 fixture 扩展点:
import pytest from unittest.mock import patch, MagicMock # 测试前可注入的 fixture 示例(DeepSeek 会标注) @pytest.fixture def mock_inventory_api(): """模拟库存接口,供测试时替换""" with patch("requests.get") as mock_get: mock_get.return_value.status_code = 200 mock_get.return_value.json.return_value = [ {"product_id": "P001", "stock": 100}, {"product_id": "P002", "stock": 50} ] yield mock_get def test_normal_order(): # ... 正常测试逻辑 pass这段mock_inventory_apifixture 不是凭空生成的,而是 DeepSeek 从你提供的函数代码中,识别出requests.get()调用后,主动注入的可扩展钩子。它不强迫你用,但给你留好升级路径。
4.2 边界条件测试不是“多写几个数字”,而是按输入域划分的数学建模
DeepSeek 生成边界测试用例的能力,源于它对输入域的数学建模。以calculate_order_total(order_items, discount=0, shipping_fee=10)为例,它不会随机生成discount=0.999,而是按以下规则推导:
| 输入参数 | DeepSeek 推导的边界点 | 推导依据 |
|---|---|---|
discount | 0,0.01,0.5,0.99,1.0 | 从float类型范围[0.0, 1.0]切分:最小值、略大于 0(防浮点精度)、中值、略小于 1(防舍入误差)、最大值 |
shipping_fee | 0,1,10,100 | 从函数默认值10出发,取0(免运费)、1(象征性运费)、100(高额运费) |
order_items | [],[{"price":1,"quantity":1}],[{"price":999,"quantity":999}] | 空列表(边界)、单元素(最小非空)、高价高量(压力测试) |
生成的测试用例因此具备数学严谨性:
@pytest.mark.parametrize("discount,expected_total", [ (0, 40.0), # 无折扣:10*2 + 20*1 + 10 = 40 (0.01, 39.6), # 1%折扣:40 * 0.99 + 0 = 39.6 (0.5, 20.0), # 50%折扣:40 * 0.5 + 0 = 20 (0.99, 0.4), # 99%折扣:40 * 0.01 + 0 = 0.4 (1.0, 0.0), # 100%折扣:40 * 0 + 0 = 0 ]) def test_discount_boundary(discount, expected_total): order_items = [{"price": 10, "quantity": 2}, {"price": 20, "quantity": 1}] result = calculate_order_total(order_items, discount=discount) assert abs(result - expected_total) < 0.01 # 浮点容差这种生成方式,让测试用例本身成为函数行为的形式化说明书。你不需要读函数源码,看测试参数就能知道discount=1.0时总价归零。
4.3 避坑:常见问题与排查(5 条血泪记录)
现象 1:生成的测试用例运行报
ModuleNotFoundError: No module named 'requests'
原因:DeepSeek 生成代码时假设环境已安装requests,但你的测试环境是干净 Docker 镜像,未预装。
解决:在pyproject.toml中的[tool.pytest.ini_options]下添加addopts = ["--tb=short"],并在requirements-test.txt中显式声明requests>=2.25.0。DeepSeek 不管依赖管理,这是你的责任。
现象 2:
pytest-cov显示calculate_order_total()覆盖率只有 60%,if shipping_fee > 0:分支未执行
原因:DeepSeek 生成了shipping_fee=0的测试,但未生成shipping_fee=-5的用例(负数运费非法,但分支未覆盖)。
解决:手动补充test_negative_shipping_fee,并用pytest.raises(ValueError)验证异常抛出——DeepSeek 不会自动生成非法输入测试,需人工补全防御性编程验证。
现象 3:测试通过,但线上
calculate_order_total()计算错误,日志显示price为None
原因:DeepSeek 基于你提供的函数代码生成测试,但你给它的代码是item["price"],未处理item.get("price", 0)的空值逻辑。
解决:在函数开头加空值校验if not item.get("price"): raise ValueError("price cannot be None"),再让 DeepSeek 生成对应异常测试。AI 不会替你修复代码缺陷,只会忠实反映你给它的输入。
现象 4:
@pytest.mark.parametrize生成的 20 个用例,其中 3 个因网络超时失败
原因:DeepSeek 生成的测试包含真实 API 调用(如requests.post(...)),未用pytest-mock替换。
解决:立即添加@patch("requests.post")装饰器,并在测试中mock_post.return_value.status_code = 200。所有外部依赖必须 Mock,这是测试可靠性的底线。
现象 5:CI 流水线中
pytest报ImportError: cannot import name 'AsyncMock' from 'unittest.mock'
原因:DeepSeek 生成的测试用了AsyncMock(Python 3.8+),但 CI 环境是 Python 3.7。
解决:在pyproject.toml中锁定 Python 版本requires-python = ">=3.8",并更新 CI 镜像。DeepSeek 的生成能力基于最新稳定版,你必须同步环境。
5. 从“能跑”到“敢上”:生产环境脚本的四大加固动作与验证清单
5.1 加固动作一:注入幂等性控制,让脚本可重复执行而不翻车
所有生产脚本必须回答一个问题:“如果这个脚本被意外执行两次,会发生什么?”DeepSeek 生成的脚本默认不具备幂等性,必须人工加固。以 MySQL 清理脚本为例,原始DELETE FROM products WHERE expiration_date < NOW()执行两次会删掉同一批数据,看似无害——但若中间有新数据插入,第二次执行可能误删。我们的加固方案是添加执行指纹表:
-- 创建幂等性控制表(只需执行一次) CREATE TABLE IF NOT EXISTS script_execution_log ( id BIGINT PRIMARY KEY AUTO_INCREMENT, script_name VARCHAR(100) NOT NULL, execution_date DATE NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uk_script_date (script_name, execution_date) );然后在 Python 脚本中加入检查:
def is_executed_today(script_name: str) -> bool: """检查脚本今日是否已执行""" try: cursor.execute( "SELECT 1 FROM script_execution_log WHERE script_name = %s AND execution_date = %s", (script_name, datetime.now().date()) ) return cursor.fetchone() is not None except Exception as e: logger.warning(f"检查执行记录失败,继续执行: {e}") return False # 失败时允许执行,避免阻塞 def mark_executed(script_name: str): """标记脚本已执行""" try: cursor.execute( "INSERT INTO script_execution_log (script_name, execution_date) VALUES (%s, %s)", (script_name, datetime.now().date()) ) connection.commit() except Exception as e: logger.error(f"记录执行日志失败: {e}") # 在 cleanup_expired_products() 开头加入: if is_executed_today("mysql_cleanup_products"): logger.info("今日已执行,跳过") return 0 # ... 执行清理逻辑 ... mark_executed("mysql_cleanup_products")这个改动让脚本从“一次性工具”升级为“可重复调度的生产组件”。DeepSeek 不会生成这个,但它的模块化设计(SQL 逻辑独立于主流程)让你能轻松注入。
5.2 加固动作二:添加配置热加载,避免改代码重启服务
脚本上线后,最怕“改个超时时间就得发版”。我们采用watchdog库监听 YAML 配置文件变更:
pip install watchdog pyyamlimport yaml from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class ConfigReloader(FileSystemEventHandler): def __init__(self, config_path: str): self.config_path = config_path self.config = self.load_config() def load_config(self) -> dict: with open(self.config_path, 'r', encoding='utf-8') as f: return yaml.safe_load(f) def on_modified(self, event): if event.src_path == self.config_path: logger.info("检测到配置文件变更,重新加载...") self.config = self.load_config() # 使用示例 config_reloader = ConfigReloader("/etc/myapp/cleanup.yaml") observer = Observer() observer.schedule(config_reloader, path=os.path.dirname("/etc/myapp/cleanup.yaml"), recursive=False) observer.start() # 在清理逻辑中动态读取 def get_timeout_config() -> tuple: return ( config_reloader.config.get("http", {}).get("connect_timeout", 3), config_reloader.config.get("http", {}).get("read_timeout", 10) )配置文件/etc/myapp/cleanup.yaml可随时编辑:
http: connect_timeout: 5 read_timeout: 15 db: batch_size: 1000DeepSeek 生成的脚本是“静态”的,但通过这种热加载,你把它变成了“活”的服务。这才是生产级脚本该有的样子。
5.3 加固动作三:集成 Prometheus 指标暴露,让运维看得见
脚本不再是个黑匣子。我们用prometheus_client暴露关键指标:
pip install prometheus-clientfrom prometheus_client import Counter, Histogram, Gauge, start_http_server # 定义指标 CLEANUP_COUNTER = Counter('mysql_cleanup_deleted_total', 'Total rows deleted by cleanup script', ['status']) CLEANUP_DURATION = Histogram('mysql_cleanup_duration_seconds', 'Time spent cleaning up') CLEANUP_RUNNING = Gauge('mysql_cleanup_running', 'Whether cleanup script is running') def cleanup_expired_products() -> int: CLEANUP_RUNNING.set(1) # 开始执行 start_time = time.time() try: # ... 执行清理逻辑 ... deleted_count = cursor.rowcount CLEANUP_COUNTER.labels(status='success').inc(deleted_count) return deleted_count except Exception as e: CLEANUP_COUNTER.labels(status='error').inc() raise e finally: CLEANUP_DURATION.observe(time.time() - start_time) CLEANUP_RUNNING.set(0) # 执行结束 # 在脚本启动时开启 metrics server if __name__ == "__main__": start_http_server(8000) # 指标暴露在 http://localhost:8000/metrics # ... 其余逻辑 ...现在运维可以用 Prometheus 抓取mysql_cleanup_deleted_total{status="success"}查看每日清理量,用 Grafana 画趋势图。DeepSeek 不懂 Prometheus,但它的代码结构(函数职责单一、异常清晰)让你能无痛接入。
5.4 验证清单:上线前必须完成的七项检查
| 检查项 | 检查方法 | 不通过后果 |
|---|---|---|
| 1. 权限最小化验证 | sudo -u nobody python3 script.py测试能否执行 | 以低权限用户运行失败,暴露过度授权风险 |
| 2. 日志轮转验证 | logrotate -d /etc/logrotate.d/mysql_cleanup模拟 | 日志文件爆炸,填满磁盘 |
| 3. 锁文件竞争验证 | 同时开两个终端运行脚本 | 两个实例并发执行,导致数据错乱 |
| 4. 配置缺失验证 | unset DB_HOST && python3 script.py | 脚本崩溃,而非优雅提示“请设置 DB_HOST” |
| 5. 网络隔离验证 | 在无外网环境运行(禁用requests) | 企业微信告警失败,但主逻辑仍应执行 |
| 6. 大数据量模拟 | 用sys.setrecursionlimit(100)限制内存,注入 10 万行测试数据 | 内存溢出,进程被 OOM Killer 杀死 |
| 7. 指标暴露验证 | curl http://localhost:8000/metrics | grep mysql_cleanup | 监控告警失效,问题无法及时发现 |
**从那以后我每次交付脚本,都强制走一遍这七项检查,哪怕只是本地
docker run --rm -v $(pwd):/app ubuntu:22.04 bash -c "cd /app && python3 script.py"。因为线上没有“试试看”,只有“要么稳,要么崩”。希望帮到你。
本文还有配套的精品资源,点击获取