# 环境巡检配置
services:
-name:"测试环境主站
    type: http
    url: "https://test.example.com/health"
timeout:10
expected_status:200

-name:"用户服务接口"
type:http
url:"https://test.example.com/api/user/health"
timeout:8

-name:"订单服务接口"
type:http
url:"https://test.example.com/api/order/health"
timeout:8

-name:"测试环境 MySQL"
type:mysql
host:"192.168.1.100"
port:3306
user:"test_user"
password:"test_password"
database:"test_db"
timeout:10

# 告警配置（可选，不填则不发通知）
alert:
enabled:true
email:
smtp_server:"smtp.qq.com"
smtp_port:587
sender:"your_email@qq.com"
password:"your_smtp授权码"
receivers:
-"team_lead@company.com"
-"qa_team@company.com"

# -*- coding: utf-8 -*-
"""
环境巡检脚本
每天上班前自动检查测试环境各服务是否正常，发现问题立即告警
依赖：requests, pyyaml（pip install requests pyyaml）
     数据库检查还需：pymysql 或 psycopg2-binary
"""

import os
import sys
import json
import time
import smtplib
import datetime
import sqlite3
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header

import requests
import yaml


# ─────────────────────────────────────────────
# 第一部分：各类服务的健康检查函数
# ─────────────────────────────────────────────

defcheck_http_service(name, url, timeout=10, expected_status=None):
"""
    检查 HTTP 服务是否可用
    :param name: 服务名称
    :param url: 健康检查接口 URL
    :param timeout: 超时时间（秒）
    :param expected_status: 期望的 HTTP 状态码，不填则只要能连通即可
    :return: dict，包含 status(ok/fail/error)、message、response_time(ms)
    """
    start = time.time()
try:
        response = requests.get(url, timeout=timeout, verify=False)
        elapsed_ms = round((time.time() - start) * 1000, 2)

if expected_status:
if response.status_code == expected_status:
return {
"status": "ok",
"message": f"HTTP {response.status_code}，响应时间 {elapsed_ms}ms",
"response_time": elapsed_ms
                }
else:
return {
"status": "fail",
"message": f"期望状态码 {expected_status}，实际 {response.status_code}",
"response_time": elapsed_ms
                }
else:
# 无法连通时 requests 会抛异常，能到这里说明通了
return {
"status": "ok",
"message": f"HTTP {response.status_code}，响应时间 {elapsed_ms}ms",
"response_time": elapsed_ms
            }
except requests.exceptions.Timeout:
return {
"status": "fail",
"message": f"连接超时（{timeout}s）",
"response_time": None
        }
except requests.exceptions.ConnectionError:
return {
"status": "error",
"message": "连接失败（服务不可达）",
"response_time": None
        }
except Exception as e:
return {
"status": "error",
"message": f"检查异常：{str(e)}",
"response_time": None
        }


defcheck_mysql_service(name, host, port, user, password, database, timeout=10):
"""
    检查 MySQL 数据库连通性
    依赖：pip install pymysql
    """
try:
import pymysql
        start = time.time()
        conn = pymysql.connect(
            host=host,
            port=port,
            user=user,
            password=password,
            database=database,
            connect_timeout=timeout
        )
        elapsed_ms = round((time.time() - start) * 1000, 2)
        conn.close()
return {
"status": "ok",
"message": f"MySQL 连通正常，耗时 {elapsed_ms}ms",
"response_time": elapsed_ms
        }
except ImportError:
return {
"status": "error",
"message": "缺少 pymysql 库，请运行: pip install pymysql",
"response_time": None
        }
except Exception as e:
return {
"status": "error",
"message": f"MySQL 连接失败：{str(e)}",
"response_time": None
        }


defcheck_redis_service(name, host, port, password=None, timeout=5):
"""
    检查 Redis 连通性
    依赖：pip install redis
    """
try:
import redis
        r = redis.Redis(host=host, port=port, password=password,
                        socket_timeout=timeout, decode_responses=True)
        r.ping()
