""" Admin API — user management and dropdown Excel upload. All routes require admin role. """ import json import logging import os import openpyxl from io import BytesIO from quart import Blueprint, jsonify, request from ..auth.middleware import admin_required from ..auth.user_store import list_users, set_role, set_active from ..api.dropdowns import save_dropdowns logger = logging.getLogger(__name__) admin_bp = Blueprint('admin', __name__, url_prefix='/api/admin') @admin_bp.route('/users', methods=['GET']) @admin_required async def get_users(): return jsonify({'users': list_users()}) @admin_bp.route('/users/', methods=['PATCH']) @admin_required async def update_user(user_id: str): body = await request.get_json() or {} user = None if 'role' in body: user = set_role(user_id, body['role']) if user is None: return jsonify({'error': 'invalid_role_or_not_found'}), 400 if 'active' in body: user = set_active(user_id, bool(body['active'])) if user is None: return jsonify({'error': 'not_found'}), 404 return jsonify({'success': True, 'user': user}) @admin_bp.route('/dropdowns/upload', methods=['POST']) @admin_required async def upload_dropdowns(): """ Upload a new Excel file (.xlsx) to update the dropdown categories. Expects multipart/form-data with field 'file'. Parses columns: A=Category name, E=Status, G=Media types (comma-separated). """ files = await request.files file = files.get('file') if not file: return jsonify({'error': 'no_file'}), 400 filename = file.filename or '' if not filename.lower().endswith('.xlsx'): return jsonify({'error': 'invalid_file_type', 'message': 'Only .xlsx files accepted'}), 400 try: data = file.read() wb = openpyxl.load_workbook(BytesIO(data)) ws = wb.active categories = [] for row in ws.iter_rows(min_row=2, values_only=True): if len(row) < 5 or not row[0]: continue name = str(row[0]).strip() status_raw = str(row[4]).strip() if row[4] else 'Active' status = 'Active' if 'active' in status_raw.lower() else 'Archived' media_raw = str(row[6]).strip() if len(row) > 6 and row[6] else '' media_types = [m.strip() for m in media_raw.split(',') if m.strip()] if media_raw else [] categories.append({'name': name, 'status': status, 'mediaTypes': media_types}) if not categories: return jsonify({'error': 'empty_file', 'message': 'No categories found in file'}), 400 save_dropdowns(categories) active_count = sum(1 for c in categories if c['status'] == 'Active') return jsonify({ 'success': True, 'total': len(categories), 'active': active_count, 'archived': len(categories) - active_count, }) except Exception as e: logger.error(f"Dropdown upload error: {e}", exc_info=True) return jsonify({'error': 'parse_error', 'message': str(e)}), 500 @admin_bp.route('/dropdowns/preview', methods=['POST']) @admin_required async def preview_dropdowns(): """Preview parsed categories from an uploaded file without saving.""" files = await request.files file = files.get('file') if not file: return jsonify({'error': 'no_file'}), 400 try: data = file.read() wb = openpyxl.load_workbook(BytesIO(data)) ws = wb.active categories = [] for row in ws.iter_rows(min_row=2, values_only=True): if len(row) < 5 or not row[0]: continue name = str(row[0]).strip() status_raw = str(row[4]).strip() if row[4] else 'Active' status = 'Active' if 'active' in status_raw.lower() else 'Archived' media_raw = str(row[6]).strip() if len(row) > 6 and row[6] else '' media_types = [m.strip() for m in media_raw.split(',') if m.strip()] if media_raw else [] categories.append({'name': name, 'status': status, 'mediaTypes': media_types}) return jsonify({'categories': categories, 'total': len(categories)}) except Exception as e: return jsonify({'error': 'parse_error', 'message': str(e)}), 500