disk_checker.py

"""
TOCP CF 卡健康檢查工具
透過 Telnet 遠端連線至 TOCP 設備 (QNX 6.4.1)，
執行 chkfsys / dcheck 檢查並記錄結果。
"""

import asyncio
import telnetlib3
import os
from datetime import datetime
import re
import sys
import questionary
# --- 設定常數 ---
IP_SEQUENCE = [
    ("TOCP 前 Port", "192.168.100.99"),
    ("TOCP 後 Port", "192.169.100.99"),
]
TELNET_PORT = 23
TELNET_USER = "root"
PORT_CHECK_TIMEOUT = 1.5

# 編譯後 __file__ 會指向暫存目錄，需改用 sys.executable 取得執行檔位置
if getattr(sys, "frozen", False):
    _BASE_DIR = os.path.dirname(sys.executable)
else:
    _BASE_DIR = os.path.dirname(os.path.abspath(__file__))
OUTPUT_DIR = os.path.join(_BASE_DIR, "disk_checker_log")
DEVICE_PATH_DEFAULT = "/dev/hd0t79"
RAW_DEVICE_PATH = "/dev/hd0"


# ==============================================================
# 工具函式
# ==============================================================

async def port_is_open(host: str, port: int, timeout: float) -> bool:
    """快速檢查 TCP 端口是否開放。"""
    try:
        _, writer = await asyncio.wait_for(
            asyncio.open_connection(host, port), timeout=timeout
        )
        writer.close()
        await writer.wait_closed()
        return True
    except (asyncio.TimeoutError, ConnectionRefusedError, OSError):
        return False


async def find_device() -> tuple[str, str] | None:
    """並行檢查所有 IP，回傳第一個可連線的裝置 (label, ip)。"""
    tasks = [
        asyncio.create_task(port_is_open(ip, TELNET_PORT, PORT_CHECK_TIMEOUT))
        for _, ip in IP_SEQUENCE
    ]
    results = await asyncio.gather(*tasks)
    for i, ok in enumerate(results):
        if ok:
            return IP_SEQUENCE[i]
    return None


# ==============================================================
# DiskChecker 主類別
# ==============================================================

