transfer_ratings.py

#!/usr/bin/env python
import argparse
import datetime
import logging
import os
import sys
import time
import traceback
from importlib import reload
from typing import Mapping

from RatS.allocine.allocine_ratings_inserter import AlloCineRatingsInserter
from RatS.allocine.allocine_ratings_parser import AlloCineRatingsParser
from RatS.base.base_ratings_inserter import RatingsInserter
from RatS.base.base_ratings_parser import RatingsParser
from RatS.base.movie_entity import Site, Movie
from RatS.base.base_exceptions import (
    RatSException,
    NoMoviesForInsertion,
    NoValidCredentialsException,
)
from RatS.criticker.criticker_ratings_inserter import CritickerRatingsInserter
from RatS.criticker.criticker_ratings_parser import CritickerRatingsParser
from RatS.filmaffinity.filmaffinity_ratings_inserter import FilmAffinityRatingsInserter
from RatS.filmaffinity.filmaffinity_ratings_parser import FilmAffinityRatingsParser
from RatS.filmtipset.filmtipset_ratings_inserter import FilmtipsetRatingsInserter
from RatS.filmtipset.filmtipset_ratings_parser import FilmtipsetRatingsParser
from RatS.icheckmovies.icheckmovies_ratings_inserter import ICheckMoviesRatingsInserter
from RatS.icheckmovies.icheckmovies_ratings_parser import ICheckMoviesRatingsParser
from RatS.imdb.imdb_ratings_inserter import IMDBRatingsInserter
from RatS.imdb.imdb_ratings_parser import IMDBRatingsParser
from RatS.letterboxd.letterboxd_ratings_inserter import LetterboxdRatingsInserter
from RatS.letterboxd.letterboxd_ratings_parser import LetterboxdRatingsParser
from RatS.listal.listal_ratings_inserter import ListalRatingsInserter
from RatS.listal.listal_ratings_parser import ListalRatingsParser
from RatS.metacritic.metacritic_ratings_inserter import MetacriticRatingsInserter
from RatS.movielens.movielens_ratings_inserter import MovielensRatingsInserter
from RatS.movielens.movielens_ratings_parser import MovielensRatingsParser
from RatS.moviepilot.moviepilot_ratings_inserter import MoviePilotRatingsInserter
from RatS.moviepilot.moviepilot_ratings_parser import MoviePilotRatingsParser
from RatS.plex.plex_ratings_inserter import PlexRatingsInserter
from RatS.plex.plex_ratings_parser import PlexRatingsParser
from RatS.rottentomatoes.rottentomatoes_ratings_inserter import RottenTomatoesRatingsInserter
from RatS.rottentomatoes.rottentomatoes_ratings_parser import RottenTomatoesRatingsParser
from RatS.tmdb.tmdb_ratings_inserter import TMDBRatingsInserter
from RatS.tmdb.tmdb_ratings_parser import TMDBRatingsParser
from RatS.trakt.trakt_ratings_inserter import TraktRatingsInserter
from RatS.trakt.trakt_ratings_parser import TraktRatingsParser
from RatS.utils import file_impex
from RatS.utils.version_check import check_version

TIMESTAMP = datetime.datetime.fromtimestamp(time.time()).strftime("%Y%m%d%H%M%S")
EXPORTS_FOLDER = os.path.abspath(os.path.join(os.path.dirname(__file__), "RatS", "exports"))

PARSERS: Mapping[Site, RatingsParser] = {
    Site.ALLOCINE: AlloCineRatingsParser,
    Site.CRITICKER: CritickerRatingsParser,
    Site.FILMAFFINITY: FilmAffinityRatingsParser,
    Site.FILMTIPSET: FilmtipsetRatingsParser,
    # Site.FLIXSTER: FlixsterRatingsParser,
    Site.ICHECKMOVIES: ICheckMoviesRatingsParser,
    Site.IMDB: IMDBRatingsParser,
    Site.LETTERBOXD: LetterboxdRatingsParser,
    Site.LISTAL: ListalRatingsParser,
    Site.MOVIELENS: MovielensRatingsParser,
    Site.MOVIEPILOT: MoviePilotRatingsParser,
    Site.PLEX: PlexRatingsParser,
    Site.ROTTENTOMATOES: RottenTomatoesRatingsParser,
    Site.TMDB: TMDBRatingsParser,
    Site.TRAKT: TraktRatingsParser,
}
INSERTERS: Mapping[Site, RatingsInserter] = {
    Site.ALLOCINE: AlloCineRatingsInserter,
    Site.CRITICKER: CritickerRatingsInserter,
    Site.FILMAFFINITY: FilmAffinityRatingsInserter,
    Site.FILMTIPSET: FilmtipsetRatingsInserter,
    # Site.FLIXSTER: FlixsterRatingsInserter,
    Site.ICHECKMOVIES: ICheckMoviesRatingsInserter,
    Site.IMDB: IMDBRatingsInserter,
    Site.LETTERBOXD: LetterboxdRatingsInserter,
    Site.LISTAL: ListalRatingsInserter,
    Site.METACRITIC: MetacriticRatingsInserter,
    Site.MOVIELENS: MovielensRatingsInserter,
    Site.MOVIEPILOT: MoviePilotRatingsInserter,
    Site.PLEX: PlexRatingsInserter,
    Site.ROTTENTOMATOES: RottenTomatoesRatingsInserter,
    Site.TMDB: TMDBRatingsInserter,
    Site.TRAKT: TraktRatingsInserter,
}


