Compare commits

...

8 commits

Author SHA1 Message Date
596769874f
Remove references to ChzzkApiResponseV1 in schema tests and update test cases to use ChzzkApiResponseV2
All checks were successful
Deploy to Server / deploy (pull_request) Successful in 13s
Deploy to Server / deploy (push) Successful in 21s
2026-04-01 04:05:59 +02:00
9ce324fd2d
Implement Chzzk campaign management features, including models, views, and templates
Some checks failed
Deploy to Server / deploy (push) Failing after 9s
2026-04-01 04:04:58 +02:00
677aedf42b
Add Chzzk campaign and reward models, import command, and schemas
Some checks failed
Deploy to Server / deploy (push) Failing after 19s
2026-03-31 21:57:12 +02:00
c852134338
Add .gitattributes to mark JSON files as linguist-generated 2026-03-31 21:56:39 +02:00
6855f2b6f6
Update actionlint version to v1.7.12 in pre-commit configuration 2026-03-31 21:56:12 +02:00
f6688ca904
Fix conditional syntax for server health check in deploy workflow
All checks were successful
Deploy to Server / deploy (push) Successful in 13s
2026-03-30 17:01:13 +02:00
f605ea9f55
Add initial chzzk app structure and configuration
Some checks failed
Deploy to Server / deploy (push) Has been cancelled
2026-03-30 17:00:10 +02:00
f1ee0528e6
Update ruff-pre-commit version to v0.15.8 2026-03-30 16:59:04 +02:00
28 changed files with 1090 additions and 11 deletions

2
.gitattributes vendored Normal file
View file

