initial commit

.gitignore

@@ -0,0 +1,63 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+.mypy_cache/
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+env/
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# Poetry
+poetry.lock
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+.hypothesis/
+.pytest_cache/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# dotenv
+.env
+.env.*
+
+# VS Code
+.vscode/
+
+# OS files
+.DS_Store
+Thumbs.db
+
+# Misc
+*.log

auth.py

@@ -0,0 +1,29 @@
+import httpx
+import structlog
+from config import AUTHELIA_URL, get_async_logger
+from hooks import add_sonarr_api_key
+
+logger: structlog.stdlib.AsyncBoundLogger = get_async_logger()
+
+
+async def authelia_login(username: str, password: str) -> httpx.AsyncClient:
+    """
+    Perform Authelia login and return an authenticated httpx.AsyncClient with Sonarr API key added.
+    """
+    client = httpx.AsyncClient(
+        event_hooks={"request": [add_sonarr_api_key]},
+        http2=True,
+        limits=httpx.Limits(
+            keepalive_expiry=60, max_connections=200, max_keepalive_connections=30
+        ),
+    )
+
+    login_url = f"{AUTHELIA_URL}/api/firstfactor"
+    payload = {"username": username, "password": password}
+    resp = await client.post(login_url, json=payload)
+    resp.raise_for_status()
+
+    # If login is successful, cookies are set in the client
+    await logger.info("Authelia login successful", username=username)
+
+    return client

config.py

@@ -0,0 +1,35 @@
+import logging
+import os
+
+import structlog
+from dotenv import load_dotenv
+
+structlog.configure(wrapper_class=structlog.make_filtering_bound_logger(logging.INFO))
+
+
+def wrap_async(logger) -> structlog.stdlib.AsyncBoundLogger:
+    return structlog.wrap_logger(
+        logger, wrapper_class=structlog.stdlib.AsyncBoundLogger
+    )
+
+
+def get_async_logger() -> structlog.stdlib.AsyncBoundLogger:
+    return wrap_async(
+        structlog.get_logger(),
+    )
+
+
+logger: structlog.stdlib.AsyncBoundLogger = structlog.getLogger()
+
+
+def getenv(key: str) -> str:
+    value = os.getenv(key)
+    if value is None:
+        raise ValueError(f"{key} must be set in the .env file.")
+    return value
+
+
+load_dotenv()
+SONARR_URL: str = getenv("SONARR_URL")
+SONARR_API_KEY: str = getenv("SONARR_API_KEY")
+AUTHELIA_URL: str = getenv("AUTHELIA_URL")

format.py

@@ -0,0 +1,58 @@
+from datetime import datetime, timezone
+
+import structlog
+
+from config import get_async_logger
+
+logger: structlog.stdlib.AsyncBoundLogger = get_async_logger()
+
+
+def relative_diff(dt_str1, dt_str2):
+    if not dt_str1 or not dt_str2:
+        return ""
+    dt1 = datetime.fromisoformat(dt_str1.replace("Z", "+00:00"))
+    dt2 = datetime.fromisoformat(dt_str2.replace("Z", "+00:00"))
+    diff = abs(dt1 - dt2)
+    seconds = int(diff.total_seconds())
+    minutes = seconds // 60
+    hours = minutes // 60
+    days = hours // 24
+
+    parts = []
+    if days > 0:
+        parts.append(f"{days}d")
+    hours = hours % 24
+    if hours > 0 or days > 0:
+        parts.append(f"{hours}h")
+    minutes = minutes % 60
+    if minutes > 0 and days == 0:
+        parts.append(f"{minutes}m")
+    return " ".join(parts) if parts else "0m"
+
+
+def relative_time(dt_str):
+    dt = datetime.fromisoformat(dt_str.replace("Z", "+00:00"))
+    now = datetime.now(timezone.utc)
+    diff = now - dt
+    seconds = int(diff.total_seconds())
+    minutes = seconds // 60
+    hours = minutes // 60
+    days = hours // 24
+    months = days // 30
+
+    if months > 0:
+        return f"{months} month{'s' if months != 1 else ''} ago"
+    elif days > 0:
+        return f"{days} day{'s' if days != 1 else ''} ago"
+    elif hours > 0:
+        return f"{hours} hour{'s' if hours != 1 else ''} ago"
+    elif minutes > 0:
+        return f"{minutes} minute{'s' if minutes != 1 else ''} ago"
+    else:
+        return "just now"
+
+
+def ellipsis(s, max_length):
+    if not isinstance(s, str) or max_length < 4 or len(s) <= max_length:
+        return s
+    return s[: max_length - 1] + "…"

hooks.py

