459 lines
16 KiB
Python
Executable File
459 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Minimal Telegram secretary: archive allowed topics and export them to Markdown."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import sqlite3
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
import uuid
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
||
|
||
|
||
APP_NAME = "Акылдаш"
|
||
APP_VERSION = "0.2.2"
|
||
FOOTER = f"{APP_NAME} v{APP_VERSION}"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Config:
|
||
token: str
|
||
chat_id: int
|
||
owner_id: int
|
||
report_thread_id: int
|
||
thread_ids: frozenset[int]
|
||
database: Path
|
||
timezone: ZoneInfo
|
||
|
||
|
||
def load_config() -> Config:
|
||
token = os.environ.get("TELEGRAM_BOT_TOKEN", "").strip()
|
||
chat_id = os.environ.get("TELEGRAM_CHAT_ID", "").strip()
|
||
owner_id = os.environ.get("TELEGRAM_OWNER_ID", "").strip()
|
||
report_thread_id = os.environ.get("TELEGRAM_REPORT_THREAD_ID", "").strip()
|
||
if not token or not chat_id or not owner_id or not report_thread_id:
|
||
raise SystemExit(
|
||
"Set TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, TELEGRAM_OWNER_ID "
|
||
"and TELEGRAM_REPORT_THREAD_ID"
|
||
)
|
||
|
||
try:
|
||
threads = frozenset(
|
||
int(value.strip())
|
||
for value in os.environ.get(
|
||
"TELEGRAM_ALLOWED_THREAD_IDS", "0,2,4,6,8"
|
||
).split(",")
|
||
if value.strip()
|
||
)
|
||
return Config(
|
||
token=token,
|
||
chat_id=int(chat_id),
|
||
owner_id=int(owner_id),
|
||
report_thread_id=int(report_thread_id),
|
||
thread_ids=threads,
|
||
database=Path(os.environ.get("BOT_DATABASE", "data/secretary.sqlite3")),
|
||
timezone=ZoneInfo(os.environ.get("APP_TIMEZONE", "Asia/Bishkek")),
|
||
)
|
||
except (ValueError, ZoneInfoNotFoundError) as error:
|
||
raise SystemExit(f"Invalid configuration: {error}") from error
|
||
|
||
|
||
def connect(database: Path) -> sqlite3.Connection:
|
||
database.parent.mkdir(parents=True, exist_ok=True)
|
||
connection = sqlite3.connect(database)
|
||
connection.row_factory = sqlite3.Row
|
||
connection.execute("PRAGMA journal_mode=WAL")
|
||
connection.executescript(
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS messages (
|
||
chat_id INTEGER NOT NULL,
|
||
message_id INTEGER NOT NULL,
|
||
thread_id INTEGER NOT NULL,
|
||
sent_at INTEGER NOT NULL,
|
||
updated_at INTEGER NOT NULL,
|
||
update_id INTEGER NOT NULL,
|
||
payload TEXT NOT NULL,
|
||
PRIMARY KEY (chat_id, message_id)
|
||
);
|
||
CREATE INDEX IF NOT EXISTS messages_thread_date
|
||
ON messages (chat_id, thread_id, sent_at);
|
||
CREATE TABLE IF NOT EXISTS state (
|
||
key TEXT PRIMARY KEY,
|
||
value TEXT NOT NULL
|
||
);
|
||
"""
|
||
)
|
||
return connection
|
||
|
||
|
||
def archive_update(
|
||
connection: sqlite3.Connection, update: dict, config: Config
|
||
) -> dict | None:
|
||
update_id = int(update["update_id"])
|
||
message = update.get("message") or update.get("edited_message")
|
||
|
||
with connection:
|
||
if message:
|
||
chat_id = int(message["chat"]["id"])
|
||
thread_id = int(message.get("message_thread_id", 0))
|
||
if chat_id == config.chat_id and thread_id in config.thread_ids:
|
||
connection.execute(
|
||
"""
|
||
INSERT INTO messages (
|
||
chat_id, message_id, thread_id, sent_at, updated_at,
|
||
update_id, payload
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?)
|
||
ON CONFLICT (chat_id, message_id) DO UPDATE SET
|
||
thread_id = excluded.thread_id,
|
||
updated_at = excluded.updated_at,
|
||
update_id = excluded.update_id,
|
||
payload = excluded.payload
|
||
""",
|
||
(
|
||
chat_id,
|
||
int(message["message_id"]),
|
||
thread_id,
|
||
int(message["date"]),
|
||
int(message.get("edit_date", message["date"])),
|
||
update_id,
|
||
json.dumps(message, ensure_ascii=False),
|
||
),
|
||
)
|
||
else:
|
||
message = None
|
||
|
||
connection.execute(
|
||
"""
|
||
INSERT INTO state (key, value) VALUES ('next_update_id', ?)
|
||
ON CONFLICT (key) DO UPDATE SET value = excluded.value
|
||
""",
|
||
(str(update_id + 1),),
|
||
)
|
||
return message
|
||
|
||
|
||
def next_update_id(connection: sqlite3.Connection) -> int:
|
||
row = connection.execute(
|
||
"SELECT value FROM state WHERE key = 'next_update_id'"
|
||
).fetchone()
|
||
return int(row["value"]) if row else 0
|
||
|
||
|
||
def export_checkpoint(
|
||
connection: sqlite3.Connection, chat_id: int, thread_id: int
|
||
) -> int:
|
||
row = connection.execute(
|
||
"SELECT value FROM state WHERE key = ?",
|
||
(f"export_checkpoint:{chat_id}:{thread_id}",),
|
||
).fetchone()
|
||
return int(row["value"]) if row else 0
|
||
|
||
|
||
def save_export_checkpoint(
|
||
connection: sqlite3.Connection, chat_id: int, thread_id: int, message_id: int
|
||
) -> None:
|
||
with connection:
|
||
connection.execute(
|
||
"""
|
||
INSERT INTO state (key, value) VALUES (?, ?)
|
||
ON CONFLICT (key) DO UPDATE SET value = excluded.value
|
||
""",
|
||
(f"export_checkpoint:{chat_id}:{thread_id}", str(message_id)),
|
||
)
|
||
|
||
|
||
def telegram_request(
|
||
config: Config, method: str, payload: dict, timeout: int = 65
|
||
) -> object:
|
||
request = urllib.request.Request(
|
||
f"https://api.telegram.org/bot{config.token}/{method}",
|
||
data=json.dumps(payload).encode(),
|
||
headers={"Content-Type": "application/json"},
|
||
)
|
||
with urllib.request.urlopen(request, timeout=timeout) as response:
|
||
result = json.load(response)
|
||
if not result.get("ok"):
|
||
raise RuntimeError(result.get("description", "Telegram API error"))
|
||
return result["result"]
|
||
|
||
|
||
def send_text(config: Config, message: dict, text: str) -> None:
|
||
payload = {
|
||
"chat_id": config.chat_id,
|
||
"text": f"{text}\n\n{FOOTER}",
|
||
"reply_parameters": {"message_id": message["message_id"]},
|
||
}
|
||
if message.get("message_thread_id"):
|
||
payload["message_thread_id"] = message["message_thread_id"]
|
||
telegram_request(config, "sendMessage", payload)
|
||
|
||
|
||
def send_document(
|
||
config: Config, thread_id: int, filename: str, content: bytes, caption: str
|
||
) -> dict:
|
||
boundary = uuid.uuid4().hex
|
||
fields = {
|
||
"chat_id": str(config.chat_id),
|
||
"message_thread_id": str(thread_id),
|
||
"caption": caption,
|
||
}
|
||
|
||
body = bytearray()
|
||
for name, value in fields.items():
|
||
body.extend(
|
||
f"--{boundary}\r\nContent-Disposition: form-data; name=\"{name}\""
|
||
f"\r\n\r\n{value}\r\n".encode()
|
||
)
|
||
body.extend(
|
||
f"--{boundary}\r\nContent-Disposition: form-data; name=\"document\"; "
|
||
f"filename=\"{filename}\"\r\nContent-Type: text/markdown; charset=utf-8"
|
||
f"\r\n\r\n".encode()
|
||
)
|
||
body.extend(content)
|
||
body.extend(f"\r\n--{boundary}--\r\n".encode())
|
||
|
||
request = urllib.request.Request(
|
||
f"https://api.telegram.org/bot{config.token}/sendDocument",
|
||
data=bytes(body),
|
||
headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
|
||
)
|
||
with urllib.request.urlopen(request, timeout=65) as response:
|
||
result = json.load(response)
|
||
if not result.get("ok"):
|
||
raise RuntimeError(result.get("description", "Telegram API error"))
|
||
return result["result"]
|
||
|
||
|
||
def author(message: dict) -> str:
|
||
sender = message.get("from", {})
|
||
name = " ".join(
|
||
part for part in (sender.get("first_name"), sender.get("last_name")) if part
|
||
)
|
||
username = sender.get("username")
|
||
if name and username:
|
||
return f"{name} (@{username})"
|
||
return name or (f"@{username}" if username else "Неизвестный участник")
|
||
|
||
|
||
def attachment_descriptions(message: dict) -> list[str]:
|
||
descriptions: list[str] = []
|
||
if message.get("photo"):
|
||
descriptions.append("фотография")
|
||
for field, label in (
|
||
("document", "документ"),
|
||
("video", "видео"),
|
||
("audio", "аудио"),
|
||
("voice", "голосовое сообщение"),
|
||
("animation", "анимация"),
|
||
("sticker", "стикер"),
|
||
):
|
||
attachment = message.get(field)
|
||
if attachment:
|
||
name = attachment.get("file_name") or attachment.get("emoji")
|
||
descriptions.append(f"{label}: {name}" if name else label)
|
||
return descriptions
|
||
|
||
|
||
def message_link(chat_id: int, message_id: int) -> str:
|
||
return f"https://t.me/c/{str(chat_id).removeprefix('-100')}/{message_id}"
|
||
|
||
|
||
def export_markdown(
|
||
connection: sqlite3.Connection,
|
||
config: Config,
|
||
thread_id: int,
|
||
days: int | None,
|
||
now: int | None = None,
|
||
after_message_id: int | None = None,
|
||
) -> str:
|
||
parameters: list[int] = [config.chat_id, thread_id]
|
||
condition = ""
|
||
if after_message_id is not None:
|
||
condition = " AND message_id > ?"
|
||
parameters.append(after_message_id)
|
||
elif days is not None:
|
||
condition = " AND sent_at >= ?"
|
||
parameters.append((now or int(time.time())) - days * 86400)
|
||
rows = connection.execute(
|
||
f"""
|
||
SELECT message_id, sent_at, payload FROM messages
|
||
WHERE chat_id = ? AND thread_id = ?{condition}
|
||
ORDER BY sent_at, message_id
|
||
""",
|
||
parameters,
|
||
).fetchall()
|
||
|
||
if after_message_id is not None:
|
||
period = (
|
||
"с начала архива"
|
||
if after_message_id == 0
|
||
else "после предыдущей отсечки"
|
||
)
|
||
else:
|
||
period = "за всё время" if days is None else f"за последние {days} дн."
|
||
lines = [
|
||
"# Обсуждение",
|
||
"",
|
||
f"Тема: `{thread_id or 'Общее'}` ",
|
||
f"Период: {period} ",
|
||
f"Экспортировано: {datetime.now(config.timezone):%Y-%m-%d %H:%M %Z}",
|
||
"",
|
||
"## Сообщения",
|
||
]
|
||
for row in rows:
|
||
message = json.loads(row["payload"])
|
||
text = message.get("text") or message.get("caption") or "[Служебное событие Telegram]"
|
||
if text.split(maxsplit=1)[0].split("@", 1)[0] in {"/help", "/status", "/export"}:
|
||
continue
|
||
sent_at = datetime.fromtimestamp(row["sent_at"], timezone.utc).astimezone(
|
||
config.timezone
|
||
)
|
||
lines.extend(
|
||
[
|
||
"",
|
||
f"### {author(message)} — {sent_at:%Y-%m-%d %H:%M}",
|
||
"",
|
||
f"[Открыть сообщение]({message_link(config.chat_id, row['message_id'])})",
|
||
]
|
||
)
|
||
reply = message.get("reply_to_message", {}).get("message_id")
|
||
if reply and reply != message.get("message_thread_id"):
|
||
lines.append(f"Ответ на сообщение: #{reply}")
|
||
lines.extend(["", text])
|
||
attachments = attachment_descriptions(message)
|
||
if attachments:
|
||
lines.extend(["", "Вложения:", *[f"- {item}" for item in attachments]])
|
||
|
||
lines.extend(["", "---", FOOTER, ""])
|
||
return "\n".join(lines)
|
||
|
||
|
||
def parse_export_days(parts: list[str]) -> int | None:
|
||
if len(parts) == 1:
|
||
return 7
|
||
if parts[1].lower() in {"all", "все"}:
|
||
return None
|
||
days = int(parts[1])
|
||
if not 1 <= days <= 3650:
|
||
raise ValueError
|
||
return days
|
||
|
||
|
||
def handle_command(
|
||
connection: sqlite3.Connection, config: Config, message: dict
|
||
) -> None:
|
||
text = message.get("text", "")
|
||
if not text.startswith("/"):
|
||
return
|
||
parts = text.split()
|
||
command = parts[0].split("@", 1)[0].lower()
|
||
thread_id = int(message.get("message_thread_id", 0))
|
||
|
||
if command == "/help":
|
||
send_text(
|
||
config,
|
||
message,
|
||
"Я сохраняю обсуждения этой группы.\n"
|
||
"/status — количество сохранённых сообщений\n"
|
||
"/export — экспорт текущей темы после предыдущей отсечки",
|
||
)
|
||
elif command == "/status":
|
||
topic_count = connection.execute(
|
||
"SELECT COUNT(*) FROM messages WHERE chat_id = ? AND thread_id = ?",
|
||
(config.chat_id, thread_id),
|
||
).fetchone()[0]
|
||
total_count = connection.execute(
|
||
"SELECT COUNT(*) FROM messages WHERE chat_id = ?", (config.chat_id,)
|
||
).fetchone()[0]
|
||
send_text(
|
||
config,
|
||
message,
|
||
f"Сохранено сообщений: {topic_count} в этой теме, {total_count} всего.",
|
||
)
|
||
elif command == "/export":
|
||
if int(message.get("from", {}).get("id", 0)) != config.owner_id:
|
||
send_text(config, message, "Экспорт доступен только владельцу бота.")
|
||
return
|
||
try:
|
||
days = parse_export_days(parts) if len(parts) > 1 else None
|
||
except (ValueError, IndexError):
|
||
send_text(config, message, "Использование: /export [1–3650|все]")
|
||
return
|
||
checkpoint = (
|
||
None
|
||
if len(parts) > 1
|
||
else export_checkpoint(connection, config.chat_id, thread_id)
|
||
)
|
||
markdown = export_markdown(
|
||
connection,
|
||
config,
|
||
thread_id,
|
||
days,
|
||
after_message_id=checkpoint,
|
||
)
|
||
stamp = datetime.now(config.timezone).strftime("%Y%m%d-%H%M")
|
||
report_tag = f"#report_{message['message_id']}"
|
||
report = send_document(
|
||
config,
|
||
config.report_thread_id,
|
||
f"discussion-{thread_id}-{stamp}.md",
|
||
markdown.encode(),
|
||
f"{report_tag}\nИсходное обсуждение: "
|
||
f"{message_link(config.chat_id, message['message_id'])}\n\n{FOOTER}",
|
||
)
|
||
send_text(
|
||
config,
|
||
message,
|
||
"━━━━━━━━━━━━━━━━\n"
|
||
"✅ ОБСУЖДЕНИЕ ЗАВЕРШЕНО\n"
|
||
f"{report_tag}\n"
|
||
f"Отчёт: {message_link(config.chat_id, report['message_id'])}\n"
|
||
"━━━━━━━━━━━━━━━━",
|
||
)
|
||
save_export_checkpoint(
|
||
connection, config.chat_id, thread_id, int(message["message_id"])
|
||
)
|
||
|
||
|
||
def run() -> None:
|
||
config = load_config()
|
||
connection = connect(config.database)
|
||
logging.info("Starting %s with database %s", FOOTER, config.database)
|
||
|
||
while True:
|
||
try:
|
||
updates = telegram_request(
|
||
config,
|
||
"getUpdates",
|
||
{
|
||
"offset": next_update_id(connection),
|
||
"timeout": 50,
|
||
"allowed_updates": ["message", "edited_message"],
|
||
},
|
||
)
|
||
for update in updates:
|
||
message = archive_update(connection, update, config)
|
||
if message:
|
||
handle_command(connection, config, message)
|
||
except (urllib.error.URLError, TimeoutError, RuntimeError, OSError, ValueError):
|
||
logging.exception("Polling failed; retrying in 5 seconds")
|
||
time.sleep(5)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
logging.basicConfig(
|
||
level=os.environ.get("LOG_LEVEL", "INFO"),
|
||
format="%(asctime)s %(levelname)s %(message)s",
|
||
)
|
||
try:
|
||
run()
|
||
except KeyboardInterrupt:
|
||
logging.info("Stopped")
|