price-builder-backend/services/service.py

import subprocess
import pandas as pd
import joblib
import json
import os
from typing import List, Dict
from schemas.schemas import LaptopCreate, LaptopResponse, PredictPriceResponse

class LaptopService:
    def __init__(self, model_path: str, feature_columns_path: str, poly_path: str, scaler_path: str):
        self.script_path = "services/ml/scripts/modelBuilders/modelBuilderLaptop.py"

        # Проверка наличия модели, если её нет — создание
        if not os.path.exists(model_path) or not os.path.exists(feature_columns_path) or not os.path.exists(poly_path) or not os.path.exists(scaler_path):
            print("Необходимые файлы модели отсутствуют. Запускаем построение модели...")
            self.run_model_builder()

        # Загрузка модели и связанных файлов
        try:
            self.model = joblib.load(model_path)
        except FileNotFoundError:
            raise Exception(f"Model file not found at {model_path}")
        except Exception as e:
            raise Exception(f"Error loading model: {str(e)}")

        try:
            self.feature_columns = joblib.load(feature_columns_path)
        except FileNotFoundError:
            raise Exception(f"Feature columns file not found at {feature_columns_path}")
        except Exception as e:
            raise Exception(f"Error loading feature columns: {str(e)}")

        try:
            self.poly_transformer = joblib.load(poly_path)
            self.scaler = joblib.load(scaler_path)
        except FileNotFoundError:
            raise Exception("Polynomial transformer or scaler file not found.")
        except Exception as e:
            raise Exception(f"Error loading polynomial transformer or scaler: {str(e)}")

    def run_model_builder(self):
        # Убедитесь, что путь к скрипту корректен
        if not os.path.exists(self.script_path):
            raise FileNotFoundError(f"Скрипт {self.script_path} не найден.")

        try:
            print(f"Запускаем скрипт {self.script_path} для создания модели ноутбуков...")
            result = subprocess.run(
                ['python', self.script_path],
                stdout=subprocess.PIPE,  # Перенаправляем stdout
                stderr=subprocess.PIPE,  # Перенаправляем stderr
                text=True  # Декодируем вывод в текст
            )

            if result.returncode != 0:
                raise Exception(f"Ошибка выполнения скрипта: {result.stderr}")
            else:
                print("Модель успешно создана.")
                print(result.stdout)

        except Exception as e:
            raise Exception(f"Не удалось выполнить скрипт с ноутбуками: {str(e)}")

    def predict_price(self, data: Dict[str, any]) -> PredictPriceResponse:
        # Преобразование данных в DataFrame
        input_df = pd.DataFrame([data])

        print("До One-Hot Encoding:")
        print(input_df.head())
        print("Колонки:", input_df.columns)

        # Применение One-Hot Encoding к категориальным признакам
        input_df = pd.get_dummies(input_df, columns=['processor', 'os', 'resolution', 'gpu', 'matrix_type'], drop_first=False)

        print("После One-Hot Encoding:")
        print(input_df.head())
        print("Колонки:", input_df.columns)

        # Добавление отсутствующих признаков
        for col in self.feature_columns:
            if col not in input_df.columns and col != 'price':
                input_df[col] = 0

        # Упорядочивание колонок
        input_df = input_df[self.feature_columns]

        # Преобразование с использованием PolynomialFeatures
        input_poly = self.poly_transformer.transform(input_df)

        # Масштабирование данных
        input_scaled = self.scaler.transform(input_poly)

        # Предсказание цены
        predicted_price = self.model.predict(input_scaled)[0]

        return PredictPriceResponse(predicted_price=round(predicted_price, 2))

    def get_unique_data(self):
        # Указываем путь к файлу
        file_path = 'services/ml/scripts/modelBuilders/columns/unique_values_laptop.json'

        # Открываем и читаем данные из файла
        with open(file_path, 'r', encoding='utf-8') as file:
            data = json.load(file)  # Загружаем данные из JSON

        # Возвращаем данные, которые будут переданы в ответ
        return data


