2023-01-14 10:23:50 +00:00
|
|
|
#!/usr/bin/env nix-shell
|
|
|
|
#!nix-shell -i python3 -p "python3.withPackages (ps: [ ps.natsort ps.requests ])"
|
2022-12-26 09:05:26 +00:00
|
|
|
"""
|
|
|
|
usage: sane-bt-search <query_string>
|
|
|
|
|
|
|
|
searches Jackett for torrent files matching the title.
|
|
|
|
returns select results and magnet links
|
|
|
|
"""
|
|
|
|
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from datetime import datetime
|
|
|
|
import natsort
|
|
|
|
import requests
|
|
|
|
import sys
|
|
|
|
import time
|
|
|
|
|
|
|
|
SERVICE = "https://jackett.uninsane.org"
|
|
|
|
ENDPOINTS = dict(
|
|
|
|
results="api/v2.0/indexers/all/results"
|
|
|
|
)
|
|
|
|
|
|
|
|
@dataclass(eq=True, order=True)
|
|
|
|
class Torrent:
|
|
|
|
seeders: int
|
|
|
|
pub_date: datetime
|
2023-01-20 02:10:07 +00:00
|
|
|
size: int
|
|
|
|
tracker: str
|
2022-12-26 09:05:26 +00:00
|
|
|
title: str
|
|
|
|
magnet: str
|
|
|
|
|
|
|
|
def __str__(self) -> str:
|
2023-01-20 02:10:07 +00:00
|
|
|
return f"{self.seeders}[S]\t{self.pub_date}\t{self.mib}M\t{self.tracker}\t{self.title}\t{self.magnet}"
|
|
|
|
|
|
|
|
@property
|
|
|
|
def mib(self) -> int:
|
|
|
|
return int(round(self.size / 1024 / 1024))
|
2022-12-26 09:05:26 +00:00
|
|
|
|
|
|
|
@staticmethod
|
|
|
|
def from_dict(d: dict) -> 'Torrent':
|
|
|
|
seeders = d.get("Seeders")
|
|
|
|
pub_date = d.get("PublishDate")
|
2023-01-20 02:10:07 +00:00
|
|
|
size = d.get("Size")
|
|
|
|
tracker = d.get("Tracker")
|
2022-12-26 09:05:26 +00:00
|
|
|
title = d.get("Title")
|
|
|
|
magnet = d.get("MagnetUri")
|
|
|
|
if seeders is not None and pub_date is not None and title is not None and magnet is not None:
|
2023-01-20 02:05:40 +00:00
|
|
|
pub_date = datetime.fromisoformat(pub_date).astimezone()
|
2023-01-20 02:10:07 +00:00
|
|
|
return Torrent(seeders, pub_date, size, tracker, title, magnet)
|
2022-12-26 09:05:26 +00:00
|
|
|
|
|
|
|
class Client:
|
|
|
|
def __init__(self):
|
|
|
|
self.apikey = open("/run/secrets/jackett_apikey").read()
|
|
|
|
|
|
|
|
def api_call(self, method: str, params: dict) -> dict:
|
|
|
|
endpoint = ENDPOINTS[method]
|
|
|
|
url = f"{SERVICE}/{endpoint}"
|
|
|
|
params = params.copy()
|
|
|
|
params.update(apikey=self.apikey, _=str(int(time.time())))
|
|
|
|
print(url)
|
|
|
|
print(params)
|
|
|
|
resp = requests.get(url, params=params)
|
|
|
|
print(resp)
|
|
|
|
return resp.json()
|
|
|
|
|
|
|
|
def query(self, q: str) -> list:
|
|
|
|
torrents = []
|
|
|
|
api_res = self.api_call("results", dict(Query=q))
|
|
|
|
for r in api_res["Results"]:
|
|
|
|
t = Torrent.from_dict(r)
|
|
|
|
if t is not None:
|
|
|
|
torrents.append(t)
|
|
|
|
|
|
|
|
return sorted(torrents, reverse=True)
|
|
|
|
|
|
|
|
q = " ".join(sys.argv[1:])
|
|
|
|
|
|
|
|
client = Client()
|
|
|
|
res = client.query(q)
|
2023-01-14 10:23:50 +00:00
|
|
|
print(f"found {len(res)} result(s)")
|
2022-12-26 09:05:26 +00:00
|
|
|
for r in res[:5]:
|
|
|
|
print(r)
|