-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathmain.py
148 lines (127 loc) · 4.95 KB
/
main.py
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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
import asyncio
import os
import tomllib
from argparse import ArgumentParser
from types import SimpleNamespace
import dotenv
from discord import Intents
from api.main import ActApi
from bot.main import ActBot
from db.main import ActDb
from utils.log import logger
log = logger(__name__)
# ----------------------------------------------------------------------------------------------------
# * PyProject
# ----------------------------------------------------------------------------------------------------
# Load and parse pyproject.toml file
with open("pyproject.toml", "rb") as file:
PYPROJECT = tomllib.load(file)
PROJECT = PYPROJECT.get("project", {})
TOOL_ACT_COMPONENTS = PYPROJECT.get("tool", {}).get("act", {}).get("components", {})
# ----------------------------------------------------------------------------------------------------
# * Main
# ----------------------------------------------------------------------------------------------------
async def main():
"""Application main entry point."""
try:
# Retrieve meta-data
name = PROJECT.get("name", "")
version = PROJECT.get("version", "")
description = PROJECT.get("description", "")
# Retrieve environment variables
dotenv.load_dotenv()
db_uri = os.getenv("MONGO_DB_URI")
bot_token = os.getenv("DISCORD_BOT_TOKEN")
server_url = os.getenv("APP_SERVER_URL")
ai_api_key = os.getenv("GEMINI_AI_API_KEY")
# Clear console & set title (Based on OS)
os.system(
f"cls && title ✈ {name}" # For Windows
if os.name == "nt"
else f'clear && printf "\033]0;🚀 {name}\007"' # For Unix-based systems
)
# Get components to enable
db_enabled, bot_enabled, api_enabled = get_components(TOOL_ACT_COMPONENTS)
# Prepare list of components to run
coroutines = []
# Create & add database component
db = None
if db_enabled:
db = ActDb(host=db_uri, name=name)
else:
log.warning("Database component is turned off.")
# Create & add bot component
bot = None
if bot_enabled:
intents = Intents.default()
intents.message_content = True
bot = ActBot(
token=bot_token or "",
command_prefix="!",
intents=intents,
db=db,
api_keys={"gemini": ai_api_key or ""},
title=name,
version=version,
description=description,
)
coroutines.append(bot.open())
else:
log.warning("Bot component is turned off.")
# Create & add api component
api = None
if api_enabled:
api = ActApi(
bot=bot,
url=server_url or "",
title=name,
version=version,
description=description,
)
coroutines.append(api.open())
else:
log.warning("API component is turned off.")
# Run all components asynchronously
await asyncio.gather(*coroutines)
except (asyncio.CancelledError, KeyboardInterrupt, Exception) as e:
if isinstance(e, KeyboardInterrupt):
log.info("Keyboard interrupt received.")
elif not isinstance(e, asyncio.CancelledError):
log.exception(e)
if db:
db.close()
if bot:
await bot.close()
if api:
await api.close()
finally:
print("\n❤ Bye!\n")
# ----------------------------------------------------------------------------------------------------
# * Title
# ----------------------------------------------------------------------------------------------------
def get_components(config: dict[str, bool]):
parser = ArgumentParser(description="run app components")
parser.add_argument(
"-d", "--db", action="store_true", help="enable db client component"
)
parser.add_argument(
"-b", "--bot", action="store_true", help="enable bot client component"
)
parser.add_argument(
"-a", "--api", action="store_true", help="enable api server component"
)
args = parser.parse_args()
db_enabled = args.db if args.db else config.get("db", False)
bot_enabled = args.bot if args.bot else config.get("bot", False)
api_enabled = args.api if args.api else config.get("api", False)
if not (bot_enabled or api_enabled):
log.warning(
f"No app component specified in command options or project settings.\n\n{parser.format_help()}"
)
raise SystemExit(1)
return (db_enabled, bot_enabled, api_enabled)
# ----------------------------------------------------------------------------------------------------
# * Run Application
# ----------------------------------------------------------------------------------------------------
if __name__ == "__main__":
asyncio.run(main())