return {"status": "ok", "message": "Redis 连通正常", "response_time": None}
except ImportError:
return {
"status": "error",
"message": "缺少 redis 库，请运行: pip install redis",
"response_time": None
        }
except Exception as e:
return {
"status": "error",
"message": f"Redis 连接失败：{str(e)}",
"response_time": None
        }


# ─────────────────────────────────────────────
# 第二部分：告警通知
# ─────────────────────────────────────────────

defsend_email_alert(subject, body, config):
"""发送邮件告警"""
try:
        msg = MIMEMultipart()
        msg['From'] = Header(config['sender'])
        msg['To'] = ','.join(config['receivers'])
        msg['Subject'] = Header(subject, 'utf-8')
        msg.attach(MIMEText(body, 'html', 'utf-8'))

        server = smtplib.SMTP(config['smtp_server'], config['smtp_port'])
        server.starttls()
        server.login(config['sender'], config['password'])
        server.sendmail(config['sender'], config['receivers'], msg.as_string())
        server.quit()
print(f"邮件已发送至: {', '.join(config['receivers'])}")
except Exception as e:
print(f"邮件发送失败: {e}")


defbuild_alert_content(results, check_time):
"""构建告警邮件正文（HTML格式）"""
    failed = [r for r in results if r['result']['status'] != 'ok']
    passed = [r for r in results if r['result']['status'] == 'ok']

    html = f"""
    <html>
    <body>
    <h2>🛠️ 测试环境巡检报告</h2>
    <p><b>检查时间:</b> {check_time}</p>
    <p><b>汇总:</b> 正常 {len(passed)} 项 | 异常 {len(failed)} 项</p>

    <h3>异常项目</h3>
    <table border='1' cellpadding='8' cellspacing='0' style='border-collapse:collapse'>
    <tr style='background:#ffcccc'><th>服务</th><th>类型</th><th>状态</th><th>详情</th></tr>
    """
for r in failed:
        html += f"<tr><td>{r['name']}</td><td>{r['type']}</td>"
        html += f"<td style='color:red'>{r['result']['status']}</td>"
        html += f"<td>{r['result']['message']}</td></tr>"

    html += "</table>"

if passed:
        html += "<h3>正常项目</h3><table border='1' cellpadding='8' cellspacing='0'><tr style='background:#ccffcc'>"
        html += "<th>服务</th><th>类型</th><th>状态</th><th>详情</th></tr>"
for r in passed:
            html += f"<tr><td>{r['name']}</td><td>{r['type']}</td>"
            html += f"<td style='color:green'>{r['result']['status']}</td>"
            html += f"<td>{r['result']['message']}</td></tr>"
        html += "</table>"

    html += "<p><i>本报告由 Python 环境巡检脚本自动生成</i></p></body></html>"
return html


# ─────────────────────────────────────────────
# 第三部分：主执行逻辑
# ─────────────────────────────────────────────

defrun_health_check(config_path='config.yaml'):
"""加载配置并执行所有健康检查"""
# 加载配置
withopen(config_path, 'r', encoding='utf-8') as f:
        config = yaml.safe_load(f)

    services = config.get('services', [])
    alert_config = config.get('alert', {})
    check_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')

print(f"\n{'='*60}")
print(f"开始巡检 | {check_time} | 共 {len(services)} 项")
print(f"{'='*60}\n")

    results = []

for svc in services:
        name = svc['name']
        svc_type = svc['type']

print(f"检查: {name} ...", end=' ', flush=True)

if svc_type == 'http':
            result = check_http_service(
                name=name,
                url=svc['url'],
                timeout=svc.get('timeout', 10),
                expected_status=svc.get('expected_status')
            )
elif svc_type == 'mysql':
            result = check_mysql_service(
                name=name,
                host=svc['host'],
                port=svc.get('port', 3306),
                user=svc['user'],
                password=svc['password'],
                database=svc.get('database', ''),
                timeout=svc.get('timeout', 10)
            )
elif svc_type == 'redis':
            result = check_redis_service(
                name=name,
                host=svc['host'],
                port=svc.get('port', 6379),
                password=svc.get('password')
            )
