usk_schedule_downloader/lib/sonarr.py

from dotenv import dotenv_values
import requests

import pytz

class Sonarr_API:
    def __init__(self,ip:str,key:str,**kwargs):
        super().__init__(**kwargs)
        self.ip=ip
        self.key=key
        self.tz = pytz.timezone("Asia/Bangkok")

        # self.db= sqlite_db(self.db_path)

        self.calendar_url = f"http://{ip}/feed/v3/calendar/Sonarr.ics?unmonitored=true&apikey={key}"

    def get_episodes(self, folder=None) -> dict:
        # print(folder)
        params = {
            # 'seriesId': id,
            'folder': folder
        }

        base_url = f"http://{self.ip}/api/v3/manualimport"
        headers = {
            "X-Api-Key": self.key,
        }

        r = requests.get(base_url, headers=headers,params=params)
        r.raise_for_status()
        # print(r.json())

        return r.json()

    def import_episodes(self, entry:dict,title_config:dict, episodes_dict:dict,mode:str="move") -> list[requests.Response]:
        base_url = f"http://{self.ip}/api/v3/command"
        headers = {
            'Accept': 'application/json, text/javascript, */*; q=0.01',
            'Content-Type': 'application/json',
            'X-Api-Key': self.key
        }
        responses=[]

        for d in episodes_dict:
            
            if not d.get('path') or not d.get('series') or not d.get('episodes'):
                continue
            for e in d['episodes']:
                season=int(entry['season'])
                if title_config['absolute_season']:
                    season=int(title_config['season'])+int(title_config['absolute_season'])
                ep=int(entry['episode'])
                if title_config['absolute']:
                    ep=int(entry['episode'])+int(title_config['absolute'])
                if not (str(season) == str(e['seasonNumber']) and  str(ep) == str(e['episodeNumber'])\
                        # and str(title_config['ID']) == str(e['seriesId'])\
                        ):

                    continue
                # print(title_config['ID'],str(entry['season']), str(e['seasonNumber']), str(entry['episode']), str(e['episodeNumber']))


                data = {
                    "name": "ManualImport",
                    "files": [{
                        "path": d['path'],
                        "seriesId": d['series']['id'],
                        "episodeIds": [e['id']],
                        "releaseGroup": d.get('releaseGroup'),
                        "quality": d.get('quality'),
                        "languages": d.get('languages'),
                        "indexerFlags": d.get('indexerFlags')
                    }],
                    "importMode": mode
                }

                r = requests.post(base_url, headers=headers, json=data)
                r.raise_for_status()
                responses.append(r)

        return responses

    def get_episode_detail(self, episodeIds) -> dict:
        # params = {
        #     'seriesId': seriesId,
        #     'episodeIds': episodeIds
        # }

        base_url = f"http://{self.ip}/api/v3/episode/{episodeIds}"
        headers = {
            "X-Api-Key": self.key,
        }
        r = requests.get(base_url, headers=headers)
        r.raise_for_status()
        # print(r.json())

        return r.json()

    def get_series_detail(self,id):
        headers = {
            'Accept': 'application/json',
            'X-Api-Key': self.key,
        }

        response = requests.get(f'http://{self.ip}/api/v3/series/{id}', headers=headers)
        return response.json()
    
    def get_episode_detail_from_season(self,seriesId,season):
        response=requests.get(
            f"http://{self.ip}/api/v3/episode",
            headers={
            "Accept": "application/json",
            "X-Api-Key": self.key
            },
            params={
            "seriesId":seriesId,
            "seasonNumber":season
            }
        )
        return response.json()


    # def get_today_schedule(self) -> list[dict]:
    #     response = requests.get(self.calendar_url)
    #     response.raise_for_status()  # Raises an error if the request failed

    #     # Parse the calendar
    #     calendar = Calendar(response.text)

    #     db_watchlist=self.db.get_watchlist()

    #     # List all events
    #     daily_list=[]

    #     for event in calendar.events:

    #         if date.today() != event.begin.date():
    #             continue

    #         title_split = event.name.split(" - ")
    #         title= title_split[0].strip()

    #         if not any(x['Title'] == title for x in db_watchlist):
    #             continue

    #         ep = title_split[-2].split('x')
    #         season = ep[-2].strip() if ep else "1"
    #         episode = ep[-1].strip() if ep else "1"
    #         episode_name = event.name.split(" - ")[2].strip()
    #         if int(season) < 1 or int(episode) < 1:
    #             continue
    
    #         detail ={
    #                 "title": title,
    #                 "season": int(season),
    #                 "episode": int(episode),
    #                 "episode_name": episode_name,
    #                 "start_timestamp": int(datetime.fromisoformat(str(event.begin)).astimezone(self.tz).timestamp()),
    #                 "end_timestamp": int(datetime.fromisoformat(str(event.end)).astimezone(self.tz).timestamp()),
    #                 "sonarr_id": event.uid.split("_")[-1],
    #             }
    #         daily_list.append(detail)

    #     return daily_list

    
if "__main__" == __name__:
    sonarr=Sonarr_API("media-server.lan:8989",dotenv_values("/root/VT_Schedule_downloader/.env")['sonarr_key'],"/root/VT_Schedule_downloader/app.sqlite")
    print(sonarr.get_episode_detail(1171,43758))