Timmy
38b4bc5377
- Add MDAuth class for authentication and session management - Add MDClient class for note operations (CRUD + export/search) - Add CLI tool mdclient_cli.py for quick operations - Support for CodiMD, HedgeDoc, and HackMD services - Multi-endpoint API compatibility Co-Authored-By: Claude Sonnet 4 <noreply@anthropic.com>
302 lines
8.9 KiB
Python
302 lines
8.9 KiB
Python
"""
|
||
MDclient 客戶端模組
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Optional, Dict, List, Any
|
||
|
||
|
||
class MDClient:
|
||
"""Markdown 筆記服務客戶端類"""
|
||
|
||
def __init__(
|
||
self,
|
||
base_url: str = "https://codimd.lotimmy.com",
|
||
cookie_file: str = "key.conf",
|
||
email: Optional[str] = None,
|
||
password: Optional[str] = None
|
||
):
|
||
"""
|
||
初始化客戶端
|
||
|
||
Args:
|
||
base_url: 伺服器位址
|
||
cookie_file: Cookie 儲存檔案路徑
|
||
email: 登入信箱（可選，用於自動登入）
|
||
password: 登入密碼（可選，用於自動登入）
|
||
"""
|
||
self.base_url = base_url.rstrip("/")
|
||
self.cookie_file = Path(cookie_file)
|
||
self.email = email
|
||
self.password = password
|
||
|
||
def _load_cookie(self) -> Optional[Dict[str, str]]:
|
||
"""從檔案載入 Cookie"""
|
||
if not self.cookie_file.exists():
|
||
return None
|
||
|
||
with open(self.cookie_file, 'r', encoding='utf-8') as f:
|
||
return json.load(f)
|
||
|
||
def _request(
|
||
self,
|
||
method: str,
|
||
endpoint: str,
|
||
data: Optional[Dict] = None,
|
||
params: Optional[Dict] = None
|
||
) -> Optional[requests.Response]:
|
||
"""
|
||
傳送帶 Cookie 的請求
|
||
|
||
Args:
|
||
method: HTTP 方法 (GET, POST, etc.)
|
||
endpoint: API 端點
|
||
data: POST 資料
|
||
params: URL 參數
|
||
|
||
Returns:
|
||
回應物件，失敗時返回 None
|
||
"""
|
||
cookies = self._load_cookie()
|
||
|
||
if not cookies:
|
||
print(f"✗ 找不到 Cookie 檔案: {self.cookie_file}")
|
||
print("請先使用 MDAuth 登入")
|
||
return None
|
||
|
||
headers = {
|
||
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
|
||
}
|
||
|
||
url = f"{self.base_url}{endpoint}"
|
||
|
||
try:
|
||
if method == 'GET':
|
||
response = requests.get(url, cookies=cookies, headers=headers, params=params, timeout=10)
|
||
elif method == 'POST':
|
||
headers['Content-Type'] = 'application/json'
|
||
response = requests.post(url, cookies=cookies, headers=headers, json=data, timeout=10)
|
||
elif method == 'DELETE':
|
||
response = requests.delete(url, cookies=cookies, headers=headers, timeout=10)
|
||
else:
|
||
raise ValueError(f"不支援的 HTTP 方法: {method}")
|
||
|
||
return response
|
||
except requests.RequestException as e:
|
||
print(f"✗ 請求失敗: {e}")
|
||
return None
|
||
|
||
def get_user_info(self) -> Optional[Dict]:
|
||
"""
|
||
取得使用者資訊
|
||
|
||
Returns:
|
||
使用者資訊字典
|
||
"""
|
||
print("正在取得使用者資訊...")
|
||
|
||
endpoints = ['/api/me', '/me', '/api/user/me']
|
||
|
||
for endpoint in endpoints:
|
||
response = self._request('GET', endpoint)
|
||
if response and response.status_code == 200:
|
||
try:
|
||
data = response.json()
|
||
print("✓ 取得使用者資訊成功")
|
||
return data
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
print("✗ 無法取得使用者資訊")
|
||
return None
|
||
|
||
def list_notes(self) -> Optional[List[Dict]]:
|
||
"""
|
||
列出所有筆記
|
||
|
||
Returns:
|
||
筆記列表
|
||
"""
|
||
print("正在取得筆記列表...")
|
||
|
||
endpoints = ['/api/me/notes', '/api/notes', '/history', '/me/notes']
|
||
|
||
for endpoint in endpoints:
|
||
response = self._request('GET', endpoint)
|
||
if response and response.status_code == 200:
|
||
try:
|
||
data = response.json()
|
||
print(f"✓ 成功取得筆記 (端點: {endpoint})")
|
||
|
||
# 處理不同的回應格式
|
||
if isinstance(data, list):
|
||
return data
|
||
elif isinstance(data, dict) and 'history' in data:
|
||
return data['history']
|
||
elif isinstance(data, dict) and 'notes' in data:
|
||
return data['notes']
|
||
|
||
return data
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
print("✗ 無法取得筆記列表")
|
||
return None
|
||
|
||
def get_note(self, note_id: str) -> Optional[Dict]:
|
||
"""
|
||
取得筆記內容
|
||
|
||
Args:
|
||
note_id: 筆記 ID
|
||
|
||
Returns:
|
||
筆記內容字典
|
||
"""
|
||
print(f"正在取得筆記: {note_id}")
|
||
|
||
# 嘗試不同的 API 端點
|
||
endpoints = [
|
||
f"/api/notes/{note_id}",
|
||
f"/api/notes/{note_id}/content",
|
||
f"/{note_id}/download",
|
||
f"/{note_id}",
|
||
]
|
||
|
||
for endpoint in endpoints:
|
||
response = self._request('GET', endpoint)
|
||
if response and response.status_code == 200:
|
||
try:
|
||
# 如果是直接訪問筆記頁面，可能需要解析 HTML
|
||
content_type = response.headers.get('content-type', '')
|
||
if 'text/html' in content_type:
|
||
# 這是 HTML 頁面，不是 API 回應
|
||
continue
|
||
|
||
data = response.json()
|
||
print(f"✓ 取得筆記成功 (端點: {endpoint})")
|
||
return data
|
||
except json.JSONDecodeError:
|
||
# 如果不是 JSON，可能是純文字內容
|
||
if response.text:
|
||
print(f"✓ 取得筆記內容 (端點: {endpoint})")
|
||
return {
|
||
'id': note_id,
|
||
'content': response.text,
|
||
'title': note_id
|
||
}
|
||
|
||
print("✗ 無法取得筆記內容")
|
||
return None
|
||
|
||
def create_note(self, title: str, content: str = "") -> Optional[Dict]:
|
||
"""
|
||
建立新筆記
|
||
|
||
Args:
|
||
title: 筆記標題
|
||
content: 筆記內容（Markdown 格式）
|
||
|
||
Returns:
|
||
建立的筆記資訊
|
||
"""
|
||
print(f"正在建立筆記: {title}")
|
||
|
||
data = {
|
||
'title': title,
|
||
'content': content,
|
||
}
|
||
|
||
response = self._request('POST', '/api/notes', data=data)
|
||
|
||
if response and response.status_code in [200, 201]:
|
||
try:
|
||
result = response.json()
|
||
print("✓ 筆記建立成功")
|
||
return result
|
||
except json.JSONDecodeError:
|
||
print("✗ 無法解析回應")
|
||
else:
|
||
status = response.status_code if response else "無"
|
||
print(f"✗ 建立筆記失敗 (狀態碼: {status})")
|
||
|
||
return None
|
||
|
||
def delete_note(self, note_id: str) -> bool:
|
||
"""
|
||
刪除筆記
|
||
|
||
Args:
|
||
note_id: 筆記 ID
|
||
|
||
Returns:
|
||
是否刪除成功
|
||
"""
|
||
print(f"正在刪除筆記: {note_id}")
|
||
response = self._request('DELETE', f"/api/notes/{note_id}")
|
||
|
||
if response and response.status_code in [200, 204]:
|
||
print("✓ 筆記刪除成功")
|
||
return True
|
||
else:
|
||
status = response.status_code if response else "無"
|
||
print(f"✗ 刪除筆記失敗 (狀態碼: {status})")
|
||
return False
|
||
|
||
def export_note(self, note_id: str, format: str = "md") -> Optional[str]:
|
||
"""
|
||
匯出筆記
|
||
|
||
Args:
|
||
note_id: 筆記 ID
|
||
format: 匯出格式 (md, pdf, html, slides)
|
||
|
||
Returns:
|
||
匯出的內容
|
||
"""
|
||
print(f"正在匯出筆記: {note_id} (格式: {format})")
|
||
|
||
endpoints = [
|
||
f"/api/notes/{note_id}/export?format={format}",
|
||
f"/{note_id}/download",
|
||
f"/{note_id}/export/{format}",
|
||
]
|
||
|
||
for endpoint in endpoints:
|
||
response = self._request('GET', endpoint)
|
||
if response and response.status_code == 200:
|
||
print("✓ 筆記匯出成功")
|
||
return response.text
|
||
|
||
print("✗ 匯出筆記失敗")
|
||
return None
|
||
|
||
def search_notes(self, query: str) -> Optional[List[Dict]]:
|
||
"""
|
||
搜尋筆記
|
||
|
||
Args:
|
||
query: 搜尋關鍵字
|
||
|
||
Returns:
|
||
符合的筆記列表
|
||
"""
|
||
print(f"正在搜尋筆記: {query}")
|
||
|
||
response = self._request('GET', '/api/search', params={'q': query})
|
||
|
||
if response and response.status_code == 200:
|
||
try:
|
||
data = response.json()
|
||
print("✓ 搜尋成功")
|
||
return data
|
||
except json.JSONDecodeError:
|
||
print("✗ 無法解析回應")
|
||
else:
|
||
status = response.status_code if response else "無"
|
||
print(f"✗ 搜尋失敗 (狀態碼: {status})")
|
||
|
||
return None
|