Service Layer

Documentation for JackettService, AuthorizationService, and PTPService

Overview

JackettSearchBot’s service layer provides clean abstractions for core functionality. All services are initialized in JackettSearchBot.__init__() and injected into command handlers.

JackettService

Handles all interactions with the Jackett API, including search queries, XML parsing, and result formatting.

Location: jackett_bot/services/jackett.py:29

Initialization

from jackett_bot.services.jackett import JackettService
import httpx

# Basic initialization
service = JackettService(
    jackett_url="http://localhost:9117",
    jackett_api_key="your_api_key_here"
)

# With custom HTTP client
custom_client = httpx.AsyncClient(timeout=30.0)
service = JackettService(
    jackett_url="http://localhost:9117",
    jackett_api_key="your_api_key_here",
    client=custom_client
)

The service automatically manages its HTTP client lifecycle. If you don’t provide a client, it creates one and closes it during cleanup.

Methods

build_search_url(query: str) → str

Constructs the Jackett API URL for a given query.

Parameters:

• query (str): Search query or IMDB ID

Returns: Fully constructed Jackett API URL

Behavior:

• Detects IMDB IDs (format: tt followed by digits)
• Uses imdbid parameter for IMDB searches
• Uses q parameter with URL encoding for text searches
• Searches across all configured indexers (/indexers/all/)

Example:

# Text search
url = service.build_search_url("Inception 2010")
# Returns: http://localhost:9117/api/v2.0/indexers/all/results/torznab/api?apikey=...&t=search&q=Inception%202010

# IMDB search
url = service.build_search_url("tt1375666")
# Returns: http://localhost:9117/api/v2.0/indexers/all/results/torznab/api?apikey=...&imdbid=tt1375666

Source: jackett_bot/services/jackett.py:41

search(query: str, golden_popcorn: bool, timeout: int) → list[SearchResult]

Performs an async search against Jackett and returns parsed results.

Parameters:

• query (str): Search query or IMDB ID
• golden_popcorn (bool): Filter for Golden Popcorn releases only (default: False)
• timeout (int): Request timeout in seconds (default: 15)

Returns: List of SearchResult objects

Example:

# Standard search
results = await service.search("The Matrix")

# Golden Popcorn only
gp_results = await service.search("Interstellar", golden_popcorn=True)

# With custom timeout
results = await service.search("tt0133093", timeout=30)

Source: jackett_bot/services/jackett.py:54

close() → None

Closes the HTTP client if owned by the service.

Usage:

await service.close()

Always call close() when shutting down to avoid resource leaks. The bot’s _shutdown_services() method handles this automatically.

Source: jackett_bot/services/jackett.py:63

SearchResult DataClass

Location: jackett_bot/services/jackett.py:12

@dataclass(frozen=True)
class SearchResult:
    title: str
    age: str      # Human-readable time (e.g., "2 d", "5 h", "30 m")
    size: str     # Human-readable size (e.g., "4.37 GB")

    def as_html(self) -> str:
        """Returns HTML-formatted result for Telegram."""
        return (
            f"<b>Title:</b> <code>{self.title}</code>\n"
            f"<b>Age:</b> {self.age}\n"
            f"<b>Size:</b> {self.size}\n"
        )

    def as_text(self) -> str:
        """Returns plain text result."""
        return f"Title: {self.title}\nAge: {self.age}\nSize: {self.size}\n"

Helper Functions

convert_size(size_bytes: int) → str

def convert_size(size_bytes: int) -> str:
    """Converts bytes to human-readable format."""
    if size_bytes == 0:
        return "0 B"

    size_name = ("B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB")
    index = int(math.floor(math.log(size_bytes, 1024)))
    power = math.pow(1024, index)
    size = round(size_bytes / power, 2)
    return f"{size} {size_name[index]}"

---

AuthorizationService

Thread-safe service for managing bot access control with both permanent (configured) and temporary (runtime) authorization lists.

Location: jackett_bot/services/auth.py:5

Initialization

from jackett_bot.services.auth import AuthorizationService

# Without bootstrap IDs
auth_service = AuthorizationService()

# With configured IDs from environment
auth_service = AuthorizationService(
    bootstrap_ids=[123456789, 987654321]
)

Authorization Model

The service maintains two separate authorization lists:

