/
start_receiver.py
77 lines (70 loc) Β· 2.7 KB
/
start_receiver.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
# -*- coding: utf-8 -*-
# @Time : 2023/10/18 δΈε10:23
import getopt
import os
import sys
from dotenv import load_dotenv
from loguru import logger
load_dotenv()
logger.remove(0)
handler_id = logger.add(
sys.stderr,
format="<level>[{level}]</level> | <level>{message}</level> | "
"<cyan>{name}:{function}:{line}</cyan> <yellow>@{time}</yellow>",
colorize=True,
backtrace=True,
enqueue=True,
level="DEBUG" if os.getenv("DEBUG", None) else "INFO",
)
logger.add(
sink="receiver.log",
format="<level>[{level}]</level> | <level>{message}</level> | "
"<cyan>{name}:{function}:{line}</cyan> <yellow>@{time}</yellow>",
level="DEBUG",
rotation="100 MB",
enqueue=True,
)
head = """
βββ βββ ββββ βββββββ βββββββββββββ ββββββ
βββ βββ βββββ ββββββββ βββββββββββββββββββββββ
βββ βββ ββββββββββββββββββ βββββββββββββββββββ
βββ βββ ββββββββββββββββββ βββββββββββββββββββ
βββββββββββββββββββ βββ ββββββ βββββββββ ββββββ βββ
βββββββββββββββββββ ββββββ βββββββββ ββββββ βββ
"""
logger.opt(record=False, exception=False, capture=False, colors=True).info(
f"<cyan>{head}</cyan>"
)
if os.getenv("DEBUG", None):
logger.warning("DEBUG MODE IS OPEN")
# Log System
if os.getenv("SENTRY_DSN", None):
try:
import sentry_sdk
sentry_sdk.init(
dsn=os.getenv("SENTRY_DSN"),
traces_sample_rate=1.0,
profiles_sample_rate=1.0,
)
except Exception as e:
logger.error(f"SENTRY ERROR: {e}")
else:
logger.success("π Create Sentry Client Successfully!")
# Tutorial
SKIP_TUTORIAL = False
SKIP_EXISTING = True
opts, args = getopt.getopt(sys.argv[1:], "h", ["no_tutorial", "tutorial"])
for op, value in opts:
if op == "--no_tutorial": # θ·εε½δ»€θ‘εζ°η --no_tutorial
SKIP_TUTORIAL = True
if op == "-h":
print("Usage: python start_receiver.py [--no_tutorial] [--tutorial]")
sys.exit()
if op == "--tutorial":
SKIP_EXISTING = False
if not SKIP_TUTORIAL:
from app.tutorial import show_tutorial
show_tutorial(skip_existing=SKIP_EXISTING, pre_step_stop=4, database_key="01")
# Run Receiver
from app.receiver import app # noqa
app.run()