coruna-lab/channel-builder/tools/tglib_to_session_files.py

#!/usr/bin/env python3
"""
tglib_to_session_files.py — Convert tglib.js JSON payload to the Telethon
"session trio" (SQLite .session + metadata .json + session_string _密钥.txt).

Usage:
    python tglib_to_session_files.py <input.json> <output_dir>

Reads the tglib.js JSON (state + db_sqlite), extracts the master MTProto
auth_key + DC id + user id, builds a Telethon SQLite session, derives a
StringSession, and writes three files into <output_dir>:
    {phone}.session        — Telethon SQLite session (binary)
    {phone}.json           — Account metadata + session_string
    {phone}_密钥.txt       — session_string plain text

Works fully offline — no Telegram connection is made.
"""
import json, base64, os, sys, tempfile, shutil

# Standard Telegram production DC endpoints (used to seed the Telethon session).
DC_ADDRS = {
    1: ("149.154.175.50", 443),
    2: ("149.154.167.51", 443),
    3: ("149.154.175.100", 443),
    4: ("149.154.167.91", 443),
    5: ("91.108.56.130", 443),
}

# Telegram Desktop official API credentials (used as defaults in metadata).
DEFAULT_API_ID = 2040
DEFAULT_API_HASH = "b18441a1ff607e10a989891a5462e627"


def extract_keys(payload: dict):
    """Extract master auth_key, dc_id, user_id, phone from tglib.js JSON."""
    state_b64 = payload.get("state")
    if not state_b64:
        raise ValueError("missing 'state' field")
    state = json.loads(base64.b64decode(state_b64))
    records = state.get("records", [])
    if not records:
        raise ValueError("no records in state")
    backup_b64 = None
    for attr in records[0].get("attributes", []):
        if isinstance(attr, dict) and "backupData" in attr:
            backup_b64 = attr["backupData"]["data"]
            break
    if not backup_b64:
        raise ValueError("no backupData in state records")
    backup = json.loads(base64.b64decode(backup_b64))
    auth_key = base64.b64decode(backup["masterDatacenterKey"])
    dc_id = backup["masterDatacenterId"]
    user_id = backup.get("peerId", 0)
    if len(auth_key) != 256:
        raise ValueError(f"auth_key must be 256 bytes, got {len(auth_key)}")
    # tglib.js stores the phone number (E.164 without +) in the top-level
    # "user_id" field; the Telegram user id is in backupData.peerId.
    phone = str(payload.get("user_id") or user_id)
    return auth_key, dc_id, user_id, phone


def make_session(tmpdir: str, auth_key: bytes, dc_id: int) -> str:
    """Create a Telethon SQLite session file with the given auth key + DC."""
    from telethon.sessions import SQLiteSession
    server, port = DC_ADDRS.get(dc_id, ("149.154.167.91", 443))
    path = os.path.join(tmpdir, "tg")
    sess = SQLiteSession(path)
    sess._conn.execute("DELETE FROM sessions")
    sess._conn.execute(
        "INSERT INTO sessions (dc_id, server_address, port, auth_key) VALUES (?,?,?,?)",
        (dc_id, server, port, auth_key),
    )
    sess._conn.commit()
    sess.close()
    return path + ".session"


def session_string_from(session_file: str) -> str:
    """Convert a Telethon SQLite session file to a StringSession string."""
    from telethon.sessions import StringSession, SQLiteSession
    return StringSession.save(SQLiteSession(session_file))


def build_metadata(user_id, phone: str, session_string: str) -> dict:
    """Build the account metadata JSON (matching the reference format)."""
    return {
        "api_id": DEFAULT_API_ID,
        "api_hash": DEFAULT_API_HASH,
        "device_model": "Telegram Desktop",
        "system_version": "Windows 10 x64",
        "app_version": "4.14.4 x64",
        "system_lang_code": "en-US",
        "lang_pack": "tdesktop",
        "lang_code": "en",
        "user_id": user_id,
        "phone": phone,
        "twofa": "",
        "password": "",
        "session_string": session_string,
        "app_id": DEFAULT_API_ID,
        "app_hash": DEFAULT_API_HASH,
        "session_file": phone,
        "device": "Telegram Desktop",
        "username": "",
        "sex": None,
        "tz_offset": 28800,
        "avatar": "img/default.png",
        "device_token": "__FIREBASE_FAILED__",
        "package_id": "",
        "installer": "",
        "ipv6": False,
        "pref_cat": 2,
        "block": False,
        "premium": False,
    }


def main():
    if len(sys.argv) != 3:
        print("usage: tglib_to_session_files.py <input.json> <output_dir>", file=sys.stderr)
        sys.exit(1)
    input_json, output_dir = sys.argv[1], sys.argv[2]
    payload = json.load(open(input_json))
    auth_key, dc_id, user_id, phone = extract_keys(payload)
    print(f"auth_key: {len(auth_key)}B, dc_id: {dc_id}, user_id: {user_id}, phone: {phone}", file=sys.stderr)

    os.makedirs(output_dir, exist_ok=True)
    tmpdir = tempfile.mkdtemp(prefix="tglib_sess_")
    try:
        session_file = make_session(tmpdir, auth_key, dc_id)
        ss = session_string_from(session_file)

        # 1) {phone}.session — copy the SQLite session file
        out_session = os.path.join(output_dir, f"{phone}.session")
        shutil.copy(session_file, out_session)

        # 2) {phone}.json — metadata + session_string
        meta = build_metadata(user_id, phone, ss)
        out_json = os.path.join(output_dir, f"{phone}.json")
        with open(out_json, "w", encoding="utf-8") as f:
            json.dump(meta, f, ensure_ascii=False, indent=4)

        # 3) {phone}_密钥.txt — session_string plain text
        out_key = os.path.join(output_dir, f"{phone}_密钥.txt")
        with open(out_key, "w", encoding="utf-8") as f:
            f.write(ss)

        print(f"wrote: {out_session}, {out_json}, {out_key}", file=sys.stderr)
    finally:
        shutil.rmtree(tmpdir, ignore_errors=True)


if __name__ == "__main__":
    main()