"""Admin API endpoints — user management.""" from datetime import datetime from flask import Blueprint, jsonify, request from flask_login import current_user from app.extensions import db from app.models.user import User from app.models.admin import Admin from app.models.user_settings import UserSettings from app.models.settings import AppSettings from app.helpers.auth import admin_required api_admin_bp = Blueprint('api_admin', __name__) @api_admin_bp.route('/users') @admin_required def list_users(): """List all users with optional filter.""" from app.helpers.asset_filter import get_allowed_markets status = request.args.get('status', 'all') query = User.query if status == 'pending': query = query.filter_by(is_approved=False, is_active=True) elif status == 'active': query = query.filter_by(is_approved=True, is_active=True) elif status == 'disabled': query = query.filter_by(is_active=False) users = query.order_by(User.created_at.desc()).all() result = [] for u in users: d = u.to_dict() d['allowed_markets'] = get_allowed_markets(u.id) result.append(d) return jsonify(result) @api_admin_bp.route('/users//approve', methods=['POST']) @admin_required def approve_user(user_id): """Approve a pending user registration.""" user = User.query.get_or_404(user_id) if user.is_approved: return jsonify({'error': 'User sudah disetujui'}), 400 user.is_approved = True user.approved_by = current_user.id user.approved_at = datetime.utcnow() db.session.commit() # Copy default fee/capital settings to new user _copy_default_settings(user.id) return jsonify({'status': 'approved', 'user': user.to_dict()}) @api_admin_bp.route('/users//disable', methods=['POST']) @admin_required def disable_user(user_id): """Disable an active user.""" user = User.query.get_or_404(user_id) user.is_active = False db.session.commit() return jsonify({'status': 'disabled', 'user': user.to_dict()}) @api_admin_bp.route('/users//enable', methods=['POST']) @admin_required def enable_user(user_id): """Re-enable a disabled user.""" user = User.query.get_or_404(user_id) user.is_active = True db.session.commit() return jsonify({'status': 'enabled', 'user': user.to_dict()}) @api_admin_bp.route('/users//allowed-markets', methods=['GET']) @admin_required def get_user_allowed_markets(user_id): """Get allowed markets for a user.""" user = User.query.get_or_404(user_id) from app.helpers.asset_filter import get_allowed_markets markets = get_allowed_markets(user_id) return jsonify({ 'user_id': user_id, 'username': user.username, 'allowed_markets': markets, }) @api_admin_bp.route('/users//allowed-markets', methods=['POST']) @admin_required def set_user_allowed_markets(user_id): """Set allowed markets for a user. Body: { "markets": ["stock", "stock_us", "crypto"] } Only valid modes are accepted. At least 'stock' must be included. """ user = User.query.get_or_404(user_id) data = request.get_json() or {} markets = data.get('markets', []) valid_modes = ('crypto', 'stock', 'stock_us') # Filter to valid modes only markets = [m for m in markets if m in valid_modes] if not markets: markets = ['stock'] # Baseline always included if 'stock' not in markets: markets.insert(0, 'stock') # Ensure stock is always first UserSettings.set(user_id, 'allowed_markets', markets, 'display') # If user's current mode is no longer allowed, reset to stock current_mode = UserSettings.get(user_id, 'asset_mode', 'stock') if current_mode not in markets: UserSettings.set(user_id, 'asset_mode', 'stock', 'display') return jsonify({ 'user_id': user_id, 'username': user.username, 'allowed_markets': markets, 'message': f'Market access updated: {", ".join(markets)}', }) @api_admin_bp.route('/stats') @admin_required def admin_stats(): """Get admin dashboard stats.""" from app.models.asset import Asset from app.models.portfolio import Portfolio total_users = User.query.count() pending = User.query.filter_by(is_approved=False, is_active=True).count() active = User.query.filter_by(is_approved=True, is_active=True).count() disabled = User.query.filter_by(is_active=False).count() total_coins = Asset.query.filter_by(is_active=True).count() # Recent registrations recent = User.query.order_by(User.created_at.desc()).limit(8).all() # Tier distribution tier_dist = db.session.query( User.subscription_tier, db.func.count(User.id) ).filter( User.is_approved == True, # noqa: E712 User.is_active == True # noqa: E712 ).group_by(User.subscription_tier).all() tier_order = {'free': 0, 'starter': 1, 'pro': 2, 'ultimate': 3} tier_distribution = sorted( [{'tier': t or 'free', 'count': c} for t, c in tier_dist], key=lambda x: tier_order.get(x['tier'], 99) ) return jsonify({ 'total_users': total_users, 'pending_users': pending, 'active_users': active, 'disabled_users': disabled, 'total_coins': total_coins, 'tier_distribution': tier_distribution, 'recent_registrations': [u.to_dict() for u in recent], }) def _copy_default_settings(user_id): """Copy default fee/capital settings from AppSettings to a new user.""" fee_keys = [ ('buy_fee_pct', 'fees', 0.31), ('sell_fee_pct', 'fees', 0.31), ('stock_buy_fee_pct', 'fees', 0.15), ('stock_sell_fee_pct', 'fees', 0.25), ('stock_us_buy_fee_pct', 'fees', 0.0), ('stock_us_sell_fee_pct', 'fees', 0.0), ] capital_keys = [ ('default_capital', 'capital', 1_000_000), ('active_capital_pct', 'capital', 75), ('reserve_capital_pct', 'capital', 25), ('stock_default_capital', 'capital', 10_000_000), ('stock_active_capital_pct', 'capital', 75), ('stock_reserve_capital_pct', 'capital', 25), ('stock_us_default_capital', 'capital', 1000), ('stock_us_active_capital_pct', 'capital', 75), ('stock_us_reserve_capital_pct', 'capital', 25), ] for key, cat, default in fee_keys + capital_keys: val = AppSettings.get(key, default) UserSettings.set(user_id, key, val, cat) # Display defaults UserSettings.set(user_id, 'asset_mode', 'stock', 'display') UserSettings.set(user_id, 'color_mode', 'dark', 'display') # Default: only Saham Indonesia — admin activates stock_us/crypto per user UserSettings.set(user_id, 'allowed_markets', ['stock'], 'display') # ── Landing Page CMS endpoints ──────────────────────────────────────── @api_admin_bp.route('/landing/
', methods=['GET']) @admin_required def get_landing_section(section): """Get a single landing page section.""" from app.helpers.landing_content import get_landing_content, DEFAULTS key = f'landing_{section}' content = get_landing_content() if key not in content: return jsonify({'error': f'Unknown section: {section}'}), 404 return jsonify(content[key]) @api_admin_bp.route('/landing/
', methods=['POST']) @admin_required def save_landing_section(section): """Save a landing page section.""" from app.helpers.landing_content import update_landing_section key = f'landing_{section}' data = request.get_json() if data is None: return jsonify({'error': 'Invalid JSON'}), 400 try: update_landing_section(key, data) return jsonify({'ok': True, 'section': section}) except ValueError as e: return jsonify({'error': str(e)}), 400 @api_admin_bp.route('/landing', methods=['GET']) @admin_required def get_all_landing(): """Get all landing page content.""" from app.helpers.landing_content import get_landing_content return jsonify(get_landing_content()) @api_admin_bp.route('/landing/reset/
', methods=['POST']) @admin_required def reset_landing_section(section): """Reset a landing page section to defaults.""" from app.helpers.landing_content import update_landing_section, DEFAULTS key = f'landing_{section}' if key not in DEFAULTS: return jsonify({'error': f'Unknown section: {section}'}), 404 update_landing_section(key, DEFAULTS[key]) return jsonify({'ok': True, 'section': section, 'data': DEFAULTS[key]})