diff --git a/.git-blame-ignore-revs b/.git-blame-ignore-revs index 637f2b6..a569984 100644 --- a/.git-blame-ignore-revs +++ b/.git-blame-ignore-revs @@ -1,5 +1,2 @@ # Changed line-length back to default 1118c03c1b21e217bb66ee2811c423fe3624d546 - -# Ruff ignoring when from IDs to human-readable names -1424978854b50d8b8ab5fd1b7a1d36292524b521 diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 00baaa9..f53995d 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -21,13 +21,13 @@ repos: - id: trailing-whitespace - repo: https://github.com/adamchainz/django-upgrade - rev: 1.31.1 + rev: 1.30.0 hooks: - id: django-upgrade args: [--target-version, "6.0"] - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.15.22 + rev: v0.15.17 hooks: - id: ruff-check args: ["--fix", "--exit-non-zero-on-fix"] diff --git a/chzzk/management/commands/import_chzzk_campaign.py b/chzzk/management/commands/import_chzzk_campaign.py index d39449a..c8944ca 100644 --- a/chzzk/management/commands/import_chzzk_campaign.py +++ b/chzzk/management/commands/import_chzzk_campaign.py @@ -144,7 +144,7 @@ class Command(BaseCommand): Args: campaign_no (int): The campaign number to import. """ - api_version: str = "v2" # TODO(TheLovinator): Add support for v1 API # ruff:ignore[missing-todo-link] + api_version: str = "v2" # TODO(TheLovinator): Add support for v1 API # noqa: TD003 url: str = f"https://api.chzzk.naver.com/service/{api_version}/drops/campaigns/{campaign_no}" resp: requests.Response = requests.get( url, diff --git a/chzzk/tasks.py b/chzzk/tasks.py index a2ef415..55f929c 100644 --- a/chzzk/tasks.py +++ b/chzzk/tasks.py @@ -9,7 +9,7 @@ logger: logging.Logger = logging.getLogger("ttvdrops.tasks") @shared_task(bind=True, queue="imports", max_retries=3, default_retry_delay=60) -def import_chzzk_campaign_task(self, campaign_no: int) -> None: # ruff:ignore[missing-type-function-argument] +def import_chzzk_campaign_task(self, campaign_no: int) -> None: # noqa: ANN001 """Import a single Chzzk campaign by its campaign number.""" try: call_command("import_chzzk_campaign", str(campaign_no)) @@ -18,7 +18,7 @@ def import_chzzk_campaign_task(self, campaign_no: int) -> None: # ruff:ignore[m @shared_task(bind=True, queue="api-fetches", max_retries=3, default_retry_delay=120) -def discover_chzzk_campaigns(self) -> None: # ruff:ignore[missing-type-function-argument] +def discover_chzzk_campaigns(self) -> None: # noqa: ANN001 """Discover and import the latest Chzzk campaigns (equivalent to --latest flag).""" try: call_command("import_chzzk_campaign", latest=True) diff --git a/config/api.py b/config/api.py deleted file mode 100644 index 1d96550..0000000 --- a/config/api.py +++ /dev/null @@ -1,19 +0,0 @@ -"""Combined v1 API router for all platforms.""" - -from ninja import NinjaAPI - -from kick.api import api as kick_router -from twitch.api import api as twitch_router - -api_v1 = NinjaAPI( - title="TTVDrops API", - version="1.0.0", - urls_namespace="api-v1", - docs_url="docs/", -) - -api_v1.add_router("/twitch/", twitch_router, url_name_prefix="twitch-api-v1") -api_v1.add_router("/kick/", kick_router, url_name_prefix="kick-api-v1") - -app_name = "api-v1" -urlpatterns = api_v1.urls[0] diff --git a/config/celery.py b/config/celery.py index 503b47d..129ab98 100644 --- a/config/celery.py +++ b/config/celery.py @@ -23,10 +23,10 @@ classes = [ ] for cls in classes: - setattr( # ruff:ignore[set-attr-with-constant] + setattr( # noqa: B010 cls, "__class_getitem__", - classmethod(lambda cls, *args, **kwargs: cls), # ruff:ignore[unused-lambda-argument] + classmethod(lambda cls, *args, **kwargs: cls), # noqa: ARG005 ) diff --git a/config/tests/test_celery.py b/config/tests/test_celery.py index 332f086..edb0a3d 100644 --- a/config/tests/test_celery.py +++ b/config/tests/test_celery.py @@ -16,7 +16,7 @@ def celery_app() -> Generator[Celery, Any]: Yields: Celery: A Celery app instance configured for testing. """ - with patch("os.environ.setdefault") as mock_setenv: # ruff:ignore[unused-variable] + with patch("os.environ.setdefault") as mock_setenv: # noqa: F841 app = Celery("config") app.config_from_object("django.conf:settings", namespace="CELERY") yield app diff --git a/config/tests/test_site_endpoint_smoke.py b/config/tests/test_site_endpoint_smoke.py index 2cc7b54..8814395 100644 --- a/config/tests/test_site_endpoint_smoke.py +++ b/config/tests/test_site_endpoint_smoke.py @@ -276,25 +276,6 @@ class SiteEndpointSmokeTest(TestCase): ("twitch:export_organizations_csv", {}, 200), ("twitch:export_organizations_json", {}, 200), # Kick endpoints - ("api-v1:kick-api-v1_list_campaigns", {}, 200), - ( - "api-v1:kick-api-v1_get_campaign", - {"kick_id": self.kick_campaign.kick_id}, - 200, - ), - ("api-v1:kick-api-v1_list_games", {}, 200), - ( - "api-v1:kick-api-v1_get_game", - {"kick_id": self.kick_category.kick_id}, - 200, - ), - ("api-v1:kick-api-v1_list_organizations", {}, 200), - ( - "api-v1:kick-api-v1_get_organization", - {"kick_id": self.kick_org.kick_id}, - 200, - ), - ("api-v1:kick-api-v1_stats", {}, 200), ("kick:dashboard", {}, 200), ("kick:campaign_list", {}, 200), ("kick:campaign_detail", {"kick_id": self.kick_campaign.kick_id}, 200), diff --git a/config/urls.py b/config/urls.py index be4ce90..f4d56ad 100644 --- a/config/urls.py +++ b/config/urls.py @@ -43,11 +43,6 @@ urlpatterns: list[URLPattern | URLResolver] = [ view=core_views.sitemap_youtube_view, name="sitemap-youtube", ), - # API v1 - path( - "api/v1/", - include("config.api", namespace="api-v1"), - ), # Core app path(route="", view=include("core.urls", namespace="core")), # Twitch app diff --git a/conftest.py b/conftest.py index 883c39d..e3b4683 100644 --- a/conftest.py +++ b/conftest.py @@ -7,7 +7,7 @@ if TYPE_CHECKING: from collections.abc import Generator -@pytest.fixture(autouse=True) +@pytest.fixture(autouse=True) # noqa: RUF076 — intentional: project-wide N+1 detection with @pytest.mark.no_zeal escape hatch def use_zeal(request: pytest.FixtureRequest) -> Generator[None]: """Enable Zeal N+1 detection context for each pytest test. diff --git a/core/tasks.py b/core/tasks.py index c26f5ae..cf7f6f0 100644 --- a/core/tasks.py +++ b/core/tasks.py @@ -9,7 +9,7 @@ logger: logging.Logger = logging.getLogger("ttvdrops.tasks") @shared_task(bind=True, queue="default", max_retries=3, default_retry_delay=300) -def submit_indexnow_task(self) -> None: # ruff:ignore[missing-type-function-argument] +def submit_indexnow_task(self) -> None: # noqa: ANN001 """Submit all site URLs to the IndexNow search index.""" try: call_command("submit_indexnow") diff --git a/core/tests/test_sitemaps.py b/core/tests/test_sitemaps.py index e6093fa..7821279 100644 --- a/core/tests/test_sitemaps.py +++ b/core/tests/test_sitemaps.py @@ -1,4 +1,4 @@ -import xml.etree.ElementTree as ET # ruff:ignore[suspicious-xml-etree-import] +import xml.etree.ElementTree as ET # noqa: S405 from typing import TYPE_CHECKING from django.urls import reverse @@ -9,7 +9,7 @@ if TYPE_CHECKING: def _extract_locs(xml_bytes: bytes) -> list[str]: - root = ET.fromstring(xml_bytes) # ruff:ignore[suspicious-xml-element-tree-usage] + root = ET.fromstring(xml_bytes) # noqa: S314 ns = {"s": "http://www.sitemaps.org/schemas/sitemap/0.9"} return [el.text for el in root.findall(".//s:loc", ns) if el.text] diff --git a/core/utils.py b/core/utils.py index 863dfca..2adc70f 100644 --- a/core/utils.py +++ b/core/utils.py @@ -1,7 +1,7 @@ from typing import Any -from xml.etree.ElementTree import Element # ruff:ignore[suspicious-xml-etree-import] -from xml.etree.ElementTree import SubElement # ruff:ignore[suspicious-xml-etree-import] -from xml.etree.ElementTree import tostring # ruff:ignore[suspicious-xml-etree-import] +from xml.etree.ElementTree import Element # noqa: S405 +from xml.etree.ElementTree import SubElement # noqa: S405 +from xml.etree.ElementTree import tostring # noqa: S405 from django.conf import settings diff --git a/core/views.py b/core/views.py index e80920e..6b20b55 100644 --- a/core/views.py +++ b/core/views.py @@ -53,7 +53,7 @@ MIN_SEARCH_RANK = 0.05 DEFAULT_SITE_DESCRIPTION = "Archive of Twitch drops, campaigns, rewards, and more." -def _build_seo_context( # ruff:ignore[too-many-arguments, too-many-positional-arguments] +def _build_seo_context( # noqa: PLR0913, PLR0917 page_title: str = "ttvdrops", page_description: str | None = None, page_url: str | None = None, @@ -91,7 +91,7 @@ def _build_seo_context( # ruff:ignore[too-many-arguments, too-many-positional-a if page_url and not page_url.startswith("http"): page_url = f"{settings.BASE_URL}{page_url}" - # TODO(TheLovinator): Instead of having so many parameters, # ruff:ignore[missing-todo-link] + # TODO(TheLovinator): Instead of having so many parameters, # noqa: TD003 # consider having a single "seo_info" parameter that # can contain all of these optional fields. This would make # it easier to extend in the future without changing the @@ -533,7 +533,7 @@ def docs_rss_view(request: HttpRequest) -> HttpResponse: # MARK: /debug/ -def debug_view(request: HttpRequest) -> HttpResponse: # ruff:ignore[too-many-locals] +def debug_view(request: HttpRequest) -> HttpResponse: # noqa: PLR0914 """Debug view showing potentially broken or inconsistent data. Returns: @@ -705,28 +705,6 @@ def debug_view(request: HttpRequest) -> HttpResponse: # ruff:ignore[too-many-lo miner_only_rewards: QuerySet[RewardCampaign] = RewardCampaign.objects.filter( twitch_id__in=miner_only_reward_ids, ).order_by("name")[:200] - - # ── Unknown-status campaigns ──────────────────────────────────────── - # Kick campaigns with incomplete dates - kick_unknown_campaigns: QuerySet[KickDropCampaign] = ( - KickDropCampaign.objects - .filter( - Q(starts_at__isnull=True) | Q(ends_at__isnull=True), - is_fully_imported=True, - ) - .select_related("category", "organization") - .order_by("name") - ) - # Twitch reward campaigns with incomplete dates - twitch_reward_unknown_campaigns: QuerySet[RewardCampaign] = ( - RewardCampaign.objects - .filter( - Q(starts_at__isnull=True) | Q(ends_at__isnull=True), - ) - .select_related("game") - .order_by("name") - ) - context: dict[str, Any] = { "now": now, "games_without_owner": games_without_owner, @@ -747,9 +725,6 @@ def debug_view(request: HttpRequest) -> HttpResponse: # ruff:ignore[too-many-lo "miner_only_rewards_count": len(miner_only_reward_ids), "sunkibot_only_rewards": sunkibot_only_rewards, "miner_only_rewards": miner_only_rewards, - # Unknown-status campaigns - "kick_unknown_campaigns": kick_unknown_campaigns, - "twitch_reward_unknown_campaigns": twitch_reward_unknown_campaigns, } seo_context: dict[str, Any] = _build_seo_context( @@ -772,8 +747,8 @@ def dataset_backups_view(request: HttpRequest) -> HttpResponse: Returns: HttpResponse: The rendered dataset backups page. """ - # TODO(TheLovinator): Instead of only using sql we should also support other formats like parquet, csv, or json. # ruff:ignore[missing-todo-link] - # TODO(TheLovinator): Upload to s3 instead. # ruff:ignore[missing-todo-link] + # TODO(TheLovinator): Instead of only using sql we should also support other formats like parquet, csv, or json. # noqa: TD003 + # TODO(TheLovinator): Upload to s3 instead. # noqa: TD003 # TODO(TheLovinator): https://developers.google.com/search/docs/appearance/structured-data/dataset#json-ld datasets_root: Path = settings.DATA_DIR / "datasets" search_dirs: list[Path] = [datasets_root] @@ -889,7 +864,7 @@ def dataset_backup_download_view( Raises: Http404: When the file is not found or is outside the data directory. """ - # TODO(TheLovinator): Use s3 instead of local disk. # ruff:ignore[missing-todo-link] + # TODO(TheLovinator): Use s3 instead of local disk. # noqa: TD003 datasets_root: Path = settings.DATA_DIR / "datasets" requested_path: Path = (datasets_root / relative_path).resolve() @@ -994,7 +969,7 @@ def search_view(request: HttpRequest) -> HttpResponse: total_results_count: int = sum(len(qs) for qs in results.values()) - # TODO(TheLovinator): Make the description more informative by including counts of each result type, e.g. "Found 5 games, 3 campaigns, and 10 drops for 'rust'." # ruff:ignore[missing-todo-link] + # TODO(TheLovinator): Make the description more informative by including counts of each result type, e.g. "Found 5 games, 3 campaigns, and 10 drops for 'rust'." # noqa: TD003 if query: page_title: str = f"Search Results for '{query}'"[:60] page_description: str = f"Found {total_results_count} results for '{query}'." diff --git a/kick/api.py b/kick/api.py deleted file mode 100644 index 50b0065..0000000 --- a/kick/api.py +++ /dev/null @@ -1,610 +0,0 @@ -from __future__ import annotations - -import datetime # ruff: ignore[typing-only-standard-library-import] -from typing import TYPE_CHECKING -from typing import Literal - -from django.db import models -from django.http import Http404 -from django.shortcuts import get_object_or_404 -from django.utils import timezone -from ninja import Router -from ninja import Schema - -from kick.models import KickCategory -from kick.models import KickChannel -from kick.models import KickDropCampaign -from kick.models import KickOrganization -from kick.models import KickUser - -if TYPE_CHECKING: - from django.db.models import QuerySet - from django.http import HttpRequest - - from kick.models import KickReward - -CampaignStatus = Literal["active", "upcoming", "expired", "unknown"] -CampaignStatusFilter = Literal["active", "upcoming", "expired", "unknown"] -VALID_STATUS_FILTERS: frozenset[str] = frozenset({ - "active", - "upcoming", - "expired", - "unknown", -}) -DEFAULT_PAGE_SIZE = 100 -MAX_PAGE_SIZE = 500 - -api = Router() - - -# MARK: Schemas -class V1StatsSchema(Schema): - """Aggregate counts for Kick drops.""" - - total_campaigns: int - active: int - upcoming: int - expired: int - partial_dates: int - total_organizations: int - total_categories: int - total_channels: int - - -class V1CategorySchema(Schema): - """Kick category (game) nested in a campaign.""" - - kick_id: int - name: str - slug: str - image_url: str - - -class V1OrganizationSummarySchema(Schema): - """Kick organization nested in a campaign.""" - - kick_id: str - name: str - logo_url: str - url: str - - -class V1UserSchema(Schema): - """Kick user nested in a channel.""" - - kick_id: int - username: str - profile_picture: str - - -class V1ChannelSchema(Schema): - """Kick channel participating in a drop campaign.""" - - kick_id: int - slug: str - url: str - user: V1UserSchema | None - - -class V1RewardSchema(Schema): - """Reward earned from a Kick drop campaign.""" - - kick_id: str - name: str - image_url: str - required_minutes_watched: int - - -class V1CampaignSummarySchema(Schema): - """Compact Kick drop campaign for list views.""" - - kick_id: str - name: str - status: str - image_url: str - starts_at: datetime.datetime | None - ends_at: datetime.datetime | None - category: V1CategorySchema | None - organization: V1OrganizationSummarySchema | None - reward_count: int - is_fully_imported: bool - - -class V1CampaignDetailSchema(V1CampaignSummarySchema): - """Full Kick drop campaign detail.""" - - connect_url: str - url: str - channels: list[V1ChannelSchema] - rewards: list[V1RewardSchema] - added_at: datetime.datetime | None - updated_at: datetime.datetime | None - - -class V1PaginationSchema(Schema): - """Common pagination fields.""" - - total: int - page: int - page_size: int - - -class V1CampaignListSchema(V1PaginationSchema): - """Paginated campaign list.""" - - items: list[V1CampaignDetailSchema] - - -class V1OrganizationListItemSchema(Schema): - """Kick organization in a paginated list.""" - - kick_id: str - name: str - logo_url: str - url: str - campaign_count: int - active_campaign_count: int - - -class V1OrganizationListSchema(V1PaginationSchema): - """Paginated organization list.""" - - items: list[V1OrganizationListItemSchema] - - -class V1OrganizationDetailSchema(Schema): - """Kick organization with its campaigns.""" - - kick_id: str - name: str - logo_url: str - url: str - campaign_count: int - active_campaign_count: int - campaigns: list[V1CampaignSummarySchema] - - -class V1CategoryListItemSchema(Schema): - """Kick category (game) in a paginated list.""" - - kick_id: int - name: str - slug: str - image_url: str - campaign_count: int - active_campaign_count: int - - -class V1CategoryListSchema(V1PaginationSchema): - """Paginated category list.""" - - items: list[V1CategoryListItemSchema] - - -class V1CategoryDetailSchema(Schema): - """Kick category (game) with its campaigns.""" - - kick_id: int - name: str - slug: str - image_url: str - campaign_count: int - active_campaign_count: int - campaigns: list[V1CampaignSummarySchema] - - -# MARK: Helpers - - -def _paginate[ModelT: models.Model]( - queryset: QuerySet[ModelT, ModelT], - *, - page: int, - page_size: int, -) -> tuple[list[ModelT], int, int, int]: - page = max(page, 1) - page_size = min(max(page_size, 1), MAX_PAGE_SIZE) - offset = (page - 1) * page_size - total = queryset.count() - items = list(queryset[offset : offset + page_size]) if offset < total else [] - return items, total, page, page_size - - -def _campaign_status( - campaign: KickDropCampaign, - now: datetime.datetime, -) -> CampaignStatus: - if campaign.starts_at and campaign.ends_at: - if campaign.starts_at <= now <= campaign.ends_at: - return "active" - if campaign.starts_at > now: - return "upcoming" - return "expired" - return "unknown" - - -def _apply_status_filter( - queryset: QuerySet[KickDropCampaign], - status: str | None, - now: datetime.datetime, -) -> QuerySet[KickDropCampaign]: - if status == "active": - return queryset.filter(starts_at__lte=now, ends_at__gte=now) - if status == "upcoming": - return queryset.filter(starts_at__gt=now, ends_at__isnull=False) - if status == "expired": - return queryset.filter(starts_at__lte=now, ends_at__lt=now) - if status == "unknown": - return queryset.filter( - models.Q(starts_at__isnull=True) | models.Q(ends_at__isnull=True), - ) - return queryset - - -# MARK: Serializers -def _serialize_category(category: KickCategory | None) -> V1CategorySchema | None: - if category is None: - return None - return V1CategorySchema( - kick_id=category.kick_id, - name=category.name, - slug=category.slug, - image_url=category.image_url, - ) - - -def _serialize_organization_summary( - organization: KickOrganization | None, -) -> V1OrganizationSummarySchema | None: - if organization is None: - return None - return V1OrganizationSummarySchema( - kick_id=organization.kick_id, - name=organization.name, - logo_url=organization.logo_url, - url=organization.url, - ) - - -def _serialize_user(user: KickUser | None) -> V1UserSchema | None: - if user is None: - return None - return V1UserSchema( - kick_id=user.kick_id, - username=user.username, - profile_picture=user.profile_picture, - ) - - -def _serialize_channel(channel: KickChannel) -> V1ChannelSchema: - return V1ChannelSchema( - kick_id=channel.kick_id, - slug=channel.slug, - url=channel.channel_url, - user=_serialize_user(channel.user), - ) - - -def _serialize_reward(reward: KickReward) -> V1RewardSchema: - return V1RewardSchema( - kick_id=reward.kick_id, - name=reward.name, - image_url=reward.full_image_url, - required_minutes_watched=reward.required_units, - ) - - -def _serialize_campaign_summary( - campaign: KickDropCampaign, - now: datetime.datetime, -) -> V1CampaignSummarySchema: - return V1CampaignSummarySchema( - kick_id=campaign.kick_id, - name=campaign.name, - status=_campaign_status(campaign, now), - image_url=campaign.image_url, - starts_at=campaign.starts_at, - ends_at=campaign.ends_at, - category=_serialize_category(campaign.category), - organization=_serialize_organization_summary(campaign.organization), - reward_count=campaign.rewards.count(), # pyright: ignore[reportAttributeAccessIssue] - is_fully_imported=campaign.is_fully_imported, - ) - - -def _serialize_campaign( - campaign: KickDropCampaign, - now: datetime.datetime, -) -> V1CampaignDetailSchema: - return V1CampaignDetailSchema( - kick_id=campaign.kick_id, - name=campaign.name, - status=_campaign_status(campaign, now), - image_url=campaign.image_url, - starts_at=campaign.starts_at, - ends_at=campaign.ends_at, - category=_serialize_category(campaign.category), - organization=_serialize_organization_summary(campaign.organization), - reward_count=campaign.rewards.count(), # pyright: ignore[reportAttributeAccessIssue] - is_fully_imported=campaign.is_fully_imported, - connect_url=campaign.connect_url, - url=campaign.url, - channels=[_serialize_channel(channel) for channel in campaign.channels.all()], - rewards=[ - _serialize_reward(reward) - for reward in campaign.rewards.all() # pyright: ignore[reportAttributeAccessIssue] - ], - added_at=campaign.added_at, - updated_at=campaign.updated_at, - ) - - -# MARK: Endpoints - - -@api.get("/", response=V1StatsSchema) -def stats(request: HttpRequest) -> V1StatsSchema: - """Return aggregate counts for Kick drops.""" - now: datetime.datetime = timezone.now() - total_campaigns: int = KickDropCampaign.objects.filter( - is_fully_imported=True, - ).count() - active: int = KickDropCampaign.objects.filter( - is_fully_imported=True, - starts_at__lte=now, - ends_at__gte=now, - ).count() - upcoming: int = KickDropCampaign.objects.filter( - is_fully_imported=True, - starts_at__gt=now, - ends_at__isnull=False, - ).count() - expired: int = KickDropCampaign.objects.filter( - is_fully_imported=True, - starts_at__lte=now, - ends_at__lt=now, - ).count() - - return V1StatsSchema( - total_campaigns=total_campaigns, - active=active, - upcoming=upcoming, - expired=expired, - partial_dates=total_campaigns - active - upcoming - expired, - total_organizations=KickOrganization.objects.count(), - total_categories=KickCategory.objects.count(), - total_channels=KickChannel.objects.count(), - ) - - -@api.get("/campaigns/", response=V1CampaignListSchema) -def list_campaigns( # ruff:ignore[too-many-positional-arguments] - request: HttpRequest, - page: int = 1, - page_size: int = DEFAULT_PAGE_SIZE, - status: CampaignStatusFilter | None = None, - game: int | None = None, - organization: str | None = None, - search: str | None = None, -) -> V1CampaignListSchema: - """Return paginated Kick drop campaigns.""" - now: datetime.datetime = timezone.now() - queryset: QuerySet[KickDropCampaign] = ( - KickDropCampaign.objects - .filter(is_fully_imported=True) - .select_related("category", "organization") - .prefetch_related("channels__user", "rewards") - .order_by("-starts_at", "kick_id") - ) - - if game is not None: - queryset = queryset.filter(category__kick_id=game) - if organization is not None: - queryset = queryset.filter(organization__kick_id=organization) - if search is not None: - queryset = queryset.filter(name__icontains=search) - - queryset = _apply_status_filter(queryset, status, now) - - items, total, current_page, current_page_size = _paginate( - queryset, - page=page, - page_size=page_size, - ) - - return V1CampaignListSchema( - total=total, - page=current_page, - page_size=current_page_size, - items=[_serialize_campaign(campaign, now) for campaign in items], - ) - - -@api.get("/campaigns/{kick_id}/", response=V1CampaignDetailSchema) -def get_campaign( - request: HttpRequest, - kick_id: str, -) -> V1CampaignDetailSchema: - """Return a single Kick drop campaign. - - Raises: - Http404: If the campaign does not exist or is not fully imported. - """ - try: - campaign: KickDropCampaign = ( - KickDropCampaign.objects - .filter(is_fully_imported=True) - .select_related("category", "organization") - .prefetch_related("channels__user", "rewards") - .get(kick_id=kick_id) - ) - except KickDropCampaign.DoesNotExist as exc: - msg = "Campaign not found" - raise Http404(msg) from exc - - return _serialize_campaign(campaign, timezone.now()) - - -@api.get("/games/", response=V1CategoryListSchema) -def list_games( - request: HttpRequest, - page: int = 1, - page_size: int = DEFAULT_PAGE_SIZE, - search: str | None = None, -) -> V1CategoryListSchema: - """Return paginated Kick categories (games).""" - now: datetime.datetime = timezone.now() - - queryset: QuerySet[KickCategory] = KickCategory.objects.annotate( - campaign_count=models.Count( - "campaigns", - filter=models.Q(campaigns__is_fully_imported=True), - ), - active_campaign_count=models.Count( - "campaigns", - filter=models.Q( - campaigns__is_fully_imported=True, - campaigns__starts_at__lte=now, - campaigns__ends_at__gte=now, - ), - ), - ).order_by("name") - - if search is not None: - queryset = queryset.filter(name__icontains=search) - - items, total, current_page, current_page_size = _paginate( - queryset, - page=page, - page_size=page_size, - ) - - return V1CategoryListSchema( - total=total, - page=current_page, - page_size=current_page_size, - items=[ - V1CategoryListItemSchema( - kick_id=cat.kick_id, - name=cat.name, - slug=cat.slug, - image_url=cat.image_url, - campaign_count=cat.campaign_count, # pyright: ignore[reportAttributeAccessIssue] - active_campaign_count=cat.active_campaign_count, # pyright: ignore[reportAttributeAccessIssue] - ) - for cat in items - ], - ) - - -@api.get("/games/{kick_id}/", response=V1CategoryDetailSchema) -def get_game( - request: HttpRequest, - kick_id: int, -) -> V1CategoryDetailSchema: - """Return a single Kick category (game) with its campaigns.""" - category: KickCategory = get_object_or_404(KickCategory, kick_id=kick_id) - now: datetime.datetime = timezone.now() - campaigns: list[KickDropCampaign] = list( - category.campaigns # pyright: ignore[reportAttributeAccessIssue] - .filter(is_fully_imported=True) - .select_related("organization") - .prefetch_related("rewards") - .order_by("-starts_at"), - ) - - return V1CategoryDetailSchema( - kick_id=category.kick_id, - name=category.name, - slug=category.slug, - image_url=category.image_url, - campaign_count=len(campaigns), - active_campaign_count=sum( - 1 for c in campaigns if _campaign_status(c, now) == "active" - ), - campaigns=[_serialize_campaign_summary(c, now) for c in campaigns], - ) - - -@api.get("/organizations/", response=V1OrganizationListSchema) -def list_organizations( - request: HttpRequest, - page: int = 1, - page_size: int = DEFAULT_PAGE_SIZE, - search: str | None = None, -) -> V1OrganizationListSchema: - """Return paginated Kick organizations.""" - now: datetime.datetime = timezone.now() - - queryset: QuerySet[KickOrganization] = KickOrganization.objects.annotate( - campaign_count=models.Count( - "campaigns", - filter=models.Q(campaigns__is_fully_imported=True), - ), - active_campaign_count=models.Count( - "campaigns", - filter=models.Q( - campaigns__is_fully_imported=True, - campaigns__starts_at__lte=now, - campaigns__ends_at__gte=now, - ), - ), - ).order_by("name") - - if search is not None: - queryset = queryset.filter(name__icontains=search) - - items, total, current_page, current_page_size = _paginate( - queryset, - page=page, - page_size=page_size, - ) - - return V1OrganizationListSchema( - total=total, - page=current_page, - page_size=current_page_size, - items=[ - V1OrganizationListItemSchema( - kick_id=org.kick_id, - name=org.name, - logo_url=org.logo_url, - url=org.url, - campaign_count=org.campaign_count, # pyright: ignore[reportAttributeAccessIssue] - active_campaign_count=org.active_campaign_count, # pyright: ignore[reportAttributeAccessIssue] - ) - for org in items - ], - ) - - -@api.get("/organizations/{kick_id}/", response=V1OrganizationDetailSchema) -def get_organization( - request: HttpRequest, - kick_id: str, -) -> V1OrganizationDetailSchema: - """Return a single Kick organization with its campaigns.""" - organization: KickOrganization = get_object_or_404( - KickOrganization, - kick_id=kick_id, - ) - now: datetime.datetime = timezone.now() - campaigns: list[KickDropCampaign] = list( - organization.campaigns # pyright: ignore[reportAttributeAccessIssue] - .filter(is_fully_imported=True) - .select_related("category") - .prefetch_related("rewards") - .order_by("-starts_at"), - ) - - return V1OrganizationDetailSchema( - kick_id=organization.kick_id, - name=organization.name, - logo_url=organization.logo_url, - url=organization.url, - campaign_count=len(campaigns), - active_campaign_count=sum( - 1 for c in campaigns if _campaign_status(c, now) == "active" - ), - campaigns=[_serialize_campaign_summary(c, now) for c in campaigns], - ) diff --git a/kick/feeds.py b/kick/feeds.py index d2d65fb..36f55b9 100644 --- a/kick/feeds.py +++ b/kick/feeds.py @@ -496,7 +496,7 @@ class KickCategoryCampaignFeed(TTVDropsBaseFeed): self._limit = None return super().__call__(request, *args, **kwargs) - def get_object(self, request: HttpRequest, kick_id: int) -> KickCategory: # ruff:ignore[unused-method-argument] + def get_object(self, request: HttpRequest, kick_id: int) -> KickCategory: # noqa: ARG002 """Return game object for this feed URL.""" return KickCategory.objects.get(kick_id=kick_id) diff --git a/kick/management/commands/import_kick_drops.py b/kick/management/commands/import_kick_drops.py index 308e202..4027227 100644 --- a/kick/management/commands/import_kick_drops.py +++ b/kick/management/commands/import_kick_drops.py @@ -130,7 +130,7 @@ class Command(BaseCommand): try: payload: dict = response.json() - except Exception as exc: # ruff:ignore[blind-except] + except Exception as exc: # noqa: BLE001 self.stderr.write(self.style.ERROR(f"Failed to parse JSON response: {exc}")) return @@ -166,7 +166,7 @@ class Command(BaseCommand): self.style.SUCCESS(f"Imported {imported}/{len(campaigns)} campaign(s)."), ) - def _import_campaign(self, data: KickDropCampaignSchema) -> None: # ruff:ignore[too-many-locals, too-many-statements] + def _import_campaign(self, data: KickDropCampaignSchema) -> None: # noqa: PLR0914, PLR0915 """Import a single campaign and all its related objects.""" # Organization org_data: KickOrganizationSchema = data.organization diff --git a/kick/models.py b/kick/models.py index 40a01c1..450434f 100644 --- a/kick/models.py +++ b/kick/models.py @@ -93,12 +93,12 @@ class KickCategory(auto_prefetch.Model): @property def get_absolute_url(self) -> str: - """The URL to the game detail page.""" + """Return the URL to the game detail page.""" return reverse("kick:game_detail", args=[self.kick_id]) @property def kick_url(self) -> str: - """The URL to the game page on Kick.""" + """Return the URL to the game page on Kick.""" return f"https://kick.com/category/{self.slug}" if self.slug else "" @@ -136,7 +136,7 @@ class KickUser(auto_prefetch.Model): @property def kick_profile_url(self) -> str: - """The Kick profile URL for this user.""" + """Return the Kick profile URL for this user.""" return f"https://kick.com/{self.username}" if self.username else "" @@ -183,7 +183,7 @@ class KickChannel(auto_prefetch.Model): @property def channel_url(self) -> str: - """The Kick channel URL.""" + """Return the Kick channel URL.""" return f"https://kick.com/{self.slug}" if self.slug else "" @@ -290,7 +290,7 @@ class KickDropCampaign(auto_prefetch.Model): @property def image_url(self) -> str: - """The image URL for the campaign.""" + """Return the image URL for the campaign.""" # Image from first drop rewards_prefetched: list[KickReward] | None = getattr( self, @@ -356,7 +356,7 @@ class KickDropCampaign(auto_prefetch.Model): @property def merged_rewards(self) -> list[KickReward]: - """Rewards de-duplicated by normalized name. + """Return rewards de-duplicated by normalized name. If both a base reward and a "(Con)" variant exist, prefer the base reward name. """ @@ -455,7 +455,7 @@ class KickReward(auto_prefetch.Model): @property def full_image_url(self) -> str: - """The absolute image URL for this reward. + """Return the absolute image URL for this reward. If the image_url is a relative path, prepend the Kick image base URL. """ diff --git a/kick/schemas.py b/kick/schemas.py index 396dbb1..2b26114 100644 --- a/kick/schemas.py +++ b/kick/schemas.py @@ -1,4 +1,4 @@ -from datetime import datetime # ruff:ignore[typing-only-standard-library-import] +from datetime import datetime # noqa: TC003 from pydantic import BaseModel from pydantic import Field diff --git a/kick/tasks.py b/kick/tasks.py index 2eebd3b..9c50a9e 100644 --- a/kick/tasks.py +++ b/kick/tasks.py @@ -9,7 +9,7 @@ logger = logging.getLogger("ttvdrops.tasks") @shared_task(bind=True, queue="api-fetches", max_retries=3, default_retry_delay=120) -def import_kick_drops(self) -> None: # ruff:ignore[missing-type-function-argument] +def import_kick_drops(self) -> None: # noqa: ANN001 """Fetch and upsert Kick drop campaigns from the public API.""" try: call_command("import_kick_drops") diff --git a/kick/tests/test_api.py b/kick/tests/test_api.py deleted file mode 100644 index ee63ae2..0000000 --- a/kick/tests/test_api.py +++ /dev/null @@ -1,810 +0,0 @@ -from __future__ import annotations - -from datetime import UTC -from datetime import datetime -from unittest import mock - -from django.db import connection -from django.test import TestCase -from django.test.utils import CaptureQueriesContext -from django.urls import reverse - -from kick.models import KickCategory -from kick.models import KickChannel -from kick.models import KickDropCampaign -from kick.models import KickOrganization -from kick.models import KickReward -from kick.models import KickUser - - -class KickCampaignApiTest(TestCase): - """Tests for the public Kick campaign JSON API.""" - - @classmethod - def setUpTestData(cls) -> None: - """Create a fully populated campaign for API contract tests.""" - cls.organization = KickOrganization.objects.create( - kick_id="org-api", - name="API Organization", - logo_url="https://example.com/org.png", - url="https://example.com/org", - ) - cls.category = KickCategory.objects.create( - kick_id=123, - name="API Game", - slug="api-game", - image_url="https://example.com/game.png", - ) - cls.campaign = KickDropCampaign.objects.create( - kick_id="campaign-api", - name="API Campaign", - status="expired", - starts_at=datetime(2026, 7, 1, tzinfo=UTC), - ends_at=datetime(2026, 8, 1, tzinfo=UTC), - connect_url="https://example.com/connect", - url="https://example.com/campaign", - organization=cls.organization, - category=cls.category, - rule_id=1, - rule_name="Watch to earn", - is_fully_imported=True, - ) - KickDropCampaign.objects.create( - kick_id="campaign-not-imported", - name="Campaign Not Imported", - starts_at=datetime(2026, 7, 1, tzinfo=UTC), - ends_at=datetime(2026, 8, 1, tzinfo=UTC), - organization=cls.organization, - category=cls.category, - ) - user = KickUser.objects.create( - kick_id=456, - username="api-streamer", - profile_picture="https://example.com/avatar.png", - ) - channel = KickChannel.objects.create( - kick_id=789, - slug="api-streamer", - user=user, - ) - cls.campaign.channels.add(channel) - KickReward.objects.create( - kick_id="reward-api", - name="API Reward", - image_url="drops/reward-image/reward.png", - required_units=30, - campaign=cls.campaign, - category=cls.category, - organization=cls.organization, - ) - - def test_list_returns_paginated_campaigns_with_nested_data(self) -> None: - """A campaign includes the fields needed by downstream importers.""" - with mock.patch( - "kick.api.timezone.now", - return_value=datetime(2026, 7, 15, tzinfo=UTC), - ): - response = self.client.get(reverse("api-v1:kick-api-v1_list_campaigns")) - - assert response.status_code == 200 - assert response["Content-Type"].startswith("application/json") - data = response.json() - assert data["total"] == 1 - assert data["page"] == 1 - assert data["page_size"] == 100 - - item = data["items"][0] - assert item["kick_id"] == "campaign-api" - assert item["name"] == "API Campaign" - assert item["status"] == "active" - assert ( - item["image_url"] - == "https://ext.cdn.kick.com/drops/reward-image/reward.png" - ) - assert item["connect_url"] == "https://example.com/connect" - assert item["url"] == "https://example.com/campaign" - assert item["starts_at"] == "2026-07-01T00:00:00Z" - assert item["ends_at"] == "2026-08-01T00:00:00Z" - assert item["category"] == { - "kick_id": 123, - "name": "API Game", - "slug": "api-game", - "image_url": "https://example.com/game.png", - } - assert item["organization"] == { - "kick_id": "org-api", - "name": "API Organization", - "logo_url": "https://example.com/org.png", - "url": "https://example.com/org", - } - assert item["reward_count"] == 1 - assert item["channels"] == [ - { - "kick_id": 789, - "slug": "api-streamer", - "url": "https://kick.com/api-streamer", - "user": { - "kick_id": 456, - "username": "api-streamer", - "profile_picture": "https://example.com/avatar.png", - }, - }, - ] - assert item["rewards"] == [ - { - "kick_id": "reward-api", - "name": "API Reward", - "image_url": "https://ext.cdn.kick.com/drops/reward-image/reward.png", - "required_minutes_watched": 30, - }, - ] - assert item["is_fully_imported"] is True - assert "added_at" in item - assert "updated_at" in item - - def test_filters_campaigns_by_game(self) -> None: - """A Kick category ID can select a downstream import subset.""" - other_category = KickCategory.objects.create( - kick_id=124, - name="Other Game", - ) - KickDropCampaign.objects.create( - kick_id="other-campaign", - name="Other Campaign", - starts_at=datetime(2026, 8, 2, tzinfo=UTC), - ends_at=datetime(2026, 9, 1, tzinfo=UTC), - organization=self.organization, - category=other_category, - is_fully_imported=True, - ) - - response = self.client.get( - reverse("api-v1:kick-api-v1_list_campaigns"), - {"game": self.category.kick_id}, - ) - - assert response.status_code == 200 - assert [item["kick_id"] for item in response.json()["items"]] == [ - self.campaign.kick_id, - ] - - def test_filters_campaigns_by_computed_status(self) -> None: - """All supported status filters are computed from campaign dates.""" - KickDropCampaign.objects.create( - kick_id="upcoming-campaign", - name="Upcoming Campaign", - starts_at=datetime(2026, 8, 2, tzinfo=UTC), - ends_at=datetime(2026, 9, 1, tzinfo=UTC), - organization=self.organization, - category=self.category, - is_fully_imported=True, - ) - KickDropCampaign.objects.create( - kick_id="expired-campaign", - name="Expired Campaign", - starts_at=datetime(2026, 6, 1, tzinfo=UTC), - ends_at=datetime(2026, 6, 2, tzinfo=UTC), - organization=self.organization, - category=self.category, - is_fully_imported=True, - ) - expected_ids = { - "active": self.campaign.kick_id, - "upcoming": "upcoming-campaign", - "expired": "expired-campaign", - } - - with mock.patch( - "kick.api.timezone.now", - return_value=datetime(2026, 7, 15, tzinfo=UTC), - ): - for status, expected_id in expected_ids.items(): - with self.subTest(status=status): - response = self.client.get( - reverse("api-v1:kick-api-v1_list_campaigns"), - {"status": status}, - ) - assert response.status_code == 200 - assert [item["kick_id"] for item in response.json()["items"]] == [ - expected_id, - ] - - def test_status_filters_exclude_campaigns_with_partial_dates(self) -> None: - """Filtered items never serialize with a contradictory unknown status.""" - KickDropCampaign.objects.create( - kick_id="partial-upcoming-campaign", - name="Partial Upcoming Campaign", - starts_at=datetime(2026, 8, 2, tzinfo=UTC), - ends_at=None, - organization=self.organization, - category=self.category, - is_fully_imported=True, - ) - KickDropCampaign.objects.create( - kick_id="partial-expired-campaign", - name="Partial Expired Campaign", - starts_at=None, - ends_at=datetime(2026, 6, 2, tzinfo=UTC), - organization=self.organization, - category=self.category, - is_fully_imported=True, - ) - - with mock.patch( - "kick.api.timezone.now", - return_value=datetime(2026, 7, 15, tzinfo=UTC), - ): - for status in ("upcoming", "expired"): - with self.subTest(status=status): - response = self.client.get( - reverse("api-v1:kick-api-v1_list_campaigns"), - {"status": status}, - ) - assert all( - item["status"] == status for item in response.json()["items"] - ) - - def test_paginates_campaigns(self) -> None: - """Page and page_size select a stable result slice.""" - for index in range(2): - KickDropCampaign.objects.create( - kick_id=f"pagination-campaign-{index}", - name=f"Pagination Campaign {index}", - starts_at=datetime(2026, 6, index + 1, tzinfo=UTC), - ends_at=datetime(2026, 6, index + 2, tzinfo=UTC), - organization=self.organization, - category=self.category, - is_fully_imported=True, - ) - - response = self.client.get( - reverse("api-v1:kick-api-v1_list_campaigns"), - {"page": 2, "page_size": 1}, - ) - - data = response.json() - assert response.status_code == 200 - assert data["total"] == 3 - assert data["page"] == 2 - assert data["page_size"] == 1 - assert [item["kick_id"] for item in data["items"]] == [ - "pagination-campaign-1", - ] - - def test_clamps_page_size(self) -> None: - """A caller cannot request an unbounded response page.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_list_campaigns"), - {"page_size": 1000}, - ) - - assert response.status_code == 200 - assert response.json()["page_size"] == 500 - - def test_clamps_pagination_minimums(self) -> None: - """Zero and negative pagination values use the first one-item page.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_list_campaigns"), - {"page": 0, "page_size": -1}, - ) - - assert response.status_code == 200 - assert response.json()["page"] == 1 - assert response.json()["page_size"] == 1 - - def test_rejects_non_integer_pagination_values(self) -> None: - """Malformed pagination values return field-specific validation errors.""" - for field in ("page", "page_size"): - with self.subTest(field=field): - response = self.client.get( - reverse("api-v1:kick-api-v1_list_campaigns"), - {field: "not-an-integer"}, - ) - assert response.status_code == 422 - assert response.json()["detail"][0]["loc"] == ["query", field] - - def test_page_beyond_result_set_returns_empty_items(self) -> None: - """An arbitrarily large page does not become a pathological DB offset.""" - page = 10**100 - - response = self.client.get( - reverse("api-v1:kick-api-v1_list_campaigns"), - {"page": page}, - ) - - assert response.status_code == 200 - assert response.json()["page"] == page - assert response.json()["items"] == [] - - def test_rejects_invalid_status(self) -> None: - """Unknown status values return a client error instead of an empty result.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_list_campaigns"), - {"status": "running"}, - ) - - assert response.status_code == 422 - assert response.json()["detail"][0]["loc"] == ["query", "status"] - - def test_rejects_empty_status(self) -> None: - """An explicitly empty status is invalid rather than an omitted filter.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_list_campaigns"), - {"status": ""}, - ) - - assert response.status_code == 422 - - def test_rejects_non_get_methods(self) -> None: - """The campaign feed is a read-only endpoint.""" - response = self.client.post(reverse("api-v1:kick-api-v1_list_campaigns")) - - assert response.status_code == 405 - - def test_rejects_invalid_game_id(self) -> None: - """A non-numeric Kick category ID returns a validation response.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_list_campaigns"), - {"game": "not-an-id"}, - ) - - assert response.status_code == 422 - assert response.json()["detail"][0]["loc"] == ["query", "game"] - - def test_returns_every_eligible_channel(self) -> None: - """The JSON API does not apply the five-channel RSS display limit.""" - for index in range(6): - channel = KickChannel.objects.create( - kick_id=1000 + index, - slug=f"extra-streamer-{index}", - ) - self.campaign.channels.add(channel) - - response = self.client.get(reverse("api-v1:kick-api-v1_list_campaigns")) - - channels = response.json()["items"][0]["channels"] - assert len(channels) == 7 - - def test_returns_every_raw_reward_in_stable_order(self) -> None: - """Rewards are neither merged nor truncated for downstream importers.""" - KickReward.objects.create( - kick_id="reward-api-con", - name="API Reward (Con)", - required_units=30, - campaign=self.campaign, - ) - KickReward.objects.create( - kick_id="reward-api-beta", - name="Beta Reward", - required_units=60, - campaign=self.campaign, - ) - - response = self.client.get(reverse("api-v1:kick-api-v1_list_campaigns")) - - rewards = response.json()["items"][0]["rewards"] - assert [reward["name"] for reward in rewards] == [ - "API Reward", - "API Reward (Con)", - "Beta Reward", - ] - - def test_empty_channels_remain_an_empty_list(self) -> None: - """Missing channel data is not presented as proof of global eligibility.""" - campaign = KickDropCampaign.objects.create( - kick_id="campaign-without-channels", - name="Campaign Without Channels", - starts_at=datetime(2026, 5, 1, tzinfo=UTC), - ends_at=datetime(2026, 5, 2, tzinfo=UTC), - organization=self.organization, - category=self.category, - is_fully_imported=True, - ) - - response = self.client.get(reverse("api-v1:kick-api-v1_list_campaigns")) - - item = next( - item - for item in response.json()["items"] - if item["kick_id"] == campaign.kick_id - ) - assert item["channels"] == [] - - def test_query_count_does_not_grow_with_campaign_count(self) -> None: - """Nested serialization avoids per-campaign database queries.""" - - def select_count() -> int: - with CaptureQueriesContext(connection) as queries: - response = self.client.get(reverse("api-v1:kick-api-v1_list_campaigns")) - assert response.status_code == 200 - return sum( - query["sql"].lstrip().upper().startswith("SELECT") - for query in queries.captured_queries - ) - - baseline = select_count() - - for index in range(8): - campaign = KickDropCampaign.objects.create( - kick_id=f"query-campaign-{index}", - name=f"Query Campaign {index}", - starts_at=datetime(2026, 4, 1, tzinfo=UTC), - ends_at=datetime(2026, 4, 2, tzinfo=UTC), - organization=self.organization, - category=self.category, - is_fully_imported=True, - ) - KickReward.objects.create( - kick_id=f"query-reward-{index}", - name=f"Query Reward {index}", - required_units=15, - campaign=campaign, - ) - - scaled = select_count() - - assert scaled == baseline - - -class KickCampaignDetailApiTest(TestCase): - """Tests for the campaign detail endpoint.""" - - @classmethod - def setUpTestData(cls) -> None: - cls.organization = KickOrganization.objects.create( - kick_id="org-detail", - name="Detail Org", - ) - cls.category = KickCategory.objects.create( - kick_id=200, - name="Detail Game", - slug="detail-game", - ) - cls.campaign = KickDropCampaign.objects.create( - kick_id="campaign-detail", - name="Detail Campaign", - starts_at=datetime(2026, 7, 1, tzinfo=UTC), - ends_at=datetime(2026, 8, 1, tzinfo=UTC), - organization=cls.organization, - category=cls.category, - rule_id=1, - rule_name="Watch to earn", - is_fully_imported=True, - ) - cls.user = KickUser.objects.create( - kick_id=457, - username="detail-streamer", - profile_picture="https://example.com/detail.png", - ) - cls.channel = KickChannel.objects.create( - kick_id=790, - slug="detail-streamer", - user=cls.user, - ) - cls.campaign.channels.add(cls.channel) - cls.reward = KickReward.objects.create( - kick_id="reward-detail", - name="Detail Reward", - required_units=45, - campaign=cls.campaign, - category=cls.category, - organization=cls.organization, - ) - - def test_detail_returns_campaign(self) -> None: - """A fully imported campaign returns its full serialized payload.""" - with mock.patch( - "kick.api.timezone.now", - return_value=datetime(2026, 7, 15, tzinfo=UTC), - ): - response = self.client.get( - reverse( - "api-v1:kick-api-v1_get_campaign", - args=[self.campaign.kick_id], - ), - ) - - assert response.status_code == 200 - data = response.json() - assert data["kick_id"] == "campaign-detail" - assert data["status"] == "active" - assert data["name"] == "Detail Campaign" - assert len(data["channels"]) == 1 - assert len(data["rewards"]) == 1 - - def test_detail_not_found(self) -> None: - """A non-existent campaign ID returns 404.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_get_campaign", args=["nonexistent-id"]), - ) - assert response.status_code == 404 - - def test_detail_not_fully_imported(self) -> None: - """A campaign that is not fully imported returns 404.""" - KickDropCampaign.objects.create( - kick_id="not-imported-detail", - name="Not Imported", - organization=self.organization, - category=self.category, - starts_at=datetime(2026, 7, 1, tzinfo=UTC), - ends_at=datetime(2026, 8, 1, tzinfo=UTC), - is_fully_imported=False, - ) - response = self.client.get( - reverse("api-v1:kick-api-v1_get_campaign", args=["not-imported-detail"]), - ) - assert response.status_code == 404 - - -class KickCampaignSearchFilterTest(TestCase): - """Tests for the search and organization filter on the campaign list.""" - - @classmethod - def setUpTestData(cls) -> None: - cls.org_a = KickOrganization.objects.create( - kick_id="org-a", - name="Studio Alpha", - ) - cls.org_b = KickOrganization.objects.create(kick_id="org-b", name="Beta Corp") - cls.cat = KickCategory.objects.create(kick_id=300, name="Test Game") - - cls.camp_alpha = KickDropCampaign.objects.create( - kick_id="alpha-camp", - name="Alpha Drop Event", - starts_at=datetime(2026, 6, 1, tzinfo=UTC), - ends_at=datetime(2026, 7, 1, tzinfo=UTC), - organization=cls.org_a, - category=cls.cat, - is_fully_imported=True, - ) - cls.camp_beta = KickDropCampaign.objects.create( - kick_id="beta-camp", - name="Beta Giveaway", - starts_at=datetime(2026, 7, 1, tzinfo=UTC), - ends_at=datetime(2026, 8, 1, tzinfo=UTC), - organization=cls.org_b, - category=cls.cat, - is_fully_imported=True, - ) - cls.camp_alpha2 = KickDropCampaign.objects.create( - kick_id="alpha-camp-2", - name="Alpha Second Wave", - starts_at=datetime(2026, 8, 1, tzinfo=UTC), - ends_at=datetime(2026, 9, 1, tzinfo=UTC), - organization=cls.org_a, - category=cls.cat, - is_fully_imported=True, - ) - - def test_filters_by_organization(self) -> None: - """An organization kick_id returns only that org's campaigns.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_list_campaigns"), - {"organization": "org-a"}, - ) - assert response.status_code == 200 - ids = [item["kick_id"] for item in response.json()["items"]] - assert sorted(ids) == ["alpha-camp", "alpha-camp-2"] - - def test_organization_filter_with_no_matches(self) -> None: - """An org with no campaigns returns an empty list.""" - KickOrganization.objects.create(kick_id="org-empty", name="Empty Org") - response = self.client.get( - reverse("api-v1:kick-api-v1_list_campaigns"), - {"organization": "org-empty"}, - ) - assert response.status_code == 200 - assert response.json()["items"] == [] - - def test_searches_by_name(self) -> None: - """A search term filters campaigns by case-insensitive name match.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_list_campaigns"), - {"search": "alpha"}, - ) - assert response.status_code == 200 - ids = [item["kick_id"] for item in response.json()["items"]] - assert sorted(ids) == ["alpha-camp", "alpha-camp-2"] - - def test_search_with_no_matches(self) -> None: - """A search term with no matches returns an empty list.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_list_campaigns"), - {"search": "zzzzz"}, - ) - assert response.status_code == 200 - assert response.json()["items"] == [] - - def test_search_case_insensitive(self) -> None: - """Search is case-insensitive.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_list_campaigns"), - {"search": "ALPHA"}, - ) - assert response.status_code == 200 - ids = [item["kick_id"] for item in response.json()["items"]] - assert sorted(ids) == ["alpha-camp", "alpha-camp-2"] - - def test_organization_filter_with_search(self) -> None: - """Organization filter and search can be combined.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_list_campaigns"), - {"organization": "org-a", "search": "Second"}, - ) - assert response.status_code == 200 - ids = [item["kick_id"] for item in response.json()["items"]] - assert ids == ["alpha-camp-2"] - - -class KickOrganizationApiTest(TestCase): - """Tests for the organization API endpoints.""" - - @classmethod - def setUpTestData(cls) -> None: - cls.org_a = KickOrganization.objects.create( - kick_id="org-list-a", - name="Alpha Studios", - logo_url="https://example.com/alpha.png", - url="https://example.com/alpha", - ) - cls.org_b = KickOrganization.objects.create( - kick_id="org-list-b", - name="Beta Games", - logo_url="https://example.com/beta.png", - url="https://example.com/beta", - ) - cat = KickCategory.objects.create(kick_id=400, name="Game") - for org in (cls.org_a, cls.org_b): - KickDropCampaign.objects.create( - kick_id=f"camp-{org.kick_id}", - name=f"{org.name} Campaign", - starts_at=datetime(2026, 7, 1, tzinfo=UTC), - ends_at=datetime(2026, 8, 1, tzinfo=UTC), - organization=org, - category=cat, - is_fully_imported=True, - ) - - def test_list_returns_organizations(self) -> None: - """Organization list returns all orgs with campaign counts.""" - response = self.client.get(reverse("api-v1:kick-api-v1_list_organizations")) - assert response.status_code == 200 - data = response.json() - assert data["total"] == 2 - names = [item["name"] for item in data["items"]] - assert names == ["Alpha Studios", "Beta Games"] - assert all(item["campaign_count"] == 1 for item in data["items"]) - - def test_detail_returns_organization_with_campaigns(self) -> None: - """Organization detail returns org metadata and its campaigns.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_get_organization", args=["org-list-a"]), - ) - assert response.status_code == 200 - data = response.json() - assert data["name"] == "Alpha Studios" - assert data["campaign_count"] == 1 - assert len(data["campaigns"]) == 1 - assert data["campaigns"][0]["name"] == "Alpha Studios Campaign" - - def test_detail_not_found(self) -> None: - """A non-existent org returns 404.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_get_organization", args=["nonexistent"]), - ) - assert response.status_code == 404 - - def test_list_pagination(self) -> None: - """Organization list respects pagination.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_list_organizations"), - {"page": 1, "page_size": 1}, - ) - assert response.status_code == 200 - data = response.json() - assert len(data["items"]) == 1 - assert data["total"] == 2 - - def test_list_search(self) -> None: - """Organization list supports search by name.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_list_organizations"), - {"search": "alpha"}, - ) - assert response.status_code == 200 - names = [item["name"] for item in response.json()["items"]] - assert names == ["Alpha Studios"] - - -class KickCategoryApiTest(TestCase): - """Tests for the category (game) API endpoints.""" - - @classmethod - def setUpTestData(cls) -> None: - cls.cat_a = KickCategory.objects.create( - kick_id=500, - name="Action Game", - slug="action-game", - image_url="https://example.com/action.png", - ) - cls.cat_b = KickCategory.objects.create( - kick_id=501, - name="Racing Game", - slug="racing-game", - image_url="https://example.com/racing.png", - ) - org = KickOrganization.objects.create( - kick_id="cat-org", - name="Category Org", - ) - for cat in (cls.cat_a, cls.cat_b): - KickDropCampaign.objects.create( - kick_id=f"camp-{cat.kick_id}", - name=f"{cat.name} Campaign", - starts_at=datetime(2026, 7, 1, tzinfo=UTC), - ends_at=datetime(2026, 8, 1, tzinfo=UTC), - organization=org, - category=cat, - is_fully_imported=True, - ) - - def test_list_returns_categories(self) -> None: - """Category list returns all categories with campaign counts.""" - response = self.client.get(reverse("api-v1:kick-api-v1_list_games")) - assert response.status_code == 200 - data = response.json() - assert data["total"] == 2 - names = [item["name"] for item in data["items"]] - assert names == ["Action Game", "Racing Game"] - assert all(item["campaign_count"] == 1 for item in data["items"]) - - def test_detail_returns_category_with_campaigns(self) -> None: - """Category detail returns metadata and its campaigns.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_get_game", args=[500]), - ) - assert response.status_code == 200 - data = response.json() - assert data["name"] == "Action Game" - assert data["campaign_count"] == 1 - assert len(data["campaigns"]) == 1 - assert data["campaigns"][0]["name"] == "Action Game Campaign" - - def test_detail_not_found(self) -> None: - """A non-existent category returns 404.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_get_game", args=[99999]), - ) - assert response.status_code == 404 - - def test_list_search(self) -> None: - """Category list supports search by name.""" - response = self.client.get( - reverse("api-v1:kick-api-v1_list_games"), - {"search": "action"}, - ) - assert response.status_code == 200 - names = [item["name"] for item in response.json()["items"]] - assert names == ["Action Game"] - - -class KickStatsApiTest(TestCase): - """Tests for the stats endpoint.""" - - @classmethod - def setUpTestData(cls) -> None: - KickOrganization.objects.create(kick_id="stats-org", name="Stats Org") - KickCategory.objects.create(kick_id=600, name="Stats Game") - KickChannel.objects.create(kick_id=800, slug="stats-channel") - - def test_stats_returns_aggregate_counts(self) -> None: - """Stats endpoint returns counts for all entity types.""" - response = self.client.get(reverse("api-v1:kick-api-v1_stats")) - assert response.status_code == 200 - data = response.json() - assert data["total_campaigns"] == 0 - assert data["total_organizations"] == 1 - assert data["total_categories"] == 1 - assert data["total_channels"] == 1 - for key in ("active", "upcoming", "expired", "partial_dates"): - assert key in data diff --git a/kick/tests/test_kick.py b/kick/tests/test_kick.py index 6d4153b..3f2925e 100644 --- a/kick/tests/test_kick.py +++ b/kick/tests/test_kick.py @@ -463,7 +463,7 @@ class KickDropCampaignMergedRewardsTest(TestCase): class ImportKickDropsCommandTest(TestCase): """Tests for the import_kick_drops management command.""" - def _run_command(self, json_payload: dict, **options: Any) -> tuple[str, str]: # ruff:ignore[any-type] + def _run_command(self, json_payload: dict, **options: Any) -> tuple[str, str]: # noqa: ANN401 mock_response = MagicMock() mock_response.json.return_value = json_payload diff --git a/kick/urls.py b/kick/urls.py index 6481ae8..1cfa7f6 100644 --- a/kick/urls.py +++ b/kick/urls.py @@ -1,7 +1,6 @@ from typing import TYPE_CHECKING from django.urls import path -from django.views.generic.base import RedirectView from kick import views from kick.feeds import KickCampaignAtomFeed @@ -30,14 +29,6 @@ urlpatterns: list[URLPattern | URLResolver] = [ view=views.dashboard, name="dashboard", ), - # Redirect old Kick v1 campaign API to the new combined API - path( - route="api/v1/campaigns/", - view=RedirectView.as_view( - pattern_name="api-v1:kick-api-v1_list_campaigns", - permanent=True, - ), - ), # /kick/campaigns/ path( route="campaigns/", diff --git a/manage.py b/manage.py index 43ab60a..1077d0a 100755 --- a/manage.py +++ b/manage.py @@ -13,7 +13,7 @@ def main() -> None: """ os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings") try: - from django.core.management import execute_from_command_line # ruff:ignore[import-outside-top-level, unsorted-imports] + from django.core.management import execute_from_command_line # noqa: PLC0415 except ImportError as exc: msg = ( "Couldn't import Django. Are you sure it's installed and " diff --git a/pyproject.toml b/pyproject.toml index edf95f9..dcaf47e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -113,7 +113,6 @@ lint.ignore = [ [tool.ruff.lint.per-file-ignores] "**/tests/**" = [ "ARG", - "D102", "FBT", "PLR0904", "PLR2004", @@ -125,9 +124,6 @@ lint.ignore = [ "SLF001", ] "**/migrations/**" = ["RUF012"] -"**/api.py" = [ - "PLR0913", # Allow many arguments for API endpoint functions -] [tool.djlint] profile = "django" diff --git a/templates/base.html b/templates/base.html index 12e5b7d..9fe04e5 100644 --- a/templates/base.html +++ b/templates/base.html @@ -203,7 +203,6 @@ Dataset | Donate | GitHub | - API Docs |
Channels | Badges | Emotes | + API Docs | Inventory
Kick @@ -234,7 +234,10 @@ Campaigns
Other sites + Steam | YouTube | + TikTok | + Discord
{% if messages %} diff --git a/templates/core/dashboard.html b/templates/core/dashboard.html index 1c95756..c19674e 100644 --- a/templates/core/dashboard.html +++ b/templates/core/dashboard.html @@ -47,7 +47,7 @@ title="Atom feed for Twitch campaigns">[atom] [discord] - [api] [explain] diff --git a/templates/core/dataset_backups.html b/templates/core/dataset_backups.html index a8a1433..db560a2 100644 --- a/templates/core/dataset_backups.html +++ b/templates/core/dataset_backups.html @@ -7,121 +7,27 @@

