mirror of
https://github.com/dtomlinson91/plex_delete_media.git
synced 2025-12-21 21:35:45 +00:00
103 lines
3.6 KiB
Python
103 lines
3.6 KiB
Python
import csv
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import re
|
|
from datetime import date
|
|
from typing import List
|
|
|
|
import panaetius
|
|
from arrapi import RadarrAPI, NotFound
|
|
|
|
BASE_URL = "http://10.1.1.50:7878"
|
|
API_KEY = os.environ["RADARR_API_KEY"]
|
|
HEADER = "plex_delete_media"
|
|
DATE_TODAY = date.today().strftime("%Y_%m_%d")
|
|
|
|
|
|
def main() -> None:
|
|
logger = get_logger()
|
|
radarr = RadarrAPI(BASE_URL, API_KEY)
|
|
|
|
all_movies = {re.sub("[^A-Za-z0-9 ]+", "", movie.title): movie for movie in radarr.all_movies()}
|
|
movies_to_delete = [movie[0] for movie in get_csv()]
|
|
|
|
space_saved: int = 0
|
|
movies_deleted = {"movies_deleted": []}
|
|
movies_not_found = {"movies_not_found": []}
|
|
|
|
_deleted_count = 0
|
|
|
|
for movie_to_delete in movies_to_delete:
|
|
movie_to_delete = re.sub("[^A-Za-z0-9 ]+", "", movie_to_delete)
|
|
if movie_to_delete in all_movies:
|
|
try:
|
|
all_movies[movie_to_delete].delete(deleteFiles=True)
|
|
except NotFound:
|
|
movies_not_found["movies_not_found"].append({"title": movie_to_delete})
|
|
movie_object = {
|
|
"title": movie_to_delete,
|
|
"year": all_movies[movie_to_delete].year,
|
|
"path": all_movies[movie_to_delete].path,
|
|
"size_on_disk_bytes": all_movies[movie_to_delete].sizeOnDisk,
|
|
"size_on_disk_gb": float(round(all_movies[movie_to_delete].sizeOnDisk / pow(10, 9), 2)),
|
|
}
|
|
_deleted_count += 1
|
|
space_saved += all_movies[movie_to_delete].sizeOnDisk
|
|
movies_deleted["movies_deleted"].append(movie_object)
|
|
logger.info(f"Deleted: {movie_object}") # noqa
|
|
else:
|
|
movies_not_found["movies_not_found"].append({"title": movie_to_delete})
|
|
|
|
_space_saved_gb = int(round(space_saved / pow(10, 9), 0))
|
|
|
|
logger.info("Movies deleted: %s", _deleted_count)
|
|
logger.info("Space saved: %sGB", _space_saved_gb)
|
|
logger.info("%s movies could not be found.", len(movies_not_found["movies_not_found"]))
|
|
|
|
summary = {
|
|
"_summary": {"movies_deleted": _deleted_count, "space_saved_gb": _space_saved_gb, "date_deleted": DATE_TODAY}
|
|
}
|
|
movies_deleted = {**summary, **movies_deleted}
|
|
save_results(movies_deleted, movies_not_found)
|
|
|
|
|
|
def get_logger() -> None:
|
|
logging_dir = pathlib.Path(__file__).parents[0] / "logs"
|
|
logging_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
config = panaetius.Config("plex_delete_media", skip_header_init=True)
|
|
panaetius.set_config(config, "logging.level", "DEBUG")
|
|
panaetius.set_config(config, "logging.path", logging_dir)
|
|
|
|
return panaetius.set_logger(config, panaetius.SimpleLogger(logging_level=config.logging_level))
|
|
|
|
|
|
def save_results(movies_deleted: List, movies_not_found: List) -> None:
|
|
results_dir = pathlib.Path(__file__).parents[0] / "results" / DATE_TODAY
|
|
results_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
movies_deleted_file = (results_dir / "movies_deleted").with_suffix(".json")
|
|
movies_not_found_file = (results_dir / "movies_not_found").with_suffix(".json")
|
|
|
|
with movies_deleted_file.open("w") as movies_deleted_file_contents:
|
|
json.dump(movies_deleted, movies_deleted_file_contents, indent=2)
|
|
|
|
with movies_not_found_file.open("w") as movies_not_found_file_contents:
|
|
json.dump(movies_not_found, movies_not_found_file_contents, indent=2)
|
|
|
|
|
|
def get_csv() -> List:
|
|
_all_movies = []
|
|
with open(
|
|
pathlib.Path(__file__).parents[0] / "data" / "movies_to_delete.csv", mode="r", encoding="utf-8"
|
|
) as csv_file:
|
|
reader = csv.reader(csv_file)
|
|
for row in reader:
|
|
_all_movies.append(row)
|
|
return _all_movies
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|