Add scheduler system with cron support
- Implement QuestionPublisherScheduler with full cron expression parser - Add schedule management in /manage command (add, edit, delete) - Create schedules and schedule_runs tables in database - Add delete_schedule method to repository - Create comprehensive SCHEDULER_GUIDE.md with examples - Add README.md with project overview - Support for active/inactive schedules - Automatic question publishing based on cron schedules - Protection against duplicate runs in the same minute Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
27
game/__init__.py
Normal file
27
game/__init__.py
Normal file
@@ -0,0 +1,27 @@
|
||||
from .api_service import (
|
||||
ApiQuestion,
|
||||
QuestionApiService,
|
||||
QuestionPayloadHandler,
|
||||
build_handler_registry,
|
||||
parse_default_question_payload,
|
||||
)
|
||||
from .database import AnswerStatistic, SQLiteWouldYouRatherRepository, ScheduleRow
|
||||
from .game import Question, WouldYouRather
|
||||
from .settings import ApiSourceSettings, AppSettings, AppSettingsService, ConfiguredApiService
|
||||
|
||||
__all__ = [
|
||||
"ApiQuestion",
|
||||
"ApiSourceSettings",
|
||||
"AppSettings",
|
||||
"AppSettingsService",
|
||||
"ConfiguredApiService",
|
||||
"AnswerStatistic",
|
||||
"Question",
|
||||
"QuestionPayloadHandler",
|
||||
"QuestionApiService",
|
||||
"ScheduleRow",
|
||||
"SQLiteWouldYouRatherRepository",
|
||||
"WouldYouRather",
|
||||
"build_handler_registry",
|
||||
"parse_default_question_payload",
|
||||
]
|
||||
133
game/api_service.py
Normal file
133
game/api_service.py
Normal file
@@ -0,0 +1,133 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Callable
|
||||
from urllib.error import HTTPError, URLError
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
from .database import SQLiteWouldYouRatherRepository
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ApiQuestion:
|
||||
text: str
|
||||
answers: list[str]
|
||||
|
||||
|
||||
QuestionPayloadHandler = Callable[[Any], ApiQuestion]
|
||||
|
||||
|
||||
def parse_default_question_payload(payload: Any) -> ApiQuestion:
|
||||
if isinstance(payload, list):
|
||||
if not payload:
|
||||
raise ValueError("API returned an empty list of questions.")
|
||||
payload = payload[0]
|
||||
|
||||
if not isinstance(payload, dict):
|
||||
raise ValueError("API response must be a JSON object or array.")
|
||||
|
||||
if "data" in payload and isinstance(payload["data"], (dict, list)):
|
||||
return parse_default_question_payload(payload["data"])
|
||||
|
||||
question_text = _extract_question_text(payload)
|
||||
answers = _extract_answers(payload)
|
||||
if len(answers) < 2:
|
||||
raise ValueError("Question must have at least 2 answers.")
|
||||
|
||||
return ApiQuestion(text=question_text, answers=answers)
|
||||
|
||||
|
||||
def build_handler_registry() -> dict[str, QuestionPayloadHandler]:
|
||||
return {
|
||||
"default": parse_default_question_payload,
|
||||
}
|
||||
|
||||
|
||||
class QuestionApiService:
|
||||
def __init__(
|
||||
self,
|
||||
url: str,
|
||||
handler: QuestionPayloadHandler | None = None,
|
||||
timeout: float = 10.0,
|
||||
headers: dict[str, str] | None = None,
|
||||
) -> None:
|
||||
self.url = url
|
||||
self.handler = handler or parse_default_question_payload
|
||||
self.timeout = timeout
|
||||
self.headers = headers or {"Accept": "application/json"}
|
||||
|
||||
def fetch_question(self) -> ApiQuestion:
|
||||
payload = self._fetch_json()
|
||||
return self.handler(payload)
|
||||
|
||||
def fetch_and_store(self, repository: SQLiteWouldYouRatherRepository) -> int:
|
||||
question = self.fetch_question()
|
||||
return repository.add_question(question.text, question.answers)
|
||||
|
||||
def _fetch_json(self) -> Any:
|
||||
request = Request(self.url, headers=self.headers, method="GET")
|
||||
try:
|
||||
with urlopen(request, timeout=self.timeout) as response:
|
||||
charset = response.headers.get_content_charset() or "utf-8"
|
||||
body = response.read().decode(charset)
|
||||
except HTTPError as error:
|
||||
raise RuntimeError(
|
||||
f"API request failed with status code {error.code}."
|
||||
) from error
|
||||
except URLError as error:
|
||||
raise RuntimeError(f"API request failed: {error.reason}.") from error
|
||||
|
||||
try:
|
||||
return json.loads(body)
|
||||
except json.JSONDecodeError as error:
|
||||
raise ValueError("API response is not valid JSON.") from error
|
||||
|
||||
|
||||
def _extract_question_text(payload: dict[str, Any]) -> str:
|
||||
for key in ("question", "text", "title", "prompt"):
|
||||
value = payload.get(key)
|
||||
if isinstance(value, str):
|
||||
normalized = value.strip()
|
||||
if normalized:
|
||||
return normalized
|
||||
raise ValueError("Question text was not found in API response.")
|
||||
|
||||
|
||||
def _extract_answers(payload: dict[str, Any]) -> list[str]:
|
||||
raw_answers: Any = None
|
||||
for key in ("answers", "options", "variants", "choices"):
|
||||
value = payload.get(key)
|
||||
if value is not None:
|
||||
raw_answers = value
|
||||
break
|
||||
|
||||
if raw_answers is None:
|
||||
raise ValueError("Answer variants were not found in API response.")
|
||||
|
||||
if isinstance(raw_answers, dict):
|
||||
iterable = raw_answers.values()
|
||||
elif isinstance(raw_answers, list):
|
||||
iterable = raw_answers
|
||||
else:
|
||||
raise ValueError("Answer variants must be a list or object.")
|
||||
|
||||
answers: list[str] = []
|
||||
for item in iterable:
|
||||
if isinstance(item, str):
|
||||
normalized = item.strip()
|
||||
if normalized:
|
||||
answers.append(normalized)
|
||||
continue
|
||||
|
||||
if isinstance(item, dict):
|
||||
for key in ("text", "answer", "title", "label"):
|
||||
value = item.get(key)
|
||||
if isinstance(value, str):
|
||||
normalized = value.strip()
|
||||
if normalized:
|
||||
answers.append(normalized)
|
||||
break
|
||||
|
||||
unique_answers = list(dict.fromkeys(answers))
|
||||
return unique_answers
|
||||
388
game/database.py
Normal file
388
game/database.py
Normal file
@@ -0,0 +1,388 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlite3
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Iterable
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class QuestionRow:
|
||||
id: int
|
||||
text: str
|
||||
used: bool
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class AnswerRow:
|
||||
id: int
|
||||
question_id: int
|
||||
text: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class QuestionWithAnswers:
|
||||
question: QuestionRow
|
||||
answers: list[AnswerRow]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class AnswerStatistic:
|
||||
answer_id: int
|
||||
answer_text: str
|
||||
count: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class QuestionListItem:
|
||||
id: int
|
||||
text: str
|
||||
used: bool
|
||||
answers_count: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ScheduleRow:
|
||||
id: int
|
||||
name: str
|
||||
cron: str
|
||||
is_active: bool
|
||||
|
||||
|
||||
class SQLiteWouldYouRatherRepository:
|
||||
def __init__(self, db_path: str | Path = "data\\game.sqlite3") -> None:
|
||||
self.db_path = Path(db_path)
|
||||
self.db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
self._initialize()
|
||||
|
||||
def _connect(self) -> sqlite3.Connection:
|
||||
connection = sqlite3.connect(self.db_path)
|
||||
connection.row_factory = sqlite3.Row
|
||||
connection.execute("PRAGMA foreign_keys = ON")
|
||||
return connection
|
||||
|
||||
def _initialize(self) -> None:
|
||||
with self._connect() as connection:
|
||||
connection.executescript(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS questions (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
text TEXT NOT NULL UNIQUE,
|
||||
used INTEGER NOT NULL DEFAULT 0 CHECK (used IN (0, 1))
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS answers (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
question_id INTEGER NOT NULL,
|
||||
text TEXT NOT NULL,
|
||||
FOREIGN KEY (question_id) REFERENCES questions(id) ON DELETE CASCADE,
|
||||
UNIQUE (question_id, text)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS statistics (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
question_id INTEGER NOT NULL,
|
||||
answer_id INTEGER NOT NULL,
|
||||
count INTEGER NOT NULL DEFAULT 0 CHECK (count >= 0),
|
||||
FOREIGN KEY (question_id) REFERENCES questions(id) ON DELETE CASCADE,
|
||||
FOREIGN KEY (answer_id) REFERENCES answers(id) ON DELETE CASCADE,
|
||||
UNIQUE (question_id, answer_id)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS schedules (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
name TEXT NOT NULL,
|
||||
cron TEXT NOT NULL,
|
||||
is_active INTEGER NOT NULL DEFAULT 1 CHECK (is_active IN (0, 1))
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS schedule_runs (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
schedule_id INTEGER NOT NULL,
|
||||
run_key TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
FOREIGN KEY (schedule_id) REFERENCES schedules(id) ON DELETE CASCADE,
|
||||
UNIQUE (schedule_id, run_key)
|
||||
);
|
||||
"""
|
||||
)
|
||||
|
||||
def add_question(self, text: str, answers: Iterable[str]) -> int:
|
||||
prepared_text = text.strip()
|
||||
prepared_answers = [answer.strip() for answer in answers if answer.strip()]
|
||||
if not prepared_text:
|
||||
raise ValueError("Question text cannot be empty.")
|
||||
if len(prepared_answers) < 2:
|
||||
raise ValueError("Question must have at least 2 answers.")
|
||||
|
||||
with self._connect() as connection:
|
||||
cursor = connection.execute(
|
||||
"INSERT INTO questions(text, used) VALUES (?, 0)",
|
||||
(prepared_text,),
|
||||
)
|
||||
question_id = int(cursor.lastrowid)
|
||||
connection.executemany(
|
||||
"INSERT INTO answers(question_id, text) VALUES (?, ?)",
|
||||
[(question_id, answer_text) for answer_text in prepared_answers],
|
||||
)
|
||||
return question_id
|
||||
|
||||
def get_question(self, question_id: int) -> QuestionWithAnswers | None:
|
||||
with self._connect() as connection:
|
||||
question_row = connection.execute(
|
||||
"SELECT id, text, used FROM questions WHERE id = ?",
|
||||
(question_id,),
|
||||
).fetchone()
|
||||
if question_row is None:
|
||||
return None
|
||||
|
||||
answer_rows = connection.execute(
|
||||
"SELECT id, question_id, text FROM answers WHERE question_id = ? ORDER BY id",
|
||||
(question_id,),
|
||||
).fetchall()
|
||||
return QuestionWithAnswers(
|
||||
question=QuestionRow(
|
||||
id=question_row["id"],
|
||||
text=question_row["text"],
|
||||
used=bool(question_row["used"]),
|
||||
),
|
||||
answers=[
|
||||
AnswerRow(
|
||||
id=answer_row["id"],
|
||||
question_id=answer_row["question_id"],
|
||||
text=answer_row["text"],
|
||||
)
|
||||
for answer_row in answer_rows
|
||||
],
|
||||
)
|
||||
|
||||
def count_questions(self) -> int:
|
||||
with self._connect() as connection:
|
||||
row = connection.execute("SELECT COUNT(*) AS total FROM questions").fetchone()
|
||||
return int(row["total"])
|
||||
|
||||
def list_questions(self, limit: int = 10, offset: int = 0) -> list[QuestionListItem]:
|
||||
if limit < 1:
|
||||
raise ValueError("Limit must be >= 1.")
|
||||
if offset < 0:
|
||||
raise ValueError("Offset must be >= 0.")
|
||||
|
||||
with self._connect() as connection:
|
||||
rows = connection.execute(
|
||||
"""
|
||||
SELECT
|
||||
q.id,
|
||||
q.text,
|
||||
q.used,
|
||||
COUNT(a.id) AS answers_count
|
||||
FROM questions q
|
||||
LEFT JOIN answers a ON a.question_id = q.id
|
||||
GROUP BY q.id, q.text, q.used
|
||||
ORDER BY q.id
|
||||
LIMIT ? OFFSET ?
|
||||
""",
|
||||
(limit, offset),
|
||||
).fetchall()
|
||||
|
||||
return [
|
||||
QuestionListItem(
|
||||
id=row["id"],
|
||||
text=row["text"],
|
||||
used=bool(row["used"]),
|
||||
answers_count=row["answers_count"],
|
||||
)
|
||||
for row in rows
|
||||
]
|
||||
|
||||
def pop_unused_question(self) -> QuestionWithAnswers | None:
|
||||
with self._connect() as connection:
|
||||
row = connection.execute(
|
||||
"SELECT id FROM questions WHERE used = 0 ORDER BY RANDOM() LIMIT 1"
|
||||
).fetchone()
|
||||
|
||||
if row is None:
|
||||
has_questions = connection.execute(
|
||||
"SELECT 1 FROM questions LIMIT 1"
|
||||
).fetchone()
|
||||
if has_questions is None:
|
||||
return None
|
||||
|
||||
connection.execute("UPDATE questions SET used = 0")
|
||||
row = connection.execute(
|
||||
"SELECT id FROM questions ORDER BY RANDOM() LIMIT 1"
|
||||
).fetchone()
|
||||
|
||||
question_id = int(row["id"])
|
||||
connection.execute("UPDATE questions SET used = 1 WHERE id = ?", (question_id,))
|
||||
|
||||
return self.get_question(question_id)
|
||||
|
||||
def record_answer(self, question_id: int, answer_id: int, increment: int = 1) -> None:
|
||||
if increment < 1:
|
||||
raise ValueError("Increment must be >= 1.")
|
||||
|
||||
with self._connect() as connection:
|
||||
answer_row = connection.execute(
|
||||
"SELECT 1 FROM answers WHERE id = ? AND question_id = ?",
|
||||
(answer_id, question_id),
|
||||
).fetchone()
|
||||
if answer_row is None:
|
||||
raise ValueError("Answer does not belong to the provided question.")
|
||||
|
||||
connection.execute(
|
||||
"""
|
||||
INSERT INTO statistics(question_id, answer_id, count)
|
||||
VALUES (?, ?, ?)
|
||||
ON CONFLICT(question_id, answer_id)
|
||||
DO UPDATE SET count = count + excluded.count
|
||||
""",
|
||||
(question_id, answer_id, increment),
|
||||
)
|
||||
|
||||
def get_question_statistics(self, question_id: int) -> list[AnswerStatistic]:
|
||||
with self._connect() as connection:
|
||||
rows = connection.execute(
|
||||
"""
|
||||
SELECT
|
||||
a.id AS answer_id,
|
||||
a.text AS answer_text,
|
||||
COALESCE(s.count, 0) AS answer_count
|
||||
FROM answers a
|
||||
LEFT JOIN statistics s
|
||||
ON s.answer_id = a.id AND s.question_id = a.question_id
|
||||
WHERE a.question_id = ?
|
||||
ORDER BY a.id
|
||||
""",
|
||||
(question_id,),
|
||||
).fetchall()
|
||||
|
||||
return [
|
||||
AnswerStatistic(
|
||||
answer_id=row["answer_id"],
|
||||
answer_text=row["answer_text"],
|
||||
count=row["answer_count"],
|
||||
)
|
||||
for row in rows
|
||||
]
|
||||
|
||||
def update_question(self, question_id: int, text: str, answers: Iterable[str]) -> None:
|
||||
prepared_text = text.strip()
|
||||
prepared_answers = [answer.strip() for answer in answers if answer.strip()]
|
||||
if not prepared_text:
|
||||
raise ValueError("Question text cannot be empty.")
|
||||
if len(prepared_answers) < 2:
|
||||
raise ValueError("Question must have at least 2 answers.")
|
||||
|
||||
with self._connect() as connection:
|
||||
exists_row = connection.execute(
|
||||
"SELECT 1 FROM questions WHERE id = ?",
|
||||
(question_id,),
|
||||
).fetchone()
|
||||
if exists_row is None:
|
||||
raise ValueError("Question does not exist.")
|
||||
|
||||
connection.execute(
|
||||
"UPDATE questions SET text = ?, used = 0 WHERE id = ?",
|
||||
(prepared_text, question_id),
|
||||
)
|
||||
connection.execute("DELETE FROM answers WHERE question_id = ?", (question_id,))
|
||||
connection.executemany(
|
||||
"INSERT INTO answers(question_id, text) VALUES (?, ?)",
|
||||
[(question_id, answer_text) for answer_text in prepared_answers],
|
||||
)
|
||||
|
||||
def add_schedule(self, name: str, cron: str, is_active: bool = True) -> int:
|
||||
prepared_name = name.strip()
|
||||
prepared_cron = cron.strip()
|
||||
if not prepared_name:
|
||||
raise ValueError("Schedule name cannot be empty.")
|
||||
if not prepared_cron:
|
||||
raise ValueError("Schedule cron cannot be empty.")
|
||||
|
||||
with self._connect() as connection:
|
||||
cursor = connection.execute(
|
||||
"INSERT INTO schedules(name, cron, is_active) VALUES (?, ?, ?)",
|
||||
(prepared_name, prepared_cron, int(is_active)),
|
||||
)
|
||||
return int(cursor.lastrowid)
|
||||
|
||||
def update_schedule(self, schedule_id: int, name: str, cron: str, is_active: bool) -> None:
|
||||
prepared_name = name.strip()
|
||||
prepared_cron = cron.strip()
|
||||
if not prepared_name:
|
||||
raise ValueError("Schedule name cannot be empty.")
|
||||
if not prepared_cron:
|
||||
raise ValueError("Schedule cron cannot be empty.")
|
||||
|
||||
with self._connect() as connection:
|
||||
updated = connection.execute(
|
||||
"""
|
||||
UPDATE schedules
|
||||
SET name = ?, cron = ?, is_active = ?
|
||||
WHERE id = ?
|
||||
""",
|
||||
(prepared_name, prepared_cron, int(is_active), schedule_id),
|
||||
)
|
||||
if updated.rowcount == 0:
|
||||
raise ValueError("Schedule does not exist.")
|
||||
|
||||
def get_schedule(self, schedule_id: int) -> ScheduleRow | None:
|
||||
with self._connect() as connection:
|
||||
row = connection.execute(
|
||||
"SELECT id, name, cron, is_active FROM schedules WHERE id = ?",
|
||||
(schedule_id,),
|
||||
).fetchone()
|
||||
if row is None:
|
||||
return None
|
||||
|
||||
return ScheduleRow(
|
||||
id=row["id"],
|
||||
name=row["name"],
|
||||
cron=row["cron"],
|
||||
is_active=bool(row["is_active"]),
|
||||
)
|
||||
|
||||
def list_schedules(self, active_only: bool = False) -> list[ScheduleRow]:
|
||||
with self._connect() as connection:
|
||||
if active_only:
|
||||
rows = connection.execute(
|
||||
"SELECT id, name, cron, is_active FROM schedules WHERE is_active = 1 ORDER BY id"
|
||||
).fetchall()
|
||||
else:
|
||||
rows = connection.execute(
|
||||
"SELECT id, name, cron, is_active FROM schedules ORDER BY id"
|
||||
).fetchall()
|
||||
|
||||
return [
|
||||
ScheduleRow(
|
||||
id=row["id"],
|
||||
name=row["name"],
|
||||
cron=row["cron"],
|
||||
is_active=bool(row["is_active"]),
|
||||
)
|
||||
for row in rows
|
||||
]
|
||||
|
||||
def record_schedule_run(self, schedule_id: int, run_key: str) -> bool:
|
||||
with self._connect() as connection:
|
||||
cursor = connection.execute(
|
||||
"""
|
||||
INSERT OR IGNORE INTO schedule_runs(schedule_id, run_key)
|
||||
VALUES (?, ?)
|
||||
""",
|
||||
(schedule_id, run_key),
|
||||
)
|
||||
return cursor.rowcount > 0
|
||||
|
||||
def delete_schedule(self, schedule_id: int) -> None:
|
||||
with self._connect() as connection:
|
||||
deleted = connection.execute(
|
||||
"DELETE FROM schedules WHERE id = ?",
|
||||
(schedule_id,),
|
||||
)
|
||||
if deleted.rowcount == 0:
|
||||
raise ValueError("Schedule does not exist.")
|
||||
|
||||
def reset_used_flags(self) -> None:
|
||||
with self._connect() as connection:
|
||||
connection.execute("UPDATE questions SET used = 0")
|
||||
49
game/game.py
Normal file
49
game/game.py
Normal file
@@ -0,0 +1,49 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
from .database import AnswerStatistic, SQLiteWouldYouRatherRepository
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Question:
|
||||
id: int
|
||||
title: str
|
||||
variants: list[str]
|
||||
variant_ids: list[int]
|
||||
|
||||
def answer_id(self, variant_index: int) -> int:
|
||||
if variant_index < 0 or variant_index >= len(self.variant_ids):
|
||||
raise IndexError("Answer variant index is out of range.")
|
||||
return self.variant_ids[variant_index]
|
||||
|
||||
|
||||
class WouldYouRather:
|
||||
def __init__(self, data: SQLiteWouldYouRatherRepository):
|
||||
self.data = data
|
||||
|
||||
def add_question(self, title: str, variants: list[str]) -> int:
|
||||
return self.data.add_question(text=title, answers=variants)
|
||||
|
||||
def pop_question(self) -> Question | None:
|
||||
question_with_answers = self.data.pop_unused_question()
|
||||
if question_with_answers is None:
|
||||
return None
|
||||
|
||||
return Question(
|
||||
id=question_with_answers.question.id,
|
||||
title=question_with_answers.question.text,
|
||||
variants=[answer.text for answer in question_with_answers.answers],
|
||||
variant_ids=[answer.id for answer in question_with_answers.answers],
|
||||
)
|
||||
|
||||
def answer_question(self, question: Question, variant_index: int, increment: int = 1) -> None:
|
||||
answer_id = question.answer_id(variant_index)
|
||||
self.data.record_answer(
|
||||
question_id=question.id,
|
||||
answer_id=answer_id,
|
||||
increment=increment,
|
||||
)
|
||||
|
||||
def get_question_statistics(self, question_id: int) -> list[AnswerStatistic]:
|
||||
return self.data.get_question_statistics(question_id)
|
||||
149
game/settings.py
Normal file
149
game/settings.py
Normal file
@@ -0,0 +1,149 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import tomllib
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .api_service import QuestionApiService
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ApiSourceSettings:
|
||||
name: str
|
||||
url: str
|
||||
handler: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class AppSettings:
|
||||
admin_ids: list[int]
|
||||
poll_channel_ids: list[int]
|
||||
question_apis: list[ApiSourceSettings]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ConfiguredApiService:
|
||||
name: str
|
||||
handler: str
|
||||
service: QuestionApiService
|
||||
|
||||
|
||||
DEFAULT_SETTINGS_TOML = """# Telegram admin user IDs
|
||||
admin_ids = []
|
||||
|
||||
# Telegram channel IDs where polls should be sent
|
||||
poll_channel_ids = []
|
||||
|
||||
[[question_apis]]
|
||||
name = "default-source"
|
||||
url = "https://example.com/api/questions/random"
|
||||
handler = "default"
|
||||
"""
|
||||
|
||||
|
||||
class AppSettingsService:
|
||||
def __init__(self, path: str | Path = "data\\settings.toml") -> None:
|
||||
self.path = Path(path)
|
||||
|
||||
def ensure_exists(self) -> None:
|
||||
if self.path.exists():
|
||||
return
|
||||
|
||||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||||
self.path.write_text(DEFAULT_SETTINGS_TOML, encoding="utf-8")
|
||||
|
||||
def load(self) -> AppSettings:
|
||||
if not self.path.exists():
|
||||
raise FileNotFoundError(
|
||||
f"Settings file does not exist: {self.path}. "
|
||||
"Call ensure_exists() to create a template file."
|
||||
)
|
||||
|
||||
raw = tomllib.loads(self.path.read_text(encoding="utf-8"))
|
||||
admin_ids = self._parse_int_list(raw, "admin_ids")
|
||||
poll_channel_ids = self._parse_int_list(raw, "poll_channel_ids")
|
||||
question_apis = self._parse_api_sources(raw)
|
||||
|
||||
return AppSettings(
|
||||
admin_ids=admin_ids,
|
||||
poll_channel_ids=poll_channel_ids,
|
||||
question_apis=question_apis,
|
||||
)
|
||||
|
||||
def load_or_create(self) -> AppSettings:
|
||||
self.ensure_exists()
|
||||
return self.load()
|
||||
|
||||
def build_api_services(
|
||||
self,
|
||||
settings: AppSettings,
|
||||
timeout: float = 10.0,
|
||||
headers: dict[str, str] | None = None,
|
||||
) -> list[ConfiguredApiService]:
|
||||
from .api_service import QuestionApiService, build_handler_registry
|
||||
|
||||
registry = build_handler_registry()
|
||||
services: list[ConfiguredApiService] = []
|
||||
for source in settings.question_apis:
|
||||
handler = registry.get(source.handler)
|
||||
if handler is None:
|
||||
raise ValueError(
|
||||
f"Unknown question API handler '{source.handler}' for source '{source.name}'."
|
||||
)
|
||||
services.append(
|
||||
ConfiguredApiService(
|
||||
name=source.name,
|
||||
handler=source.handler,
|
||||
service=QuestionApiService(
|
||||
url=source.url,
|
||||
handler=handler,
|
||||
timeout=timeout,
|
||||
headers=headers,
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
return services
|
||||
|
||||
def _parse_int_list(self, raw: dict, key: str) -> list[int]:
|
||||
value = raw.get(key, [])
|
||||
if not isinstance(value, list):
|
||||
raise ValueError(f"'{key}' must be a list of integers.")
|
||||
|
||||
parsed: list[int] = []
|
||||
for item in value:
|
||||
if isinstance(item, bool) or not isinstance(item, int):
|
||||
raise ValueError(f"'{key}' must contain only integers.")
|
||||
parsed.append(item)
|
||||
return parsed
|
||||
|
||||
def _parse_api_sources(self, raw: dict) -> list[ApiSourceSettings]:
|
||||
value = raw.get("question_apis", [])
|
||||
if not isinstance(value, list):
|
||||
raise ValueError("'question_apis' must be an array of tables.")
|
||||
|
||||
parsed: list[ApiSourceSettings] = []
|
||||
for index, item in enumerate(value):
|
||||
if not isinstance(item, dict):
|
||||
raise ValueError(f"'question_apis[{index}]' must be a table.")
|
||||
|
||||
name = item.get("name")
|
||||
url = item.get("url")
|
||||
handler = item.get("handler", "default")
|
||||
if not isinstance(name, str) or not name.strip():
|
||||
raise ValueError(f"'question_apis[{index}].name' must be a non-empty string.")
|
||||
if not isinstance(url, str) or not url.strip():
|
||||
raise ValueError(f"'question_apis[{index}].url' must be a non-empty string.")
|
||||
if not isinstance(handler, str) or not handler.strip():
|
||||
raise ValueError(
|
||||
f"'question_apis[{index}].handler' must be a non-empty string."
|
||||
)
|
||||
|
||||
parsed.append(
|
||||
ApiSourceSettings(name=name.strip(), url=url.strip(), handler=handler.strip())
|
||||
)
|
||||
|
||||
return parsed
|
||||
Reference in New Issue
Block a user