-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
136 lines (113 loc) · 4.27 KB
/
main.py
File metadata and controls
136 lines (113 loc) · 4.27 KB
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
# Copyright NGGT.LightKeeper. All Rights Reserved.
import argparse
import threading
import subprocess
from pathlib import Path
from typing import List
import os
import sys
BASE_DIR = Path(__file__).resolve().parent
def run_subprocess(command: List[str]):
"""Runs a command in a subprocess."""
try:
# The process will inherit stdout and stderr from the parent,
# so the output will be printed directly to the console.
subprocess.run(command, check=True)
except subprocess.CalledProcessError as e:
print(f"Command '{' '.join(command)}' returned non-zero exit status {e.returncode}")
# Optionally, exit or handle the error as needed
sys.exit(e.returncode)
def runbotserver():
"""Launches the Telegram bot server."""
print("Starting Telegram bot server...")
run_subprocess(["python", str(BASE_DIR / "bot.py")])
def runwebserver():
"""Starts the Django development web server."""
print("Starting Django development server...")
run_subprocess(["python", str(BASE_DIR / "manage.py"), "runserver"])
def makemigrations():
"""Creates new database migrations."""
print("Creating database migrations...")
run_subprocess(["python", str(BASE_DIR / "manage.py"), "makemigrations"])
def migrate():
"""Applies database migrations."""
print("Applying database migrations...")
run_subprocess(["python", str(BASE_DIR / "manage.py"), "migrate"])
def setup_django():
"""Configures Django to run management commands."""
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'telegram_bot_django.settings')
import django
django.setup()
def export_db(path: str):
"""Exports the database to a JSON file."""
setup_django()
from django.core.management import call_command
try:
with open(path, 'w', encoding='utf-8') as f:
call_command('dumpdata', '--indent', '2', stdout=f)
print(f'Database exported successfully to {path}')
except Exception as e:
print(f"Failed to export database: {e}")
def import_db(path: str):
"""Imports the database from a JSON file."""
setup_django()
from django.core.management import call_command
try:
call_command('loaddata', path)
print(f'Database imported successfully from {path}')
except Exception as e:
print(f"Failed to import database: {e}")
mess_help = '''
Available commands:
runserver - Runs both the bot and web servers.
runbotserver - Runs the Telegram bot server only.
runwebserver - Runs the Django web server only.
makemigrations - Creates new database migrations.
migrate - Applies database migrations.
db import <file> - Imports database from a JSON file.
db export <file> - Exports database to a JSON file.
help - Shows this help message.
'''
def main():
"""Parses command line arguments and executes the corresponding command."""
parser = argparse.ArgumentParser(description="Manage the Telegram bot and web server.")
parser.add_argument('command', nargs='+', help="The command to execute.")
args = parser.parse_args()
command = args.command
cmd = command[0]
if cmd == 'db':
if len(command) < 3:
print("Usage: python main.py db <import|export> <path>")
return
action = command[1]
path = command[2]
print(f"Running command: db {action}")
if action == 'export':
export_db(path)
elif action == 'import':
import_db(path)
else:
print(f"Unknown db command: '{action}'. Use 'import' or 'export'.")
return
if cmd == 'runserver':
print("Running both bot and web servers in parallel.")
thread1 = threading.Thread(target=runbotserver)
thread2 = threading.Thread(target=runwebserver)
thread1.start()
thread2.start()
thread1.join()
thread2.join()
elif cmd == 'runbotserver':
runbotserver()
elif cmd == 'runwebserver':
runwebserver()
elif cmd == 'makemigrations':
makemigrations()
elif cmd == 'migrate':
migrate()
elif cmd == 'help':
print(mess_help)
else:
print(f"Unknown command: '{cmd}'. Use 'help' to see available commands.")
if __name__ == '__main__':
main()