| """ |
| Created By: ishwor subedi |
| Date: 2024-08-02 |
| """ |
| import base64 |
| import string |
|
|
| import requests |
| import yaml |
| import configparser |
| from src.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 |
|
|