liuziyi

一、Python腳本（完整代碼）

import subprocess
import os
import socket
import re
import glob
import cx_Oracle
import argparse
from datetime import datetime

def parse_args():
    """解析命令行參數(shù)，支持靈活配置巡檢參數(shù)"""
    parser = argparse.ArgumentParser(description="Oracle數(shù)據(jù)庫(kù)與服務(wù)器巡檢腳本（Python版）")
    parser.add_argument("--db-host", required=True, help="數(shù)據(jù)庫(kù)主機(jī)地址（如127.0.0.1）")
    parser.add_argument("--db-port", default=1521, type=int, help="數(shù)據(jù)庫(kù)端口（默認(rèn)1521）")
    parser.add_argument("--db-sid", required=True, help="Oracle SID（如orcl11g）")
    parser.add_argument("--db-user", default="sys", help="數(shù)據(jù)庫(kù)用戶名（默認(rèn)sys）")
    parser.add_argument("--db-pwd", required=True, help="數(shù)據(jù)庫(kù)密碼")
    parser.add_argument("--html-output", default="./oracle_Healthcheck_{}.html", help="HTML報(bào)告輸出路徑（默認(rèn)當(dāng)前目錄）")
    return parser.parse_args()

def get_server_info():
    """獲取服務(wù)器層面信息：主機(jī)名、OS版本、存儲(chǔ)、內(nèi)存、CPU"""
    server_info = {}

    # 1. 主機(jī)名
    server_info["hostname"] = socket.gethostname()

    # 2. 操作系統(tǒng)版本（讀取/etc/os-release）
    try:
        with open("/etc/os-release", "r", encoding="utf-8") as f:
            for line in f:
                if line.startswith("PRETTY_NAME="):
                    server_info["os_version"] = line.strip().split('"')[1]
                    break
    except Exception as e:
        server_info["os_version"] = f"獲取失敗：{str(e)}"

    # 3. 根目錄存儲(chǔ)使用情況（df -h /）
    try:
        df_result = subprocess.check_output(["df", "-h", "/"], stderr=subprocess.STDOUT, text=True)
        # 提取關(guān)鍵行（排除標(biāo)題行）
        df_line = [line for line in df_result.splitlines() if line.endswith("/")][0]
        df_parts = df_line.split()
        server_info["storage"] = f"Disk Usage: {df_parts[2]}/{df_parts[1]} ({df_parts[4]})"
    except Exception as e:
        server_info["storage"] = f"獲取失敗：{str(e)}"

    # 4. 內(nèi)存使用情況（free -h）
    try:
        free_result = subprocess.check_output(["free", "-h"], stderr=subprocess.STDOUT, text=True)
        free_line = free_result.splitlines()[1]  # 第二行為內(nèi)存詳情
        free_parts = free_line.split()
        total_mem = free_parts[1]
        used_mem = free_parts[2]
        # 修正原bash腳本的計(jì)算錯(cuò)誤（原$3*0.1/$2邏輯錯(cuò)誤，改為實(shí)際使用率）
        used_percent = (float(free_parts[2].replace("G", "")) / float(total_mem.replace("G", ""))) * 100
        server_info["memory"] = f"Total Memory: {total_mem}, Used Memory: {used_mem}, Memory Usage: {used_percent:.2f}%"
    except Exception as e:
        server_info["memory"] = f"獲取失敗：{str(e)}"

    # 5. CPU使用率（top -bn1 提取用戶+系統(tǒng)CPU）
    try:
        top_result = subprocess.check_output(["top", "-bn1"], stderr=subprocess.STDOUT, text=True)
        cpu_line = [line for line in top_result.splitlines() if line.startswith("Cpu(s)")][0]
        cpu_parts = re.findall(r"\d+\.\d+", cpu_line)
        user_cpu = float(cpu_parts[0])
        sys_cpu = float(cpu_parts[2])
        server_info["cpu"] = f"{user_cpu + sys_cpu:.1f}%"
    except Exception as e:
        server_info["cpu"] = f"獲取失敗：{str(e)}"

    return server_info

