Files
Timmy b34c503826 Add admin commands for database operations
Features:
- Add admin command group for database operations
- admin list-users: List all users from SQLite/PostgreSQL/MySQL
- admin user-notes: List notes for specific user
- Add CodiMDDatabase class for direct DB access
2026-04-07 12:22:24 +08:00

133 lines
4.3 KiB
Python

"""
User operations for CodiMD CLI.
"""
import json
import re
from typing import Dict, Any, Optional
from .client import CodiMDClient
class UserManager:
"""Manager for user operations."""
def __init__(self, client: CodiMDClient):
self.client = client
def login(self, email: str, password: str) -> Dict[str, Any]:
"""Login with email and password.
Returns the session cookies and user info.
"""
# First, get the login page to obtain CSRF token if needed
login_page = self.client.get("/login")
# Try to extract CSRF token from the page
csrf_token = None
csrf_match = re.search(r'name="csrf".*?value="([^"]+)"', login_page.text)
if csrf_match:
csrf_token = csrf_match.group(1)
# Prepare login data
login_data = {
"email": email,
"password": password,
}
if csrf_token:
login_data["_csrf"] = csrf_token
# Perform login
response = self.client.post(
"/login",
data=login_data,
allow_redirects=True
)
# Check if login was successful by trying to access /me
me_response = self.client.get("/me")
if me_response.status_code == 200:
user_data = me_response.json()
return {
"success": True,
"user": user_data,
"cookies": self.client.get_cookies()
}
# Login failed
if response.status_code == 302:
# Redirect might indicate failure or success
location = response.headers.get("Location", "")
if "login" in location.lower():
raise PermissionError("Login failed: Invalid credentials")
return {
"success": True,
"cookies": self.client.get_cookies()
}
raise PermissionError(f"Login failed: {response.status_code}")
def get_me(self) -> Dict[str, Any]:
"""Get current user information."""
response = self.client.get("/me")
if response.status_code == 403:
raise PermissionError("Not authenticated")
if response.status_code != 200:
raise Exception(f"Failed to get user info: {response.status_code}")
return response.json()
def export_my_data(self) -> Dict[str, Any]:
"""Export all user data."""
response = self.client.post(
"/me/export",
data={},
headers={"Content-Type": "application/x-www-form-urlencoded"}
)
if response.status_code == 403:
raise PermissionError("Not authenticated")
if response.status_code != 200:
raise Exception(f"Failed to export data: {response.status_code}")
return response.json()
def delete_user(self, delete_token: Optional[str] = None) -> bool:
"""Delete current user account."""
url = f"/me/delete/{delete_token}" if delete_token else "/me/delete"
response = self.client.get(url)
if response.status_code == 403:
raise PermissionError("Not authenticated or invalid token")
if response.status_code != 200:
raise Exception(f"Failed to delete user: {response.status_code}")
return True
def get_avatar(self, username: str) -> bytes:
"""Get user avatar SVG."""
response = self.client.get(f"/user/{username}/avatar.svg")
if response.status_code != 200:
raise Exception(f"Failed to get avatar: {response.status_code}")
return response.content
def list_all_users(self) -> list:
"""List all users (admin only).
This requires direct database access as CodiMD doesn't expose
a user list API. You'll need to provide database configuration.
"""
# Since CodiMD doesn't have a built-in user list API,
# we return a note explaining this limitation
raise NotImplementedError(
"CodiMD doesn't have a built-in API to list all users. "
"This feature requires direct database access. "
"You can query the database directly:\n"
" PostgreSQL: SELECT id, email, profile FROM Users;\n"
" MySQL: SELECT id, email, profile FROM Users;\n"
" SQLite: SELECT id, email, profile FROM Users;"
)