@ -0,0 +1,2 @@
chzzk/tests/*.json linguist-generated=true
twitch/tests/*.json linguist-generated=true

View file

@ -40,6 +40,7 @@ jobs:
sudo -u ttvdrops uv run python /home/ttvdrops/ttvdrops/manage.py collectstatic --noinput
sudo systemctl restart ttvdrops
- name: Check if server is up
if : ${{ success() && github.ref == 'refs/heads/master' }}
run: |
sleep 2
curl -f https://ttvdrops.lovinator.space/ || exit 1

View file

@ -27,7 +27,7 @@ repos:
args: [--target-version, "6.0"]
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.15.5
rev: v0.15.8
hooks:
- id: ruff-check
args: ["--fix", "--exit-non-zero-on-fix"]
@ -40,6 +40,6 @@ repos:
args: ["--py311-plus"]
- repo: https://github.com/rhysd/actionlint
rev: v1.7.11
rev: v1.7.12
hooks:
- id: actionlint

View file

@ -10,6 +10,7 @@
"buildx",
"chatbadge",
"chatbadgeset",
"chzzk",
"collectstatic",
"colorama",
"createsuperuser",

0
chzzk/__init__.py Normal file
View file

7
chzzk/apps.py Normal file
View file

@ -0,0 +1,7 @@
from django.apps import AppConfig
class ChzzkConfig(AppConfig):
"""Config for chzzk app."""
name = "chzzk"

View file

View file

View file

@ -0,0 +1,128 @@
from typing import TYPE_CHECKING
from typing import Any
if TYPE_CHECKING:
import argparse
from chzzk.schemas import ChzzkCampaignV2
from typing import TYPE_CHECKING
import requests
from django.core.management.base import BaseCommand
from django.utils import timezone
from chzzk.models import ChzzkCampaign
from chzzk.models import ChzzkReward
from chzzk.schemas import ChzzkApiResponseV2
if TYPE_CHECKING:
import argparse
CHZZK_API_URLS: list[tuple[str, str]] = [
("v1", "https://api.chzzk.naver.com/service/v1/drops/campaigns/{campaign_no}"),
("v2", "https://api.chzzk.naver.com/service/v2/drops/campaigns/{campaign_no}"),
]
USER_AGENT = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:151.0) Gecko/20100101 Firefox/151.0"
)
class Command(BaseCommand):
"""Django management command to scrape Chzzk drops campaigns from both v1 and v2 APIs and store them in the database."""
help = "Scrape Chzzk drops campaigns from both v1 and v2 APIs and store them."
def add_arguments(self, parser: argparse.ArgumentParser) -> None:
"""Add command-line arguments for the management command."""
parser.add_argument("campaign_no", type=int, help="Campaign number to fetch")
def handle(self, **options) -> None:
"""Main handler for the management command. Fetches campaign data from both API versions, validates, and stores them."""
campaign_no: int = int(options["campaign_no"])
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,
timeout=2,
headers={
"Accept": "application/json",
"User-Agent": USER_AGENT,
},
)
resp.raise_for_status()
data: dict[str, Any] = resp.json()
campaign_data: ChzzkCampaignV2
campaign_data = ChzzkApiResponseV2.model_validate(data).content
# Prepare raw JSON defaults for both API versions so DB inserts won't fail
raw_json_v1_val = data if api_version == "v1" else {}
raw_json_v2_val = data if api_version == "v2" else {}
# Save campaign
campaign_obj, created = ChzzkCampaign.objects.update_or_create(
campaign_no=campaign_data.campaign_no,
defaults={
"title": campaign_data.title,
"image_url": campaign_data.image_url,
"description": campaign_data.description,
"category_type": campaign_data.category_type,
"category_id": campaign_data.category_id,
"category_value": campaign_data.category_value,
"pc_link_url": campaign_data.pc_link_url,
"mobile_link_url": campaign_data.mobile_link_url,
"service_id": campaign_data.service_id,
"state": campaign_data.state,
"start_date": campaign_data.start_date,
"end_date": campaign_data.end_date,
"has_ios_based_reward": campaign_data.has_ios_based_reward,
"drops_campaign_not_started": campaign_data.drops_campaign_not_started,
"campaign_reward_type": getattr(
campaign_data,
"campaign_reward_type",
"",
),
"reward_type": getattr(campaign_data, "reward_type", ""),
"account_link_url": campaign_data.account_link_url,
"scraped_at": timezone.now(),
"scrape_status": "success",
"raw_json_v1": raw_json_v1_val,
"raw_json_v2": raw_json_v2_val,
},
)
if created:
self.stdout.write(
self.style.SUCCESS(f"Created campaign {campaign_no}"),
)
for reward in campaign_data.reward_list:
reward_, created = ChzzkReward.objects.update_or_create(
campaign=campaign_obj,
reward_no=reward.reward_no,
defaults={
"image_url": reward.image_url,
"title": reward.title,
"reward_type": reward.reward_type,
"campaign_reward_type": getattr(
reward,
"campaign_reward_type",
"",
),
"condition_type": reward.condition_type,
"condition_for_minutes": reward.condition_for_minutes,
"ios_based_reward": reward.ios_based_reward,
"code_remaining_count": reward.code_remaining_count,
},
)
if created:
self.stdout.write(
self.style.SUCCESS(
f" Created reward {reward_.reward_no} for campaign {campaign_no}",
),
)
self.stdout.write(
self.style.SUCCESS(f"Imported campaign {campaign_no}"),
)

View file

@ -0,0 +1,100 @@
# Generated by Django 6.0.3 on 2026-04-01 01:57
import django.db.models.deletion
import django.utils.timezone
from django.db import migrations
from django.db import models
class Migration(migrations.Migration):
"""Initial migration for chzzk app, creating ChzzkCampaign and ChzzkReward models."""
initial = True
dependencies = []
operations = [
migrations.CreateModel(
name="ChzzkCampaign",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
("campaign_no", models.BigIntegerField()),
("title", models.CharField(max_length=255)),
("image_url", models.URLField()),
("description", models.TextField()),
("category_type", models.CharField(max_length=64)),
("category_id", models.CharField(max_length=128)),
("category_value", models.CharField(max_length=128)),
("pc_link_url", models.URLField()),
("mobile_link_url", models.URLField()),
("service_id", models.CharField(max_length=128)),
("state", models.CharField(max_length=64)),
("start_date", models.DateTimeField()),
("end_date", models.DateTimeField()),
("has_ios_based_reward", models.BooleanField()),
("drops_campaign_not_started", models.BooleanField()),
(
"campaign_reward_type",
models.CharField(blank=True, default="", max_length=64),
),
(
"reward_type",
models.CharField(blank=True, default="", max_length=64),
),
("account_link_url", models.URLField()),
("scraped_at", models.DateTimeField(default=django.utils.timezone.now)),
("source_api", models.CharField(max_length=16)),
("scrape_status", models.CharField(default="success", max_length=32)),
("raw_json_v1", models.JSONField(blank=True, null=True)),
("raw_json_v2", models.JSONField(blank=True, null=True)),
],
options={
"ordering": ["-start_date"],
},
),
migrations.CreateModel(
name="ChzzkReward",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
("reward_no", models.BigIntegerField()),
("image_url", models.URLField()),
("title", models.CharField(max_length=255)),
("reward_type", models.CharField(max_length=64)),
(
"campaign_reward_type",
models.CharField(blank=True, default="", max_length=64),
),
("condition_type", models.CharField(max_length=64)),
("condition_for_minutes", models.IntegerField()),
("ios_based_reward", models.BooleanField()),
("code_remaining_count", models.IntegerField()),
(
"campaign",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name="rewards",
to="chzzk.chzzkcampaign",
),
),
],
options={
"unique_together": {("campaign", "reward_no")},
},
),
]

View file

63
chzzk/models.py Normal file
View file

@ -0,0 +1,63 @@
from django.db import models
from django.utils import timezone
class ChzzkCampaign(models.Model):
"""Chzzk campaign, including scraping metadata."""
campaign_no = models.BigIntegerField()
title = models.CharField(max_length=255)
image_url = models.URLField()
description = models.TextField()
category_type = models.CharField(max_length=64)
category_id = models.CharField(max_length=128)
category_value = models.CharField(max_length=128)
pc_link_url = models.URLField()
mobile_link_url = models.URLField()
service_id = models.CharField(max_length=128)
state = models.CharField(max_length=64)
start_date = models.DateTimeField()
end_date = models.DateTimeField()
has_ios_based_reward = models.BooleanField()
drops_campaign_not_started = models.BooleanField()
campaign_reward_type = models.CharField(max_length=64, blank=True, default="")
reward_type = models.CharField(max_length=64, blank=True, default="")
account_link_url = models.URLField()
# Scraping metadata
scraped_at = models.DateTimeField(default=timezone.now)
source_api = models.CharField(max_length=16)
scrape_status = models.CharField(max_length=32, default="success")
raw_json_v1 = models.JSONField(null=True, blank=True)
raw_json_v2 = models.JSONField(null=True, blank=True)
class Meta:
ordering = ["-start_date"]
def __str__(self) -> str:
return f"{self.title} (#{self.campaign_no})"
class ChzzkReward(models.Model):
"""Chzzk reward belonging to a campaign."""
campaign = models.ForeignKey(
ChzzkCampaign,
related_name="rewards",
on_delete=models.CASCADE,
)
reward_no = models.BigIntegerField()
image_url = models.URLField()
title = models.CharField(max_length=255)
reward_type = models.CharField(max_length=64)
campaign_reward_type = models.CharField(max_length=64, blank=True, default="")
condition_type = models.CharField(max_length=64)
condition_for_minutes = models.IntegerField()
ios_based_reward = models.BooleanField()
code_remaining_count = models.IntegerField()
class Meta:
unique_together = ("campaign", "reward_no")
def __str__(self) -> str:
return f"{self.title} (#{self.reward_no})"

54
chzzk/schemas.py Normal file
View file

@ -0,0 +1,54 @@
from typing import Any
from pydantic import BaseModel
from pydantic import Field
class ChzzkRewardV2(BaseModel):
"""Pydantic schema for api v2 reward object."""
title: str
reward_no: int = Field(..., alias="rewardNo")
image_url: str = Field(..., alias="imageUrl")
reward_type: str = Field(..., alias="rewardType")
condition_type: str = Field(..., alias="conditionType")
condition_for_minutes: int = Field(..., alias="conditionForMinutes")
ios_based_reward: bool = Field(..., alias="iosBasedReward")
code_remaining_count: int = Field(..., alias="codeRemainingCount")
model_config = {"extra": "forbid"}
class ChzzkCampaignV2(BaseModel):
"""Pydantic schema for api v2 campaign object."""
title: str
state: str
description: str
campaign_no: int = Field(..., alias="campaignNo")
image_url: str = Field(..., alias="imageUrl")
category_type: str = Field(..., alias="categoryType")
category_id: str = Field(..., alias="categoryId")
category_value: str = Field(..., alias="categoryValue")
pc_link_url: str = Field(..., alias="pcLinkUrl")
mobile_link_url: str = Field(..., alias="mobileLinkUrl")
service_id: str = Field(..., alias="serviceId")
start_date: str = Field(..., alias="startDate")
end_date: str = Field(..., alias="endDate")
reward_list: list[ChzzkRewardV2] = Field(..., alias="rewardList")
has_ios_based_reward: bool = Field(..., alias="hasIosBasedReward")
drops_campaign_not_started: bool = Field(..., alias="dropsCampaignNotStarted")
reward_type: str | None = Field(None, alias="rewardType")
account_link_url: str = Field(..., alias="accountLinkUrl")
model_config = {"extra": "forbid"}
class ChzzkApiResponseV2(BaseModel):
"""Pydantic schema for api v2 API response."""
code: int
message: Any | None
content: ChzzkCampaignV2
model_config = {"extra": "forbid"}

0
chzzk/tests/__init__.py Normal file
View file

View file

@ -0,0 +1,37 @@
import json
from pathlib import Path
import pydantic
import pytest
from chzzk.schemas import ChzzkApiResponseV2
TESTS_DIR = Path(__file__).parent
@pytest.mark.parametrize(
("fname", "data_model"),
[
("v2_905.json", ChzzkApiResponseV2),
],
)
def test_chzzk_schema_strict(fname: str, data_model: type) -> None:
"""Test that the schema strictly validates the given JSON file against the provided Pydantic model."""
with Path(TESTS_DIR / fname).open(encoding="utf-8") as f:
data = json.load(f)
# Should not raise
data_model.model_validate(data)
@pytest.mark.parametrize(
("fname", "data_model"),
[
("v1_905.json", ChzzkApiResponseV2),
],
)
def test_chzzk_schema_cross_fail(fname: str, data_model: type) -> None:
"""Test that the schema fails when validating the wrong JSON file/model combination."""
with Path(TESTS_DIR / fname).open(encoding="utf-8") as f:
data = json.load(f)
with pytest.raises(pydantic.ValidationError):
data_model.model_validate(data)

58
chzzk/tests/v1_905.json generated Normal file
View file

@ -0,0 +1,58 @@
{
"code": 200,
"message": null,
"content": {
"campaignNo": 905,
"title": "붉은사막 드롭스 #2",
"imageUrl": "https://nng-phinf.pstatic.net/MjAyNjAzMjRfMzcg/MDAxNzc0MzU4Mzk3MjM0.JQAXkEYe2ntJ5gvzr5U5egn78DalY24mi1hodGyPYcog.tko3iBzm7dOqDKdjZAZX2ozWrj-tKjVcN3v3ieQaNzQg.JPEG/KR_260319_sony_masterimage_2160x2160.jpg",
"description": "치지직에서 붉은사막 방송을 시청하고, 특별한 보상을 획득해 보세요.",
"categoryType": "GAME",
"categoryId": "CrimsonDesert",
"categoryValue": "붉은사막",
"pcLinkUrl": "https://event.pearlabyss.com/CrimsonDesert/Drops",
"mobileLinkUrl": "https://event.pearlabyss.com/CrimsonDesert/Drops",
"serviceId": "Pearl_Abyss_Event",
"state": "PROMOTED",
"startDate": "2026-03-26 09:00:00",
"endDate": "2026-04-02 08:59:00",
"rewardList": [
{
"rewardNo": 2521,
"imageUrl": "https://nng-phinf.pstatic.net/MjAyNjAzMjRfMTIg/MDAxNzc0MzU4NTgxMTY5.WFRtLuWw0qciEmMlnm79culfGW1lj-KwvXTaZLkbzMIg.PxoTwZ6v_i1N7Vkyj--OxmeuEJsRvQDWgjT1sOB6kfog.PNG/Week-2-60-min.png",
"title": "푸른 정찰대 등자",
"rewardType": "TIME_BASED",
"campaignRewardType": "IN_APP",
"conditionType": "TIME_BASED",
"conditionForMinutes": 60,
"iosBasedReward": false,
"codeRemainingCount": 0
},
{
"rewardNo": 2522,
"imageUrl": "https://nng-phinf.pstatic.net/MjAyNjAzMjRfMTYw/MDAxNzc0MzU4NjEwODkz.6SMLoYofJ6A2a9Nd1IlKn_CC3JLyR8ecxkt8cO-rzcAg.kQSP_H08VoS1qXGiB_3BBMYT_0IT07YPz09n0lNmD1cg.PNG/Week-2-120-min.png",
"title": "푸른 정찰대 마면",
"rewardType": "TIME_BASED",
"campaignRewardType": "IN_APP",
"conditionType": "TIME_BASED",
"conditionForMinutes": 120,
"iosBasedReward": false,
"codeRemainingCount": 0
},
{
"rewardNo": 2523,
"imageUrl": "https://nng-phinf.pstatic.net/MjAyNjAzMjRfMjU3/MDAxNzc0MzU4NjM5MDgw.7u6E8ybLp7x9MNESI3od1Sf3AE_6AobSMFfb18MhsNog.M5ZvA1SSjaOObB2TB-P9Rw2rtMcV8zXipL0i7adILdwg.PNG/Week-2-180-min.png",
"title": "푸른 정찰대 안장",
"rewardType": "TIME_BASED",
"campaignRewardType": "IN_APP",
"conditionType": "TIME_BASED",
"conditionForMinutes": 180,
"iosBasedReward": false,
"codeRemainingCount": 0
}
],
"hasIosBasedReward": false,
"dropsCampaignNotStarted": false,
"campaignRewardType": "IN_APP",
"accountLinkUrl": "https://event.pearlabyss.com/CrimsonDesert/Drops"
}
}

55
chzzk/tests/v2_905.json generated Normal file
View file

@ -0,0 +1,55 @@
{
"code": 200,
"message": null,
"content": {
"campaignNo": 905,
"title": "붉은사막 드롭스 #2",
"imageUrl": "https://nng-phinf.pstatic.net/MjAyNjAzMjRfMzcg/MDAxNzc0MzU4Mzk3MjM0.JQAXkEYe2ntJ5gvzr5U5egn78DalY24mi1hodGyPYcog.tko3iBzm7dOqDKdjZAZX2ozWrj-tKjVcN3v3ieQaNzQg.JPEG/KR_260319_sony_masterimage_2160x2160.jpg",
"description": "치지직에서 붉은사막 방송을 시청하고, 특별한 보상을 획득해 보세요.",
"categoryType": "GAME",
"categoryId": "CrimsonDesert",
"categoryValue": "붉은사막",
"pcLinkUrl": "https://event.pearlabyss.com/CrimsonDesert/Drops",
"mobileLinkUrl": "https://event.pearlabyss.com/CrimsonDesert/Drops",
"serviceId": "Pearl_Abyss_Event",
"state": "PROMOTED",
"startDate": "2026-03-26 09:00:00",
"endDate": "2026-04-02 08:59:00",
"rewardList": [
{
"rewardNo": 2521,
"imageUrl": "https://nng-phinf.pstatic.net/MjAyNjAzMjRfMTIg/MDAxNzc0MzU4NTgxMTY5.WFRtLuWw0qciEmMlnm79culfGW1lj-KwvXTaZLkbzMIg.PxoTwZ6v_i1N7Vkyj--OxmeuEJsRvQDWgjT1sOB6kfog.PNG/Week-2-60-min.png",
"title": "푸른 정찰대 등자",
"rewardType": "IN_APP",
"conditionType": "TIME_BASED",
"conditionForMinutes": 60,
"iosBasedReward": false,
"codeRemainingCount": 0
},
{
"rewardNo": 2522,
"imageUrl": "https://nng-phinf.pstatic.net/MjAyNjAzMjRfMTYw/MDAxNzc0MzU4NjEwODkz.6SMLoYofJ6A2a9Nd1IlKn_CC3JLyR8ecxkt8cO-rzcAg.kQSP_H08VoS1qXGiB_3BBMYT_0IT07YPz09n0lNmD1cg.PNG/Week-2-120-min.png",
"title": "푸른 정찰대 마면",
"rewardType": "IN_APP",
"conditionType": "TIME_BASED",
"conditionForMinutes": 120,
"iosBasedReward": false,
"codeRemainingCount": 0
},
{
"rewardNo": 2523,
"imageUrl": "https://nng-phinf.pstatic.net/MjAyNjAzMjRfMjU3/MDAxNzc0MzU4NjM5MDgw.7u6E8ybLp7x9MNESI3od1Sf3AE_6AobSMFfb18MhsNog.M5ZvA1SSjaOObB2TB-P9Rw2rtMcV8zXipL0i7adILdwg.PNG/Week-2-180-min.png",
"title": "푸른 정찰대 안장",
"rewardType": "IN_APP",
"conditionType": "TIME_BASED",
"conditionForMinutes": 180,
"iosBasedReward": false,
"codeRemainingCount": 0
}
],
"hasIosBasedReward": false,
"dropsCampaignNotStarted": false,
"rewardType": "IN_APP",
"accountLinkUrl": "https://event.pearlabyss.com/CrimsonDesert/Drops"
}
}

49
chzzk/urls.py Normal file
View file

@ -0,0 +1,49 @@
from typing import TYPE_CHECKING
from django.urls import path
from chzzk import views
if TYPE_CHECKING:
from django.urls.resolvers import URLPattern
app_name = "chzzk"
urlpatterns: list[URLPattern] = [
# /chzzk/
path(
"",
views.dashboard_view,
name="dashboard",
),
# /chzzk/campaigns/
path(
"campaigns/",
views.CampaignListView.as_view(),
name="campaign_list",
),
# /chzzk/campaigns/<campaign_no>/
path(
"campaigns/<int:campaign_no>/",
views.campaign_detail_view,
name="campaign_detail",
),
# /chzzk/rss/campaigns
path(
"rss/campaigns",
views.ChzzkCampaignFeed(),
name="campaign_feed",
),
# /chzzk/atom/campaigns
path(
"atom/campaigns",
views.ChzzkCampaignFeed(),
name="campaign_feed_atom",
),
# /chzzk/discord/campaigns
path(
"discord/campaigns",
views.ChzzkCampaignFeed(),
name="campaign_feed_discord",
),
]

206
chzzk/views.py Normal file
View file

@ -0,0 +1,206 @@
from typing import TYPE_CHECKING
from django.db.models.query import QuerySet
from django.shortcuts import render
from django.urls import reverse
from django.utils import timezone
from django.utils.html import format_html
from django.utils.safestring import SafeText
from django.views import generic
from chzzk import models
from twitch.feeds import TTVDropsBaseFeed
if TYPE_CHECKING:
import datetime
from django.http import HttpResponse
from django.http.request import HttpRequest
from pytest_django.asserts import QuerySet
def dashboard_view(request: HttpRequest) -> HttpResponse:
"""View function for the dashboard page showing all the active chzzk campaigns.
Args:
request (HttpRequest): The incoming HTTP request.
Returns:
HttpResponse: The HTTP response containing the rendered dashboard page.
"""
active_campaigns: QuerySet[models.ChzzkCampaign, models.ChzzkCampaign] = (
models.ChzzkCampaign.objects.filter(end_date__gte=timezone.now()).order_by(
"-start_date",
)
)
return render(
request=request,
template_name="chzzk/dashboard.html",
context={
"active_campaigns": active_campaigns,
},
)
class CampaignListView(generic.ListView):
"""List view showing all chzzk campaigns."""
model = models.ChzzkCampaign
template_name = "chzzk/campaign_list.html"
context_object_name = "campaigns"
paginate_by = 25
def campaign_detail_view(request: HttpRequest, campaign_no: int) -> HttpResponse:
"""View function for the campaign detail page showing information about a single chzzk campaign.
Args:
request (HttpRequest): The incoming HTTP request.
campaign_no (int): The campaign number of the campaign to display.
Returns:
HttpResponse: The HTTP response containing the rendered campaign detail page.
"""
campaign: models.ChzzkCampaign = models.ChzzkCampaign.objects.get(
campaign_no=campaign_no,
)
rewards: QuerySet[models.ChzzkReward, models.ChzzkReward] = campaign.rewards.all() # pyright: ignore[reportAttributeAccessIssue]
return render(
request=request,
template_name="chzzk/campaign_detail.html",
context={
"campaign": campaign,
"rewards": rewards,
},
)
class ChzzkCampaignFeed(TTVDropsBaseFeed):
"""RSS feed for the latest chzzk campaigns."""
title: str = "chzzk campaigns"
link: str = "/chzzk/campaigns/"
description: str = "Latest chzzk campaigns"
_limit: int | None = None
def __call__(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
"""Allow an optional 'limit' query parameter to specify the number of items in the feed.
Args:
request (HttpRequest): The incoming HTTP request.
*args: Additional positional arguments.
**kwargs: Additional keyword arguments.
Returns:
HttpResponse: The HTTP response containing the feed.
"""
if request.GET.get("limit"):
try:
self._limit = int(request.GET.get("limit", 50))
except ValueError, TypeError:
self._limit = None
return super().__call__(request, *args, **kwargs)
def items(self) -> QuerySet[models.ChzzkCampaign, models.ChzzkCampaign]:
"""Return the latest chzzk campaigns with raw_json data, ordered by start date.
Returns:
QuerySet: A queryset of ChzzkCampaign objects.
"""
limit: int = self._limit if self._limit is not None else 50
return models.ChzzkCampaign.objects.filter(raw_json__isnull=False).order_by(
"-start_date",
)[:limit]
def item_title(self, item: models.ChzzkCampaign) -> str:
"""Return the title of the feed item, which is the campaign title.
Args:
item (ChzzkCampaign): The campaign object for the feed item.
Returns:
str: The title of the feed item.
"""
return item.title
def item_description(self, item: models.ChzzkCampaign) -> SafeText:
"""Return the description of the feed item, which includes the campaign image and description.
Args:
item (ChzzkCampaign): The campaign object for the feed item.
Returns:
SafeText: The HTML description of the feed item.
"""
parts: list[SafeText] = []
if getattr(item, "image_url", ""):
parts.append(
format_html(
'<img src="{}" alt="{}" width="320" height="180" />',
item.image_url,
item.title,
),
)
if getattr(item, "description", ""):
parts.append(format_html("<p>{}</p>", item.description))
# Link back to the PC detail URL when available
if getattr(item, "pc_link_url", ""):
parts.append(
format_html('<p><a href="{}">Details</a></p>', item.pc_link_url),
)
return SafeText("".join(str(p) for p in parts))
def item_link(self, item: models.ChzzkCampaign) -> str:
"""Return the URL for the feed item, which is the campaign detail page.
Args:
item (ChzzkCampaign): The campaign object for the feed item.
Returns:
str: The URL for the feed item.
"""
return reverse("chzzk:campaign_detail", args=[item.pk])
def item_pubdate(self, item: models.ChzzkCampaign) -> datetime.datetime:
"""Return the publication date of the feed item, which is the campaign start date.
Args:
item (ChzzkCampaign): The campaign object for the feed item.
Returns:
datetime.datetime: The publication date of the feed item.
"""
return getattr(item, "start_date", timezone.now()) or timezone.now()
def item_updateddate(self, item: models.ChzzkCampaign) -> datetime.datetime:
"""Return the last updated date of the feed item, which is the campaign scraped date.
Args:
item (ChzzkCampaign): The campaign object for the feed item.
Returns:
datetime.datetime: The last updated date of the feed item.
"""
return getattr(item, "scraped_at", timezone.now()) or timezone.now()
def item_author_name(self, item: models.ChzzkCampaign) -> str:
"""Return the author name for the feed item. Since we don't have a specific author, return a default value.
Args:
item (ChzzkCampaign): The campaign object for the feed item.
Returns:
str: The author name for the feed item.
"""
return item.category_id or "Unknown Category"
def feed_url(self) -> str:
"""Return the URL of the feed itself.
Returns:
str: The URL of the feed.
"""
return reverse("chzzk:campaign_feed")

View file

@ -140,10 +140,11 @@ INSTALLED_APPS: list[str] = [
"django.contrib.staticfiles",
"django.contrib.postgres",
# Internal apps
"twitch.apps.TwitchConfig",
"kick.apps.KickConfig",
"youtube.apps.YoutubeConfig",
"chzzk.apps.ChzzkConfig",
"core.apps.CoreConfig",
"kick.apps.KickConfig",
"twitch.apps.TwitchConfig",
"youtube.apps.YoutubeConfig",
# Third-party apps
"django_celery_results",
"django_celery_beat",
@ -171,10 +172,35 @@ TEMPLATES: list[dict[str, Any]] = [
},
]
DATABASES: dict[str, dict[str, Any]] = (
{"default": {"ENGINE": "django.db.backends.sqlite3", "NAME": ":memory:"}}
if TESTING
else {
def configure_databases(*, testing: bool, base_dir: Path) -> dict[str, dict[str, Any]]:
"""Configure Django databases based on environment variables and testing mode.
Args:
testing (bool): Whether the application is running in testing mode.
base_dir (Path): The base directory of the project, used for SQLite file location.
Returns:
dict[str, dict[str, Any]]: The DATABASES setting for Django.
"""
use_sqlite: bool = env_bool("USE_SQLITE", default=False)
if testing:
return {
"default": {
"ENGINE": "django.db.backends.sqlite3",
"NAME": ":memory:",
},
}
if use_sqlite:
return {
"default": {
"ENGINE": "django.db.backends.sqlite3",
"NAME": str(base_dir / "db.sqlite3"),
},
}
# Default: PostgreSQL
return {
"default": {
"ENGINE": "django.db.backends.postgresql",
"NAME": os.getenv("POSTGRES_DB", "ttvdrops"),
@ -187,6 +213,11 @@ DATABASES: dict[str, dict[str, Any]] = (
"OPTIONS": {"connect_timeout": env_int("DB_CONNECT_TIMEOUT", 10)},
},
}
DATABASES: dict[str, dict[str, Any]] = configure_databases(
testing=TESTING,
base_dir=BASE_DIR,
)
if not TESTING:

View file

@ -3,6 +3,7 @@ import os
import sys
from contextlib import contextmanager
from typing import TYPE_CHECKING
from typing import Any
import pytest
from django.contrib.sessions.models import Session
@ -249,14 +250,14 @@ def test_email_settings_from_env(
assert reloaded.SERVER_EMAIL == "me@example.com"
def test_database_settings_when_not_testing(
def test_database_settings_when_use_sqlite_enabled(
monkeypatch: pytest.MonkeyPatch,
reload_settings_module: Callable[..., ModuleType],
) -> None:
"""When not running tests, DATABASES should use the Postgres configuration."""
# Ensure the module believes it's not running tests
"""When USE_SQLITE=1 and not testing, DATABASES should use SQLite file."""
monkeypatch.setattr("sys.argv", ["manage.py", "runserver"])
monkeypatch.delenv("PYTEST_VERSION", raising=False)
monkeypatch.setenv("USE_SQLITE", "1")
reloaded: ModuleType = reload_settings_module(
TESTING=None,
@ -271,7 +272,36 @@ def test_database_settings_when_not_testing(
)
assert reloaded.TESTING is False
db_cfg = reloaded.DATABASES["default"]
db_cfg: dict[str, Any] = reloaded.DATABASES["default"]
assert db_cfg["ENGINE"] == "django.db.backends.sqlite3"
# Should use a file, not in-memory
assert db_cfg["NAME"].endswith("db.sqlite3")
def test_database_settings_when_not_testing(
monkeypatch: pytest.MonkeyPatch,
reload_settings_module: Callable[..., ModuleType],
) -> None:
"""When not running tests, DATABASES should use the Postgres configuration."""
# Ensure the module believes it's not running tests and USE_SQLITE is unset
monkeypatch.setattr("sys.argv", ["manage.py", "runserver"])
monkeypatch.delenv("PYTEST_VERSION", raising=False)
monkeypatch.setenv("USE_SQLITE", "0")
reloaded: ModuleType = reload_settings_module(
TESTING=None,
PYTEST_VERSION=None,
POSTGRES_DB="prod_db",
POSTGRES_USER="prod_user",
POSTGRES_PASSWORD="secret",
POSTGRES_HOST="db.host",
POSTGRES_PORT="5433",
CONN_MAX_AGE="120",
CONN_HEALTH_CHECKS="0",
)
assert reloaded.TESTING is False
db_cfg: dict[str, Any] = reloaded.DATABASES["default"]
assert db_cfg["ENGINE"] == "django.db.backends.postgresql"
assert db_cfg["NAME"] == "prod_db"
assert db_cfg["USER"] == "prod_user"

View file

@ -49,6 +49,8 @@ urlpatterns: list[URLPattern | URLResolver] = [
path(route="twitch/", view=include("twitch.urls", namespace="twitch")),
# Kick app
path(route="kick/", view=include("kick.urls", namespace="kick")),
# Chzzk app
path(route="chzzk/", view=include("chzzk.urls", namespace="chzzk")),
# YouTube app
path(route="youtube/", view=include("youtube.urls", namespace="youtube")),
]

View file

@ -228,6 +228,10 @@
<a href="{% url 'kick:game_list' %}">Games</a> |
<a href="{% url 'kick:organization_list' %}">Organizations</a>
<br />
<strong>Chzzk</strong>
<a href="{% url 'chzzk:dashboard' %}">Dashboard</a> |
<a href="{% url 'chzzk:campaign_list' %}">Campaigns</a>
<br />
<strong>Other sites</strong>
<a href="#">Steam</a> |
<a href="{% url 'youtube:index' %}">YouTube</a> |

View file

@ -0,0 +1,52 @@
{% extends "base.html" %}
{% block title %}
{{ campaign.title }}
{% endblock title %}
{% block content %}
<main>
<h1>{{ campaign.title }}</h1>
<nav>
<a href="{% url 'chzzk:dashboard' %}">chzzk</a> &gt; <a href="{% url 'chzzk:campaign_list' %}">Campaigns</a> &gt; {{ campaign.title }}
</nav>
{% if campaign.image_url %}
<img src="{{ campaign.image_url }}"
alt="{{ campaign.title }}"
height="auto"
width="auto"
style="max-width:200px;
height:auto;
border-radius:8px" />
{% endif %}
{% if campaign.description %}
<section class="description">
{{ campaign.description|linebreaksbr }}
</section>
{% endif %}
<section class="times">
{% if campaign.starts_at %}
<div>
Starts: <time datetime="{{ campaign.starts_at|date:'c' }}">{{ campaign.starts_at|date:'M d, Y H:i' }}</time>
</div>
{% endif %}
{% if campaign.ends_at %}
<div>
Ends: <time datetime="{{ campaign.ends_at|date:'c' }}">{{ campaign.ends_at|date:'M d, Y H:i' }}</time>
</div>
{% endif %}
</section>
<hr />
<h2>Rewards</h2>
{% if rewards %}
<ul>
{% for r in rewards %}<li>{{ r.title }} — {{ r.condition_for_minutes }} minutes of watch time</li>{% endfor %}
</ul>
{% else %}
<p>No rewards available.</p>
{% endif %}
{% if campaign.external_url %}
<p>
<a class="btn" href="{{ campaign.external_url }}">View on chzzk</a>
</p>
{% endif %}
</main>
{% endblock content %}

View file

@ -0,0 +1,62 @@
{% extends "base.html" %}
{% block title %}
chzzk Campaigns
{% endblock title %}
{% block extra_head %}
<link rel="alternate"
type="application/rss+xml"
title="All chzzk campaigns (RSS)"
href="{% url 'chzzk:campaign_feed' %}" />
{% endblock extra_head %}
{% block content %}
<main>
<h1>chzzk campaigns</h1>
<nav>
<a href="{% url 'chzzk:dashboard' %}">chzzk</a> > Campaigns
</nav>
<!-- <div class="feeds">
<a href="{% url 'chzzk:campaign_feed' %}" title="RSS feed for all campaigns">[rss]</a>
</div> -->
{% if campaigns %}
<table>
<thead>
<tr>
<th>Name</th>
<th>Organization</th>
<th>Start</th>
<th>End</th>
</tr>
</thead>
<tbody>
{% for c in campaigns %}
<tr>
<td>
<a href="{% url 'chzzk:campaign_detail' c.pk %}">{{ c.title }}</a>
</td>
<td>
{% if c.organization %}
<a href="{% url 'chzzk:organization_detail' c.organization.pk %}">{{ c.organization.name }}</a>
{% endif %}
</td>
<td>
{% if c.start_date %}
<time datetime="{{ c.start_date|date:'c' }}">{{ c.start_date|date:'M d, Y' }}</time> ({{ c.start_date|timesince }} ago)
{% endif %}
</td>
<td>
{% if c.end_date %}
<time datetime="{{ c.end_date|date:'c' }}">{{ c.end_date|date:'M d, Y' }}</time> ({{ c.end_date|timesince }} ago)
{% endif %}
</td>
</tr>
{% endfor %}
</tbody>
</table>
{% if is_paginated %}
{% include "includes/pagination.html" with page_obj=page_obj %}
{% endif %}
{% else %}
<p>No campaigns found.</p>
{% endif %}
</main>
{% endblock content %}

View file

@ -0,0 +1,137 @@
{% extends "base.html" %}
{% block title %}
Chzzk Drops
{% endblock title %}
{% block extra_head %}
<link rel="alternate"
type="application/rss+xml"
title="All Chzzk campaigns (RSS)"
href="{% url 'chzzk:campaign_feed' %}" />
<link rel="alternate"
type="application/atom+xml"
title="All Chzzk campaigns (Atom)"
href="{% url 'chzzk:campaign_feed_atom' %}" />
<link rel="alternate"
type="application/atom+xml"
title="All Chzzk campaigns (Discord)"
href="{% url 'chzzk:campaign_feed_discord' %}" />
{% endblock extra_head %}
{% block content %}
<main>
<h1>Active Chzzk Drops</h1>
<div>CHZZK is South Korean alternative to Twitch.</div>
<!--RSS Feeds
<div>
<a href="{% url 'chzzk:campaign_feed' %}" title="RSS feed for all campaigns">[rss]</a>
<a href="{% url 'chzzk:campaign_feed_atom' %}" title="Atom feed for all campaigns">[atom]</a>
<a href="{% url 'chzzk:campaign_feed_discord' %}" title="Discord feed for all campaigns">[discord]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
</div> -->
<hr />
{% if active_campaigns %}
{% for campaign in active_campaigns %}
<!-- {{ campaign }} -->
<article>
<header>
<h2>{{ campaign.category_value }}</h2>
<div style="font-size: 0.9rem; color: #666;">Status: {{ campaign.state }} | Campaign #{{ campaign.campaign_no }}</div>
</header>
<div style="display: flex; gap: 1rem;">
<div style="flex-shrink: 0;">
{% if campaign.image_url %}
<img src="{{ campaign.image_url }}"
width="200"
height="200"
alt="{{ campaign.title }} image"
style="width: 200px;
height: auto;
border-radius: 8px" />
{% else %}
<div style="width: 200px;
height: 200px;
background-color: #eee;
display: flex;
align-items: center;
justify-content: center;
border-radius: 8px">No Image</div>
{% endif %}
</div>
<div style="flex: 1;">
<h3>
<a href="{% url 'chzzk:campaign_detail' campaign.campaign_no %}">{{ campaign.title }}</a>
</h3>
{% if campaign.description %}
<p style="margin: 0.25rem 0; font-style: italic;">{{ campaign.description }}</p>
{% endif %}
<!-- Start -->
{% if campaign.start_date %}
<p style="margin: 0.25rem 0;">
<strong>Starts:</strong>
<time datetime="{{ campaign.start_date|date:'c' }}"
title="{{ campaign.start_date|date:'DATETIME_FORMAT' }}">
{{ campaign.start_date|date:"M d, Y H:i" }}
</time>
{% if campaign.start_date < now %}
(started {{ campaign.start_date|timesince }} ago)
{% else %}
(in {{ campaign.start_date|timeuntil }})
{% endif %}
</p>
{% endif %}
<!-- End -->
{% if campaign.end_date %}
<p style="margin: 0.25rem 0;">
<strong>Ends:</strong>
<time datetime="{{ campaign.end_date|date:'c' }}"
title="{{ campaign.end_date|date:'DATETIME_FORMAT' }}">
{{ campaign.end_date|date:"M d, Y H:i" }}
</time>
{% if campaign.end_date < now %}
(ended {{ campaign.end_date|timesince }} ago)
{% else %}
(in {{ campaign.end_date|timeuntil }})
{% endif %}
</p>
{% endif %}
{% if campaign.reward_type %}
<p style="margin: 0.25rem 0;">
<strong>Reward Type:</strong> {{ campaign.reward_type }}
</p>
{% endif %}
<p style="margin: 0.25rem 0;">
{% if campaign.account_link_url %}<a href="{{ campaign.account_link_url }}">Connect account</a> |{% endif %}
{% if campaign.pc_link_url %}<a href="{{ campaign.pc_link_url }}" target="_blank">View on Chzzk</a>{% endif %}
</p>
<div style="margin-top: 0.5rem; font-size: 0.9rem;">
<strong>Participating channels:</strong>
<p style="margin: 0.25rem 0 0 0;">{{ campaign.category_value }} is game wide.</p>
</div>
{% if campaign.rewards.all %}
<div style="margin-top: 0.75rem;">
<strong>Rewards:</strong>
<ul style="margin: 0.25rem 0 0 0; padding-left: 1rem;">
{% for reward in campaign.rewards.all %}
<li>
{% if reward.image_url %}
<img src="{{ reward.image_url }}"
alt="{{ reward.title }}"
width="56"
height="56"
style="vertical-align: middle;
border-radius: 4px" />
{% endif %}
{{ reward.title }} ({{ reward.condition_for_minutes }} min)
</li>
{% endfor %}
</ul>
</div>
{% endif %}
</div>
</div>
</article>
{% endfor %}
{% else %}
<p>No active Chzzk drop campaigns at the moment. Check back later!</p>
{% endif %}
</main>
{% endblock content %}

View file

@ -620,7 +620,7 @@ class ExampleJsonImportTests(TestCase):
command = Command()
repo_root: Path = Path(__file__).resolve().parents[2]
example_path: Path = repo_root / "example.json"
example_path: Path = repo_root / "twitch" / "tests" / "example.json"
responses = json.loads(example_path.read_text(encoding="utf-8"))