class DiskChecker:
    """TOCP 設備 CF 卡檢查器。"""

    def __init__(self):
        self.reader = None
        self.writer = None
        self.device_label = ""
        self.device_ip = ""
        self.current_partition = DEVICE_PATH_DEFAULT
        self.available_partitions = []
        self.log_lines: list[str] = []
        self._base_log: list[str] = []  # 連線/磁碟使用率等基礎資訊，每次存檔都保留

    # ---------- 連線 / 登入 ----------

    async def connect(self) -> bool:
        """自動偵測並連線至可用的 TOCP 設備。"""
        print("正在並行檢查所有可用裝置...")
        device = await find_device()
        if device is None:
            print(f"所有裝置在 {PORT_CHECK_TIMEOUT} 秒內均無回應。")
            return False

        self.device_label, self.device_ip = device
        print(f"找到裝置：{self.device_label} ({self.device_ip})")
        print(f"正在建立 Telnet 連線...")

        try:
            self.reader, self.writer = await asyncio.wait_for(
                telnetlib3.open_connection(self.device_ip, TELNET_PORT, shell=None),
                timeout=5.0,
            )
        except Exception as e:
            print(f"無法建立 Telnet 連線：{e}")
            return False

        # 登入
        await asyncio.sleep(1)
        self.writer.write(TELNET_USER + "\n")
        await self.writer.drain()
        await asyncio.sleep(1)
        await self.reader.read(4096)  # 清除歡迎訊息
        print(f"✅ 連線成功！已登入為 {TELNET_USER}\n")
        
        # 連線後自動偵測分區
        await self.detect_partitions()
        
        return True

    async def detect_partitions(self):
        """偵測系統上所有可用的 QNX 分區 (/dev/hd0t*)。"""
        print("正在偵測可用分區...")
        try:
            output = await self.run_command("ls /dev/hd0t*", timeout=5.0)
            # 過濾出類似 /dev/hd0t77, /dev/hd0t79 的路徑
            parts = []
            for token in output.split():
                # 過濾出類似 /dev/hd0t77, /dev/hd0t79 的路徑，排除包含 * 的項目
                if token.startswith("/dev/hd0t") and token != "/dev/hd0" and "*" not in token:
                    parts.append(token.strip())
            
            if parts:
                self.available_partitions = sorted(parts)
                print(f"找到分區：{', '.join(self.available_partitions)}")
                # 如果原本預設的分區在列表內，就保持預設，否則選第一個
                if self.current_partition not in self.available_partitions:
                    self.current_partition = self.available_partitions[0]
            else:
                print("⚠️ 未找到任何 hd0t* 分區，維持預設值。")
                
        except Exception as e:
            print(f"分區偵測失敗：{e}")

    async def disconnect(self):
        """關閉連線。"""
        if self.writer:
            self.writer.close()
            await self.writer.wait_closed()
            print("連線已關閉。")

    # ---------- 指令執行 ----------

    async def _flush_reader(self):
        """清空 reader buffer 中的殘留資料，避免污染下一個指令。"""
        while True:
            try:
                chunk = await asyncio.wait_for(self.reader.read(4096), timeout=0.3)
                if not chunk:
                    break
            except asyncio.TimeoutError:
                break
            except (ConnectionResetError, ConnectionAbortedError, OSError, EOFError):
                raise ConnectionError("與設備的連線已中斷")

    async def run_command(self, cmd: str, timeout: float = 10.0,
                          realtime: bool = False, show_progress: bool = False) -> str:
        """
        送出指令並讀取完整回應。
        realtime=True 時會即時印出過濾後的輸出。
        show_progress=True 時會嘗試解析 dcheck 輸出並顯示進度條。
        """
        # 清空前一個指令可能殘留的資料
        await self._flush_reader()

        # 使用唯一 marker 防止跨指令匹配
        self._cmd_seq = getattr(self, "_cmd_seq", 0) + 1
        marker = f"__CMD_DONE_{self._cmd_seq}__"
        
        # 分開傳送指令與 marker，避免 shell 解析問題
        self.writer.write(cmd + "\n")
        await self.writer.drain()
        
        # 稍微等待一下讓指令開始執行
        if "dcheck" in cmd:
             await asyncio.sleep(0.5)

        self.writer.write(f"echo {marker}\n")
        await self.writer.drain()

        output = ""
        loop = asyncio.get_running_loop()
        deadline = loop.time() + timeout
        
        spinner = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]
        idx = 0
        
        # 記錄開始時間與上次更新時間
        start_t = loop.time()
        last_spinner_update = start_t

        while loop.time() < deadline:
            remaining = deadline - loop.time()
            chunk_timeout = 0.1  #縮短 timeout 以便頻繁刷新 spinner

            try:
                chunk = await asyncio.wait_for(
                    self.reader.read(8192), timeout=chunk_timeout
                )
                if chunk:
                    output += chunk
                    
                    if realtime:
                        # 即時印出但過濾掉 marker 和 command echo
                        lines = chunk.split("\n")
                        for line in lines:
                            stripped = line.strip()
                            if marker in stripped:
                                continue
                            if stripped == cmd.strip():
                                continue
                            if stripped == f"echo {marker}":
                                continue
                            if stripped == "#":
                                continue
                            
                            # --- 即時輸出過濾邏輯 (與 Log 過濾保持一致) ---
                            # 過濾只包含數字、空白、橫槓的行（倒數 & dcheck 進度）
                            if re.match(r'^[\d\s\-]+$', stripped):
                                continue

                            # 過濾 "Comparing new bitmap" 標題
                            if "comparing new bitmap" in stripped.lower():
                                continue

                            if stripped:
                                # 清除 spinner 行，印出內容
                                print(f"\r{' '*50}\r{line}", flush=True)

                    # --- 互動模式偵測 ---
                    # 檢查是否有常見的詢問句
                    lower_chunk = chunk.lower()
                    if any(p in lower_chunk for p in ["(y/n)", "[y/n]", "continue?", "confirm?"]):
                        print(f"\n\n{'!'*50}")
                        print("⚠️  偵測到設備可能在等待輸入！")
                        print(f"內容: {chunk.strip()}")
                        print(f"{'!'*50}")
                        
                        # 暫停 Spinner
                        user_input = await asyncio.to_thread(input, "請輸入要傳送的回應 (預設 y): ")
                        if not user_input.strip():
                            user_input = "y"
                        
                        print(f"傳送回應: {user_input}")
                        self.writer.write(user_input + "\n")
                        await self.writer.drain()
                        
                        # 重置超時，給予更多時間處理
                        deadline = loop.time() + timeout 

                    if marker in output:
                        break
            except asyncio.TimeoutError:
                pass # 繼續執行下方更新邏輯
            except (ConnectionResetError, ConnectionAbortedError, OSError, EOFError):
                raise ConnectionError("與設備的連線已中斷")
            
            # 定期更新 Spinner (不管有沒有收到資料)
            if show_progress:
                now = loop.time()
                if now - last_spinner_update > 0.5: # 至少 0.5 秒更新一次
                    idx = (idx + 1) % len(spinner)
                    elapsed = now - start_t
                    print(f"\r{spinner[idx]} 正在掃描中... 已耗時: {elapsed:.1f} 秒", end="", flush=True)
                    last_spinner_update = now
        
        if show_progress:
            print(f"\r{' '*50}", end="\r") # 清除最後的 spinner
            

        # 清理輸出：先移除 telnet 殘留的 null bytes，避免 marker 被拆散
        output = output.replace("\x00", "")

        # 移除 marker 與指令回顯
        lines = output.split("\n")
        cleaned = []
        for line in lines:
            stripped = line.strip()
            if marker in stripped:
                continue
            if stripped == cmd.strip():
                continue
            if stripped == f"echo {marker}":
                continue
            if stripped == "#":
                continue
            cleaned.append(line)

        return "\n".join(cleaned).strip()

    # ---------- 檢查功能 ----------

    async def check_filesystem_full(self) -> dict:
        """完整檔案系統檢查 (chkfsys -uv)，強制掃描。"""
        print(f"\n{'='*50}")
        print(f"  完整檔案系統檢查 (chkfsys -uv {self.current_partition})")
        print(f"  這可能需要數十秒，請稍候...")
        print(f"{'='*50}")
        self._log(f"\n--- 完整檔案系統檢查 ({self.current_partition}) ---")

        output = await self.run_command(
            f"yes | chkfsys -uv {self.current_partition}", timeout=120.0, realtime=True
        )
        print()  # realtime 輸出後換行
        filtered_log = self._filter_garbage(output)
        self._log(filtered_log)

        return self._parse_chkfsys(output)

    def _filter_garbage(self, output: str) -> str:
        """過濾過大的輸出中的無意義內容（倒數、進度等），其餘內容（含檔案清單）完整保留。"""
        # 清除 telnet 殘留的 null bytes，再正規化換行
        normalized = output.replace("\x00", "").replace("\r", "\n")
        lines = normalized.splitlines()

        # 匹配「只有數字、空白、橫槓」的行
        # 涵蓋：chkfsys 倒數 (486 485 484 ...)、dcheck 進度 (353094  767877 - 20)
        noise_pattern = re.compile(r'^[\d\s\-]+$')

        clean_lines = []
        for line in lines:
            s_line = line.strip()

            if not s_line:
                # 簡單合併多個空行，只留一個
                if clean_lines and not clean_lines[-1].strip():
                    continue
                clean_lines.append(line)
                continue

            # 過濾 "Comparing new bitmap" 標題
            if "comparing new bitmap" in s_line.lower():
                continue

            # 過濾只包含數字、空白、橫槓的行（倒數 & dcheck 進度）
            if noise_pattern.match(s_line):
                continue

            clean_lines.append(line)

        return "\n".join(clean_lines)

    async def get_disk_size(self, device: str) -> int:
        """嘗試取得磁碟總區塊數。"""
        # 嘗試方法 1: 使用 fdisk query
        try:
            output = await self.run_command(f"fdisk {device} query_part", timeout=2.0)
            # 解析輸出尋找區塊數，若失敗回傳 0
            # 這裡假設輸出可能包含區塊數
            nums = [int(s) for s in output.split() if s.isdigit()]
            if nums:
                return max(nums)
        except:
            pass
        
        # 嘗試方法 2: 硬編碼估計值 (如果 query 失敗)
        # 根據經驗，若是完整硬碟，通常有數百萬個區塊
        # 1GB 硬碟 = 1,073,741,824 bytes / 512 bytes/block ≈ 2,097,152 blocks
        if device == RAW_DEVICE_PATH:
            return 2097152  # 1GB 硬碟的區塊數
        return 0

    async def check_bad_blocks_full_disk(self) -> dict:
        """完整 CF 卡壞軌掃描 (dcheck -rv /dev/hd0)。"""
        print(f"\n{'='*50}")
        print(f"  完整 CF 卡壞軌掃描 (dcheck)")
        print(f"  ⏱️  掃描約需 2 分鐘，請耐心等候...")
        print(f"{'='*50}")
        self._log(f"\n--- 完整 CF 卡壞軌掃描 (Whole Disk) ---")

        # 嘗試取得總區塊數以顯示進度
        total_blocks = await self.get_disk_size(RAW_DEVICE_PATH)
        if total_blocks == 0:
            print("無法取得總區塊數，將使用預估值顯示進度...")
            total_blocks = 2000000 # 200萬區塊預估

        start_time = datetime.now()
        # 啟用進度條模式
        output = await self.run_command(
            f"dcheck -rv {RAW_DEVICE_PATH}",
            timeout=600.0,
            realtime=False,
            show_progress=True,
        )
        duration = (datetime.now() - start_time).total_seconds()
        
        time_msg = f"\n[執行時間] {duration:.2f} 秒"
        print(time_msg)
        
        filtered_log = self._filter_garbage(output)
        self._log(filtered_log)
        self._log(time_msg)

        return self._parse_dcheck(output, total_scanned=total_blocks)

    async def get_disk_usage(self) -> str:
        """取得磁碟使用率 (df)。"""
        output = await self.run_command("df")
        return output

    async def get_system_log(self) -> str:
        """取得完整系統日誌 (sloginfo)。"""
        print(f"\n{'='*50}")
        print(f"  系統日誌 (sloginfo)")
        print(f"{'='*50}")
        self._log(f"\n--- 系統日誌 ---")

        output = await self.run_command("sloginfo", timeout=10.0)
        print(output)
        self._log(output)
        return output

    # ---------- 結果解析 ----------

    def _parse_chkfsys(self, output: str) -> dict:
        """解析 chkfsys 輸出，判斷結果。"""
        result = {"status": "UNKNOWN", "detail": "", "raw": output}
        lower = output.lower()
        
        # 狀態優先級：ERROR > FIXED > DIRTY > CLEAN > OK
        
        if "corruption" in lower or "error" in lower or "cross-link" in lower:
            result["status"] = "ERROR"
            result["detail"] = "❌ 嚴重錯誤：偵測到檔案系統損壞、結構錯誤或交叉連結！"
        elif "fixed" in lower or "deleted" in lower or "recovered" in lower:
            result["status"] = "FIXED"
            result["detail"] = "⚠️ 警告：偵測到問題但已自動修復 (檔案刪除或復原)。"
        elif "dirty" in lower:
             # 有時候 -u 會忽略 dirty 標記繼續掃，但若是檢查報告提及 dirty 仍需注意
            result["status"] = "FIXED" 
            result["detail"] = "⚠️ 警告：檔案系統標記為 Dirty，已執行檢查與修復。"
        elif "ok" in lower or len(output.strip()) == 0:
             # 有些版本的 chkfsys 成功時可能沒說話或只說 OK
             result["status"] = "OK"
             result["detail"] = "✅ 掃描完成 (無異常訊息)。"
        else:
            # 沒捕捉到明確關鍵字，顯示原始輸出讓使用者自己判斷
            result["status"] = "MANUAL_CHECK"
            result["detail"] = "⚠️ 無法自動判讀結果，請參閱上方原始輸出訊息！"

        return result

    def _parse_dcheck(self, output: str, total_scanned: int = 0) -> dict:
        """解析 dcheck 輸出，判斷結果。"""
        result = {
            "status": "UNKNOWN", 
            "bad_blocks": 0, 
            "detail": "", 
            "raw": output,
            "scanned_count": total_scanned
        }
        lower = output.lower()

        # 優先檢查是否有壞軌報告
        # dcheck 輸出範例: "Block 12345 is bad"
        if "is bad" in lower:
             # 統計壞軌數量
            bad_count = lower.count("is bad")
            result["bad_blocks"] = bad_count
            result["status"] = "BAD_BLOCKS"
            result["detail"] = f"❌ 嚴重警告：發現 {bad_count} 個壞軌！請儘速更換 CF 卡。"
            
        elif "read error" in lower or "seek error" in lower or "fail" in lower:
            result["status"] = "ERROR"
            result["detail"] = "❌ 掃描失敗：發生讀取錯誤或是硬體通訊異常。"
            
        elif "error" in lower:
             # 有些 minor error 可能不影響使用，但仍需標示
             result["status"] = "ERROR"
             result["detail"] = "⚠️ 掃描過程中發生未預期的錯誤。"
             
        elif "bad" not in lower and "error" not in lower and len(output.strip()) > 10:
             # 有輸出但沒捕捉到關鍵字
            result["status"] = "MANUAL_CHECK"
            result["detail"] = "⚠️ 無法自動判讀結果，請參閱上方原始輸出訊息！"
            
        else:
            result["status"] = "OK"
            result["detail"] = "✅ 掃描完成，CF 卡狀況良好，未發現壞軌。"

        return result

    # ---------- 結果顯示 ----------

    def print_summary(self, results: list[dict]):
        """印出所有檢查的摘要報告。"""
        print(f"\n{'='*50}")
        print(f"  檢查結果摘要")
        print(f"{'='*50}")

        summary_lines = []
        for r in results:
            status = r.get("status", "UNKNOWN")
            detail = r.get("detail", "")
            scanned = r.get("scanned_count", 0)

            if status in ("CLEAN", "OK"):
                icon = "✅"
            elif status == "FIXED":
                icon = "⚠️"
            elif status == "MANUAL_CHECK":
                icon = "👀"
            else:
                icon = "❌"

            line = f"  {icon} [{status}] {detail}"
            print(line)
            summary_lines.append(line)
            
            if scanned > 0:
                scanned_msg = f"      └─ 掃描磁區數: {scanned:,}"
                print(scanned_msg)
                summary_lines.append(scanned_msg)

        self._log("\n--- 檢查結果摘要 ---")
        for line in summary_lines:
            self._log(line)

    # ---------- 日誌記錄 ----------

    def _log(self, text: str):
        """將文字加入日誌緩衝。"""
        self.log_lines.append(text)

    def save_log(self):
        """將日誌存檔至 output 資料夾，存檔後只保留基礎資訊避免重複累積。"""
        os.makedirs(OUTPUT_DIR, exist_ok=True)
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"disk_check_{timestamp}.log"
        filepath = os.path.join(OUTPUT_DIR, filename)

        header = [
            f"TOCP CF 卡健康檢查報告",
            f"裝置：{self.device_label} ({self.device_ip})",
            f"時間：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
            f"{'='*50}",
        ]

        with open(filepath, "w", encoding="utf-8") as f:
            f.write("\n".join(header) + "\n")
            # 清除殘留的 telnet 控制字元，確保 log 為純文字
            content = "\n".join(self.log_lines)
            content = content.replace("\x00", "").replace("\r", "")
            f.write(content + "\n")

        print(f"\n📄 報告已儲存至：{filepath}")
        # 清空本次檢查資料，但保留基礎資訊供下次存檔使用
        self.log_lines = list(self._base_log)
        return filepath


