128 lines
5 KiB
Python
128 lines
5 KiB
Python
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}"),
|
|
)
|