Implement automatic poll answer statistics tracking

Features:
- Add poll_mappings table to link Telegram poll_id with question_id
- Add PollStatisticsHandler to automatically record user votes
- Implement save_poll_mapping, get_question_id_by_poll methods
- Implement get_answer_id_by_option_index for answer lookup
- Update scheduler to save poll mappings on publish
- Add comprehensive test suite (test_statistics.py)
- Add STATISTICS_GUIDE.md with detailed documentation
- Update README.md with statistics features

Statistics features:
✅ Automatic vote counting for each answer option
✅ Percentage distribution display
✅ Anonymous tracking (no user data stored)
✅ View statistics via /manage → 📊 Statistics
✅ Support for multiple poll publications per question

All tests passing successfully!

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
2026-04-05 12:27:45 +02:00
parent 25380e7c1f
commit ab8e129277
7 changed files with 601 additions and 4 deletions

View File

@@ -104,6 +104,14 @@ class SQLiteWouldYouRatherRepository:
FOREIGN KEY (schedule_id) REFERENCES schedules(id) ON DELETE CASCADE,
UNIQUE (schedule_id, run_key)
);
CREATE TABLE IF NOT EXISTS poll_mappings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
poll_id TEXT NOT NULL UNIQUE,
question_id INTEGER NOT NULL,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (question_id) REFERENCES questions(id) ON DELETE CASCADE
);
"""
)
@@ -383,6 +391,39 @@ class SQLiteWouldYouRatherRepository:
if deleted.rowcount == 0:
raise ValueError("Schedule does not exist.")
def save_poll_mapping(self, poll_id: str, question_id: int) -> None:
"""Сохраняет связь между poll_id Telegram и question_id в базе."""
with self._connect() as connection:
connection.execute(
"INSERT OR REPLACE INTO poll_mappings(poll_id, question_id) VALUES (?, ?)",
(poll_id, question_id),
)
def get_question_id_by_poll(self, poll_id: str) -> int | None:
"""Получает question_id по poll_id Telegram."""
with self._connect() as connection:
row = connection.execute(
"SELECT question_id FROM poll_mappings WHERE poll_id = ?",
(poll_id,),
).fetchone()
return int(row["question_id"]) if row else None
def get_answer_id_by_option_index(
self, question_id: int, option_index: int
) -> int | None:
"""Получает answer_id по индексу опции в опросе."""
with self._connect() as connection:
row = connection.execute(
"""
SELECT id FROM answers
WHERE question_id = ?
ORDER BY id
LIMIT 1 OFFSET ?
""",
(question_id, option_index),
).fetchone()
return int(row["id"]) if row else None
def reset_used_flags(self) -> None:
with self._connect() as connection:
connection.execute("UPDATE questions SET used = 0")