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
133 lines
4.3 KiB
Python
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;"
|
|
)
|