# ==============================================================
# 主程式
# ==============================================================

async def main():
    print("╔══════════════════════════════════════╗")
    print("║   TOCP CF 卡健康檢查工具 v1.0.2       ║")
    print("╚══════════════════════════════════════╝")
    print()

    checker = DiskChecker()

    # 連線
    if not await checker.connect():
        input("按 Enter 鍵結束...")
        return

    # 顯示磁碟使用率
    print("--- CF 卡使用率 ---")
    df_output = await checker.get_disk_usage()
    print(df_output)
    checker._log(f"\n--- CF 卡使用率 ---\n{df_output}")
    # 保存基礎資訊，每次存檔時都會包含
    checker._base_log = list(checker.log_lines)

    try:
        while True:
            # 使用 questionary 建立互動式選單
            choices = [
                questionary.Choice("1. CF卡 檔案系統檢查 (chkfsys)", value="1"),
                questionary.Choice("2. CF卡 磁區壞軌掃描 (dcheck)", value="2"),
                questionary.Choice("3. 全面檢查 (CF卡檔案系統 + 磁區壞軌掃描)", value="3"),
                questionary.Choice("4. 系統日誌", value="4"),
                questionary.Choice("5. 結束程式", value="5"),
            ]

            print(f"\n{'─'*40}")
            print(f"目前目標分區：[{checker.current_partition}]")
            
            # 使用非同步上下文執行 questionary
            answer = await asyncio.to_thread(
                questionary.select(
                    "請選擇檢查項目：",
                    choices=choices,
                    use_indicator=True,
                    style=questionary.Style([
                        ('qmark', 'fg:#E91E63 bold'),       # token in front of the question
                        ('question', 'bold'),               # question text
                        ('answer', 'fg:#f44336 bold'),      # submitted answer text behind the question
                        ('pointer', 'fg:#673ab7 bold'),     # pointer used in select and checkbox prompts
                        ('highlighted', 'fg:#673ab7 bold'), # pointed-at choice in select and checkbox prompts
                        ('selected', 'fg:#cc5454'),         # style for a selected choice of a checkbox
                        ('separator', 'fg:#cc5454'),        # separator in lists
                        ('instruction', ''),                # user instructions for select, rawselect, checkbox
                        ('text', ''),                       # plain text
                        ('disabled', 'fg:#858585 italic')   # disabled choices for select, rawselect, checkbox
                    ])
                ).ask
            )
            
            if answer is None: # 使用者按 Ctrl+C 取消
                break
                
            choice = answer

            results = []

            try:
                if choice == "1":
                    # 原本的完整檢查 (chkfsys -uv)
                    r = await checker.check_filesystem_full()
                    results.append(r)

                elif choice == "2":
                    # 原本的整顆硬碟壞軌掃描 (dcheck -rv hd0)
                    r = await checker.check_bad_blocks_full_disk()
                    results.append(r)

                elif choice == "3":
                    # 全面檢查
                    print("\n開始全面檢查，請耐心等候...\n")
                    r1 = await checker.check_filesystem_full()
                    results.append(r1)
                    r2 = await checker.check_bad_blocks_full_disk() # 改用整顆硬碟掃描
                    results.append(r2)

                elif choice == "4":
                    await checker.get_system_log()

                elif choice == "5":
                    print("\n使用者選擇結束程式。")
                    break

            except ConnectionError as e:
                print(f"\n❌ 連線中斷：{e}")
                print("設備可能已斷線或重啟，程式即將結束。")
                break

            # 顯示摘要 & 存檔
            if results:
                checker.print_summary(results)
                checker.save_log()

    finally:
        await checker.disconnect()

    input("\n按 Enter 鍵關閉視窗...")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n偵測到使用者中斷 (Ctrl+C)，程式已終止。")