def get_db_info(db_host, db_port, db_sid, db_user, db_pwd):
    """獲取數(shù)據(jù)庫(kù)層面信息，返回字典格式"""
    db_info = {}
    dsn = cx_Oracle.makedsn(db_host, db_port, sid=db_sid)
    conn = None

    try:
        # 以SYSDBA權(quán)限連接數(shù)據(jù)庫(kù)
        conn = cx_Oracle.connect(user=db_user, password=db_pwd, dsn=dsn, mode=cx_Oracle.SYSDBA)
        cursor = conn.cursor()

        # 1. 數(shù)據(jù)庫(kù)異常日志（最后200行含ERROR的內(nèi)容）
        log_path = f"/u01/app/oracle/diag/rdbms/{db_sid}/{db_sid}/trace/*.log"
        error_logs = []
        for log_file in glob.glob(log_path):
            try:
                # 讀取文件最后200行
                with open(log_file, "r", encoding="utf-8", errors="ignore") as f:
                    lines = f.readlines()[-200:]
                    # 篩選含ERROR/WARNING的行
                    for line in lines:
                        if re.search(r"error|warning", line, re.IGNORECASE):
                            error_logs.append(f"[{os.path.basename(log_file)}] {line.strip()}")
            except Exception as e:
                error_logs.append(f"讀取{os.path.basename(log_file)}失敗：{str(e)}")
        db_info["error_logs"] = "\n".join(error_logs) if error_logs else "無(wú)報(bào)錯(cuò)信息"

        # 2. 數(shù)據(jù)庫(kù)備份情況（近1天的備份）
        backup_sql = """
            SELECT fname backup_file_name, status, device_type, completion_time backup_finish_time
            FROM v$backup_files
            WHERE file_type = 'PIECE' AND bs_completion_time > SYSDATE - 1
        """
        cursor.execute(backup_sql)
        backup_rows = cursor.fetchall()
        backup_header = "BACKUP_FILE_NAME | STATUS | DEVICE_TYPE | BACKUP_FINISH_TIME\n"
        backup_header += "-" * 80 + "\n"
        backup_content = backup_header + "\n".join([f"{row[0]} | {row[1]} | {row[2]} | {row[3]}" for row in backup_rows])
        db_info["backups"] = backup_content if backup_rows else "近1天無(wú)備份記錄"

        # 3. 異常計(jì)劃任務(wù)（近2天未成功的任務(wù)）
        task_sql = """
            SELECT job_name, job_status, job_start_time
            FROM dba_autotask_job_history
            WHERE job_start_time > SYSDATE - 2 AND job_status != 'SUCCEEDED'
        """
        cursor.execute(task_sql)
        task_rows = cursor.fetchall()
        task_content = "\n".join([f"任務(wù)名：{row[0]} | 狀態(tài)：{row[1]} | 開(kāi)始時(shí)間：{row[2]}" for row in task_rows])
        db_info["abnormal_tasks"] = task_content if task_rows else "近2天無(wú)異常任務(wù)"

        # 4. 數(shù)據(jù)庫(kù)活動(dòng)會(huì)話
        session_sql = """
            SELECT vs.username, COUNT(*) active_session
            FROM v$session vs
            WHERE vs.status = 'ACTIVE'
            GROUP BY vs.username
        """
        cursor.execute(session_sql)
        session_rows = cursor.fetchall()
        session_content = "\n".join([f"用戶名：{row[0]} | 活動(dòng)會(huì)話數(shù)：{row[1]}" for row in session_rows])
        db_info["active_sessions"] = session_content if session_rows else "無(wú)活動(dòng)會(huì)話"

        # 5. 近1天歸檔日志數(shù)量
        archive_sql = "SELECT COUNT(*) FROM v$archived_log WHERE first_time > SYSDATE - 1"
        cursor.execute(archive_sql)
        db_info["archive_count"] = str(cursor.fetchone()[0])

        # 6. 表空間使用率
        tablespace_sql = """
            SELECT df.tablespace_name, ROUND((df.bytes - NVL(free.bytes, 0))/df.bytes*100, 2) used_percent
            FROM (SELECT tablespace_name, SUM(bytes) bytes FROM dba_data_files GROUP BY tablespace_name) df
            LEFT JOIN (SELECT tablespace_name, SUM(bytes) bytes FROM dba_free_space GROUP BY tablespace_name) free
            ON df.tablespace_name = free.tablespace_name
        """
        cursor.execute(tablespace_sql)
        ts_rows = cursor.fetchall()
        ts_content = "\n".join([f"表空間：{row[0]} | 使用率：{row[1]}%" for row in ts_rows])
        db_info["tablespace_usage"] = ts_content

        # 7. 異常狀態(tài)用戶（近30天鎖定/7天內(nèi)過(guò)期）
        user_sql = """
            SELECT username, lock_date, expiry_date
            FROM dba_users
            WHERE account_status != 'OPEN'
              AND created >= (SELECT TRUNC(created) FROM dba_users WHERE username = 'SYS') + 0.99999
              AND (lock_date >= SYSDATE - 30 OR (expiry_date BETWEEN SYSDATE -7 AND SYSDATE +7))
            ORDER BY created
        """
        cursor.execute(user_sql)
        user_rows = cursor.fetchall()
        user_content = "\n".join([f"用戶名：{row[0]} | 鎖定時(shí)間：{row[1]} | 過(guò)期時(shí)間：{row[2]}" for row in user_rows])
        db_info["abnormal_users"] = user_content if user_rows else "無(wú)異常狀態(tài)用戶"

        # 8. 歸檔日志開(kāi)啟狀態(tài)（log_archive_start參數(shù)）
        archive_status_sql = "SELECT value FROM v$parameter WHERE name = 'log_archive_start'"
        cursor.execute(archive_status_sql)
        db_info["archive_status"] = cursor.fetchone()[0].strip()

        # 9. 閃回區(qū)大小（轉(zhuǎn)換為GB，增強(qiáng)可讀性）
        flashback_sql = "SELECT value FROM v$parameter WHERE name = 'db_recovery_file_dest_size'"
        cursor.execute(flashback_sql)
        flashback_bytes = int(cursor.fetchone()[0])
        flashback_gb = flashback_bytes / (1024 ** 3)  # 字節(jié)轉(zhuǎn)GB
        db_info["flashback_size"] = f"{flashback_gb:.2f} GB"

        # 10. 鎖表數(shù)量
        lock_sql = "SELECT COUNT(*) FROM v$lock GROUP BY DECODE(request, 0, 'held', 'waiting')"
        cursor.execute(lock_sql)
        lock_count = len(cursor.fetchall())
        db_info["lock_count"] = str(lock_count)

    except cx_Oracle.Error as e:
        # 捕獲Oracle數(shù)據(jù)庫(kù)錯(cuò)誤
        db_info["db_error"] = f"數(shù)據(jù)庫(kù)連接/查詢失敗：{str(e)}"
    except Exception as e:
        db_info["db_error"] = f"未知錯(cuò)誤：{str(e)}"
    finally:
        if conn:
            conn.close()

    return db_info

