From b34c503826fb847e53235c95b40309aa0054ea33 Mon Sep 17 00:00:00 2001 From: Timmy Date: Tue, 7 Apr 2026 12:22:24 +0800 Subject: [PATCH] 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 --- cli_anything/codimd/codimd_cli.py | 109 +++++++++++++++++++ cli_anything/codimd/core/__init__.py | 2 + cli_anything/codimd/core/db.py | 150 +++++++++++++++++++++++++++ cli_anything/codimd/core/user.py | 17 +++ 4 files changed, 278 insertions(+) create mode 100644 cli_anything/codimd/core/db.py diff --git a/cli_anything/codimd/codimd_cli.py b/cli_anything/codimd/codimd_cli.py index 990c133..ca23cd3 100644 --- a/cli_anything/codimd/codimd_cli.py +++ b/cli_anything/codimd/codimd_cli.py @@ -122,6 +122,12 @@ def config(): pass +@cli.group() +def admin(): + """Admin operations (require database access).""" + pass + + # Note commands @note.command("list") @click.option("--limit", type=int, help="Limit number of results") @@ -599,6 +605,109 @@ def config_set(key, value): click.echo(f"Set {key} = {value}") +# Admin commands +@admin.command("list-users") +@click.option("--db-type", type=click.Choice(["sqlite", "postgresql", "mysql"]), default="sqlite", help="Database type") +@click.option("--db-path", type=Path, help="SQLite database path") +@click.option("--db-host", help="Database host (for PostgreSQL/MySQL)") +@click.option("--db-name", help="Database name (for PostgreSQL/MySQL)") +@click.option("--db-user", help="Database user (for PostgreSQL/MySQL)") +@click.option("--db-password", help="Database password (for PostgreSQL/MySQL)") +@click.option("--db-port", type=int, default=5432, help="Database port (default: 5432 for PostgreSQL, 3306 for MySQL)") +@handle_error +def admin_list_users(db_type, db_path, db_host, db_name, db_user, db_password, db_port): + """List all users in the CodiMD database.""" + from cli_anything.codimd.core import CodiMDDatabase + + if db_type == "sqlite": + if not db_path: + # Try default CodiMD SQLite path + default_paths = [ + "./codimd/db.sqlite", + "./db.sqlite", + "/var/lib/codimd/db.sqlite" + ] + for path in default_paths: + if Path(path).exists(): + db_path = path + break + if not db_path: + raise ValueError( + "SQLite database path not found. " + "Please specify with --db-path or ensure db.sqlite is in current directory." + ) + users = CodiMDDatabase.list_users_sqlite(str(db_path)) + + elif db_type == "postgresql": + if not all([db_host, db_name, db_user]): + raise ValueError("--db-host, --db-name, and --db-user are required for PostgreSQL") + if not db_password: + db_password = click.prompt("Database password", hide_input=True) + users = CodiMDDatabase.list_users_postgresql(db_host, db_name, db_user, db_password, db_port) + + elif db_type == "mysql": + if not all([db_host, db_name, db_user]): + raise ValueError("--db-host, --db-name, and --db-user are required for MySQL") + if not db_password: + db_password = click.prompt("Database password", hide_input=True) + users = CodiMDDatabase.list_users_mysql(db_host, db_name, db_user, db_password, db_port) + + if ctx["json_mode"]: + output_json(format_success({"users": users, "count": len(users)})) + else: + if not users: + click.echo("No users found.") + else: + click.echo(f"Total users: {len(users)}") + for u in users: + user_id = u.get("id", "N/A")[:8] + "..." if len(u.get("id", "")) > 8 else u.get("id", "N/A") + email = u.get("email") or "N/A" + + # Get name from profile + profile = u.get("profile") + if profile: + if isinstance(profile, str): + try: + profile = json.loads(profile) + except: + profile = {} + name = profile.get("name") or profile.get("displayName") or "N/A" + else: + name = "N/A" + + created = u.get("createdAt", "N/A") + click.echo(f" {user_id} | {email:30s} | {name}") + if created != "N/A": + click.echo(f" Created: {created}") + + +@admin.command("user-notes") +@click.argument("user_id") +@click.option("--db-path", type=Path, default="./db.sqlite", help="SQLite database path") +@handle_error +def admin_user_notes(user_id, db_path): + """List all notes for a specific user.""" + from cli_anything.codimd.core import CodiMDDatabase + + if not Path(db_path).exists(): + raise FileNotFoundError(f"Database not found: {db_path}") + + notes = CodiMDDatabase.list_user_notes_sqlite(str(db_path), user_id) + + if ctx["json_mode"]: + output_json(format_success({"user_id": user_id, "notes": notes, "count": len(notes)})) + else: + if not notes: + click.echo(f"No notes found for user {user_id}") + else: + click.echo(f"Notes for user {user_id}: {len(notes)} total") + for n in notes: + note_id = n.get("shortid", n.get("id", "N/A")) + title = n.get("title") or "Untitled" + permission = n.get("permission", "N/A") + click.echo(f" {note_id} | {title[:50]} | [{permission}]") + + # REPL mode @cli.command("repl") @handle_error diff --git a/cli_anything/codimd/core/__init__.py b/cli_anything/codimd/core/__init__.py index 08a38ea..b0d021e 100644 --- a/cli_anything/codimd/core/__init__.py +++ b/cli_anything/codimd/core/__init__.py @@ -9,6 +9,7 @@ from .note import NoteManager from .user import UserManager from .export import ExportManager from .revision import RevisionManager +from .db import CodiMDDatabase __all__ = [ "Config", @@ -18,4 +19,5 @@ __all__ = [ "UserManager", "ExportManager", "RevisionManager", + "CodiMDDatabase", ] diff --git a/cli_anything/codimd/core/db.py b/cli_anything/codimd/core/db.py new file mode 100644 index 0000000..ca35557 --- /dev/null +++ b/cli_anything/codimd/core/db.py @@ -0,0 +1,150 @@ +""" +Database operations for CodiMD CLI. + +Direct database access for admin operations not exposed via API. +""" + +import json +from typing import List, Dict, Any, Optional +from pathlib import Path + + +class CodiMDDatabase: + """Direct database access for CodiMD.""" + + @staticmethod + def list_users_sqlite(db_path: str) -> List[Dict[str, Any]]: + """List users from SQLite database.""" + try: + import sqlite3 + except ImportError: + raise ImportError("sqlite3 required for SQLite access") + + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + cursor = conn.cursor() + + cursor.execute(""" + SELECT id, email, profile, createdAt, updatedAt + FROM Users + ORDER BY createdAt DESC + """) + + users = [] + for row in cursor.fetchall(): + user = dict(row) + # Parse profile JSON if it's a string + if user.get('profile') and isinstance(user['profile'], str): + try: + user['profile'] = json.loads(user['profile']) + except: + pass + users.append(user) + + conn.close() + return users + + @staticmethod + def list_users_postgresql(host: str, database: str, user: str, password: str, + port: int = 5432) -> List[Dict[str, Any]]: + """List users from PostgreSQL database.""" + try: + import psycopg2 + except ImportError: + raise ImportError("psycopg2 required for PostgreSQL access") + + conn = psycopg2.connect( + host=host, + database=database, + user=user, + password=password, + port=port + ) + cursor = conn.cursor() + + cursor.execute(""" + SELECT id, email, profile, "createdAt", "updatedAt" + FROM "Users" + ORDER BY "createdAt" DESC + """) + + users = [] + for row in cursor.fetchall(): + user = { + 'id': row[0], + 'email': row[1], + 'profile': row[2], + 'createdAt': row[3], + 'updatedAt': row[4] + } + # Parse profile JSON if it's a string + if user.get('profile') and isinstance(user['profile'], str): + try: + user['profile'] = json.loads(user['profile']) + except: + pass + users.append(user) + + conn.close() + return users + + @staticmethod + def list_users_mysql(host: str, database: str, user: str, password: str, + port: int = 3306) -> List[Dict[str, Any]]: + """List users from MySQL/MariaDB database.""" + try: + import mysql.connector + except ImportError: + raise ImportError("mysql-connector-python required for MySQL access") + + conn = mysql.connector.connect( + host=host, + database=database, + user=user, + password=password, + port=port + ) + cursor = conn.cursor(dictionary=True) + + cursor.execute(""" + SELECT id, email, profile, createdAt, updatedAt + FROM Users + ORDER BY createdAt DESC + """) + + users = cursor.fetchall() + + # Parse profile JSON + for user in users: + if user.get('profile') and isinstance(user['profile'], str): + try: + user['profile'] = json.loads(user['profile']) + except: + pass + + conn.close() + return users + + @staticmethod + def list_user_notes_sqlite(db_path: str, user_id: str) -> List[Dict[str, Any]]: + """List notes for a specific user from SQLite database.""" + try: + import sqlite3 + except ImportError: + raise ImportError("sqlite3 required for SQLite access") + + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + cursor = conn.cursor() + + cursor.execute(""" + SELECT id, shortid, alias, title, permission, + "lastchangeAt", "createdAt", viewcount + FROM Notes + WHERE "ownerId" = ? + ORDER BY "lastchangeAt" DESC + """, (user_id,)) + + notes = [dict(row) for row in cursor.fetchall()] + conn.close() + return notes diff --git a/cli_anything/codimd/core/user.py b/cli_anything/codimd/core/user.py index c552bb3..b4b7827 100644 --- a/cli_anything/codimd/core/user.py +++ b/cli_anything/codimd/core/user.py @@ -113,3 +113,20 @@ class UserManager: 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;" + )