gilfanovrandClaude Opus 5.5
8a06244ef4
Пользователь tgexport, сессия /var/lib/tg-export (600), тексты в /mnt/backup/tg (группа astra), докачка раз в сутки без медиа, итог в backup-metrics. Вход делает владелец. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
902 lines
38 KiB
Python
902 lines
38 KiB
Python
#!/usr/bin/env python3
|
||
"""Полная выгрузка истории личного Telegram-аккаунта через Telethon.
|
||
|
||
Возможности:
|
||
* логин по номеру телефона (MTProto, свой api_id/api_hash);
|
||
* индекс всех диалогов (личные, боты, группы, каналы);
|
||
* выгрузка всех сообщений в JSONL (сырой `to_dict()` — ничего не теряется)
|
||
плюс опциональный человекочитаемый .txt;
|
||
* скачивание медиа с лимитом на размер;
|
||
* возобновление: повторный запуск догружает только новое;
|
||
* аккуратная обработка FloodWait и приватных/недоступных чатов.
|
||
|
||
Команды:
|
||
python export.py login — только авторизация (создаёт .session)
|
||
python export.py list — вывести список диалогов
|
||
python export.py run — выгрузить историю
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import base64
|
||
import datetime as dt
|
||
import json
|
||
import os
|
||
import re
|
||
import signal
|
||
import sys
|
||
import unicodedata
|
||
from pathlib import Path
|
||
|
||
from telethon import TelegramClient, errors, utils
|
||
from telethon.tl import types
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# конфиг
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
ROOT = Path(__file__).resolve().parent
|
||
|
||
|
||
def load_dotenv(path: Path) -> None:
|
||
"""Минимальный .env-лоадер, чтобы не тащить python-dotenv."""
|
||
if not path.exists():
|
||
return
|
||
for raw in path.read_text(encoding="utf-8").splitlines():
|
||
line = raw.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
key, _, val = line.partition("=")
|
||
key, val = key.strip(), val.strip().strip("'\"")
|
||
os.environ.setdefault(key, val)
|
||
|
||
|
||
def require_credentials() -> tuple[int, str]:
|
||
api_id = os.environ.get("TG_API_ID")
|
||
api_hash = os.environ.get("TG_API_HASH")
|
||
if not api_id or not api_hash:
|
||
sys.exit(
|
||
"Нет TG_API_ID / TG_API_HASH.\n"
|
||
"Получи их на https://my.telegram.org → API development tools,\n"
|
||
f"затем скопируй {ROOT / '.env.example'} в {ROOT / '.env'} и заполни."
|
||
)
|
||
try:
|
||
return int(api_id), api_hash
|
||
except ValueError:
|
||
sys.exit(f"TG_API_ID должен быть числом, получено: {api_id!r}")
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# утилиты
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
class Encoder(json.JSONEncoder):
|
||
"""Сериализует всё, что встречается в Telethon-объектах."""
|
||
|
||
def default(self, o): # noqa: D102
|
||
if isinstance(o, (dt.datetime, dt.date, dt.time)):
|
||
return o.isoformat()
|
||
if isinstance(o, dt.timedelta):
|
||
return o.total_seconds()
|
||
if isinstance(o, (bytes, bytearray, memoryview)):
|
||
return base64.b64encode(bytes(o)).decode("ascii")
|
||
if isinstance(o, set):
|
||
return sorted(o, key=str)
|
||
return str(o)
|
||
|
||
|
||
_SLUG_BAD = re.compile(r"[^\w\-. ]+", re.UNICODE)
|
||
|
||
|
||
def slugify(name: str, limit: int = 48) -> str:
|
||
name = unicodedata.normalize("NFKC", name or "").strip()
|
||
name = _SLUG_BAD.sub("", name)
|
||
name = re.sub(r"\s+", "_", name).strip("._-")
|
||
return (name[:limit] or "unnamed")
|
||
|
||
|
||
def dialog_kind(entity) -> str:
|
||
if isinstance(entity, types.User):
|
||
return "bot" if getattr(entity, "bot", False) else "user"
|
||
if isinstance(entity, types.Chat) or isinstance(entity, types.ChatForbidden):
|
||
return "group"
|
||
if isinstance(entity, types.Channel):
|
||
return "supergroup" if getattr(entity, "megagroup", False) else "channel"
|
||
return "unknown"
|
||
|
||
|
||
def human_size(n: int | None) -> str:
|
||
if not n:
|
||
return "0B"
|
||
for unit in ("B", "K", "M", "G", "T"):
|
||
if n < 1024 or unit == "T":
|
||
return f"{n:.0f}{unit}" if unit == "B" else f"{n:.1f}{unit}"
|
||
n /= 1024
|
||
return f"{n:.1f}T"
|
||
|
||
|
||
def parse_date(value: str | None) -> dt.datetime | None:
|
||
if not value:
|
||
return None
|
||
try:
|
||
parsed = dt.datetime.fromisoformat(value)
|
||
except ValueError:
|
||
sys.exit(f"Не смог разобрать дату {value!r}. Формат: YYYY-MM-DD или YYYY-MM-DDTHH:MM")
|
||
if parsed.tzinfo is None:
|
||
parsed = parsed.replace(tzinfo=dt.timezone.utc)
|
||
return parsed
|
||
|
||
|
||
STOP = False
|
||
|
||
|
||
def install_signal_handlers() -> None:
|
||
def handler(signum, frame): # noqa: ARG001
|
||
global STOP
|
||
if STOP:
|
||
sys.exit(130)
|
||
STOP = True
|
||
print("\n[!] Останавливаюсь после текущего сообщения; прогресс сохранён. Ctrl-C ещё раз — выход.",
|
||
file=sys.stderr)
|
||
|
||
signal.signal(signal.SIGINT, handler)
|
||
signal.signal(signal.SIGTERM, handler)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# состояние (для возобновления)
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
class State:
|
||
def __init__(self, path: Path):
|
||
self.path = path
|
||
self.data = {"last_id": 0, "count": 0, "media": 0, "updated_at": None}
|
||
if path.exists():
|
||
try:
|
||
self.data.update(json.loads(path.read_text(encoding="utf-8")))
|
||
except (ValueError, OSError):
|
||
pass
|
||
|
||
@property
|
||
def last_id(self) -> int:
|
||
return int(self.data.get("last_id") or 0)
|
||
|
||
def bump(self, msg_id: int, media: int = 0) -> None:
|
||
self.data["last_id"] = max(self.last_id, int(msg_id))
|
||
self.data["count"] = int(self.data.get("count") or 0) + 1
|
||
self.data["media"] = int(self.data.get("media") or 0) + media
|
||
|
||
def save(self) -> None:
|
||
self.data["updated_at"] = dt.datetime.now(dt.timezone.utc).isoformat()
|
||
tmp = self.path.with_suffix(".tmp")
|
||
tmp.write_text(json.dumps(self.data, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
tmp.replace(self.path)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# текстовое представление
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
def render_line(msg, sender_name: str) -> str:
|
||
when = msg.date.astimezone().strftime("%Y-%m-%d %H:%M:%S") if msg.date else "?"
|
||
body = (msg.message or "").replace("\n", "\n ")
|
||
tags = []
|
||
if msg.media is not None and not isinstance(msg.media, types.MessageMediaWebPage):
|
||
tags.append(type(msg.media).__name__.replace("MessageMedia", "").lower())
|
||
if getattr(msg, "reply_to_msg_id", None):
|
||
tags.append(f"reply→{msg.reply_to_msg_id}")
|
||
if getattr(msg, "fwd_from", None):
|
||
tags.append("forwarded")
|
||
if getattr(msg, "action", None):
|
||
tags.append(type(msg.action).__name__.replace("MessageAction", "action:"))
|
||
suffix = f" [{', '.join(tags)}]" if tags else ""
|
||
return f"[{when}] #{msg.id} {sender_name}:{suffix}\n {body}\n" if body else \
|
||
f"[{when}] #{msg.id} {sender_name}:{suffix}\n"
|
||
|
||
|
||
async def sender_label(client, msg, cache: dict) -> str:
|
||
sid = msg.sender_id
|
||
if sid is None:
|
||
return "—"
|
||
if sid in cache:
|
||
return cache[sid]
|
||
label = str(sid)
|
||
try:
|
||
sender = await msg.get_sender()
|
||
if sender is not None:
|
||
if isinstance(sender, types.User):
|
||
parts = [sender.first_name or "", sender.last_name or ""]
|
||
label = " ".join(p for p in parts if p) or (sender.username or str(sid))
|
||
else:
|
||
label = getattr(sender, "title", None) or str(sid)
|
||
except (errors.RPCError, ValueError, TypeError):
|
||
pass
|
||
cache[sid] = label
|
||
return label
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# читаемое представление (messages.json)
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
def media_info(msg, record: dict) -> dict | None:
|
||
media = getattr(msg, "media", None)
|
||
if media is None:
|
||
return None
|
||
|
||
if isinstance(media, types.MessageMediaWebPage):
|
||
url = getattr(getattr(media, "webpage", None), "url", None)
|
||
return {"type": "link", "url": url} if url else None
|
||
|
||
info: dict = {"type": type(media).__name__.replace("MessageMedia", "").lower()}
|
||
|
||
handle = getattr(msg, "file", None)
|
||
for key in ("name", "mime_type", "size", "duration"):
|
||
val = getattr(handle, key, None)
|
||
if val:
|
||
info[key] = round(val, 1) if key == "duration" else val
|
||
if info.get("size"):
|
||
info["size_human"] = human_size(info["size"])
|
||
|
||
for attr in getattr(getattr(media, "document", None), "attributes", None) or []:
|
||
if isinstance(attr, types.DocumentAttributeSticker):
|
||
info["type"] = "sticker"
|
||
info["emoji"] = attr.alt
|
||
elif isinstance(attr, types.DocumentAttributeAudio) and attr.voice:
|
||
info["type"] = "voice"
|
||
|
||
if isinstance(media, types.MessageMediaPoll):
|
||
poll = media.poll
|
||
info["question"] = getattr(poll.question, "text", poll.question)
|
||
info["answers"] = [getattr(a.text, "text", a.text) for a in poll.answers]
|
||
elif isinstance(media, (types.MessageMediaGeo, types.MessageMediaGeoLive)):
|
||
geo = media.geo
|
||
info["lat"], info["long"] = getattr(geo, "lat", None), getattr(geo, "long", None)
|
||
elif isinstance(media, types.MessageMediaContact):
|
||
info["contact"] = " ".join(
|
||
p for p in (media.first_name, media.last_name, media.phone_number) if p
|
||
)
|
||
|
||
for key in ("_media_file", "_media_skipped", "_media_error"):
|
||
if key in record:
|
||
info[key.lstrip("_").removeprefix("media_")] = record[key]
|
||
return info
|
||
|
||
|
||
def forward_info(msg) -> dict | None:
|
||
fwd = getattr(msg, "fwd_from", None)
|
||
if fwd is None:
|
||
return None
|
||
info: dict = {}
|
||
if fwd.from_name:
|
||
info["from_name"] = fwd.from_name
|
||
if fwd.from_id is not None:
|
||
try:
|
||
info["from_id"] = utils.get_peer_id(fwd.from_id)
|
||
except (TypeError, ValueError):
|
||
info["from_id"] = str(fwd.from_id)
|
||
if fwd.date:
|
||
info["original_date"] = fwd.date.astimezone().isoformat()
|
||
if fwd.post_author:
|
||
info["post_author"] = fwd.post_author
|
||
return info or None
|
||
|
||
|
||
def reactions_info(msg) -> list | None:
|
||
results = getattr(getattr(msg, "reactions", None), "results", None) or []
|
||
out = []
|
||
for res in results:
|
||
emoji = getattr(res.reaction, "emoticon", None) \
|
||
or getattr(res.reaction, "document_id", None)
|
||
out.append({"reaction": str(emoji), "count": res.count})
|
||
return out or None
|
||
|
||
|
||
def simplify_message(msg, sender: str, record: dict) -> dict:
|
||
"""Плоское, читаемое представление сообщения для messages.json."""
|
||
out: dict = {
|
||
"id": msg.id,
|
||
"date": msg.date.astimezone().isoformat() if msg.date else None,
|
||
"from": sender,
|
||
"from_id": msg.sender_id,
|
||
"outgoing": bool(getattr(msg, "out", False)),
|
||
"text": getattr(msg, "message", "") or "",
|
||
}
|
||
if getattr(msg, "reply_to_msg_id", None):
|
||
out["reply_to"] = msg.reply_to_msg_id
|
||
if getattr(msg, "edit_date", None):
|
||
out["edited"] = msg.edit_date.astimezone().isoformat()
|
||
if fwd := forward_info(msg):
|
||
out["forwarded"] = fwd
|
||
if media := media_info(msg, record):
|
||
out["media"] = media
|
||
if reactions := reactions_info(msg):
|
||
out["reactions"] = reactions
|
||
|
||
action = getattr(msg, "action", None)
|
||
if action is not None:
|
||
out["service_action"] = type(action).__name__.replace("MessageAction", "")
|
||
if duration := getattr(action, "duration", None):
|
||
out["service_action_duration_sec"] = duration
|
||
if getattr(msg, "views", None):
|
||
out["views"] = msg.views
|
||
if getattr(msg, "via_bot_id", None):
|
||
out["via_bot_id"] = msg.via_bot_id
|
||
if getattr(msg, "pinned", False):
|
||
out["pinned"] = True
|
||
if getattr(msg, "grouped_id", None):
|
||
out["album_id"] = str(msg.grouped_id)
|
||
return out
|
||
|
||
|
||
# медиа без файла: качать нечего, для них download_media вернёт None.
|
||
# MessageMediaContact сюда НЕ входит — Telethon сохраняет контакт как .vcard.
|
||
NO_FILE_MEDIA = frozenset({
|
||
"MessageMediaWebPage", "MessageMediaGeo", "MessageMediaGeoLive",
|
||
"MessageMediaPoll", "MessageMediaDice", "MessageMediaVenue",
|
||
"MessageMediaGame", "MessageMediaUnsupported", "MessageMediaStory",
|
||
"MessageMediaInvoice", "MessageMediaGiveaway",
|
||
"MessageMediaGiveawayResults", "MessageMediaEmpty",
|
||
})
|
||
|
||
|
||
def raw_media_tags(media: dict) -> set[str]:
|
||
"""Метки медиа из сырого to_dict(): тип, mime, расширение, вид документа.
|
||
|
||
Позволяет отсеивать по `--skip-media webp` / `sticker` / `video` ещё до
|
||
запроса к API — по данным, которые уже лежат в messages.jsonl.
|
||
"""
|
||
if not media:
|
||
return set()
|
||
kind = media.get("_", "")
|
||
tags = {kind.replace("MessageMedia", "").lower()}
|
||
if kind == "MessageMediaPhoto":
|
||
tags |= {"photo", "jpg", "jpeg", "image/jpeg"}
|
||
|
||
doc = media.get("document") or {}
|
||
mime = (doc.get("mime_type") or "").lower()
|
||
if mime:
|
||
tags.add(mime)
|
||
tags.add(mime.rsplit("/", 1)[-1])
|
||
|
||
for attr in doc.get("attributes") or []:
|
||
name = attr.get("_", "")
|
||
if name == "DocumentAttributeSticker":
|
||
tags.add("sticker")
|
||
elif name == "DocumentAttributeAudio":
|
||
tags.add("voice" if attr.get("voice") else "audio")
|
||
elif name == "DocumentAttributeVideo":
|
||
tags.add("round" if attr.get("round_message") else "video")
|
||
elif name == "DocumentAttributeAnimated":
|
||
tags.add("gif")
|
||
elif name == "DocumentAttributeFilename":
|
||
filename = attr.get("file_name") or ""
|
||
if "." in filename:
|
||
tags.add(filename.rsplit(".", 1)[-1].lower())
|
||
return tags
|
||
|
||
|
||
def has_downloadable_media(msg) -> bool:
|
||
media = getattr(msg, "media", None)
|
||
return media is not None and type(media).__name__ not in NO_FILE_MEDIA
|
||
|
||
|
||
def load_media_index(folder: Path) -> dict[str, dict]:
|
||
path = folder / "media_index.json"
|
||
if not path.exists():
|
||
return {}
|
||
try:
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
except (ValueError, OSError):
|
||
return {}
|
||
|
||
|
||
def assemble_pretty(folder: Path, header: dict) -> int:
|
||
"""Собирает messages.json из накопленного _pretty.jsonl (устойчиво к докачке)."""
|
||
src = folder / "_pretty.jsonl"
|
||
if not src.exists():
|
||
return 0
|
||
by_id: dict[int, dict] = {}
|
||
with src.open(encoding="utf-8") as handle:
|
||
for line in handle:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
rec = json.loads(line)
|
||
except ValueError:
|
||
continue
|
||
by_id[rec["id"]] = rec
|
||
|
||
# подмешиваем результаты догрузки медиа, если она уже прошла
|
||
for msg_id, entry in load_media_index(folder).items():
|
||
rec = by_id.get(int(msg_id))
|
||
if rec is None or "media" not in rec:
|
||
continue
|
||
# не ссылаемся на файлы, которых уже нет на диске (удалили вручную и т.п.)
|
||
if "file" in entry and not (folder / entry["file"]).exists():
|
||
continue
|
||
rec["media"].update({k: v for k, v in entry.items() if v is not None})
|
||
|
||
messages = [by_id[key] for key in sorted(by_id)]
|
||
payload = {
|
||
"chat": header,
|
||
"messages_count": len(messages),
|
||
"first_message_date": messages[0]["date"] if messages else None,
|
||
"last_message_date": messages[-1]["date"] if messages else None,
|
||
"messages": messages,
|
||
}
|
||
(folder / "messages.json").write_text(
|
||
json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8"
|
||
)
|
||
return len(messages)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# выгрузка одного диалога
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
async def export_dialog(client, dialog, out_root: Path, args) -> dict:
|
||
entity = dialog.entity
|
||
kind = dialog_kind(entity)
|
||
folder = out_root / "chats" / f"{dialog.id}_{slugify(dialog.name)}"
|
||
media_dir = folder / "media"
|
||
folder.mkdir(parents=True, exist_ok=True)
|
||
|
||
state = State(folder / "_state.json")
|
||
start_from = 0 if args.full else state.last_id
|
||
|
||
header = {
|
||
"id": dialog.id,
|
||
"name": dialog.name,
|
||
"kind": kind,
|
||
"username": getattr(entity, "username", None),
|
||
"phone": getattr(entity, "phone", None),
|
||
"is_archived": bool(getattr(dialog, "archived", False)),
|
||
"exported_at": dt.datetime.now(dt.timezone.utc).isoformat(),
|
||
}
|
||
(folder / "meta.json").write_text(
|
||
json.dumps({**header, "entity": entity.to_dict()},
|
||
ensure_ascii=False, indent=2, cls=Encoder),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
excluded_types = {t.lower().lstrip(".") for t in args.skip_media}
|
||
mode = "w" if args.full else "a"
|
||
jsonl = (folder / "messages.jsonl").open(mode, encoding="utf-8")
|
||
txt = (folder / "messages.txt").open(mode, encoding="utf-8") if args.txt else None
|
||
pretty = (folder / "_pretty.jsonl").open(mode, encoding="utf-8") if args.pretty else None
|
||
cache: dict = {}
|
||
new_msgs = 0
|
||
new_media = 0
|
||
media_bytes = 0
|
||
errors_count = 0
|
||
|
||
try:
|
||
async for msg in client.iter_messages(
|
||
entity,
|
||
reverse=True, # от старых к новым — так работает докачка
|
||
offset_id=start_from,
|
||
offset_date=args._since,
|
||
limit=args.limit,
|
||
wait_time=args.wait or None,
|
||
):
|
||
if STOP:
|
||
break
|
||
if args._until and msg.date and msg.date > args._until:
|
||
break
|
||
|
||
record = msg.to_dict()
|
||
|
||
if args.media and has_downloadable_media(msg):
|
||
size = getattr(getattr(msg, "file", None), "size", None) or 0
|
||
hit = raw_media_tags(record.get("media") or {}) & excluded_types
|
||
if hit:
|
||
record["_media_skipped"] = f"type_excluded:{','.join(sorted(hit))}"
|
||
elif args.max_media_mb and size > args.max_media_mb * 1024 * 1024:
|
||
record["_media_skipped"] = f"too_large:{human_size(size)}"
|
||
else:
|
||
media_dir.mkdir(parents=True, exist_ok=True)
|
||
try:
|
||
path = await client.download_media(msg, file=str(media_dir / f"{msg.id}"))
|
||
if path:
|
||
record["_media_file"] = str(Path(path).relative_to(folder))
|
||
new_media += 1
|
||
media_bytes += size or Path(path).stat().st_size
|
||
except errors.FloodWaitError as exc:
|
||
print(f" FloodWait {exc.seconds}s на медиа #{msg.id} — жду", file=sys.stderr)
|
||
await asyncio.sleep(exc.seconds + 1)
|
||
except (errors.RPCError, OSError, ValueError) as exc:
|
||
record["_media_error"] = f"{type(exc).__name__}: {exc}"
|
||
errors_count += 1
|
||
|
||
jsonl.write(json.dumps(record, ensure_ascii=False, cls=Encoder) + "\n")
|
||
|
||
label = await sender_label(client, msg, cache) if (txt or pretty) else ""
|
||
if txt is not None:
|
||
txt.write(render_line(msg, label))
|
||
if pretty is not None:
|
||
pretty.write(json.dumps(simplify_message(msg, label, record),
|
||
ensure_ascii=False, cls=Encoder) + "\n")
|
||
|
||
state.bump(msg.id, media=1 if "_media_file" in record else 0)
|
||
new_msgs += 1
|
||
if new_msgs % 500 == 0:
|
||
jsonl.flush()
|
||
for stream in (txt, pretty):
|
||
if stream is not None:
|
||
stream.flush()
|
||
state.save()
|
||
print(f" … {new_msgs} новых (всего {state.data['count']})", flush=True)
|
||
finally:
|
||
jsonl.close()
|
||
for stream in (txt, pretty):
|
||
if stream is not None:
|
||
stream.close()
|
||
state.save()
|
||
if args.pretty:
|
||
assemble_pretty(folder, header)
|
||
|
||
return {
|
||
"id": dialog.id,
|
||
"name": dialog.name,
|
||
"kind": kind,
|
||
"new_messages": new_msgs,
|
||
"total_messages": state.data["count"],
|
||
"new_media": new_media,
|
||
"media_bytes": media_bytes,
|
||
"media_errors": errors_count,
|
||
"folder": str(folder.relative_to(out_root)),
|
||
}
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# отбор диалогов
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
def dialog_matches(dialog, args) -> bool:
|
||
kind = dialog_kind(dialog.entity)
|
||
if args.kinds and kind not in args.kinds:
|
||
return False
|
||
if not args.archived and getattr(dialog, "archived", False):
|
||
return False
|
||
|
||
if args.skip:
|
||
for needle in args.skip:
|
||
if _needle_hits(dialog, needle):
|
||
return False
|
||
if args.only:
|
||
return any(_needle_hits(dialog, needle) for needle in args.only)
|
||
return True
|
||
|
||
|
||
def _needle_hits(dialog, needle: str) -> bool:
|
||
needle = needle.strip()
|
||
if not needle:
|
||
return False
|
||
if needle.lstrip("-").isdigit() and int(needle) == dialog.id:
|
||
return True
|
||
username = getattr(dialog.entity, "username", None) or ""
|
||
if needle.lstrip("@").lower() == username.lower() and username:
|
||
return True
|
||
return needle.lower() in (dialog.name or "").lower()
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# команды
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
async def cmd_login(client, args) -> None: # noqa: ARG001
|
||
me = await client.get_me()
|
||
print(f"OK: {me.first_name or ''} {me.last_name or ''} "
|
||
f"(@{me.username or '—'}, id={me.id}). Сессия: {args.session}")
|
||
|
||
|
||
async def cmd_list(client, args) -> None:
|
||
rows = []
|
||
async for dialog in client.iter_dialogs():
|
||
if not dialog_matches(dialog, args):
|
||
continue
|
||
rows.append((dialog.id, dialog_kind(dialog.entity), dialog.name or "",
|
||
getattr(dialog.entity, "username", None) or "",
|
||
getattr(dialog, "unread_count", 0)))
|
||
width = max((len(r[2]) for r in rows), default=10)
|
||
print(f"{'ID':>15} {'ТИП':<10} {'НАЗВАНИЕ':<{min(width, 50)}} USERNAME")
|
||
for did, kind, name, username, _ in rows:
|
||
print(f"{did:>15} {kind:<10} {name[:50]:<{min(width, 50)}} "
|
||
f"{'@' + username if username else ''}")
|
||
print(f"\nвсего: {len(rows)}")
|
||
|
||
|
||
async def cmd_media(client, args) -> None:
|
||
"""Догружает медиа к уже выгруженным сообщениям.
|
||
|
||
Идёт не по API-истории, а по локальному messages.jsonl, поэтому:
|
||
* не зависит от прогресса `run` и не мешает ему;
|
||
* инкрементальна — уже скачанное фиксируется в media_index.json;
|
||
* может гоняться повторно, пока `run` дописывает историю.
|
||
"""
|
||
out_root = Path(args.out).expanduser().resolve()
|
||
folders = sorted(p for p in (out_root / "chats").glob("*") if p.is_dir())
|
||
if not folders:
|
||
sys.exit(f"В {out_root / 'chats'} нет выгруженных чатов — сначала запусти `run`.")
|
||
|
||
limit_bytes = int(args.max_media_mb * 1024 * 1024) if args.max_media_mb else 0
|
||
grand_new = grand_bytes = grand_skip = grand_err = 0
|
||
|
||
for folder in folders:
|
||
try:
|
||
header = json.loads((folder / "meta.json").read_text(encoding="utf-8"))
|
||
except (ValueError, OSError):
|
||
print(f"[!] {folder.name}: нет читаемого meta.json — пропуск", file=sys.stderr)
|
||
continue
|
||
|
||
needles = args.only or []
|
||
if needles and not any(
|
||
n.lstrip("@").lower() in ((header.get("username") or "").lower(),
|
||
str(header.get("id")))
|
||
or n.lower() in (header.get("name") or "").lower()
|
||
for n in needles
|
||
):
|
||
continue
|
||
|
||
raw = folder / "messages.jsonl"
|
||
if not raw.exists():
|
||
continue
|
||
|
||
index = load_media_index(folder)
|
||
excluded = {t.lower().lstrip(".") for t in args.skip_media}
|
||
todo: list[int] = []
|
||
by_type_excluded = 0
|
||
with raw.open(encoding="utf-8") as handle:
|
||
for line in handle:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
rec = json.loads(line)
|
||
except ValueError:
|
||
continue # хвост может быть недописан, если `run` ещё идёт
|
||
media = rec.get("media")
|
||
if not media or media.get("_") in NO_FILE_MEDIA:
|
||
index.pop(str(rec["id"]), None) # чистим ложные nothing_to_download
|
||
continue
|
||
if excluded and (raw_media_tags(media) & excluded):
|
||
index.pop(str(rec["id"]), None) # забываем то, что решили не хранить
|
||
by_type_excluded += 1
|
||
continue
|
||
if rec.get("_media_file"):
|
||
continue
|
||
# готово только то, у чего реально есть файл: записи skipped/error
|
||
# переспрашиваются — иначе поднятый --max-media-mb ничего не даст
|
||
if "file" in index.get(str(rec["id"]), {}):
|
||
continue
|
||
todo.append(rec["id"])
|
||
|
||
print(f"\n{folder.name}: к догрузке {len(todo)} медиа "
|
||
f"(уже есть {len(index)}"
|
||
+ (f", исключено по типу {by_type_excluded}" if by_type_excluded else "")
|
||
+ ")", flush=True)
|
||
if not todo:
|
||
continue
|
||
|
||
entity = await client.get_entity(int(header["id"]))
|
||
media_dir = folder / "media"
|
||
media_dir.mkdir(parents=True, exist_ok=True)
|
||
sem = asyncio.Semaphore(max(1, args.jobs))
|
||
done = new = skipped = failed = 0
|
||
got_bytes = 0
|
||
|
||
async def fetch(msg) -> None:
|
||
nonlocal done, new, skipped, failed, got_bytes
|
||
async with sem:
|
||
if msg is None or STOP:
|
||
return
|
||
size = getattr(getattr(msg, "file", None), "size", None) or 0
|
||
if limit_bytes and size > limit_bytes:
|
||
index[str(msg.id)] = {"skipped": f"too_large:{human_size(size)}"}
|
||
skipped += 1
|
||
else:
|
||
try:
|
||
path = await client.download_media(
|
||
msg, file=str(media_dir / f"{msg.id}"))
|
||
if path:
|
||
index[str(msg.id)] = {
|
||
"file": str(Path(path).relative_to(folder))}
|
||
new += 1
|
||
got_bytes += size or Path(path).stat().st_size
|
||
else:
|
||
index[str(msg.id)] = {"error": "nothing_to_download"}
|
||
failed += 1
|
||
except errors.FloodWaitError as exc:
|
||
print(f" FloodWait {exc.seconds}s — жду", file=sys.stderr, flush=True)
|
||
await asyncio.sleep(exc.seconds + 1)
|
||
return
|
||
except (errors.RPCError, OSError, ValueError) as exc:
|
||
index[str(msg.id)] = {"error": f"{type(exc).__name__}: {exc}"}
|
||
failed += 1
|
||
done += 1
|
||
if done % 200 == 0:
|
||
(folder / "media_index.json").write_text(
|
||
json.dumps(index, ensure_ascii=False, indent=1), encoding="utf-8")
|
||
print(f" … {done}/{len(todo)} +{new} файлов "
|
||
f"({human_size(got_bytes)}), крупных пропущено {skipped}, "
|
||
f"ошибок {failed}", flush=True)
|
||
|
||
for start in range(0, len(todo), 100):
|
||
if STOP:
|
||
break
|
||
chunk = todo[start:start + 100]
|
||
try:
|
||
msgs = await client.get_messages(entity, ids=chunk)
|
||
except errors.FloodWaitError as exc:
|
||
print(f" FloodWait {exc.seconds}s на выборке — жду", file=sys.stderr, flush=True)
|
||
await asyncio.sleep(exc.seconds + 1)
|
||
msgs = await client.get_messages(entity, ids=chunk)
|
||
await asyncio.gather(*(fetch(m) for m in msgs))
|
||
|
||
(folder / "media_index.json").write_text(
|
||
json.dumps(index, ensure_ascii=False, indent=1), encoding="utf-8")
|
||
if args.pretty:
|
||
assemble_pretty(folder, header)
|
||
print(f" итого: +{new} файлов ({human_size(got_bytes)}), "
|
||
f"крупных пропущено {skipped}, ошибок {failed}", flush=True)
|
||
grand_new += new
|
||
grand_bytes += got_bytes
|
||
grand_skip += skipped
|
||
grand_err += failed
|
||
|
||
print(f"\nМедиа{' (прервано)' if STOP else ''}: +{grand_new} файлов "
|
||
f"({human_size(grand_bytes)}), пропущено по размеру {grand_skip}, "
|
||
f"ошибок {grand_err}.")
|
||
|
||
|
||
async def cmd_run(client, args) -> None:
|
||
out_root = Path(args.out).expanduser().resolve()
|
||
(out_root / "chats").mkdir(parents=True, exist_ok=True)
|
||
|
||
dialogs = []
|
||
async for dialog in client.iter_dialogs():
|
||
if dialog_matches(dialog, args):
|
||
dialogs.append(dialog)
|
||
|
||
me = await client.get_me()
|
||
print(f"Аккаунт: @{me.username or me.id} · диалогов к выгрузке: {len(dialogs)} · вывод: {out_root}\n")
|
||
|
||
index, summary = [], []
|
||
for n, dialog in enumerate(dialogs, 1):
|
||
if STOP:
|
||
break
|
||
kind = dialog_kind(dialog.entity)
|
||
print(f"[{n}/{len(dialogs)}] {kind:<10} {dialog.name} (id={dialog.id})", flush=True)
|
||
try:
|
||
result = await export_dialog(client, dialog, out_root, args)
|
||
except errors.FloodWaitError as exc:
|
||
print(f" FloodWait {exc.seconds}s — жду и повторяю", file=sys.stderr)
|
||
await asyncio.sleep(exc.seconds + 1)
|
||
result = await export_dialog(client, dialog, out_root, args)
|
||
except (errors.ChannelPrivateError, errors.ChatAdminRequiredError,
|
||
errors.UserBannedInChannelError) as exc:
|
||
print(f" пропуск: {type(exc).__name__}", file=sys.stderr)
|
||
result = {"id": dialog.id, "name": dialog.name, "kind": kind,
|
||
"error": type(exc).__name__}
|
||
except errors.RPCError as exc:
|
||
print(f" ошибка: {type(exc).__name__}: {exc}", file=sys.stderr)
|
||
result = {"id": dialog.id, "name": dialog.name, "kind": kind,
|
||
"error": f"{type(exc).__name__}: {exc}"}
|
||
|
||
index.append(result)
|
||
summary.append(result)
|
||
if "error" not in result:
|
||
print(f" +{result['new_messages']} сообщений, +{result['new_media']} медиа "
|
||
f"({human_size(result['media_bytes'])}), всего {result['total_messages']}",
|
||
flush=True)
|
||
|
||
(out_root / "dialogs.json").write_text(
|
||
json.dumps(index, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
total_msgs = sum(r.get("new_messages", 0) for r in summary)
|
||
total_media = sum(r.get("new_media", 0) for r in summary)
|
||
total_bytes = sum(r.get("media_bytes", 0) for r in summary)
|
||
failed = [r for r in summary if "error" in r]
|
||
print(f"\nГотово{' (прервано)' if STOP else ''}: "
|
||
f"{len(summary)} диалогов, +{total_msgs} сообщений, "
|
||
f"+{total_media} медиа ({human_size(total_bytes)}).")
|
||
if failed:
|
||
print(f"Недоступны ({len(failed)}): " +
|
||
", ".join(f"{r['name']} [{r['error']}]" for r in failed[:10]) +
|
||
(" …" if len(failed) > 10 else ""))
|
||
print(f"Каталог: {out_root}")
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# CLI
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
KINDS = ("user", "bot", "group", "supergroup", "channel")
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
p = argparse.ArgumentParser(
|
||
description="Выгрузка истории личного Telegram через Telethon",
|
||
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
||
)
|
||
p.add_argument("command", choices=("login", "list", "run", "media"))
|
||
p.add_argument("--session", default=str(ROOT / "tg_export.session"),
|
||
help="файл сессии Telethon")
|
||
p.add_argument("--out", default=str(ROOT / "out"), help="каталог для выгрузки")
|
||
|
||
sel = p.add_argument_group("отбор диалогов")
|
||
sel.add_argument("--kinds", nargs="*", choices=KINDS, default=list(KINDS),
|
||
help="какие типы чатов выгружать")
|
||
sel.add_argument("--only", nargs="*", default=[], metavar="NEEDLE",
|
||
help="только эти чаты: id, @username или часть названия")
|
||
sel.add_argument("--skip", nargs="*", default=[], metavar="NEEDLE",
|
||
help="исключить чаты (id / @username / часть названия)")
|
||
sel.add_argument("--archived", action="store_true", help="включить архив")
|
||
|
||
rng = p.add_argument_group("границы истории")
|
||
rng.add_argument("--since", metavar="YYYY-MM-DD", help="только сообщения не раньше даты")
|
||
rng.add_argument("--until", metavar="YYYY-MM-DD", help="только сообщения не позже даты")
|
||
rng.add_argument("--limit", type=int, default=None,
|
||
help="максимум сообщений на чат (для пробного прогона)")
|
||
rng.add_argument("--full", action="store_true",
|
||
help="игнорировать сохранённый прогресс и перезаписать всё с нуля")
|
||
|
||
med = p.add_argument_group("медиа и вывод")
|
||
med.add_argument("--media", action=argparse.BooleanOptionalAction, default=True,
|
||
help="скачивать медиафайлы")
|
||
med.add_argument("--max-media-mb", type=float, default=50,
|
||
help="не качать файлы крупнее (0 — без лимита)")
|
||
med.add_argument("--skip-media", nargs="*", default=[], metavar="TAG",
|
||
help="не качать медиа с такими метками: расширение (webp, mp4), "
|
||
"mime (image/webp) или вид (sticker, video, voice, round, gif)")
|
||
med.add_argument("--pretty", action=argparse.BooleanOptionalAction, default=True,
|
||
help="писать читаемый messages.json (плоские поля, отступы)")
|
||
med.add_argument("--txt", action=argparse.BooleanOptionalAction, default=True,
|
||
help="дополнительно писать читаемый messages.txt")
|
||
med.add_argument("--wait", type=float, default=0.0,
|
||
help="пауза между запросами истории, сек (мягче к лимитам)")
|
||
med.add_argument("--jobs", type=int, default=4,
|
||
help="параллельных загрузок в команде media")
|
||
return p
|
||
|
||
|
||
async def main_async(args) -> None:
|
||
api_id, api_hash = require_credentials()
|
||
client = TelegramClient(args.session, api_id, api_hash)
|
||
client.flood_sleep_threshold = 24 * 60 * 60 # сам ждёт длинные FloodWait
|
||
|
||
await client.start(
|
||
phone=lambda: input("Номер телефона (в формате +7...): ").strip(),
|
||
password=lambda: input("Пароль двухфакторной аутентификации: ").strip(),
|
||
)
|
||
try:
|
||
handler = {"login": cmd_login, "list": cmd_list,
|
||
"run": cmd_run, "media": cmd_media}[args.command]
|
||
await handler(client, args)
|
||
finally:
|
||
await client.disconnect()
|
||
|
||
|
||
def main() -> None:
|
||
load_dotenv(ROOT / ".env")
|
||
args = build_parser().parse_args()
|
||
args._since = parse_date(args.since)
|
||
args._until = parse_date(args.until)
|
||
if args.max_media_mb <= 0:
|
||
args.max_media_mb = 0
|
||
install_signal_handlers()
|
||
try:
|
||
asyncio.run(main_async(args))
|
||
except (errors.PhoneNumberInvalidError, errors.SessionPasswordNeededError) as exc:
|
||
sys.exit(f"Ошибка авторизации: {type(exc).__name__}: {exc}")
|
||
except KeyboardInterrupt:
|
||
sys.exit(130)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|