else:
            result = {'status': 'error', 'message': f'未知类型: {svc_type}', 'response_time': None}

# 输出结果
        status_icon = {'ok': '✅', 'fail': '⚠️', 'error': '❌'}.get(result['status'], '❓')
print(f"{status_icon}{result['message']}")

        results.append({
'name': name,
'type': svc_type,
'result': result,
'check_time': check_time
        })

# ── 汇总报告 ──
print(f"\n{'='*60}")
print(f"巡检汇总报告")
print(f"{'='*60}")

    ok_count = sum(1for r in results if r['result']['status'] == 'ok')
    fail_count = sum(1for r in results if r['result']['status'] == 'fail')
    error_count = sum(1for r in results if r['result']['status'] == 'error')

print(f" 正常: {ok_count}")
print(f" 异常: {fail_count}")
print(f" 错误: {error_count}")
print(f"  总计:   {len(results)}")
print(f"{'='*60}\n")

# ── 发送告警 ──
if alert_config.get('enabled') and (fail_count > 0or error_count > 0):
print("正在发送告警通知...")
        html_body = build_alert_content(results, check_time)
        subject = f"测试环境巡检异常 | {fail_count + error_count} 项出问题"
        send_email_alert(subject, html_body, alert_config['email'])

return results


# ─────────────────────────────────────────────
# 第四部分：程序入口 + Windows 任务计划设置说明
# ─────────────────────────────────────────────

if __name__ == '__main__':
import argparse

    parser = argparse.ArgumentParser(description='测试环境健康检查脚本')
    parser.add_argument('--config', default='config.yaml', help='配置文件路径')
    parser.add_argument('--output', default='report.json', help='JSON结果输出路径')
    args = parser.parse_args()

# 忽略 SSL 警告（测试环境常见）
import urllib3
    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

    results = run_health_check(args.config)

# 写 JSON 结果文件
withopen(args.output, 'w', encoding='utf-8') as f:
        json.dump(results, f, ensure_ascii=False, indent=2)
print(f"📄 结果已保存至: {args.output}")

# 以退出码告知是否全通过（方便 CI/定时任务判断）
    has_issue = any(r['result']['status'] != 'ok'for r in results)
    sys.exit(1if has_issue else0)

Windows 定时任务设置（上班前自动跑）

不想手动运行？设置好定时任务，每天自动执行：

方法：任务计划程序

1. 创建基本任务：控制面板 → 管理工具 → 任务计划程序 → 创建基本任务

2. 设置触发器：

• 触发时间：每天早上 8:30（上班前30分钟）
• 每天重复：勾选

3. 设置操作：

• 程序：python.exe 的完整路径（如 C:\Python312\python.exe）
• 参数：env_health_check.py --config config.yaml
• 起始位置：D:\tools\env_health_check\（脚本所在目录）

4. 完成，以后每天 8:30 脚本自动运行，有问题直接发邮件通知你。

扩展：接入钉钉/企业微信 Webhook

如果你们用钉钉或企业微信，把通知函数替换成 Webhook 方式更方便：

defsend_dingtalk_alert(message, webhook_url):
"""钉钉群机器人 Webhook 通知"""
    payload = {
"msgtype": "text",
"text": {"content": f"【环境巡检告警】\n{message}"}
    }
    resp = requests.post(webhook_url, json=payload)
return resp.json()

defsend_wxwork_alert(message, webhook_url):
"""企业微信群机器人 Webhook 通知"""
    payload = {
"msgtype": "markdown",
"markdown": {
"content": f"### 🛠️ 测试环境巡检告警\n>{message}"
        }
    }
    resp = requests.post(webhook_url, json=payload)
return resp.json()

使用建议

总结

这个脚本解决的核心问题就一个：把环境检查从"被动发现"变成"主动告知"。

成本极低——一个下午写好，配好定时任务，以后每天早上省下 15~30 分钟的"环境排查时间"。工具虽然简单，但实用价值很高，尤其适合测试环境不太稳定的团队。