api.py - master - netogithubik/pyapi-diplom-basic

api.py
226 строк8 KB
rskuro
add comments/ try-exepts / corrected the code to comply with PEP8
11 фев 2026, 18:55
11 фев 2026, 18:5596794cd
import requests
from pathlib import Path
from pprint import pprint
import io


class DogClient:
    BASE_DOG_URL = "https://dog.ceo/api/breed/"
    BASE_DISK_URL = "https://cloud-api.yandex.net/v1/"
    RESPONSE_CODES = {
        "OK_200": 200,
        "OK_202": 202,
        "CREATED": 201,
        "AlREADY_EXISTS": 409
    }

    def __init__(self):
        self.headers = None

    # Получаем ссылку на фото собаки
    def get_dog_photo(self, breed: str, sub_breed: str = ""):
        path = f"{breed}/{sub_breed}" if sub_breed != "" else breed
        url = f"{self.BASE_DOG_URL}{path}/images/random"
        try:
            response = requests.get(url)
            response_json = response.json()
            response.raise_for_status()
            if response_json["status"] == "success":
                return response_json["message"]
        except requests.exceptions.HTTPError:
            raise requests.exceptions.HTTPError(
                f"""Файл по адресу: {response.url} не найден.
Проверьте путь или существование объекта."""
            )

    # Создаём папку
    def create_folder(self, breed: str):
        url = f"{self.BASE_DISK_URL}disk/resources"
        headers = self.headers
        params = {"path": breed}

        try:
            response = requests.put(
                url,
                headers=headers,
                params=params,
            )
            if response.status_code not in (
                self.RESPONSE_CODES["CREATED"],
                self.RESPONSE_CODES["AlREADY_EXISTS"]
            ):
                response.raise_for_status()
            if response.status_code == self.RESPONSE_CODES["CREATED"]:
                print(f"Создаём папку {breed}")
        except requests.exceptions.HTTPError:
            raise requests.exceptions.HTTPError(
                f"Ошибка при создании папки {breed}"
            )

    # Загружаем фото собаки в ЯДиск
    def upload_dog_photo(
        self,
        photo_url: str,
        breed: str,
        sub_breed: str = ""
    ):
        file_name = photo_url.split("/")[-1]
        path = (
            f"{breed}/{breed}_{sub_breed}_{file_name}"
            if sub_breed != ""
            else f"{breed}/{breed}_{file_name}"
        )
        is_file_already_exists = self.is_file_already_exists(path)

        if not is_file_already_exists:
            self.create_folder(breed)
            yandex_url = f"{self.BASE_DISK_URL}disk/resources/upload"
            params = {
                "url": photo_url,
                "path": path,
                }

            try:
                response = requests.post(
                    yandex_url,
                    headers=self.headers,
                    params=params,
                )
                if response.status_code == self.RESPONSE_CODES["OK_202"]:
                    response_json = response.json()
                    if "href" in response_json:
                        self.operation_checker(
                            response_json["href"],
                            f"{path.split("/")[1]}",
                            path
                        )
                else:
                    raise requests.exceptions.HTTPError
            except requests.exceptions.HTTPError as e:
                print(e)
        else:
            print(f"Файл {file_name} не был загружен, так как уже существует")

    # Получаем все ссылки на фото собаки, включая под_породы
    def get_dog_photo_urls(self, breed: str):
        try:
            response = requests.get(f"{self.BASE_DOG_URL}{breed}/list")
            response_json = response.json()
            if (response_json["status"] == "success" and
                    len(response_json["message"]) > 0):
                sub_breeds = response_json["message"]
                for sub_breed in sub_breeds:
                    url = self.get_dog_photo(breed, sub_breed)
                    self.upload_dog_photo(url, breed, sub_breed)
            elif (response_json["status"] == "success" and
                    len(response_json["message"]) == 0):
                url = self.get_dog_photo(breed)
                self.upload_dog_photo(url, breed)
            else:
                response.raise_for_status()
        except requests.exceptions.HTTPError:
            raise requests.exceptions.HTTPError(
                f"""Ошибка при загрузке списка под-пород.
Порода {breed} не найдена."""
            )

    # Проверяем статус операции и выводим логи
    def operation_checker(
        self,
        operation_url: str,
        file_name: str,
        file_path: str
    ):
        operation_id = (
            operation_url.split("operations/")[1]
            if "operations/" in operation_url
            else operation_url
        )
        url = f"https://cloud-api.yandex.net/v1/disk/operations/{operation_id}"
        try:
            response_pre = requests.get(url, headers=self.headers)
            response = response_pre.json()
            print(f"\nЗагружаем файл {file_name}")
            if response["status"] == "failed":
                print(response)
                raise Exception
            while (response["status"] != "success"):
                if response["status"] == "in-progress":
                    print("Загрузка...")
                    response = requests.get(url, headers=self.headers).json()
                elif response["status"] == "failed":
                    print(response)
                    raise Exception
            print(f"Файл {file_name} загружен")
            print('--')
            self.get_file_info(file_path)
        except Exception as e:
            print(e)

    # Получаем инфу о файле(когда загрузка завершена)
    def get_file_info(self, file_path: str):
        url = f"{self.BASE_DISK_URL}disk/resources"
        params = {
            "path": file_path,
        }
        response = requests.get(url, headers=self.headers, params=params)
        self.upload_json(response.text, url, file_path)

    # Загружаем json файл с инфой
    def upload_json(self, json_str: str, url: str, file_path: str,):
        path = f"{file_path.split(".")[0]}.json"
        file_in_memory = io.BytesIO()
        file_in_memory.write(json_str.encode('utf-8'))
        file_in_memory.seek(0)
        upload_url = f"{url}/upload"
        params = {
            "path": path,
        }
        files = {
            'file': (
                'uploaded_file.json',
                file_in_memory,
                'application/json'
            )
        }
        try:
            print(f"Создаём файл {path.split('/')[1]}")
            response = requests.get(
                upload_url,
                params=params,
                headers=self.headers
            ).json()
            if "href" in response:
                requests.put(
                    response["href"],
                    params=params,
                    headers=self.headers,
                    files=files
                )
                print(f"Файл {path.split('/')[1]} создан")
            else:
                response.raise_for_status()
        except requests.exceptions.HTTPError:
            raise requests.exceptions.HTTPError(
                f"Ошибка при загрузке json файла {path.split('/')[1]}"
            )

    # Проверяем, существует ли уже файл
    def is_file_already_exists(self, file_path: str) -> bool:
        url = f"{self.BASE_DISK_URL}disk/resources"
        params = {
            "path": file_path,
        }
        try:
            response = requests.get(url, headers=self.headers, params=params)
            return response.status_code == self.RESPONSE_CODES["OK_200"]
        except requests.RequestException:
            return False

    # Точка входа для юзера
    def save_photo(self, token: str, breed: str):
        if not token or not breed:
            raise ValueError("Token и breed обязательны")
        self.yandex_token = token
        self.headers = {"Authorization": f"OAuth {self.yandex_token}"}
        self.get_dog_photo_urls(breed)