184 lines
6.2 KiB
Python
184 lines
6.2 KiB
Python
"""
|
|
Note operations for CodiMD CLI.
|
|
"""
|
|
|
|
import json
|
|
import re
|
|
import base64
|
|
from typing import Optional, List, Dict, Any
|
|
from datetime import datetime
|
|
import uuid
|
|
|
|
from .client import CodiMDClient
|
|
|
|
|
|
class NoteManager:
|
|
"""Manager for note operations."""
|
|
|
|
def __init__(self, client: CodiMDClient):
|
|
self.client = client
|
|
|
|
@staticmethod
|
|
def encode_note_id(note_id: str) -> str:
|
|
"""Encode UUID note ID to base64url format."""
|
|
# Remove dashes and convert to bytes
|
|
hex_str = note_id.replace("-", "")
|
|
raw_bytes = bytes.fromhex(hex_str)
|
|
# Encode to base64url
|
|
return base64.urlsafe_b64encode(raw_bytes).decode("utf-8").rstrip("=")
|
|
|
|
@staticmethod
|
|
def decode_note_id(encoded_id: str) -> str:
|
|
"""Decode base64url note ID to UUID format."""
|
|
# Add padding if needed
|
|
padding = 4 - len(encoded_id) % 4
|
|
if padding != 4:
|
|
encoded_id += "=" * padding
|
|
# Decode from base64url
|
|
raw_bytes = base64.urlsafe_b64decode(encoded_id)
|
|
hex_str = raw_bytes.hex()
|
|
# Format as UUID
|
|
parts = [
|
|
hex_str[0:8], hex_str[8:12], hex_str[12:16],
|
|
hex_str[16:20], hex_str[20:32]
|
|
]
|
|
return "-".join(parts)
|
|
|
|
def list_my_notes(self) -> List[Dict[str, Any]]:
|
|
"""List notes owned by current user."""
|
|
response = self.client.get("/api/notes/myNotes")
|
|
if response.status_code == 403:
|
|
raise PermissionError("Not authenticated or not authorized")
|
|
if response.status_code != 200:
|
|
raise Exception(f"Failed to list notes: {response.status_code}")
|
|
|
|
data = response.json()
|
|
return data.get("myNotes", [])
|
|
|
|
def get_note(self, note_id: str) -> Dict[str, Any]:
|
|
"""Get note content and metadata."""
|
|
response = self.client.get(f"/{note_id}")
|
|
if response.status_code == 404:
|
|
raise FileNotFoundError(f"Note not found: {note_id}")
|
|
if response.status_code == 403:
|
|
raise PermissionError("No permission to view this note")
|
|
if response.status_code != 200:
|
|
raise Exception(f"Failed to get note: {response.status_code}")
|
|
|
|
# Parse response for note data
|
|
# The note page contains the data in HTML, we need to extract it
|
|
# For now, return a basic structure
|
|
return {
|
|
"id": note_id,
|
|
"status": "success"
|
|
}
|
|
|
|
def create_note(self, content: str = "", alias: str = None) -> Dict[str, Any]:
|
|
"""Create a new note."""
|
|
if content:
|
|
response = self.client.post(
|
|
"/new",
|
|
data=content.encode("utf-8"),
|
|
headers={"Content-Type": "text/markdown"}
|
|
)
|
|
else:
|
|
response = self.client.get("/new")
|
|
|
|
if response.status_code != 200:
|
|
raise Exception(f"Failed to create note: {response.status_code}")
|
|
|
|
# Extract note ID from response or redirect
|
|
# The response might be a redirect to the new note
|
|
if response.history:
|
|
# Get the final URL
|
|
final_url = response.url
|
|
note_id = final_url.rstrip("/").split("/")[-1]
|
|
else:
|
|
# Parse response to find note ID
|
|
note_id = self._extract_note_id_from_response(response)
|
|
|
|
return {
|
|
"id": note_id,
|
|
"url": f"{self.client.server}/{note_id}",
|
|
"status": "created"
|
|
}
|
|
|
|
def _extract_note_id_from_response(self, response) -> str:
|
|
"""Extract note ID from response."""
|
|
# Try to extract from JSON response
|
|
try:
|
|
data = response.json()
|
|
return data.get("id") or data.get("noteId")
|
|
except:
|
|
pass
|
|
|
|
# Extract from HTML or use a default
|
|
match = re.search(r'/([a-zA-Z0-9_-]+)', response.url or "")
|
|
if match:
|
|
return match.group(1)
|
|
|
|
return "unknown"
|
|
|
|
def update_note(self, note_id: str, content: str) -> Dict[str, Any]:
|
|
"""Update note content."""
|
|
response = self.client.put(
|
|
f"/api/notes/{note_id}",
|
|
json={"content": content},
|
|
headers={"Content-Type": "application/json"}
|
|
)
|
|
|
|
if response.status_code == 403:
|
|
raise PermissionError("No permission to edit this note")
|
|
if response.status_code == 404:
|
|
raise FileNotFoundError(f"Note not found: {note_id}")
|
|
if response.status_code != 200:
|
|
raise Exception(f"Failed to update note: {response.status_code}")
|
|
|
|
return response.json()
|
|
|
|
def delete_note(self, note_id: str) -> bool:
|
|
"""Delete a note."""
|
|
response = self.client.delete(f"/api/notes/{note_id}")
|
|
|
|
if response.status_code == 403:
|
|
raise PermissionError("No permission to delete this note")
|
|
if response.status_code == 404:
|
|
raise FileNotFoundError(f"Note not found: {note_id}")
|
|
if response.status_code != 200:
|
|
raise Exception(f"Failed to delete note: {response.status_code}")
|
|
|
|
return True
|
|
|
|
def get_note_info(self, note_id: str) -> Dict[str, Any]:
|
|
"""Get note metadata without content."""
|
|
# Use the /:noteId/info endpoint
|
|
response = self.client.get(f"/{noteId}/info")
|
|
|
|
if response.status_code == 404:
|
|
raise FileNotFoundError(f"Note not found: {note_id}")
|
|
if response.status_code == 403:
|
|
raise PermissionError("No permission to view this note")
|
|
if response.status_code != 200:
|
|
raise Exception(f"Failed to get note info: {response.status_code}")
|
|
|
|
try:
|
|
return response.json()
|
|
except:
|
|
return {"id": note_id, "status": "success"}
|
|
|
|
def get_publish_link(self, note_id: str) -> Dict[str, Any]:
|
|
"""Get or create publish link for note."""
|
|
response = self.client.get(f"/{noteId}/publish")
|
|
|
|
if response.status_code == 404:
|
|
raise FileNotFoundError(f"Note not found: {note_id}")
|
|
if response.status_code != 200:
|
|
raise Exception(f"Failed to get publish link: {response.status_code}")
|
|
|
|
# Extract short ID from response
|
|
return {
|
|
"url": response.url,
|
|
"note_id": note_id,
|
|
"status": "published"
|
|
}
|