libs.functions.search

  1"""
  2libs/functions/search.py
  3"""
  4
  5import logging
  6import re
  7from contextlib import closing
  8from typing import cast
  9
 10import libs.global_value as g
 11from cls.score import GameResult
 12from cls.timekit import ExtendedDatetime as ExtDt
 13from cls.types import SlackSearchData
 14from libs.utils import dbutil, formatter, validator
 15
 16SlackSearchDict = dict[str, SlackSearchData]
 17DBSearchDict = dict[str, GameResult]
 18
 19
 20def slack_messages(word: str) -> SlackSearchDict:
 21    """slackログからメッセージを検索して返す
 22
 23    Args:
 24        word (str): 検索するワード
 25
 26    Returns:
 27        SlackSearchDict: 検索した結果
 28    """
 29
 30    # 検索クエリ
 31    after = ExtDt(days=-g.cfg.search.after).format("ymd", "-")
 32    query = f"{word} in:{g.cfg.search.channel} after:{after}"
 33    logging.info("query=%s", query)
 34
 35    # データ取得
 36    response = g.webclient.search_messages(
 37        query=query,
 38        sort="timestamp",
 39        sort_dir="asc",
 40        count=100
 41    )
 42    matches = response["messages"]["matches"]  # 1ページ目
 43    for p in range(2, response["messages"]["paging"]["pages"] + 1):
 44        response = g.webclient.search_messages(
 45            query=query,
 46            sort="timestamp",
 47            sort_dir="asc",
 48            count=100,
 49            page=p
 50        )
 51        matches += response["messages"]["matches"]  # 2ページ目以降
 52
 53    # 必要なデータだけ辞書に格納
 54    data: SlackSearchDict = cast(SlackSearchDict, {})
 55    for x in matches:
 56        if isinstance(x, dict):
 57            data[x["ts"]] = {
 58                "channel_id": str(cast(dict, x["channel"]).get("id", "")),
 59                "user_id": str(x.get("user", "")),
 60                "text": str(x.get("text", "")),
 61            }
 62
 63    return data
 64
 65
 66def get_message_details(matches: SlackSearchDict) -> SlackSearchDict:
 67    """メッセージ詳細情報取得
 68
 69    Args:
 70        matches (SlackSearchDict): 対象データ
 71
 72    Returns:
 73        SlackSearchDict: 詳細情報追加データ
 74    """
 75
 76    # 詳細情報取得
 77    for key, val in matches.items():
 78        res: dict = {}
 79        if isinstance(channel_id := val.get("channel_id"), str):
 80            conversations = g.app.client.conversations_replies(channel=channel_id, ts=key)
 81            if (msg := conversations.get("messages")):
 82                res = cast(dict, msg[0])
 83        else:
 84            continue
 85
 86        if res:
 87            # 各種時間取得
 88            matches[key].update({"event_ts": res.get("ts")})  # イベント発生時間
 89            matches[key].update({"thread_ts": res.get("thread_ts")})  # スレッドの先頭
 90            matches[key].update({"edited_ts": cast(dict, res.get("edited", {})).get("ts")})  # 編集時間
 91            # リアクション取得
 92            reaction_ok, reaction_ng = reactions_list(res)
 93            matches[key].update({"reaction_ok": reaction_ok})
 94            matches[key].update({"reaction_ng": reaction_ng})
 95            # スレッド内フラグ
 96            if val.get("event_ts") == val.get("thread_ts") or val.get("thread_ts") is None:
 97                matches[key].update({"in_thread": False})
 98            else:
 99                matches[key].update({"in_thread": True})
