Features: 1) Add SEO meta utilities and builder functions for schemas; 2) Implement SEO-related APIs in viewsets for categories, brands, and products; 3) Add SeoMeta model and serializer to manage SEO metadata.

Fixes: 1) Resolve annotations compatibility with mypy `type: ignore` adjustments; 2) Correct `resolve_price_with_currency` method to ensure proper float returns; 3) Handle scenarios for empty or null queryset in vendor methods.

Extra: 1) Refactor utility functions for better type annotations and robustness; 2) Minor formatting corrections and removal of redundant mypy strict setting; 3) Improve method return types for consistency.
This commit is contained in:
Egor Pavlovich Gorbunov 2025-08-18 14:26:09 +03:00
parent 45e9ffa143
commit ab10a7a0b7
12 changed files with 335 additions and 31 deletions

View file

@ -7,7 +7,7 @@ from .models import Post, PostTag
@register(Post)
class PostAdmin(SummernoteModelAdminMixin, FieldsetsMixin, ActivationActionsMixin, ModelAdmin):
class PostAdmin(SummernoteModelAdminMixin, FieldsetsMixin, ActivationActionsMixin, ModelAdmin): # type: ignore [misc]
list_display = ("title", "author", "slug", "created", "modified")
list_filter = ("author", "tags", "created", "modified")
search_fields = ("title", "content", "slug")

View file

@ -1,10 +1,10 @@
from graphene import Mutation
class BaseMutation(Mutation):
def __init__(self, *args, **kwargs):
class BaseMutation(Mutation): # type: ignore [misc]
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
@staticmethod
def mutate(**kwargs):
def mutate(**kwargs) -> None:
pass

View file

