momentry_core/docs/PYTHON.md

Python 開發規範

項目	內容
建立者	Warren
建立時間	2026-03-16
文件版本	V1.0

---

版本歷史

版本	日期	目的	操作人	工具/模型
V1.0	2026-03-16	創建文件	Warren	OpenCode / MiniMax M2.5
V1.1	2026-03-21	新增 RedisPublisher API 文檔	OpenCode	-

---

概述

本文檔定義 Momentry 專案中 Python 程式碼的開發標準與最佳實踐。

---

版本管理

鎖定版本

版本	用途	路徑
Python 3.11.14	Momentry venv	/Users/accusys/momentry_core_0.1/venv/bin/python
Python 3.11.14	系統安裝	/opt/homebrew/bin/python3.11
Python 3.14.3	系統預設	/opt/homebrew/bin/python3
Python 3.9.6	系統預設 (備用)	/usr/bin/python3

版本選擇原則

• Momentry 專案：使用 venv 中的 Python 3.11.14
• 新專案：建議使用 venv 管理
• 系統工具：可使用系統預設版本

---

腳本規範

Shebang 宣告

所有 Momentry Python 腳本必須在第一行宣告明確的 Python 路徑：

#!/opt/homebrew/bin/python3.11

錯誤範例：

#!/usr/bin/env python3      # 會解析到系統預設 (3.14.3)
#!/usr/bin/python3         # 會使用系統 Python (3.9.6)

正確範例：

#!/opt/homebrew/bin/python3.11
import sys
...

檔案結構

scripts/
├── asr_processor.py          # ASR 處理腳本
├── thumbnail_extractor.py     # 縮圖提取腳本
└── new_script.py             # 新腳本模板

腳本模板

#!/opt/homebrew/bin/python3.11
"""
腳本名稱
簡短描述腳本功能

用法:
    python3.11 script.py <args>

作者: Momentry Team
版本: 1.0.0
"""

import argparse
import json
import logging
import sys
from pathlib import Path

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    stream=sys.stderr,
)
logger = logging.getLogger(__name__)


def main():
    parser = argparse.ArgumentParser(description="腳本功能描述")
    parser.add_argument("input", help="輸入檔案或參數")
    parser.add_argument("-o", "--output", default="output.json", help="輸出檔案")
    parser.add_argument("-v", "--verbose", action="store_true", help="詳細輸出")
    parser.add_argument("-c", "--count", type=int, default=10, help="數量")
    
    args = parser.parse_args()
    
    if args.verbose:
        logger.setLevel(logging.DEBUG)
    
    # 業務邏輯
    result = process_data(args.input, args.count)
    
    # 輸出 JSON結果
    print(json.dumps(result))


def process_data(input_path: str, count: int) -> dict:
    """處理資料並返回結果"""
    logger.info(f"Processing: {input_path}")
    
    # TODO: 實作業務邏輯
    
    return {
        "status": "success",
        "input": input_path,
        "count": count,
    }


if __name__ == "__main__":
    main()

---

與 Rust 整合

使用 venv (目前採用)

Momentry 使用 venv 管理 Python 環境，避免與系統其他程式衝突。

建立 venv

# 建立虛擬環境
cd /Users/accusys/momentry_core_0.1
/opt/homebrew/bin/python3.11 -m venv venv

# 啟用虛擬環境
source venv/bin/activate

# 安裝依賴
pip install -r requirements.txt

從 Rust 呼叫 venv Python

use std::path::Path;

let script_path = Path::new(env!("CARGO_MANIFEST_DIR"))
    .join("scripts")
    .join("asr_processor.py");

// 使用 venv 中的 Python
let venv_python = Path::new(env!("CARGO_MANIFEST_DIR"))
    .join("venv")
    .join("bin")
    .join("python");

let output = Command::new(venv_python)
    .arg(script_path)
    .arg(video_path)
    .output()
    .context("Failed to run processor")?;

優點：

