wachter-bot/app.py at dev · wachter-org/wachter-bot · GitHub

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
from telegram.ext import (
    ApplicationBuilder,
    CommandHandler,
    filters,
    MessageHandler,
    CallbackQueryHandler,
    ChatMemberHandler,
    PicklePersistence,
)
from ptbcontrib.ptb_jobstores.sqlalchemy import PTBSQLAlchemyJobStore

import grpc
from opentelemetry import metrics
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import (
    OTLPMetricExporter,
)
from opentelemetry.sdk import metrics as sdkmetrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import (
    AggregationTemporality,
    PeriodicExportingMetricReader,
)
from opentelemetry.sdk.resources import Resource

from src.custom_filters import filter_bot_added
from src.logging import tg_logger
from src import handlers
import os


temporality_delta = {
    sdkmetrics.Counter: AggregationTemporality.DELTA,
    sdkmetrics.UpDownCounter: AggregationTemporality.DELTA,
    sdkmetrics.Histogram: AggregationTemporality.DELTA,
    sdkmetrics.ObservableCounter: AggregationTemporality.DELTA,
    sdkmetrics.ObservableUpDownCounter: AggregationTemporality.DELTA,
    sdkmetrics.ObservableGauge: AggregationTemporality.DELTA,
}


def main():
    dsn = os.environ.get("UPTRACE_DSN")

    exporter = OTLPMetricExporter(
        endpoint="otlp.uptrace.dev:4317",
        headers=(("uptrace-dsn", dsn),),
        timeout=5,
        compression=grpc.Compression.Gzip,
        preferred_temporality=temporality_delta,
    )
    reader = PeriodicExportingMetricReader(exporter)

    resource = Resource(
        attributes={
            "service.name": "wachter",
            "service.version": "1.1.0",
            "deployment.environment": os.environ.get("DEPLOYMENT_ENVIRONMENT"),
        }
    )
    provider = MeterProvider(metric_readers=[reader], resource=resource)
    metrics.set_meter_provider(provider)

    application = (
        ApplicationBuilder()
        .persistence(PicklePersistence(filepath="persistent_storage.pickle"))
        .token(os.environ["TELEGRAM_TOKEN"])
        .build()
    )
    if "PERSISTENCE_DATABASE_URL" in os.environ:
        tg_logger.info(f"Using SQLAlchemy job store with PERSISTENCE_DATABASE_URL")
        application.job_queue.scheduler.add_jobstore(
            PTBSQLAlchemyJobStore(
                application=application,
                url=os.environ["PERSISTENCE_DATABASE_URL"],
            )
        )
    else:
        tg_logger.info("No PERSISTENCE_DATABASE_URL set, using in-memory job store")

    application.add_handler(CommandHandler("help", handlers.help_handler))
    application.add_handler(CommandHandler("listjobs", handlers.list_jobs_handler))

    # group UX
    application.add_handler(
        ChatMemberHandler(
            handlers.my_chat_member_handler,
            ChatMemberHandler.MY_CHAT_MEMBER,
        )
    )
    application.add_handler(
        MessageHandler(
            filters.Entity("hashtag") & filters.ChatType.GROUPS,
            handlers.on_hashtag_message,
        )
    )
    application.add_handler(
        MessageHandler(
            filters.StatusUpdate.NEW_CHAT_MEMBERS & filter_bot_added,
            handlers.on_new_chat_members,
        )
    )

    # admin UX
    application.add_handler(CommandHandler("start", handlers.start_handler))
    application.add_handler(CallbackQueryHandler(handlers.button_handler))
    application.add_handler(MessageHandler(filters.TEXT, handlers.message_handler))
    application.add_error_handler(handlers.error_handler)

    # Remove any existing metrics_exporter jobs to avoid duplicates
    existing_jobs = [job for job in application.job_queue.jobs() if job.name == "metrics_exporter"]
    if existing_jobs:
        tg_logger.info(f"Removing {len(existing_jobs)} existing metrics_exporter job(s)")
        for job in existing_jobs:
            job.schedule_removal()

    job = application.job_queue.run_repeating(
        handlers.group.group_handler.db_metrics_reader_helper,
        3600,
        name="metrics_exporter",
    )
    tg_logger.info(f"Scheduled metrics_exporter job: {job.name}")

    tg_logger.info("Bot has started successfully")
    application.run_polling()


if __name__ == "__main__":
    main()