"""Admin backend: dashboard KPIs + user management. Admin-only.""" from flask import Blueprint, render_template, request, redirect, url_for, flash from flask_login import current_user from flask_babel import gettext as _ from app.extensions import db from app.models.user import User from app.models.listing import Listing from app.models.plan import Plan from app.models.trust import TrustEvent from app.models.audit import AuditLog from app.models.report import Report from app.models.enums import UserStatus, TrustEventType, ListingStatus from app.utils import admin_required from app.services import admin_users as usvc from app.services import admin_dashboard as dash from app.services import moderation as msvc from app.services.settings import get_setting, set_setting from app.services import audit admin_bp = Blueprint("admin", __name__) PER_PAGE = 25 @admin_bp.route("/admin") @admin_required def dashboard(): kpis = { "active_listings": dash.active_listings_count(), "new_users_7d": dash.new_users_count(7), "new_users_30d": dash.new_users_count(30), "mrr_cents": dash.mrr_cents(), "revenue_30d_cents": dash.revenue_30d_cents(), "flag_queue_depth": dash.flag_queue_depth(), } return render_template("admin/dashboard.html", kpis=kpis) @admin_bp.route("/admin/users") @admin_required def users(): page = request.args.get("page", 1, type=int) role = request.args.get("role", type=str) or None status = request.args.get("status", type=str) or None q = request.args.get("q", type=str) or None pagination = usvc.search_query(role=role, status=status, q=q).paginate( page=page, per_page=PER_PAGE, error_out=False) return render_template("admin/users.html", pagination=pagination, filters=request.args) @admin_bp.route("/admin/users/") @admin_required def user_detail(user_id): user = User.query.get_or_404(user_id) listings = (Listing.query.filter_by(user_id=user.id) .order_by(Listing.created_at.desc()).limit(50).all()) trust_events = (user.trust_events.order_by(TrustEvent.created_at.desc()) .limit(50).all()) audit_history = (AuditLog.query.filter_by(target_type="user", target_id=user.id) .order_by(AuditLog.created_at.desc()).limit(50).all()) plans = Plan.query.order_by(Plan.sort_order).all() return render_template("admin/user_detail.html", user=user, listings=listings, trust_events=trust_events, audit_history=audit_history, plans=plans, TrustEventType=TrustEventType) def _guard_self_action(user): if user.id == current_user.id: flash(_("You cannot perform this action on your own account."), "danger") return True return False @admin_bp.route("/admin/users//ban", methods=["POST"]) @admin_required def ban_user(user_id): user = User.query.get_or_404(user_id) if _guard_self_action(user): return redirect(url_for("admin.user_detail", user_id=user.id)) usvc.set_status(user, UserStatus.banned, actor=current_user) db.session.commit() flash(_("User banned."), "warning") return redirect(url_for("admin.user_detail", user_id=user.id)) @admin_bp.route("/admin/users//suspend", methods=["POST"]) @admin_required def suspend_user(user_id): user = User.query.get_or_404(user_id) if _guard_self_action(user): return redirect(url_for("admin.user_detail", user_id=user.id)) usvc.set_status(user, UserStatus.suspended, actor=current_user) db.session.commit() flash(_("User suspended."), "warning") return redirect(url_for("admin.user_detail", user_id=user.id)) @admin_bp.route("/admin/users//activate", methods=["POST"]) @admin_required def activate_user(user_id): user = User.query.get_or_404(user_id) usvc.set_status(user, UserStatus.active, actor=current_user) db.session.commit() flash(_("User activated."), "success") return redirect(url_for("admin.user_detail", user_id=user.id)) @admin_bp.route("/admin/users//tier", methods=["POST"]) @admin_required def set_tier(user_id): user = User.query.get_or_404(user_id) plan_id = request.form.get("plan_id", type=int) plan = db.session.get(Plan, plan_id) if plan_id else None usvc.set_tier(user, plan, actor=current_user) db.session.commit() flash(_("Tier updated."), "success") return redirect(url_for("admin.user_detail", user_id=user.id)) @admin_bp.route("/admin/users//trust", methods=["POST"]) @admin_required def trust_adjust(user_id): user = User.query.get_or_404(user_id) delta = request.form.get("delta", type=int) or 0 event_type = request.form.get("event_type", type=str) usvc.adjust_trust(user, TrustEventType(event_type), delta, actor=current_user) db.session.commit() flash(_("Trust adjusted."), "success") return redirect(url_for("admin.user_detail", user_id=user.id)) # --- listing moderation --- @admin_bp.route("/admin/listings") @admin_required def listings(): page = request.args.get("page", 1, type=int) pagination = msvc.flag_queue(page=page, per_page=PER_PAGE) flag_threshold = get_setting("flag_threshold", 5) keyword_blocklist = get_setting("keyword_blocklist", []) return render_template("admin/listings.html", pagination=pagination, flag_threshold=flag_threshold, keyword_blocklist=keyword_blocklist) @admin_bp.route("/admin/listings/settings", methods=["POST"]) @admin_required def listings_settings(): threshold = request.form.get("flag_threshold", type=int) or 5 raw_blocklist = request.form.get("keyword_blocklist", "") blocklist = [line.strip() for line in raw_blocklist.splitlines() if line.strip()] set_setting("flag_threshold", threshold) set_setting("keyword_blocklist", blocklist) audit.log_action(current_user, "settings.updated", "setting", None, meta={"flag_threshold": threshold, "keyword_blocklist": blocklist}) db.session.commit() flash(_("Moderation settings updated."), "success") return redirect(url_for("admin.listings")) @admin_bp.route("/admin/listings//approve", methods=["POST"]) @admin_required def approve_listing(listing_id): listing = Listing.query.get_or_404(listing_id) msvc.approve(listing, actor=current_user) db.session.commit() flash(_("Listing approved."), "success") return redirect(url_for("admin.listings")) @admin_bp.route("/admin/listings//hide", methods=["POST"]) @admin_required def hide_listing(listing_id): listing = Listing.query.get_or_404(listing_id) msvc.hide(listing, actor=current_user) db.session.commit() flash(_("Listing hidden."), "warning") return redirect(url_for("admin.listings")) @admin_bp.route("/admin/listings//remove", methods=["POST"]) @admin_required def remove_listing(listing_id): listing = Listing.query.get_or_404(listing_id) msvc.remove(listing, actor=current_user) db.session.commit() flash(_("Listing removed."), "danger") return redirect(url_for("admin.listings")) # --- reports queue --- @admin_bp.route("/admin/reports") @admin_required def reports(): candidates = (Report.query.join(Listing, Report.listing_id == Listing.id) .filter(Listing.status == ListingStatus.flagged) .order_by(Report.created_at.desc()).all()) dismissed_ids = {a.target_id for a in AuditLog.query.filter_by(target_type="report", action="report.dismissed").all()} open_reports = [r for r in candidates if r.id not in dismissed_ids] return render_template("admin/reports.html", reports=open_reports) @admin_bp.route("/admin/reports//dismiss", methods=["POST"]) @admin_required def dismiss_report(report_id): report = Report.query.get_or_404(report_id) audit.log_action(current_user, "report.dismissed", "report", report.id, meta={"listing_id": report.listing_id, "reporter_id": report.reporter_id}) db.session.commit() flash(_("Report dismissed."), "info") return redirect(url_for("admin.reports")) @admin_bp.route("/admin/reports//escalate", methods=["POST"]) @admin_required def escalate_report(report_id): report = Report.query.get_or_404(report_id) audit.log_action(current_user, "report.escalated", "report", report.id, meta={"listing_id": report.listing_id, "reporter_id": report.reporter_id}) db.session.commit() flash(_("Report escalated."), "warning") return redirect(url_for("admin.reports"))