mirror of
https://github.com/maxdorninger/MediaManager.git
synced 2026-04-23 01:05:39 +02:00
89 lines
3.8 KiB
Python
89 lines
3.8 KiB
Python
import logging
|
|
import xml.etree.ElementTree as ET
|
|
from datetime import datetime
|
|
from email.utils import parsedate_to_datetime
|
|
|
|
from media_manager.indexer.schemas import IndexerQueryResult
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class TorznabMixin:
|
|
def process_search_result(self, xml: str) -> list[IndexerQueryResult]:
|
|
result_list: list[IndexerQueryResult] = []
|
|
xml_tree = ET.fromstring(xml) # noqa: S314 # trusted source, since it is user controlled
|
|
xmlns = {
|
|
"torznab": "http://torznab.com/schemas/2015/feed",
|
|
"atom": "http://www.w3.org/2005/Atom",
|
|
}
|
|
for item in xml_tree.findall("channel/item"):
|
|
try:
|
|
flags = []
|
|
seeders = 0
|
|
age = 0
|
|
indexer_name = "unknown"
|
|
|
|
if item.find("jackettindexer") is not None:
|
|
indexer_name = item.find("jackettindexer").text
|
|
if item.find("prowlarrindexer") is not None:
|
|
indexer_name = item.find("prowlarrindexer").text
|
|
|
|
is_usenet = (
|
|
item.find("enclosure").attrib["type"] != "application/x-bittorrent"
|
|
)
|
|
|
|
attributes = list(item.findall("torznab:attr", xmlns))
|
|
for attribute in attributes:
|
|
if is_usenet:
|
|
if attribute.attrib["name"] == "usenetdate":
|
|
posted_date = parsedate_to_datetime(
|
|
attribute.attrib["value"]
|
|
)
|
|
now = datetime.now(datetime.UTC)
|
|
age = int((now - posted_date).total_seconds())
|
|
else:
|
|
if attribute.attrib["name"] == "seeders":
|
|
seeders = int(attribute.attrib["value"])
|
|
|
|
if attribute.attrib["name"] == "downloadvolumefactor":
|
|
download_volume_factor = float(attribute.attrib["value"])
|
|
if download_volume_factor == 0:
|
|
flags.append("freeleech")
|
|
if download_volume_factor == 0.5:
|
|
flags.append("halfleech")
|
|
if download_volume_factor == 0.75:
|
|
flags.append("freeleech75")
|
|
if download_volume_factor == 0.25:
|
|
flags.append("freeleech25")
|
|
|
|
if attribute.attrib["name"] == "uploadvolumefactor":
|
|
upload_volume_factor = int(attribute.attrib["value"])
|
|
if upload_volume_factor == 2:
|
|
flags.append("doubleupload")
|
|
|
|
title = item.find("title").text
|
|
size_str = item.find("size")
|
|
if size_str is None or size_str.text is None:
|
|
log.warning(f"Torznab item {title} has no size, skipping.")
|
|
continue
|
|
try:
|
|
size = int(size_str.text or "0")
|
|
except ValueError:
|
|
log.warning(f"Torznab item {title} has invalid size, skipping.")
|
|
continue
|
|
|
|
result = IndexerQueryResult(
|
|
title=title or "unknown",
|
|
download_url=str(item.find("enclosure").attrib["url"]),
|
|
seeders=seeders,
|
|
flags=flags,
|
|
size=size,
|
|
usenet=is_usenet,
|
|
age=age,
|
|
indexer=indexer_name,
|
|
)
|
|
result_list.append(result)
|
|
except Exception:
|
|
log.exception("1 Torznab search result failed")
|
|
return result_list
|