84 lines
3 KiB
Python
84 lines
3 KiB
Python
import sqlite3
|
|
import config
|
|
|
|
|
|
def _connect():
|
|
conn = sqlite3.connect(config.DB_FILE)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def init_db():
|
|
with _connect() as conn:
|
|
conn.execute("""
|
|
CREATE TABLE IF NOT EXISTS messages (
|
|
task_id INTEGER PRIMARY KEY,
|
|
message_id INTEGER NOT NULL,
|
|
task_number TEXT
|
|
)
|
|
""")
|
|
# Универсальная таблица справочников: entity = 'status'|'priority'|'service'|...
|
|
conn.execute("""
|
|
CREATE TABLE IF NOT EXISTS ref_data (
|
|
entity TEXT NOT NULL,
|
|
id INTEGER NOT NULL,
|
|
name TEXT NOT NULL,
|
|
PRIMARY KEY (entity, id)
|
|
)
|
|
""")
|
|
conn.commit()
|
|
|
|
|
|
# ── messages ──────────────────────────────────────────────────────────────────
|
|
|
|
def get_message_id(task_id: int):
|
|
with _connect() as conn:
|
|
row = conn.execute(
|
|
"SELECT message_id FROM messages WHERE task_id = ?", (task_id,)
|
|
).fetchone()
|
|
return row["message_id"] if row else None
|
|
|
|
|
|
def save_message_id(task_id: int, message_id: int, task_number: str = None):
|
|
with _connect() as conn:
|
|
conn.execute(
|
|
"INSERT OR REPLACE INTO messages (task_id, message_id, task_number) VALUES (?, ?, ?)",
|
|
(task_id, message_id, task_number),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
# ── ref_data ──────────────────────────────────────────────────────────────────
|
|
|
|
def save_ref_data(entity: str, items: list[tuple[int, str]]):
|
|
"""Перезаписывает весь справочник entity списком пар (id, name)."""
|
|
with _connect() as conn:
|
|
conn.execute("DELETE FROM ref_data WHERE entity = ?", (entity,))
|
|
conn.executemany(
|
|
"INSERT INTO ref_data (entity, id, name) VALUES (?, ?, ?)",
|
|
[(entity, item_id, name) for item_id, name in items],
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def resolve(entity: str, item_id: int | None) -> str | None:
|
|
"""Возвращает name по entity + id. None если не найдено."""
|
|
if item_id is None:
|
|
return None
|
|
with _connect() as conn:
|
|
row = conn.execute(
|
|
"SELECT name FROM ref_data WHERE entity = ? AND id = ?",
|
|
(entity, item_id),
|
|
).fetchone()
|
|
return row["name"] if row else None
|
|
|
|
|
|
def get_all_ref(entity: str) -> list[dict]:
|
|
"""Возвращает все записи справочника как список {id, name}."""
|
|
with _connect() as conn:
|
|
rows = conn.execute(
|
|
"SELECT id, name FROM ref_data WHERE entity = ? ORDER BY name",
|
|
(entity,),
|
|
).fetchall()
|
|
return [{"id": r["id"], "name": r["name"]} for r in rows]
|
|
|