feat: added lidarr support
This commit is contained in:
Kamil
2
lidarr/__init__.py
Normal file
2
lidarr/__init__.py
Normal file
@@ -0,0 +1,2 @@
|
||||
from .client import LidarrClient
|
||||
__all__ = ["LidarrClient"]
|
||||
174
lidarr/classes.py
Normal file
174
lidarr/classes.py
Normal file
@@ -0,0 +1,174 @@
|
||||
from dataclasses import dataclass, field
|
||||
from typing import List, Optional
|
||||
|
||||
@dataclass
|
||||
class Image:
|
||||
url: str
|
||||
coverType: str
|
||||
extension: str
|
||||
remoteUrl: str
|
||||
|
||||
@dataclass
|
||||
class Link:
|
||||
url: str
|
||||
name: str
|
||||
|
||||
@dataclass
|
||||
class Ratings:
|
||||
votes: int
|
||||
value: float
|
||||
|
||||
@dataclass
|
||||
class AddOptions:
|
||||
monitor: str
|
||||
albumsToMonitor: List[str]
|
||||
monitored: bool
|
||||
searchForMissingAlbums: bool
|
||||
|
||||
@dataclass
|
||||
class Statistics:
|
||||
albumCount: int
|
||||
trackFileCount: int
|
||||
trackCount: int
|
||||
totalTrackCount: int
|
||||
sizeOnDisk: int
|
||||
percentOfTracks: float
|
||||
|
||||
@dataclass
|
||||
class Member:
|
||||
name: str
|
||||
instrument: str
|
||||
images: List[Image]
|
||||
|
||||
@dataclass
|
||||
class Artist:
|
||||
mbId: Optional[str] = None
|
||||
tadbId: Optional[int] = None
|
||||
discogsId: Optional[int] = None
|
||||
allMusicId: Optional[str] = None
|
||||
overview: str = ""
|
||||
artistType: str = ""
|
||||
disambiguation: str = ""
|
||||
links: List[Link] = field(default_factory=list)
|
||||
nextAlbum: str = ""
|
||||
lastAlbum: str = ""
|
||||
images: List[Image] = field(default_factory=list)
|
||||
members: List[Member] = field(default_factory=list)
|
||||
remotePoster: str = ""
|
||||
path: str = ""
|
||||
qualityProfileId: int = 0
|
||||
metadataProfileId: int = 0
|
||||
monitored: bool = False
|
||||
monitorNewItems: str = ""
|
||||
rootFolderPath: Optional[str] = None
|
||||
folder: str = ""
|
||||
genres: List[str] = field(default_factory=list)
|
||||
cleanName: str = ""
|
||||
sortName: str = ""
|
||||
tags: List[int] = field(default_factory=list)
|
||||
added: str = ""
|
||||
addOptions: Optional[AddOptions] = None
|
||||
ratings: Optional[Ratings] = None
|
||||
statistics: Optional[Statistics] = None
|
||||
status : str = ""
|
||||
ended : bool = False
|
||||
artistName : str = ""
|
||||
foreignArtistId : str = ""
|
||||
id : int = 0
|
||||
|
||||
|
||||
@dataclass
|
||||
class Media:
|
||||
mediumNumber: int
|
||||
mediumName: str
|
||||
mediumFormat: str
|
||||
|
||||
@dataclass
|
||||
class Release:
|
||||
id: int
|
||||
albumId: int
|
||||
foreignReleaseId: str
|
||||
title: str
|
||||
status: str
|
||||
duration: int
|
||||
trackCount: int
|
||||
media: List[Media]
|
||||
mediumCount: int
|
||||
disambiguation: str
|
||||
country: List[str]
|
||||
label: List[str]
|
||||
format: str
|
||||
monitored: bool
|
||||
|
||||
@dataclass
|
||||
class Album:
|
||||
id: int = 0
|
||||
title: str = ""
|
||||
disambiguation: str = ""
|
||||
overview: str = ""
|
||||
artistId: int = 0
|
||||
foreignAlbumId: str = ""
|
||||
monitored: bool = False
|
||||
anyReleaseOk: bool = False
|
||||
profileId: int = 0
|
||||
duration: int = 0
|
||||
albumType: str = ""
|
||||
secondaryTypes: List[str] = field(default_factory=list)
|
||||
mediumCount: int = 0
|
||||
ratings: Ratings = None
|
||||
releaseDate: str = ""
|
||||
releases: List[Release] = field(default_factory=list)
|
||||
genres: List[str] = field(default_factory=list)
|
||||
media: List[Media] = field(default_factory=list)
|
||||
artist: Artist = field(default_factory=Artist)
|
||||
images: List[Image] = field(default_factory=list)
|
||||
links: List[Link] = field(default_factory=list)
|
||||
lastSearchTime: str = ""
|
||||
statistics: Statistics = None
|
||||
addOptions: Optional[dict] = field(default_factory=dict)
|
||||
remoteCover: str = ""
|
||||
|
||||
@dataclass
|
||||
class RootFolder:
|
||||
id: int = 0
|
||||
name: str = ""
|
||||
path: str = ""
|
||||
defaultMetadataProfileId: int = 0
|
||||
defaultQualityProfileId: int = 0
|
||||
defaultMonitorOption: str = ""
|
||||
defaultNewItemMonitorOption: str = ""
|
||||
defaultTags: List[int] = field(default_factory=list)
|
||||
accessible: bool = False
|
||||
freeSpace: int = 0
|
||||
totalSpace: int = 0
|
||||
|
||||
@dataclass
|
||||
class Quality:
|
||||
id: int = 0
|
||||
name: str = ""
|
||||
|
||||
@dataclass
|
||||
class Item:
|
||||
id: int = 0
|
||||
name: str = ""
|
||||
quality: Quality = field(default_factory=Quality)
|
||||
items: List[str] = field(default_factory=list)
|
||||
allowed: bool = False
|
||||
|
||||
@dataclass
|
||||
class FormatItem:
|
||||
id: int = 0
|
||||
format: int = 0
|
||||
name: str = ""
|
||||
score: int = 0
|
||||
|
||||
@dataclass
|
||||
class QualityProfile:
|
||||
id: int = 0
|
||||
name: str = ""
|
||||
upgradeAllowed: bool = False
|
||||
cutoff: int = 0
|
||||
items: List[Item] = field(default_factory=list)
|
||||
minFormatScore: int = 0
|
||||
cutoffFormatScore: int = 0
|
||||
formatItems: List[FormatItem] = field(default_factory=list)
|
||||
143
lidarr/client.py
Normal file
143
lidarr/client.py
Normal file
@@ -0,0 +1,143 @@
|
||||
import json
|
||||
import re
|
||||
from flask import jsonify
|
||||
import requests
|
||||
from typing import List, Optional
|
||||
from .classes import Album, Artist, QualityProfile, RootFolder
|
||||
import logging
|
||||
l = logging.getLogger(__name__)
|
||||
|
||||
class LidarrClient:
|
||||
def __init__(self, base_url: str, api_token: str):
|
||||
self.base_url = base_url
|
||||
self.api_token = api_token
|
||||
self.headers = {
|
||||
'X-Api-Key': self.api_token
|
||||
}
|
||||
|
||||
def _get(self, endpoint: str, params: Optional[dict] = None):
|
||||
response = requests.get(f"{self.base_url}{endpoint}", headers=self.headers, params=params)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
def _post(self, endpoint: str, json: dict):
|
||||
response = requests.post(f"{self.base_url}{endpoint}", headers=self.headers, json=json)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def _put(self, endpoint: str, json: dict):
|
||||
response = requests.put(f"{self.base_url}{endpoint}", headers=self.headers, json=json)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def get_album(self, album_id: int) -> Album:
|
||||
l.debug(f"Getting album {album_id}")
|
||||
data = self._get(f"/api/v1/album/{album_id}")
|
||||
return Album(**data)
|
||||
|
||||
def get_artist(self, artist_id: int) -> Artist:
|
||||
l.debug(f"Getting artist {artist_id}")
|
||||
data = self._get(f"/api/v1/artist/{artist_id}")
|
||||
return Artist(**data)
|
||||
|
||||
def search(self, term: str) -> List[object]:
|
||||
l.debug(f"Searching for {term}")
|
||||
data = self._get("/api/v1/search", params={"term": term})
|
||||
results = []
|
||||
for item in data:
|
||||
if 'artist' in item:
|
||||
results.append(Artist(**item['artist']))
|
||||
elif 'album' in item:
|
||||
results.append(Album(**item['album']))
|
||||
return results
|
||||
# A method which takes a List[object] end external URL as parameter, and returns the object from the List[object] which has the same external URL as the parameter.
|
||||
def get_object_by_external_url(self, objects: List[object], external_url: str) -> object:
|
||||
l.debug(f"Getting object by external URL {external_url}")
|
||||
# We need to check whether the external_url matches intl-[a-zA-Z]{2}\/ it has to be replaced by an empty string
|
||||
external_url = re.sub(r"intl-[a-zA-Z]{2}\/", "", external_url)
|
||||
for obj in objects:
|
||||
# object can either be an Album or an Artist, so it can be verified and casted
|
||||
if isinstance(obj, Album):
|
||||
for link in obj.links:
|
||||
if link['url'] == external_url:
|
||||
return obj
|
||||
elif isinstance(obj, Artist):
|
||||
for link in obj.links:
|
||||
if link['url'] == external_url:
|
||||
return obj
|
||||
|
||||
return None
|
||||
# A method to get all Albums from List[object] where the name equals the parameter name
|
||||
def get_albums_by_name(self, objects: List[object], name: str) -> List[Album]:
|
||||
l.debug(f"Getting albums by name {name}")
|
||||
albums = []
|
||||
for obj in objects:
|
||||
if isinstance(obj, Album) and obj.title == name:
|
||||
artist = Artist(**obj.artist)
|
||||
obj.artist = artist
|
||||
albums.append(obj)
|
||||
return albums
|
||||
|
||||
# a method to get all artists from List[object] where the name equals the parameter name
|
||||
def get_artists_by_name(self, objects: List[object], name: str) -> List[Artist]:
|
||||
l.debug(f"Getting artists by name {name}")
|
||||
artists = []
|
||||
for obj in objects:
|
||||
if isinstance(obj, Artist) and obj.artistName == name:
|
||||
artists.append(obj)
|
||||
return artists
|
||||
|
||||
def create_album(self, album: Album) -> Album:
|
||||
l.debug(f"Creating album {album.title}")
|
||||
json_artist = album.artist.__dict__
|
||||
album.artist = json_artist
|
||||
data = self._post("/api/v1/album", json=album.__dict__)
|
||||
return Album(**data)
|
||||
|
||||
def update_album(self, album_id: int, album: Album) -> Album:
|
||||
l.debug(f"Updating album {album_id}")
|
||||
json_artist = album.artist.__dict__
|
||||
album.artist = json_artist
|
||||
data = self._put(f"/api/v1/album/{album_id}", json=album.__dict__)
|
||||
return Album(**data)
|
||||
|
||||
def create_artist(self, artist: Artist) -> Artist:
|
||||
l.debug(f"Creating artist {artist.artistName}")
|
||||
data = self._post("/api/v1/artist", json=artist.__dict__)
|
||||
return Artist(**data)
|
||||
|
||||
def update_artist(self, artist_id: int, artist: Artist) -> Artist:
|
||||
l.debug(f"Updating artist {artist_id}")
|
||||
data = self._put(f"/api/v1/artist/{artist_id}", json=artist.__dict__)
|
||||
return Artist(**data)
|
||||
|
||||
# shorthand method to set artist to monitored
|
||||
def monitor_artist(self, artist: Artist):
|
||||
artist.monitored = True
|
||||
l.debug(f"Monitoring artist {artist.artistName}")
|
||||
if artist.id == 0:
|
||||
artist = self.create_artist(artist)
|
||||
else:
|
||||
self.update_artist(artist.id, artist)
|
||||
# shorthand method to set album to monitored
|
||||
def monitor_album(self, album: Album):
|
||||
album.monitored = True
|
||||
|
||||
l.debug(f"Monitoring album {album.title}")
|
||||
if album.id == 0:
|
||||
album = self.create_album(album)
|
||||
else:
|
||||
self.update_album(album.id, album)
|
||||
|
||||
# a method to query /api/v1/rootfolder and return a List[RootFolder]
|
||||
def get_root_folders(self) -> List[RootFolder]:
|
||||
l.debug("Getting root folders")
|
||||
data = self._get("/api/v1/rootfolder")
|
||||
return [RootFolder(**folder) for folder in data]
|
||||
|
||||
# a method to query /api/v1/qualityprofile and return a List[QualityProfile]
|
||||
def get_quality_profiles(self) -> List[QualityProfile]:
|
||||
l.debug("Getting quality profiles")
|
||||
data = self._get("/api/v1/qualityprofile")
|
||||
return [QualityProfile(**profile) for profile in data]
|
||||
|
||||
|
||||
Reference in New Issue
Block a user