Spaces:
Running
Running
File size: 2,042 Bytes
3fdcc70 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 |
import os
import functools
import signal
from pathlib import Path
RESOURCE_ROOT = os.environ.get("RESOURCE_ROOT", "./client_resources")
def get_real_path(path):
if path is None:
return None
if RESOURCE_ROOT in path:
return path
return os.path.join(RESOURCE_ROOT, path)
def get_root_dir():
return RESOURCE_ROOT
def md2plain(md):
plain_text = md.replace(" ", " ")
plain_text = plain_text.replace("<br>", "\n")
plain_text = plain_text.replace("\<", "<")
plain_text = plain_text.replace("\>", ">")
return plain_text
def plain2md(plain_text: str):
md_text = plain_text.replace("<", "\<")
md_text = md_text.replace(">", "\>")
md_text = md_text.replace("\n", "<br>")
# md_text = md_text + "<br>"
md_text = md_text.replace(" ", " ")
return md_text
def transform_msgs(history_msgs: list = []):
if history_msgs is None:
return []
filtered_msg = []
for item in history_msgs:
if isinstance(item[0], str):
item[0] = md2plain(item[0])
if isinstance(item[1], str):
item[1] = md2plain(item[1])
if isinstance(item[1], str) and item[1].startswith(
"The whole process will take some time, please be patient."
):
item[1] = None
filtered_msg.append(item)
return filtered_msg
def timeout(sec):
"""
timeout decorator
:param sec: function raise TimeoutError after ? seconds
"""
def decorator(func):
@functools.wraps(func)
def wrapped_func(*args, **kwargs):
def _handle_timeout(signum, frame):
err_msg = f"Function {func.__name__} timed out after {sec} seconds"
raise TimeoutError(err_msg)
signal.signal(signal.SIGALRM, _handle_timeout)
signal.alarm(sec)
try:
result = func(*args, **kwargs)
finally:
signal.alarm(0)
return result
return wrapped_func
return decorator
|