"""Listings: browse/search/detail + owner CRUD + image management + dev media.""" import os from datetime import datetime from flask import (Blueprint, render_template, request, redirect, url_for, flash, abort, current_app, send_from_directory) from flask_login import login_required, current_user from flask_babel import gettext as _ from app.extensions import db, limiter from app.models.category import Category from app.models.listing import Listing, ListingImage from app.models.enums import ListingStatus, ReportReason from app.services import listings as svc from app.services.geo import geocode_zip from app.services.images import process_upload, delete_image_files, ImageError from app.services import reports as rsvc from app.services import reviews as rev_svc from app.blueprints.listings.forms import ListingForm, ImageUploadForm listings_bp = Blueprint("listings", __name__) PER_PAGE = 20 def _category_choices(): cats = Category.query.filter_by(is_active=True).order_by( Category.sort_order, Category.name).all() return cats def _parse_attributes(category): """Pull attr_ fields from the submitted form.""" raw = {} for f in category.fields(): raw[f["name"]] = request.form.get("attr_" + f["name"]) return raw # --- browse + search --- @listings_bp.route("/listings") def browse(): page = request.args.get("page", 1, type=int) category_id = request.args.get("category", type=int) q = request.args.get("q", type=str) state = request.args.get("state", type=str) zip_code = request.args.get("zip", type=str) radius = request.args.get("radius", type=int) min_price = request.args.get("min_price", type=int) max_price = request.args.get("max_price", type=int) condition = request.args.get("condition", type=str) base = svc.browse_query(category_id=category_id, q=q, state=state, min_price=(min_price * 100 if min_price else None), max_price=(max_price * 100 if max_price else None), condition=condition) base = svc.order_default(base) # promoted listings pinned to top when keyword searched from app.services.ads import promoted_listings as get_promoted promoted = get_promoted(q) if q else [] near = None if zip_code and radius: geo = geocode_zip(zip_code) if geo: lat, lng = geo[0], geo[1] results = svc.search_with_radius(base, lat, lng, radius) total = len(results) start = (page - 1) * PER_PAGE items = results[start:start + PER_PAGE] near = {"zip": zip_code, "radius": radius, "total": total} # prepend promoted (no distance) promoted_pairs = [(l, None) for l in promoted] promoted_ids = {l.id for l in promoted} items = promoted_pairs + [(l, d) for l, d in items if l.id not in promoted_ids] return render_template("listings/browse.html", results=items, near=near, categories=_category_choices(), filters=request.args, page=page, has_next=start + PER_PAGE < total, promoted_ids={l.id for l in promoted}) flash(_("ZIP not found; showing all results."), "warning") pagination = base.paginate(page=page, per_page=PER_PAGE, error_out=False) promoted_ids = {l.id for l in promoted} organic = [(l, None) for l in pagination.items if l.id not in promoted_ids] results = [(l, None) for l in promoted] + organic return render_template("listings/browse.html", results=results, near=None, categories=_category_choices(), filters=request.args, page=page, has_next=pagination.has_next, promoted_ids={l.id for l in promoted}) # --- detail --- @listings_bp.route("/listings/") def detail(listing_id): listing = Listing.query.get_or_404(listing_id) is_owner = current_user.is_authenticated and listing.user_id == current_user.id if not listing.is_live and not (is_owner or (current_user.is_authenticated and current_user.is_moderator)): abort(404) if not is_owner: listing.view_count = (listing.view_count or 0) + 1 db.session.commit() # determine if current user can leave a review can_review = False if (current_user.is_authenticated and not is_owner and listing.status == ListingStatus.sold): from app.models.review import Review already = Review.query.filter_by( listing_id=listing.id, author_id=current_user.id).first() can_review = already is None listing_rating = rev_svc.seller_rating(listing.user_id) return render_template("listings/detail.html", listing=listing, is_owner=is_owner, can_review=can_review, listing_rating=listing_rating) # --- create --- @listings_bp.route("/listings/new", methods=["GET", "POST"]) @login_required @limiter.limit("30 per hour", methods=["POST"]) def create(): if not svc.can_create(current_user): flash(_("You've reached your plan's active-listing limit."), "warning") return redirect(url_for("listings.mine")) form = ListingForm() cats = _category_choices() form.category_id.choices = [(c.id, c.name) for c in cats] category = None if request.method == "POST" and form.category_id.data: category = db.session.get(Category, form.category_id.data) if form.validate_on_submit() and category: raw_attrs = _parse_attributes(category) price_cents = int(form.price.data * 100) if form.price.data is not None else None try: listing = svc.create_listing( current_user, category, title=form.title.data, body=form.body.data, lang=form.lang.data, price_cents=price_cents, zip_code=form.zip.data, raw_attributes=raw_attrs) except svc.ListingError as e: for fld, msg in (e.field_errors or {}).items(): flash(_("%(f)s: %(m)s", f=fld, m=msg), "danger") if not e.field_errors: flash(str(e), "danger") return render_template("listings/form.html", form=form, categories=cats, category=category, listing=None) # process any uploaded images, honoring per-plan cap _save_images(form.images.data, listing) flash(_("Listing published."), "success") return redirect(url_for("listings.detail", listing_id=listing.id)) return render_template("listings/form.html", form=form, categories=cats, category=category, listing=None) # --- edit --- @listings_bp.route("/listings//edit", methods=["GET", "POST"]) @login_required def edit(listing_id): listing = Listing.query.get_or_404(listing_id) if listing.user_id != current_user.id and not current_user.is_moderator: abort(403) cats = _category_choices() form = ListingForm(obj=None) form.category_id.choices = [(c.id, c.name) for c in cats] if request.method == "GET": form.category_id.data = listing.category_id form.title.data = listing.title form.body.data = listing.body form.lang.data = listing.lang.value form.price.data = (listing.price_cents / 100) if listing.price_cents else None form.zip.data = listing.zip category = db.session.get(Category, form.category_id.data or listing.category_id) if form.validate_on_submit() and category: raw_attrs = _parse_attributes(category) price_cents = int(form.price.data * 100) if form.price.data is not None else None try: svc.update_listing(listing, category, title=form.title.data, body=form.body.data, lang=form.lang.data, price_cents=price_cents, zip_code=form.zip.data, raw_attributes=raw_attrs) except svc.ListingError as e: for fld, msg in (e.field_errors or {}).items(): flash(_("%(f)s: %(m)s", f=fld, m=msg), "danger") return render_template("listings/form.html", form=form, categories=cats, category=category, listing=listing) _save_images(form.images.data, listing) flash(_("Listing updated."), "success") return redirect(url_for("listings.detail", listing_id=listing.id)) return render_template("listings/form.html", form=form, categories=cats, category=category, listing=listing) # --- delete --- @listings_bp.route("/listings//delete", methods=["POST"]) @login_required def delete(listing_id): listing = Listing.query.get_or_404(listing_id) if listing.user_id != current_user.id and not current_user.is_moderator: abort(403) for img in list(listing.images): delete_image_files(img) db.session.delete(listing) db.session.commit() flash(_("Listing deleted."), "info") return redirect(url_for("listings.mine")) # --- mark sold --- @listings_bp.route("/listings//sold", methods=["POST"]) @login_required def mark_sold(listing_id): listing = Listing.query.get_or_404(listing_id) if listing.user_id != current_user.id: abort(403) listing.status = ListingStatus.sold db.session.commit() flash(_("Marked as sold."), "success") return redirect(url_for("listings.detail", listing_id=listing.id)) # --- report --- @listings_bp.route("/listings//report", methods=["POST"]) @login_required @limiter.limit("20 per hour", methods=["POST"]) def report(listing_id): listing = Listing.query.get_or_404(listing_id) reason_raw = request.form.get("reason", type=str) note = request.form.get("note", type=str) try: reason = ReportReason(reason_raw) except ValueError: flash(_("Invalid report reason."), "danger") return redirect(url_for("listings.detail", listing_id=listing.id)) try: rsvc.create_report(listing, current_user, reason, note=note) except rsvc.ReportError as e: flash(str(e), "warning") else: flash(_("Thanks — this listing has been reported for review."), "success") return redirect(url_for("listings.detail", listing_id=listing.id)) # --- leave review --- @listings_bp.route("/listings//review", methods=["GET", "POST"]) @login_required def leave_review(listing_id): listing = Listing.query.get_or_404(listing_id) if listing.status != ListingStatus.sold: flash(_("You can only review sold listings."), "warning") return redirect(url_for("listings.detail", listing_id=listing.id)) if listing.user_id == current_user.id: flash(_("You cannot review your own listing."), "warning") return redirect(url_for("listings.detail", listing_id=listing.id)) if request.method == "POST": rating = request.form.get("rating", type=int) body = request.form.get("body", "").strip() try: rev_svc.create_review(listing, current_user, rating, body) flash(_("Review submitted. Thanks!"), "success") return redirect(url_for("listings.detail", listing_id=listing.id)) except rev_svc.ReviewError as e: flash(str(e), "danger") return render_template("listings/review_form.html", listing=listing) # --- my listings --- @listings_bp.route("/my/listings") @login_required def mine(): items = (Listing.query.filter_by(user_id=current_user.id) .order_by(Listing.created_at.desc()).all()) cap = svc._limit(current_user, "active_listings", 3) return render_template("listings/mine.html", items=items, active=svc.active_count(current_user), cap=cap) # --- image delete --- @listings_bp.route("/listings//images//delete", methods=["POST"]) @login_required def delete_image(listing_id, image_id): listing = Listing.query.get_or_404(listing_id) if listing.user_id != current_user.id and not current_user.is_moderator: abort(403) img = ListingImage.query.get_or_404(image_id) if img.listing_id != listing.id: abort(404) delete_image_files(img) db.session.delete(img) db.session.commit() flash(_("Photo removed."), "info") return redirect(url_for("listings.edit", listing_id=listing.id)) # --- dev media serving (Nginx serves /media in prod) --- @listings_bp.route("/media/") def media(rel): root = current_app.config.get("MEDIA_ROOT") or os.path.join( current_app.instance_path, "media") return send_from_directory(root, rel) def _save_images(file_list, listing): if not file_list: return cap = svc.image_cap(current_user) existing = len(listing.images) saved = 0 for fs in file_list: if not getattr(fs, "filename", ""): continue if cap is not None and existing + saved >= cap: flash(_("Photo limit (%(n)s) reached for your plan.", n=cap), "warning") break try: img = process_upload(fs, listing.id, sort_order=existing + saved) db.session.add(img) saved += 1 except ImageError as e: flash(_("Image skipped: %(m)s", m=str(e)), "warning") if saved: db.session.commit()