"""Admin user-management service. All query/aggregation logic for the ``/admin/users`` management page lives here; routes stay thin. Premium status is resolved from the ``subscriptions`` table (the single source of truth) via the batched helper, so the listing never runs a per-row premium lookup. """ from __future__ import annotations from datetime import timedelta from flask import current_app from sqlalchemy import case, func, or_ from sqlalchemy.orm import selectinload from app.extensions import db from app.models import ( Conversation, Favorite, Listing, ListingStatus, Offer, Subscription, SubscriptionStatus, User, UserProfile, ) from app.models.base import utcnow from app.models.enums import AccountStatus, UserRole from app.moderation.services import get_user_report_counts from app.premium.services import get_active_subscription, get_premium_seller_ids, premium_days_remaining # Filter / sort option lists — exposed so the template and query stay in sync. USER_SORT_OPTIONS = [ ("newest", "Newest"), ("oldest", "Oldest"), ("most_listings", "Most Listings"), ("most_views", "Most Views"), ("premium_first", "Premium First"), ] PREMIUM_FILTER_OPTIONS = [("all", "All"), ("premium", "Premium"), ("free", "Free")] STATUS_FILTER_OPTIONS = [("all", "All"), ("active", "Active"), ("suspended", "Suspended")] ROLE_FILTER_OPTIONS = [("all", "All"), ("student", "Student"), ("admin", "Admin")] def _currently_featured(): return db.and_( Listing.is_featured.is_(True), Listing.featured_until.isnot(None), Listing.featured_until > utcnow(), ) def _active_subscription_select(): return db.select(Subscription.user_id).where( Subscription.status == SubscriptionStatus.ACTIVE, Subscription.end_date > utcnow() ) def get_user_summary() -> dict: """Platform-wide user summary cards (aggregation queries).""" now = utcnow() total_users = db.session.scalar(db.select(func.count(User.id))) or 0 premium_users = db.session.scalar( db.select(func.count(func.distinct(Subscription.user_id))) .where(Subscription.status == SubscriptionStatus.ACTIVE, Subscription.end_date > now) ) or 0 new_users_30d = db.session.scalar( db.select(func.count(User.id)).where(User.created_at >= now - timedelta(days=30)) ) or 0 suspended_users = db.session.scalar( db.select(func.count(User.id)).where(User.account_status == AccountStatus.SUSPENDED) ) or 0 total_listings = db.session.scalar( db.select(func.count(Listing.id)).where(Listing.deleted_at.is_(None)) ) or 0 return { "total_users": int(total_users), "premium_users": int(premium_users), "free_users": int(total_users - premium_users), "new_users_30d": int(new_users_30d), "suspended_users": int(suspended_users), "avg_listings_per_user": round(total_listings / total_users, 1) if total_users else 0.0, } def get_users(page=1, search=None, premium=None, status=None, role=None, sort="newest"): """Return ``(pagination, rows)`` for the user table. ``rows`` are per-user display dicts (name, premium flag, listing/featured/view counts, status…). Filtering and sorting happen in SQL (so pagination is correct across pages); the per-page aggregate counts and premium flags are resolved in two bounded queries — never one-per-row. """ per_page = current_app.config.get("DEFAULT_PAGE_SIZE", 20) active_sub = _active_subscription_select() query = db.select(User).options(selectinload(User.profile)) if search and search.strip(): esc = search.strip().replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") like = f"%{esc}%" query = query.where(or_( User.email.ilike(like, escape="\\"), User.public_id.ilike(like, escape="\\"), User.id.in_(db.select(UserProfile.user_id).where(UserProfile.full_name.ilike(like, escape="\\"))), )) premium = (premium or "all").lower() if premium in ("premium", "active"): query = query.where(User.id.in_(active_sub)) elif premium == "free": query = query.where(~User.id.in_(active_sub)) status = (status or "all").lower() if status == "active": query = query.where(User.account_status == AccountStatus.ACTIVE) elif status == "suspended": query = query.where(User.account_status == AccountStatus.SUSPENDED) role = (role or "all").lower() if role == "student": query = query.where(User.role == UserRole.STUDENT) elif role in ("admin", "super_admin"): query = query.where(User.role == UserRole.SUPER_ADMIN) if sort in ("most_listings", "most_views"): agg = ( db.select( Listing.seller_id.label("sid"), func.count(Listing.id).label("lc"), func.coalesce(func.sum(Listing.view_count), 0).label("vc"), ) .where(Listing.deleted_at.is_(None)) .group_by(Listing.seller_id) .subquery() ) query = query.outerjoin(agg, agg.c.sid == User.id) order_col = agg.c.lc if sort == "most_listings" else agg.c.vc query = query.order_by(func.coalesce(order_col, 0).desc(), User.created_at.desc()) elif sort == "premium_first": query = query.order_by(User.id.in_(active_sub).desc(), User.created_at.desc()) elif sort == "oldest": query = query.order_by(User.created_at.asc()) else: # newest query = query.order_by(User.created_at.desc()) pagination = db.paginate(query, page=page, per_page=per_page, error_out=False) return pagination, _build_rows(pagination.items) def _build_rows(users) -> list[dict]: """Attach per-user listing/featured/view counts and premium flag (2 queries).""" user_ids = [u.id for u in users] counts: dict[int, tuple[int, int, int]] = {} if user_ids: for sid, lc, fc, vc in db.session.execute( db.select( Listing.seller_id, func.count(Listing.id), func.coalesce(func.sum(case((_currently_featured(), 1), else_=0)), 0), func.coalesce(func.sum(Listing.view_count), 0), ) .where(Listing.seller_id.in_(user_ids), Listing.deleted_at.is_(None)) .group_by(Listing.seller_id) ).all(): counts[sid] = (int(lc), int(fc), int(vc)) premium_ids = get_premium_seller_ids(user_ids) if user_ids else set() rows = [] for u in users: lc, fc, vc = counts.get(u.id, (0, 0, 0)) rows.append({ "public_id": u.public_id, "full_name": u.profile.full_name if u.profile else u.email.split("@")[0], "email": u.email, "profile_picture": u.profile.profile_picture if u.profile else None, "role": u.role.value, "is_premium": u.id in premium_ids, "listings": lc, "featured": fc, "views": vc, "joined": u.created_at, "last_login": u.last_login, "account_status": u.account_status.value, }) return rows def get_user_detail(user: User) -> dict: """Full per-user detail: listing summary, premium + subscription history, recent activity and a few listings — reusing existing models via aggregation.""" agg = db.session.execute( db.select( func.count(Listing.id), func.coalesce(func.sum(case((Listing.status == ListingStatus.AVAILABLE, 1), else_=0)), 0), func.coalesce(func.sum(case((_currently_featured(), 1), else_=0)), 0), func.coalesce(func.sum(Listing.view_count), 0), func.coalesce(func.sum(Listing.favorite_count), 0), ).where(Listing.seller_id == user.id, Listing.deleted_at.is_(None)) ).one() total_listings, active_listings, featured, views, favorites = (int(x) for x in agg) offers = db.session.scalar( db.select(func.count(Offer.id)).where(Offer.seller_id == user.id) ) or 0 conversations = db.session.scalar( db.select(func.count(Conversation.id)).where(Conversation.seller_id == user.id) ) or 0 history = db.session.scalars( db.select(Subscription).where(Subscription.user_id == user.id).order_by(Subscription.start_date.desc()) ).all() active = get_active_subscription(user) listings = db.session.scalars( db.select(Listing).options(selectinload(Listing.images)) .where(Listing.seller_id == user.id, Listing.deleted_at.is_(None)) .order_by(Listing.created_at.desc()).limit(6) ).all() return { "summary": { "total_listings": total_listings, "active_listings": active_listings, "featured_listings": featured, "total_views": views, "total_favorites": favorites, "total_offers": int(offers), "total_conversations": int(conversations), }, "is_premium": active is not None, "subscription_active": active, "subscription_history": history, "days_remaining": premium_days_remaining(user), "listings": listings, "recent_activity": _recent_activity(user), "report_counts": get_user_report_counts(user), } def _recent_activity(user: User, limit: int = 6) -> list[dict]: """Merged latest activity across offers received and favorites (bounded).""" items: list[dict] = [] for created_at, amount, title in db.session.execute( db.select(Offer.created_at, Offer.amount, Listing.title) .join(Listing, Offer.listing_id == Listing.id) .where(Offer.seller_id == user.id).order_by(Offer.created_at.desc()).limit(limit) ).all(): items.append({"icon": "bi-tag-fill", "text": f"Offer received: ₹{amount:,.0f} on {title}", "timestamp": created_at}) for created_at, title in db.session.execute( db.select(Favorite.created_at, Listing.title) .join(Listing, Favorite.listing_id == Listing.id) .where(Listing.seller_id == user.id).order_by(Favorite.created_at.desc()).limit(limit) ).all(): items.append({"icon": "bi-heart-fill", "text": f"Someone favorited {title}", "timestamp": created_at}) items.sort(key=lambda i: i["timestamp"], reverse=True) return items[:limit] def set_account_status(user: User, status: AccountStatus) -> None: """Set a user's account status (suspend / reactivate).""" user.account_status = status db.session.commit()