Alexzf01/AAAChatGPT

1# -*- coding:utf-8 -*-2from __future__ import annotations3from typing import TYPE_CHECKING, Any, Callable, Dict, List, Tuple, Type4import logging5import json6import os7import datetime8import hashlib9import csv10import requests11import re12import html13import sys14import subprocess15 16import gradio as gr17from pypinyin import lazy_pinyin18import tiktoken19import mdtex2html20from markdown import markdown21from pygments import highlight22from pygments.lexers import get_lexer_by_name23from pygments.formatters import HtmlFormatter24import pandas as pd25 26from modules.presets import *27from . import shared28from modules.config import retrieve_proxy29 30if TYPE_CHECKING:31    from typing import TypedDict32 33    class DataframeData(TypedDict):34        headers: List[str]35        data: List[List[str | int | bool]]36 37 38def count_token(message):39    encoding = tiktoken.get_encoding("cl100k_base")40    input_str = f"role: {message['role']}, content: {message['content']}"41    length = len(encoding.encode(input_str))42    return length43 44 45def markdown_to_html_with_syntax_highlight(md_str):46    def replacer(match):47        lang = match.group(1) or "text"48        code = match.group(2)49 50        try:51            lexer = get_lexer_by_name(lang, stripall=True)52        except ValueError:53            lexer = get_lexer_by_name("text", stripall=True)54 55        formatter = HtmlFormatter()56        highlighted_code = highlight(code, lexer, formatter)57 58        return f'<pre><code class="{lang}">{highlighted_code}</code></pre>'59 60    code_block_pattern = r"```(\w+)?\n([\s\S]+?)\n```"61    md_str = re.sub(code_block_pattern, replacer, md_str, flags=re.MULTILINE)62 63    html_str = markdown(md_str)64    return html_str65 66 67def normalize_markdown(md_text: str) -> str:68    lines = md_text.split("\n")69    normalized_lines = []70    inside_list = False71 72    for i, line in enumerate(lines):73        if re.match(r"^(\d+\.|-|\*|\+)\s", line.strip()):74            if not inside_list and i > 0 and lines[i - 1].strip() != "":75                normalized_lines.append("")76            inside_list = True77            normalized_lines.append(line)78        elif inside_list and line.strip() == "":79            if i < len(lines) - 1 and not re.match(80                r"^(\d+\.|-|\*|\+)\s", lines[i + 1].strip()81            ):82                normalized_lines.append(line)83            continue84        else:85            inside_list = False86            normalized_lines.append(line)87 88    return "\n".join(normalized_lines)89 90 91def convert_mdtext(md_text):92    code_block_pattern = re.compile(r"```(.*?)(?:```|$)", re.DOTALL)93    inline_code_pattern = re.compile(r"`(.*?)`", re.DOTALL)94    code_blocks = code_block_pattern.findall(md_text)95    non_code_parts = code_block_pattern.split(md_text)[::2]96 97    result = []98    for non_code, code in zip(non_code_parts, code_blocks + [""]):99        if non_code.strip():100            non_code = normalize_markdown(non_code)101            if inline_code_pattern.search(non_code):102                result.append(markdown(non_code, extensions=["tables"]))103            else:104                result.append(mdtex2html.convert(non_code, extensions=["tables"]))105        if code.strip():106            # _, code = detect_language(code)  # 暂时去除代码高亮功能，因为在大段代码的情况下会出现问题107            # code = code.replace("\n\n", "\n") # 暂时去除代码中的空行，因为在大段代码的情况下会出现问题108            code = f"\n```{code}\n\n```"109            code = markdown_to_html_with_syntax_highlight(code)110            result.append(code)111    result = "".join(result)112    result += ALREADY_CONVERTED_MARK113    return result114 115 116def convert_asis(userinput):117    return (118        f'<p style="white-space:pre-wrap;">{html.escape(userinput)}</p>'119        + ALREADY_CONVERTED_MARK120    )121 122 123def detect_converted_mark(userinput):124    if userinput.endswith(ALREADY_CONVERTED_MARK):125        return True126    else:127        return False128 129 130def detect_language(code):131    if code.startswith("\n"):132        first_line = ""133    else:134        first_line = code.strip().split("\n", 1)[0]135    language = first_line.lower() if first_line else ""136    code_without_language = code[len(first_line) :].lstrip() if first_line else code137    return language, code_without_language138 139 140def construct_text(role, text):141    return {"role": role, "content": text}142 143 144def construct_user(text):145    return construct_text("user", text)146 147 148def construct_system(text):149    return construct_text("system", text)150 151 152def construct_assistant(text):153    return construct_text("assistant", text)154 155 156def construct_token_message(tokens: List[int]):157    token_sum = 0158    for i in range(len(tokens)):159        token_sum += sum(tokens[: i + 1])160    return f"Token 计数: {sum(tokens)}，本次对话累计消耗了 {token_sum} tokens"161 162 163def delete_first_conversation(history, previous_token_count):164    if history:165        del history[:2]166        del previous_token_count[0]167    return (168        history,169        previous_token_count,170        construct_token_message(previous_token_count),171    )172 173 174def delete_last_conversation(chatbot, history, previous_token_count):175    if len(chatbot) > 0 and standard_error_msg in chatbot[-1][1]:176        logging.info("由于包含报错信息，只删除chatbot记录")177        chatbot.pop()178        return chatbot, history179    if len(history) > 0:180        logging.info("删除了一组对话历史")181        history.pop()182        history.pop()183    if len(chatbot) > 0:184        logging.info("删除了一组chatbot对话")185        chatbot.pop()186    if len(previous_token_count) > 0:187        logging.info("删除了一组对话的token计数记录")188        previous_token_count.pop()189    return (190        chatbot,191        history,192        previous_token_count,193        construct_token_message(previous_token_count),194    )195 196 197def save_file(filename, system, history, chatbot, user_name):198    logging.info(f"{user_name} 保存对话历史中……")199    os.makedirs(HISTORY_DIR / user_name, exist_ok=True)200    if filename.endswith(".json"):201        json_s = {"system": system, "history": history, "chatbot": chatbot}202        print(json_s)203        with open(os.path.join(HISTORY_DIR / user_name, filename), "w") as f:204            json.dump(json_s, f)205    elif filename.endswith(".md"):206        md_s = f"system: \n- {system} \n"207        for data in history:208            md_s += f"\n{data['role']}: \n- {data['content']} \n"209        with open(os.path.join(HISTORY_DIR / user_name, filename), "w", encoding="utf8") as f:210            f.write(md_s)211    logging.info(f"{user_name} 保存对话历史完毕")212    return os.path.join(HISTORY_DIR / user_name, filename)213 214 215def save_chat_history(filename, system, history, chatbot, user_name):216    if filename == "":217        return218    if not filename.endswith(".json"):219        filename += ".json"220    return save_file(filename, system, history, chatbot, user_name)221 222 223def export_markdown(filename, system, history, chatbot, user_name):224    if filename == "":225        return226    if not filename.endswith(".md"):227        filename += ".md"228    return save_file(filename, system, history, chatbot, user_name)229 230 231def load_chat_history(filename, system, history, chatbot, user_name):232    logging.info(f"{user_name} 加载对话历史中……")233    if type(filename) != str:234        filename = filename.name235    try:236        with open(os.path.join(HISTORY_DIR / user_name, filename), "r") as f:237            json_s = json.load(f)238        try:239            if type(json_s["history"][0]) == str:240                logging.info("历史记录格式为旧版，正在转换……")241                new_history = []242                for index, item in enumerate(json_s["history"]):243                    if index % 2 == 0:244                        new_history.append(construct_user(item))245                    else:246                        new_history.append(construct_assistant(item))247                json_s["history"] = new_history248                logging.info(new_history)249        except:250            # 没有对话历史251            pass252        logging.info(f"{user_name} 加载对话历史完毕")253        return filename, json_s["system"], json_s["history"], json_s["chatbot"]254    except FileNotFoundError:255        logging.info(f"{user_name} 没有找到对话历史文件，不执行任何操作")256        return filename, system, history, chatbot257 258 259def sorted_by_pinyin(list):260    return sorted(list, key=lambda char: lazy_pinyin(char)[0][0])261 262 263def get_file_names(dir, plain=False, filetypes=[".json"]):264    logging.info(f"获取文件名列表，目录为{dir}，文件类型为{filetypes}，是否为纯文本列表{plain}")265    files = []266    try:267        for type in filetypes:268            files += [f for f in os.listdir(dir) if f.endswith(type)]269    except FileNotFoundError:270        files = []271    files = sorted_by_pinyin(files)272    if files == []:273        files = [""]274    logging.debug(f"files are:{files}")275    if plain:276        return files277    else:278        return gr.Dropdown.update(choices=files)279 280 281def get_history_names(plain=False, user_name=""):282    logging.info(f"从用户 {user_name} 中获取历史记录文件名列表")283    return get_file_names(HISTORY_DIR / user_name, plain)284 285 286def load_template(filename, mode=0):287    logging.info(f"加载模板文件{filename}，模式为{mode}（0为返回字典和下拉菜单，1为返回下拉菜单，2为返回字典）")288    lines = []289    logging.info("Loading template...")290    if filename.endswith(".json"):291        with open(os.path.join(TEMPLATES_DIR, filename), "r", encoding="utf8") as f:292            lines = json.load(f)293        lines = [[i["act"], i["prompt"]] for i in lines]294    else:295        with open(296            os.path.join(TEMPLATES_DIR, filename), "r", encoding="utf8"297        ) as csvfile:298            reader = csv.reader(csvfile)299            lines = list(reader)300        lines = lines[1:]301    if mode == 1:302        return sorted_by_pinyin([row[0] for row in lines])303    elif mode == 2:304        return {row[0]: row[1] for row in lines}305    else:306        choices = sorted_by_pinyin([row[0] for row in lines])307        return {row[0]: row[1] for row in lines}, gr.Dropdown.update(308            choices=choices309        )310 311 312def get_template_names(plain=False):313    logging.info("获取模板文件名列表")314    return get_file_names(TEMPLATES_DIR, plain, filetypes=[".csv", "json"])315 316 317def get_template_content(templates, selection, original_system_prompt):318    logging.info(f"应用模板中，选择为{selection}，原始系统提示为{original_system_prompt}")319    try:320        return templates[selection]321    except:322        return original_system_prompt323 324 325def reset_state():326    logging.info("重置状态")327    return [], [], [], construct_token_message([0])328 329 330def reset_textbox():331    logging.debug("重置文本框")332    return gr.update(value="")333 334 335def reset_default():336    default_host = shared.state.reset_api_host()337    retrieve_proxy("")338    return gr.update(value=default_host), gr.update(value=""), "API-Host 和代理已重置"339 340 341def change_api_host(host):342    shared.state.set_api_host(host)343    msg = f"API-Host更改为了{host}"344    logging.info(msg)345    return msg346 347 348def change_proxy(proxy):349    retrieve_proxy(proxy)350    os.environ["HTTPS_PROXY"] = proxy351    msg = f"代理更改为了{proxy}"352    logging.info(msg)353    return msg354 355 356def hide_middle_chars(s):357    if s is None:358        return ""359    if len(s) <= 8:360        return s361    else:362        head = s[:4]363        tail = s[-4:]364        hidden = "*" * (len(s) - 8)365        return head + hidden + tail366 367 368def submit_key(key):369    key = key.strip()370    msg = f"API密钥更改为了{hide_middle_chars(key)}"371    logging.info(msg)372    return key, msg373 374 375def replace_today(prompt):376    today = datetime.datetime.today().strftime("%Y-%m-%d")377    return prompt.replace("{current_date}", today)378 379 380def get_geoip():381    try:382        with retrieve_proxy():383            response = requests.get("https://ipapi.co/json/", timeout=5)384        data = response.json()385    except:386        data = {"error": True, "reason": "连接ipapi失败"}387    if "error" in data.keys():388        logging.warning(f"无法获取IP地址信息。\n{data}")389        if data["reason"] == "RateLimited":390            return (391                f"获取IP地理位置失败，因为达到了检测IP的速率限制。聊天功能可能仍然可用。"392            )393        else:394            return f"获取IP地理位置失败。原因：{data['reason']}。你仍然可以使用聊天功能。"395    else:396        country = data["country_name"]397        if country == "China":398            text = "**您的IP区域：中国。请立即检查代理设置，在不受支持的地区使用API可能导致账号被封禁。**"399        else:400            text = f"您的IP区域：{country}。"401        logging.info(text)402        return text403 404 405def find_n(lst, max_num):406    n = len(lst)407    total = sum(lst)408 409    if total < max_num:410        return n411 412    for i in range(len(lst)):413        if total - lst[i] < max_num:414            return n - i - 1415        total = total - lst[i]416    return 1417 418 419def start_outputing():420    logging.debug("显示取消按钮，隐藏发送按钮")421    return gr.Button.update(visible=True), gr.Button.update(visible=False)422 423 424def end_outputing():425    return (426        gr.Button.update(visible=True),427        gr.Button.update(visible=False),428    )429 430 431def cancel_outputing():432    logging.info("中止输出……")433    shared.state.interrupt()434 435 436def transfer_input(inputs):437    # 一次性返回，降低延迟438    textbox = reset_textbox()439    outputing = start_outputing()440    return (441        inputs,442        gr.update(value=""),443        gr.Button.update(visible=True),444        gr.Button.update(visible=False),445    )446 447 448 449def run(command, desc=None, errdesc=None, custom_env=None, live=False):450    if desc is not None:451        print(desc)452    if live:453        result = subprocess.run(command, shell=True, env=os.environ if custom_env is None else custom_env)454        if result.returncode != 0:455            raise RuntimeError(f"""{errdesc or 'Error running command'}.456Command: {command}457Error code: {result.returncode}""")458 459        return ""460    result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, env=os.environ if custom_env is None else custom_env)461    if result.returncode != 0:462        message = f"""{errdesc or 'Error running command'}.463Command: {command}464Error code: {result.returncode}465stdout: {result.stdout.decode(encoding="utf8", errors="ignore") if len(result.stdout)>0 else '<empty>'}466stderr: {result.stderr.decode(encoding="utf8", errors="ignore") if len(result.stderr)>0 else '<empty>'}467"""468        raise RuntimeError(message)469    return result.stdout.decode(encoding="utf8", errors="ignore")470 471def versions_html():472    git = os.environ.get('GIT', "git")473    python_version = ".".join([str(x) for x in sys.version_info[0:3]])474    try:475        commit_hash = run(f"{git} rev-parse HEAD").strip()476    except Exception:477        commit_hash = "<none>"478    if commit_hash != "<none>":479        short_commit = commit_hash[0:7]480        commit_info = f"<a style=\"text-decoration:none\" href=\"https://github.com/GaiZhenbiao/ChuanhuChatGPT/commit/{short_commit}\">{short_commit}</a>"481    else:482        commit_info = "unknown \U0001F615"483    return f"""484Python: <span title="{sys.version}">{python_version}</span>485 • 486Gradio: {gr.__version__}487 • 488Commit: {commit_info}489"""490 491def add_source_numbers(lst, source_name = "Source", use_source = True):492    if use_source:493        return [f'[{idx+1}]\t "{item[0]}"\n{source_name}: {item[1]}' for idx, item in enumerate(lst)]494    else:495        return [f'[{idx+1}]\t "{item}"' for idx, item in enumerate(lst)]496 497def add_details(lst):498    nodes = []499    for index, txt in enumerate(lst):500        brief = txt[:25].replace("\n", "")501        nodes.append(502            f"<details><summary>{brief}...</summary><p>{txt}</p></details>"503        )504    return nodes505 506 507def sheet_to_string(sheet):508    result = ""509    for index, row in sheet.iterrows():510        row_string = ""511        for column in sheet.columns:512            row_string += f"{column}: {row[column]}, "513        row_string = row_string.rstrip(", ")514        row_string += "."515        result += row_string + "\n"516    return result517 518def excel_to_string(file_path):519    # 读取Excel文件中的所有工作表520    excel_file = pd.read_excel(file_path, engine='openpyxl', sheet_name=None)521 522    # 初始化结果字符串523    result = ""524 525    # 遍历每一个工作表526    for sheet_name, sheet_data in excel_file.items():527        # 将工作表名称添加到结果字符串528        result += f"Sheet: {sheet_name}\n"529 530        # 处理当前工作表并添加到结果字符串531        result += sheet_to_string(sheet_data)532 533        # 在不同工作表之间添加分隔符534        result += "\n" + ("-" * 20) + "\n\n"535 536    return result537