rpi: выгрузка Telegram второго аккаунта (tg_export с Mac) — stacks/rpi-tg-export
Пользователь tgexport, сессия /var/lib/tg-export (600), тексты в /mnt/backup/tg (группа astra), докачка раз в сутки без медиа, итог в backup-metrics. Вход делает владелец. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
gilfanovr
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
8373a87826
commit
8a06244ef4
@@ -0,0 +1,901 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user