Intoval/privateChatGPT

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 116 117def count_token(message):118    encoding = tiktoken.get_encoding("cl100k_base")119    input_str = f"role: {message['role']}, content: {message['content']}"120    length = len(encoding.encode(input_str))121    return length122 123 124def markdown_to_html_with_syntax_highlight(md_str):125    def replacer(match):126        lang = match.group(1) or "text"127        code = match.group(2)128 129        try:130            lexer = get_lexer_by_name(lang, stripall=True)131        except ValueError:132            lexer = get_lexer_by_name("text", stripall=True)133 134        formatter = HtmlFormatter()135        highlighted_code = highlight(code, lexer, formatter)136 137        return f'<pre><code class="{lang}">{highlighted_code}</code></pre>'138 139    code_block_pattern = r"```(\w+)?\n([\s\S]+?)\n```"140    md_str = re.sub(code_block_pattern, replacer, md_str, flags=re.MULTILINE)141 142    html_str = markdown(md_str)143    return html_str144 145 146def normalize_markdown(md_text: str) -> str:147    lines = md_text.split("\n")148    normalized_lines = []149    inside_list = False150 151    for i, line in enumerate(lines):152        if re.match(r"^(\d+\.|-|\*|\+)\s", line.strip()):153            if not inside_list and i > 0 and lines[i - 1].strip() != "":154                normalized_lines.append("")155            inside_list = True156            normalized_lines.append(line)157        elif inside_list and line.strip() == "":158            if i < len(lines) - 1 and not re.match(159                r"^(\d+\.|-|\*|\+)\s", lines[i + 1].strip()160            ):161                normalized_lines.append(line)162            continue163        else:164            inside_list = False165            normalized_lines.append(line)166 167    return "\n".join(normalized_lines)168 169 170def convert_mdtext(md_text):171    code_block_pattern = re.compile(r"```(.*?)(?:```|$)", re.DOTALL)172    inline_code_pattern = re.compile(r"`(.*?)`", re.DOTALL)173    code_blocks = code_block_pattern.findall(md_text)174    non_code_parts = code_block_pattern.split(md_text)[::2]175 176    result = []177    for non_code, code in zip(non_code_parts, code_blocks + [""]):178        if non_code.strip():179            non_code = normalize_markdown(non_code)180            if inline_code_pattern.search(non_code):181                result.append(markdown(non_code, extensions=["tables"]))182            else:183                result.append(mdtex2html.convert(non_code, extensions=["tables"]))184        if code.strip():185            # _, code = detect_language(code)  # 暂时去除代码高亮功能，因为在大段代码的情况下会出现问题186            # code = code.replace("\n\n", "\n") # 暂时去除代码中的空行，因为在大段代码的情况下会出现问题187            code = f"\n```{code}\n\n```"188            code = markdown_to_html_with_syntax_highlight(code)189            result.append(code)190    result = "".join(result)191    result += ALREADY_CONVERTED_MARK192    return result193 194 195def convert_asis(userinput):196    return (197        f'<p style="white-space:pre-wrap;">{html.escape(userinput)}</p>'198        + ALREADY_CONVERTED_MARK199    )200 201 202def detect_converted_mark(userinput):203    try:204        if userinput.endswith(ALREADY_CONVERTED_MARK):205            return True206        else:207            return False208    except:209        return True210 211 212def detect_language(code):213    if code.startswith("\n"):214        first_line = ""215    else:216        first_line = code.strip().split("\n", 1)[0]217    language = first_line.lower() if first_line else ""218    code_without_language = code[len(first_line) :].lstrip() if first_line else code219    return language, code_without_language220 221 222def construct_text(role, text):223    return {"role": role, "content": text}224 225 226def construct_user(text):227    return construct_text("user", text)228 229 230def construct_system(text):231    return construct_text("system", text)232 233 234def construct_assistant(text):235    return construct_text("assistant", text)236 237 238def save_file(filename, system, history, chatbot, user_name):239    logging.debug(f"{user_name} 保存对话历史中……")240    os.makedirs(os.path.join(HISTORY_DIR, user_name), exist_ok=True)241    if filename.endswith(".json"):242        json_s = {"system": system, "history": history, "chatbot": chatbot}243        print(json_s)244        with open(os.path.join(HISTORY_DIR, user_name, filename), "w") as f:245            json.dump(json_s, f)246    elif filename.endswith(".md"):247        md_s = f"system: \n- {system} \n"248        for data in history:249            md_s += f"\n{data['role']}: \n- {data['content']} \n"250        with open(os.path.join(HISTORY_DIR, user_name, filename), "w", encoding="utf8") as f:251            f.write(md_s)252    logging.debug(f"{user_name} 保存对话历史完毕")253    return os.path.join(HISTORY_DIR, user_name, filename)254 255 256def sorted_by_pinyin(list):257    return sorted(list, key=lambda char: lazy_pinyin(char)[0][0])258 259 260def get_file_names(dir, plain=False, filetypes=[".json"]):261    logging.debug(f"获取文件名列表，目录为{dir}，文件类型为{filetypes}，是否为纯文本列表{plain}")262    files = []263    try:264        for type in filetypes:265            files += [f for f in os.listdir(dir) if f.endswith(type)]266    except FileNotFoundError:267        files = []268    files = sorted_by_pinyin(files)269    if files == []:270        files = [""]271    logging.debug(f"files are:{files}")272    if plain:273        return files274    else:275        return gr.Dropdown.update(choices=files)276 277 278def get_history_names(plain=False, user_name=""):279    logging.debug(f"从用户 {user_name} 中获取历史记录文件名列表")280    return get_file_names(os.path.join(HISTORY_DIR, user_name), plain)281 282 283def load_template(filename, mode=0):284    logging.debug(f"加载模板文件{filename}，模式为{mode}（0为返回字典和下拉菜单，1为返回下拉菜单，2为返回字典）")285    lines = []286    if filename.endswith(".json"):287        with open(os.path.join(TEMPLATES_DIR, filename), "r", encoding="utf8") as f:288            lines = json.load(f)289        lines = [[i["act"], i["prompt"]] for i in lines]290    else:291        with open(292            os.path.join(TEMPLATES_DIR, filename), "r", encoding="utf8"293        ) as csvfile:294            reader = csv.reader(csvfile)295            lines = list(reader)296        lines = lines[1:]297    if mode == 1:298        return sorted_by_pinyin([row[0] for row in lines])299    elif mode == 2:300        return {row[0]: row[1] for row in lines}301    else:302        choices = sorted_by_pinyin([row[0] for row in lines])303        return {row[0]: row[1] for row in lines}, gr.Dropdown.update(304            choices=choices305        )306 307 308def get_template_names(plain=False):309    logging.debug("获取模板文件名列表")310    return get_file_names(TEMPLATES_DIR, plain, filetypes=[".csv", "json"])311 312 313def get_template_content(templates, selection, original_system_prompt):314    logging.debug(f"应用模板中，选择为{selection}，原始系统提示为{original_system_prompt}")315    try:316        return templates[selection]317    except:318        return original_system_prompt319 320 321def reset_textbox():322    logging.debug("重置文本框")323    return gr.update(value="")324 325 326def reset_default():327    default_host = shared.state.reset_api_host()328    retrieve_proxy("")329    return gr.update(value=default_host), gr.update(value=""), "API-Host 和代理已重置"330 331 332def change_api_host(host):333    shared.state.set_api_host(host)334    msg = f"API-Host更改为了{host}"335    logging.info(msg)336    return msg337 338 339def change_proxy(proxy):340    retrieve_proxy(proxy)341    os.environ["HTTPS_PROXY"] = proxy342    msg = f"代理更改为了{proxy}"343    logging.info(msg)344    return msg345 346 347def hide_middle_chars(s):348    if s is None:349        return ""350    if len(s) <= 8:351        return s352    else:353        head = s[:4]354        tail = s[-4:]355        hidden = "*" * (len(s) - 8)356        return head + hidden + tail357 358 359def submit_key(key):360    key = key.strip()361    msg = f"API密钥更改为了{hide_middle_chars(key)}"362    logging.info(msg)363    return key, msg364 365 366def replace_today(prompt):367    today = datetime.datetime.today().strftime("%Y-%m-%d")368    return prompt.replace("{current_date}", today)369 370 371def get_geoip():372    try:373        with retrieve_proxy():374            response = requests.get("https://ipapi.co/json/", timeout=5)375        data = response.json()376    except:377        data = {"error": True, "reason": "连接ipapi失败"}378    if "error" in data.keys():379        logging.warning(f"无法获取IP地址信息。\n{data}")380        if data["reason"] == "RateLimited":381            return (382                i18n("您的IP区域：未知。")383            )384        else:385            return i18n("获取IP地理位置失败。原因：") + f"{data['reason']}" + i18n("。你仍然可以使用聊天功能。")386    else:387        country = data["country_name"]388        if country == "China":389            text = "**您的IP区域：中国。请立即检查代理设置，在不受支持的地区使用API可能导致账号被封禁。**"390        else:391            text = i18n("您的IP区域：") + f"{country}。"392        logging.info(text)393        return text394 395 396def find_n(lst, max_num):397    n = len(lst)398    total = sum(lst)399 400    if total < max_num:401        return n402 403    for i in range(len(lst)):404        if total - lst[i] < max_num:405            return n - i - 1406        total = total - lst[i]407    return 1408 409 410def start_outputing():411    logging.debug("显示取消按钮，隐藏发送按钮")412    return gr.Button.update(visible=False), gr.Button.update(visible=True)413 414 415def end_outputing():416    return (417        gr.Button.update(visible=True),418        gr.Button.update(visible=False),419    )420 421 422def cancel_outputing():423    logging.info("中止输出……")424    shared.state.interrupt()425 426 427def transfer_input(inputs):428    # 一次性返回，降低延迟429    textbox = reset_textbox()430    outputing = start_outputing()431    return (432        inputs,433        gr.update(value=""),434        gr.Button.update(visible=False),435        gr.Button.update(visible=True),436    )437 438 439 440def run(command, desc=None, errdesc=None, custom_env=None, live=False):441    if desc is not None:442        print(desc)443    if live:444        result = subprocess.run(command, shell=True, env=os.environ if custom_env is None else custom_env)445        if result.returncode != 0:446            raise RuntimeError(f"""{errdesc or 'Error running command'}.447Command: {command}448Error code: {result.returncode}""")449 450        return ""451    result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, env=os.environ if custom_env is None else custom_env)452    if result.returncode != 0:453        message = f"""{errdesc or 'Error running command'}.454Command: {command}455Error code: {result.returncode}456stdout: {result.stdout.decode(encoding="utf8", errors="ignore") if len(result.stdout)>0 else '<empty>'}457stderr: {result.stderr.decode(encoding="utf8", errors="ignore") if len(result.stderr)>0 else '<empty>'}458"""459        raise RuntimeError(message)460    return result.stdout.decode(encoding="utf8", errors="ignore")461 462def versions_html():463    git = os.environ.get('GIT', "git")464    python_version = ".".join([str(x) for x in sys.version_info[0:3]])465    try:466        commit_hash = run(f"{git} rev-parse HEAD").strip()467    except Exception:468        commit_hash = "<none>"469    if commit_hash != "<none>":470        short_commit = commit_hash[0:7]471        commit_info = f"<a style=\"text-decoration:none\" href=\"https://github.com/GaiZhenbiao/ChuanhuChatGPT/commit/{short_commit}\">{short_commit}</a>"472    else:473        commit_info = "unknown \U0001F615"474    return f"""475Python: <span title="{sys.version}">{python_version}</span>476 • 477Gradio: {gr.__version__}478 • 479Commit: {commit_info}480"""481 482def add_source_numbers(lst, source_name = "Source", use_source = True):483    if use_source:484        return [f'[{idx+1}]\t "{item[0]}"\n{source_name}: {item[1]}' for idx, item in enumerate(lst)]485    else:486        return [f'[{idx+1}]\t "{item}"' for idx, item in enumerate(lst)]487 488def add_details(lst):489    nodes = []490    for index, txt in enumerate(lst):491        brief = txt[:25].replace("\n", "")492        nodes.append(493            f"<details><summary>{brief}...</summary><p>{txt}</p></details>"494        )495    return nodes496 497 498def sheet_to_string(sheet, sheet_name = None):499    result = []500    for index, row in sheet.iterrows():501        row_string = ""502        for column in sheet.columns:503            row_string += f"{column}: {row[column]}, "504        row_string = row_string.rstrip(", ")505        row_string += "."506        result.append(row_string)507    return result508 509def excel_to_string(file_path):510    # 读取Excel文件中的所有工作表511    excel_file = pd.read_excel(file_path, engine='openpyxl', sheet_name=None)512 513    # 初始化结果字符串514    result = []515 516    # 遍历每一个工作表517    for sheet_name, sheet_data in excel_file.items():518 519        # 处理当前工作表并添加到结果字符串520        result += sheet_to_string(sheet_data, sheet_name=sheet_name)521 522 523    return result524 525def get_last_day_of_month(any_day):526    # The day 28 exists in every month. 4 days later, it's always next month527    next_month = any_day.replace(day=28) + datetime.timedelta(days=4)528    # subtracting the number of the current day brings us back one month529    return next_month - datetime.timedelta(days=next_month.day)530 531def get_model_source(model_name, alternative_source):532    if model_name == "gpt2-medium":533        return "https://huggingface.co/gpt2-medium"534