@@ -0,0 +1,18 @@
+import httpx
+import structlog
+
+from config import SONARR_API_KEY, SONARR_URL, get_async_logger
+
+logger: structlog.stdlib.AsyncBoundLogger = get_async_logger()
+
+
+async def add_sonarr_api_key(request: httpx.Request) -> None:
+    """
+    Event hook to inject the Sonarr API key into requests to the Sonarr domain and /api path.
+    """
+    if SONARR_URL and request.url.host in SONARR_URL and "/api" in request.url.path:
+        await logger.debug(
+            "applying sonarr api key",
+            sonarr_url=SONARR_URL,
+        )
+        request.headers["X-Api-Key"] = SONARR_API_KEY

main.py

@@ -0,0 +1,230 @@
+from anyio import run
+import os
+from collections import defaultdict
+from typing import Dict
+
+import structlog
+import httpx
+import asyncio
+
+from auth import authelia_login
+from config import get_async_logger
+from format import ellipsis
+import anyio
+from collections import Counter
+from sonarr import (
+    get_all_series,
+    get_episodes_for_series,
+    get_history_for_episode,
+    get_series,
+)
+
+logger: structlog.stdlib.AsyncBoundLogger = get_async_logger()
+
+
+async def process_series(client, series_id: int) -> Dict[str, int]:
+    """
+    For a given series, count the number of files per indexer by analyzing episode history.
+    Returns a dictionary mapping indexer names to file counts.
+    """
+    series = await get_series(client, series_id)
+    episodes = await get_episodes_for_series(client, series_id)
+    indexer_counts: Dict[str, int] = defaultdict(lambda: 0)
+
+    for ep in episodes:
+        # Skip episodes without files
+        if not ep.get("hasFile", False):
+            continue
+
+        indexer = "unknown"
+        episode_detail = f"{ellipsis(series['title'], 12)} S{ep['seasonNumber']:02}E{ep['episodeNumber']:02} - {ellipsis(ep.get('title', 'Unknown'), 18)}"
+
+        # Retrieves all history events for the episode
+        try:
+            history = await get_history_for_episode(client, ep["id"])
+        except Exception as e:
+            if (
+                hasattr(e, "response")
+                and getattr(e.response, "status_code", None) == 404
+            ):
+                await logger.warning(
+                    "History not found for episode (404)",
+                    episode_id=ep["id"],
+                    episode=episode_detail,
+                )
+                continue
+            else:
+                await logger.error(
+                    "Error fetching history for episode",
+                    episode_id=ep["id"],
+                    episode=episode_detail,
+                    error=str(e),
+                )
+                continue
+
+        # Get the episode's episodeFileId
+        target_file_id = ep.get("episodeFileId")
+        if not target_file_id:
+            await logger.error(
+                "No episode file for episode",
+                episode_id=ep["id"],
+                episode=episode_detail,
+            )
+            continue
+
+        # Find the 'downloadFolderImported' event with the matching data.fileId
+        def is_import_event(event: Dict, file_id: int) -> bool:
+            return event.get("eventType") == "downloadFolderImported" and event.get(
+                "data", {}
+            ).get("fileId") == str(file_id)
+
+        import_event = next(
+            (
+                event
+                for event in history["records"]
+                if is_import_event(event, target_file_id)
+            ),
+            None,
+        )
+        if not import_event:
+            await logger.debug(
+                "No import event found for episode file",
+                episode_id=ep["id"],
+                episode=episode_detail,
+                target_file_id=target_file_id,
+            )
+            continue
+
+        # Acquire the event's downloadId
+        download_id = import_event.get("downloadId")
+        if not download_id:
+            await logger.error(
+                "No downloadId found in import event",
+                episode_id=ep["id"],
+                episode=episode_detail,
+                target_file_id=target_file_id,
+            )
+            continue
+
+        # Find the 'grabbed' event with the matching downloadId
+        def is_grabbed_event(event: Dict, download_id: str) -> bool:
+            return (
+                event.get("eventType") == "grabbed"
+                and event.get("downloadId") == download_id
+            )
+
+        grabbed_event = next(
+            (
+                event
+                for event in history["records"]
+                if is_grabbed_event(event, download_id)
+            ),
+            None,
+        )
+        if not grabbed_event:
+            await logger.debug(
+                "No 'grabbed' event found",
+                episode_id=ep["id"],
+                download_id=ellipsis(download_id, 20),
+                episode=episode_detail,
+            )
+            continue
+
+        # Extract the indexer from the 'grabbed' event
+        indexer = grabbed_event.get("data", {}).get("indexer")
+        if not indexer:
+            await logger.warning(
+                "No indexer provided in 'grabbed' event",
+                episode_id=ep["id"],
+                episode=episode_detail,
+                download_id=ellipsis(download_id, 20),
+            )
+
+            indexer = "unknown"
+
+            # Normalize indexer names by removing the "(Prowlarr)" suffix
+            indexer = indexer[:-11] if indexer.endswith("(Prowlarr)") else indexer
+
+        indexer_counts[indexer] += 1
+
+    return indexer_counts
+
+
+async def main():
+    """
+    Entrypoint: Authenticates with Authelia, fetches all Sonarr series, and logs per-series indexer statistics.
+    """
+    username = os.getenv("AUTHELIA_USERNAME")
+    password = os.getenv("AUTHELIA_PASSWORD")
+
+    if not username or not password:
+        await logger.critical(
+            "Missing Authelia credentials",
+            AUTHELIA_USERNAME=username,
+            AUTHELIA_PASSWORD=bool(password),
+        )
+        raise Exception(
+            "AUTHELIA_USERNAME and AUTHELIA_PASSWORD must be set in the .env file."
+        )
+
+    # Request counter setup
+    request_counter = {"count": 0, "active": 0}
+    counter_lock = anyio.Lock()
+
+    async def count_requests(request):
+        async with counter_lock:
+            request_counter["count"] += 1
+            request_counter["active"] += 1
+
+    async def count_responses(response):
+        async with counter_lock:
+            request_counter["active"] -= 1
+
+    # Attach the event hooks to the client
+    client = await authelia_login(username, password)
+    if hasattr(client, "event_hooks"):
+        client.event_hooks.setdefault("request", []).append(count_requests)
+        client.event_hooks.setdefault("response", []).append(count_responses)
+
+    series_list = await get_all_series(client)
+
+    total_indexer_counts = Counter()
+
+    async def process_and_log(series):
+        indexer_counts = await process_series(client, series["id"])
+        if any(indexer_counts.keys()):
+            await logger.debug(
+                "Processed series",
+                series_title=series["title"],
+                series_id=series["id"],
+                indexers=dict(indexer_counts),
+            )
+            total_indexer_counts.update(indexer_counts)
+
+    async def print_rps(interval=3):
+        while True:
+            await anyio.sleep(interval)
+            async with counter_lock:
+                rps = request_counter["count"] / interval
+                active = request_counter["active"]
+                request_counter["count"] = 0
+            await logger.info(
+                "Requests per second and active requests",
+                rps=rps,
+                active_requests=active,
+                interval=interval,
+            )
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(print_rps, 3)
+        for series in series_list:
+            tg.start_soon(process_and_log, series)
+
+    await logger.info(
+        "Total indexer counts across all series",
+        indexers=dict(total_indexer_counts),
+    )
+
+
+if __name__ == "__main__":
+    run(main)

