libs.functions.search

libs/functions/search.py

 1"""
 2libs/functions/search.py
 3"""
 4
 5from contextlib import closing
 6from typing import TYPE_CHECKING
 7
 8import libs.global_value as g
 9from cls.score import GameResult
10from libs.utils import dbutil
11
12if TYPE_CHECKING:
13    from libs.types import RemarkDict
14
15
16def for_db_score(first_ts: float) -> list[GameResult]:
17    """データベースからスコアを検索して返す
18
19    Args:
20        first_ts (float): 検索を開始する時刻
21
22    Returns:
23        list[GameResult]: 検索した結果
24    """
25
26    data: list = []
27    with closing(dbutil.connection(g.cfg.setting.database_file)) as conn:
28        curs = conn.cursor()
29        rows = curs.execute(
30            "select * from result where ts >= ? and source like ?",
31            (str(first_ts), f"{g.adapter.interface_type}_%"))
32        for row in rows.fetchall():
33            data.append(GameResult(**dict(row)))
34
35    return data
36
37
38def for_db_remarks(first_ts: float) -> list["RemarkDict"]:
39    """データベースからメモを検索して返す
40
41    Args:
42        first_ts (float): 検索を開始する時刻
43
44    Returns:
45        list[RemarkDict]: 検索した結果
46    """
47
48    data: list["RemarkDict"] = []
49    with closing(dbutil.connection(g.cfg.setting.database_file)) as cur:
50        # 記録済みメモ内容
51        rows = cur.execute(
52            dbutil.query("REMARKS_SELECT"),
53            (str(first_ts), f"{g.adapter.interface_type}_%")
54        )
55        for row in rows.fetchall():
56            data.append({
57                "thread_ts": row["thread_ts"],
58                "event_ts": row["event_ts"],
59                "name": row["name"],
60                "matter": row["matter"],
61                "source": row["source"],
62            })
63
64    return data
def for_db_score(first_ts: float) -> list[cls.score.GameResult]:
17def for_db_score(first_ts: float) -> list[GameResult]:
18    """データベースからスコアを検索して返す
19
20    Args:
21        first_ts (float): 検索を開始する時刻
22
23    Returns:
24        list[GameResult]: 検索した結果
25    """
26
27    data: list = []
28    with closing(dbutil.connection(g.cfg.setting.database_file)) as conn:
29        curs = conn.cursor()
30        rows = curs.execute(
31            "select * from result where ts >= ? and source like ?",
32            (str(first_ts), f"{g.adapter.interface_type}_%"))
33        for row in rows.fetchall():
34            data.append(GameResult(**dict(row)))
35
36    return data

データベースからスコアを検索して返す

Arguments:
  • first_ts (float): 検索を開始する時刻
Returns:

list[GameResult]: 検索した結果

def for_db_remarks(first_ts: float) -> list[libs.types.RemarkDict]:
39def for_db_remarks(first_ts: float) -> list["RemarkDict"]:
40    """データベースからメモを検索して返す
41
42    Args:
43        first_ts (float): 検索を開始する時刻
44
45    Returns:
46        list[RemarkDict]: 検索した結果
47    """
48
49    data: list["RemarkDict"] = []
50    with closing(dbutil.connection(g.cfg.setting.database_file)) as cur:
51        # 記録済みメモ内容
52        rows = cur.execute(
53            dbutil.query("REMARKS_SELECT"),
54            (str(first_ts), f"{g.adapter.interface_type}_%")
55        )
56        for row in rows.fetchall():
57            data.append({
58                "thread_ts": row["thread_ts"],
59                "event_ts": row["event_ts"],
60                "name": row["name"],
61                "matter": row["matter"],
62                "source": row["source"],
63            })
64
65    return data

データベースからメモを検索して返す

Arguments:
  • first_ts (float): 検索を開始する時刻
Returns:

list[RemarkDict]: 検索した結果