Dataset Backups

About this dataset

-

This site tracks and publishes open drop campaign data from Twitch and Kick.

+

This site tracks and publishes open Twitch and Kick drop campaign data.

- Campaign metadata in the downloadable datasets and public JSON API is released under - CC0 so you can reuse it freely. The underlying source data is scraped from Twitch and Kick - APIs and pages; we do not control the upstream content and cannot guarantee its accuracy or - permanence. Linked third-party images, logos, and trademarks may remain subject to separate rights. + The exported datasets on this page are released under CC0 so you can reuse them freely. + The underlying source data is scraped from Twitch/Kick APIs and pages, so we do not control the + upstream content and cannot guarantee upstream accuracy or permanence.

-

Note that some drops have missing or incomplete data due to Twitch API limitations.

+

Note that some drops has missing or incomplete data due to Twitch API limitations.

Reward campaign data is sourced from SunkwiBOT/twitch-drops-api. - Raw JSON exports are available in the same repository as + Raw JSON exports are available in the repository: drops.json and rewards.json.

- Kick data is fetched directly from the public - Kick campaigns API - on a recurring schedule. -

-

- Need a custom format for your workflow or research pipeline? - Open a GitHub issue + Need a special format for your workflow or research pipeline? + Contact me via GitHub issues and describe what you need.