100
101    return matches
102
103
104def for_slack_score() -> SlackSearchDict:
105    """過去ログからスコア記録を検索して返す
106
107    Returns:
108        SlackSearchDict: 検索した結果
109    """
110
111    matches = slack_messages(g.cfg.search.keyword)
112
113    # ゲーム結果の抽出
114    for key in list(matches.keys()):
115        if (detection := validator.pattern(matches[key].get("text", ""))):
116            detection.calc(ts=key)
117            if matches[key].get("user_id", "") in g.cfg.setting.ignore_userid:  # 除外ユーザからのポストは破棄
118                logging.info("skip ignore user: %s (%s)", matches[key]["user_id"], detection)
119                matches.pop(key)
120                continue
121            matches[key]["score"] = detection
122            matches[key].pop("text")
123        else:  # 不一致は破棄
124            matches.pop(key)
125
126    # 結果が無い場合は空の辞書を返して後続の処理をスキップ
127    if not matches:
128        return cast(SlackSearchDict, {})
129
130    matches = get_message_details(matches)
131    g.msg.channel_type = "search_messages"
132    return matches
133
134
135def for_slack_remarks() -> SlackSearchDict:
136    """slackログからメモを検索して返す
137
138    Returns:
139        SlackSearchDict: 検索した結果
140    """
141
142    matches = slack_messages(g.cfg.cw.remarks_word)
143
144    # メモの抽出
145    for key in list(matches.keys()):
146        if re.match(rf"^{g.cfg.cw.remarks_word}", matches[key].get("text", "")):  # キーワードが先頭に存在するかチェック
147            text = matches[key]["text"].replace(g.cfg.cw.remarks_word, "").strip().split()
148            if matches[key].get("user_id", "") in g.cfg.setting.ignore_userid:  # 除外ユーザからのポストは破棄
149                logging.info("skip ignore user: %s, (%s)", matches[key]["user_id"], text)
150                matches.pop(key)
151                continue
152            matches[key]["remarks"] = []
153            g.params.update(unregistered_replace=False)  # 名前ブレを修正(ゲスト無効)
154            for name, matter in zip(text[0::2], text[1::2]):
155                matches[key]["remarks"].append((formatter.name_replace(name, False), matter))
156            matches[key].pop("text")
157        else:  # 不一致は破棄
158            matches.pop(key)
159
160    # 結果が無い場合は空の辞書を返して後続の処理をスキップ
161    if not matches:
162        return cast(SlackSearchDict, {})
163
164    matches = get_message_details(matches)
165    g.msg.channel_type = "search_messages"
166    return matches
167
168
169def for_db_score(first_ts: float | bool = False) -> DBSearchDict:
170    """データベースからスコアを検索して返す
171
172    Args:
173        first_ts (Union[float, bool], optional): 検索を開始する時刻. Defaults to False.
174
175    Returns:
176        DBSearchDict: 検索した結果
177    """
178
179    if not first_ts:
180        return {}
181
182    data: DBSearchDict = {}
183    with closing(dbutil.get_connection()) as conn:
184        curs = conn.cursor()
185        rows = curs.execute("select * from result where ts >= ?", (str(first_ts),))
186        for row in rows.fetchall():
187            ts = str(dict(row).get("ts", ""))
188            result = GameResult()
189            result.set(**dict(row))
190            data[ts] = result
191
192    return data
193
194
195def for_db_remarks(first_ts: float | bool = False) -> list:
196    """データベースからメモを検索して返す
197
198    Args:
199        first_ts (Union[float, bool], optional): 検索を開始する時刻. Defaults to False.
200
201    Returns:
202        list: 検索した結果
203    """
204
205    if not first_ts:
206        return []
207
208    # データベースからデータ取得
209    data: list = []
210    with closing(dbutil.get_connection()) as cur:
211        # 記録済みメモ内容
212        rows = cur.execute("select * from remarks where thread_ts>=?", (str(first_ts),))
213        for row in rows.fetchall():
214            data.append({
215                "thread_ts": row["thread_ts"],
216                "event_ts": row["event_ts"],
217                "name": row["name"],
218                "matter": row["matter"],
219            })
220
221    return data
222
223
224def reactions_list(msg: dict) -> tuple[list, list]:
225    """botが付けたリアクションを取得
226
227    Args:
228        msg (dict): メッセージ内容
229
230    Returns:
231        tuple[list, list]:
232        - reaction_ok: okが付いているメッセージのタイムスタンプ
233        - reaction_ng: ngが付いているメッセージのタイムスタンプ
234    """
235
236    reaction_ok: list = []
237    reaction_ng: list = []
238
239    if msg.get("reactions"):
240        for reactions in msg.get("reactions", {}):
241            if isinstance(reactions, dict) and g.bot_id in reactions.get("users", []):
242                match reactions.get("name"):
243                    case g.cfg.setting.reaction_ok:
244                        reaction_ok.append(msg.get("ts"))
245                    case g.cfg.setting.reaction_ng:
246                        reaction_ng.append(msg.get("ts"))
247
248    return (reaction_ok, reaction_ng)