"""Admin listing-management service. All query / aggregation / mutation logic for the ``/admin/listings`` management page lives here; routes stay thin. Filtering and sorting happen in SQL (so pagination is correct across pages), per-page counts are resolved with bounded GROUP BY / batched queries (never one-per-row), and every admin action goes through a single service function so the routes never loop or query. Premium seller status is resolved from the ``subscriptions`` table (the single source of truth) via the batched :func:`get_premium_seller_ids` helper. The "currently featured" definition mirrors :mod:`app.marketplace.services` — a listing is featured only while ``is_featured`` is set *and* ``featured_until`` is still in the future — so an expired feature never counts. """ 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.marketplace.services import ( FEATURE_DURATION_DAYS, expire_featured_listing, featured_days_remaining, is_listing_currently_featured, ) from app.models import ( Category, Conversation, Favorite, Listing, ListingStatus, ListingView, Offer, User, UserProfile, ) from app.models.base import utcnow from app.moderation.services import get_open_report_listing_ids from app.premium.services import get_premium_seller_ids # Filter / sort option lists — exposed so the template and query stay in sync. LISTING_SORT_OPTIONS = [ ("newest", "Newest"), ("oldest", "Oldest"), ("most_views", "Most Views"), ("most_favorites", "Most Favorites"), ("most_offers", "Most Offers"), ] LISTING_STATUS_FILTERS = [ ("all", "All"), ("active", "Active"), ("reserved", "Reserved"), ("sold", "Sold"), ("deleted", "Deleted"), ] FEATURED_FILTER_OPTIONS = [ ("all", "All"), ("featured", "Featured"), ("not_featured", "Not Featured"), ] # Admin actions that mutate a listing. Every quick / bulk action funnels through # :func:`apply_listing_action`, so both the single-row routes and the bulk route # share one validated code path. LISTING_ACTIONS = ("feature", "unfeature", "available", "reserved", "sold", "delete", "restore") _STATUS_ACTION_MAP = { "available": ListingStatus.AVAILABLE, "reserved": ListingStatus.RESERVED, "sold": ListingStatus.SOLD, } def _currently_featured(): """SQL predicate: featured AND still within its live featured window.""" return db.and_( Listing.is_featured.is_(True), Listing.featured_until.isnot(None), Listing.featured_until > utcnow(), ) # --- Summary --------------------------------------------------------------- def get_listing_summary() -> dict: """Platform-wide listing summary cards (single-pass aggregation). All counts except *Deleted* are scoped to live (non-deleted) listings, so *Total Listings* matches the analytics dashboard. *Deleted* counts soft-deleted rows (``deleted_at`` set) separately. """ live = Listing.deleted_at.is_(None) row = 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((Listing.status == ListingStatus.SOLD, 1), else_=0)), 0), func.coalesce(func.sum(case((Listing.status == ListingStatus.RESERVED, 1), else_=0)), 0), func.coalesce(func.sum(case((_currently_featured(), 1), else_=0)), 0), func.coalesce(func.avg(Listing.view_count), 0), func.coalesce(func.avg(Listing.favorite_count), 0), ).where(live) ).one() total, active, sold, reserved, featured, avg_views, avg_favorites = row deleted = db.session.scalar( db.select(func.count(Listing.id)).where(Listing.deleted_at.isnot(None)) ) or 0 return { "total_listings": int(total), "active_listings": int(active), "sold_listings": int(sold), "reserved_listings": int(reserved), "featured_listings": int(featured), "deleted_listings": int(deleted), "avg_views": round(float(avg_views), 1), "avg_favorites": round(float(avg_favorites), 1), } # --- Listing table --------------------------------------------------------- def get_listings(page=1, search=None, status=None, category=None, featured=None, seller=None, sort="newest"): """Return ``(pagination, rows)`` for the listing table. ``rows`` are per-listing display dicts. Filtering and sorting are performed in SQL; the per-page offer counts and seller-premium flags are resolved in two bounded queries (a GROUP BY over offers and one batched premium lookup), never one-per-row. """ per_page = current_app.config.get("DEFAULT_PAGE_SIZE", 20) query = db.select(Listing).options( selectinload(Listing.images), selectinload(Listing.category), selectinload(Listing.seller).selectinload(User.profile), ) # Status scopes the default view to live listings; "deleted" opts in. status = (status or "all").lower() if status == "deleted": query = query.where(Listing.deleted_at.isnot(None)) else: query = query.where(Listing.deleted_at.is_(None)) if status == "active": query = query.where(Listing.status == ListingStatus.AVAILABLE) elif status == "reserved": query = query.where(Listing.status == ListingStatus.RESERVED) elif status == "sold": query = query.where(Listing.status == ListingStatus.SOLD) featured = (featured or "all").lower() if featured in ("featured", "1", "true", "yes"): query = query.where(_currently_featured()) elif featured in ("not_featured", "0", "false", "no"): query = query.where(~_currently_featured()) if category and category != "all": query = query.where( Listing.category_id.in_(db.select(Category.id).where(Category.public_id == category)) ) if seller and seller != "all": query = query.where( Listing.seller_id.in_(db.select(User.id).where(User.public_id == seller)) ) if search and search.strip(): esc = search.strip().replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") like = f"%{esc}%" query = query.where(or_( Listing.title.ilike(like, escape="\\"), Listing.public_id.ilike(like, escape="\\"), Listing.seller_id.in_(db.select(User.id).where(User.email.ilike(like, escape="\\"))), Listing.seller_id.in_( db.select(UserProfile.user_id).where(UserProfile.full_name.ilike(like, escape="\\")) ), )) if sort == "most_offers": agg = ( db.select(Offer.listing_id.label("lid"), func.count(Offer.id).label("oc")) .group_by(Offer.listing_id) .subquery() ) query = query.outerjoin(agg, agg.c.lid == Listing.id) query = query.order_by(func.coalesce(agg.c.oc, 0).desc(), Listing.created_at.desc()) elif sort == "most_views": query = query.order_by(Listing.view_count.desc(), Listing.created_at.desc()) elif sort == "most_favorites": query = query.order_by(Listing.favorite_count.desc(), Listing.created_at.desc()) elif sort == "oldest": query = query.order_by(Listing.created_at.asc()) else: # newest query = query.order_by(Listing.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(listings) -> list[dict]: """Attach per-listing offer counts and seller-premium flags (2 queries). Views / favorites come from the cached counters already loaded on each row, so only the offer count needs a bounded GROUP BY. """ listing_ids = [l.id for l in listings] offer_counts: dict[int, int] = {} if listing_ids: for lid, oc in db.session.execute( db.select(Offer.listing_id, func.count(Offer.id)) .where(Offer.listing_id.in_(listing_ids)) .group_by(Offer.listing_id) ).all(): offer_counts[lid] = int(oc) premium_ids = get_premium_seller_ids([l.seller_id for l in listings]) if listings else set() reported_ids = get_open_report_listing_ids(listing_ids) if listing_ids else set() rows = [] for l in listings: seller = l.seller profile = seller.profile if seller else None rows.append({ "public_id": l.public_id, "title": l.title, "thumbnail": l.images[0].image_path if l.images else None, "seller_name": profile.full_name if profile else (seller.email.split("@")[0] if seller else "—"), "seller_email": seller.email if seller else "—", "seller_public_id": seller.public_id if seller else None, "seller_is_premium": bool(seller and seller.id in premium_ids), "category": l.category.name if l.category else "—", "price": l.price, "condition": l.condition.value, "views": l.view_count, "favorites": l.favorite_count, "offers": offer_counts.get(l.id, 0), "status": l.status.value, "is_featured": is_listing_currently_featured(l), "is_deleted": l.deleted_at is not None, "is_reported": l.id in reported_ids, "created": l.created_at, }) return rows def get_category_filter_options() -> list[tuple[str, str]]: """``(public_id, name)`` options for the category filter, ``all`` first.""" categories = db.session.scalars( db.select(Category).order_by(Category.display_order, Category.name) ).all() return [("all", "All Categories")] + [(c.public_id, c.name) for c in categories] # --- Listing detail -------------------------------------------------------- def get_listing_detail(listing: Listing) -> dict: """Full per-listing detail: analytics, featured status, view-history summary and recent activity — reusing existing models via bounded aggregation.""" offers = db.session.scalar( db.select(func.count(Offer.id)).where(Offer.listing_id == listing.id) ) or 0 conversations = db.session.scalar( db.select(func.count(Conversation.id)).where(Conversation.listing_id == listing.id) ) or 0 view_events, unique_viewers, unique_sessions, last_viewed = db.session.execute( db.select( func.count(ListingView.id), func.count(func.distinct(ListingView.viewer_id)), func.count(func.distinct(ListingView.session_id)), func.max(ListingView.viewed_at), ).where(ListingView.listing_id == listing.id) ).one() seller_premium = bool(get_premium_seller_ids([listing.seller_id])) is_reported = bool(get_open_report_listing_ids([listing.id])) return { "analytics": { "views": listing.view_count, "favorites": listing.favorite_count, "offers": int(offers), "conversations": int(conversations), }, "is_reported": is_reported, "view_history": { "events": int(view_events or 0), "unique_viewers": int(unique_viewers or 0), "unique_sessions": int(unique_sessions or 0), "last_viewed": last_viewed, }, "is_featured": is_listing_currently_featured(listing), "featured_until": listing.featured_until, "featured_days_left": featured_days_remaining(listing), "seller_is_premium": seller_premium, "recent_activity": _recent_activity(listing), } def _recent_activity(listing: Listing, limit: int = 8) -> list[dict]: """Merged latest activity on this listing (offers + favorites), bounded.""" items: list[dict] = [] for created_at, amount, email in db.session.execute( db.select(Offer.created_at, Offer.amount, User.email) .join(User, Offer.buyer_id == User.id) .where(Offer.listing_id == listing.id) .order_by(Offer.created_at.desc()).limit(limit) ).all(): items.append({"icon": "bi-tag-fill", "text": f"Offer of ₹{amount:,.0f} from {email}", "timestamp": created_at}) for created_at, email in db.session.execute( db.select(Favorite.created_at, User.email) .join(User, Favorite.user_id == User.id) .where(Favorite.listing_id == listing.id) .order_by(Favorite.created_at.desc()).limit(limit) ).all(): items.append({"icon": "bi-heart-fill", "text": f"Saved by {email}", "timestamp": created_at}) items.sort(key=lambda i: i["timestamp"], reverse=True) return items[:limit] # --- Mutations ------------------------------------------------------------- def feature_listing_admin(listing: Listing, *, commit: bool = True) -> bool: """Admin override: feature a listing for ``FEATURE_DURATION_DAYS`` days. Unlike the seller-facing :func:`app.marketplace.services.feature_listing`, this bypasses the ownership and premium-capability checks (an admin acts on behalf of the platform), but still refuses to feature a deleted or unavailable listing so the "featured => available" invariant holds. Returns ``True`` when the listing was (re)featured. """ if listing.deleted_at is not None or listing.status != ListingStatus.AVAILABLE: return False now = utcnow() listing.is_featured = True listing.featured_at = now listing.featured_until = now + timedelta(days=FEATURE_DURATION_DAYS) if commit: db.session.commit() return True def unfeature_listing_admin(listing: Listing, *, commit: bool = True) -> None: """Admin override: clear a listing's featured state (reuses the marketplace clearing routine so the behaviour matches expiry/un-feature everywhere).""" expire_featured_listing(listing, commit=commit) def change_listing_status(listing: Listing, action: str, *, editor=None, commit: bool = True) -> bool: """Apply a lifecycle transition to a listing (admin override). ``action`` is one of ``available`` / ``reserved`` / ``sold`` (plain status change), ``delete`` (soft delete) or ``restore``. Featured state is cleared whenever a listing leaves the available catalog, preserving "featured => available". Returns ``True`` when a change was applied. """ editor_id = getattr(editor, "id", None) if action == "delete": if listing.deleted_at is not None: return False listing.deleted_at = utcnow() listing.status = ListingStatus.REMOVED listing.updated_by = editor_id _clear_featured(listing) elif action == "restore": if listing.deleted_at is None: return False listing.deleted_at = None listing.status = ListingStatus.AVAILABLE listing.updated_by = editor_id elif action in _STATUS_ACTION_MAP: if listing.deleted_at is not None: return False target = _STATUS_ACTION_MAP[action] if listing.status == target: return False listing.status = target listing.updated_by = editor_id if target in (ListingStatus.SOLD, ListingStatus.RESERVED): _clear_featured(listing) else: return False if commit: db.session.commit() return True def _clear_featured(listing: Listing) -> None: listing.is_featured = False listing.featured_at = None listing.featured_until = None def apply_listing_action(listing: Listing, action: str, *, editor=None, commit: bool = True) -> bool: """Single dispatch point for every quick / bulk action on a listing.""" if action == "feature": return feature_listing_admin(listing, commit=commit) if action == "unfeature": if not is_listing_currently_featured(listing) and not listing.is_featured: return False unfeature_listing_admin(listing, commit=commit) return True return change_listing_status(listing, action, editor=editor, commit=commit) def bulk_update_listings(action: str, public_ids, *, editor=None) -> int: """Apply ``action`` to many listings in one query + one commit. Resolves all target listings with a single ``IN`` query (no per-id lookup), applies the shared :func:`apply_listing_action` to each, and commits once. Returns the number of listings actually changed. The loop lives here in the service, never in the route. """ if action not in LISTING_ACTIONS: return 0 ids = [pid for pid in (public_ids or []) if pid] if not ids: return 0 listings = db.session.scalars( db.select(Listing).where(Listing.public_id.in_(ids)) ).all() changed = 0 for listing in listings: if apply_listing_action(listing, action, editor=editor, commit=False): changed += 1 if changed: db.session.commit() return changed