-
-

Kick JSON API

-

- Fully imported Kick campaigns are also available through the public paginated - Kick campaign JSON API. - Responses include campaign dates, game and organization metadata, rewards, and the full - channel list stored for each campaign. -

-

- Use page and page_size for pagination. Filter by Kick category ID with - game, or by the computed campaign state with - status=active, status=upcoming, status=expired, - or status=unknown. - Page numbers and page sizes have a minimum of 1, and the maximum page size is 500. -

-
{
-  "total": 1,
-  "page": 1,
-  "page_size": 100,
-  "items": [{
-    "kick_id": "...",
-    "name": "...",
-    "status": "active",
-    "starts_at": "2026-07-01T00:00:00Z",
-    "ends_at": "2026-08-01T00:00:00Z",
-    "category": { "kick_id": 123, "name": "..." },
-    "organization": { "kick_id": "...", "name": "..." },
-    "channels": [],
-    "rewards": [{ "name": "...", "required_minutes_watched": 30 }]
-  }]
-}
-

- Timestamps use ISO 8601 in UTC and may be null when the source does not provide a date. Campaigns with - incomplete dates receive the computed status unknown and are excluded from status-based filtering. - Each item includes every stored channel and raw reward without RSS truncation or reward-name merging. - Consumers can stop pagination when items is empty or when - page * page_size >= total. -