pyproject.toml

@@ -0,0 +1,18 @@
+[tool.poetry]
+name = "indexer-analyze"
+version = "0.1.0"
+description = ""
+authors = ["Xevion <xevion@xevion.dev>"]
+readme = "README.md"
+
+[tool.poetry.dependencies]
+python = "^3.10"
+python-dotenv = "^1.1.0"
+structlog = "^25.3.0"
+httpx = {extras = ["http2"], version = "^0.28.1"}
+anyio = "^4.9.0"
+
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

sonarr.py

@@ -0,0 +1,44 @@
+import httpx
+from config import SONARR_URL, SONARR_API_KEY, logger
+
+
+async def get_all_series(client: httpx.AsyncClient):
+    """
+    Fetch all series from Sonarr.
+    """
+    url = f"{SONARR_URL}/api/v3/series"
+    resp = await client.get(url)
+    resp.raise_for_status()
+    return resp.json()
+
+
+async def get_series(client: httpx.AsyncClient, series_id: int):
+    """
+    Fetch a single series by ID from Sonarr.
+    """
+    url = f"{SONARR_URL}/api/v3/series/{series_id}"
+    resp = await client.get(url)
+    resp.raise_for_status()
+    return resp.json()
+
+
+async def get_episodes_for_series(client: httpx.AsyncClient, series_id: int):
+    """
+    Fetch all episodes for a given series from Sonarr.
+    """
+    url = f"{SONARR_URL}/api/v3/episode?seriesId={series_id}"
+    resp = await client.get(url)
+    resp.raise_for_status()
+    return resp.json()
+
+
+async def get_history_for_episode(client: httpx.AsyncClient, episode_id: int):
+    """
+    Fetch history for a given episode from Sonarr.
+    """
+    resp = await client.get(
+        SONARR_URL + "/api/v3/history",
+        params={"episodeId": episode_id, "pageSize": 100},
+    )
+    resp.raise_for_status()
+    return resp.json()