Nathanwoodburn.github.io/cache_helper.py

"""
Cache helper module for expensive API calls and configuration.
Provides centralized caching with TTL for external API calls.
"""

import datetime
import os
import json
import requests
from functools import lru_cache


# Cache storage for NC_CONFIG with timestamp
_nc_config_cache = {"data": None, "timestamp": 0}
_nc_config_ttl = 3600  # 1 hour cache


def get_nc_config():
    """
    Get NC_CONFIG with caching (1 hour TTL).
    Falls back to default config on error.

    Returns:
        dict: Configuration dictionary
    """
    global _nc_config_cache
    current_time = datetime.datetime.now().timestamp()

    # Check if cache is valid
    if (
        _nc_config_cache["data"]
        and (current_time - _nc_config_cache["timestamp"]) < _nc_config_ttl
    ):
        return _nc_config_cache["data"]

    # Fetch new config
    try:
        config = requests.get(
            "https://cloud.woodburn.au/s/4ToXgFe3TnnFcN7/download/website-conf.json",
            timeout=5,
        ).json()
        _nc_config_cache = {"data": config, "timestamp": current_time}
        return config
    except Exception as e:
        print(f"Error fetching NC_CONFIG: {e}")
        # Return cached data if available, otherwise default
        if _nc_config_cache["data"]:
            return _nc_config_cache["data"]
        return {"time-zone": 10, "message": ""}


# Cache storage for git data
_git_data_cache = {"data": None, "timestamp": 0}
_git_data_ttl = 300  # 5 minutes cache


def get_git_latest_activity():
    """
    Get latest git activity with caching (5 minute TTL).

    Returns:
        dict: Git activity data or default values
    """
    global _git_data_cache
    current_time = datetime.datetime.now().timestamp()

    # Check if cache is valid
    if (
        _git_data_cache["data"]
        and (current_time - _git_data_cache["timestamp"]) < _git_data_ttl
    ):
        return _git_data_cache["data"]

    # Fetch new data
    try:
        git = requests.get(
            "https://git.woodburn.au/api/v1/users/nathanwoodburn/activities/feeds?only-performed-by=true&limit=1",
            headers={
                "Authorization": os.getenv("GIT_AUTH") or os.getenv("git_token") or ""
            },
            timeout=5,
        )
        git_data = git.json()
        if git_data and len(git_data) > 0:
            result = git_data[0]
            _git_data_cache = {"data": result, "timestamp": current_time}
            return result
    except Exception as e:
        print(f"Error fetching git data: {e}")

    # Return cached or default
    if _git_data_cache["data"]:
        return _git_data_cache["data"]

    return {
        "repo": {
            "html_url": "https://nathan.woodburn.au",
            "name": "nathanwoodburn.github.io",
            "description": "Personal website",
        }
    }


# Cache storage for projects
_projects_cache = {"data": None, "timestamp": 0}
_projects_ttl = 7200  # 2 hours cache


def get_projects(limit=3):
    """
    Get projects list with caching (2 hour TTL).

    Args:
        limit (int): Number of projects to return

    Returns:
        list: List of project dictionaries
    """
    global _projects_cache
    current_time = datetime.datetime.now().timestamp()

    # Check if cache is valid
    if (
        _projects_cache["data"]
        and (current_time - _projects_cache["timestamp"]) < _projects_ttl
    ):
        return _projects_cache["data"][:limit]

    # Fetch new data
    try:
        projects = []
        projectsreq = requests.get(
            "https://git.woodburn.au/api/v1/users/nathanwoodburn/repos", timeout=5
        )
        projects = projectsreq.json()

        # Check for pagination
        pageNum = 2
        while 'rel="next"' in projectsreq.headers.get("link", ""):
            projectsreq = requests.get(
                f"https://git.woodburn.au/api/v1/users/nathanwoodburn/repos?page={pageNum}",
                timeout=5,
            )
            projects += projectsreq.json()
            pageNum += 1
            # Safety limit
            if pageNum > 10:
                break

        # Process projects
        for project in projects:
            if project.get("avatar_url") in ("https://git.woodburn.au/", ""):
                project["avatar_url"] = "/favicon.png"
            project["name"] = project["name"].replace("_", " ").replace("-", " ")

        # Sort by last updated
        projects_sorted = sorted(
            projects, key=lambda x: x.get("updated_at", ""), reverse=True
        )

        # Remove duplicates by name
        seen_names = set()
        unique_projects = []
        for project in projects_sorted:
            if project["name"] not in seen_names:
                unique_projects.append(project)
                seen_names.add(project["name"])

        _projects_cache = {"data": unique_projects, "timestamp": current_time}
        return unique_projects[:limit]
    except Exception as e:
        print(f"Error fetching projects: {e}")
        if _projects_cache["data"]:
            return _projects_cache["data"][:limit]
        return []


# Cache storage for uptime status
_uptime_cache = {"data": None, "timestamp": 0}
_uptime_ttl = 300  # 5 minutes cache


def get_uptime_status():
    """
    Get uptime status with caching (5 minute TTL).

    Returns:
        bool: True if services are up, False otherwise
    """
    global _uptime_cache
    current_time = datetime.datetime.now().timestamp()

    # Check if cache is valid
    if (
        _uptime_cache["data"] is not None
        and (current_time - _uptime_cache["timestamp"]) < _uptime_ttl
    ):
        return _uptime_cache["data"]

    # Fetch new data
    try:
        uptime = requests.get(
            "https://uptime.woodburn.au/api/status-page/main/badge", timeout=5
        )
        content = uptime.content.decode("utf-8").lower()
        status = "maintenance" in content or uptime.content.count(b"Up") > 1
        _uptime_cache = {"data": status, "timestamp": current_time}
        return status
    except Exception as e:
        print(f"Error fetching uptime: {e}")
        # Return cached or default (assume up)
        if _uptime_cache["data"] is not None:
            return _uptime_cache["data"]
        return True


# Cached wallet data loaders
@lru_cache(maxsize=1)
def get_wallet_tokens():
    """
    Get wallet tokens with caching.

    Returns:
        list: List of token dictionaries
    """
    try:
        with open(".well-known/wallets/.tokens") as file:
            return json.load(file)
    except Exception as e:
        print(f"Error loading tokens: {e}")
        return []


@lru_cache(maxsize=1)
def get_coin_names():
    """
    Get coin names with caching.

    Returns:
        dict: Dictionary of coin names
    """
    try:
        with open(".well-known/wallets/.coins") as file:
            return json.load(file)
    except Exception as e:
        print(f"Error loading coin names: {e}")
        return {}


@lru_cache(maxsize=1)
def get_wallet_domains():
    """
    Get wallet domains with caching.

    Returns:
        dict: Dictionary of wallet domains
    """
    try:
        if os.path.isfile(".well-known/wallets/.domains"):
            with open(".well-known/wallets/.domains") as file:
                return json.load(file)
    except Exception as e:
        print(f"Error loading domains: {e}")
    return {}