sources/uSwagger.py - master - VolcanoByte/vbCinema

sources/uSwagger.py
151 строка5 KB
Eugene Gavrilyuk
Sync on start and finish
01 апр 2026, 14:46
01 апр 2026, 14:4641ca630
"""
Набор методов для получения информации о фильме с
https://kinopoiskapiunofficial.tech
"""

import json
from typing import LiteralString

import requests
import os.path

from uDB import db
from uConsts import StoreConsts, COVERS_DIR
from uGlobal import store, MessageLanguageConsts as mlc


class SwaggerException(Exception):
    """Исключения, возникающие при получении информации"""

    def __init__(self, _code: int = 200):
        match _code:
            case 200:
                # Запрос выполнен успешно
                self.message = mlc.swaggerCode200
            case 401:
                # Пустой или неправильный токен
                self.message = mlc.swaggerCode401
            case 402:
                # Превышен лимит запросов(или дневной, или общий)
                self.message = mlc.swaggerCode402
            case 404:
                self.message = mlc.swaggerCode404
            case 429:
                # Слишком много запросов. Общий лимит - 20 запросов в секунду
                self.message = mlc.swaggerCode429
            case _:
                # Иначе: возвращен неизвестный код
                self.message = mlc.swaggerCodeUnknown % _code


def saveJSON(data: json, filename: str):
    """Сохранение полученной информации в формате JSON в файл (для
    отладочных целей)"""
    with open(filename, "w") as json_file:
        json.dump(data, json_file)
        json_file.close()


def getHeaders() -> dict:
    """Подготовка словаря для заголовка запроса"""
    if apiKey := store.value(StoreConsts.OptionsToken, ''):
        return {
            'X-API-KEY': apiKey,
            'Content-Type': 'application/json'
        }
    else:
        raise SwaggerException(401)


def getJSON(url: str) -> json:
    """Получение JSON из запроса"""
    response = requests.get(url, headers=getHeaders())

    if response.status_code != 200:
        raise SwaggerException(response.status_code)

    j = json.loads(response.content)

    return j


def getInfo(_id: str) -> json:
    """Получение общей информации о фильме в JSON"""
    url = f'https://kinopoiskapiunofficial.tech/api/v2.2/films/{_id}'
    j = getJSON(url)

    return j


def getKeyword(keyword: str) -> json:
    """Получение списка фильмов по ключевому слову"""
    url = f'https://kinopoiskapiunofficial.tech/api/v2.1/films/search-by-keyword?keyword={keyword}&page=1'
    j = getJSON(url)

    return j


def getSimilar(idMovie) -> json:
    """Получение списка фильмов, похожих на указанный в id"""
    url = f'https://kinopoiskapiunofficial.tech/api/v2.2/films/{idMovie}/similars'
    j = getJSON(url)

    return j


def getSequelsAndPrequels(idMovie) -> json:
    """Получение списка сиквелов и проиквелов к фильму, указанному в idMovie"""
    url = (f'https://kinopoiskapiunofficial.tech/api/v2.1/films/'
           f'{idMovie}/sequels_and_prequels')
    j = getJSON(url)

    return j


def getStuff(idMovie) -> json:
    """Получение списка актеров и режисеров фильма, указанного в idMovie"""
    url = f'https://kinopoiskapiunofficial.tech/api/v1/staff?filmId={idMovie}'
    j = getJSON(url)

    return j


def getCoversFolder() -> str:
    _base = os.path.dirname(db.filename)

    _covers = os.path.join(_base, COVERS_DIR)
    if not os.path.exists(_covers):
        os.makedirs(_covers)

    return str(_covers)


def getCoverFilename(idMovie):
    if isinstance(idMovie, int):
        _id = str(idMovie)
    else:
        _id = idMovie

    _filename = os.path.join(getCoversFolder(), _id + '.jpg')
    return _filename


def getCover(idMovie: str, url: str = None):
    """Скачивание обложки к фильму.  В параметр url может передаваться
    ссылка на обложку из предварительно полученной информации (getInfo)"""
    filename = getCoverFilename(idMovie)
    if url is None:
        j = getInfo(idMovie)
        _url = j.get('posterUrlPreview', None)
    else:
        _url = url

    if _url is None:
        raise SwaggerException(404)

    if not os.path.exists(filename):
        cover = requests.get(_url)
        if cover.status_code == 200:
            with open(filename, "wb") as cover_file:
                cover_file.write(cover.content)
                cover_file.close()