def generate_html_report(server_info, db_info, db_sid, output_path):
    """生成HTML巡檢報(bào)告"""
    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    html_content = f"""
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>Oracle 健康巡檢報(bào)告 - {db_sid}</title>
    <style>
        body {{ font-family: Arial, sans-serif; margin: 20px; }}
        h1, h2 {{ color: #333; border-bottom: 1px solid #ddd; padding-bottom: 5px; }}
        table {{ width: 100%; border-collapse: collapse; margin: 15px 0; }}
        th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
        th {{ background-color: #f5f5f5; }}
        pre {{ background-color: #f8f8f8; padding: 10px; border-radius: 4px; overflow-x: auto; }}
    </style>
</head>
<body>
    <h1>XXXX項(xiàng)目</h1>
    <h2>Oracle 健康巡檢報(bào)告 - {db_sid}</h2>
    <p>巡檢時(shí)間：{current_time} &nbsp;&nbsp;&nbsp; 巡檢人員：zhh</p>

    <!-- 服務(wù)器層面巡檢 -->
    <h2>一、服務(wù)器層面巡檢</h2>
    <table>
        <tr><th>巡檢項(xiàng)</th><th>結(jié)果</th></tr>
        <tr><td>主機(jī)名</td><td>{server_info['hostname']}</td></tr>
        <tr><td>操作系統(tǒng)版本</td><td>{server_info['os_version']}</td></tr>
        <tr><td>根目錄存儲(chǔ)使用情況</td><td>{server_info['storage']}</td></tr>
        <tr><td>內(nèi)存使用情況</td><td>{server_info['memory']}</td></tr>
        <tr><td>CPU使用率</td><td>{server_info['cpu']}</td></tr>
    </table>

    <!-- 數(shù)據(jù)庫(kù)層面巡檢 -->
    <h2>二、數(shù)據(jù)庫(kù)層面巡檢（{db_sid}）</h2>
    <table>
        <tr><th>巡檢項(xiàng)</th><th>查詢邏輯</th><th>結(jié)果</th></tr>
"""

    # 數(shù)據(jù)庫(kù)巡檢項(xiàng)拼接（處理可能的連接錯(cuò)誤）
    if "db_error" in db_info:
        html_content += f"""
        <tr><td colspan="3" style="color: red;">{db_info['db_error']}</td></tr>
"""
    else:
        db_items = [
            ("數(shù)據(jù)庫(kù)異常日志", "tail 日志文件最后200行 + 篩選ERROR/WARNING", f"<pre>{db_info['error_logs']}</pre>"),
            ("近1天備份情況", "查詢v$backup_files", f"<pre>{db_info['backups']}</pre>"),
            ("近2天異常計(jì)劃任務(wù)", "查詢dba_autotask_job_history", f"<pre>{db_info['abnormal_tasks']}</pre>"),
            ("活動(dòng)會(huì)話", "查詢v$session（STATUS='ACTIVE'）", f"<pre>{db_info['active_sessions']}</pre>"),
            ("近1天歸檔日志數(shù)量", "查詢v$archived_log", db_info['archive_count']),
            ("表空間使用率", "dba_data_files + dba_free_space計(jì)算", f"<pre>{db_info['tablespace_usage']}</pre>"),
            ("異常狀態(tài)用戶", "查詢dba_users（鎖定/過(guò)期）", f"<pre>{db_info['abnormal_users']}</pre>"),
            ("歸檔日志開(kāi)啟狀態(tài)", "查詢v$parameter（log_archive_start）", db_info['archive_status']),
            ("閃回區(qū)大小", "查詢v$parameter（db_recovery_file_dest_size）", db_info['flashback_size']),
            ("鎖表分組數(shù)量", "查詢v$lock分組統(tǒng)計(jì)", db_info['lock_count'])
        ]
        for item_name, logic, result in db_items:
            html_content += f"""
        <tr><td>{item_name}</td><td>{logic}</td><td>{result}</td></tr>
"""

    # HTML尾部
    html_content += """
    </table>
</body>
</html>
"""

    # 寫(xiě)入HTML文件
    final_output = output_path.format(db_sid)
    with open(final_output, "w", encoding="utf-8") as f:
        f.write(html_content)
    print(f"巡檢報(bào)告已生成：{final_output}")

