102 lines
3.8 KiB
Python
102 lines
3.8 KiB
Python
"""Telegram ingestion via Telethon (MTProto).
|
|
|
|
This logs in as the real user account, not a bot — the Bot API cannot read
|
|
personal DMs, so there is no bot-shaped way to do this (see docs/project-plan.md
|
|
Phase 12 step 4). It runs strictly non-interactively: the session file at
|
|
TELEGRAM_SESSION_PATH must already exist, created once by hand with
|
|
`python ingest/telegram_login.py`. If it doesn't, or it has been invalidated,
|
|
this module warns and returns nothing rather than blocking a scheduled run on a
|
|
phone-code prompt nobody is there to answer.
|
|
|
|
Reading messages does not acknowledge them — Telethon only marks a chat read on
|
|
an explicit `send_read_acknowledge()`, which is never called here.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
MAX_BODY_CHARS = 2000
|
|
DEFAULT_SESSION_PATH = "/data/telegram.session"
|
|
|
|
|
|
def _session_path():
|
|
return os.environ.get("TELEGRAM_SESSION_PATH", DEFAULT_SESSION_PATH)
|
|
|
|
|
|
async def _fetch_async(lookback_hours, api_id, api_hash, max_dialogs, max_per_dialog):
|
|
from telethon import TelegramClient
|
|
|
|
since = datetime.now(timezone.utc) - timedelta(hours=lookback_hours)
|
|
messages = []
|
|
|
|
client = TelegramClient(_session_path(), api_id, api_hash)
|
|
await client.connect()
|
|
try:
|
|
if not await client.is_user_authorized():
|
|
LOG.warning(
|
|
"telegram: session at %s is not authorized — run `python ingest/telegram_login.py` once",
|
|
_session_path(),
|
|
)
|
|
return []
|
|
|
|
async for dialog in client.iter_dialogs(limit=max_dialogs):
|
|
try:
|
|
async for message in client.iter_messages(dialog.entity, limit=max_per_dialog):
|
|
if message.date is None or message.date < since:
|
|
break
|
|
if not message.message:
|
|
continue
|
|
sender = await message.get_sender()
|
|
sender_name = getattr(sender, "username", None) or getattr(
|
|
sender, "first_name", None
|
|
)
|
|
messages.append(
|
|
{
|
|
"source": "telegram",
|
|
"chat": dialog.name,
|
|
"from": sender_name or "unknown",
|
|
"outgoing": bool(message.out),
|
|
"timestamp": message.date.isoformat(),
|
|
"body": message.message[:MAX_BODY_CHARS],
|
|
}
|
|
)
|
|
except Exception:
|
|
LOG.warning("telegram: could not read dialog %r, skipping", dialog.name, exc_info=True)
|
|
finally:
|
|
await client.disconnect()
|
|
|
|
return messages
|
|
|
|
|
|
def fetch(lookback_hours):
|
|
api_id = os.environ.get("TELEGRAM_API_ID", "").strip()
|
|
api_hash = os.environ.get("TELEGRAM_API_HASH", "").strip()
|
|
max_dialogs = int(os.environ.get("TELEGRAM_MAX_DIALOGS", "25"))
|
|
max_per_dialog = int(os.environ.get("TELEGRAM_MAX_MESSAGES_PER_DIALOG", "50"))
|
|
|
|
if not (api_id and api_hash):
|
|
LOG.warning("telegram: TELEGRAM_API_ID/TELEGRAM_API_HASH not set, skipping")
|
|
return []
|
|
|
|
if not os.path.exists(_session_path()):
|
|
LOG.warning(
|
|
"telegram: no session file at %s — run `python ingest/telegram_login.py` once, skipping",
|
|
_session_path(),
|
|
)
|
|
return []
|
|
|
|
try:
|
|
messages = asyncio.run(
|
|
_fetch_async(lookback_hours, int(api_id), api_hash, max_dialogs, max_per_dialog)
|
|
)
|
|
except Exception:
|
|
LOG.warning("telegram: ingestion failed, returning nothing", exc_info=True)
|
|
return []
|
|
|
|
LOG.info("telegram: %d message(s) in the last %sh", len(messages), lookback_hours)
|
|
return messages
|