def main():
    args = parse_args()
    init_logging(args)
    check_version()
    try:
        execute(args)
    except RatSException as e:
        logging.error(str(e))
        sys.exit(1)


def parse_args():
    argparser = argparse.ArgumentParser()
    argparser.add_argument(
        "-s",
        "--source",
        help="Source of the movie ratings",
        required=True,
    )
    argparser.add_argument(
        "-d",
        "--destination",
        help="Destination for the ratings",
        required=False,
        action="append",
    )
    argparser.add_argument(
        "-D",
        "--all-destinations",
        help="Try to insert in all available destinations",
        action="store_true",
    )
    argparser.add_argument(
        "-f",
        "--file",
        help="Import ratings from this file instead of parser "
        "(you still have provide the -s/--source argument "
        "to determine which data to use for inserting)",
        type=str,
        required=False,
    )
    argparser.add_argument(
        "--log",
        help="log level",
        type=str,
        default="INFO",
        choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
        required=False,
    )
    argparser.add_argument(
        "-x",
        "--show_browser",
        help="show the browser doing his work",
        action="store_true",
        required=False,
    )
    args = argparser.parse_args()
    return args


def get_parser_from_arg(site: Site) -> RatingsParser:
    try:
        return PARSERS.get(site)
    except (KeyError, ValueError):
        logging.critical(f"No parser matching '{site.name}' found.")
        logging.info(f"Available parsers: {[parser for parser in PARSERS]}")
        sys.exit(1)


def get_inserter_from_arg(site: Site) -> RatingsInserter:
    try:
        return INSERTERS.get(site)
    except (KeyError, ValueError):
        logging.critical(f"No inserter matching '{site.name}' found.")
        logging.info(f"Available inserters: {[inserter for inserter in INSERTERS]}")
        sys.exit(1)


def init_logging(args):
    log_level = getattr(logging, args.log.upper(), None)
    reload(logging)
    logging.basicConfig(
        level=log_level,
        format="[%(asctime)s] %(levelname)s - %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
        stream=sys.stdout,
    )


def execute(args):
    try:
        site: Site = Site(args.source.upper())
        parser: RatingsParser = get_parser_from_arg(site)(args)
        movies: list[Movie] = execute_parsing(args, parser)
        execute_inserting(args, movies, parser)
    except RatSException as e:
        logging.error(str(e))


def execute_inserting(args, movies: list[Movie], parser: RatingsParser):
    if not args.all_destinations and not args.destination:
        return
    destinations: list[Site] = (
        [inserter for inserter in INSERTERS.keys()]
        if args.all_destinations
        else [Site(destination.upper()) for destination in args.destination]
    )
    _filter_source_site_from_destinations(destinations, parser.site.site_name)
    if destinations:
        if len(movies) == 0:
            NoMoviesForInsertion("There are no files to be inserted. Did the parser run properly?")
        # INSERT THE DATA
        for destination in destinations:
            try:
                inserter: RatingsInserter = get_inserter_from_arg(destination)(args)
                insert_movie_ratings(inserter, movies, parser.site.site)
            except RatSException as e:
                logging.error(str(e))


def _filter_source_site_from_destinations(destinations: list[str], parser_site_name: str):
    if parser_site_name.upper() in destinations:
        destinations.remove(parser_site_name.upper())
        logging.warning(f"Will not insert ratings to their source. Skipping {parser_site_name}.")


def execute_parsing(args, parser: RatingsParser) -> list[Movie]:
    if not parser.site.CREDENTIALS_VALID:
        raise NoValidCredentialsException(f"No valid credentials found for {parser.site.site_name}. Skipping parsing.")
    if args.file:
        # LOAD FROM FILE
        movies: list[Movie] = load_data_from_file(args.file)
        parser.site.browser_handler.kill()
    else:
        # PARSE DATA
        movies: list[Movie] = parse_data_from_source(parser)
    return movies


def parse_data_from_source(parser: RatingsParser) -> list[Movie]:
    try:
        movies: list[Movie] = parser.parse()
    except RatSException as e:
        logging.error(str(e))
        sys.exit(1)

    json_filename: str = f"{TIMESTAMP}_{type(parser.site).__name__}.json"
    file_impex.save_movies_to_json(movies, folder=EXPORTS_FOLDER, filename=json_filename)
    logging.info(
        f"===== {parser.site.site_name}: saved {len(movies)} parsed movies to {EXPORTS_FOLDER}/{json_filename}"
    )
    return movies


def load_data_from_file(filename: str) -> list[Movie]:
    movies: list[Movie] = file_impex.load_movies_from_json(folder=EXPORTS_FOLDER, filename=filename)
    logging.info(f"===== loaded {len(movies)} movies from {EXPORTS_FOLDER}/{filename}")
    return movies


def insert_movie_ratings(inserter: RatingsInserter, movies: list[Movie], source: Site):
    if inserter.site.CREDENTIALS_VALID:
        try:
            inserter.insert(movies, source)
        except RatSException as e:
            logging.error(str(e))
        except Exception:  # pylint: disable=broad-except
            inserter.site.browser_handler.kill()
            logging.error(f"There was an exception inside {inserter.site.site_name} (see below). - Skipping insertion.")
            traceback.print_exc()
    else:
        logging.warning(f"No valid credentials found for {inserter.site.site_name}. Skipping insertion.")


if __name__ == "__main__":
    main()