libs.functions.tools.comparison

libs/functions/tools/comparison.py

View Source

 1"""
 2libs/functions/tools/comparison.py
 3"""
 4
 5import os
 6from typing import TYPE_CHECKING, cast
 7
 8from slack_bolt import App
 9from slack_sdk import WebClient
10
11import libs.global_value as g
12from integrations import factory
13from integrations.slack.events import comparison
14from libs import configuration
15
16if TYPE_CHECKING:
17    from integrations.slack.adapter import ServiceAdapter
18    from libs.datamodels import ComparisonResults
19
20
21def main():
22    """データ突合処理"""
23
24    if g.args.compar:
25        try:
26            g.adapter = cast("ServiceAdapter", g.adapter)
27            app = App(token=os.environ["SLACK_BOT_TOKEN"])
28            configuration.read_memberslist(False)
29        except Exception as err:
30            raise RuntimeError(err) from err
31
32        adapter_slack = factory.select_adapter("slack", g.cfg)
33        adapter_slack.api.webclient = WebClient(token=os.environ["SLACK_WEB_TOKEN"])
34        adapter_slack.api.appclient = app.client
35        adapter_slack.conf.bot_id = app.client.auth_test()["user_id"]
36        g.adapter = adapter_slack
37
38        adapter_std = factory.select_adapter("standard_io", g.cfg)
39        m = adapter_std.parser()
40        m.data.channel_id = adapter_slack.functions.get_channel_id()
41
42        comparison.main(m)
43        print(cast("ComparisonResults", m.status.message).output("summary"))

def main(): View Source

22def main():
23    """データ突合処理"""
24
25    if g.args.compar:
26        try:
27            g.adapter = cast("ServiceAdapter", g.adapter)
28            app = App(token=os.environ["SLACK_BOT_TOKEN"])
29            configuration.read_memberslist(False)
30        except Exception as err:
31            raise RuntimeError(err) from err
32
33        adapter_slack = factory.select_adapter("slack", g.cfg)
34        adapter_slack.api.webclient = WebClient(token=os.environ["SLACK_WEB_TOKEN"])
35        adapter_slack.api.appclient = app.client
36        adapter_slack.conf.bot_id = app.client.auth_test()["user_id"]
37        g.adapter = adapter_slack
38
39        adapter_std = factory.select_adapter("standard_io", g.cfg)
40        m = adapter_std.parser()
41        m.data.channel_id = adapter_slack.functions.get_channel_id()
42
43        comparison.main(m)
44        print(cast("ComparisonResults", m.status.message).output("summary"))

データ突合処理