""" Products API routes for categories, websites, products and prices. """ from typing import List, Optional import re import time from decimal import Decimal, InvalidOperation import csv import io from ninja import Router, Query, File from ninja.files import UploadedFile from ninja_jwt.authentication import JWTAuth from ninja.pagination import paginate, PageNumberPagination from django.conf import settings from django.db.models import Count, Min, Max, Q, Prefetch, F from django.db import transaction, IntegrityError from django.shortcuts import get_object_or_404 from django.views.decorators.cache import cache_page from .models import Category, Website, Product, ProductPrice from .schemas import ( CategoryOut, CategoryIn, WebsiteOut, WebsiteIn, WebsiteFilter, ProductOut, ProductIn, ProductWithPricesOut, ProductPriceOut, ProductPriceIn, ProductFilter, ProductSearchFilter, ImportResultOut, MyProductOut, ) from apps.favorites.models import Favorite router = Router() category_router = Router() website_router = Router() # ==================== Category Routes ==================== @category_router.get("/", response=List[CategoryOut]) @cache_page(settings.CACHE_TTL_SECONDS) def list_categories(request): """Get all categories.""" return Category.objects.all() @category_router.get("/{slug}", response=CategoryOut) @cache_page(settings.CACHE_TTL_SECONDS) def get_category_by_slug(request, slug: str): """Get category by slug.""" return get_object_or_404(Category, slug=slug) def require_admin(user): """Check if user is admin.""" from ninja.errors import HttpError if not user or user.role != 'admin' or not user.is_active: raise HttpError(403, "仅管理员可执行此操作") def normalize_category_slug(name: str, slug: str) -> str: """Normalize category slug and ensure it's not empty.""" raw_slug = (slug or "").strip() if raw_slug: return raw_slug base = re.sub(r"\s+", "-", name.strip().lower()) base = re.sub(r"[^a-z0-9-]", "", base) if not base: base = f"category-{int(time.time())}" if Category.objects.filter(slug=base).exists(): suffix = 1 while Category.objects.filter(slug=f"{base}-{suffix}").exists(): suffix += 1 base = f"{base}-{suffix}" return base @category_router.post("/", response=CategoryOut, auth=JWTAuth()) def create_category(request, data: CategoryIn): """Create a new category.""" name = (data.name or "").strip() if not name: raise HttpError(400, "分类名称不能为空") slug = normalize_category_slug(name, data.slug) if len(slug) > 100: raise HttpError(400, "分类标识过长") try: category = Category.objects.create( name=name, slug=slug, description=data.description, icon=data.icon, parent_id=data.parent_id, sort_order=data.sort_order or 0, ) except IntegrityError: raise HttpError(400, "分类标识已存在") return category # ==================== Website Routes ==================== @website_router.get("/", response=List[WebsiteOut]) @paginate(PageNumberPagination, page_size=20) @cache_page(settings.CACHE_TTL_SECONDS) def list_websites(request, filters: WebsiteFilter = Query(...)): """Get all websites with optional filters.""" queryset = Website.objects.all() if filters.category_id: queryset = queryset.filter(category_id=filters.category_id) if filters.is_verified is not None: queryset = queryset.filter(is_verified=filters.is_verified) return queryset @website_router.get("/{website_id}", response=WebsiteOut) @cache_page(settings.CACHE_TTL_SECONDS) def get_website(request, website_id: int): """Get website by ID.""" return get_object_or_404(Website, id=website_id) @website_router.post("/", response=WebsiteOut, auth=JWTAuth()) def create_website(request, data: WebsiteIn): """Create a new website. Any authenticated user can create.""" website = Website.objects.create(**data.dict()) return website # ==================== Product Routes ==================== @router.post("/import/", response=ImportResultOut, auth=JWTAuth()) def import_products_csv(request, file: UploadedFile = File(...)): """Import products/websites/prices from CSV.""" content = file.read() try: text = content.decode("utf-8") except UnicodeDecodeError: text = content.decode("utf-8-sig", errors="ignore") reader = csv.DictReader(io.StringIO(text)) result = ImportResultOut() def parse_bool(value: Optional[str], default=False) -> bool: if value is None: return default val = str(value).strip().lower() if val in ("1", "true", "yes", "y", "on"): return True if val in ("0", "false", "no", "n", "off"): return False return default def parse_decimal(value: Optional[str], field: str, line_no: int) -> Optional[Decimal]: if value in (None, ""): return None try: return Decimal(str(value).strip()) except (InvalidOperation, ValueError): result.errors.append(f"第{line_no}行字段{field}格式错误") return None for idx, row in enumerate(reader, start=2): if not row: continue product_name = (row.get("product_name") or "").strip() category_slug = (row.get("category_slug") or "").strip() category_name = (row.get("category_name") or "").strip() website_name = (row.get("website_name") or "").strip() website_url = (row.get("website_url") or "").strip() price_value = (row.get("price") or "").strip() product_url = (row.get("url") or "").strip() if not product_name or not (category_slug or category_name) or not website_name or not website_url or not price_value or not product_url: result.errors.append(f"第{idx}行缺少必填字段") continue try: with transaction.atomic(): category = None if category_slug: category, created = Category.objects.get_or_create( slug=category_slug, defaults={ "name": category_name or category_slug, "description": row.get("category_desc") or None, }, ) if created: result.created_categories += 1 if not category: category, created = Category.objects.get_or_create( name=category_name, defaults={ "slug": category_slug or category_name, "description": row.get("category_desc") or None, }, ) if created: result.created_categories += 1 website, created = Website.objects.get_or_create( name=website_name, defaults={ "url": website_url, "logo": row.get("website_logo") or None, "description": row.get("website_desc") or None, "category": category, "is_verified": parse_bool(row.get("is_verified"), False), }, ) if created: result.created_websites += 1 else: if website.url != website_url: website.url = website_url if row.get("website_logo"): website.logo = row.get("website_logo") if row.get("website_desc"): website.description = row.get("website_desc") if website.category_id != category.id: website.category = category website.save() product, created = Product.objects.get_or_create( name=product_name, category=category, defaults={ "description": row.get("product_desc") or None, "image": row.get("product_image") or None, }, ) if created: result.created_products += 1 else: updated = False if row.get("product_desc") and product.description != row.get("product_desc"): product.description = row.get("product_desc") updated = True if row.get("product_image") and product.image != row.get("product_image"): product.image = row.get("product_image") updated = True if updated: product.save() price = parse_decimal(row.get("price"), "price", idx) if price is None: continue original_price = parse_decimal(row.get("original_price"), "original_price", idx) currency = (row.get("currency") or "CNY").strip() or "CNY" in_stock = parse_bool(row.get("in_stock"), True) price_record, created = ProductPrice.objects.update_or_create( product=product, website=website, defaults={ "price": price, "original_price": original_price, "currency": currency, "url": product_url, "in_stock": in_stock, }, ) if created: result.created_prices += 1 else: result.updated_prices += 1 except Exception: result.errors.append(f"第{idx}行处理失败") continue return result @router.get("/recommendations/", response=List[ProductOut]) def recommend_products(request, limit: int = 12): """Get recommended products based on favorites or popularity.""" # 限制 limit 最大值 if limit < 1: limit = 1 if limit > 100: limit = 100 user = getattr(request, "auth", None) # 只显示已审核通过的商品 base_queryset = Product.objects.select_related("category").filter(status='approved') if user: favorite_product_ids = list( Favorite.objects.filter(user=user).values_list("product_id", flat=True) ) category_ids = list( Product.objects.filter(id__in=favorite_product_ids, status='approved') .values_list("category_id", flat=True) .distinct() ) if category_ids: base_queryset = base_queryset.filter(category_id__in=category_ids).exclude( id__in=favorite_product_ids ) queryset = ( base_queryset.annotate(favorites_count=Count("favorites", distinct=True)) .order_by("-favorites_count", "-created_at")[:limit] ) return list(queryset) @router.get("/", response=List[ProductOut]) @paginate(PageNumberPagination, page_size=20) @cache_page(settings.CACHE_TTL_SECONDS) def list_products(request, filters: ProductFilter = Query(...)): """Get all approved products with optional filters.""" # 只显示已审核通过的商品 queryset = Product.objects.select_related("category").filter(status='approved') if filters.category_id: queryset = queryset.filter(category_id=filters.category_id) if filters.search: queryset = queryset.filter( Q(name__icontains=filters.search) | Q(description__icontains=filters.search) ) needs_price_stats = ( filters.min_price is not None or filters.max_price is not None or (filters.sort_by or "").lower() in ("price_asc", "price_desc") ) if needs_price_stats: queryset = queryset.annotate(lowest_price=Min("prices__price")) if filters.min_price is not None: queryset = queryset.filter(lowest_price__gte=filters.min_price) if filters.max_price is not None: queryset = queryset.filter(lowest_price__lte=filters.max_price) sort_by = (filters.sort_by or "newest").lower() if sort_by == "oldest": queryset = queryset.order_by("created_at") elif sort_by == "price_asc": queryset = queryset.order_by(F("lowest_price").asc(nulls_last=True), "-created_at") elif sort_by == "price_desc": queryset = queryset.order_by(F("lowest_price").desc(nulls_last=True), "-created_at") else: queryset = queryset.order_by("-created_at") return queryset @router.get("/{product_id}", response=ProductOut) @cache_page(settings.CACHE_TTL_SECONDS) def get_product(request, product_id: int): """Get product by ID.""" return get_object_or_404(Product, id=product_id) @router.get("/{product_id}/with-prices", response=ProductWithPricesOut) @cache_page(settings.CACHE_TTL_SECONDS) def get_product_with_prices(request, product_id: int): """Get product with all prices from different websites.""" product = get_object_or_404(Product, id=product_id) prices = ProductPrice.objects.filter(product=product).select_related('website') price_list = [] for pp in prices: price_list.append(ProductPriceOut( id=pp.id, product_id=pp.product_id, website_id=pp.website_id, website_name=pp.website.name, website_logo=pp.website.logo, price=pp.price, original_price=pp.original_price, currency=pp.currency, url=pp.url, in_stock=pp.in_stock, last_checked=pp.last_checked, )) # Calculate price range price_stats = prices.aggregate( lowest=Min('price'), highest=Max('price') ) return ProductWithPricesOut( id=product.id, name=product.name, description=product.description, image=product.image, category_id=product.category_id, created_at=product.created_at, updated_at=product.updated_at, prices=price_list, lowest_price=price_stats['lowest'], highest_price=price_stats['highest'], ) @router.get("/search/", response=List[ProductWithPricesOut]) @paginate(PageNumberPagination, page_size=20) @cache_page(settings.CACHE_TTL_SECONDS) def search_products(request, q: str, filters: ProductSearchFilter = Query(...)): """Search approved products by name or description.""" prices_prefetch = Prefetch( "prices", queryset=ProductPrice.objects.select_related("website"), ) # 只搜索已审核通过的商品 products = ( Product.objects.select_related("category") .filter(Q(name__icontains=q) | Q(description__icontains=q), status='approved') ) if filters.category_id: products = products.filter(category_id=filters.category_id) needs_price_stats = ( filters.min_price is not None or filters.max_price is not None or (filters.sort_by or "").lower() in ("price_asc", "price_desc") ) if needs_price_stats: products = products.annotate(lowest_price=Min("prices__price")) if filters.min_price is not None: products = products.filter(lowest_price__gte=filters.min_price) if filters.max_price is not None: products = products.filter(lowest_price__lte=filters.max_price) sort_by = (filters.sort_by or "newest").lower() if sort_by == "oldest": products = products.order_by("created_at") elif sort_by == "price_asc": products = products.order_by(F("lowest_price").asc(nulls_last=True), "-created_at") elif sort_by == "price_desc": products = products.order_by(F("lowest_price").desc(nulls_last=True), "-created_at") else: products = products.order_by("-created_at") products = products.prefetch_related(prices_prefetch) result = [] for product in products: prices = list(product.prices.all()) price_list = [ ProductPriceOut( id=pp.id, product_id=pp.product_id, website_id=pp.website_id, website_name=pp.website.name, website_logo=pp.website.logo, price=pp.price, original_price=pp.original_price, currency=pp.currency, url=pp.url, in_stock=pp.in_stock, last_checked=pp.last_checked, ) for pp in prices ] lowest_price = min((pp.price for pp in prices), default=None) highest_price = max((pp.price for pp in prices), default=None) result.append(ProductWithPricesOut( id=product.id, name=product.name, description=product.description, image=product.image, category_id=product.category_id, created_at=product.created_at, updated_at=product.updated_at, prices=price_list, lowest_price=lowest_price, highest_price=highest_price, )) return result @router.post("/", response=ProductOut, auth=JWTAuth()) def create_product(request, data: ProductIn): """Create a new product. Admin creates approved, others create pending.""" user = request.auth is_admin = user and user.role == 'admin' and user.is_active product = Product.objects.create( name=data.name, description=data.description, image=data.image, category_id=data.category_id, status='approved' if is_admin else 'pending', submitted_by=user, ) return product @router.get("/my/", response=List[MyProductOut], auth=JWTAuth()) @paginate(PageNumberPagination, page_size=20) def my_products(request, status: Optional[str] = None): """Get current user's submitted products.""" user = request.auth queryset = Product.objects.filter(submitted_by=user).order_by('-created_at') if status: queryset = queryset.filter(status=status) return queryset @router.post("/prices/", response=ProductPriceOut, auth=JWTAuth()) def add_product_price(request, data: ProductPriceIn): """Add a price for a product. Admin or product owner can add.""" user = request.auth is_admin = user and user.role == 'admin' and user.is_active # 检查商品是否存在并验证权限 product = get_object_or_404(Product, id=data.product_id) if not is_admin and product.submitted_by_id != user.id: from ninja.errors import HttpError raise HttpError(403, "只能为自己提交的商品添加价格") price = ProductPrice.objects.create(**data.dict()) website = price.website return ProductPriceOut( id=price.id, product_id=price.product_id, website_id=price.website_id, website_name=website.name, website_logo=website.logo, price=price.price, original_price=price.original_price, currency=price.currency, url=price.url, in_stock=price.in_stock, last_checked=price.last_checked, )