1. Configured IDs (_configured_ids)

   • Loaded from AUTHORIZED_CHAT_IDS environment variable
   • Permanent for the bot’s lifetime
   • Cannot be removed via /unauth
   • Persisted across restarts (via config file)

2. Temporary IDs (_temporary_ids)

   • Added via /auth command at runtime
   • Stored in-memory only
   • Can be removed via /unauth or /unauthall
   • Cleared when bot restarts

Methods

add_authorized(entity_id: int) → bool

Adds a temporary authorization.

Parameters:

• entity_id (int): User ID or chat ID to authorize

Returns: True if added, False if already authorized

Example:

if auth_service.add_authorized(123456789):
    print("User authorized")
else:
    print("User already authorized")

Source: jackett_bot/services/auth.py:17

remove_authorized(entity_id: int) → bool

Removes a temporary authorization.

Parameters:

• entity_id (int): User ID or chat ID to unauthorize

Returns: True if removed, False if not in temporary list

This method only removes temporary authorizations. Configured IDs cannot be removed at runtime.

Example:

if auth_service.remove_authorized(123456789):
    print("Authorization removed")
else:
    print("Not a temporary authorization")

Source: jackett_bot/services/auth.py:25

clear_authorized() → int

Clears all temporary authorizations.

Returns: Count of removed authorizations

Example:

count = auth_service.clear_authorized()
print(f"Removed {count} temporary authorizations")

Source: jackett_bot/services/auth.py:33

is_configured_id_authorized(entity_id: int) → bool

Checks if an ID is in the configured (permanent) list.

Source: jackett_bot/services/auth.py:47

is_temporary_id_authorized(entity_id: int) → bool

Checks if an ID is in the temporary list.

Source: jackett_bot/services/auth.py:52

list_configured_ids() → list[int]

Returns sorted list of all configured IDs.

Source: jackett_bot/services/auth.py:39

list_temporary_ids() → list[int]

Returns sorted list of all temporary IDs.

Source: jackett_bot/services/auth.py:43

Thread Safety

All methods use threading.RLock() for thread-safe access:

def add_authorized(self, entity_id: int) -> bool:
    normalized_id = self._normalize_id(entity_id)
    with self._lock:
        if normalized_id in self._configured_ids or normalized_id in self._temporary_ids:
            return False
        self._temporary_ids.add(normalized_id)
        return True

The reentrant lock (RLock) allows the same thread to acquire the lock multiple times, preventing deadlocks in nested calls.

---

PTPService

Simple service for checking PassThePopcorn availability.

Location: jackett_bot/services/ptp.py:4

Initialization

from jackett_bot.services.ptp import PTPService

# Default base URL
ptp_service = PTPService()

# Custom base URL
ptp_service = PTPService(base_url="https://custom-ptp-url.com")

# With custom HTTP client
import httpx
custom_client = httpx.AsyncClient()
ptp_service = PTPService(client=custom_client)

Methods

is_available(timeout: int) → bool

Checks if PTP is reachable and responding.

Parameters:

• timeout (int): Request timeout in seconds (default: 5)

Returns: True if PTP responds with successful status, False on any HTTP error

Example:

if await ptp_service.is_available():
    print("PTP is available")
else:
    print("PTP is down or unreachable")

Implementation:

async def is_available(self, timeout: int = 5) -> bool:
    try:
        response = await self._client.get(self.base_url, timeout=timeout)
        response.raise_for_status()
        return True
    except httpx.HTTPError:
        return False

Source: jackett_bot/services/ptp.py:10

close() → None

Closes the HTTP client if owned by the service.

Source: jackett_bot/services/ptp.py:18

---

Service Integration

Services are wired together in the JackettSearchBot class:

class JackettSearchBot:
    def __init__(self, config: BotConfig | None = None):
        self.config = config or BotConfig.from_env("config.env")
        
        # Initialize services
        self.auth_service = AuthorizationService(
            bootstrap_ids=self.config.authorized_chat_ids,
        )
        self.jackett_service = JackettService(
            jackett_url=self.config.jackett_url,
            jackett_api_key=self.config.jackett_api_key,
        )
        self.ptp_service = PTPService()
        
        # Inject into command handlers
        self.handlers = CommandHandlers(
            config=self.config,
            auth_service=self.auth_service,
            jackett_service=self.jackett_service,
            ptp_service=self.ptp_service,
            logger=self.logger,
        )

This dependency injection pattern makes services easy to test and swap out if needed.