diff --git a/.clineignore b/.clineignore new file mode 100644 index 0000000..41ab61b --- /dev/null +++ b/.clineignore @@ -0,0 +1,16 @@ +# 1. Python 虛擬環境與快取 (吃錢最大凶手，絕對不要讓 AI 讀！) +cmts_api_env/ +__pycache__/ +*.pyc + +# 2. 其他 AI 工具的隱藏資料夾 +.continue/ + +# 3. 動態生成的 JSON 快取資料 (非常耗 Token，且對 AI 寫程式幫助不大) +*_cache.json +filters_*.json +*.bk + +# 4. 版控與系統隱藏檔 (如果有這兩個的話) +.git/ +.vscode/ diff --git a/.clinerules b/.clinerules new file mode 100644 index 0000000..ddc4f83 --- /dev/null +++ b/.clinerules @@ -0,0 +1,41 @@ +# Harmonic CMTS Manager - AI 開發守則與架構白皮書 + +## 1. 專案架構概覽與技術棧 +- **定位**: 專為有線電視網路終端設備 (CMTS) 設計的企業級 Web 管理系統。 +- **後端**: Python 3.10+, FastAPI (Async-first), AsyncSSH, Netmiko, asyncpg. +- **前端**: 原生 Vanilla JS (ES Modules), HTML5, CSS3, Xterm.js. +- **資料庫**: PostgreSQL (主要), JSON File Cache (高可用性降級備援). + +### 📂 目錄與檔案結構 +- **進入點**: `main.py` +- **路由管理**: API 路由統一放置於 `routers/` 目錄。 +- **前端介面**: `index.html` 與 `static/` 目錄。 +- **核心邏輯**: + - `cmts_scraper.py`: 負責底層爬蟲與資料處理。 + - `shared.py`: 放置共用函式 (如兩階段解析法)。 + - `config.py`: 配置轉譯器 (CLI Generator)。 +- **資料庫**: 連線與 ORM 邏輯在 `database.py`，初始化腳本為 `init_db.py`。 +- **併發控制**: `lock.py` (路徑階層鎖), `leaf_options.py` (SSE 頻道分流)。 + +## 2. 核心架構與業務邏輯 (Architecture & Logic) +- **兩階段解析法**: 依賴「縮排」與「`!`」劃分區塊，並動態降維成深層巢狀結構，使用 `deep_merge` 確保資料不遺失。 +- **動態探測**: 透過發送 `[指令] ?` 動態學習資料結構，並採用 `BATCH_SIZE = 30` 搭配 `asyncio.sleep()` 進行非同步批次處理。 +- **併發與狀態廣播**: 支援「父子繼層攔截」的路徑階層鎖，並利用 `asyncio.Queue` 實作 SSE 頻道分流，即時推播進度。 +- **配置轉譯器**: 自動補齊父層級路徑，處理 `no [指令]` 刪除邏輯，並具備 `admin-state` 的生命週期防呆機制。 + +## 3. 🚨 AI 開發絕對約束 (Directives for AI) + +### ⚙️ 系統與環境規範 +1. **套件管理**: 若需安裝新套件，請提醒我手動在 `cmts_api_env` 中安裝，並更新 `requirements.txt`。 +2. **資料讀取限制**: 請勿隨意讀取 `*_cache.json` 檔案的內容，若需了解資料結構，請參考 `cmts_scraper.py` 中的定義。 +3. **狀態同步**: 完成重大修改或一個 Phase 後，必須主動更新 `PROJECT_STATE.md` 記錄最新進度與待辦事項。 + +### 💻 程式碼風格與後端規範 +4. **語言與風格**: 註解與對話請一律使用**繁體中文**。Python 程式碼請遵循 PEP8 規範，並加上適當的 Type Hints (型別提示)。 +5. **Async-First (非同步絕對優先)**: 嚴禁使用阻塞的同步 I/O。同步函數必須封裝進 `await asyncio.to_thread()`。 +6. **嚴守「雙軌並行」**: 必須在任何 API 請求明確傳遞並驗證 `config_type` ('running' 或 'full')。絕對禁止 `running` 污染 `full` 快取。 +7. **無痛切換 (Feature Toggle)**: 必須保留 `USE_DB` 開關，若 DB 連線異常，必須能自動退回使用 JSON 檔案讀寫。 +8. **無聲錯誤是原罪**: 所有設備互動模組必須使用 `try-except`，並回傳標準 JSON `{"status": "error", "message": "..."}`。嚴禁 FastAPI 直接拋出 500。 + +### 🎨 前端規範 +9. **DOM 神聖不可侵犯**: 在前端 JS 中，嚴禁為了視覺美化刪除 `leaf-container`, `data-path`, `data-original` 等錨點。隱藏請用 `display: none`。 diff --git a/.continue/rules/new-rule.md b/.continue/rules/new-rule.md new file mode 100644 index 0000000..ddc4f83 --- /dev/null +++ b/.continue/rules/new-rule.md @@ -0,0 +1,41 @@ +# Harmonic CMTS Manager - AI 開發守則與架構白皮書 + +## 1. 專案架構概覽與技術棧 +- **定位**: 專為有線電視網路終端設備 (CMTS) 設計的企業級 Web 管理系統。 +- **後端**: Python 3.10+, FastAPI (Async-first), AsyncSSH, Netmiko, asyncpg. +- **前端**: 原生 Vanilla JS (ES Modules), HTML5, CSS3, Xterm.js. +- **資料庫**: PostgreSQL (主要), JSON File Cache (高可用性降級備援). + +### 📂 目錄與檔案結構 +- **進入點**: `main.py` +- **路由管理**: API 路由統一放置於 `routers/` 目錄。 +- **前端介面**: `index.html` 與 `static/` 目錄。 +- **核心邏輯**: + - `cmts_scraper.py`: 負責底層爬蟲與資料處理。 + - `shared.py`: 放置共用函式 (如兩階段解析法)。 + - `config.py`: 配置轉譯器 (CLI Generator)。 +- **資料庫**: 連線與 ORM 邏輯在 `database.py`，初始化腳本為 `init_db.py`。 +- **併發控制**: `lock.py` (路徑階層鎖), `leaf_options.py` (SSE 頻道分流)。 + +## 2. 核心架構與業務邏輯 (Architecture & Logic) +- **兩階段解析法**: 依賴「縮排」與「`!`」劃分區塊，並動態降維成深層巢狀結構，使用 `deep_merge` 確保資料不遺失。 +- **動態探測**: 透過發送 `[指令] ?` 動態學習資料結構，並採用 `BATCH_SIZE = 30` 搭配 `asyncio.sleep()` 進行非同步批次處理。 +- **併發與狀態廣播**: 支援「父子繼層攔截」的路徑階層鎖，並利用 `asyncio.Queue` 實作 SSE 頻道分流，即時推播進度。 +- **配置轉譯器**: 自動補齊父層級路徑，處理 `no [指令]` 刪除邏輯，並具備 `admin-state` 的生命週期防呆機制。 + +## 3. 🚨 AI 開發絕對約束 (Directives for AI) + +### ⚙️ 系統與環境規範 +1. **套件管理**: 若需安裝新套件，請提醒我手動在 `cmts_api_env` 中安裝，並更新 `requirements.txt`。 +2. **資料讀取限制**: 請勿隨意讀取 `*_cache.json` 檔案的內容，若需了解資料結構，請參考 `cmts_scraper.py` 中的定義。 +3. **狀態同步**: 完成重大修改或一個 Phase 後，必須主動更新 `PROJECT_STATE.md` 記錄最新進度與待辦事項。 + +### 💻 程式碼風格與後端規範 +4. **語言與風格**: 註解與對話請一律使用**繁體中文**。Python 程式碼請遵循 PEP8 規範，並加上適當的 Type Hints (型別提示)。 +5. **Async-First (非同步絕對優先)**: 嚴禁使用阻塞的同步 I/O。同步函數必須封裝進 `await asyncio.to_thread()`。 +6. **嚴守「雙軌並行」**: 必須在任何 API 請求明確傳遞並驗證 `config_type` ('running' 或 'full')。絕對禁止 `running` 污染 `full` 快取。 +7. **無痛切換 (Feature Toggle)**: 必須保留 `USE_DB` 開關，若 DB 連線異常，必須能自動退回使用 JSON 檔案讀寫。 +8. **無聲錯誤是原罪**: 所有設備互動模組必須使用 `try-except`，並回傳標準 JSON `{"status": "error", "message": "..."}`。嚴禁 FastAPI 直接拋出 500。 + +### 🎨 前端規範 +9. **DOM 神聖不可侵犯**: 在前端 JS 中，嚴禁為了視覺美化刪除 `leaf-container`, `data-path`, `data-original` 等錨點。隱藏請用 `display: none`。 diff --git a/PROJECT_STATE.md b/PROJECT_STATE.md new file mode 100644 index 0000000..72a2c36 --- /dev/null +++ b/PROJECT_STATE.md @@ -0,0 +1,28 @@ +# 📖 Harmonic CMTS Manager - Project State (Living Document) + +> ⚠️ **AI 助手請注意**：本專案的核心架構與開發規範已移至 `.clinerules`。此檔案僅作為「專案進度存檔」與「待辦事項追蹤」使用。 + +--- + +## ✅ 已完成開發階段 (Completed Phases) + +### Phase 1: PostgreSQL 高可用性架構升級 (Completed) +- [x] 成功導入 `asyncpg`，建立 `database.py` 管理非同步資料庫連線池。 +- [x] 建立 `cmts_options`, `device_status`, `system_filters` 資料表，嚴格遵守 `config_type` 雙軌隔離的主鍵設計。 +- [x] 實踐完整的「**Zero-Downtime Fallback 機制**」：資料庫連線異常時，自動捕捉錯誤並退回使用 JSON 檔案讀寫，防止 FastAPI Crash。 +- [x] 調整 `init_db.py` 為非同步啟動腳本。 + +--- + +## 🚧 目前開發階段 (Current Phase) + +### Phase 2: 設備配置備份與快照機制 (Configuration Backup & Snapshots) +- [ ] **資料庫擴充**：在 PostgreSQL 中建立 `config_backups` 資料表 (核心欄位需包含 `id`, `host`, `timestamp`, `raw_cli`, `parsed_tree`, `snapshot_name`)。 +- [ ] **後端 API 實作**：新增「手動建立快照 (Manual Snapshot)」與「取得歷史快照列表」的 RESTful API 路由。 +- [ ] **自動備份攔截**：在執行任何 `generate_cli` (寫入設備變更) 之前，實作自動觸發背景備份 `running config` 的防呆機制。 + +--- + +## 🐛 已知問題與未來計畫 (Known Issues & Backlog) +- **[未來計畫] Phase 3: 歷史預覽 (History Preview)**：前端 UI 支援點擊歷史快照，將 `parsed_tree` 載入 `tree-ui.js` 並強制標示為「唯讀模式 (Read-only)」。 +- **[未來計畫] Phase 4: 智慧還原 (Smart Recovery)**：實作 Diff-based Rollback 演算法。比對歷史備份與當前樹狀結構，自動產生反向 CLI 指令 (`no [指令]`)，並在推送到設備前提供 Dry-Run 預覽確認視窗。 diff --git a/cmts_scraper.py b/cmts_scraper.py index fed0efb..012a572 100644 --- a/cmts_scraper.py +++ b/cmts_scraper.py @@ -5,6 +5,8 @@ import re import json import os import time +import database +from shared import USE_DB def parse_question_mark_output(output: str) -> dict: """解析 '?' 回傳內容，支援無 Description、子命令判定與 dhcp-relay 混合欄位""" @@ -258,50 +260,75 @@ async def sync_cmts_leaves_async(host, username, password, leaf_paths: list, con finally: if conn: conn.close() - # --- 步驟 3：寫入快取檔 --- + # --- 步驟 3：寫入快取 (DB or JSON) --- try: - # 🌟 2. 動態決定快取檔名 (加入 IP 隔離) - safe_host = host.replace(".", "_") - cache_file = f"{safe_host}_{config_type}_cache.json" - - # ========================================== - # 🌟 優化 2-A：將「讀取」舊快取檔丟到背景執行緒 - # ========================================== - def read_json_from_file(filepath): - if os.path.exists(filepath): - with open(filepath, "r", encoding="utf-8") as f: - return json.load(f) - return {} - - cache_data = await asyncio.to_thread(read_json_from_file, cache_file) - - # 🌟 新增：寫入 Metadata (版本與掃描時間) - if "__metadata__" not in cache_data: - cache_data["__metadata__"] = {} - - if cmts_version != "unknown": - cache_data["__metadata__"]["cmts_version"] = cmts_version - - cache_data["__metadata__"]["last_scanned"] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) - + db_success = False current_ts = int(time.time()) - for p in batch: - if p in result_data: - result_data[p]["updated_at"] = current_ts - cache_data[p] = result_data[p] + formatted_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + + # 1. 嘗試寫入資料庫 + if USE_DB: + try: + # 寫入 metadata + if cmts_version != "unknown": + await database.upsert_device_status(host, config_type, {"cmts_version": cmts_version, "last_scanned": formatted_time}) + else: + await database.upsert_device_status(host, config_type, {"last_scanned": formatted_time}) + + # 寫入 options + db_write_count = 0 + for p in batch: + if p in result_data: + result_data[p]["updated_at"] = current_ts + success = await database.upsert_leaf_option(host, config_type, p, result_data[p]) + if success: + db_write_count += 1 + + if db_write_count > 0: + db_success = True + print(f"💾 [DB] 第 {batch_idx + 1}/{len(batches)} 批次已非同步寫入資料庫！") + else: + print("⚠️ [Fallback] 資料庫寫入 0 筆，退回寫入 JSON 快取...") + except Exception as e: + print(f"⚠️ [Fallback] 資料庫寫入發生例外: {e}，自動切換至 JSON 快取寫入...") + + # 2. 如果 DB 寫入失敗或 USE_DB=False，則 Fallback 寫入 JSON + if not db_success: + # 🌟 動態決定快取檔名 (加入 IP 隔離) + safe_host = host.replace(".", "_") + cache_file = f"{safe_host}_{config_type}_cache.json" + + def read_json_from_file(filepath): + if os.path.exists(filepath): + with open(filepath, "r", encoding="utf-8") as f: + return json.load(f) + return {} - # ========================================== - # 🌟 優化 2-B：將「寫入」新快取檔丟到背景執行緒 - # ========================================== - def write_json_to_file(filepath, data): - with open(filepath, "w", encoding="utf-8") as f: - json.dump(data, f, indent=4, ensure_ascii=False) - - await asyncio.to_thread(write_json_to_file, cache_file, cache_data) - - print(f"💾 [Debug] 第 {batch_idx + 1}/{len(batches)} 批次已非同步寫入快取檔！(版本: {cmts_version}, 檔案: {cache_file})") + cache_data = await asyncio.to_thread(read_json_from_file, cache_file) + + # 🌟 新增：寫入 Metadata (版本與掃描時間) + if "__metadata__" not in cache_data: + cache_data["__metadata__"] = {} + + if cmts_version != "unknown": + cache_data["__metadata__"]["cmts_version"] = cmts_version + + cache_data["__metadata__"]["last_scanned"] = formatted_time + + for p in batch: + if p in result_data: + result_data[p]["updated_at"] = current_ts + cache_data[p] = result_data[p] + + def write_json_to_file(filepath, data): + with open(filepath, "w", encoding="utf-8") as f: + json.dump(data, f, indent=4, ensure_ascii=False) + + await asyncio.to_thread(write_json_to_file, cache_file, cache_data) + print(f"💾 [JSON] 第 {batch_idx + 1}/{len(batches)} 批次已非同步寫入快取檔！(版本: {cmts_version}, 檔案: {cache_file})") + except Exception as e: - print(f"⚠️ 提早寫入快取失敗: {e}") + print(f"⚠️ 寫入快取失敗 (DB 與 JSON 皆失敗): {e}") if batch_idx < len(batches) - 1: await asyncio.sleep(2) diff --git a/database.py b/database.py new file mode 100644 index 0000000..3cac799 --- /dev/null +++ b/database.py @@ -0,0 +1,243 @@ +import asyncpg +import json +import logging +from typing import Dict, List, Optional, Any + +# ========================================== +# 💡 PostgreSQL 連線與操作 (高可用性版) +# ========================================== + +logger = logging.getLogger(__name__) + +# DB 設定 (從您的 init_db.py 中提取) +DB_CONFIG = { + "database": "cmts_nms", + "user": "swpa", + "password": "swpa4920", + "host": "127.0.0.1", + "port": "5432" +} + +_pool: Optional[asyncpg.Pool] = None + +async def init_db_pool(): + """初始化非同步連線池""" + global _pool + try: + if _pool is None: + _pool = await asyncpg.create_pool( + database=DB_CONFIG["database"], + user=DB_CONFIG["user"], + password=DB_CONFIG["password"], + host=DB_CONFIG["host"], + port=int(DB_CONFIG["port"]), + min_size=1, + max_size=10 + ) + logger.info("✅ PostgreSQL Connection Pool Initialized.") + except Exception as e: + logger.error(f"❌ Failed to initialize DB Pool: {e}") + +async def close_db_pool(): + """關閉非同步連線池""" + global _pool + if _pool: + await _pool.close() + _pool = None + logger.info("🔌 PostgreSQL Connection Pool Closed.") + +async def get_pool() -> Optional[asyncpg.Pool]: + if _pool is None: + await init_db_pool() + return _pool + +# ------------------------------------------ +# CRUD Functions for cmts_options +# ------------------------------------------ + +async def upsert_leaf_option(host: str, config_type: str, path: str, data: dict) -> bool: + """將選項寫入或更新至資料庫""" + pool = await get_pool() + if not pool: + return False + + query = """ + INSERT INTO cmts_options (host, config_type, path, data) + VALUES ($1, $2, $3, $4) + ON CONFLICT (host, config_type, path) + DO UPDATE SET data = EXCLUDED.data, updated_at = CURRENT_TIMESTAMP; + """ + try: + async with pool.acquire() as conn: + await conn.execute(query, host, config_type, path, json.dumps(data)) + return True + except asyncpg.PostgresError as e: + logger.error(f"❌ DB Error (upsert_leaf_option): {e}") + return False + except Exception as e: + logger.error(f"❌ Unknown Error (upsert_leaf_option): {e}") + return False + +async def get_all_leaf_options(host: str, config_type: str) -> Optional[Dict[str, Any]]: + """取得特定 host 與 config_type 的所有選項""" + pool = await get_pool() + if not pool: + return None + + query = """ + SELECT path, data FROM cmts_options + WHERE host = $1 AND config_type = $2; + """ + try: + async with pool.acquire() as conn: + records = await conn.fetch(query, host, config_type) + result = {} + for record in records: + # asyncpg returns strings for JSON if not explicitly configured with type mapping, + # but usually it's fine to just json.loads it. + data_val = record['data'] + if isinstance(data_val, str): + result[record['path']] = json.loads(data_val) + else: + result[record['path']] = data_val + return result + except asyncpg.PostgresError as e: + logger.error(f"❌ DB Error (get_all_leaf_options): {e}") + return None + except Exception as e: + logger.error(f"❌ Unknown Error (get_all_leaf_options): {e}") + return None + +async def delete_leaf_options(host: str, config_type: str, paths: List[str]) -> int: + """刪除特定路徑的選項快取""" + pool = await get_pool() + if not pool or not paths: + return -1 + + query = """ + DELETE FROM cmts_options + WHERE host = $1 AND config_type = $2 AND path = ANY($3); + """ + try: + async with pool.acquire() as conn: + # Execute returns the command tag, e.g., 'DELETE 5' + status = await conn.execute(query, host, config_type, paths) + deleted_count = int(status.split()[-1]) + return deleted_count + except asyncpg.PostgresError as e: + logger.error(f"❌ DB Error (delete_leaf_options): {e}") + return -1 + except Exception as e: + logger.error(f"❌ Unknown Error (delete_leaf_options): {e}") + return -1 + +# ------------------------------------------ +# CRUD Functions for device_status +# ------------------------------------------ + +async def upsert_device_status(host: str, config_type: str, metadata: dict) -> bool: + """更新設備的 metadata""" + pool = await get_pool() + if not pool: + return False + + # Extract known fields + cmts_version = metadata.get("cmts_version", "unknown") + last_scanned = metadata.get("last_scanned", None) + + query = """ + INSERT INTO device_status (host, config_type, cmts_version, last_scanned) + VALUES ($1, $2, $3, $4) + ON CONFLICT (host, config_type) + DO UPDATE SET + cmts_version = EXCLUDED.cmts_version, + last_scanned = EXCLUDED.last_scanned, + updated_at = CURRENT_TIMESTAMP; + """ + try: + async with pool.acquire() as conn: + await conn.execute(query, host, config_type, cmts_version, last_scanned) + return True + except asyncpg.PostgresError as e: + logger.error(f"❌ DB Error (upsert_device_status): {e}") + return False + except Exception as e: + logger.error(f"❌ Unknown Error (upsert_device_status): {e}") + return False + +async def get_device_status(host: str, config_type: str) -> Optional[Dict[str, Any]]: + """獲取設備 metadata""" + pool = await get_pool() + if not pool: + return None + + query = """ + SELECT cmts_version, last_scanned FROM device_status + WHERE host = $1 AND config_type = $2; + """ + try: + async with pool.acquire() as conn: + record = await conn.fetchrow(query, host, config_type) + if record: + return { + "cmts_version": record["cmts_version"], + "last_scanned": record["last_scanned"] + } + return None + except asyncpg.PostgresError as e: + logger.error(f"❌ DB Error (get_device_status): {e}") + return None + except Exception as e: + logger.error(f"❌ Unknown Error (get_device_status): {e}") + return None + +# ------------------------------------------ +# CRUD Functions for system_filters (Tree Filters) +# ------------------------------------------ +async def upsert_tree_filters(config_type: str, hidden_keys: List[str]) -> bool: + """寫入全域的樹狀圖隱藏節點名單""" + pool = await get_pool() + if not pool: + return False + + # Use 'global' as a dummy host to keep schema simple if needed, + # but a dedicated system_filters table is better. + query = """ + INSERT INTO system_filters (config_type, hidden_keys) + VALUES ($1, $2) + ON CONFLICT (config_type) + DO UPDATE SET hidden_keys = EXCLUDED.hidden_keys, updated_at = CURRENT_TIMESTAMP; + """ + try: + async with pool.acquire() as conn: + await conn.execute(query, config_type, hidden_keys) + return True + except asyncpg.PostgresError as e: + logger.error(f"❌ DB Error (upsert_tree_filters): {e}") + return False + except Exception as e: + logger.error(f"❌ Unknown Error (upsert_tree_filters): {e}") + return False + +async def get_tree_filters(config_type: str) -> Optional[List[str]]: + """讀取全域的樹狀圖隱藏節點名單""" + pool = await get_pool() + if not pool: + return None + + query = """ + SELECT hidden_keys FROM system_filters + WHERE config_type = $1; + """ + try: + async with pool.acquire() as conn: + record = await conn.fetchrow(query, config_type) + if record: + return record["hidden_keys"] + return [] + except asyncpg.PostgresError as e: + logger.error(f"❌ DB Error (get_tree_filters): {e}") + return None + except Exception as e: + logger.error(f"❌ Unknown Error (get_tree_filters): {e}") + return None diff --git a/init_db.py b/init_db.py index e0e8235..d732216 100644 --- a/init_db.py +++ b/init_db.py @@ -1,66 +1,71 @@ -import psycopg2 -from psycopg2.extras import RealDictCursor +import asyncio +import asyncpg +import logging -# 💡 資料庫連線設定 (請確認密碼與步驟一設定的相同) +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +# DB 設定 DB_CONFIG = { - "dbname": "cmts_nms", + "database": "cmts_nms", "user": "swpa", - "password": "swpa4920", # 替換為您的密碼 - "host": "127.0.0.1", # 如果 DB 在同一台機器上 + "password": "swpa4920", + "host": "127.0.0.1", "port": "5432" } -def init_database(): +async def init_database(): try: - # 建立連線 - print("🔄 正在連線到 PostgreSQL...") - conn = psycopg2.connect(**DB_CONFIG) - cursor = conn.cursor() + logger.info("🔄 正在連線到 PostgreSQL (asyncpg)...") + conn = await asyncpg.connect( + database=DB_CONFIG["database"], + user=DB_CONFIG["user"], + password=DB_CONFIG["password"], + host=DB_CONFIG["host"], + port=int(DB_CONFIG["port"]) + ) - # 1. 建立 CLI 字典表 (Global-Ready Schema) - print("🛠️ 正在建立 cli_schema_dictionary 資料表...") - cursor.execute(""" - CREATE TABLE IF NOT EXISTS cli_schema_dictionary ( - id SERIAL PRIMARY KEY, - path VARCHAR(255) UNIQUE NOT NULL, - node_type VARCHAR(50) DEFAULT 'leaf', - options JSONB DEFAULT '[]'::jsonb, - description TEXT, - updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + # 1. 建立選項快取表 cmts_options + logger.info("🛠️ 正在建立 cmts_options 資料表...") + await conn.execute(""" + CREATE TABLE IF NOT EXISTS cmts_options ( + host VARCHAR(255) NOT NULL, + config_type VARCHAR(50) NOT NULL, + path VARCHAR(500) NOT NULL, + data JSONB NOT NULL, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (host, config_type, path) ); - CREATE INDEX IF NOT EXISTS idx_cli_schema_path ON cli_schema_dictionary(path); """) - # 2. 建立系統狀態表 - print("🛠️ 正在建立 system_metadata 資料表...") - cursor.execute(""" - CREATE TABLE IF NOT EXISTS system_metadata ( - key VARCHAR(50) PRIMARY KEY, - value VARCHAR(255) NOT NULL, + # 2. 建立設備狀態表 device_status + logger.info("🛠️ 正在建立 device_status 資料表...") + await conn.execute(""" + CREATE TABLE IF NOT EXISTS device_status ( + host VARCHAR(255) NOT NULL, + config_type VARCHAR(50) NOT NULL, + cmts_version VARCHAR(100) DEFAULT 'unknown', + last_scanned VARCHAR(100), + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (host, config_type) + ); + """) + + # 3. 建立過濾器設定表 system_filters + logger.info("🛠️ 正在建立 system_filters 資料表...") + await conn.execute(""" + CREATE TABLE IF NOT EXISTS system_filters ( + config_type VARCHAR(50) PRIMARY KEY, + hidden_keys TEXT[] DEFAULT '{}', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) - # 3. 寫入初始狀態預設值 (使用 ON CONFLICT 避免重複執行時報錯) - print("📝 寫入系統狀態預設值...") - cursor.execute(""" - INSERT INTO system_metadata (key, value) - VALUES ('last_schema_sync_time', '1970-01-01 00:00:00') - ON CONFLICT (key) DO NOTHING; - - INSERT INTO system_metadata (key, value) - VALUES ('last_schema_sync_cos_version', 'Unknown') - ON CONFLICT (key) DO NOTHING; - """) - - # 提交變更並關閉連線 - conn.commit() - cursor.close() - conn.close() - print("✅ 資料庫初始化完成！所有資料表已準備就緒。") + await conn.close() + logger.info("✅ 資料庫初始化完成！所有資料表已準備就緒。") except Exception as e: - print(f"❌ 資料庫初始化失敗: {e}") + logger.error(f"❌ 資料庫初始化失敗: {e}") if __name__ == "__main__": - init_database() + asyncio.run(init_database()) diff --git a/main.py b/main.py index f362f6f..c6dd9fa 100644 --- a/main.py +++ b/main.py @@ -3,11 +3,21 @@ import os from fastapi import FastAPI from fastapi.staticfiles import StaticFiles from fastapi.responses import HTMLResponse +from contextlib import asynccontextmanager +import database # 引入我們剛剛拆分出來的路由模組 from routers import query, config, terminal, lock, leaf_options -app = FastAPI(title="Harmonic CMTS Manager", version="2.0") +@asynccontextmanager +async def lifespan(app: FastAPI): + # Startup + await database.init_db_pool() + yield + # Shutdown + await database.close_db_pool() + +app = FastAPI(title="Harmonic CMTS Manager", version="2.0", lifespan=lifespan) # 掛載靜態檔案目錄 (對應 static/style.css 與 static/app.js) app.mount("/static", StaticFiles(directory="static"), name="static") diff --git a/requirements.txt b/requirements.txt index cb81df0..388133a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -36,3 +36,4 @@ uvicorn==0.46.0 uvloop==0.22.1 watchfiles==1.1.1 websockets==16.0 +asyncpg==0.30.0 diff --git a/routers/config.py b/routers/config.py index 6c73529..9e8bd6f 100644 --- a/routers/config.py +++ b/routers/config.py @@ -2,25 +2,36 @@ import re import json import os +import database from fastapi import APIRouter, HTTPException from pydantic import BaseModel from typing import List from netmiko import ConnectHandler # 🌟 移除了舊的 load_settings, save_settings，改用專屬的雙軌機制 -from shared import CMTS_DEVICE, cmts_config_lock, parse_cli_to_tree, deep_split_tree +from shared import CMTS_DEVICE, cmts_config_lock, parse_cli_to_tree, deep_split_tree, USE_DB from collections import defaultdict router = APIRouter() # ========================================== -# 🌟 雙軌過濾器檔案讀寫輔助函數 +# 🌟 雙軌過濾器檔案讀寫輔助函數 (加入高可用性 Fallback) # ========================================== def get_filter_file_path(config_type: str) -> str: # 確保檔名安全，只允許 'running' 或 'full' safe_type = "full" if config_type == "full" else "running" return f"filters_{safe_type}.json" -def load_tree_filters(config_type: str) -> list: +async def load_tree_filters(config_type: str) -> list: + if USE_DB: + try: + db_filters = await database.get_tree_filters(config_type) + if db_filters is not None: + return db_filters + else: + print("⚠️ [Fallback] 資料庫無法取得過濾器，自動切換至 JSON 快取讀取...") + except Exception as e: + print(f"⚠️ [Fallback] 資料庫讀取過濾器發生例外: {e}，自動切換至 JSON 快取讀取...") + file_path = get_filter_file_path(config_type) if os.path.exists(file_path): try: @@ -32,13 +43,24 @@ def load_tree_filters(config_type: str) -> list: return [] return [] -def save_tree_filters(config_type: str, hidden_keys: list): - file_path = get_filter_file_path(config_type) - try: - with open(file_path, 'w', encoding='utf-8') as f: - json.dump({"hidden_keys": hidden_keys}, f, ensure_ascii=False, indent=4) - except Exception as e: - print(f"儲存過濾器檔案失敗: {e}") +async def save_tree_filters(config_type: str, hidden_keys: list): + db_success = False + if USE_DB: + try: + if await database.upsert_tree_filters(config_type, hidden_keys): + db_success = True + else: + print("⚠️ [Fallback] 資料庫儲存過濾器失敗，自動切換至 JSON 快取寫入...") + except Exception as e: + print(f"⚠️ [Fallback] 資料庫寫入過濾器發生例外: {e}，自動切換至 JSON 快取寫入...") + + if not db_success: + file_path = get_filter_file_path(config_type) + try: + with open(file_path, 'w', encoding='utf-8') as f: + json.dump({"hidden_keys": hidden_keys}, f, ensure_ascii=False, indent=4) + except Exception as e: + print(f"儲存過濾器檔案失敗: {e}") # ========================================== @@ -248,7 +270,7 @@ async def get_full_config(host: str, username: str, password: str = "", skip_fil # 🌟 深度過濾攔截器：改用雙軌過濾器讀取邏輯 # ========================================== if not skip_filter: - hidden_keys = load_tree_filters(config_type) + hidden_keys = await load_tree_filters(config_type) for path in hidden_keys: keys = path.split('::') @@ -280,12 +302,12 @@ class SettingsRequest(BaseModel): async def get_tree_filters(config_type: str = "running"): """獲取目前的樹狀圖隱藏名單""" # 🌟 根據 config_type 讀取對應的 JSON - hidden_keys = load_tree_filters(config_type) + hidden_keys = await load_tree_filters(config_type) return {"status": "success", "data": hidden_keys} @router.post("/settings/tree-filters") async def update_tree_filters(req: SettingsRequest): """更新樹狀圖隱藏名單並存檔""" # 🌟 根據 config_type 寫入對應的 JSON - save_tree_filters(req.config_type, req.hidden_keys) + await save_tree_filters(req.config_type, req.hidden_keys) return {"status": "success", "message": f"系統設定 ({req.config_type}) 已更新"} diff --git a/routers/leaf_options.py b/routers/leaf_options.py index 22110ac..f29a31e 100644 --- a/routers/leaf_options.py +++ b/routers/leaf_options.py @@ -4,9 +4,9 @@ from fastapi.responses import StreamingResponse from pydantic import BaseModel from typing import List import json, os, asyncio, re - +import database from cmts_scraper import sync_cmts_leaves_async -from shared import CMTS_DEVICE +from shared import CMTS_DEVICE, USE_DB router = APIRouter(tags=["Options"]) @@ -84,6 +84,21 @@ class SyncOptionsRequest(BaseModel): @router.get("/cmts-leaf-options") async def get_leaf_options(host: str, config_type: str = "running"): + if USE_DB: + try: + db_options = await database.get_all_leaf_options(host, config_type) + if db_options is not None: + # 取得 metadata + metadata = await database.get_device_status(host, config_type) + if metadata: + db_options["__metadata__"] = metadata + return db_options + else: + # 若回傳 None 表示 DB 連線異常，執行 Fallback + print("⚠️ [Fallback] 資料庫無法取得資料，自動切換至 JSON 快取讀取...") + except Exception as e: + print(f"⚠️ [Fallback] 資料庫讀取發生例外: {e}，自動切換至 JSON 快取讀取...") + cache_file = get_cache_file(host, config_type) if not os.path.exists(cache_file): return {} @@ -157,25 +172,45 @@ async def sync_leaf_options(request: SyncOptionsRequest, background_tasks: Backg @router.post("/clear_cache") async def clear_specific_cache(host: str, config_type: str = "running", paths_to_clear: list = Body(...)): cleared_count = 0 + + # 1. 嘗試清除 DB + if USE_DB: + try: + db_cleared = await database.delete_leaf_options(host, config_type, paths_to_clear) + if db_cleared >= 0: + cleared_count = db_cleared + else: + print("⚠️ [Fallback] 資料庫清除失敗，自動切換至 JSON 快取清除...") + except Exception as e: + print(f"⚠️ [Fallback] 資料庫清除發生例外: {e}，自動切換至 JSON 快取清除...") + + # 2. 清除 JSON (如果 DB 沒清掉，或 USE_DB=False，或者是連同舊檔一起清確保乾淨) cache_file = get_cache_file(host, config_type) if not os.path.exists(cache_file): + # 如果是走 DB，且有清掉，回傳 DB 的結果 + if USE_DB and cleared_count > 0: + return {"status": "success", "cleared_count": cleared_count} return {"status": "success", "cleared_count": 0, "message": "快取檔案不存在"} try: with open(cache_file, "r", encoding="utf-8") as f: cache_data = json.load(f) + json_cleared_count = 0 for path in paths_to_clear: if path in cache_data: del cache_data[path] - cleared_count += 1 + json_cleared_count += 1 - if cleared_count > 0: + if json_cleared_count > 0: with open(cache_file, "w", encoding="utf-8") as f: json.dump(cache_data, f, ensure_ascii=False, indent=4) - return {"status": "success", "cleared_count": cleared_count} + # 回傳較大的那個數值 + final_count = max(cleared_count, json_cleared_count) + return {"status": "success", "cleared_count": final_count} except Exception as e: - return {"status": "error", "message": f"清除快取時發生錯誤: {str(e)}"} + # 確保回傳標準 JSON 格式 + return {"status": "error", "message": f"資料庫連線與快取存取皆失敗: {str(e)}"} diff --git a/shared.py b/shared.py index 7e6f82f..f1ee0c4 100644 --- a/shared.py +++ b/shared.py @@ -6,6 +6,7 @@ import re from collections import defaultdict DEBUG_MODE = False +USE_DB = True # PostgreSQL Feature Toggle def debug_print(msg: str): if DEBUG_MODE: