Ten31-Portal/backend/ten31portal/cli.py

"""CLI commands for Ten31Portal (also driven by the StartOS service Actions)."""

import argparse
import os
import sys

from sqlmodel import Session, select

from ten31portal import config
from ten31portal.auth import hash_password
from ten31portal.database import engine
from ten31portal.db_init import run_migrations
from ten31portal.models import (
    AuditLog, CapitalAccountStatement, Document, EntityAccess, User, UserRole,
    ValuationRound,
)


def _find_user(session: Session, username: str | None, email: str | None) -> User | None:
    """Look a user up by username (preferred) or email."""
    if username:
        u = session.exec(select(User).where(User.username == username)).first()
        if u:
            return u
    if email:
        return session.exec(select(User).where(User.email == email)).first()
    return None


def create_user(args: argparse.Namespace) -> None:
    """Create a new user account."""
    run_migrations()

    with Session(engine) as session:
        if session.exec(select(User).where(User.username == args.username)).first():
            print(f"Error: user with username {args.username} already exists.", file=sys.stderr)
            sys.exit(1)
        if args.email and session.exec(select(User).where(User.email == args.email)).first():
            print(f"Error: user with email {args.email} already exists.", file=sys.stderr)
            sys.exit(1)

        try:
            role = UserRole(args.role)
        except ValueError:
            print(f"Error: invalid role '{args.role}'. Must be one of: {', '.join(r.value for r in UserRole)}", file=sys.stderr)
            sys.exit(1)

        user = User(
            name=args.name,
            username=args.username,
            email=args.email or None,
            password_hash=hash_password(args.password),
            role=role,
            is_service_admin=bool(getattr(args, "service_admin", False)),
        )
        session.add(user)
        session.commit()
        print(f"Created user: {user.name} ({user.username}) with role {user.role.value}")


def _clear_admin_password_file() -> None:
    """Remove the recorded initial admin password (it is no longer valid once changed)."""
    try:
        os.remove(config.ADMIN_PASSWORD_FILE)
    except OSError:
        pass


def reset_password(args: argparse.Namespace) -> None:
    """Reset a user's password (and re-enable their login)."""
    run_migrations()

    with Session(engine) as session:
        user = _find_user(session, args.username, args.email)
        if user is None:
            who = args.username or args.email
            print(f"Error: no user found for '{who}'.", file=sys.stderr)
            sys.exit(1)
        user.password_hash = hash_password(args.password)
        user.login_enabled = True
        is_admin = user.is_service_admin
        session.add(user)
        session.commit()
        # Resetting the built-in admin invalidates the initial password recorded on first boot.
        if is_admin:
            _clear_admin_password_file()
        print(f"Password reset for {user.name} ({user.username}).")


def show_admin_password(args: argparse.Namespace) -> None:
    """Print the randomly-generated initial admin password recorded on first boot."""
    path = config.ADMIN_PASSWORD_FILE
    if not os.path.exists(path):
        print(
            "No stored initial password. It was either set explicitly at install time, or the "
            "admin password has already been changed. Use Reset Password to set a new one."
        )
        return
    with open(path) as f:
        pw = f.read().strip()
    print(pw if pw else "(the recorded initial password is empty)")


def list_users(args: argparse.Namespace) -> None:
    """Print all user accounts."""
    run_migrations()

    with Session(engine) as session:
        users = session.exec(select(User).order_by(User.role, User.name)).all()  # type: ignore[arg-type]
        if not users:
            print("No users.")
            return
        print(f"{'USERNAME':<20} {'NAME':<24} {'ROLE':<14} {'STATUS':<10} EMAIL")
        print("-" * 84)
        for u in users:
            status = "active" if u.is_active else "disabled"
            tag = "  [SERVICE ADMIN]" if u.is_service_admin else ""
            print(f"{u.username:<20} {u.name:<24} {u.role.value:<14} {status:<10} {u.email or '-'}{tag}")
        print(f"\n{len(users)} user(s). The Service Admin cannot be deleted.")


def delete_user(args: argparse.Namespace) -> None:
    """Delete a user account and its dependent rows. The Service Admin is protected."""
    run_migrations()

    from ten31portal import storage

    with Session(engine) as session:
        user = _find_user(session, args.username, args.email)
        if user is None:
            who = args.username or args.email
            print(f"Error: no user found for '{who}'.", file=sys.stderr)
            sys.exit(1)
        if user.is_service_admin:
            print(f"Error: '{user.username}' is the Service Admin and cannot be deleted.", file=sys.stderr)
            sys.exit(1)

        uid = user.id

        # Entity-access grants and capital-account statements are this user's own data.
        for acc in session.exec(select(EntityAccess).where(EntityAccess.user_id == uid)).all():
            session.delete(acc)
        for stmt in session.exec(
            select(CapitalAccountStatement).where(CapitalAccountStatement.investor_user_id == uid)
        ).all():
            session.delete(stmt)

        # Documents addressed privately to this investor are removed (file + row); documents
        # they uploaded stay, with the uploader cleared.
        for doc in session.exec(select(Document).where(Document.investor_user_id == uid)).all():
            try:
                storage.delete_file(doc.storage_path)
            except OSError as exc:
                # Don't abort the whole deletion, but surface it — a swallowed disk/permission
                # error would silently orphan the file on the data volume.
                print(f"Warning: could not delete file {doc.storage_path}: {exc}", file=sys.stderr)
            session.delete(doc)
        for doc in session.exec(select(Document).where(Document.uploaded_by == uid)).all():
            doc.uploaded_by = None
            session.add(doc)

        # Preserve history/rounds by clearing the references to this user.
        for rnd in session.exec(
            select(ValuationRound).where(
                (ValuationRound.submitted_by == uid) | (ValuationRound.approved_by == uid)
            )
        ).all():
            if rnd.submitted_by == uid:
                rnd.submitted_by = None
            if rnd.approved_by == uid:
                rnd.approved_by = None
            session.add(rnd)
        for log in session.exec(select(AuditLog).where(AuditLog.actor_user_id == uid)).all():
            log.actor_user_id = None
            session.add(log)

        # Detach any linked sub-accounts so they log in on their own again.
        for sub in session.exec(select(User).where(User.primary_account_id == uid)).all():
            sub.primary_account_id = None
            session.add(sub)

        name, username = user.name, user.username
        session.delete(user)
        session.commit()
        print(f"Deleted user {name} ({username}).")


def dedupe_holdings(args: argparse.Namespace) -> None:
    """Remove duplicate holdings/positions left by old double-imports (fixes inflated Invested)."""
    run_migrations()
    from ten31portal.models import Entity
    from ten31portal.routers.import_router import dedupe_entity

    with Session(engine) as session:
        if args.entity_id:
            entities = [e for e in [session.get(Entity, args.entity_id)] if e]
        else:
            entities = session.exec(select(Entity).order_by(Entity.name)).all()  # type: ignore[arg-type]
        total_h = total_p = 0
        for ent in entities:
            res = dedupe_entity(ent.id, session)
            if res["removed_holdings"] or res["removed_positions"]:
                print(f"{ent.name}: removed {res['removed_holdings']} holding(s), "
                      f"{res['removed_positions']} position(s)")
            total_h += res["removed_holdings"]
            total_p += res["removed_positions"]
        session.commit()
        if total_h or total_p:
            print(f"\nDone. Removed {total_h} duplicate holding(s) and {total_p} position(s).")
        else:
            print("No duplicates found — nothing to clean up.")


def reset_holdings(args: argparse.Namespace) -> None:
    """Clear a fund's holdings/positions/rounds so it can be re-imported from scratch."""
    run_migrations()
    from ten31portal.models import Entity
    from ten31portal.routers.import_router import reset_entity_holdings

    with Session(engine) as session:
        entity = None
        if args.entity_id:
            entity = session.get(Entity, args.entity_id)
        elif args.name:
            entity = session.exec(select(Entity).where(Entity.name == args.name)).first()
        if entity is None:
            print(f"Error: no fund found for '{args.name or args.entity_id}'. "
                  f"Check the exact name with list-funds.", file=sys.stderr)
            sys.exit(1)
        res = reset_entity_holdings(entity.id, session)
        session.commit()
        print(f"Cleared {entity.name}: removed {res['holdings']} holding(s), "
              f"{res['positions']} position(s), {res['rounds']} round(s). "
              f"Re-import the fund's NAV to repopulate it.")


def list_funds(args: argparse.Namespace) -> None:
    """Print every entity's id and name (so the exact name is known for reset-holdings)."""
    run_migrations()
    from ten31portal.models import Entity

    with Session(engine) as session:
        entities = session.exec(select(Entity).order_by(Entity.name)).all()  # type: ignore[arg-type]
        if not entities:
            print("No funds yet.")
            return
        for e in entities:
            print(f"[{e.id}] {e.name} ({e.type.value})")


def main() -> None:
    parser = argparse.ArgumentParser(prog="ten31portal-cli")
    sub = parser.add_subparsers(dest="command")

    create = sub.add_parser("create-user", help="Provision a new user")
    create.add_argument("--name", required=True)
    create.add_argument("--username", required=True)
    create.add_argument("--email", required=False, default=None)
    create.add_argument("--role", required=True, choices=[r.value for r in UserRole])
    create.add_argument("--password", required=True)
    create.add_argument("--service-admin", action="store_true",
                        help="Mark as the protected built-in Service Admin")

    reset = sub.add_parser("reset-password", help="Reset a user's password")
    reset.add_argument("--username", required=False, default=None)
    reset.add_argument("--email", required=False, default=None)
    reset.add_argument("--password", required=True)

    sub.add_parser("list-users", help="List all user accounts")

    sub.add_parser("show-admin-password", help="Show the initial admin password from first boot")

    delete = sub.add_parser("delete-user", help="Delete a user (not the Service Admin)")
    delete.add_argument("--username", required=False, default=None)
    delete.add_argument("--email", required=False, default=None)

    dedupe = sub.add_parser("dedupe-holdings", help="Remove duplicate holdings/positions")
    dedupe.add_argument("--entity-id", type=int, required=False, default=None,
                        help="Limit to one entity; omit to clean all")

    sub.add_parser("list-funds", help="List entities (funds/SPVs) with their ids")

    reset = sub.add_parser("reset-holdings", help="Clear a fund's holdings to re-import fresh")
    reset.add_argument("--name", required=False, default=None, help="Exact fund name")
    reset.add_argument("--entity-id", type=int, required=False, default=None)

    args = parser.parse_args()
    if args.command == "create-user":
        create_user(args)
    elif args.command == "reset-password":
        reset_password(args)
    elif args.command == "list-users":
        list_users(args)
    elif args.command == "show-admin-password":
        show_admin_password(args)
    elif args.command == "delete-user":
        delete_user(args)
    elif args.command == "dedupe-holdings":
        dedupe_holdings(args)
    elif args.command == "list-funds":
        list_funds(args)
    elif args.command == "reset-holdings":
        reset_holdings(args)
    else:
        parser.print_help()


if __name__ == "__main__":
    main()