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
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
|
#!/usr/bin/env python
import os
import json
import sys
import importlib.util
def main():
(bot_name, bot_func, raw_data) = parse_argv("run-bot.py")
bot_data = parse_raw_data(raw_data)
root_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
setup_env(root_dir, bot_name)
bot_tools_path = os.path.join(root_dir, f"bots/{bot_name}/tools.py")
run(bot_tools_path, bot_func, bot_data)
def parse_raw_data(data):
if not data:
raise ValueError("No JSON data")
try:
return json.loads(data)
except Exception:
raise ValueError("Invalid JSON data")
def parse_argv(this_file_name):
argv = sys.argv[:] + [None] * max(0, 4 - len(sys.argv))
bot_name = argv[0]
bot_func = ""
bot_data = None
if bot_name.endswith(this_file_name):
bot_name = sys.argv[1]
bot_func = sys.argv[2]
bot_data = sys.argv[3]
else:
bot_name = os.path.basename(bot_name)
bot_func = sys.argv[1]
bot_data = sys.argv[2]
if bot_name.endswith(".py"):
bot_name = bot_name[:-3]
return bot_name, bot_func, bot_data
def setup_env(root_dir, bot_name):
os.environ["LLM_ROOT_DIR"] = root_dir
load_env(os.path.join(root_dir, ".env"))
os.environ["LLM_BOT_NAME"] = bot_name
os.environ["LLM_BOT_ROOT_DIR"] = os.path.join(root_dir, "bots", bot_name)
os.environ["LLM_BOT_CACHE_DIR"] = os.path.join(root_dir, "cache", bot_name)
def load_env(file_path):
try:
with open(file_path, "r") as f:
for line in f:
line = line.strip()
if line.startswith("#") or line == "":
continue
key, *value = line.split("=")
os.environ[key.strip()] = "=".join(value).strip()
except FileNotFoundError:
pass
def run(bot_path, bot_func, bot_data):
try:
spec = importlib.util.spec_from_file_location(
os.path.basename(bot_path), bot_path
)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
except:
raise Exception(f"Unable to load bot tools at '{bot_path}'")
if not hasattr(mod, bot_func):
raise Exception(f"Not module function '{bot_func}' at '{bot_path}'")
value = getattr(mod, bot_func)(**bot_data)
dump_value(value)
def dump_value(value):
if value is None:
return
value_type = type(value).__name__
if value_type in ("str", "int", "float", "bool"):
print(value)
elif value_type == "dict" or value_type == "list":
value_str = json.dumps(value, indent=2)
assert value == json.loads(value_str)
print(value_str)
if __name__ == "__main__":
main()
|