def main():
    # 解析參數(shù)
    args = parse_args()
    # 設(shè)置Oracle客戶端環(huán)境（根據(jù)實(shí)際環(huán)境調(diào)整，若已配置可注釋）
    os.environ["LD_LIBRARY_PATH"] = "/u01/app/oracle/product/11.2.0/db_1/lib"  # 示例路徑

    # 1. 獲取服務(wù)器信息
    print("正在獲取服務(wù)器信息...")
    server_info = get_server_info()

    # 2. 獲取數(shù)據(jù)庫(kù)信息
    print("正在獲取數(shù)據(jù)庫(kù)信息...")
    db_info = get_db_info(
        db_host=args.db_host,
        db_port=args.db_port,
        db_sid=args.db_sid,
        db_user=args.db_user,
        db_pwd=args.db_pwd
    )

    # 3. 生成HTML報(bào)告
    print("正在生成HTML報(bào)告...")
    generate_html_report(server_info, db_info, args.db_sid, args.html_output)
    print("巡檢完成！")

if __name__ == "__main__":
    main()

二、使用前準(zhǔn)備

1. 安裝依賴庫(kù)

   # 安裝cx_Oracle（用于Oracle數(shù)據(jù)庫(kù)連接）
   pip install cx_Oracle
   

2. 配置Oracle客戶端

   • 若服務(wù)器未安裝Oracle數(shù)據(jù)庫(kù)，需單獨(dú)安裝Oracle Instant Client（輕量級(jí)客戶端）。
   • 設(shè)置環(huán)境變量LD_LIBRARY_PATH指向客戶端庫(kù)路徑（腳本中已包含示例，需根據(jù)實(shí)際路徑調(diào)整）。

3. 權(quán)限要求

   • 執(zhí)行腳本的用戶需具備：
     • 讀取服務(wù)器文件權(quán)限（/etc/os-release、Oracle日志文件）。
     • 執(zhí)行系統(tǒng)命令權(quán)限（df、free、top）。
     • Oracle數(shù)據(jù)庫(kù)SYSDBA權(quán)限（用于查詢系統(tǒng)視圖）。

三、執(zhí)行命令示例

python oracle_healthcheck.py \
  --db-host 192.168.1.100 \
  --db-port 1521 \
  --db-sid orcl11g \
  --db-user sys \
  --db-pwd YourSysPassword \
  --html-output ./oracle_report_{}.html