Danielzero/GPT3.5

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 37def predict(current_model, *args):38    iter = current_model.predict(*args)39    for i in iter:40        yield i41 42def billing_info(current_model):43    return current_model.billing_info()44 45def set_key(current_model, *args):46    return current_model.set_key(*args)47 48def load_chat_history(current_model, *args):49    return current_model.load_chat_history(*args)50 51def interrupt(current_model, *args):52    return current_model.interrupt(*args)53 54def reset(current_model, *args):55    return current_model.reset(*args)56 57def retry(current_model, *args):58    iter = current_model.retry(*args)59    for i in iter:60        yield i61 62def delete_first_conversation(current_model, *args):63    return current_model.delete_first_conversation(*args)64 65def delete_last_conversation(current_model, *args):66    return current_model.delete_last_conversation(*args)67 68def set_system_prompt(current_model, *args):69    return current_model.set_system_prompt(*args)70 71def save_chat_history(current_model, *args):72    return current_model.save_chat_history(*args)73 74def export_markdown(current_model, *args):75    return current_model.export_markdown(*args)76 77def load_chat_history(current_model, *args):78    return current_model.load_chat_history(*args)79 80def set_token_upper_limit(current_model, *args):81    return current_model.set_token_upper_limit(*args)82 83def set_temperature(current_model, *args):84    current_model.set_temperature(*args)85 86def set_top_p(current_model, *args):87    current_model.set_top_p(*args)88 89def set_n_choices(current_model, *args):90    current_model.set_n_choices(*args)91 92def set_stop_sequence(current_model, *args):93    current_model.set_stop_sequence(*args)94 95def set_max_tokens(current_model, *args):96    current_model.set_max_tokens(*args)97 98def set_presence_penalty(current_model, *args):99    current_model.set_presence_penalty(*args)100 101def set_frequency_penalty(current_model, *args):102    current_model.set_frequency_penalty(*args)103 104def set_logit_bias(current_model, *args):105    current_model.set_logit_bias(*args)106 107def set_user_identifier(current_model, *args):108    current_model.set_user_identifier(*args)109 110def set_single_turn(current_model, *args):111    current_model.set_single_turn(*args)112 113def handle_file_upload(current_model, *args):114    return current_model.handle_file_upload(*args)115 116def like(current_model, *args):117    return current_model.like(*args)118 119def dislike(current_model, *args):120    return current_model.dislike(*args)121 122 123def count_token(message):124    encoding = tiktoken.get_encoding("cl100k_base")125    input_str = f"role: {message['role']}, content: {message['content']}"126    length = len(encoding.encode(input_str))127    return length128 129 130def markdown_to_html_with_syntax_highlight(md_str):131    def replacer(match):132        lang = match.group(1) or "text"133        code = match.group(2)134 135        try:136            lexer = get_lexer_by_name(lang, stripall=True)137        except ValueError:138            lexer = get_lexer_by_name("text", stripall=True)139 140        formatter = HtmlFormatter()141        highlighted_code = highlight(code, lexer, formatter)142 143        return f'<pre><code class="{lang}">{highlighted_code}</code></pre>'144 145    code_block_pattern = r"```(\w+)?\n([\s\S]+?)\n```"146    md_str = re.sub(code_block_pattern, replacer, md_str, flags=re.MULTILINE)147 148    html_str = markdown(md_str)149    return html_str150 151 152def normalize_markdown(md_text: str) -> str:153    lines = md_text.split("\n")154    normalized_lines = []155    inside_list = False156 157    for i, line in enumerate(lines):158        if re.match(r"^(\d+\.|-|\*|\+)\s", line.strip()):159            if not inside_list and i > 0 and lines[i - 1].strip() != "":160                normalized_lines.append("")161            inside_list = True162            normalized_lines.append(line)163        elif inside_list and line.strip() == "":164            if i < len(lines) - 1 and not re.match(165                r"^(\d+\.|-|\*|\+)\s", lines[i + 1].strip()166            ):167                normalized_lines.append(line)168            continue169        else:170            inside_list = False171            normalized_lines.append(line)172 173    return "\n".join(normalized_lines)174 175 176def convert_mdtext(md_text):177    code_block_pattern = re.compile(r"```(.*?)(?:```|$)", re.DOTALL)178    inline_code_pattern = re.compile(r"`(.*?)`", re.DOTALL)179    code_blocks = code_block_pattern.findall(md_text)180    non_code_parts = code_block_pattern.split(md_text)[::2]181 182    result = []183    for non_code, code in zip(non_code_parts, code_blocks + [""]):184        if non_code.strip():185            non_code = normalize_markdown(non_code)186            if inline_code_pattern.search(non_code):187                result.append(markdown(non_code, extensions=["tables"]))188            else:189                result.append(mdtex2html.convert(non_code, extensions=["tables"]))190        if code.strip():191            # _, code = detect_language(code)  # 暂时去除代码高亮功能，因为在大段代码的情况下会出现问题192            # code = code.replace("\n\n", "\n") # 暂时去除代码中的空行，因为在大段代码的情况下会出现问题193            code = f"\n```{code}\n\n```"194            code = markdown_to_html_with_syntax_highlight(code)195            result.append(code)196    result = "".join(result)197    result += ALREADY_CONVERTED_MARK198    return result199 200 201def convert_asis(userinput):202    return (203        f'<p style="white-space:pre-wrap;">{html.escape(userinput)}</p>'204        + ALREADY_CONVERTED_MARK205    )206 207 208def detect_converted_mark(userinput):209    try:210        if userinput.endswith(ALREADY_CONVERTED_MARK):211            return True212        else:213            return False214    except:215        return True216 217 218def detect_language(code):219    if code.startswith("\n"):220        first_line = ""221    else:222        first_line = code.strip().split("\n", 1)[0]223    language = first_line.lower() if first_line else ""224    code_without_language = code[len(first_line) :].lstrip() if first_line else code225    return language, code_without_language226 227 228def construct_text(role, text):229    return {"role": role, "content": text}230 231 232def construct_user(text):233    return construct_text("user", text)234 235 236def construct_system(text):237    return construct_text("system", text)238 239 240def construct_assistant(text):241    return construct_text("assistant", text)242 243 244def save_file(filename, system, history, chatbot, user_name):245    logging.debug(f"{user_name} 保存对话历史中……")246    os.makedirs(os.path.join(HISTORY_DIR, user_name), exist_ok=True)247    if filename.endswith(".json"):248        json_s = {"system": system, "history": history, "chatbot": chatbot}249        print(json_s)250        with open(os.path.join(HISTORY_DIR, user_name, filename), "w") as f:251            json.dump(json_s, f)252    elif filename.endswith(".md"):253        md_s = f"system: \n- {system} \n"254        for data in history:255            md_s += f"\n{data['role']}: \n- {data['content']} \n"256        with open(os.path.join(HISTORY_DIR, user_name, filename), "w", encoding="utf8") as f:257            f.write(md_s)258    logging.debug(f"{user_name} 保存对话历史完毕")259    return os.path.join(HISTORY_DIR, user_name, filename)260 261 262def sorted_by_pinyin(list):263    return sorted(list, key=lambda char: lazy_pinyin(char)[0][0])264 265 266def get_file_names(dir, plain=False, filetypes=[".json"]):267    logging.debug(f"获取文件名列表，目录为{dir}，文件类型为{filetypes}，是否为纯文本列表{plain}")268    files = []269    try:270        for type in filetypes:271            files += [f for f in os.listdir(dir) if f.endswith(type)]272    except FileNotFoundError:273        files = []274    files = sorted_by_pinyin(files)275    if files == []:276        files = [""]277    logging.debug(f"files are:{files}")278    if plain:279        return files280    else:281        return gr.Dropdown.update(choices=files)282 283 284def get_history_names(plain=False, user_name=""):285    logging.debug(f"从用户 {user_name} 中获取历史记录文件名列表")286    return get_file_names(os.path.join(HISTORY_DIR, user_name), plain)287 288 289def load_template(filename, mode=0):290    logging.debug(f"加载模板文件{filename}，模式为{mode}（0为返回字典和下拉菜单，1为返回下拉菜单，2为返回字典）")291    lines = []292    if filename.endswith(".json"):293        with open(os.path.join(TEMPLATES_DIR, filename), "r", encoding="utf8") as f:294            lines = json.load(f)295        lines = [[i["act"], i["prompt"]] for i in lines]296    else:297        with open(298            os.path.join(TEMPLATES_DIR, filename), "r", encoding="utf8"299        ) as csvfile:300            reader = csv.reader(csvfile)301            lines = list(reader)302        lines = lines[1:]303    if mode == 1:304        return sorted_by_pinyin([row[0] for row in lines])305    elif mode == 2:306        return {row[0]: row[1] for row in lines}307    else:308        choices = sorted_by_pinyin([row[0] for row in lines])309        return {row[0]: row[1] for row in lines}, gr.Dropdown.update(310            choices=choices311        )312 313 314def get_template_names(plain=False):315    logging.debug("获取模板文件名列表")316    return get_file_names(TEMPLATES_DIR, plain, filetypes=[".csv", "json"])317 318 319def get_template_content(templates, selection, original_system_prompt):320    logging.debug(f"应用模板中，选择为{selection}，原始系统提示为{original_system_prompt}")321    try:322        return templates[selection]323    except:324        return original_system_prompt325 326 327def reset_textbox():328    logging.debug("重置文本框")329    return gr.update(value="")330 331 332def reset_default():333    default_host = shared.state.reset_api_host()334    retrieve_proxy("")335    return gr.update(value=default_host), gr.update(value=""), "API-Host 和代理已重置"336 337 338def change_api_host(host):339    shared.state.set_api_host(host)340    msg = f"API-Host更改为了{host}"341    logging.info(msg)342    return msg343 344 345def change_proxy(proxy):346    retrieve_proxy(proxy)347    os.environ["HTTPS_PROXY"] = proxy348    msg = f"代理更改为了{proxy}"349    logging.info(msg)350    return msg351 352 353def hide_middle_chars(s):354    if s is None:355        return ""356    if len(s) <= 8:357        return s358    else:359        head = s[:4]360        tail = s[-4:]361        hidden = "*" * (len(s) - 8)362        return head + hidden + tail363 364 365def submit_key(key):366    key = key.strip()367    msg = f"API密钥更改为了{hide_middle_chars(key)}"368    logging.info(msg)369    return key, msg370 371 372def replace_today(prompt):373    today = datetime.datetime.today().strftime("%Y-%m-%d")374    return prompt.replace("{current_date}", today)375 376 377def get_geoip():378    try:379        with retrieve_proxy():380            response = requests.get("https://ipapi.co/json/", timeout=5)381        data = response.json()382    except:383        data = {"error": True, "reason": "连接ipapi失败"}384    if "error" in data.keys():385        logging.warning(f"无法获取IP地址信息。\n{data}")386        if data["reason"] == "RateLimited":387            return (388                i18n("您的IP区域：未知。")389            )390        else:391            return i18n("获取IP地理位置失败。原因：") + f"{data['reason']}" + i18n("。你仍然可以使用聊天功能。")392    else:393        country = data["country_name"]394        if country == "China":395            text = "**您的IP区域：中国。请立即检查代理设置，在不受支持的地区使用API可能导致账号被封禁。**"396        else:397            text = i18n("您的IP区域：") + f"{country}。"398        logging.info(text)399        return text400 401 402def find_n(lst, max_num):403    n = len(lst)404    total = sum(lst)405 406    if total < max_num:407        return n408 409    for i in range(len(lst)):410        if total - lst[i] < max_num:411            return n - i - 1412        total = total - lst[i]413    return 1414 415 416def start_outputing():417    logging.debug("显示取消按钮，隐藏发送按钮")418    return gr.Button.update(visible=False), gr.Button.update(visible=True)419 420 421def end_outputing():422    return (423        gr.Button.update(visible=True),424        gr.Button.update(visible=False),425    )426 427 428def cancel_outputing():429    logging.info("中止输出……")430    shared.state.interrupt()431 432 433def transfer_input(inputs):434    # 一次性返回，降低延迟435    textbox = reset_textbox()436    outputing = start_outputing()437    return (438        inputs,439        gr.update(value=""),440        gr.Button.update(visible=False),441        gr.Button.update(visible=True),442    )443 444 445 446def run(command, desc=None, errdesc=None, custom_env=None, live=False):447    if desc is not None:448        print(desc)449    if live:450        result = subprocess.run(command, shell=True, env=os.environ if custom_env is None else custom_env)451        if result.returncode != 0:452            raise RuntimeError(f"""{errdesc or 'Error running command'}.453Command: {command}454Error code: {result.returncode}""")455 456        return ""457    result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, env=os.environ if custom_env is None else custom_env)458    if result.returncode != 0:459        message = f"""{errdesc or 'Error running command'}.460            Command: {command}461            Error code: {result.returncode}462            stdout: {result.stdout.decode(encoding="utf8", errors="ignore") if len(result.stdout)>0 else '<empty>'}463            stderr: {result.stderr.decode(encoding="utf8", errors="ignore") if len(result.stderr)>0 else '<empty>'}464            """465        raise RuntimeError(message)466    return result.stdout.decode(encoding="utf8", errors="ignore")467 468def versions_html():469    git = os.environ.get('GIT', "git")470    python_version = ".".join([str(x) for x in sys.version_info[0:3]])471    try:472        commit_hash = run(f"{git} rev-parse HEAD").strip()473    except Exception:474        commit_hash = "<none>"475    if commit_hash != "<none>":476        short_commit = commit_hash[0:7]477        commit_info = f"<a style=\"text-decoration:none\" href=\"https://github.com/GaiZhenbiao/ChuanhuChatGPT/commit/{short_commit}\">{short_commit}</a>"478    else:479        commit_info = "unknown \U0001F615"480    return f"""481        Python: <span title="{sys.version}">{python_version}</span>482         • 483        Gradio: {gr.__version__}484         • 485        Commit: {commit_info}486        """487 488def add_source_numbers(lst, source_name = "Source", use_source = True):489    if use_source:490        return [f'[{idx+1}]\t "{item[0]}"\n{source_name}: {item[1]}' for idx, item in enumerate(lst)]491    else:492        return [f'[{idx+1}]\t "{item}"' for idx, item in enumerate(lst)]493 494def add_details(lst):495    nodes = []496    for index, txt in enumerate(lst):497        brief = txt[:25].replace("\n", "")498        nodes.append(499            f"<details><summary>{brief}...</summary><p>{txt}</p></details>"500        )501    return nodes502 503 504def sheet_to_string(sheet, sheet_name = None):505    result = []506    for index, row in sheet.iterrows():507        row_string = ""508        for column in sheet.columns:509            row_string += f"{column}: {row[column]}, "510        row_string = row_string.rstrip(", ")511        row_string += "."512        result.append(row_string)513    return result514 515def excel_to_string(file_path):516    # 读取Excel文件中的所有工作表517    excel_file = pd.read_excel(file_path, engine='openpyxl', sheet_name=None)518 519    # 初始化结果字符串520    result = []521 522    # 遍历每一个工作表523    for sheet_name, sheet_data in excel_file.items():524 525        # 处理当前工作表并添加到结果字符串526        result += sheet_to_string(sheet_data, sheet_name=sheet_name)527 528 529    return result530 531def get_last_day_of_month(any_day):532    # The day 28 exists in every month. 4 days later, it's always next month533    next_month = any_day.replace(day=28) + datetime.timedelta(days=4)534    # subtracting the number of the current day brings us back one month535    return next_month - datetime.timedelta(days=next_month.day)536 537def get_model_source(model_name, alternative_source):538    if model_name == "gpt2-medium":539        return "https://huggingface.co/gpt2-medium"540 541def refresh_ui_elements_on_load(current_model, selected_model_name):542    return toggle_like_btn_visibility(selected_model_name)543 544def toggle_like_btn_visibility(selected_model_name):545    if selected_model_name == "xmchat":546        return gr.update(visible=True)547    else:548        return gr.update(visible=False)549