Rename bin_public/utils/tools.py to bin_public/utils/utils.py

bin_public/utils/{tools.py → utils.py}

@@ -5,31 +5,38 @@ import logging
 import json
 import os
 import datetime
-import hashlib
 import csv
+import requests
+import re
+import html
+import sys
+import subprocess
 
-import gradio as gr
 from pypinyin import lazy_pinyin
 import tiktoken
+import mdtex2html
+from markdown import markdown
+from pygments import highlight
+from pygments.lexers import get_lexer_by_name
+from pygments.formatters import HtmlFormatter
 
 from bin_public.config.presets import *
+import bin_public.utils.shared as shared
 
-# logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] [%(filename)s:%(lineno)d] %(message)s")
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] [%(filename)s:%(lineno)d] %(message)s",
+)
 
 if TYPE_CHECKING:
     from typing import TypedDict
 
+
     class DataframeData(TypedDict):
         headers: List[str]
         data: List[List[str | int | bool]]
 
 
-initial_prompt = "You are a helpful assistant."
-API_URL = "https://api.openai.com/v1/chat/completions"
-HISTORY_DIR = "history"
-TEMPLATES_DIR = "templates"
-
-
 def count_token(message):
     encoding = tiktoken.get_encoding("cl100k_base")
     input_str = f"role: {message['role']}, content: {message['content']}"
@@ -37,36 +44,99 @@ def count_token(message):
     return length
 
 
-def parse_text(text):
-    lines = text.split("\n")
-    lines = [line for line in lines if line != ""]
-    count = 0
+def markdown_to_html_with_syntax_highlight(md_str):
+    def replacer(match):
+        lang = match.group(1) or "text"
+        code = match.group(2)
+
+        try:
+            lexer = get_lexer_by_name(lang, stripall=True)
+        except ValueError:
+            lexer = get_lexer_by_name("text", stripall=True)
+
+        formatter = HtmlFormatter()
+        highlighted_code = highlight(code, lexer, formatter)
+
+        return f'<pre><code class="{lang}">{highlighted_code}</code></pre>'
+
+    code_block_pattern = r"```(\w+)?\n([\s\S]+?)\n```"
+    md_str = re.sub(code_block_pattern, replacer, md_str, flags=re.MULTILINE)
+
+    html_str = markdown(md_str)
+    return html_str
+
+
+def normalize_markdown(md_text: str) -> str:
+    lines = md_text.split("\n")
+    normalized_lines = []
+    inside_list = False
+
     for i, line in enumerate(lines):
-        if "```" in line:
-            count += 1
-            items = line.split('`')
-            if count % 2 == 1:
-                lines[i] = f'<pre><code class="language-{items[-1]}">'
-            else:
-                lines[i] = f'<br></code></pre>'
+        if re.match(r"^(\d+\.|-|\*|\+)\s", line.strip()):
+            if not inside_list and i > 0 and lines[i - 1].strip() != "":
+                normalized_lines.append("")
+            inside_list = True
+            normalized_lines.append(line)
+        elif inside_list and line.strip() == "":
+            if i < len(lines) - 1 and not re.match(
+                    r"^(\d+\.|-|\*|\+)\s", lines[i + 1].strip()
+            ):
+                normalized_lines.append(line)
+            continue
         else:
-            if i > 0:
-                if count % 2 == 1:
-                    line = line.replace("`", "\`")
-                    line = line.replace("<", "&lt;")
-                    line = line.replace(">", "&gt;")
-                    line = line.replace(" ", "&nbsp;")
-                    line = line.replace("*", "&ast;")
-                    line = line.replace("_", "&lowbar;")
-                    line = line.replace("-", "&#45;")
-                    line = line.replace(".", "&#46;")
-                    line = line.replace("!", "&#33;")
-                    line = line.replace("(", "&#40;")
-                    line = line.replace(")", "&#41;")
-                    line = line.replace("$", "&#36;")
-                lines[i] = "<br>" + line
-    text = "".join(lines)
-    return text
+            inside_list = False
+            normalized_lines.append(line)
+
+    return "\n".join(normalized_lines)
+
+
+def convert_mdtext(md_text):
+    code_block_pattern = re.compile(r"```(.*?)(?:```|$)", re.DOTALL)
+    inline_code_pattern = re.compile(r"`(.*?)`", re.DOTALL)
+    code_blocks = code_block_pattern.findall(md_text)
+    non_code_parts = code_block_pattern.split(md_text)[::2]
+
+    result = []
+    for non_code, code in zip(non_code_parts, code_blocks + [""]):
+        if non_code.strip():
+            non_code = normalize_markdown(non_code)
+            if inline_code_pattern.search(non_code):
+                result.append(markdown(non_code, extensions=["tables"]))
+            else:
+                result.append(mdtex2html.convert(non_code, extensions=["tables"]))
+        if code.strip():
+            # _, code = detect_language(code)  # 暂时去除代码高亮功能，因为在大段代码的情况下会出现问题
+            # code = code.replace("\n\n", "\n") # 暂时去除代码中的空行，因为在大段代码的情况下会出现问题
+            code = f"\n```{code}\n\n```"
+            code = markdown_to_html_with_syntax_highlight(code)
+            result.append(code)
+    result = "".join(result)
+    result += ALREADY_CONVERTED_MARK
+    return result
+
+
+def convert_asis(userinput):
+    return (
+            f'<p style="white-space:pre-wrap;">{html.escape(userinput)}</p>'
+            + ALREADY_CONVERTED_MARK
+    )
+
+
+def detect_converted_mark(userinput):
+    if userinput.endswith(ALREADY_CONVERTED_MARK):
+        return True
+    else:
+        return False
+
+
+def detect_language(code):
+    if code.startswith("\n"):
+        first_line = ""
+    else:
+        first_line = code.strip().split("\n", 1)[0]
+    language = first_line.lower() if first_line else ""
+    code_without_language = code[len(first_line):].lstrip() if first_line else code
+    return language, code_without_language
 
 
 def construct_text(role, text):
@@ -89,6 +159,17 @@ def construct_token_message(token, stream=False):
     return f"Token 计数: {token}"
 
 
+def delete_first_conversation(history, previous_token_count):
+    if history:
+        del history[:2]
+        del previous_token_count[0]
+    return (
+        history,
+        previous_token_count,
+        construct_token_message(sum(previous_token_count)),
+    )
+
+
 def delete_last_conversation(chatbot, history, previous_token_count):
     if len(chatbot) > 0 and standard_error_msg in chatbot[-1][1]:
         logging.info("由于包含报错信息，只删除chatbot记录")
@@ -210,7 +291,7 @@ def load_template(filename, mode=0):
         lines = [[i["act"], i["prompt"]] for i in lines]
     else:
         with open(
-            os.path.join(TEMPLATES_DIR, filename), "r", encoding="utf8"
+                os.path.join(TEMPLATES_DIR, filename), "r", encoding="utf8"
         ) as csvfile:
             reader = csv.reader(csvfile)
             lines = list(reader)
@@ -245,20 +326,19 @@ def reset_state():
 
 
 def reset_textbox():
+    logging.debug("重置文本框")
     return gr.update(value="")
 
 
 def reset_default():
-    global API_URL
-    API_URL = "https://api.openai.com/v1/chat/completions"
+    newurl = shared.state.reset_api_url()
     os.environ.pop("HTTPS_PROXY", None)
     os.environ.pop("https_proxy", None)
-    return gr.update(value=API_URL), gr.update(value=""), "API URL 和代理已重置"
+    return gr.update(value=newurl), gr.update(value=""), "API URL 和代理已重置"
 
 
 def change_api_url(url):
-    global API_URL
-    API_URL = url
+    shared.state.set_api_url(url)
     msg = f"API地址更改为了{url}"
     logging.info(msg)
     return msg
@@ -288,12 +368,138 @@ def submit_key(key):
     return key, msg
 
 
-def sha1sum(filename):
-    sha1 = hashlib.sha1()
-    sha1.update(filename.encode("utf-8"))
-    return sha1.hexdigest()
-
-
 def replace_today(prompt):
     today = datetime.datetime.today().strftime("%Y-%m-%d")