class TVService:
    def __init__(self, model_path: str, feature_columns_path: str, poly_path: str, scaler_path: str):
        self.script_path = "services/ml/scripts/modelBuilders/modelBuilderTV.py"

        # Проверка наличия модели, если её нет — создание
        if not os.path.exists(model_path) or not os.path.exists(feature_columns_path) or not os.path.exists(
                poly_path) or not os.path.exists(scaler_path):
            print("Необходимые файлы модели отсутствуют. Запускаем построение модели...")
            self.run_model_builder()

        try:
            self.model = joblib.load(model_path)
        except FileNotFoundError:
            raise Exception(f"Model file not found at {model_path}")
        except Exception as e:
            raise Exception(f"Error loading model: {str(e)}")

        try:
            self.feature_columns = joblib.load(feature_columns_path)
        except FileNotFoundError:
            raise Exception(f"Feature columns file not found at {feature_columns_path}")
        except Exception as e:
            raise Exception(f"Error loading feature columns: {str(e)}")

        try:
            self.poly_transformer = joblib.load(poly_path)
            self.scaler = joblib.load(scaler_path)
        except FileNotFoundError:
            raise Exception("Polynomial transformer or scaler file not found.")
        except Exception as e:
            raise Exception(f"Error loading polynomial transformer or scaler: {str(e)}")

    def predict_price(self, data: Dict[str, any]) -> PredictPriceResponse:
        input_df = pd.DataFrame([data])

        print("До One-Hot Encoding:")
        print(input_df.head())
        print("Колонки:", input_df.columns)

        # Применение One-Hot Encoding
        input_df = pd.get_dummies(input_df,
                                  columns=['display', 'tuners', 'features', 'os', 'color', 'power_of_volume'],
                                  drop_first=False)

        # Преобразование булевых значений в числовые
        input_df = input_df.astype(int)

        print("После One-Hot Encoding:")
        print(input_df.head())
        print("Колонки:", input_df.columns)

        # Добавление отсутствующих признаков
        missing_columns = [col for col in self.feature_columns if col not in input_df.columns and col != 'price']
        missing_df = pd.DataFrame(0, index=input_df.index, columns=missing_columns)
        input_df = pd.concat([input_df, missing_df], axis=1)

        # Упорядочение столбцов
        input_df = input_df.reindex(columns=self.feature_columns, fill_value=0)

        # Полиномиальные и масштабированные данные
        input_poly = self.poly_transformer.transform(input_df)
        input_scaled = self.scaler.transform(input_poly)

        # Предсказание
        predicted_price = self.model.predict(input_scaled)[0]
        return PredictPriceResponse(predicted_price=round(predicted_price, 2))

    def run_model_builder(self):
        # Убедитесь, что путь к скрипту корректен
        if not os.path.exists(self.script_path):
            raise FileNotFoundError(f"Скрипт {self.script_path} не найден.")

        try:
            print(f"Запускаем скрипт {self.script_path} для создания модели телевизоров...")
            result = subprocess.run(
                ['python', self.script_path],
                stdout=subprocess.PIPE,  # Перенаправляем stdout
                stderr=subprocess.PIPE,  # Перенаправляем stderr
                text=True  # Декодируем вывод в текст
            )

            if result.returncode != 0:
                raise Exception(f"Ошибка выполнения скрипта: {result.stderr}")
            else:
                print("Модель успешно создана.")
                print(result.stdout)

        except Exception as e:
            raise Exception(f"Не удалось выполнить скрипт с телевизорами: {str(e)}")

    def get_unique_data(self):
        # Указываем путь к файлу
        file_path = 'services/ml/scripts/modelBuilders/columns/unique_values_tv.json'

        # Открываем и читаем данные из файла
        with open(file_path, 'r', encoding='utf-8') as file:
            data = json.load(file)  # Загружаем данные из JSON

        # Возвращаем данные, которые будут переданы в ответ
        return data