zhongyou
/
scm-harmonic-cmts-admin
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: 1.PostgreSQL 高可用性架構升級, 2.導入continue.dev與Roo Code偕同開發模式。

This commit is contained in:
swpa 2026-05-20 17:19:40 +08:00
parent 289204780a
commit 3cebbb2824
12 changed files with 577 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
.clineignore Normal file
View File

@ -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/

41
.clinerules Normal file
View File

@ -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`。

41
.continue/rules/new-rule.md Normal file
View File

@ -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`。

28
PROJECT_STATE.md Normal file
View File

@ -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 預覽確認視窗。

51
cmts_scraper.py
View File

@ -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,15 +260,44 @@ 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 隔離)
db_success = False
current_ts = int(time.time())
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"
# ==========================================
# 🌟 優化 2-A：將「讀取」舊快取檔丟到背景執行緒
# ==========================================
def read_json_from_file(filepath):
if os.path.exists(filepath):
with open(filepath, "r", encoding="utf-8") as f:
@ -282,26 +313,22 @@ async def sync_cmts_leaves_async(host, username, password, leaf_paths: list, con
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())
cache_data["__metadata__"]["last_scanned"] = formatted_time
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]
# ==========================================
# 🌟 優化 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"💾 [JSON] 第 {batch_idx + 1}/{len(batches)} 批次已非同步寫入快取檔！(版本: {cmts_version}, 檔案: {cache_file})")
print(f"💾 [Debug] 第 {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)

243
database.py Normal file
View File

@ -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

99
init_db.py
View File

@ -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())

12
main.py
View File

@ -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")

1
requirements.txt
View File

@ -36,3 +36,4 @@ uvicorn==0.46.0
uvloop==0.22.1
watchfiles==1.1.1
websockets==16.0
asyncpg==0.30.0

36
routers/config.py
View File

@ -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,7 +43,18 @@ def load_tree_filters(config_type: str) -> list:
return []
return []
def save_tree_filters(config_type: str, hidden_keys: list):
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:
@ -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}) 已更新"}

47
routers/leaf_options.py
View File

@ -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)}"}

1
shared.py
View File

@ -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: