Update Ruff and fix its errors
This commit is contained in:
parent
c54ceeb7a8
commit
1424978854
Signed by:
Joakim Hellsén
Joakim Hellsén
SSH key fingerprint:
SHA256:/9h/CsExpFp+PRhsfA0xznFx2CGfTT5R/kpuFfUgEQk
39 changed files with 183 additions and 175 deletions
|
|
@ -21,13 +21,13 @@ repos:
|
|||
- id: trailing-whitespace
|
||||
|
||||
- repo: https://github.com/adamchainz/django-upgrade
|
||||
rev: 1.30.0
|
||||
rev: 1.31.1
|
||||
hooks:
|
||||
- id: django-upgrade
|
||||
args: [--target-version, "6.0"]
|
||||
|
||||
- repo: https://github.com/astral-sh/ruff-pre-commit
|
||||
rev: v0.15.17
|
||||
rev: v0.15.22
|
||||
hooks:
|
||||
- id: ruff-check
|
||||
args: ["--fix", "--exit-non-zero-on-fix"]
|
||||
|
|
|
|||
|
|
@ -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 # noqa: TD003
|
||||
api_version: str = "v2" # TODO(TheLovinator): Add support for v1 API # ruff:ignore[missing-todo-link]
|
||||
url: str = f"https://api.chzzk.naver.com/service/{api_version}/drops/campaigns/{campaign_no}"
|
||||
resp: requests.Response = requests.get(
|
||||
url,
|
||||
|
|
|
|||
|
|
@ -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: # noqa: ANN001
|
||||
def import_chzzk_campaign_task(self, campaign_no: int) -> None: # ruff:ignore[missing-type-function-argument]
|
||||
"""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: # noqa: ANN001
|
|||
|
||||
|
||||
@shared_task(bind=True, queue="api-fetches", max_retries=3, default_retry_delay=120)
|
||||
def discover_chzzk_campaigns(self) -> None: # noqa: ANN001
|
||||
def discover_chzzk_campaigns(self) -> None: # ruff:ignore[missing-type-function-argument]
|
||||
"""Discover and import the latest Chzzk campaigns (equivalent to --latest flag)."""
|
||||
try:
|
||||
call_command("import_chzzk_campaign", latest=True)
|
||||
|
|
|
|||
|
|
@ -23,10 +23,10 @@ classes = [
|
|||
]
|
||||
|
||||
for cls in classes:
|
||||
setattr( # noqa: B010
|
||||
setattr( # ruff:ignore[set-attr-with-constant]
|
||||
cls,
|
||||
"__class_getitem__",
|
||||
classmethod(lambda cls, *args, **kwargs: cls), # noqa: ARG005
|
||||
classmethod(lambda cls, *args, **kwargs: cls), # ruff:ignore[unused-lambda-argument]
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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: # noqa: F841
|
||||
with patch("os.environ.setdefault") as mock_setenv: # ruff:ignore[unused-variable]
|
||||
app = Celery("config")
|
||||
app.config_from_object("django.conf:settings", namespace="CELERY")
|
||||
yield app
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ if TYPE_CHECKING:
|
|||
from collections.abc import Generator
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True) # noqa: RUF076 — intentional: project-wide N+1 detection with @pytest.mark.no_zeal escape hatch
|
||||
@pytest.fixture(autouse=True)
|
||||
def use_zeal(request: pytest.FixtureRequest) -> Generator[None]:
|
||||
"""Enable Zeal N+1 detection context for each pytest test.
|
||||
|
||||
|
|
|
|||
|
|
@ -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: # noqa: ANN001
|
||||
def submit_indexnow_task(self) -> None: # ruff:ignore[missing-type-function-argument]
|
||||
"""Submit all site URLs to the IndexNow search index."""
|
||||
try:
|
||||
call_command("submit_indexnow")
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
import xml.etree.ElementTree as ET # noqa: S405
|
||||
import xml.etree.ElementTree as ET # ruff:ignore[suspicious-xml-etree-import]
|
||||
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) # noqa: S314
|
||||
root = ET.fromstring(xml_bytes) # ruff:ignore[suspicious-xml-element-tree-usage]
|
||||
ns = {"s": "http://www.sitemaps.org/schemas/sitemap/0.9"}
|
||||
return [el.text for el in root.findall(".//s:loc", ns) if el.text]
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
from typing import Any
|
||||
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 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 django.conf import settings
|
||||
|
||||
|
|
|
|||
|
|
@ -53,7 +53,7 @@ MIN_SEARCH_RANK = 0.05
|
|||
DEFAULT_SITE_DESCRIPTION = "Archive of Twitch drops, campaigns, rewards, and more."
|
||||
|
||||
|
||||
def _build_seo_context( # noqa: PLR0913, PLR0917
|
||||
def _build_seo_context( # ruff:ignore[too-many-arguments, too-many-positional-arguments]
|
||||
page_title: str = "ttvdrops",
|
||||
page_description: str | None = None,
|
||||
page_url: str | None = None,
|
||||
|
|
@ -91,7 +91,7 @@ def _build_seo_context( # noqa: PLR0913, PLR0917
|
|||
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, # noqa: TD003
|
||||
# TODO(TheLovinator): Instead of having so many parameters, # ruff:ignore[missing-todo-link]
|
||||
# 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: # noqa: PLR0914
|
||||
def debug_view(request: HttpRequest) -> HttpResponse: # ruff:ignore[too-many-locals]
|
||||
"""Debug view showing potentially broken or inconsistent data.
|
||||
|
||||
Returns:
|
||||
|
|
@ -772,8 +772,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. # noqa: TD003
|
||||
# TODO(TheLovinator): Upload to s3 instead. # noqa: TD003
|
||||
# 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): 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 +889,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. # noqa: TD003
|
||||
# TODO(TheLovinator): Use s3 instead of local disk. # ruff:ignore[missing-todo-link]
|
||||
|
||||
datasets_root: Path = settings.DATA_DIR / "datasets"
|
||||
requested_path: Path = (datasets_root / relative_path).resolve()
|
||||
|
|
@ -994,7 +994,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'." # noqa: TD003
|
||||
# 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]
|
||||
if query:
|
||||
page_title: str = f"Search Results for '{query}'"[:60]
|
||||
page_description: str = f"Found {total_results_count} results for '{query}'."
|
||||
|
|
|
|||
|
|
@ -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: # noqa: ARG002
|
||||
def get_object(self, request: HttpRequest, kick_id: int) -> KickCategory: # ruff:ignore[unused-method-argument]
|
||||
"""Return game object for this feed URL."""
|
||||
return KickCategory.objects.get(kick_id=kick_id)
|
||||
|
||||
|
|
|
|||
|
|
@ -130,7 +130,7 @@ class Command(BaseCommand):
|
|||
|
||||
try:
|
||||
payload: dict = response.json()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
except Exception as exc: # ruff:ignore[blind-except]
|
||||
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: # noqa: PLR0914, PLR0915
|
||||
def _import_campaign(self, data: KickDropCampaignSchema) -> None: # ruff:ignore[too-many-locals, too-many-statements]
|
||||
"""Import a single campaign and all its related objects."""
|
||||
# Organization
|
||||
org_data: KickOrganizationSchema = data.organization
|
||||
|
|
|
|||
|
|
@ -93,12 +93,12 @@ class KickCategory(auto_prefetch.Model):
|
|||
|
||||
@property
|
||||
def get_absolute_url(self) -> str:
|
||||
"""Return the URL to the game detail page."""
|
||||
"""The URL to the game detail page."""
|
||||
return reverse("kick:game_detail", args=[self.kick_id])
|
||||
|
||||
@property
|
||||
def kick_url(self) -> str:
|
||||
"""Return the URL to the game page on Kick."""
|
||||
"""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:
|
||||
"""Return the Kick profile URL for this user."""
|
||||
"""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:
|
||||
"""Return the Kick channel URL."""
|
||||
"""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:
|
||||
"""Return the image URL for the campaign."""
|
||||
"""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]:
|
||||
"""Return rewards de-duplicated by normalized name.
|
||||
"""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:
|
||||
"""Return the absolute image URL for this reward.
|
||||
"""The absolute image URL for this reward.
|
||||
|
||||
If the image_url is a relative path, prepend the Kick image base URL.
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
from datetime import datetime # noqa: TC003
|
||||
from datetime import datetime # ruff:ignore[typing-only-standard-library-import]
|
||||
|
||||
from pydantic import BaseModel
|
||||
from pydantic import Field
|
||||
|
|
|
|||
|
|
@ -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: # noqa: ANN001
|
||||
def import_kick_drops(self) -> None: # ruff:ignore[missing-type-function-argument]
|
||||
"""Fetch and upsert Kick drop campaigns from the public API."""
|
||||
try:
|
||||
call_command("import_kick_drops")
|
||||
|
|
|
|||
|
|
@ -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]: # noqa: ANN401
|
||||
def _run_command(self, json_payload: dict, **options: Any) -> tuple[str, str]: # ruff:ignore[any-type]
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = json_payload
|
||||
|
|
|
|||
|
|
@ -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 # noqa: PLC0415
|
||||
from django.core.management import execute_from_command_line # ruff:ignore[import-outside-top-level, unsorted-imports]
|
||||
except ImportError as exc:
|
||||
msg = (
|
||||
"Couldn't import Django. Are you sure it's installed and "
|
||||
|
|
|
|||
|
|
@ -16,7 +16,7 @@ import argparse
|
|||
import hashlib
|
||||
import json
|
||||
import shutil
|
||||
import subprocess # noqa: S404
|
||||
import subprocess # ruff:ignore[suspicious-subprocess-import]
|
||||
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( # noqa: S603
|
||||
result: subprocess.CompletedProcess[str] = subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
|
||||
[git, "--git-dir", git_dir, *args],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import datetime # noqa: TC003
|
||||
import datetime # ruff:ignore[typing-only-standard-library-import]
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import Literal
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@ class TwitchConfig(AppConfig):
|
|||
default_auto_field = "django.db.models.BigAutoField"
|
||||
name = "twitch"
|
||||
|
||||
def ready(self) -> None: # noqa: D102
|
||||
def ready(self) -> None: # ruff:ignore[undocumented-public-method]
|
||||
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 # noqa: I001, PLC0415
|
||||
from django.db.models.signals import post_save # noqa: PLC0415
|
||||
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 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
|
||||
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]
|
||||
|
||||
post_save.connect(on_game_saved, sender=Game)
|
||||
post_save.connect(on_drop_campaign_saved, sender=DropCampaign)
|
||||
|
|
|
|||
|
|
@ -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 # noqa: F811
|
||||
from core.base_url import get_current_site # ruff:ignore[redefined-while-unused]
|
||||
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. # noqa: TD003
|
||||
# TODO(TheLovinator): Refactor to avoid this mess. # ruff:ignore[missing-todo-link]
|
||||
try:
|
||||
from django.contrib.sites import shortcuts as sites_shortcuts # noqa: I001, PLC0415
|
||||
from django.contrib.sites import shortcuts as sites_shortcuts # ruff:ignore[unsorted-imports, import-outside-top-level]
|
||||
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 # noqa: TD003
|
||||
# TODO(TheLovinator): Improve detection of global emotes # ruff:ignore[missing-todo-link]
|
||||
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: # noqa: ARG002
|
||||
def get_object(self, request: HttpRequest, twitch_id: str) -> Game: # ruff:ignore[unused-method-argument]
|
||||
"""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: # noqa: ARG002
|
||||
def get_object(self, request: HttpRequest, twitch_id: str) -> Game: # ruff:ignore[unused-method-argument]
|
||||
"""Retrieve the Game instance for the given Twitch ID.
|
||||
|
||||
Returns:
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ class Command(BaseCommand):
|
|||
|
||||
help = "Backfill image dimensions for existing cached images"
|
||||
|
||||
def handle(self, *args, **options) -> None: # noqa: ARG002, PLR0915
|
||||
def handle(self, *args, **options) -> None: # ruff:ignore[unused-method-argument, too-many-statements]
|
||||
"""Execute the command."""
|
||||
total_updated = 0
|
||||
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ import io
|
|||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess # noqa: S404
|
||||
import subprocess # ruff:ignore[suspicious-subprocess-import]
|
||||
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}"') # noqa: S608
|
||||
cursor = connection.execute(f'SELECT * FROM "{table}"') # ruff:ignore[hardcoded-sql-expression]
|
||||
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') # noqa: S608
|
||||
handle.write(f'INSERT INTO "{table}" VALUES ({values});\n') # ruff:ignore[hardcoded-sql-expression]
|
||||
|
||||
|
||||
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( # noqa: S603
|
||||
process = subprocess.Popen( # ruff:ignore[subprocess-without-shell-equals-true]
|
||||
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}"') # noqa: S608
|
||||
cursor.execute(f'SELECT * FROM "{table}"') # ruff:ignore[hardcoded-sql-expression]
|
||||
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]
|
||||
|
|
|
|||
|
|
@ -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: # noqa: SIM102
|
||||
if isinstance(wrapped_data, list) and wrapped_data: # ruff:ignore[collapsible-if]
|
||||
# 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() # noqa: PLW2901
|
||||
line: str = line.strip() # ruff:ignore[redefined-loop-name]
|
||||
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( # noqa: PLR0915
|
||||
def process_responses( # ruff:ignore[too-many-statements]
|
||||
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: # noqa: ARG002
|
||||
def handle(self, *args, **options) -> None: # ruff:ignore[unused-method-argument]
|
||||
"""Main entry point for the command."""
|
||||
colorama_init(autoreset=True)
|
||||
|
||||
|
|
|
|||
|
|
@ -49,7 +49,7 @@ class Command(BaseCommand):
|
|||
),
|
||||
)
|
||||
|
||||
def handle(self, *args: Any, **options: Any) -> None: # noqa: ANN401, ARG002
|
||||
def handle(self, *args: Any, **options: Any) -> None: # ruff:ignore[any-type, unused-method-argument]
|
||||
"""Execute the command to detach the organization and optionally re-import data.
|
||||
|
||||
Args:
|
||||
|
|
|
|||
|
|
@ -39,7 +39,7 @@ class Command(BaseCommand):
|
|||
help="Re-download even if a local box art file already exists.",
|
||||
)
|
||||
|
||||
def handle( # noqa: PLR0914, PLR0915
|
||||
def handle( # ruff:ignore[too-many-locals, too-many-statements]
|
||||
self,
|
||||
*_args: str,
|
||||
**options: str | bool | int | None,
|
||||
|
|
|
|||
|
|
@ -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: # noqa: ARG002
|
||||
def handle(self, *args, **options) -> None: # ruff:ignore[unused-method-argument]
|
||||
"""Main entry point for the command.
|
||||
|
||||
Raises:
|
||||
|
|
|
|||
|
|
@ -27,7 +27,7 @@ import hashlib
|
|||
import json
|
||||
import os
|
||||
import pathlib
|
||||
import subprocess # noqa: S404
|
||||
import subprocess # ruff:ignore[suspicious-subprocess-import]
|
||||
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: # noqa: ANN401, ARG002, PLR0914, PLR0915
|
||||
def handle(self, *args: Any, **options: Any) -> None: # ruff:ignore[any-type, unused-method-argument, too-many-locals, too-many-statements]
|
||||
"""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: # noqa: BLE001
|
||||
except Exception as exc: # ruff:ignore[blind-except]
|
||||
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: # noqa: BLE001
|
||||
except Exception as exc: # ruff:ignore[blind-except]
|
||||
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( # noqa: PLR0913
|
||||
def _process_historical( # ruff:ignore[too-many-arguments]
|
||||
self,
|
||||
source: str,
|
||||
*,
|
||||
|
|
@ -423,7 +423,7 @@ class Command(BaseCommand):
|
|||
max_commits=max_commits,
|
||||
skip_existing=skip_existing,
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
except Exception as exc: # ruff:ignore[blind-except]
|
||||
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( # noqa: PLR0913
|
||||
def _import_historical_file( # ruff:ignore[too-many-arguments]
|
||||
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) # noqa: S603
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, check=check) # ruff:ignore[subprocess-without-shell-equals-true]
|
||||
return result.stdout
|
||||
|
||||
def _rmtree(self, path: str) -> None:
|
||||
|
|
@ -659,7 +659,7 @@ class Command(BaseCommand):
|
|||
)
|
||||
|
||||
@staticmethod
|
||||
def _strip_typename(data: Any) -> Any: # noqa: ANN401
|
||||
def _strip_typename(data: Any) -> Any: # ruff:ignore[any-type]
|
||||
"""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( # noqa: PLR0914, PLR0915
|
||||
def _generate_report( # ruff:ignore[too-many-locals, too-many-statements]
|
||||
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 # noqa: PLC0415
|
||||
from collections import Counter # ruff:ignore[import-outside-top-level]
|
||||
|
||||
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 [ # noqa: PLR1702
|
||||
for label, raw_url, model_cls, id_field in [ # ruff:ignore[too-many-nested-blocks]
|
||||
("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: # noqa: BLE001
|
||||
except Exception as exc: # ruff:ignore[blind-except]
|
||||
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( # noqa: PLR0913
|
||||
def _parse_and_import_drops( # ruff:ignore[too-many-arguments]
|
||||
self,
|
||||
raw_data: list[dict[str, Any]],
|
||||
source: str,
|
||||
|
|
@ -976,7 +976,7 @@ class Command(BaseCommand):
|
|||
|
||||
return total_campaigns
|
||||
|
||||
def _process_sunkwibot_reward( # noqa: PLR0915
|
||||
def _process_sunkwibot_reward( # ruff:ignore[too-many-statements]
|
||||
self,
|
||||
reward: SunkwiBotRewardSchema,
|
||||
game_obj: Game,
|
||||
|
|
@ -1133,7 +1133,7 @@ class Command(BaseCommand):
|
|||
self,
|
||||
tbd_schema: SunkwiBotTimeBasedDropSchema,
|
||||
campaign_obj: DropCampaign,
|
||||
source: str, # noqa: ARG002
|
||||
source: str, # ruff:ignore[unused-method-argument]
|
||||
*,
|
||||
verbose: bool,
|
||||
) -> None:
|
||||
|
|
@ -1335,7 +1335,7 @@ class Command(BaseCommand):
|
|||
skip_existing=skip_existing,
|
||||
)
|
||||
|
||||
def _parse_and_import_rewards( # noqa: PLR0913
|
||||
def _parse_and_import_rewards( # ruff:ignore[too-many-arguments]
|
||||
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( # noqa: PLR0913
|
||||
def _get_or_create_game( # ruff:ignore[too-many-arguments]
|
||||
self,
|
||||
twitch_id: str,
|
||||
display_name: str,
|
||||
box_art_url: str,
|
||||
source: str, # noqa: ARG002
|
||||
source: str, # ruff:ignore[unused-method-argument]
|
||||
*,
|
||||
verbose: bool,
|
||||
org_obj: Organization | None = None,
|
||||
|
|
|
|||
|
|
@ -37,7 +37,7 @@ class Command(BaseCommand):
|
|||
help="Path to directory to watch for JSON files",
|
||||
)
|
||||
|
||||
def handle(self, *args, **options) -> None: # noqa: ARG002
|
||||
def handle(self, *args, **options) -> None: # ruff:ignore[unused-method-argument]
|
||||
"""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. # noqa: TD003
|
||||
# TODO(TheLovinator): Implement actual file watching using watchdog or similar library. # ruff:ignore[missing-todo-link]
|
||||
json_files: list[Path] = [
|
||||
f for f in watch_path.iterdir() if f.suffix == ".json" and f.is_file()
|
||||
]
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ from django.db import migrations
|
|||
from django.db import models
|
||||
|
||||
|
||||
def migrate_operation_name_to_list(apps, schema_editor) -> None: # noqa: ANN001
|
||||
def migrate_operation_name_to_list(apps, schema_editor) -> None: # ruff:ignore[missing-type-function-argument]
|
||||
"""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: # noqa: ANN001
|
|||
campaign.save(update_fields=["operation_names"])
|
||||
|
||||
|
||||
def reverse_operation_names_to_string(apps, schema_editor) -> None: # noqa: ANN001
|
||||
def reverse_operation_names_to_string(apps, schema_editor) -> None: # ruff:ignore[missing-type-function-argument]
|
||||
"""Convert operation_names list back to operation_name string (first item only)."""
|
||||
DropCampaign = apps.get_model("twitch", "DropCampaign")
|
||||
for campaign in DropCampaign.objects.all():
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
from django.db import migrations
|
||||
|
||||
|
||||
def mark_all_drops_fully_imported(apps, schema_editor) -> None: # noqa: ANN001
|
||||
def mark_all_drops_fully_imported(apps, schema_editor) -> None: # ruff:ignore[missing-type-function-argument]
|
||||
"""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.
|
||||
|
|
|
|||
|
|
@ -137,7 +137,7 @@ class Game(auto_prefetch.Model):
|
|||
default="",
|
||||
)
|
||||
|
||||
box_art = models.URLField( # noqa: DJ001
|
||||
box_art = models.URLField( # ruff:ignore[django-nullable-model-string-field]
|
||||
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]:
|
||||
"""Return orgs that own games with campaigns for this game."""
|
||||
"""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:
|
||||
"""Return the best available name for the game."""
|
||||
"""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:
|
||||
"""Return Twitch directory URL with drops filter when slug exists."""
|
||||
# TODO(TheLovinator): If no slug, get from Twitch API or IGDB? # noqa: TD003
|
||||
"""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]
|
||||
|
||||
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:
|
||||
"""Return the best available URL for the game's box art (local first)."""
|
||||
"""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:
|
||||
"""Return dashboard-safe box art URL without touching deferred image fields."""
|
||||
"""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:
|
||||
"""Return display name fallback used by channel-facing pages."""
|
||||
"""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): # noqa: PLR0904
|
||||
class DropCampaign(auto_prefetch.Model): # ruff:ignore[too-many-public-methods]
|
||||
"""Represents a Twitch drop campaign."""
|
||||
|
||||
twitch_id = models.TextField(
|
||||
|
|
@ -1270,7 +1270,7 @@ class DropCampaign(auto_prefetch.Model): # noqa: PLR0904
|
|||
|
||||
@property
|
||||
def clean_name(self) -> str:
|
||||
"""Return the campaign name without the game name prefix.
|
||||
"""The campaign name without the game name prefix.
|
||||
|
||||
Examples:
|
||||
"Ravendawn - July 2" -> "July 2"
|
||||
|
|
@ -1302,7 +1302,7 @@ class DropCampaign(auto_prefetch.Model): # noqa: PLR0904
|
|||
|
||||
@property
|
||||
def single_reward_benefit(self) -> DropBenefit | None:
|
||||
"""Return the only unique reward benefit for this campaign, if it has one."""
|
||||
"""The only unique reward benefit for this campaign, if it has one."""
|
||||
benefits: list[DropBenefit] = []
|
||||
seen_benefit_keys: set[int | str] = set()
|
||||
|
||||
|
|
@ -1321,7 +1321,7 @@ class DropCampaign(auto_prefetch.Model): # noqa: PLR0904
|
|||
|
||||
@property
|
||||
def single_reward_image_best_url(self) -> str:
|
||||
"""Return the best image URL for a campaign that has exactly one reward."""
|
||||
"""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 +1329,12 @@ class DropCampaign(auto_prefetch.Model): # noqa: PLR0904
|
|||
|
||||
@property
|
||||
def meta_image_url(self) -> str:
|
||||
"""Return the preferred campaign image URL for SEO metadata."""
|
||||
"""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:
|
||||
"""Return the best URL for the campaign image.
|
||||
"""The best URL for the campaign image.
|
||||
|
||||
Priority:
|
||||
1. Local cached image file
|
||||
|
|
@ -1361,7 +1361,7 @@ class DropCampaign(auto_prefetch.Model): # noqa: PLR0904
|
|||
|
||||
@property
|
||||
def listing_image_url(self) -> str:
|
||||
"""Return a campaign image URL optimized for list views.
|
||||
"""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 +1375,7 @@ class DropCampaign(auto_prefetch.Model): # noqa: PLR0904
|
|||
|
||||
@property
|
||||
def dashboard_image_url(self) -> str:
|
||||
"""Return dashboard-safe campaign or single-reward image URL."""
|
||||
"""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 +1383,7 @@ class DropCampaign(auto_prefetch.Model): # noqa: PLR0904
|
|||
|
||||
@property
|
||||
def duration_iso(self) -> str:
|
||||
"""Return the campaign duration in ISO 8601 format (e.g., 'P3DT4H30M').
|
||||
"""The campaign duration in ISO 8601 format (e.g., 'P3DT4H30M').
|
||||
|
||||
This is used for the <time> element's datetime attribute to provide
|
||||
machine-readable duration. If start_at or end_at is missing, returns
|
||||
|
|
@ -1421,7 +1421,7 @@ class DropCampaign(auto_prefetch.Model): # noqa: PLR0904
|
|||
|
||||
@property
|
||||
def sorted_benefits(self) -> list[DropBenefit]:
|
||||
"""Return a sorted list of benefits for the campaign."""
|
||||
"""A sorted list of benefits for the campaign."""
|
||||
benefits: list[DropBenefit] = []
|
||||
for drop in self.time_based_drops.all(): # pyright: ignore[reportAttributeAccessIssue]
|
||||
benefits.extend(drop.benefits.all()) # pyright: ignore[reportAttributeAccessIssue]
|
||||
|
|
@ -1570,7 +1570,7 @@ class DropBenefit(auto_prefetch.Model):
|
|||
|
||||
@property
|
||||
def image_best_url(self) -> str:
|
||||
"""Return the best URL for the benefit image (local first)."""
|
||||
"""The best URL for the benefit image (local first)."""
|
||||
try:
|
||||
if self.image_file:
|
||||
file_name: str = getattr(self.image_file, "name", "")
|
||||
|
|
@ -1917,7 +1917,7 @@ class RewardCampaign(auto_prefetch.Model):
|
|||
|
||||
@property
|
||||
def image_best_url(self) -> str:
|
||||
"""Return the best URL for the reward campaign image (local first)."""
|
||||
"""The best URL for the reward campaign image (local first)."""
|
||||
try:
|
||||
if self.image_file and getattr(self.image_file, "url", None):
|
||||
return self.image_file.url
|
||||
|
|
@ -2022,14 +2022,14 @@ class ChatBadge(auto_prefetch.Model):
|
|||
verbose_name="Description",
|
||||
)
|
||||
|
||||
click_action = models.TextField( # noqa: DJ001
|
||||
click_action = models.TextField( # ruff:ignore[django-nullable-model-string-field]
|
||||
help_text="The action to take when clicking on the badge (e.g., 'visit_url').",
|
||||
verbose_name="Click Action",
|
||||
blank=True,
|
||||
null=True,
|
||||
)
|
||||
|
||||
click_url = models.URLField( # noqa: DJ001
|
||||
click_url = models.URLField( # ruff:ignore[django-nullable-model-string-field]
|
||||
help_text="The URL to navigate to when clicking on the badge.",
|
||||
verbose_name="Click URL",
|
||||
max_length=500,
|
||||
|
|
|
|||
|
|
@ -8,11 +8,11 @@ from django.db.models import Count
|
|||
logger = logging.getLogger("ttvdrops.signals")
|
||||
|
||||
|
||||
def _dispatch(task_fn: Any, pk: int) -> None: # noqa: ANN401
|
||||
def _dispatch(task_fn: Any, pk: int) -> None: # ruff:ignore[any-type]
|
||||
"""Dispatch a Celery task, logging rather than raising when the broker is unavailable."""
|
||||
try:
|
||||
task_fn.delay(pk)
|
||||
except Exception: # noqa: BLE001
|
||||
except Exception: # ruff:ignore[blind-except]
|
||||
logger.debug(
|
||||
"Could not dispatch %s(%d) — broker may be unavailable.",
|
||||
task_fn.name,
|
||||
|
|
@ -20,49 +20,57 @@ def _dispatch(task_fn: Any, pk: int) -> None: # noqa: ANN401
|
|||
)
|
||||
|
||||
|
||||
def on_game_saved(sender: Any, instance: Any, created: bool, **kwargs: Any) -> None: # noqa: ANN401, FBT001
|
||||
def on_game_saved(sender: Any, instance: Any, created: bool, **kwargs: Any) -> None: # ruff:ignore[any-type, boolean-type-hint-positional-argument]
|
||||
"""Dispatch a box-art download task when a new Game is created."""
|
||||
if created:
|
||||
from twitch.tasks import download_game_image # noqa: PLC0415
|
||||
from twitch.tasks import ( # ruff:ignore[import-outside-top-level]
|
||||
download_game_image,
|
||||
)
|
||||
|
||||
_dispatch(download_game_image, instance.pk)
|
||||
|
||||
|
||||
def on_drop_campaign_saved(
|
||||
sender: Any, # noqa: ANN401
|
||||
instance: Any, # noqa: ANN401
|
||||
created: bool, # noqa: FBT001
|
||||
**kwargs: Any, # noqa: ANN401
|
||||
sender: Any, # ruff:ignore[any-type]
|
||||
instance: Any, # ruff:ignore[any-type]
|
||||
created: bool, # ruff:ignore[boolean-type-hint-positional-argument]
|
||||
**kwargs: Any, # ruff:ignore[any-type]
|
||||
) -> None:
|
||||
"""Dispatch an image download task when a new DropCampaign is created."""
|
||||
if created:
|
||||
from twitch.tasks import download_campaign_image # noqa: PLC0415
|
||||
from twitch.tasks import ( # ruff:ignore[import-outside-top-level]
|
||||
download_campaign_image,
|
||||
)
|
||||
|
||||
_dispatch(download_campaign_image, instance.pk)
|
||||
|
||||
|
||||
def on_drop_benefit_saved(
|
||||
sender: Any, # noqa: ANN401
|
||||
instance: Any, # noqa: ANN401
|
||||
created: bool, # noqa: FBT001
|
||||
**kwargs: Any, # noqa: ANN401
|
||||
sender: Any, # ruff:ignore[any-type]
|
||||
instance: Any, # ruff:ignore[any-type]
|
||||
created: bool, # ruff:ignore[boolean-type-hint-positional-argument]
|
||||
**kwargs: Any, # ruff:ignore[any-type]
|
||||
) -> None:
|
||||
"""Dispatch an image download task when a new DropBenefit is created."""
|
||||
if created:
|
||||
from twitch.tasks import download_benefit_image # noqa: PLC0415
|
||||
from twitch.tasks import ( # ruff:ignore[import-outside-top-level]
|
||||
download_benefit_image,
|
||||
)
|
||||
|
||||
_dispatch(download_benefit_image, instance.pk)
|
||||
|
||||
|
||||
def on_reward_campaign_saved(
|
||||
sender: Any, # noqa: ANN401
|
||||
instance: Any, # noqa: ANN401
|
||||
created: bool, # noqa: FBT001
|
||||
**kwargs: Any, # noqa: ANN401
|
||||
sender: Any, # ruff:ignore[any-type]
|
||||
instance: Any, # ruff:ignore[any-type]
|
||||
created: bool, # ruff:ignore[boolean-type-hint-positional-argument]
|
||||
**kwargs: Any, # ruff:ignore[any-type]
|
||||
) -> None:
|
||||
"""Dispatch an image download task when a new RewardCampaign is created."""
|
||||
if created:
|
||||
from twitch.tasks import download_reward_campaign_image # noqa: PLC0415
|
||||
from twitch.tasks import ( # ruff:ignore[import-outside-top-level]
|
||||
download_reward_campaign_image,
|
||||
)
|
||||
|
||||
_dispatch(download_reward_campaign_image, instance.pk)
|
||||
|
||||
|
|
@ -72,8 +80,8 @@ def _refresh_allowed_campaign_counts(channel_ids: set[int]) -> None:
|
|||
if not channel_ids:
|
||||
return
|
||||
|
||||
from twitch.models import Channel # noqa: PLC0415
|
||||
from twitch.models import DropCampaign # noqa: PLC0415
|
||||
from twitch.models import Channel # ruff:ignore[import-outside-top-level]
|
||||
from twitch.models import DropCampaign # ruff:ignore[import-outside-top-level]
|
||||
|
||||
through_model: type[Channel] = DropCampaign.allow_channels.through
|
||||
counts_by_channel: dict[int, int] = {
|
||||
|
|
@ -96,19 +104,19 @@ def _refresh_allowed_campaign_counts(channel_ids: set[int]) -> None:
|
|||
Channel.objects.bulk_update(channels, ["allowed_campaign_count"])
|
||||
|
||||
|
||||
def on_drop_campaign_allow_channels_changed( # noqa: PLR0913, PLR0917
|
||||
sender: Any, # noqa: ANN401
|
||||
instance: Any, # noqa: ANN401
|
||||
def on_drop_campaign_allow_channels_changed( # ruff:ignore[too-many-arguments, too-many-positional-arguments]
|
||||
sender: Any, # ruff:ignore[any-type]
|
||||
instance: Any, # ruff:ignore[any-type]
|
||||
action: str,
|
||||
reverse: bool, # noqa: FBT001
|
||||
model: Any, # noqa: ANN401
|
||||
reverse: bool, # ruff:ignore[boolean-type-hint-positional-argument]
|
||||
model: Any, # ruff:ignore[any-type]
|
||||
pk_set: set[int] | None,
|
||||
**kwargs: Any, # noqa: ANN401
|
||||
**kwargs: Any, # ruff:ignore[any-type]
|
||||
) -> None:
|
||||
"""Keep Channel.allowed_campaign_count in sync for allow_channels M2M changes."""
|
||||
if action == "pre_clear" and not reverse:
|
||||
# post_clear does not expose removed channel IDs; snapshot before clearing.
|
||||
instance._pre_clear_channel_ids = set( # pyright: ignore[reportAttributeAccessIssue] # noqa: SLF001
|
||||
instance._pre_clear_channel_ids = set( # pyright: ignore[reportAttributeAccessIssue] # ruff:ignore[private-member-access]
|
||||
instance.allow_channels.values_list("pk", flat=True), # pyright: ignore[reportAttributeAccessIssue]
|
||||
)
|
||||
return
|
||||
|
|
|
|||
|
|
@ -21,7 +21,7 @@ logger: logging.Logger = logging.getLogger("ttvdrops.tasks")
|
|||
|
||||
|
||||
@shared_task(bind=True, queue="imports", max_retries=3, default_retry_delay=60)
|
||||
def scan_pending_twitch_files(self) -> None: # noqa: ANN001
|
||||
def scan_pending_twitch_files(self) -> None: # ruff:ignore[missing-type-function-argument]
|
||||
"""Scan TTVDROPS_PENDING_DIR for JSON files and dispatch an import task for each."""
|
||||
pending_dir: str = os.getenv("TTVDROPS_PENDING_DIR", "")
|
||||
if not pending_dir:
|
||||
|
|
@ -44,9 +44,9 @@ def scan_pending_twitch_files(self) -> None: # noqa: ANN001
|
|||
|
||||
|
||||
@shared_task(bind=True, queue="imports", max_retries=3, default_retry_delay=60)
|
||||
def import_twitch_file(self, file_path: str) -> None: # noqa: ANN001
|
||||
def import_twitch_file(self, file_path: str) -> None: # ruff:ignore[missing-type-function-argument]
|
||||
"""Import a single Twitch JSON drop file via BetterImportDrops logic."""
|
||||
from twitch.management.commands.better_import_drops import Command as Importer # noqa: I001, PLC0415
|
||||
from twitch.management.commands.better_import_drops import Command as Importer # ruff:ignore[unsorted-imports, import-outside-top-level]
|
||||
|
||||
path = Path(file_path)
|
||||
if not path.is_file():
|
||||
|
|
@ -103,7 +103,7 @@ def _convert_to_modern_formats(source: Path) -> None:
|
|||
|
||||
try:
|
||||
img = _open_and_prepare_image(source)
|
||||
except Exception: # noqa: BLE001
|
||||
except Exception: # ruff:ignore[blind-except]
|
||||
logger.debug("Format conversion failed for %s.", source)
|
||||
return
|
||||
|
||||
|
|
@ -116,7 +116,7 @@ def _open_and_prepare_image(source: Path) -> Image:
|
|||
Returns:
|
||||
An RGB-mode PIL Image ready for saving in modern formats.
|
||||
"""
|
||||
from PIL import Image # noqa: PLC0415
|
||||
from PIL import Image # ruff:ignore[import-outside-top-level]
|
||||
|
||||
with Image.open(source) as raw:
|
||||
if raw.mode in {"RGBA", "LA"} or (
|
||||
|
|
@ -140,7 +140,7 @@ def _save_modern_formats(img: Image, source: Path) -> None:
|
|||
out: Path = source.with_suffix(ext)
|
||||
try:
|
||||
img.save(out, fmt, quality=80)
|
||||
except Exception: # noqa: BLE001
|
||||
except Exception: # ruff:ignore[blind-except]
|
||||
logger.debug("Could not convert %s to %s.", source, fmt)
|
||||
|
||||
|
||||
|
|
@ -150,11 +150,11 @@ def _save_modern_formats(img: Image, source: Path) -> None:
|
|||
|
||||
|
||||
@shared_task(bind=True, queue="image-downloads", max_retries=3, default_retry_delay=300)
|
||||
def download_game_image(self, game_pk: int) -> None: # noqa: ANN001
|
||||
def download_game_image(self, game_pk: int) -> None: # ruff:ignore[missing-type-function-argument]
|
||||
"""Download and cache the box art image for a single Game."""
|
||||
from twitch.models import Game # noqa: PLC0415
|
||||
from twitch.utils import is_twitch_box_art_url # noqa: PLC0415
|
||||
from twitch.utils import normalize_twitch_box_art_url # noqa: PLC0415
|
||||
from twitch.models import Game # ruff:ignore[import-outside-top-level, unsorted-imports]
|
||||
from twitch.utils import is_twitch_box_art_url # ruff:ignore[import-outside-top-level]
|
||||
from twitch.utils import normalize_twitch_box_art_url # ruff:ignore[import-outside-top-level]
|
||||
|
||||
try:
|
||||
game = Game.objects.get(pk=game_pk)
|
||||
|
|
@ -172,9 +172,9 @@ def download_game_image(self, game_pk: int) -> None: # noqa: ANN001
|
|||
|
||||
|
||||
@shared_task(bind=True, queue="image-downloads", max_retries=3, default_retry_delay=300)
|
||||
def download_campaign_image(self, campaign_pk: int) -> None: # noqa: ANN001
|
||||
def download_campaign_image(self, campaign_pk: int) -> None: # ruff:ignore[missing-type-function-argument]
|
||||
"""Download and cache the image for a single DropCampaign."""
|
||||
from twitch.models import DropCampaign # noqa: PLC0415
|
||||
from twitch.models import DropCampaign # ruff:ignore[import-outside-top-level]
|
||||
|
||||
try:
|
||||
campaign = DropCampaign.objects.get(pk=campaign_pk)
|
||||
|
|
@ -191,9 +191,9 @@ def download_campaign_image(self, campaign_pk: int) -> None: # noqa: ANN001
|
|||
|
||||
|
||||
@shared_task(bind=True, queue="image-downloads", max_retries=3, default_retry_delay=300)
|
||||
def download_benefit_image(self, benefit_pk: int) -> None: # noqa: ANN001
|
||||
def download_benefit_image(self, benefit_pk: int) -> None: # ruff:ignore[missing-type-function-argument]
|
||||
"""Download and cache the image for a single DropBenefit."""
|
||||
from twitch.models import DropBenefit # noqa: PLC0415
|
||||
from twitch.models import DropBenefit # ruff:ignore[import-outside-top-level]
|
||||
|
||||
try:
|
||||
benefit = DropBenefit.objects.get(pk=benefit_pk)
|
||||
|
|
@ -214,9 +214,9 @@ def download_benefit_image(self, benefit_pk: int) -> None: # noqa: ANN001
|
|||
|
||||
|
||||
@shared_task(bind=True, queue="image-downloads", max_retries=3, default_retry_delay=300)
|
||||
def download_reward_campaign_image(self, reward_pk: int) -> None: # noqa: ANN001
|
||||
def download_reward_campaign_image(self, reward_pk: int) -> None: # ruff:ignore[missing-type-function-argument]
|
||||
"""Download and cache the image for a single RewardCampaign."""
|
||||
from twitch.models import RewardCampaign # noqa: PLC0415
|
||||
from twitch.models import RewardCampaign # ruff:ignore[import-outside-top-level]
|
||||
|
||||
try:
|
||||
reward = RewardCampaign.objects.get(pk=reward_pk)
|
||||
|
|
@ -245,7 +245,7 @@ def download_all_images() -> None:
|
|||
|
||||
|
||||
@shared_task(bind=True, queue="api-fetches", max_retries=3, default_retry_delay=120)
|
||||
def import_chat_badges(self) -> None: # noqa: ANN001
|
||||
def import_chat_badges(self) -> None: # ruff:ignore[missing-type-function-argument]
|
||||
"""Fetch and upsert Twitch global chat badges via the Helix API."""
|
||||
try:
|
||||
call_command("import_chat_badges")
|
||||
|
|
|
|||
|
|
@ -46,7 +46,7 @@ def get_format_url(image_url: str, fmt: str) -> str:
|
|||
|
||||
|
||||
@register.simple_tag
|
||||
def picture( # noqa: PLR0913, PLR0917
|
||||
def picture( # ruff:ignore[too-many-arguments, too-many-positional-arguments]
|
||||
src: str,
|
||||
alt: str = "",
|
||||
width: int | None = None,
|
||||
|
|
|
|||
|
|
@ -1080,7 +1080,7 @@ def test_campaign_feed_queries_bounded(
|
|||
_build_campaign(game, i)
|
||||
|
||||
url: str = reverse("core:campaign_feed")
|
||||
# TODO(TheLovinator): 14 queries is still quite high for a feed - we should be able to optimize this further, but this is a good starting point to prevent regressions for now. # noqa: TD003
|
||||
# TODO(TheLovinator): 14 queries is still quite high for a feed - we should be able to optimize this further, but this is a good starting point to prevent regressions for now. # ruff:ignore[missing-todo-link]
|
||||
with django_assert_num_queries(14, exact=False):
|
||||
response: _MonkeyPatchedWSGIResponse = client.get(url)
|
||||
|
||||
|
|
@ -1331,7 +1331,7 @@ def test_docs_rss_queries_bounded(
|
|||
|
||||
url: str = reverse("core:docs_rss")
|
||||
|
||||
# TODO(TheLovinator): 31 queries is still quite high for a feed - we should be able to optimize this further, but this is a good starting point to prevent regressions for now. # noqa: TD003
|
||||
# TODO(TheLovinator): 31 queries is still quite high for a feed - we should be able to optimize this further, but this is a good starting point to prevent regressions for now. # ruff:ignore[missing-todo-link]
|
||||
with django_assert_num_queries(31, exact=False):
|
||||
response: _MonkeyPatchedWSGIResponse = client.get(url)
|
||||
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@ if TYPE_CHECKING:
|
|||
|
||||
|
||||
@pytest.mark.django_db(transaction=True)
|
||||
def test_0021_backfills_allowed_campaign_count() -> None: # noqa: PLR0914
|
||||
def test_0021_backfills_allowed_campaign_count() -> None: # ruff:ignore[too-many-locals]
|
||||
"""Migration 0021 should backfill cached allowed campaign counts."""
|
||||
migrate_from: list[tuple[str, str]] = [
|
||||
("twitch", "0020_rewardcampaign_tw_reward_ends_starts_idx"),
|
||||
|
|
|
|||
|
|
@ -2555,7 +2555,7 @@ class TestRewardCampaignViews:
|
|||
game.owners.add(org)
|
||||
return game
|
||||
|
||||
def _create_reward_campaign( # noqa: PLR0913
|
||||
def _create_reward_campaign( # ruff:ignore[too-many-arguments]
|
||||
self,
|
||||
twitch_id: str,
|
||||
*,
|
||||
|
|
|
|||
|
|
@ -196,7 +196,7 @@ def _build_breadcrumb_schema(items: list[dict[str, str | int]]) -> dict[str, Any
|
|||
Returns:
|
||||
BreadcrumbList schema dict.
|
||||
"""
|
||||
# TODO(TheLovinator): Replace dict with something more structured, like a dataclass or namedtuple, for better type safety and readability. # noqa: TD003
|
||||
# TODO(TheLovinator): Replace dict with something more structured, like a dataclass or namedtuple, for better type safety and readability. # ruff:ignore[missing-todo-link]
|
||||
|
||||
breadcrumb_items: list[dict[str, str | int]] = []
|
||||
for position, item in enumerate(items, start=1):
|
||||
|
|
@ -384,7 +384,7 @@ def organization_detail_view(request: HttpRequest, twitch_id: str) -> HttpRespon
|
|||
|
||||
|
||||
# MARK: /campaigns/
|
||||
def drop_campaign_list_view(request: HttpRequest) -> HttpResponse: # noqa: PLR0914
|
||||
def drop_campaign_list_view(request: HttpRequest) -> HttpResponse: # ruff:ignore[too-many-locals]
|
||||
"""Function-based view for drop campaigns list.
|
||||
|
||||
Args:
|
||||
|
|
@ -483,7 +483,7 @@ def drop_campaign_list_view(request: HttpRequest) -> HttpResponse: # noqa: PLR0
|
|||
|
||||
|
||||
# MARK: /campaigns/<twitch_id>/
|
||||
def drop_campaign_detail_view(request: HttpRequest, twitch_id: str) -> HttpResponse: # noqa: PLR0914
|
||||
def drop_campaign_detail_view(request: HttpRequest, twitch_id: str) -> HttpResponse: # ruff:ignore[too-many-locals]
|
||||
"""Function-based view for a drop campaign detail.
|
||||
|
||||
Args:
|
||||
|
|
@ -536,7 +536,7 @@ def drop_campaign_detail_view(request: HttpRequest, twitch_id: str) -> HttpRespo
|
|||
reverse("twitch:campaign_detail", args=[campaign.twitch_id]),
|
||||
)
|
||||
|
||||
# TODO(TheLovinator): If the campaign has specific allowed channels, we could list those as potential locations instead of just linking to Twitch homepage. # noqa: TD003
|
||||
# TODO(TheLovinator): If the campaign has specific allowed channels, we could list those as potential locations instead of just linking to Twitch homepage. # ruff:ignore[missing-todo-link]
|
||||
campaign_event: dict[str, Any] = {
|
||||
"@context": "https://schema.org",
|
||||
"@type": "Event",
|
||||
|
|
@ -590,7 +590,7 @@ def drop_campaign_detail_view(request: HttpRequest, twitch_id: str) -> HttpRespo
|
|||
campaign_schema: dict[str, Any] = campaign_event
|
||||
|
||||
# Breadcrumb schema for navigation
|
||||
# TODO(TheLovinator): We should have a game.get_display_name() method that encapsulates the logic of choosing between display_name, name, and twitch_id. # noqa: TD003
|
||||
# TODO(TheLovinator): We should have a game.get_display_name() method that encapsulates the logic of choosing between display_name, name, and twitch_id. # ruff:ignore[missing-todo-link]
|
||||
game_name: str = (
|
||||
campaign.game.display_name or campaign.game.name or campaign.game.twitch_id
|
||||
)
|
||||
|
|
@ -702,7 +702,7 @@ class GameDetailView(DetailView):
|
|||
"""Return game queryset optimized for the game detail page."""
|
||||
return Game.for_detail_view()
|
||||
|
||||
def get_context_data(self, **kwargs) -> dict[str, Any]: # noqa: PLR0914
|
||||
def get_context_data(self, **kwargs) -> dict[str, Any]: # ruff:ignore[too-many-locals]
|
||||
"""Add additional context data.
|
||||
|
||||
Args:
|
||||
|
|
@ -849,7 +849,7 @@ def dashboard(request: HttpRequest) -> HttpResponse:
|
|||
dashboard_data: dict[str, Any] = DropCampaign.dashboard_context(now)
|
||||
|
||||
# WebSite schema with SearchAction for sitelinks search box
|
||||
# TODO(TheLovinator): Should this be on all pages instead of just the dashboard? # noqa: TD003
|
||||
# TODO(TheLovinator): Should this be on all pages instead of just the dashboard? # ruff:ignore[missing-todo-link]
|
||||
website_schema: dict[str, str | dict[str, str | dict[str, str]]] = {
|
||||
"@context": "https://schema.org",
|
||||
"@type": "WebSite",
|
||||
|
|
@ -1184,7 +1184,7 @@ class ChannelDetailView(DetailView):
|
|||
channel: Channel = get_object_or_404(queryset, twitch_id=twitch_id)
|
||||
return channel
|
||||
|
||||
def get_context_data(self, **kwargs) -> dict[str, Any]: # noqa: PLR0914
|
||||
def get_context_data(self, **kwargs) -> dict[str, Any]: # ruff:ignore[too-many-locals]
|
||||
"""Add additional context data.
|
||||
|
||||
Args:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue