wutheringwaves/scrape.py

"""Fetch articles from the Wuthering Waves website and saves them locally in JSON format.

It retrieves the article menu and individual articles, prettifies the JSON output, and sets file timestamps based on article creation dates.
"""  # noqa: CPY001

import asyncio
import json

# Set up logging
import logging
import os
import time
from datetime import UTC, datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, LiteralString

import aiofiles
import httpx

if TYPE_CHECKING:
    from collections.abc import Coroutine

logging.basicConfig(
    level=logging.INFO,
    format="%(message)s",
)

logger: logging.Logger = logging.getLogger("wutheringwaves")


async def fetch_json(url: str, client: httpx.AsyncClient) -> dict[Any, Any] | None:
    """Fetch JSON data from a URL.

    Args:
        url (str): The URL to fetch data from.
        client (httpx.AsyncClient): The HTTP client to use for the request.

    Returns:
        dict[Any, Any] | None: The parsed JSON data if successful, None otherwise.

    """
    try:
        response: httpx.Response = await client.get(url)
        response.raise_for_status()
        return response.json()
    except (httpx.RequestError, json.JSONDecodeError):
        logger.exception("Error fetching %s:", url)
        return None


async def save_prettified_json(data: dict[Any, Any], filepath: Path) -> bool:
    """Save JSON data to a file with pretty formatting.

    Args:
        data (dict[Any, Any]): The JSON data to save.
        filepath (Path): The path to the file where the data will be saved.

    Returns:
        bool: True if the data was saved successfully, False otherwise.

    """
    try:
        async with aiofiles.open(filepath, "w", encoding="utf-8") as f:
            await f.write(json.dumps(data, indent=2, ensure_ascii=False))
    except Exception:
        logger.exception("Error saving %s:", filepath)
        return False
    else:
        return True


def set_file_timestamp(filepath: Path, timestamp_str: str) -> bool:
    """Set file's modification time based on ISO timestamp string.

    Args:
        filepath (Path): The path to the file.
        timestamp_str (str): The ISO timestamp string.

    Returns:
        bool: True if the timestamp was set successfully, False otherwise.

    """
    try:
        # Parse the timestamp string
        dt: datetime = datetime.strptime(timestamp_str, "%Y-%m-%d %H:%M:%S").replace(tzinfo=UTC)

        # Convert to Unix timestamp
        timestamp: float = dt.timestamp()

        # Set the file's modification time
        os.utime(filepath, (timestamp, timestamp))
    except ValueError:
        logger.info("Error setting timestamp for %s", filepath)
        return False
    else:
        logger.info("Timestamp for %s set to %s", filepath, dt.isoformat())
        return True


async def main() -> Literal[1, 0]:  # noqa: C901, PLR0912, PLR0915
    """Fetch and save articles from the Wuthering Waves website.

    Returns:
        Literal[1, 0]: 1 if an error occurred, 0 otherwise.

    """
    # Setup
    current_time = int(time.time() * 1000)  # Current time in milliseconds
    base_url = "https://hw-media-cdn-mingchao.kurogame.com/akiwebsite/website2.0/json/G152/en"
    article_menu_url: str = f"{base_url}/ArticleMenu.json?t={current_time}"
    article_base_url: LiteralString = f"{base_url}/article/"
    output_dir = Path("articles")
    output_dir.mkdir(exist_ok=True)

    logger.info("Fetching article menu from %s", article_menu_url)

    async with httpx.AsyncClient(timeout=30.0) as client:
        # Fetch the article menu
        menu_data: dict[Any, Any] | None = await fetch_json(article_menu_url, client)
        if not menu_data:
            logger.error("Error: Fetched ArticleMenu.json is empty")
            return 1

        # Save and prettify the menu JSON
        menu_file: Path = output_dir / "ArticleMenu.json"
        if await save_prettified_json(menu_data, menu_file):
            logger.info("Menu JSON saved and prettified to %s", menu_file)

        # Extract article IDs
        logger.info("Extracting article IDs...")
        article_ids: list[str] = [str(item["articleId"]) for item in menu_data if item.get("articleId")]

        if not article_ids:
            logger.warning("No article IDs found. Please check the JSON structure of ArticleMenu.json.")
            logger.warning("Full menu response for debugging:")
            logger.warning(json.dumps(menu_data, indent=2))
            return 1

        # Get list of already downloaded article IDs
        existing_files: list[str] = [file.stem for file in output_dir.glob("*.json") if file.stem != "ArticleMenu"]

        # Filter out already downloaded articles
        article_ids = [article_id for article_id in article_ids if article_id not in existing_files]

        # Download each article
        download_tasks: list[Coroutine[Any, Any, dict[Any, Any] | None]] = []
        for article_id in article_ids:
            article_url: str = f"{article_base_url}{article_id}.json?t={current_time}"
            output_file: Path = output_dir / f"{article_id}.json"

            logger.info("Downloading article %s from %s", article_id, article_url)
            download_tasks.append(fetch_json(article_url, client))

        # Wait for all downloads to complete
        results: list[dict[Any, Any] | BaseException | None] = await asyncio.gather(*download_tasks, return_exceptions=True)

        # Process the downloaded articles
        for i, result in enumerate(results):
            article_id: str = article_ids[i]
            output_file = output_dir / f"{article_id}.json"

            if isinstance(result, Exception):
                logger.error("Error downloading article %s: %s", article_id, result)
                continue

            if not result:
                logger.warning("Downloaded article %s is empty or invalid", article_id)
                continue

            # Save the article JSON
            if isinstance(result, dict) and await save_prettified_json(result, output_file):
                logger.info("Successfully downloaded and prettified %s", output_file)

        # Update file timestamps based on createTime
        for file in output_dir.glob("*.json"):
            article_id = file.stem
            if article_id == "ArticleMenu":
                continue

            # Find the corresponding article in the menu
            create_time = None
            for item in menu_data:
                if str(item.get("articleId", "")) == article_id and "createTime" in item:
                    create_time = item["createTime"]
                    break

            if not create_time:
                logger.info("Warning: no createTime for %s in menu data - skipping", article_id)
                continue

            logger.info("Setting %s timestamp to %s", file, create_time)
            if not set_file_timestamp(file, create_time):
                logger.error("  failed to update timestamp")

    logger.info("Script finished. Articles are in the '%s' directory.", output_dir)
    return 0


if __name__ == "__main__":
    asyncio.run(main())