core/utils/utils.py

import base64
import string

import requests
import yaml
import configparser
from core.models.utls import FollowUps
from langchain_core.output_parsers import JsonOutputParser


def load_config(file_path):
    with open(file_path, 'r') as file:
        config = yaml.safe_load(file)
    return config


def load_ini_config(file_path):
    config = configparser.ConfigParser()
    config.read(file_path)
    return config


def config_read():
    conf = load_ini_config('config.ini')
    return conf


def encode_to_base64(dct: dict):
    for key in dct:
        if type(dct[key]) == str:
            dct[key] = base64.b64encode(dct[key].encode("utf-8", errors="replace")).decode("utf-8", errors="replace")
        elif type(dct[key]) == dict:
            dct[key] = encode_to_base64(dct[key])
    return dct


def decode_base64(dct: dict):
    if type(dct["output"]) == str:
        dct["output"] = base64.b64decode(dct["output"].encode("utf-8", errors="replace")).decode("utf-8",
                                                                                                 errors="replace")
    else:
        for key in dct["output"]:
            dct["output"][key] = base64.b64decode(dct["output"][key].encode("utf-8", errors="replace")).decode("utf-8",
                                                                                                               errors="replace")
    return dct


def get_ip_info(ip: str):
    try:
        response = requests.get(f"https://ipinfo.io/{ip}/json")
        data = response.json()
        return data.get("city", "Unknown")
    except Exception as e:
        return "Unknown"


def json_parser():
    json_parser = JsonOutputParser(pydantic_object=FollowUps)

    return json_parser


def clean_text(text: str):
    text = text.replace("\n", " ")
    text = text.translate(str.maketrans('', '', string.punctuation.replace(".", "")))
    return text