-    return prompt.replace("{current_date}", today)
+    return prompt.replace("{current_date}", today)
+
+
+def get_geoip():
+    response = requests.get("https://ipapi.co/json/", timeout=5)
+    try:
+        data = response.json()
+    except:
+        data = {"error": True, "reason": "连接ipapi失败"}
+    if "error" in data.keys():
+        logging.warning(f"无法获取IP地址信息。\n{data}")
+        if data["reason"] == "RateLimited":
+            return (
+                f"获取IP地理位置失败，因为达到了检测IP的速率限制。聊天功能可能仍然可用，但请注意，如果您的IP地址在不受支持的地区，您可能会遇到问题。"
+            )
+        else:
+            return f"获取IP地理位置失败。原因：{data['reason']}。你仍然可以使用聊天功能。"
+    else:
+        country = data["country_name"]
+        if country == "China":
+            text = "**您的IP区域：中国。请立即检查代理设置，在不受支持的地区使用API可能导致账号被封禁。**"
+        else:
+            text = f"您的IP区域：{country}。"
+        logging.info(text)
+        return text
+
+
+def find_n(lst, max_num):
+    n = len(lst)
+    total = sum(lst)
+
+    if total < max_num:
+        return n
+
+    for i in range(len(lst)):
+        if total - lst[i] < max_num:
+            return n - i - 1
+        total = total - lst[i]
+    return 1
+
+
+def start_outputing():
+    logging.debug("显示取消按钮，隐藏发送按钮")
+    return gr.Button.update(visible=False), gr.Button.update(visible=True)
+
+
+def end_outputing():
+    return (
+        gr.Button.update(visible=True),
+        gr.Button.update(visible=False),
+    )
+
+
+def cancel_outputing():
+    logging.info("中止输出……")
+    shared.state.interrupt()
+
+
+def transfer_input(inputs):
+    # 一次性返回，降低延迟
+    textbox = reset_textbox()
+    outputing = start_outputing()
+    return (
+        inputs,
+        gr.update(value=""),
+        gr.Button.update(visible=False),
+        gr.Button.update(visible=True),
+    )
+
+
+def get_proxies():
+    # 获取环境变量中的代理设置
+    http_proxy = os.environ.get("HTTP_PROXY") or os.environ.get("http_proxy")
+    https_proxy = os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy")
+
+    # 如果存在代理设置，使用它们
+    proxies = {}
+    if http_proxy:
+        logging.info(f"使用 HTTP 代理: {http_proxy}")
+        proxies["http"] = http_proxy
+    if https_proxy:
+        logging.info(f"使用 HTTPS 代理: {https_proxy}")
+        proxies["https"] = https_proxy
+
+    if proxies == {}:
+        proxies = None
+
+    return proxies
+
+
+def run(command, desc=None, errdesc=None, custom_env=None, live=False):
+    if desc is not None:
+        print(desc)
+    if live:
+        result = subprocess.run(command, shell=True, env=os.environ if custom_env is None else custom_env)
+        if result.returncode != 0:
+            raise RuntimeError(f"""{errdesc or 'Error running command'}.
+Command: {command}
+Error code: {result.returncode}""")
+
+        return ""
+    result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True,
+                            env=os.environ if custom_env is None else custom_env)
+    if result.returncode != 0:
+        message = f"""{errdesc or 'Error running command'}.
+Command: {command}
+Error code: {result.returncode}
+stdout: {result.stdout.decode(encoding="utf8", errors="ignore") if len(result.stdout) > 0 else '<empty>'}
+stderr: {result.stderr.decode(encoding="utf8", errors="ignore") if len(result.stderr) > 0 else '<empty>'}
+"""
+        raise RuntimeError(message)
+    return result.stdout.decode(encoding="utf8", errors="ignore")
+
+
+def versions_html():
+    git = os.environ.get('GIT', "git")
+    python_version = ".".join([str(x) for x in sys.version_info[0:3]])
+    try:
+        commit_hash = run(f"{git} rev-parse HEAD").strip()
+    except Exception:
+        commit_hash = "<none>"
+    if commit_hash != "<none>":
+        short_commit = commit_hash[0:7]
+        commit_info = f"<a style=\"text-decoration:none\" href=\"https://github.com/GaiZhenbiao/ChuanhuChatGPT/commit/{short_commit}\">{short_commit}</a>"
+    else:
+        commit_info = "unknown \U0001F615"
+    return f"""
+Python: <span title="{sys.version}">{python_version}</span>
+ • 
+Gradio: {gr.__version__}
+ • 
+Commit: {commit_info}
+"""