A very lightweight Python script to download torrent files (less than X days old) from the https://showrss.info feed.

showtime.py

#!/usr/bin/env python3

"""
A very lightweight Python script to download torrent files (less than X days old) from the https://showrss.info feed.

Requirements:

    - feedparser
"""

from os import walk
import feedparser
import re
import signal
import sys
from datetime import datetime, timedelta
from pathlib import Path
from time import mktime
from types import FrameType
from typing import Optional, Generator

# Get your feed url from: https://showrss.info/feeds
# Leave the "Link type" option as "Use magnets" in feed (recommended).
_USER_FEED_URL = 'https://showrss.info/user/000000.rss?magnets=true&namespaces=true&name=null&quality=hd&re=yes'

# Set the path where the torrent files wil be created.
_TORRENTS_PATH = '/home/rubenhortas/showtime/torrents'

# Number of days back from the current date to search for torrents.
_THRESHOLD_DAYS = 7

_SHOWRSS_FEED_URL_PATTERN = re.compile(
    r'^https://showrss\.info/user/\d+\.rss\?'
    r'magnets=(true|false)&'
    r'namespaces=(true|false)&'
    r'name=(null|clean)&'
    r'quality=(null|any|sd|hd|fhd|anyhd)&'
    r're=(null|yes|no)$'
)
_REGEX_MAGNET_URI = re.compile(r'xt=urn:btih:([^&/]+)')


def _handle_sigint(signal: int, frame: Optional[FrameType]) -> None:
    sys.exit(0)


class Cache:
    _SEPARATOR = ','
    _DATE_FORMAT = '%Y-%m-%d'
    _FILE_NAME = 'showtime.cache'

    def __init__(self):
        self._cache_file = Path(__file__).parent / self._FILE_NAME
        self._threshold_date = (datetime.today() - timedelta(days=_THRESHOLD_DAYS)).replace(hour=0, minute=0, second=0, microsecond=0)
        self._downloaded: set[str] = set()
        self._get_downloaded()

    def is_new(self, published_date: datetime, file_name: str) -> bool:
        entry = self._create_entry(published_date, file_name)
        return published_date >= self._threshold_date and entry not in self._downloaded

    def add(self, published_date: datetime, file_name: str) -> None:
        entry = self._create_entry(published_date, file_name)
        self._downloaded.add(entry)

    def save(self) -> None:
        sorted_entries = sorted(self._downloaded, reverse=True)
        self._cache_file.write_text('\n'.join(sorted_entries) + '\n', encoding='UTF-8')

    def _create_entry(self, published_date: datetime, file_name: str) -> str:
        return f'{published_date.strftime(self._DATE_FORMAT)}{self._SEPARATOR}{file_name}'

    def _get_downloaded(self) -> None:
        if not self._cache_file.exists():
            return

        with self._cache_file.open('r', encoding='UTF-8') as file:
            clean_lines = (stripped for line in file if (stripped := line.strip()))

            for line in clean_lines:
                try:
                    parts = line.split(self._SEPARATOR)
                    date_str = parts[0]
                    date_val = datetime.strptime(date_str, self._DATE_FORMAT)

                    if date_val >= self._threshold_date:
                        self._downloaded.add(line)
                except (ValueError, IndexError):
                    continue

class Torrent:
    def __init__(self, torrents_path: Path, title: str, parsed_date: tuple, magnet_uri: str):
        self._torrents_path = torrents_path
        self._title = re.sub(r'[\\/*?:"<>|]', "", title)  # Sanitized title
        self._magnet_uri = magnet_uri
        self.file_name = f'{title}.torrent'
        self.published_date = datetime.fromtimestamp(mktime(parsed_date))

    def __str__(self):
        return self.file_name

    def save_to_disk(self) -> bool:
        # Extract the hash info (BTIH) from the magnet link
        match = _REGEX_MAGNET_URI.search(self._magnet_uri)

        if match:
            # Create the contents of the .torrent file in Bencode format.
            # Bencode format: d10:magnet-uri[uri_length]:[uri]e
            data = f"d10:magnet-uri{len(self._magnet_uri)}:{self._magnet_uri}e"

            try:
                target_path = self._torrents_path / self.file_name
                target_path.write_text(data, encoding='UTF-8')

                return True
            except (IOError, OSError) as e:
                print(f"Error writing torrent file {self.file_name}: {e}")

        return False


def _is_valid_showrss_feed_url() -> bool:
    if not _USER_FEED_URL:
        return False

    return bool(_SHOWRSS_FEED_URL_PATTERN.match(_USER_FEED_URL))


def _download_torrents() -> Generator['Torrent', None, None]:
    torrents_path = Path(_TORRENTS_PATH)
    torrents_path.mkdir(parents=True, exist_ok=True)

    cache = Cache()
    feed = feedparser.parse(_USER_FEED_URL)

    for entry in feed.entries:
        try:
            torrent = Torrent(torrents_path, str(entry.title), tuple(entry.published_parsed), str(entry.link))

            if cache.is_new(torrent.published_date, torrent.file_name) and torrent.save_to_disk():
                cache.add(torrent.published_date, torrent.file_name)

                yield torrent
        except Exception:
            continue

    cache.save()


if __name__ == '__main__':
    signal.signal(signal.SIGINT, _handle_sigint)

    try:
        if _is_valid_showrss_feed_url():
            print('Downloading torrents...')

            for torrent in _download_torrents():
                print(torrent)

            print('Done')
        else:
            print(f"Invalid feed '{_USER_FEED_URL}'")
    except Exception as e:
        print(f"Unexpected error: {e}")
        sys.exit(1)