@ -4,6 +4,8 @@ import logging
from typing import Any, Optional, Self
from constance import config
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from django.contrib.gis.db.models import PointField
from django.contrib.postgres.indexes import GinIndex
from django.core.cache import cache
@ -30,6 +32,8 @@ from django.db.models import (
PositiveIntegerField,
QuerySet,
TextField,
UUIDField,
URLField,
)
from django.db.models.indexes import Index
from django.http import Http404
@ -171,7 +175,7 @@ class Vendor(ExportModelOperationsMixin("vendor"), NiceModel): # type: ignore [
def __str__(self) -> str:
return self.name
def save(self, **kwargs) -> Self:
def save(self, **kwargs) -> None:
users = self.users.filter(is_active=True)
users = users.exclude(attributes__icontains="is_business")
if users.count() > 0:
@ -1418,8 +1422,8 @@ class Order(ExportModelOperationsMixin("order"), NiceModel): # type: ignore [mi
@property
def is_business(self) -> bool:
return (self.user.attributes.get("is_business", False) if self.user else False) or (
self.attributes.get("is_business", False) if self.attributes else False
return (self.attributes.get("is_business", False) if self.attributes else False) or (
self.user.attributes.get("is_business", False) if self.user else False
)
def save(self, **kwargs: dict) -> Self:
@ -2006,3 +2010,20 @@ class Feedback(ExportModelOperationsMixin("feedback"), NiceModel): # type: igno
class Meta:
verbose_name = _("feedback")
verbose_name_plural = _("feedbacks")
class SeoMeta(NiceModel):
uuid = None
content_type = ForeignKey(ContentType, on_delete=CASCADE)
object_id = UUIDField()
content_object = GenericForeignKey("content_type", "object_id")
meta_title = CharField(max_length=70, blank=True)
meta_description = CharField(max_length=180, blank=True)
canonical_override = URLField(blank=True)
robots = CharField(max_length=40, blank=True, default="index,follow")
social_image = ImageField(upload_to="seo/", blank=True, null=True)
extras = JSONField(blank=True, null=True)
class Meta:
unique_together = ("content_type", "object_id")

13
core/serializers/seo.py Normal file
View file

@ -0,0 +1,13 @@
from rest_framework.fields import CharField, DictField, ListField
from rest_framework.serializers import Serializer
class SeoSnapshotSerializer(Serializer):
title = CharField()
description = CharField()
canonical = CharField()
robots = CharField()
hreflang = ListField(child=DictField(), required=False)
open_graph = DictField()
twitter = DictField()
json_ld = ListField(child=DictField())

View file

@ -95,7 +95,7 @@ def set_default_caches_task() -> tuple[bool, str]:
@shared_task(queue="default")
def remove_stale_product_images() -> tuple[bool, str]:
def remove_stale_product_images() -> tuple[bool, str] | None:
"""
Removes stale product images from the products directory by identifying directories
whose names do not match any UUIDs currently present in the database.

View file

@ -27,7 +27,7 @@ def get_random_code() -> str:
return get_random_string(20)
def get_product_uuid_as_path(instance, filename):
def get_product_uuid_as_path(instance, filename: str = "") -> str:
"""
Generates a unique file path for a product using its UUID.
@ -50,7 +50,7 @@ def get_product_uuid_as_path(instance, filename):
return "products" + "/" + str(instance.product.uuid) + "/" + filename
def get_brand_name_as_path(instance, filename):
def get_brand_name_as_path(instance, filename: str = "") -> str:
return "brands/" + str(instance.name) + "/" + filename
@ -90,7 +90,7 @@ def is_url_safe(url: str) -> bool:
return bool(re.match(r"^https://", url, re.IGNORECASE))
def format_attributes(attributes: str | None = None):
def format_attributes(attributes: str | None = None) -> dict:
if not attributes:
return {}
@ -110,7 +110,7 @@ def format_attributes(attributes: str | None = None):
return result
def get_project_parameters():
def get_project_parameters() -> dict:
parameters = cache.get("parameters", {})
if not parameters:
@ -122,7 +122,7 @@ def get_project_parameters():
return parameters
def resolve_translations_for_elasticsearch(instance, field_name):
def resolve_translations_for_elasticsearch(instance, field_name) -> None:
field = getattr(instance, f"{field_name}_{LANGUAGE_CODE}", "")
filled_field = getattr(instance, field_name, "")
if not field:

View file

@ -0,0 +1,98 @@
from constance import config
from django.conf import settings
def org_schema():
return {
"@context": "https://schema.org",
"@type": "Organization",
"name": config.COMPANY_NAME,
"url": f"https://{config.BASE_DOMAIN}/",
"logo": f"https://{config.BASE_DOMAIN}/static/logo.png",
}
def website_schema():
return {
"@context": "https://schema.org",
"@type": "WebSite",
"name": config.PROJECT_NAME,
"url": f"https://{config.BASE_DOMAIN}/",
"potentialAction": {
"@type": "SearchAction",
"target": f"https://{config.BASE_DOMAIN}/search?q={{query}}",
"query-input": "required name=query",
},
}
def breadcrumb_schema(items):
return {
"@context": "https://schema.org",
"@type": "BreadcrumbList",
"itemListElement": [
{"@type": "ListItem", "position": i + 1, "name": name, "item": url} for i, (name, url) in enumerate(items)
],
}
def item_list_schema(urls):
return {
"@context": "https://schema.org",
"@type": "ItemList",
"itemListElement": [{"@type": "ListItem", "position": i + 1, "url": u} for i, u in enumerate(urls)],
}
def product_schema(product, images, rating=None):
offers = []
for stock in product.stocks.all():
offers.append(
{
"@type": "Offer",
"price": round(stock.price, 2),
"priceCurrency": settings.CURRENCY_CODE,
"availability": "https://schema.org/InStock" if stock.quantity > 0 else "https://schema.org/OutOfStock",
"sku": stock.sku,
"url": f"https://{config.BASE_DOMAIN}/product/{product.slug}",
}
)
data = {
"@context": "https://schema.org",
"@type": "Product",
"name": product.name,
"description": product.description or "",
"sku": product.partnumber or "",
"brand": {"@type": "Brand", "name": product.brand.name} if product.brand else None,
"image": [img.image.url for img in images] or [],
"offers": offers[:1] if offers else None,
}
if rating and rating["count"] > 0:
data["aggregateRating"] = {
"@type": "AggregateRating",
"ratingValue": rating["value"],
"reviewCount": rating["count"],
}
return {k: v for k, v in data.items() if v not in (None, [], {})}
def category_schema(category, url):
return {
"@context": "https://schema.org",
"@type": "CollectionPage",
"name": category.name,
"description": category.description or "",
"url": url,
}
def brand_schema(brand, url, logo_url=None):
data = {
"@context": "https://schema.org",
"@type": "Brand",
"name": brand.name,
"url": url,
}
if logo_url:
data["logo"] = logo_url
return data

View file

@ -79,13 +79,15 @@ class AbstractVendor:
instance.
"""
def __init__(self, vendor_name: str = None, currency: str = "USD") -> None:
def __init__(self, vendor_name: str | None = None, currency: str = "USD") -> None:
self.vendor_name = vendor_name
self.currency = currency
self.blocked_attributes = []
self.blocked_attributes: list[Any] = []
@staticmethod
def chunk_data(data: dict = list, num_chunks: int = 20) -> list[dict[Any, Any]] | None:
def chunk_data(data: list[Any] | None = None, num_chunks: int = 20) -> list[list[Any]] | list[Any]:
if not data:
return []
total = len(data)
if total == 0:
return []
@ -226,7 +228,7 @@ class AbstractVendor:
return round(price, 2)
def resolve_price_with_currency(self, price: float | int | Decimal, provider: str, currency: str = ""):
def resolve_price_with_currency(self, price: float | int | Decimal, provider: str, currency: str = "") -> float:
rates = get_rates(provider)
rate = rates.get(currency or self.currency)
@ -234,7 +236,7 @@ class AbstractVendor:
if not rate:
raise RatesError(f"No rate found for {currency or self.currency} in {rates} with probider {provider}...")
return round(price / rate, 2) if rate else round(price, 2)
return float(round(price / rate, 2)) if rate else float(round(price, 2))
@staticmethod
def round_price_marketologically(price: float) -> float:
@ -294,6 +296,8 @@ class AbstractVendor:
def prepare_for_stock_update(self, method: str = "deactivate") -> None:
products = self.get_products_queryset()
if products is None:
return
# noinspection PyUnreachableCode
match method:
@ -316,7 +320,12 @@ class AbstractVendor:
raise ValueError(f"Invalid method {inactivation_method!r} for products cleaner...")
while True:
batch_ids = list(self.get_products_queryset().filter(**filter_kwargs).values_list("pk", flat=True)[:size])
products = self.get_products_queryset()
if products is None:
return
batch_ids = list(products.filter(**filter_kwargs).values_list("pk", flat=True)[:size])
if not batch_ids:
break
with suppress(Exception):
@ -325,11 +334,11 @@ class AbstractVendor:
Product.objects.filter(pk__in=batch_ids).delete()
def delete_belongings(self) -> None:
self.get_products_queryset().delete()
self.get_stocks_queryset().delete()
self.get_attribute_values_queryset().delete()
self.get_products_queryset().delete() # type: ignore [union-attr]
self.get_stocks_queryset().delete() # type: ignore [union-attr]
self.get_attribute_values_queryset().delete() # type: ignore [union-attr]
def process_attribute(self, key: str, value, product: Product, attr_group: AttributeGroup) -> None:
def process_attribute(self, key: str, value: Any, product: Product, attr_group: AttributeGroup) -> None:
if not value:
return
@ -357,12 +366,12 @@ class AbstractVendor:
attribute = Attribute.objects.filter(name=key, group=attr_group).order_by("uuid").first() # type: ignore [assignment]
attribute.is_active = True
attribute.value_type = attr_value_type
attribute.save()
attribute.save() # type: ignore [no-untyped-call]
except IntegrityError:
return
attribute.categories.add(product.category)
attribute.save()
attribute.save() # type: ignore [no-untyped-call]
if not is_created:
return
@ -374,10 +383,10 @@ class AbstractVendor:
defaults={"is_active": True},
)
def update_stock(self):
def update_stock(self) -> None:
pass
def update_order_products_statuses(self):
def update_order_products_statuses(self) -> None:
pass

View file

@ -1,7 +1,10 @@
import logging
import uuid
from contextlib import suppress
from uuid import UUID
from constance import config
from django.conf import settings
from django.db.models import Q, Prefetch
from django.http import Http404
from django.shortcuts import get_object_or_404
@ -101,9 +104,19 @@ from core.serializers import (
WishlistDetailSerializer,
WishlistSimpleSerializer,
)
from core.serializers.seo import SeoSnapshotSerializer
from core.utils import format_attributes
from core.utils.messages import permission_denied_message
from core.utils.nominatim import fetch_address_suggestions
from core.utils.seo_builders import (
org_schema,
website_schema,
breadcrumb_schema,
product_schema,
category_schema,
item_list_schema,
brand_schema,
)
from evibes.settings import DEBUG
from payments.serializers import TransactionProcessSerializer
@ -277,6 +290,64 @@ class CategoryViewSet(EvibesViewSet):
return qs
return qs.filter(is_active=True)
@action(detail=True, methods=["get"], url_path="seo")
def seo(self, request, **kwargs):
lookup_key = getattr(self, "lookup_url_kwarg", "pk")
lookup_val = kwargs.get(lookup_key)
category = get_object_or_404(Category.objects.select_related("parent"), slug=str(lookup_val))
with suppress(Exception):
category = Category.objects.select_related("parent").get(uuid=UUID(str(lookup_val)))
title = f"{category.name} | {config.PROJECT_NAME}"
description = (category.description or "")[:180]
canonical = f"https://{config.BASE_DOMAIN}/{settings.LANGUAGE_CODE}/catalog/{category.slug}"
og = {
"title": title,
"description": description,
"type": "website",
"url": canonical,
"image": category.image.url if getattr(category, "image", None) else "",
}
tw = {"card": "summary_large_image", "title": title, "description": description}
crumbs = [("Home", f"https://{config.BASE_DOMAIN}/")]
if category.get_ancestors().exists():
for c in category.get_ancestors():
crumbs.append((c.name, f"https://{config.BASE_DOMAIN}/{settings.LANGUAGE_CODE}/catalog/{c.slug}"))
crumbs.append((category.name, canonical))
json_ld = [org_schema(), website_schema(), breadcrumb_schema(crumbs), category_schema(category, canonical)]
product_urls = []
qs = (
Product.objects.filter(
is_active=True,
category=category,
brand__is_active=True,
stocks__vendor__is_active=True,
)
.only("slug")
.distinct()[:24]
)
for p in qs:
product_urls.append(f"https://{config.BASE_DOMAIN}/{settings.LANGUAGE_CODE}/product/{p.slug}")
if product_urls:
json_ld.append(item_list_schema(product_urls))
payload = {
"title": title,
"description": description,
"canonical": canonical,
"robots": "index,follow",
"open_graph": og,
"twitter": tw,
"json_ld": json_ld,
}
return Response(SeoSnapshotSerializer(payload).data)
class BrandViewSet(EvibesViewSet):
"""
@ -320,6 +391,56 @@ class BrandViewSet(EvibesViewSet):
return queryset
@action(detail=True, methods=["get"], url_path="seo")
def seo(self, request, **kwargs):
lookup_key = getattr(self, "lookup_url_kwarg", "pk")
lookup_val = kwargs.get(lookup_key)
brand = get_object_or_404(Brand, slug=str(lookup_val))
with suppress(Exception):
brand = Brand.objects.get(uuid=UUID(str(lookup_val)))
title = f"{brand.name} | {config.PROJECT_NAME}"
description = (brand.description or "")[:180]
canonical = f"https://{config.BASE_DOMAIN}/{settings.LANGUAGE_CODE}/brand/{brand.slug}"
logo_url = (
brand.big_logo.url
if getattr(brand, "big_logo", None)
else (brand.small_logo.url if getattr(brand, "small_logo", None) else None)
)
og = {
"title": title,
"description": description,
"type": "website",
"url": canonical,
"image": logo_url or "",
}
tw = {"card": "summary_large_image", "title": title, "description": description}
crumbs = [
("Home", f"https://{config.BASE_DOMAIN}/"),
(brand.name, canonical),
]
json_ld = [
org_schema(),
website_schema(),
breadcrumb_schema(crumbs),
brand_schema(brand, canonical, logo_url=logo_url),
]
payload = {
"title": title,
"description": description,
"canonical": canonical,
"robots": "index,follow",
"open_graph": og,
"twitter": tw,
"json_ld": json_ld,
}
return Response(SeoSnapshotSerializer(payload).data)
@extend_schema_view(**PRODUCT_SCHEMA)
class ProductViewSet(EvibesViewSet):
@ -406,6 +527,47 @@ class ProductViewSet(EvibesViewSet):
name = "Product"
return Response(status=status.HTTP_404_NOT_FOUND, data={"detail": _(f"{name} does not exist: {uuid}")})
@action(detail=True, methods=["get"], url_path="seo")
def seo(self, request, slug):
p = get_object_or_404(Product.objects.select_related("brand", "category"), slug=slug)
images = list(p.images.all()[:6])
rating = {"value": p.rating, "count": p.feedbacks_count}
title = f"{p.name} | {config.PROJECT_NAME}"
description = (p.description or "")[:180]
canonical = f"https://{config.BASE_DOMAIN}/products/{p.slug}"
og = {
"title": title,
"description": description,
"type": "product",
"url": canonical,
"image": images[0].image.url if images else "",
}
tw = {"card": "summary_large_image", "title": title, "description": description}
crumbs = []
if p.category:
crumbs.append(("Home", f"https://{config.BASE_DOMAIN}/"))
for c in p.category.get_ancestors(include_self=True):
crumbs.append((c.name, f"https://{config.BASE_DOMAIN}/c/{c.slug}"))
crumbs.append((p.name, canonical))
json_ld = [org_schema(), website_schema()]
if crumbs:
json_ld.append(breadcrumb_schema(crumbs))
json_ld.append(product_schema(p, images, rating=rating))
payload = {
"title": title,
"description": description,
"canonical": canonical,
"robots": "index,follow",
"open_graph": og,
"twitter": tw,
"json_ld": json_ld,
}
return Response(SeoSnapshotSerializer(payload).data)
class VendorViewSet(EvibesViewSet):
"""
@ -663,7 +825,7 @@ class OrderViewSet(EvibesViewSet):
serializer.is_valid(raise_exception=True)
lookup_val = kwargs.get(self.lookup_field)
try:
order = Order.objects.get(uuid=lookup_val)
order = Order.objects.get(uuid=str(lookup_val))
if not (request.user.has_perm("core.add_orderproduct") or request.user == order.user):
raise PermissionDenied(permission_denied_message)

View file

@ -4,6 +4,8 @@ from storages.backends.ftp import FTPStorage
class AbsoluteFTPStorage(FTPStorage): # type: ignore
# noinspection PyProtectedMember
def _get_config(self):
cfg = super()._get_config()
url = urlparse(self.location)

View file

@ -21,7 +21,7 @@ class TransactionInline(admin.TabularInline):
@register(Balance)
class BalanceAdmin(ActivationActionsMixin, ModelAdmin):
class BalanceAdmin(ActivationActionsMixin, ModelAdmin): # type: ignore [misc]
inlines = (TransactionInline,)
list_display = ("user", "amount")
search_fields = ("user__email",)
@ -33,7 +33,7 @@ class BalanceAdmin(ActivationActionsMixin, ModelAdmin):
@register(Transaction)
class TransactionAdmin(ActivationActionsMixin, ModelAdmin):
class TransactionAdmin(ActivationActionsMixin, ModelAdmin): # type: ignore [misc]
list_display = ("balance", "amount", "order", "modified", "created")
search_fields = ("balance__user__email", "currency", "payment_method")
list_filter = ("currency", "payment_method")

View file

@ -103,7 +103,6 @@ testing = ["pytest", "pytest-django", "coverage"]
linting = ["black", "isort", "flake8", "bandit"]
[tool.mypy]
strict = true
disable_error_code = ["no-redef", "import-untyped"]
exclude = ["*/migrations/*", "storefront/*"]
plugins = ["mypy_django_plugin.main", "mypy_drf_plugin.main"]