This repository has been archived by the owner on May 10, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 21
/
main.py
executable file
·90 lines (71 loc) · 2.43 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
#!/bin/env python
"""The main entry point for the stickerfinder."""
from contextlib import contextmanager
import typer
from sqlalchemy_utils.functions import database_exists, create_database, drop_database
from stickerfinder.config import config
from stickerfinder.db import engine, base
from stickerfinder.models import * # noqa
from stickerfinder.stickerfinder import init_app
cli = typer.Typer()
@contextmanager
def wrap_echo(msg: str):
typer.echo(f"{msg}... ", nl=False)
yield
typer.echo("done.")
@cli.command()
def initdb(exist_ok: bool = False, drop_existing: bool = False):
"""Set up the database.
Can be used to remove an existing database.
"""
db_url = engine.url
typer.echo(f"Using database at {db_url}")
if database_exists(db_url):
if drop_existing:
with wrap_echo("Dropping database"):
drop_database(db_url)
elif not exist_ok:
typer.echo(
"Database already exists, aborting.\n"
"Use --exist-ok if you are sure the database is uninitialized and contains no data.\n"
"Use --drop-existing if you want to recreate it.",
err=True,
)
return
with wrap_echo("Creating database"):
create_database(db_url)
pass
with engine.connect() as con:
with wrap_echo("Installing extensions"):
con.execute("CREATE EXTENSION IF NOT EXISTS pgcrypto;")
con.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm;")
pass
with wrap_echo("Creating metadata"):
base.metadata.create_all()
pass
typer.echo("Database initialization complete.")
@cli.command()
def run():
"""Actually start the bot."""
updater = init_app()
if config["webhook"]["enabled"]:
typer.echo("Starting the bot in webhook mode.")
domain = config["webhook"]["domain"]
token = config["webhook"]["token"]
print("Starting up")
updater.start_webhook(
listen="127.0.0.1",
port=config["webhook"]["port"],
url_path=config["webhook"]["token"],
webhook_url=f"{domain}{token}",
cert=config["webhook"]["cert_path"],
)
else:
typer.echo("Starting the bot in polling mode.")
updater.start_polling()
print("Starting up")
updater.idle()
if __name__ == "__main__":
cli()
#!/bin/env python
"""Start the bot."""