"""Report & moderation service layer (Phase 7). Owns the full report lifecycle: students create reports, Super Admins triage and resolve them. All query / aggregation / mutation logic lives here so routes stay thin. Moderator state changes only mutate the report; the audit trail is written once by the route via :func:`app.admin.audit.record_admin_action`, so there is no duplicated audit code here. Filtering happens in SQL (correct pagination); per-page reporter / seller / moderator names are eager-loaded with ``selectinload`` (no N+1); summary and analytics use COUNT / GROUP BY aggregation. """ from __future__ import annotations from datetime import datetime, time, 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 ( Category, Conversation, Listing, Message, MessageType, Notification, NotificationType, Offer, RelatedEntityType, Report, ReportEntityType, ReportReason, ReportStatus, User, UserProfile, ) from app.models.base import utcnow TREND_WINDOW_DAYS = 30 _TOP_LIMIT = 5 class ReportError(Exception): """Base error for the reporting workflow.""" class CannotReportOwnListingError(ReportError): """Raised when a user tries to report their own listing.""" class DuplicateReportError(ReportError): """Raised when the reporter already has an open report for the entity.""" # Option lists — exposed so templates and queries stay in sync. REPORT_REASON_OPTIONS = [(r.value, r.value.replace("_", " ").title()) for r in ReportReason] REPORT_STATUS_FILTERS = [("all", "All")] + [(s.value, s.value.replace("_", " ").title()) for s in ReportStatus] REPORT_REASON_FILTERS = [("all", "All reasons")] + REPORT_REASON_OPTIONS REPORT_ENTITY_TYPE_FILTERS = [("all", "All types"), ("listing", "Listing"), ("user", "User")] REPORT_ASSIGNED_FILTERS = [("all", "All"), ("assigned", "Assigned"), ("unassigned", "Unassigned")] REPORT_SORT_OPTIONS = [("newest", "Newest"), ("oldest", "Oldest")] _OPEN_STATUSES = (ReportStatus.PENDING, ReportStatus.UNDER_REVIEW) def reason_label(reason) -> str: value = reason.value if hasattr(reason, "value") else str(reason) return value.replace("_", " ").title() def status_label(status) -> str: value = status.value if hasattr(status, "value") else str(status) return value.replace("_", " ").title() # --- Write path (students) ------------------------------------------------- def get_open_report(reporter: User, listing: Listing) -> Report | None: """Return this reporter's still-open report for the listing, if any.""" return db.session.scalar( db.select(Report).where( Report.reporter_id == reporter.id, Report.entity_type == ReportEntityType.LISTING, Report.entity_public_id == listing.public_id, Report.status.in_(_OPEN_STATUSES), ) ) def create_report(reporter: User, listing: Listing, reason: str, description: str | None = None) -> Report: """Create a PENDING report against a listing. Enforces the two business rules server-side: a user cannot report their own listing, and cannot file a second report for the same listing while an earlier one is still open. """ if listing.seller_id == reporter.id: raise CannotReportOwnListingError("You can't report your own listing.") if get_open_report(reporter, listing) is not None: raise DuplicateReportError("You already have an open report for this listing.") try: reason_enum = ReportReason(reason) except ValueError as exc: raise ReportError("Please choose a valid reason.") from exc report = Report( reporter_id=reporter.id, entity_type=ReportEntityType.LISTING, entity_public_id=listing.public_id, listing_id=listing.id, reported_user_id=listing.seller_id, reason=reason_enum, description=(description or None), status=ReportStatus.PENDING, created_at=utcnow(), ) db.session.add(report) db.session.commit() return report # --- Read path (admin table) ----------------------------------------------- def get_report_or_404(public_id: str) -> Report: return db.first_or_404(db.select(Report).where(Report.public_id == public_id)) # Backwards-compatible alias matching the documented service surface. def get_report(public_id: str) -> Report: """Fetch a single report by public id (or abort 404).""" return get_report_or_404(public_id) def get_reporter_reports(user: User) -> list[dict]: """A reporter's own reports (newest first) for the user-facing "My Reports" page. Privacy-safe: never exposes moderator notes or internal state — only the reporter's own submission and its current status.""" reports = db.session.scalars( db.select(Report).options(selectinload(Report.listing)) .where(Report.reporter_id == user.id) .order_by(Report.created_at.desc(), Report.id.desc()) ).all() return [{ "public_id": r.public_id, "short_id": r.public_id[:8], "entity_type": r.entity_type.value, "listing_title": r.listing.title if r.listing else None, "listing_public_id": (r.listing.public_id if r.listing and r.listing.deleted_at is None else None), "reason": reason_label(r.reason), "status": r.status.value, "status_label": status_label(r.status), "is_open": r.status in _OPEN_STATUSES, "created_at": r.created_at, "resolved_at": r.resolved_at, } for r in reports] def get_report_summary() -> dict: """Summary cards via a single conditional-aggregation pass (+ time windows).""" now = utcnow() today_start = datetime.combine(now.date(), time.min) month_start = datetime.combine(now.date().replace(day=1), time.min) def _status(status): return func.coalesce(func.sum(case((Report.status == status, 1), else_=0)), 0) row = db.session.execute( db.select( _status(ReportStatus.PENDING), _status(ReportStatus.UNDER_REVIEW), _status(ReportStatus.RESOLVED), _status(ReportStatus.DISMISSED), func.coalesce(func.sum(case((Report.created_at >= today_start, 1), else_=0)), 0), func.coalesce(func.sum(case((Report.created_at >= month_start, 1), else_=0)), 0), ) ).one() pending, under_review, resolved, dismissed, today, this_month = (int(x) for x in row) return { "pending": pending, "under_review": under_review, "resolved": resolved, "dismissed": dismissed, "today": today, "this_month": this_month, } def get_reports(page=1, search=None, status=None, reason=None, entity_type=None, moderator=None, assigned=None, seller=None, sort="newest"): """Return ``(pagination, rows)`` for the moderation table (SQL-filtered).""" per_page = current_app.config.get("DEFAULT_PAGE_SIZE", 20) query = db.select(Report).options( selectinload(Report.reporter).selectinload(User.profile), selectinload(Report.reported_user).selectinload(User.profile), selectinload(Report.moderator).selectinload(User.profile), selectinload(Report.listing), ) status = (status or "all").lower() if status != "all": try: query = query.where(Report.status == ReportStatus(status)) except ValueError: pass reason = (reason or "all").lower() if reason != "all": try: query = query.where(Report.reason == ReportReason(reason)) except ValueError: pass entity_type = (entity_type or "all").lower() if entity_type != "all": try: query = query.where(Report.entity_type == ReportEntityType(entity_type)) except ValueError: pass if moderator and moderator != "all": query = query.where( Report.moderator_id.in_(db.select(User.id).where(User.public_id == moderator)) ) assigned = (assigned or "all").lower() if assigned == "assigned": query = query.where(Report.moderator_id.isnot(None)) elif assigned == "unassigned": query = query.where(Report.moderator_id.is_(None)) if seller and seller != "all": query = query.where( Report.reported_user_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_( Report.entity_public_id.ilike(like, escape="\\"), Report.public_id.ilike(like, escape="\\"), Report.reporter_id.in_(db.select(User.id).where(User.email.ilike(like, escape="\\"))), Report.reported_user_id.in_(db.select(User.id).where(User.email.ilike(like, escape="\\"))), Report.listing_id.in_(db.select(Listing.id).where(Listing.title.ilike(like, escape="\\"))), )) if sort == "oldest": query = query.order_by(Report.created_at.asc(), Report.id.asc()) else: # newest query = query.order_by(Report.created_at.desc(), Report.id.desc()) pagination = db.paginate(query, page=page, per_page=per_page, error_out=False) return pagination, [_row(r) for r in pagination.items] def _name(user) -> str: if user is None: return "—" return user.profile.full_name if user.profile and user.profile.full_name else user.email def _row(report: Report) -> dict: listing = report.listing return { "public_id": report.public_id, "short_id": report.public_id[:8], "created_at": report.created_at, "reporter_name": _name(report.reporter), "reporter_email": report.reporter.email if report.reporter else "—", "entity_type": report.entity_type.value, "entity_name": (listing.title if listing else report.entity_public_id[:8]), "entity_public_id": report.entity_public_id, "seller_name": _name(report.reported_user), "seller_public_id": report.reported_user.public_id if report.reported_user else None, "reason": reason_label(report.reason), "status": report.status.value, "status_label": status_label(report.status), "moderator_name": _name(report.moderator) if report.moderator else None, "is_open": report.status in _OPEN_STATUSES, } # --- Report detail --------------------------------------------------------- def get_report_detail(report: Report) -> dict: """Rich detail for one report: listing, parties, previous reports for the same seller, moderator timeline and audit history — bounded queries.""" listing = None if report.listing_id: listing = db.session.scalar( db.select(Listing).options(selectinload(Listing.images), selectinload(Listing.category)) .where(Listing.id == report.listing_id) ) previous = [] if report.reported_user_id: prev_reports = db.session.scalars( db.select(Report).options(selectinload(Report.reporter)) .where(Report.reported_user_id == report.reported_user_id, Report.id != report.id) .order_by(Report.created_at.desc()).limit(10) ).all() previous = [{ "short_id": r.public_id[:8], "public_id": r.public_id, "reason": reason_label(r.reason), "status": r.status.value, "status_label": status_label(r.status), "created_at": r.created_at, } for r in prev_reports] # Moderator timeline derived from the report's own state (consistent icons). timeline = [{"icon": "bi-file-earmark-text", "text": f"Report submitted by {_name(report.reporter)}", "timestamp": report.created_at}] if report.moderator_id: timeline.append({"icon": "bi-person-check", "text": f"Assigned to {_name(report.moderator)}", "timestamp": report.created_at}) if report.resolved_at: verb = "Resolved" if report.status == ReportStatus.RESOLVED else "Dismissed" timeline.append({"icon": "bi-check-circle" if report.status == ReportStatus.RESOLVED else "bi-slash-circle", "text": f"{verb} by {_name(report.moderator)}", "timestamp": report.resolved_at}) audit_history = _audit_history(report.public_id) return { "listing": listing, "reporter_name": _name(report.reporter), "reporter_email": report.reporter.email if report.reporter else "—", "seller_name": _name(report.reported_user), "seller_public_id": report.reported_user.public_id if report.reported_user else None, "moderator_name": _name(report.moderator) if report.moderator_id else None, "reason": reason_label(report.reason), "previous_reports": previous, "seller_history": _seller_history(report), "timeline": timeline, "audit_history": audit_history, "conversations": get_listing_conversations(report), "notification_delivery": _notification_delivery(report), } def _notification_delivery(report: Report) -> dict: """Whether the reporter / reported user have been notified about this report. Derived from the notifications actually stored (all moderation notices link to ``REPORT`` + ``report.id``), so it reflects real delivery, not just intent.""" def _notified(user_id): if user_id is None: return False return db.session.scalar( db.select(Notification.id).where( Notification.user_id == user_id, Notification.related_entity_type == RelatedEntityType.REPORT, Notification.related_entity_id == report.id, ).limit(1) ) is not None reporter_notified = _notified(report.reporter_id) reported_notified = _notified(report.reported_user_id) return { "show": report.resolved_at is not None or reporter_notified or reported_notified, "reporter_notified": reporter_notified, "reported_notified": reported_notified, "sent_at": report.resolved_at, } # Consistent icons for the audit log (chronological moderator action feed). _AUDIT_ICONS = { "report_assigned": "bi-person-check", "report_under_review": "bi-search", "report_resolved": "bi-check-circle", "report_dismissed": "bi-slash-circle", "report_listing_removed": "bi-trash3", "report_seller_suspended": "bi-slash-circle", "seller_warned": "bi-exclamation-triangle", } def _seller_history(report: Report) -> dict | None: """Read-only moderation history for the reported seller — reuses existing tables (reports, listings, audit log) with a few bounded aggregate queries. Returns ``None`` when the report has no seller. Never fabricates data.""" seller_id = report.reported_user_id if not seller_id: return None from app.models import AdminAuditLog status_rows = db.session.execute( db.select(Report.status, func.count()) .where(Report.reported_user_id == seller_id) .group_by(Report.status) ).all() by_status = {status: int(count) for status, count in status_rows} total_reports = sum(by_status.values()) open_reports = sum(v for s, v in by_status.items() if s in _OPEN_STATUSES) resolved_reports = by_status.get(ReportStatus.RESOLVED, 0) listing_row = db.session.execute( db.select( func.coalesce(func.sum(case((Listing.deleted_at.is_(None), 1), else_=0)), 0), func.coalesce(func.sum(case((Listing.deleted_at.isnot(None), 1), else_=0)), 0), ).where(Listing.seller_id == seller_id) ).one() warnings = db.session.scalar( db.select(func.count(AdminAuditLog.id)).where( AdminAuditLog.action == "seller_warned", AdminAuditLog.entity_public_id.in_( db.select(Report.public_id).where(Report.reported_user_id == seller_id) ), ) ) or 0 return { "total_reports": total_reports, "open_reports": open_reports, "resolved_reports": resolved_reports, "warnings": int(warnings), "active_listings": int(listing_row[0]), "removed_listings": int(listing_row[1]), } # --- Conversation history (admin-only) ------------------------------------- def _offer_view(offer: Offer, conversation: Conversation) -> dict: """View model for one offer in the moderator timeline. A counter-offer is stored (see :func:`app.negotiation.services.counter_offer`) with ``buyer_id`` pointing at the seller, so the *maker* of any offer is its ``buyer_id``. That lets us label buyer offers vs. seller counter-offers. """ is_counter = offer.buyer_id == conversation.seller_id maker = conversation.seller if is_counter else conversation.buyer return { "kind": "offer", "offer_kind": "counter" if is_counter else "offer", "role": "seller" if is_counter else "buyer", "sender": _name(maker), "timestamp": offer.created_at, "amount": offer.amount, "content": offer.optional_message, "status": offer.status.value, "status_label": status_label(offer.status), } def _message_view(message: Message, conversation: Conversation) -> dict: """View model for one chat message (text or system event).""" is_system = message.message_type == MessageType.SYSTEM if is_system: role, sender = "system", "System" elif message.sender_id == conversation.seller_id: role, sender = "seller", _name(conversation.seller) else: role, sender = "buyer", _name(conversation.buyer) return { "kind": "system" if is_system else "message", "offer_kind": None, "role": role, "sender": sender, "timestamp": message.created_at, "amount": None, "content": message.message, "status": None, "status_label": None, } def get_listing_conversations(report: Report) -> dict: """Admin-only chat + negotiation history for the reported listing. Loads every conversation about the reported listing (each unique per buyer/seller/listing) with its messages, offers and both parties eager-loaded, then reuses :func:`app.negotiation.services.get_conversation_timeline` to merge each conversation's messages and offers into one chronological timeline. Bounded and free of N+1 lookups. Only ever rendered inside the admin report detail page — never exposed to students. """ empty = {"groups": [], "conversation_count": 0, "total_messages": 0, "total_offers": 0} if not report.listing_id: return empty from app.negotiation.services import get_conversation_timeline conversations = db.session.scalars( db.select(Conversation).options( selectinload(Conversation.messages).selectinload(Message.sender).selectinload(User.profile), selectinload(Conversation.offers), selectinload(Conversation.buyer).selectinload(User.profile), selectinload(Conversation.seller).selectinload(User.profile), ) .where(Conversation.listing_id == report.listing_id) .order_by(Conversation.created_at.asc()) ).all() groups = [] total_messages = total_offers = 0 for convo in conversations: events = [] for item in get_conversation_timeline(convo): if isinstance(item, Offer): events.append(_offer_view(item, convo)) else: events.append(_message_view(item, convo)) msg_count = sum(1 for e in events if e["kind"] in ("message", "system")) offer_count = sum(1 for e in events if e["kind"] == "offer") total_messages += msg_count total_offers += offer_count groups.append({ "public_id": convo.public_id, "buyer_name": _name(convo.buyer), "seller_name": _name(convo.seller), "started_at": convo.conversation_started_at or convo.created_at, "status": convo.status.value, "message_count": msg_count, "offer_count": offer_count, "events": events, }) return { "groups": groups, "conversation_count": len(groups), "total_messages": total_messages, "total_offers": total_offers, } def _audit_history(report_public_id: str) -> list[dict]: """Audit-log entries recorded against this report (bounded).""" from app.admin.audit import action_label from app.models import AdminAuditLog rows = db.session.scalars( db.select(AdminAuditLog).options(selectinload(AdminAuditLog.admin).selectinload(User.profile)) .where(AdminAuditLog.entity_public_id == report_public_id) .order_by(AdminAuditLog.created_at.desc()) ).all() history = [] for log in rows: admin = log.admin history.append({ "label": action_label(log.action), "icon": _AUDIT_ICONS.get(log.action, "bi-shield-check"), "admin_name": (admin.profile.full_name if admin and admin.profile and admin.profile.full_name else (admin.email if admin else "—")), "timestamp": log.created_at, }) return history # --- Moderator actions ----------------------------------------------------- # # Each mutates only the report (and, for the two entity actions, the target). The # route records the audit entry, so audit code is never duplicated here. def assign_report(report: Report, moderator: User) -> Report: """Assign the report to a moderator and move it into review.""" report.moderator_id = moderator.id if report.status == ReportStatus.PENDING: report.status = ReportStatus.UNDER_REVIEW db.session.commit() return report def mark_under_review(report: Report, moderator: User) -> Report: """Explicitly move a report to under-review (assigning if needed).""" report.moderator_id = report.moderator_id or moderator.id report.status = ReportStatus.UNDER_REVIEW db.session.commit() return report def resolve_report(report: Report, moderator: User, notes: str | None = None) -> Report: """Close a report as resolved.""" report.status = ReportStatus.RESOLVED report.moderator_id = report.moderator_id or moderator.id report.resolved_at = utcnow() if notes: report.moderator_notes = notes db.session.commit() return report def dismiss_report(report: Report, moderator: User, notes: str | None = None) -> Report: """Close a report as dismissed (no action warranted).""" report.status = ReportStatus.DISMISSED report.moderator_id = report.moderator_id or moderator.id report.resolved_at = utcnow() if notes: report.moderator_notes = notes db.session.commit() return report def remove_reported_listing(report: Report, moderator: User) -> bool: """Soft-delete the reported listing and resolve the report. Reuses the marketplace soft-delete so the listing leaves the catalog exactly as an owner deletion would. Returns ``True`` when a listing was removed. """ from app.marketplace.services import soft_delete_listing listing = report.listing if listing is None or listing.deleted_at is not None: resolve_report(report, moderator) return False soft_delete_listing(listing, moderator) resolve_report(report, moderator) return True def suspend_reported_seller(report: Report, moderator: User) -> bool: """Suspend the reported seller and resolve the report. Reuses the user-management status change. Won't suspend an admin account or an already-suspended user. Returns ``True`` when the seller was suspended. """ from app.admin.user_management import set_account_status from app.models.enums import AccountStatus, UserRole seller = report.reported_user resolved = False if seller is not None and seller.role != UserRole.SUPER_ADMIN and seller.account_status != AccountStatus.SUSPENDED: set_account_status(seller, AccountStatus.SUSPENDED) resolved = True resolve_report(report, moderator) return resolved DEFAULT_WARNING_MESSAGE = ( "Your listing has been reported for violating CampusCart guidelines. " "Please review your listing. Repeated violations may result in account restrictions." ) def warn_reported_seller(report: Report, moderator: User, *, reason: str | None = None, message: str | None = None): """Send an in-app warning to the reported seller (no email). Reuses the generic notification infrastructure — a single SYSTEM notification titled "Warning from CampusCart", linked to the reported listing. This does **not** change the report's status (a warning is a lighter touch than resolve/suspend); the moderator can still resolve or escalate afterwards. The audit entry is written by the route, like every other moderator action. Returns the created ``Notification``, or ``None`` when the report has no seller to warn. """ from app.notifications.services import MAX_MESSAGE_LENGTH, create_notification seller = report.reported_user if seller is None: return None listing = report.listing reason = (reason or "").strip() or None message = (message or "").strip() or None parts = [] if reason: parts.append(f"Reason: {reason}.") parts.append(message or DEFAULT_WARNING_MESSAGE) if listing is not None: parts.append(f"Reported listing: “{listing.title}”.") parts.append(f"Issued on {utcnow().strftime('%d %b %Y')}.") body = " ".join(parts) if len(body) > MAX_MESSAGE_LENGTH: body = body[: MAX_MESSAGE_LENGTH - 1].rstrip() + "…" return create_notification( seller, NotificationType.SYSTEM, "Warning from CampusCart", body, # Linked to the report (uniform with the other moderation notices) so the # notification opens the Community Guidelines and delivery is detectable. related_entity_type=RelatedEntityType.REPORT, related_entity_id=report.id, ) # --- Moderation outcome notifications -------------------------------------- # # When a review completes, both parties are told the outcome using the shared # notification service. Copy is privacy-safe: the reporter is never told the # specific punishment, only that appropriate action was (or wasn't) taken. # Reporter-facing copy, keyed by the report's final outcome. _REPORTER_NOTICE = { "resolved": ( "Report Reviewed", "Thank you for helping keep Student Bazaar safe. We reviewed your report " "and appropriate moderation action has been taken.", ), "dismissed": ( "Report Reviewed", "We reviewed your report but did not find sufficient evidence to take " "moderation action at this time.", ), } # Reported-user copy, keyed by the concrete action taken against them. _REPORTED_NOTICE = { "warning": ( "Account Warning", "Your account has received an official warning following a moderation " "review. Please review our Community Guidelines to avoid further action.", ), "listing_hidden": ( "Listing Hidden", "One of your listings has been hidden following a moderation review.", ), "listing_removed": ( "Listing Removed", "One of your listings has been removed because it violated our Community Guidelines.", ), "suspended": ( "Account Suspended", "Your account has been temporarily suspended following a moderation review.", ), "banned": ( "Account Restricted", "Your account has been permanently restricted due to serious or repeated violations.", ), } # Moderator action -> (reporter outcome, reported-user action or None). Only the # actions that *complete* a review appear here; ``None`` means the reported user # receives no notification (no concrete action was taken). _ACTION_OUTCOME = { "resolve": ("resolved", None), "dismiss": ("dismissed", None), "remove_listing": ("resolved", "listing_removed"), "suspend_seller": ("resolved", "suspended"), } # Actions that finish a review — the route shows the reporter/reported banner # only for these (assignment / under-review are not completions). REPORT_COMPLETION_ACTIONS = frozenset(_ACTION_OUTCOME) # Makes every moderation notification actionable. Keyed by the notification's # (server-owned) title, so the destination + call-to-action stay in one place and # are shared by the notification router (redirect) and the list UI (button). All # these notifications carry ``related_entity_type=REPORT`` so `open_notification` # resolves them here. Endpoints take no arguments (My Reports / My Listings / # Community Guidelines are all whole-page views). MOD_NOTIFICATION_LINKS = { "Report Reviewed": {"endpoint": "marketplace.my_reports", "label": "View Report", "icon": "bi-flag"}, "Listing Removed": {"endpoint": "marketplace.my_listings", "label": "View My Listings", "icon": "bi-box-seam"}, "Listing Hidden": {"endpoint": "marketplace.my_listings", "label": "View My Listings", "icon": "bi-box-seam"}, "Account Warning": {"endpoint": "main.terms", "label": "View Community Guidelines", "icon": "bi-journal-text"}, "Warning from CampusCart": {"endpoint": "main.terms", "label": "View Community Guidelines", "icon": "bi-journal-text"}, "Account Suspended": {"endpoint": "main.terms", "label": "Learn More", "icon": "bi-info-circle"}, "Account Restricted": {"endpoint": "main.terms", "label": "Learn More", "icon": "bi-info-circle"}, } def moderation_notification_link(notification) -> dict | None: """Return the ``{endpoint, label, icon}`` for a moderation notification, or ``None`` when it isn't one we route. Used by both the router and the list UI.""" if notification.related_entity_type != RelatedEntityType.REPORT: return None return MOD_NOTIFICATION_LINKS.get(notification.title) def _notification_already_sent(user_id: int, title: str, report_id: int) -> bool: """True when this exact moderation notification already exists (per user, title and report) — prevents duplicates on a repeated action/POST.""" return db.session.scalar( db.select(Notification.id).where( Notification.user_id == user_id, Notification.title == title, Notification.related_entity_type == RelatedEntityType.REPORT, Notification.related_entity_id == report_id, ).limit(1) ) is not None def _safe_notify(user, notification_type, title, message, report_id) -> bool: """Send one moderation notification. Never raises: a delivery failure must not roll back the moderation action that already committed (it is only logged). De-duplicated per (user, title, report). Returns whether one was created.""" from app.notifications.services import create_notification if user is None: return False try: if _notification_already_sent(user.id, title, report_id): return False create_notification( user, notification_type, title, message, related_entity_type=RelatedEntityType.REPORT, related_entity_id=report_id, ) return True except Exception: # pragma: no cover - defensive; delivery must never break moderation db.session.rollback() current_app.logger.exception( "Failed to send moderation notification to user %s", getattr(user, "id", None) ) return False def notify_report_parties(report: Report, action: str) -> dict: """Notify the reporter and (only when a concrete action was taken) the reported user after a report review completes. Reuses the shared notification service and never raises — moderation has already committed by the time this runs. Returns a summary the route uses for the moderator confirmation banner: ``{reporter_notified, reported_notified, reported_action}``. """ summary = {"reporter_notified": False, "reported_notified": False, "reported_action": None} outcome = _ACTION_OUTCOME.get(action) if outcome is None: return summary reporter_outcome, reported_action = outcome summary["reported_action"] = reported_action title, message = _REPORTER_NOTICE[reporter_outcome] summary["reporter_notified"] = _safe_notify( report.reporter, NotificationType.REPORT, title, message, report.id ) # Feed activity for the reporter (in addition to the notification). if report.reporter is not None: from app.activity.services import publish_activity publish_activity(report.reporter, "report_resolved", "Report reviewed", message=message, related_entity_type=RelatedEntityType.REPORT, related_entity_id=report.id) if reported_action is not None and report.reported_user is not None: r_title, r_message = _REPORTED_NOTICE[reported_action] summary["reported_notified"] = _safe_notify( report.reported_user, NotificationType.SYSTEM, r_title, r_message, report.id ) return summary # --- Cross-module helpers (badges, counts, dashboard) ---------------------- def get_open_report_listing_ids(items) -> set[int]: """Listing ids (from a page of listings or ids) that have an open report. One query for the whole page — powers the "🚩 Reported" badge without an N+1 lookup per row. """ ids: set[int] = set() for item in items or (): lid = getattr(item, "id", item) if isinstance(lid, int): ids.add(lid) if not ids: return set() rows = db.session.scalars( db.select(Report.listing_id).where( Report.listing_id.in_(ids), Report.status.in_(_OPEN_STATUSES) ) ).all() return {lid for lid in rows if lid is not None} def listing_has_open_report(listing: Listing) -> bool: return bool(get_open_report_listing_ids([listing.id])) def get_user_report_counts(user: User) -> dict: """Open / resolved report counts for a seller (User Management integration).""" row = db.session.execute( db.select( func.coalesce(func.sum(case((Report.status.in_(_OPEN_STATUSES), 1), else_=0)), 0), func.coalesce(func.sum(case((Report.status == ReportStatus.RESOLVED, 1), else_=0)), 0), ).where(Report.reported_user_id == user.id) ).one() return {"open": int(row[0]), "resolved": int(row[1])} def pending_reports_count() -> int: return db.session.scalar( db.select(func.count(Report.id)).where(Report.status == ReportStatus.PENDING) ) or 0 def get_recent_reports(limit: int = 5) -> list[dict]: """Latest reports for the dashboard 'Recent Reports' section.""" reports = db.session.scalars( db.select(Report).options( selectinload(Report.reporter).selectinload(User.profile), selectinload(Report.reported_user).selectinload(User.profile), selectinload(Report.listing), ).order_by(Report.created_at.desc(), Report.id.desc()).limit(limit) ).all() return [_row(r) for r in reports] # --- Analytics ------------------------------------------------------------- def get_moderation_analytics() -> dict: """Moderation analytics via SQL aggregation (constant query count).""" now = utcnow() month_start = datetime.combine(now.date().replace(day=1), time.min) start_date = now.date() - timedelta(days=TREND_WINDOW_DAYS - 1) reports_this_month = db.session.scalar( db.select(func.count(Report.id)).where(Report.created_at >= month_start) ) or 0 # Average resolution time (hours) across closed reports — one query, averaged # in Python for cross-database portability. closed = db.session.execute( db.select(Report.created_at, Report.resolved_at).where(Report.resolved_at.isnot(None)) ).all() avg_resolution_hours = None if closed: deltas = [(res - cre).total_seconds() / 3600 for cre, res in closed] avg_resolution_hours = round(sum(deltas) / len(deltas), 1) reason_row = db.session.execute( db.select(Report.reason, func.count()).group_by(Report.reason).order_by(func.count().desc()) ).first() most_common_reason = reason_label(reason_row[0]) if reason_row else None top_categories = [ (name, int(c)) for name, c in db.session.execute( db.select(Category.name, func.count(Report.id)) .select_from(Report).join(Listing, Report.listing_id == Listing.id) .join(Category, Listing.category_id == Category.id) .group_by(Category.name).order_by(func.count(Report.id).desc()).limit(_TOP_LIMIT) ).all() ] seller_counts = db.session.execute( db.select(Report.reported_user_id, func.count()) .where(Report.reported_user_id.isnot(None)) .group_by(Report.reported_user_id).order_by(func.count().desc()).limit(_TOP_LIMIT) ).all() top_sellers = [] if seller_counts: seller_ids = [sid for sid, _ in seller_counts] users = { u.id: u for u in db.session.scalars( db.select(User).options(selectinload(User.profile)).where(User.id.in_(seller_ids)) ).all() } top_sellers = [ {"name": _name(users.get(sid)), "public_id": users[sid].public_id if sid in users else None, "count": int(c)} for sid, c in seller_counts if sid in users ] # 30-day trend (one GROUP BY, zero-filled). rows = db.session.execute( db.select(func.date(Report.created_at), func.count()) .where(Report.created_at >= datetime.combine(start_date, time.min)) .group_by(func.date(Report.created_at)) ).all() counts = {str(d)[:10]: int(n) for d, n in rows} trend = [ {"date": (start_date + timedelta(days=i)).isoformat(), "count": counts.get((start_date + timedelta(days=i)).isoformat(), 0)} for i in range(TREND_WINDOW_DAYS) ] return { "reports_this_month": int(reports_this_month), "avg_resolution_hours": avg_resolution_hours, "most_common_reason": most_common_reason, "top_categories": top_categories, "top_sellers": top_sellers, "trend": {"labels": [d["date"][5:] for d in trend], "values": [d["count"] for d in trend]}, }