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