116 lines
3.6 KiB
Python
116 lines
3.6 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
|