starred/MediaManager-maxdorninger-1
1
0
Fork 0
mirror of https://github.com/maxdorninger/MediaManager.git synced 2026-04-17 15:43:28 +02:00
Code Issues Packages Projects Releases Wiki Activity

work on the tv module, automatically gets metadata from tmdb, you can add seasons and shows to the db

Browse Source
This commit is contained in:
maxDorninger
2025-03-01 20:37:30 +01:00
parent 069d7e9236
commit b890b9e8dc
7 changed files with 168 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
MediaManager/src/auth/__init__.py
View File

@@ -1,5 +1,6 @@
import logging
from datetime import datetime, timedelta, timezone
from typing import Annotated
import jwt
from fastapi import Depends, HTTPException, status, APIRouter
@@ -29,12 +30,13 @@ oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/v1/token")
router = APIRouter()
async def get_current_user(token: str = Depends(oauth2_scheme), config = Depends(AuthConfig)) -> UserInternal:
async def get_current_user(token: str = Depends(oauth2_scheme)) -> UserInternal:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
config = AuthConfig()
log.debug("token: " + token)
try:
payload = jwt.decode(token, config.jwt_signing_key, algorithms=[config.jwt_signing_algorithm])
@@ -55,8 +57,9 @@ async def get_current_user(token: str = Depends(oauth2_scheme), config = Depends
return user
def create_access_token(data: dict, expires_delta: timedelta | None = None, config = Depends(AuthConfig)):
def create_access_token(data: dict, expires_delta: timedelta | None = None):
to_encode = data.copy()
config = AuthConfig()
if expires_delta:
expire = datetime.now(timezone.utc) + expires_delta
else:

5
MediaManager/src/config/__init__.py
View File

@@ -16,6 +16,10 @@ class DbConfig(BaseModel):
def password(self):
return self._password
class TvConfig(BaseModel):
api_key: str = os.getenv("TMDB_API_KEY")
class IndexerConfig(BaseModel):
default_indexer: Literal["tmdb"] = os.getenv("INDEXER") or "tmdb"
@@ -44,6 +48,7 @@ def load_config():
log.info(f"loaded config: DbConfig: {DbConfig().__str__()}")
log.info(f"loaded config: IndexerConfig: {IndexerConfig().__str__()}")
log.info(f"loaded config: AuthConfig: {AuthConfig().__str__()}")
log.info(f"loaded config: TvConfig: {TvConfig().__str__()}")
if __name__ == "__main__":

71
MediaManager/src/database/tv.py
View File

@@ -1,63 +1,5 @@
from typing import Literal, List
from uuid import UUID, uuid4
from pydantic import BaseModel
from database import PgDatabase, log
# NOTE: use tmdbsimple for api calls
class Episode(BaseModel):
number: int
title: str
class Season(BaseModel):
number: int
episodes: List[Episode]
def get_episode_count(self)-> int:
return self.episodes.__len__()
class Show(BaseModel):
id: UUID = uuid4()
external_id: int
indexer: Literal["tmdb"]
name: str
seasons: List[Season]
def get_season_count(self)-> int:
return self.seasons.__len__()
def get_episode_count(self) -> int:
episode_count = 0
for season in self.seasons:
episode_count += season.get_episode_count()
return episode_count
def save_show(self) -> None:
with PgDatabase() as db:
db.connection.execute("""
INSERT INTO tv_show (
id,
external_id,
indexer,
name,
episode_count,
season_count
)VALUES(%s,%s,%s,%s,%s,%s);
""",
(self.id,
self.external_id,
self.indexer,
self.name,
self.get_episode_count(),
self.get_season_count(),
)
)
log.info("added show: " + self.__str__())
# TODO: add NOT NULL and default values to DB
def init_table():
@@ -66,10 +8,12 @@ def init_table():
CREATE TABLE IF NOT EXISTS tv_show (
id UUID PRIMARY KEY,
external_id TEXT,
indexer TEXT,
metadata_provider TEXT,
name TEXT,
episode_count INTEGER,
season_count INTEGER
season_count INTEGER,
UNIQUE (external_id, metadata_provider)
);""")
log.info("tv_show Table initialized successfully")
db.connection.execute("""
@@ -78,17 +22,16 @@ def init_table():
season_number INTEGER,
episode_count INTEGER,
CONSTRAINT PK_season PRIMARY KEY (show_id,season_number)
);""")
log.info("tv_seasonTable initialized successfully")
db.connection.execute("""
CREATE TABLE IF NOT EXISTS tv_episode (
season INTEGER,
season_number INTEGER,
show_id uuid,
episode_number INTEGER,
title TEXT,
CONSTRAINT PK_episode PRIMARY KEY (season,show_id,episode_number),
FOREIGN KEY (season, show_id) REFERENCES tv_season(season_number,show_id)
CONSTRAINT PK_episode PRIMARY KEY (season_number,show_id,episode_number),
FOREIGN KEY (season_number, show_id) REFERENCES tv_season(season_number,show_id)
);""")
log.info("tv_episode Table initialized successfully")

2
MediaManager/src/main.py
View File

@@ -6,6 +6,7 @@ from fastapi import FastAPI
import config
import database
import tv.router
from auth import password
from routers import users
@@ -18,6 +19,7 @@ database.init_db()
app = FastAPI(root_path="/api/v1")
app.include_router(users.router, tags=["users"])
app.include_router(password.router, tags=["authentication"])
app.include_router(tv.router.router, tags=["tv"])

0
MediaManager/src/routers/tv.py
View File

106
MediaManager/src/tv/__init__.py Normal file
View File

@@ -0,0 +1,106 @@
import logging
import pprint
from typing import Literal, List, Any
from uuid import UUID, uuid4
import requests
from pydantic import BaseModel
from config import TvConfig
from database import PgDatabase
import tmdbsimple as tmdb
# NOTE: use tmdbsimple for api calls
class Episode(BaseModel):
number: int
title: str
class Season(BaseModel):
number: int
episodes: List[Episode]
def get_episode_count(self) -> int:
return self.episodes.__len__()
class Show(BaseModel):
id: UUID = uuid4()
external_id: int
metadata_provider: str
name: str
seasons: List[Season] = []
def get_season_count(self) -> int:
return self.seasons.__len__()
def get_episode_count(self) -> int:
episode_count = 0
for season in self.seasons:
episode_count += season.get_episode_count()
return episode_count
def save_show(self) -> None:
with PgDatabase() as db:
db.connection.execute("""
INSERT INTO tv_show (
id,
external_id,
metadata_provider,
name,
episode_count,
season_count
)VALUES(%s,%s,%s,%s,%s,%s);
""",
(self.id,
self.external_id,
self.metadata_provider,
self.name,
self.get_episode_count(),
self.get_season_count(),
)
)
log.info("added show: " + self.__str__())
def get_data_from_tmdb(self) -> None:
data = tmdb.TV(self.external_id).info()
log.debug("data from tmdb: " + pprint.pformat(data))
self.name = data["original_name"]
self.metadata_provider = "tmdb"
def add_season(self, season_number: int) -> None:
data = tmdb.TV_Seasons(self.external_id, season_number).info()
log.debug("data from tmdb: " + pprint.pformat(data))
episodes: List[Episode] = []
for episode in data["episodes"]:
episodes.append(Episode(title=episode["name"],number=episode["episode_number"]))
season = Season(number=season_number, episodes=episodes)
self.seasons.append(season)
def add_seasons(self, season_numbers: List[int]) -> None:
for season_number in season_numbers:
self.add_season(season_number)
def get_all_shows() -> List[Show]:
with PgDatabase() as db:
result = db.connection.execute("""
SELECT * FROM tv_show
""").fetchall()
return result
def get_show(id: UUID) -> Show:
with PgDatabase() as db:
result = db.connection.execute("""
SELECT * FROM tv_show WHERE id = %s
""", (id,)).fetchone()
return Show(**result)
config = TvConfig()
log = logging.getLogger(__name__)
tmdb.API_KEY = config.api_key

43
MediaManager/src/tv/router.py Normal file
View File

@@ -0,0 +1,43 @@
from typing import Annotated
from uuid import UUID
import psycopg.errors
from fastapi import APIRouter, Depends
import auth
from database.users import User, UserInternal
from tv import Show, get_all_shows, tmdb, log, get_show
router = APIRouter(
prefix="/tv",
)
@router.post("/show", status_code=201,dependencies=[Depends(auth.get_current_user)])
def post_add_show_route(show_id: int, metadata_provider: str = "tmdb"):
show: Show = Show(external_id=show_id, metadata_provider=metadata_provider, name="temp_name_set_in_post_show_route")
show.get_data_from_tmdb()
try:
show.save_show()
except psycopg.errors.UniqueViolation:
log.info("Show already exists " + show.__str__())
return show
return show
@router.post("/{show_id}/{season}", status_code=201,dependencies=[Depends(auth.get_current_user)])
def post_add_season_route(show_id: UUID, season: int):
show = get_show(show_id)
show.add_season(season)
show.save_show()
return get_show(show_id)
@router.get("/show")
def get_shows_route():
return get_all_shows()
@router.get("/search")
def search_show_route(query: str):
search = tmdb.Search()
return search.tv(query=query)