-

- An empty channels list means no channel restriction data is stored. It does not - by itself guarantee that a campaign is available on every channel. -

-
-
-

Twitch JSON API

-

- Fully imported Twitch campaigns are available through the public paginated - Twitch campaign JSON API. - Responses include campaign dates, game and organization metadata, allowed channels, drops, - and rewards for each campaign. -

-

- Use page and page_size for pagination. Filter by Twitch game ID with - game, or by the computed campaign state with - status=active, status=upcoming, status=expired, - or status=unknown. - Page numbers and page sizes have a minimum of 1, and the maximum page size is 500. -

-
{
-  "total": 1,
-  "page": 1,
-  "page_size": 100,
-  "items": [{
-    "twitch_id": "...",
-    "name": "...",
-    "game": { "twitch_id": "...", "display_name": "..." },
-    "organization": { "twitch_id": "...", "name": "...", "display_name": "..." },
-    "start_at": "2026-07-01T00:00:00Z",
-    "end_at": "2026-08-01T00:00:00Z",
-    "status": "active",
-    "allowed_channels": [],
-    "drops": [{ "name": "...", "benefits": ["..."] }]
-  }]
-}
-

- Timestamps use ISO 8601 in UTC and may be null when the source does not provide a date. Campaigns with - incomplete dates receive the computed status unknown and are excluded from status-based filtering. -

-

- Additional endpoints include - games, - organizations, - channels, - reward campaigns, and - chat badges - all public and paginated. Each list endpoint has a corresponding detail view by ID. -

-
{% if datasets %} @@ -148,7 +54,7 @@ {% endfor %}
-

{{ dataset_count }} dataset{% if dataset_count != 1 %}s{% endif %} available.

+

Found {{ dataset_count }} datasets.

{% else %}

No dataset backups found.

{% endif %} diff --git a/templates/core/debug.html b/templates/core/debug.html index c62e4b7..32c41e3 100644 --- a/templates/core/debug.html +++ b/templates/core/debug.html @@ -127,38 +127,6 @@

None ✅

{% endif %} -
-

Kick Campaigns With Unknown Status ({{ kick_unknown_campaigns|length }})

- {% if kick_unknown_campaigns %} - - {% else %} -

None ✅

- {% endif %} -
-
-

Twitch Reward Campaigns With Unknown Status ({{ twitch_reward_unknown_campaigns|length }})

- {% if twitch_reward_unknown_campaigns %} - - {% else %} -

None ✅

- {% endif %} -

Duplicate Campaign Names Per Game ({{ duplicate_name_campaigns|length }})

{% if duplicate_name_campaigns %} diff --git a/templates/core/docs_rss.html b/templates/core/docs_rss.html index 49e41e2..6d3d163 100644 --- a/templates/core/docs_rss.html +++ b/templates/core/docs_rss.html @@ -18,7 +18,7 @@

