libs.functions.search

  1"""
  2libs/data/search.py
  3"""
  4
  5import logging
  6from contextlib import closing
  7from typing import TYPE_CHECKING
  8
  9import libs.global_value as g
 10from libs.domain.score import GameResult
 11from libs.utils import dbutil
 12from libs.utils.timekit import ExtendedDatetime as ExtDt
 13
 14if TYPE_CHECKING:
 15    from libs.types import RemarkDict
 16
 17
 18def for_db_score(first_ts: float) -> list[GameResult]:
 19    """
 20    データベースからスコアを検索して返す
 21
 22    Args:
 23        first_ts (float): 検索を開始する時刻
 24
 25    Returns:
 26        list[GameResult]: 検索した結果
 27
 28    """
 29    data: list[GameResult] = []
 30    rows = dbutil.execute(
 31        "select * from result where ts >= :first_ts and source like :source",
 32        {"first_ts": str(first_ts), "source": f"{g.adapter.interface_type}_%"},
 33    )
 34
 35    for row in rows:
 36        data.append(GameResult(**row))
 37
 38    logging.debug(data)
 39    return data
 40
 41
 42def for_db_remarks(first_ts: float) -> list["RemarkDict"]:
 43    """
 44    データベースからメモを検索して返す
 45
 46    Args:
 47        first_ts (float): 検索を開始する時刻
 48
 49    Returns:
 50        list[RemarkDict]: 検索した結果
 51
 52    """
 53    data: list["RemarkDict"] = []
 54    with closing(dbutil.connection(g.cfg.setting.database_file)) as cur:
 55        # 記録済みメモ内容
 56        rows = cur.execute(
 57            dbutil.query("REMARKS_SELECT"),
 58            (str(first_ts), f"{g.adapter.interface_type}_%"),
 59        )
 60        for row in rows.fetchall():
 61            data.append(
 62                {
 63                    "thread_ts": row["thread_ts"],
 64                    "event_ts": row["event_ts"],
 65                    "name": row["name"],
 66                    "matter": row["matter"],
 67                    "source": row["source"],
 68                }
 69            )
 70    logging.debug(data)
 71    return data
 72
 73
 74def first_record(rule_list: list[str]) -> ExtDt:
 75    """
 76    最初のゲーム記録時間を返す
 77
 78    Args:
 79        rule_list (list[str]): ルール識別子
 80
 81    Returns:
 82        ExtendedDatetime: 最初のゲーム記録時間
 83
 84    """
 85    ret = ExtDt()
 86    rule_dict = {f"rule_{idx}": name for idx, name in enumerate(set(rule_list))}
 87
 88    try:
 89        with closing(dbutil.connection(g.cfg.setting.database_file)) as conn:
 90            table_count = conn.execute(
 91                "select count() from sqlite_master where type='view' and name='game_results';",
 92            ).fetchall()[0][0]
 93
 94            if table_count:
 95                sql = "select min(playtime) from game_results where rule_version in (<<rule_list>>);".replace("<<rule_list>>", ":" + ", :".join(rule_dict))
 96                record = conn.execute(sql, rule_dict).fetchall()[0][0]
 97                if record:
 98                    ret = ExtDt(str(record)) - {"hour": 0, "minute": 0, "second": 0, "microsecond": 0, "hours": g.cfg.setting.time_adjust}
 99    except AttributeError:
100        ret = ExtDt()
101
102    return ret