函数封装与代码复用

当同一逻辑在脚本中出现两次以上，或单文件超过百行时，应提取为函数。函数让运维脚本更易测试、Review 和在不同项目间复用。

---

函数定义与返回值

def format_uptime(seconds: int) -> str:
    """将秒数格式化为运维可读字符串。"""
    days, rem = divmod(seconds, 86400)
    hours, rem = divmod(rem, 3600)
    minutes, _ = divmod(rem, 60)
    parts = []
    if days:
        parts.append(f"{days}d")
    if hours:
        parts.append(f"{hours}h")
    parts.append(f"{minutes}m")
    return " ".join(parts)


print(format_uptime(90061))  # 1d 1h 1m

类型注解（int → str）不强制运行时检查，但 IDE 与静态工具能提前发现错误。

---

默认参数与关键字参数

def ssh_command(
    host: str,
    cmd: str,
    user: str = "root",
    port: int = 22,
    timeout: int = 30,
) -> dict:
    """模拟 SSH 执行，返回结构化结果。"""
    # 实际项目会用 paramiko/subprocess，此处仅演示接口
    return {
        "host": host,
        "user": user,
        "port": port,
        "cmd": cmd,
        "exit_code": 0,
        "stdout": "ok",
        "timeout": timeout,
    }


# 位置参数 + 关键字参数
result = ssh_command("10.0.0.5", "uptime", user="deploy", timeout=60)
print(result["host"], result["stdout"])

警告

默认参数不要用可变对象（如 []、{}），否则多次调用会共享同一列表。需要时用 None 再在函数内创建。

def append_host(hosts=None):
    if hosts is None:
        hosts = []
    hosts.append("new")
    return hosts

---

文档字符串与 main 入口

def disk_alert(used_pct: float, threshold: float = 85.0) -> str | None:
    """
    磁盘使用率告警。

    Args:
        used_pct: 当前使用率 0-100
        threshold: 告警阈值，默认 85

    Returns:
        告警文案；未超阈值返回 None
    """
    if used_pct >= threshold:
        return f"磁盘使用率 {used_pct:.1f}% >= {threshold}%"
    return None


def main() -> int:
    checks = [("web-01", 72.0), ("db-01", 91.5)]
    exit_code = 0
    for host, pct in checks:
        msg = disk_alert(pct)
        if msg:
            print(f"[ALERT] {host}: {msg}")
            exit_code = 1
        else:
            print(f"[OK] {host}: {pct}%")
    return exit_code


if __name__ == "__main__":
    raise SystemExit(main())

用 return 退出码（0 成功、非 0 失败）便于 CI/CD 与 cron 捕获脚本结果。

---

实战：批量巡检骨架

def check_host(host: str, metrics: dict) -> list[str]:
    """单主机检查，返回告警消息列表。"""
    issues = []
    if metrics.get("cpu", 0) >= 85:
        issues.append(f"CPU={metrics['cpu']}%")
    if metrics.get("disk", 0) >= 90:
        issues.append(f"DISK={metrics['disk']}%")
    return issues


def run_inspection(inventory: dict[str, dict]) -> int:
    """inventory: 主机名 -> 指标 dict。返回异常主机数。"""
    bad = 0
    for name, metrics in inventory.items():
        issues = check_host(name, metrics)
        if issues:
            bad += 1
            print(f"FAIL {name}: {', '.join(issues)}")
        else:
            print(f"OK   {name}")
    return bad


if __name__ == "__main__":
    data = {
        "web-01": {"cpu": 40, "disk": 50},
        "web-02": {"cpu": 88, "disk": 55},
    }
    failed = run_inspection(data)
    raise SystemExit(1 if failed else 0)

---

小结

实践	说明
单一职责	一个函数只做一件事（检查、格式化、发送）
类型注解	参数与返回值写清楚，降低协作成本
docstring	说明参数含义与返回值
main() + 退出码	脚本可被调度系统正确判定成败

第一阶段完结。进入 第二阶段：数据处理，从字符串与日志解析开始。