|
import os
|
|
import sys
|
|
import pandas as pd
|
|
import joblib
|
|
from anime_recommender.loggers.logging import logging
|
|
from anime_recommender.exception.exception import AnimeRecommendorException
|
|
from anime_recommender.constant import *
|
|
from huggingface_hub import HfApi, HfFolder
|
|
|
|
def export_data_to_dataframe(dataframe: pd.DataFrame, file_path: str) -> pd.DataFrame:
|
|
"""
|
|
Saves a given Pandas DataFrame to a CSV file.
|
|
|
|
Args:
|
|
dataframe (pd.DataFrame): The DataFrame to be saved.
|
|
file_path (str): The file path where the DataFrame should be stored.
|
|
|
|
Returns:
|
|
pd.DataFrame: The same DataFrame that was saved.
|
|
"""
|
|
try:
|
|
logging.info(f"Saving DataFrame to file: {file_path}")
|
|
dir_path = os.path.dirname(file_path)
|
|
os.makedirs(dir_path, exist_ok=True)
|
|
dataframe.to_csv(file_path, index=False, header=True)
|
|
logging.info(f"DataFrame saved successfully to {file_path}.")
|
|
return dataframe
|
|
except Exception as e:
|
|
logging.error(f"Error saving DataFrame to {file_path}: {e}")
|
|
raise AnimeRecommendorException(e, sys)
|
|
|
|
def load_csv_data(file_path: str) -> pd.DataFrame:
|
|
"""
|
|
Loads a CSV file into a Pandas DataFrame.
|
|
|
|
Args:
|
|
file_path (str): The file path of the CSV file.
|
|
|
|
Returns:
|
|
pd.DataFrame: The loaded DataFrame.
|
|
"""
|
|
try:
|
|
logging.info(f"Loading CSV data from file: {file_path}")
|
|
df = pd.read_csv(file_path)
|
|
logging.info("CSV file loaded successfully.")
|
|
return df
|
|
except Exception as e:
|
|
logging.error(f"Error loading CSV file {file_path}: {e}")
|
|
raise AnimeRecommendorException(e, sys) from e
|
|
|
|
def save_model(model: object, file_path: str) -> None:
|
|
"""
|
|
Saves a machine learning model to a file using joblib.
|
|
|
|
Args:
|
|
model (object): The model object to be saved.
|
|
file_path (str): The file path where the model should be stored.
|
|
"""
|
|
try:
|
|
logging.info("Entered the save_model method.")
|
|
os.makedirs(os.path.dirname(file_path), exist_ok=True)
|
|
with open(file_path, "wb") as file_obj:
|
|
joblib.dump(model, file_obj)
|
|
logging.info(f"Model saved successfully to {file_path}.")
|
|
except Exception as e:
|
|
logging.error(f"Error saving model to {file_path}: {e}")
|
|
raise AnimeRecommendorException(e, sys) from e
|
|
|
|
def load_object(file_path: str) -> object:
|
|
"""
|
|
Loads a model or object from a file using joblib.
|
|
|
|
Args:
|
|
file_path (str): The file path of the saved model.
|
|
|
|
Returns:
|
|
object: The loaded model.
|
|
"""
|
|
try:
|
|
logging.info(f"Attempting to load object from {file_path}")
|
|
if not os.path.exists(file_path):
|
|
error_msg = f"The file: {file_path} does not exist."
|
|
logging.error(error_msg)
|
|
raise Exception(error_msg)
|
|
with open(file_path, "rb") as file_obj:
|
|
logging.info("Object loaded successfully.")
|
|
return joblib.load(file_obj)
|
|
except Exception as e:
|
|
logging.error(f"Error loading object from {file_path}: {e}")
|
|
raise AnimeRecommendorException(e, sys) from e
|
|
|
|
def upload_model_to_huggingface(model_path: str, repo_id: str, filename: str):
|
|
"""
|
|
Uploads a trained model file to the specified Hugging Face repository.
|
|
|
|
Args:
|
|
model_path (str): Local path of the trained model file.
|
|
repo_id (str): Hugging Face repository ID (e.g., 'krishnaveni76/anime-recommendation-models').
|
|
filename (str): Name of the file when uploaded to Hugging Face.
|
|
"""
|
|
try:
|
|
api = HfApi()
|
|
api.upload_file(
|
|
path_or_fileobj=model_path,
|
|
path_in_repo=filename,
|
|
repo_id=repo_id,
|
|
repo_type="model"
|
|
)
|
|
logging.info(f"Successfully uploaded {filename} to Hugging Face: {repo_id}")
|
|
except Exception as e:
|
|
logging.info(f"Error uploading model to Hugging Face: {str(e)}")
|
|
raise AnimeRecommendorException(e, sys) from e |