| 1 | import json |
| 2 | import os |
| 3 | import re |
| 4 | import Levenshtein |
| 5 | from FilmAgent_root.FilmAgent.LLMCaller import * |
| 6 | |
| 7 | def read_json(input_path): |
| 8 | with open(input_path, 'r', encoding='utf-8',errors='ignore') as f: |
| 9 | r = toString(json.load(f)) |
| 10 | r = r.replace("�",".") |
| 11 | return json.loads(r) |
| 12 | |
| 13 | |
| 14 | def write_json(output_path, output_data): |
| 15 | with open(output_path, 'w', encoding='utf-8') as f: |
| 16 | json.dump(output_data, f, ensure_ascii=False) |
| 17 | |
| 18 | |
| 19 | def read_prompt(input_path): |
| 20 | with open(input_path, 'r', encoding='utf-8') as f: |
| 21 | return f.read() |
| 22 | |
| 23 | |
| 24 | def log_prompt(prompt_log_path, input): |
| 25 | if not isinstance(input, str): |
| 26 | input = toString(input) |
| 27 | with open(prompt_log_path, "a", encoding="utf-8") as log_file: |
| 28 | log_file.write(f"{input}\n") |
| 29 | log_file.write("#######################################################\n") |
| 30 | |
| 31 | |
| 32 | def cretae_new_path(path, filetype): |
| 33 | if not os.path.exists(path): |
| 34 | os.makedirs(path) |
| 35 | files = os.listdir(path) |
| 36 | if len(files) == 0: |
| 37 | return os.path.join(path, f'0.{filetype}') |
| 38 | else: |
| 39 | max = 0 |
| 40 | for file in files: |
| 41 | max = max if max>=int(os.path.splitext(file)[0]) else int(os.path.splitext(file)[0]) |
| 42 | return os.path.join(path, str(max+1) + f'.{filetype}') |
| 43 | |
| 44 | |
| 45 | def find_latest_file(path): |
| 46 | files = os.listdir(path) |
| 47 | max = 0 |
| 48 | f = "" |
| 49 | for file in files: |
| 50 | if max < int(os.path.splitext(file)[0]): |
| 51 | max = int(os.path.splitext(file)[0]) |
| 52 | f = file |
| 53 | |
| 54 | return os.path.join(path, f) |
| 55 | |
| 56 | |
| 57 | def toString(input): |
| 58 | return json.dumps(input, ensure_ascii=False, separators=(",", ":")) |
| 59 | |
| 60 | |
| 61 | def prompt_format(prompt, params): |
| 62 | text = prompt |
| 63 | for key, value in params.items(): |
| 64 | if isinstance(value, (dict, list)): |
| 65 | value = toString(value) |
| 66 | if isinstance(value, (int, float)): |
| 67 | value = str(value) |
| 68 | text = text.replace(key, value) |
| 69 | return text |
| 70 | |
| 71 | |
| 72 | def GPTResponse2JSON(response): |
| 73 | json_string = response |
| 74 | prompt = f"Modify the following string so that it can be correctly parsed by the json.loads() method:\n{json_string}\n\nYou should just return the modified string." |
| 75 | if "```json" in json_string: |
| 76 | json_string = json_string.replace("```","") |
| 77 | json_string = json_string.replace("json","") |
| 78 | json_string = json_string.strip() |
| 79 | try: |
| 80 | result = json.loads(json_string) |
| 81 | except: |
| 82 | result = json.loads(clean_text(GPTCall(prompt))) |
| 83 | |
| 84 | return result |
| 85 | |
| 86 | |
| 87 | def clean_text(text): |
| 88 | # Only keep json content |
| 89 | pattern = r"```json(.*?)```" |
| 90 | match = re.search(pattern, text, re.DOTALL) |
| 91 | if match: |
| 92 | text = match.group(1) |
| 93 | |
| 94 | # Remove some unexpected characters |
| 95 | punctuation_map = { |
| 96 | "，": ",", |
| 97 | "。": ".", |
| 98 | "！": "!", |
| 99 | "？": "?", |
| 100 | "：": ":", |
| 101 | "；": ";", |
| 102 | "“": "\"", |
| 103 | "”": "\"", |
| 104 | "‘": "\'", |
| 105 | "’": "\'", |
| 106 | "（": "(", |
| 107 | "）": ")", |
| 108 | "【": "[", |
| 109 | "】": "]", |
| 110 | "——": "-", |
| 111 | "…": "...", |
| 112 | "–": "-", |
| 113 | "—": "-", |
| 114 | "�": "." |
| 115 | } |
| 116 | pattern = r'[^a-zA-Z0-9\s\.,!?;:\'"\-({})\[\]]' |
| 117 | for chinese, english in punctuation_map.items(): |
| 118 | text = text.replace(chinese, english) |
| 119 | text = re.sub(pattern, '', text) |
| 120 | text = text.strip() |
| 121 | |
| 122 | return text |
| 123 | |
| 124 | |
| 125 | def get_number(text): |
| 126 | pattern = r'[^0-9]' |
| 127 | text = re.sub(pattern, '', text) |
| 128 | return int(text) |
| 129 | |
| 130 | |
| 131 | def contains_digit(string): |
| 132 | return bool(re.search(r'\d', string)) |
| 133 | |
| 134 | |
| 135 | def translate_digit(string): |
| 136 | prompt = f"Convert the numbers in the following sentence into correct English expressions:\n\n{string}\n\nYour answer should only contain the following JSON content:\n" + '{"Converted-sentence": "..."}' |
| 137 | result = GPTResponse2JSON(GPTCall(prompt)) |
| 138 | return list(result.values())[0] |
| 139 | |
| 140 | |
| 141 | def calculate_similarity(str1, str2): |
| 142 | distance = Levenshtein.distance(str1.lower(), str2.lower()) |
| 143 | max_len = max(len(str1), len(str2)) |
| 144 | similarity = 1 - distance / max_len |
| 145 | return similarity |
| 146 | |
| 147 | |
| 148 | def return_most_similar(string, string_list): |
| 149 | max_similarity = 0 |
| 150 | tgt = 0 |
| 151 | for id,item in enumerate(string_list): |
| 152 | current_similarity = calculate_similarity(string, item) |
| 153 | if current_similarity > max_similarity: |
| 154 | max_similarity = current_similarity |
| 155 | tgt = id |
| 156 | |
| 157 | return string_list[tgt] |
| 158 | |
| 159 | |
| 160 | def GetValueFromDictArray(dictarray, key1, key2, value1): |
| 161 | for item in dictarray: |
| 162 | if item[key1] == value1: |
| 163 | return item[key2] |
| 164 |