• 專案依賴隔離
• 不同專案可使用不同 Python 版本
• 易於重現環境
• 不影響系統其他程式

---

依賴管理

venv 目錄結構

momentry_core_0.1/
├── venv/                    # 虛擬環境
│   ├── bin/
│   │   ├── python           # Python 3.11.14
│   │   ├── pip
│   │   └── ...
│   └── lib/python3.11/     # 安裝的套件
├── requirements.txt          # 依賴列表
├── scripts/                  # Python 腳本
│   ├── asr_processor.py
│   └── thumbnail_extractor.py
└── src/                     # Rust 程式碼

使用虛擬環境

# 啟用虛擬環境
source venv/bin/activate

# 安裝依賴
pip install faster-whisper

# 退出虛擬環境
deactivate

退出虛擬環境

deactivate

### 依賴列表格式

建立 `requirements.txt`：

faster-whisper>=1.0.0 ffmpeg-python>=0.2.0 Pillow>=10.0.0

### 安裝專案依賴

```bash
# 使用 python3.11 安裝
/opt/homebrew/bin/python3.11 -m pip install -r requirements.txt

---

RedisPublisher 進度發布

概述

redis_publisher.py 提供統一的進度發布介面，用於 Python 處理器向 Rust 端的 TUI 即時回報進度。

基本用法

#!/opt/homebrew/bin/python3.11
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from redis_publisher import RedisPublisher

def process_video(video_path: str, uuid: str):
    pub = RedisPublisher(uuid)
    
    pub.info("asr", "Starting ASR processing")
    pub.progress("asr", current=50, total=100, message="Processing segment")
    pub.complete("asr", "Transcription complete")

API 參考

方法	說明	範例
info(proc, msg)	發布資訊訊息	pub.info("asr", "Model loaded")
progress(proc, cur, tot, msg)	發布進度	pub.progress("asr", 50, 100, "...")
complete(proc, msg)	發布完成	pub.complete("asr", "Done")
error(proc, msg)	發布錯誤	pub.error("asr", "Failed")
warning(proc, msg)	發布警告	pub.warning("asr", "Retry...")
percentage(proc, pct, msg)	發布百分比	pub.percentage("asr", 50.5, "50%")

結構化訊息格式

from redis_publisher import MessageType, ProgressContext

# 使用 Context Manager
with ProgressContext(pub, "asr"):
    # 自動發布開始/完成/錯誤
    run_asr()

# 帶 extra 資料
pub.progress("asr", current=50, total=100, message="...",
             extra={"fps": 30.5, "model": "tiny"})

環境變數

變數	預設值	說明
REDIS_URL	redis://:accusys@localhost:6379	Redis 連線 URL
REDIS_PASSWORD	accusys	Redis 密碼

---

程式碼規範

Import 排序

# 1. 標準庫
import sys
import os
import json
import logging
from pathlib import Path
from typing import Optional

# 2. 第三方庫
import numpy as np
import pandas as pd
from faster_whisper import WhisperModel

# 3. 本地模組
from . import local_module
from ..package import module

命名規範

類型	規範	範例
模組/檔案	snake_case	asr_processor.py
類別	PascalCase	class VideoProcessor
函數/變數	snake_case	def process_video()
常量	UPPER_SNAKE_CASE	MAX_WORKERS = 4
私有成員	_leading_underscore	_private_method()

類型提示

from typing import Optional, List, Dict

def process_video(
    video_path: str,
    options: Optional[Dict[str, int]] = None,
) -> List[Dict[str, float]]:
    """處理影片並返回結果"""
    ...

錯誤處理

import logging
from pathlib import Path

logger = logging.getLogger(__name__)


def process_video(video_path: str) -> dict:
    path = Path(video_path)
    
    if not path.exists():
        logger.error(f"Video file not found: {video_path}")
        raise FileNotFoundError(f"Video not found: {video_path}")
    
    try:
        result = _do_process(path)
        logger.info(f"Processed successfully: {path}")
        return result
    except Exception as e:
        logger.exception(f"Processing failed: {e}")
        raise

日誌規範

import logging
import sys

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    stream=sys.stderr,
)
logger = logging.getLogger(__name__)

# 使用說明
logger.info("Starting process...")
logger.debug(f"Input: {input_path}")
logger.warning(f"Using fallback: {reason}")
logger.error(f"Failed: {error}")

---

測試規範

測試結構

tests/
├── __init__.py
├── test_asr_processor.py
└── test_thumbnail_extractor.py

測試範例

import pytest
from pathlib import Path
import sys

sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))

from asr_processor import run_asr


def test_asr_processor_with_valid_video(tmp_path):
    video_path = tmp_path / "test.mp4"
    output_path = tmp_path / "output.json"
    
    # 建立測試影片
    video_path.write_text("dummy")
    
    # 執行
    result = run_asr(str(video_path), str(output_path))
    
    # 斷言
    assert output_path.exists()
    assert result["segments"]


def test_asr_processor_with_invalid_video():
    with pytest.raises(FileNotFoundError):
        run_asr("/nonexistent/video.mp4", "/tmp/output.json")

執行測試

# 使用 python3.11 執行測試
/opt/homebrew/bin/python3.11 -m pytest tests/ -v

# 包含覆蓋率
/opt/homebrew/bin/python3.11 -m pytest tests/ --cov=scripts

---

監控配置

監控腳本

在 monitor/config/monitor_config.yaml 中配置：

service:
  - name: "python"
    type: "process"
    process_name: "python3"
    enabled: true
    check_interval: 60
    version_lock: "3.11.14"
    scripts:
      - "/Users/accusys/momentry_core_0.1/scripts/asr_processor.py"
      - "/Users/accusys/momentry_core_0.1/scripts/thumbnail_extractor.py"

檢查版本

# 執行 Python 監控
bash /Users/accusys/momentry_core_0.1/monitor/control/monitor_control.sh check python

# 查看資料庫記錄
psql -U accusys -h localhost -d momentry -c "SELECT * FROM python_version_baseline;"

---

CI/CD 考量

GitHub Actions 範例

name: Python Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: macos-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python 3.11
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      
      - name: Run tests
        run: |
          python -m pytest tests/ -v

---

常見問題

Q: 為什麼腳本要用 #!/opt/homebrew/bin/python3.11 而不是 #!/usr/bin/env python3？

A: #!/usr/bin/env python3 會解析 PATH 中的第一個 python3，在 macOS 上可能是：

• /opt/homebrew/bin/python3 (3.14.3)
• /usr/bin/python3 (3.9.6)

明確指定路徑可確保使用正確版本。

Q: Rust 呼叫 Python 腳本時如何確保版本正確？

A: 有三種方式：

1. Rust 程式碼中使用明確路徑：Command::new("/opt/homebrew/bin/python3.11")
2. 設定環境變數 PATH
3. 建立系統別名（不推薦，影響其他程式）

Q: 如何管理多個 Python 版本？

A: 建議使用：

• pyenv：管理多個 Python 版本
• venv：隔離專案依賴
• Docker：容器化環境

---

檢查清單

新增 Python 腳本時確認：

• 使用 #!/opt/homebrew/bin/python3.11 shebang
• 包含 docstring 說明功能
• 使用 argparse 處理命令行參數
• 使用 logging 進行日誌輸出
• 錯誤處理適當
• 類型提示完整
• 更新監控配置
• 建立測試案例

---

版本速查

版本	用途	路徑
3.11.14	Momentry venv	/Users/accusys/momentry_core_0.1/venv/bin/python
3.11.14	系統安裝	/opt/homebrew/bin/python3.11
3.14.3	系統預設	/opt/homebrew/bin/python3

---

相關文件

• NODEJS.md - Node.js 開發指南
• RUST_DEVELOPMENT.md - Rust 開發規範
• monitor_config.yaml - 監控配置
• python_monitor.sh - Python 監控腳本