Twitch JSON API documentation is available at - /api/v1/docs/. + /twitch/api/v1/docs.

Twitch campaign feeds accept ?limit=50 to change item count and diff --git a/templates/twitch/campaign_detail.html b/templates/twitch/campaign_detail.html index 94fa6eb..85ec252 100644 --- a/templates/twitch/campaign_detail.html +++ b/templates/twitch/campaign_detail.html @@ -121,7 +121,7 @@ [discord] [explain] - [api] {% endif %} diff --git a/templates/twitch/campaign_list.html b/templates/twitch/campaign_list.html index b2ce75b..9fbe54d 100644 --- a/templates/twitch/campaign_list.html +++ b/templates/twitch/campaign_list.html @@ -33,7 +33,7 @@ title="Atom feed for all campaigns">[atom] [discord] - [api] [explain] [atom] [discord] - [api] [explain] diff --git a/templates/twitch/game_detail.html b/templates/twitch/game_detail.html index a5b7e56..fd52995 100644 --- a/templates/twitch/game_detail.html +++ b/templates/twitch/game_detail.html @@ -91,7 +91,7 @@ title="Discord feed for {{ game.display_name }} rewards">[discord]

- [api] [explain]
diff --git a/templates/twitch/games_grid.html b/templates/twitch/games_grid.html index fd60bd8..b85eb34 100644 --- a/templates/twitch/games_grid.html +++ b/templates/twitch/games_grid.html @@ -32,7 +32,7 @@ title="Atom feed for all games">[atom] [discord] - [api] [explain] [atom] [discord] - [api] [explain] [atom] [discord] - [api] [explain] [atom] [discord] - [api] [explain] {% if reward_campaign.external_url %} diff --git a/templates/twitch/reward_campaign_list.html b/templates/twitch/reward_campaign_list.html index b6e25bf..24c1d11 100644 --- a/templates/twitch/reward_campaign_list.html +++ b/templates/twitch/reward_campaign_list.html @@ -30,7 +30,7 @@
[explain] - [api]
diff --git a/tools/extract_historical_drops.py b/tools/extract_historical_drops.py index 9af65a0..65db357 100755 --- a/tools/extract_historical_drops.py +++ b/tools/extract_historical_drops.py @@ -16,7 +16,7 @@ import argparse import hashlib import json import shutil -import subprocess # ruff:ignore[suspicious-subprocess-import] +import subprocess # noqa: S404 import tempfile from pathlib import Path from typing import Any @@ -60,7 +60,7 @@ def run_git(git_dir: str, *args: str, check: bool = True) -> str: if not git: msg = "Git executable not found in PATH." raise FileNotFoundError(msg) - result: subprocess.CompletedProcess[str] = subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true] + result: subprocess.CompletedProcess[str] = subprocess.run( # noqa: S603 [git, "--git-dir", git_dir, *args], capture_output=True, text=True, diff --git a/twitch/api.py b/twitch/api.py index 26af923..96a409d 100644 --- a/twitch/api.py +++ b/twitch/api.py @@ -1,6 +1,6 @@ from __future__ import annotations -import datetime # ruff:ignore[typing-only-standard-library-import] +import datetime # noqa: TC003 from dataclasses import dataclass from typing import TYPE_CHECKING from typing import Literal @@ -9,7 +9,7 @@ from django.db import models from django.http import Http404 from django.shortcuts import get_object_or_404 from django.utils import timezone -from ninja import Router +from ninja import NinjaAPI from ninja import Schema from twitch.models import Channel @@ -28,7 +28,7 @@ if TYPE_CHECKING: from twitch.models import DropBenefit from twitch.models import TimeBasedDrop -V1StatusFilter = Literal["active", "upcoming", "expired", "unknown"] +V1StatusFilter = Literal["active", "upcoming", "expired"] DEFAULT_PAGE_SIZE = 100 MAX_PAGE_SIZE = 500 @@ -253,7 +253,11 @@ class V1ChatBadgeSetListSchema(V1PaginationSchema): items: list[V1ChatBadgeSetSchema] -api = Router() +api = NinjaAPI( + title="TTVDrops Twitch API", + version="1.0.0", + urls_namespace="twitch:twitch-api-v1", +) def _paginate[ModelT: models.Model]( @@ -301,11 +305,6 @@ def _apply_status_filter[ModelT: models.Model]( return queryset.filter(**{f"{start_field}__gt": now}) if status == "expired": return queryset.filter(**{f"{end_field}__lt": now}) - if status == "unknown": - return queryset.filter( - models.Q(**{f"{start_field}__isnull": True}) - | models.Q(**{f"{end_field}__isnull": True}), - ) return queryset diff --git a/twitch/apps.py b/twitch/apps.py index 67d0d6c..d15fdbd 100644 --- a/twitch/apps.py +++ b/twitch/apps.py @@ -15,7 +15,7 @@ class TwitchConfig(AppConfig): default_auto_field = "django.db.models.BigAutoField" name = "twitch" - def ready(self) -> None: # ruff:ignore[undocumented-public-method] + def ready(self) -> None: # noqa: D102 logger: logging.Logger = logging.getLogger("ttvdrops.apps") # Patch FieldFile.open to swallow FileNotFoundError and provide @@ -39,18 +39,18 @@ class TwitchConfig(AppConfig): # Register post_save signal handlers that dispatch image download tasks # when new Twitch records are created. - from django.db.models.signals import m2m_changed # ruff:ignore[unsorted-imports, import-outside-top-level] - from django.db.models.signals import post_save # ruff:ignore[import-outside-top-level] + from django.db.models.signals import m2m_changed # noqa: I001, PLC0415 + from django.db.models.signals import post_save # noqa: PLC0415 - from twitch.models import DropBenefit # ruff:ignore[import-outside-top-level] - from twitch.models import DropCampaign # ruff:ignore[import-outside-top-level] - from twitch.models import Game # ruff:ignore[import-outside-top-level] - from twitch.models import RewardCampaign # ruff:ignore[import-outside-top-level] - from twitch.signals import on_drop_benefit_saved # ruff:ignore[import-outside-top-level] - from twitch.signals import on_drop_campaign_allow_channels_changed # ruff:ignore[import-outside-top-level] - from twitch.signals import on_drop_campaign_saved # ruff:ignore[import-outside-top-level] - from twitch.signals import on_game_saved # ruff:ignore[import-outside-top-level] - from twitch.signals import on_reward_campaign_saved # ruff:ignore[import-outside-top-level] + from twitch.models import DropBenefit # noqa: PLC0415 + from twitch.models import DropCampaign # noqa: PLC0415 + from twitch.models import Game # noqa: PLC0415 + from twitch.models import RewardCampaign # noqa: PLC0415 + from twitch.signals import on_drop_benefit_saved # noqa: PLC0415 + from twitch.signals import on_drop_campaign_allow_channels_changed # noqa: PLC0415 + from twitch.signals import on_drop_campaign_saved # noqa: PLC0415 + from twitch.signals import on_game_saved # noqa: PLC0415 + from twitch.signals import on_reward_campaign_saved # noqa: PLC0415 post_save.connect(on_game_saved, sender=Game) post_save.connect(on_drop_campaign_saved, sender=DropCampaign) diff --git a/twitch/feeds.py b/twitch/feeds.py index 3c9e904..b5c503c 100644 --- a/twitch/feeds.py +++ b/twitch/feeds.py @@ -22,7 +22,7 @@ from django.utils.html import format_html_join from django.utils.safestring import SafeText from core.base_url import build_absolute_uri -from core.base_url import get_current_site # ruff:ignore[redefined-while-unused] +from core.base_url import get_current_site # noqa: F811 from twitch.models import Channel from twitch.models import ChatBadge from twitch.models import DropCampaign @@ -185,9 +185,9 @@ class TTVDropsBaseFeed(Feed): Returns: SyndicationFeed: The feed generator instance with the correct site and URL context for absolute URL generation. """ - # TODO(TheLovinator): Refactor to avoid this mess. # ruff:ignore[missing-todo-link] + # TODO(TheLovinator): Refactor to avoid this mess. # noqa: TD003 try: - from django.contrib.sites import shortcuts as sites_shortcuts # ruff:ignore[unsorted-imports, import-outside-top-level] + from django.contrib.sites import shortcuts as sites_shortcuts # noqa: I001, PLC0415 except ImportError: sites_shortcuts = None @@ -575,7 +575,7 @@ def generate_channels_html(item: Model, *, hide_paid: bool = False) -> list[Safe ) if "twitch-chat-badges-guide" in getattr(game, "details_url", ""): - # TODO(TheLovinator): Improve detection of global emotes # ruff:ignore[missing-todo-link] + # TODO(TheLovinator): Improve detection of global emotes # noqa: TD003 parts.append( format_html( "{}", @@ -1177,7 +1177,7 @@ class GameCampaignFeed(TTVDropsBaseFeed): self._hide_paid = _query_bool(request, "hide_paid") return super().__call__(request, *args, **kwargs) - def get_object(self, request: HttpRequest, twitch_id: str) -> Game: # ruff:ignore[unused-method-argument] + def get_object(self, request: HttpRequest, twitch_id: str) -> Game: # noqa: ARG002 """Retrieve the Game instance for the given Twitch ID. Returns: @@ -1727,7 +1727,7 @@ class GameRewardCampaignFeed(TTVDropsBaseFeed): self._limit = None return super().__call__(request, *args, **kwargs) - def get_object(self, request: HttpRequest, twitch_id: str) -> Game: # ruff:ignore[unused-method-argument] + def get_object(self, request: HttpRequest, twitch_id: str) -> Game: # noqa: ARG002 """Retrieve the Game instance for the given Twitch ID. Returns: diff --git a/twitch/management/commands/backfill_image_dimensions.py b/twitch/management/commands/backfill_image_dimensions.py index 3a0b468..9d876b0 100644 --- a/twitch/management/commands/backfill_image_dimensions.py +++ b/twitch/management/commands/backfill_image_dimensions.py @@ -14,7 +14,7 @@ class Command(BaseCommand): help = "Backfill image dimensions for existing cached images" - def handle(self, *args, **options) -> None: # ruff:ignore[unused-method-argument, too-many-statements] + def handle(self, *args, **options) -> None: # noqa: ARG002, PLR0915 """Execute the command.""" total_updated = 0 diff --git a/twitch/management/commands/backup_db.py b/twitch/management/commands/backup_db.py index bcdbbd5..753e844 100644 --- a/twitch/management/commands/backup_db.py +++ b/twitch/management/commands/backup_db.py @@ -2,7 +2,7 @@ import io import json import os import shutil -import subprocess # ruff:ignore[suspicious-subprocess-import] +import subprocess # noqa: S404 from compression import zstd from datetime import datetime from pathlib import Path @@ -191,11 +191,11 @@ def _write_table_rows( connection: SQLite connection. table: Table name. """ - cursor = connection.execute(f'SELECT * FROM "{table}"') # ruff:ignore[hardcoded-sql-expression] + cursor = connection.execute(f'SELECT * FROM "{table}"') # noqa: S608 columns = [description[0] for description in cursor.description] for row in cursor.fetchall(): values = ", ".join(_sql_literal(row[idx]) for idx in range(len(columns))) - handle.write(f'INSERT INTO "{table}" VALUES ({values});\n') # ruff:ignore[hardcoded-sql-expression] + handle.write(f'INSERT INTO "{table}" VALUES ({values});\n') # noqa: S608 def _write_indexes( @@ -268,7 +268,7 @@ def _write_postgres_dump(output_path: Path, tables: list[str]) -> None: for table in tables: cmd.extend(["-t", f"public.{table}"]) - process = subprocess.Popen( # ruff:ignore[subprocess-without-shell-equals-true] + process = subprocess.Popen( # noqa: S603 cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, @@ -339,7 +339,7 @@ def _write_json_dump(output_path: Path, tables: list[str]) -> None: data: dict[str, list[dict]] = {} with django_connection.cursor() as cursor: for table in tables: - cursor.execute(f'SELECT * FROM "{table}"') # ruff:ignore[hardcoded-sql-expression] + cursor.execute(f'SELECT * FROM "{table}"') # noqa: S608 columns: list[str] = [col[0] for col in cursor.description] rows = cursor.fetchall() data[table] = [dict(zip(columns, row, strict=False)) for row in rows] diff --git a/twitch/management/commands/better_import_drops.py b/twitch/management/commands/better_import_drops.py index 03ed611..81acda0 100644 --- a/twitch/management/commands/better_import_drops.py +++ b/twitch/management/commands/better_import_drops.py @@ -403,7 +403,7 @@ def repair_partially_broken_json(raw_text: str) -> str: wrapped: str = f"[{raw_text}]" wrapped_data = json.loads(wrapped) # Validate that all items look like GraphQL responses - if isinstance(wrapped_data, list) and wrapped_data: # ruff:ignore[collapsible-if] + if isinstance(wrapped_data, list) and wrapped_data: # noqa: SIM102 # Check if all items have "data" or "extensions" (GraphQL response structure) if all( isinstance(item, dict) and ("data" in item or "extensions" in item) @@ -452,7 +452,7 @@ def repair_partially_broken_json(raw_text: str) -> str: valid_lines: list[dict[str, Any]] = [] for line in lines: - line: str = line.strip() # ruff:ignore[redefined-loop-name] + line: str = line.strip() # noqa: PLW2901 if line and line.startswith("{"): try: fixed_line: str = json_repair.repair_json(line, logging=False) @@ -749,7 +749,7 @@ class Command(BaseCommand): return channel_obj - def process_responses( # ruff:ignore[too-many-statements] + def process_responses( # noqa: PLR0915 self, responses: list[dict[str, Any]], file_path: Path, @@ -1180,7 +1180,7 @@ class Command(BaseCommand): msg: str = f"Path does not exist: {input_path}" raise CommandError(msg) - def handle(self, *args, **options) -> None: # ruff:ignore[unused-method-argument] + def handle(self, *args, **options) -> None: # noqa: ARG002 """Main entry point for the command.""" colorama_init(autoreset=True) diff --git a/twitch/management/commands/cleanup_unknown_organizations.py b/twitch/management/commands/cleanup_unknown_organizations.py index e6ec0a9..3f07263 100644 --- a/twitch/management/commands/cleanup_unknown_organizations.py +++ b/twitch/management/commands/cleanup_unknown_organizations.py @@ -49,7 +49,7 @@ class Command(BaseCommand): ), ) - def handle(self, *args: Any, **options: Any) -> None: # ruff:ignore[any-type, unused-method-argument] + def handle(self, *args: Any, **options: Any) -> None: # noqa: ANN401, ARG002 """Execute the command to detach the organization and optionally re-import data. Args: diff --git a/twitch/management/commands/download_box_art.py b/twitch/management/commands/download_box_art.py index 2a32cb2..3d1b513 100644 --- a/twitch/management/commands/download_box_art.py +++ b/twitch/management/commands/download_box_art.py @@ -39,7 +39,7 @@ class Command(BaseCommand): help="Re-download even if a local box art file already exists.", ) - def handle( # ruff:ignore[too-many-locals, too-many-statements] + def handle( # noqa: PLR0914, PLR0915 self, *_args: str, **options: str | bool | int | None, diff --git a/twitch/management/commands/import_chat_badges.py b/twitch/management/commands/import_chat_badges.py index 2e0d463..43906be 100644 --- a/twitch/management/commands/import_chat_badges.py +++ b/twitch/management/commands/import_chat_badges.py @@ -50,7 +50,7 @@ class Command(BaseCommand): help="Twitch Access Token (optional - will be obtained automatically if not provided)", ) - def handle(self, *args, **options) -> None: # ruff:ignore[unused-method-argument] + def handle(self, *args, **options) -> None: # noqa: ARG002 """Main entry point for the command. Raises: diff --git a/twitch/management/commands/import_twitch_drops_api.py b/twitch/management/commands/import_twitch_drops_api.py index 8791b18..f8119ec 100644 --- a/twitch/management/commands/import_twitch_drops_api.py +++ b/twitch/management/commands/import_twitch_drops_api.py @@ -27,7 +27,7 @@ import hashlib import json import os import pathlib -import subprocess # ruff:ignore[suspicious-subprocess-import] +import subprocess # noqa: S404 import tempfile from typing import TYPE_CHECKING from typing import Any @@ -194,7 +194,7 @@ class Command(BaseCommand): ), ) - def handle(self, *args: Any, **options: Any) -> None: # ruff:ignore[any-type, unused-method-argument, too-many-locals, too-many-statements] + def handle(self, *args: Any, **options: Any) -> None: # noqa: ANN401, ARG002, PLR0914, PLR0915 """Main entry point for the command. Raises: @@ -301,7 +301,7 @@ class Command(BaseCommand): crash_on_error=crash_on_error, skip_existing=skip_existing, ) - except Exception as exc: # ruff:ignore[blind-except] + except Exception as exc: # noqa: BLE001 msg = f"Failed to import drops.json: {exc}" self.stderr.write(self.style.ERROR(msg)) errors.append(msg) @@ -315,7 +315,7 @@ class Command(BaseCommand): crash_on_error=crash_on_error, skip_existing=skip_existing, ) - except Exception as exc: # ruff:ignore[blind-except] + except Exception as exc: # noqa: BLE001 msg = f"Failed to import rewards.json: {exc}" self.stderr.write(self.style.ERROR(msg)) errors.append(msg) @@ -353,7 +353,7 @@ class Command(BaseCommand): # Historical import via git clone # ------------------------------------------------------------------ - def _process_historical( # ruff:ignore[too-many-arguments] + def _process_historical( # noqa: PLR0913 self, source: str, *, @@ -423,7 +423,7 @@ class Command(BaseCommand): max_commits=max_commits, skip_existing=skip_existing, ) - except Exception as exc: # ruff:ignore[blind-except] + except Exception as exc: # noqa: BLE001 errors.append(str(exc)) finally: if own_clone: @@ -431,7 +431,7 @@ class Command(BaseCommand): return drops_count, rewards_count, errors - def _import_historical_file( # ruff:ignore[too-many-arguments] + def _import_historical_file( # noqa: PLR0913 self, git_dir: str, file_path: str, @@ -581,7 +581,7 @@ class Command(BaseCommand): Stdout of the git command. """ cmd = ["git", "--git-dir", git_dir, *args] - result = subprocess.run(cmd, capture_output=True, text=True, check=check) # ruff:ignore[subprocess-without-shell-equals-true] + result = subprocess.run(cmd, capture_output=True, text=True, check=check) # noqa: S603 return result.stdout def _rmtree(self, path: str) -> None: @@ -659,7 +659,7 @@ class Command(BaseCommand): ) @staticmethod - def _strip_typename(data: Any) -> Any: # ruff:ignore[any-type] + def _strip_typename(data: Any) -> Any: # noqa: ANN401 """Recursively remove ``__typename`` keys from parsed JSON data. Old commits in the repo included GraphQL ``__typename`` metadata which @@ -683,7 +683,7 @@ class Command(BaseCommand): return [Command._strip_typename(item) for item in data] return data - def _generate_report( # ruff:ignore[too-many-locals, too-many-statements] + def _generate_report( # noqa: PLR0914, PLR0915 self, drops_url: str, rewards_url: str, @@ -707,7 +707,7 @@ class Command(BaseCommand): Raises: ValidationError: If any item fails schema validation and crash_on_error is True. """ - from collections import Counter # ruff:ignore[import-outside-top-level] + from collections import Counter # noqa: PLC0415 self.stdout.write("=" * 60) self.stdout.write(self.style.SUCCESS("COMPARISON REPORT")) @@ -719,14 +719,14 @@ class Command(BaseCommand): # Track differences for field-level comparison sample_limit = 5 - for label, raw_url, model_cls, id_field in [ # ruff:ignore[too-many-nested-blocks] + for label, raw_url, model_cls, id_field in [ # noqa: PLR1702 ("drops", drops_url, DropCampaign, "twitch_id"), ("rewards", rewards_url, RewardCampaign, "twitch_id"), ]: self.stdout.write(f"\n--- {label.upper()} ---") try: raw_data = self._fetch_json(raw_url, f"{label}.json") - except Exception as exc: # ruff:ignore[blind-except] + except Exception as exc: # noqa: BLE001 self.stderr.write( self.style.ERROR(f" Failed to fetch {label}.json: {exc}"), ) @@ -894,7 +894,7 @@ class Command(BaseCommand): self.stdout.write("\n" + "=" * 60) - def _parse_and_import_drops( # ruff:ignore[too-many-arguments] + def _parse_and_import_drops( # noqa: PLR0913 self, raw_data: list[dict[str, Any]], source: str, @@ -976,7 +976,7 @@ class Command(BaseCommand): return total_campaigns - def _process_sunkwibot_reward( # ruff:ignore[too-many-statements] + def _process_sunkwibot_reward( # noqa: PLR0915 self, reward: SunkwiBotRewardSchema, game_obj: Game, @@ -1133,7 +1133,7 @@ class Command(BaseCommand): self, tbd_schema: SunkwiBotTimeBasedDropSchema, campaign_obj: DropCampaign, - source: str, # ruff:ignore[unused-method-argument] + source: str, # noqa: ARG002 *, verbose: bool, ) -> None: @@ -1335,7 +1335,7 @@ class Command(BaseCommand): skip_existing=skip_existing, ) - def _parse_and_import_rewards( # ruff:ignore[too-many-arguments] + def _parse_and_import_rewards( # noqa: PLR0913 self, raw_data: list[dict[str, Any]], source: str, @@ -1563,12 +1563,12 @@ class Command(BaseCommand): self._org_cache[twitch_id] = org_obj return org_obj - def _get_or_create_game( # ruff:ignore[too-many-arguments] + def _get_or_create_game( # noqa: PLR0913 self, twitch_id: str, display_name: str, box_art_url: str, - source: str, # ruff:ignore[unused-method-argument] + source: str, # noqa: ARG002 *, verbose: bool, org_obj: Organization | None = None, diff --git a/twitch/management/commands/watch_imports.py b/twitch/management/commands/watch_imports.py index 96e8549..d1f6ca3 100644 --- a/twitch/management/commands/watch_imports.py +++ b/twitch/management/commands/watch_imports.py @@ -37,7 +37,7 @@ class Command(BaseCommand): help="Path to directory to watch for JSON files", ) - def handle(self, *args, **options) -> None: # ruff:ignore[unused-method-argument] + def handle(self, *args, **options) -> None: # noqa: ARG002 """Main entry point for the watch command. Args: @@ -87,7 +87,7 @@ class Command(BaseCommand): importer_command: An instance of the BetterImportDropsCommand to handle the import logic. watch_path: The directory path to watch for JSON files. """ - # TODO(TheLovinator): Implement actual file watching using watchdog or similar library. # ruff:ignore[missing-todo-link] + # TODO(TheLovinator): Implement actual file watching using watchdog or similar library. # noqa: TD003 json_files: list[Path] = [ f for f in watch_path.iterdir() if f.suffix == ".json" and f.is_file() ] diff --git a/twitch/migrations/0007_rename_operation_name_to_operation_names.py b/twitch/migrations/0007_rename_operation_name_to_operation_names.py index bbe6ac2..63190e1 100644 --- a/twitch/migrations/0007_rename_operation_name_to_operation_names.py +++ b/twitch/migrations/0007_rename_operation_name_to_operation_names.py @@ -5,7 +5,7 @@ from django.db import migrations from django.db import models -def migrate_operation_name_to_list(apps, schema_editor) -> None: # ruff:ignore[missing-type-function-argument] +def migrate_operation_name_to_list(apps, schema_editor) -> None: # noqa: ANN001 """Convert operation_name string values to operation_names list.""" DropCampaign = apps.get_model("twitch", "DropCampaign") for campaign in DropCampaign.objects.all(): @@ -14,7 +14,7 @@ def migrate_operation_name_to_list(apps, schema_editor) -> None: # ruff:ignore[ campaign.save(update_fields=["operation_names"]) -def reverse_operation_names_to_string(apps, schema_editor) -> None: # ruff:ignore[missing-type-function-argument] +def reverse_operation_names_to_string(apps, schema_editor) -> None: # noqa: ANN001 """Convert operation_names list back to operation_name string (first item only).""" DropCampaign = apps.get_model("twitch", "DropCampaign") for campaign in DropCampaign.objects.all(): diff --git a/twitch/migrations/0016_mark_all_drops_fully_imported.py b/twitch/migrations/0016_mark_all_drops_fully_imported.py index 5a9ef00..54b4003 100644 --- a/twitch/migrations/0016_mark_all_drops_fully_imported.py +++ b/twitch/migrations/0016_mark_all_drops_fully_imported.py @@ -1,7 +1,7 @@ from django.db import migrations -def mark_all_drops_fully_imported(apps, schema_editor) -> None: # ruff:ignore[missing-type-function-argument] +def mark_all_drops_fully_imported(apps, schema_editor) -> None: # noqa: ANN001 """Marks all existing DropCampaigns as fully imported. This was needed to ensure that the Twitch API view only returns campaigns that are ready for display. diff --git a/twitch/models.py b/twitch/models.py index 7ff20f7..6aeb0f1 100644 --- a/twitch/models.py +++ b/twitch/models.py @@ -137,7 +137,7 @@ class Game(auto_prefetch.Model): default="", ) - box_art = models.URLField( # ruff:ignore[django-nullable-model-string-field] + box_art = models.URLField( # noqa: DJ001 verbose_name="Box art URL", max_length=500, blank=True, @@ -231,12 +231,12 @@ class Game(auto_prefetch.Model): @property def organizations(self) -> models.QuerySet[Organization]: - """Orgs that own games with campaigns for this game.""" + """Return orgs that own games with campaigns for this game.""" return Organization.objects.filter(games__drop_campaigns__game=self).distinct() @property def get_game_name(self) -> str: - """The best available name for the game.""" + """Return the best available name for the game.""" if self.display_name: return self.display_name if self.name: @@ -247,8 +247,8 @@ class Game(auto_prefetch.Model): @property def twitch_directory_url(self) -> str: - """Twitch directory URL with drops filter when slug exists.""" - # TODO(TheLovinator): If no slug, get from Twitch API or IGDB? # ruff:ignore[missing-todo-link] + """Return Twitch directory URL with drops filter when slug exists.""" + # TODO(TheLovinator): If no slug, get from Twitch API or IGDB? # noqa: TD003 if self.slug: return f"https://www.twitch.tv/directory/category/{self.slug}?filter=drops" @@ -256,7 +256,7 @@ class Game(auto_prefetch.Model): @property def box_art_best_url(self) -> str: - """The best available URL for the game's box art (local first).""" + """Return the best available URL for the game's box art (local first).""" try: if self.box_art_file and getattr(self.box_art_file, "url", None): return self.box_art_file.url @@ -271,7 +271,7 @@ class Game(auto_prefetch.Model): @property def dashboard_box_art_url(self) -> str: - """Dashboard-safe box art URL without touching deferred image fields.""" + """Return dashboard-safe box art URL without touching deferred image fields.""" return normalize_twitch_box_art_url(self.box_art or "") @classmethod @@ -570,7 +570,7 @@ class Channel(auto_prefetch.Model): @property def preferred_name(self) -> str: - """Display name fallback used by channel-facing pages.""" + """Return display name fallback used by channel-facing pages.""" return self.display_name or self.name or self.twitch_id def detail_description(self, total_campaigns: int) -> str: @@ -580,7 +580,7 @@ class Channel(auto_prefetch.Model): # MARK: DropCampaign -class DropCampaign(auto_prefetch.Model): # ruff:ignore[too-many-public-methods] +class DropCampaign(auto_prefetch.Model): # noqa: PLR0904 """Represents a Twitch drop campaign.""" twitch_id = models.TextField( @@ -801,10 +801,6 @@ class DropCampaign(auto_prefetch.Model): # ruff:ignore[too-many-public-methods] queryset = queryset.filter(start_at__gt=now) elif status == "expired": queryset = queryset.filter(end_at__lt=now) - elif status == "unknown": - queryset = queryset.filter( - Q(start_at__isnull=True) | Q(end_at__isnull=True), - ) return queryset @classmethod @@ -1270,7 +1266,7 @@ class DropCampaign(auto_prefetch.Model): # ruff:ignore[too-many-public-methods] @property def clean_name(self) -> str: - """The campaign name without the game name prefix. + """Return the campaign name without the game name prefix. Examples: "Ravendawn - July 2" -> "July 2" @@ -1302,7 +1298,7 @@ class DropCampaign(auto_prefetch.Model): # ruff:ignore[too-many-public-methods] @property def single_reward_benefit(self) -> DropBenefit | None: - """The only unique reward benefit for this campaign, if it has one.""" + """Return the only unique reward benefit for this campaign, if it has one.""" benefits: list[DropBenefit] = [] seen_benefit_keys: set[int | str] = set() @@ -1321,7 +1317,7 @@ class DropCampaign(auto_prefetch.Model): # ruff:ignore[too-many-public-methods] @property def single_reward_image_best_url(self) -> str: - """The best image URL for a campaign that has exactly one reward.""" + """Return the best image URL for a campaign that has exactly one reward.""" benefit: DropBenefit | None = self.single_reward_benefit if not benefit: return "" @@ -1329,12 +1325,12 @@ class DropCampaign(auto_prefetch.Model): # ruff:ignore[too-many-public-methods] @property def meta_image_url(self) -> str: - """The preferred campaign image URL for SEO metadata.""" + """Return the preferred campaign image URL for SEO metadata.""" return self.single_reward_image_best_url or self.image_best_url @property def image_best_url(self) -> str: - """The best URL for the campaign image. + """Return the best URL for the campaign image. Priority: 1. Local cached image file @@ -1361,7 +1357,7 @@ class DropCampaign(auto_prefetch.Model): # ruff:ignore[too-many-public-methods] @property def listing_image_url(self) -> str: - """A campaign image URL optimized for list views. + """Return a campaign image URL optimized for list views. This intentionally avoids traversing drops/benefits to prevent N+1 queries in list pages that render many campaigns. @@ -1375,7 +1371,7 @@ class DropCampaign(auto_prefetch.Model): # ruff:ignore[too-many-public-methods] @property def dashboard_image_url(self) -> str: - """Dashboard-safe campaign or single-reward image URL.""" + """Return dashboard-safe campaign or single-reward image URL.""" benefit: DropBenefit | None = self.single_reward_benefit if benefit and benefit.image_asset_url: return benefit.image_asset_url @@ -1383,7 +1379,7 @@ class DropCampaign(auto_prefetch.Model): # ruff:ignore[too-many-public-methods] @property def duration_iso(self) -> str: - """The campaign duration in ISO 8601 format (e.g., 'P3DT4H30M'). + """Return the campaign duration in ISO 8601 format (e.g., 'P3DT4H30M'). This is used for the