Script to convert a decrypted Conversations database to a Telegram data export, to be imported by signalbackup-tools.

con2tg4sig.py

#!/usr/bin/env python3

"""
Conversion from a Conversations [1] decrypted database [2] to a
Telegram data export [3], to be imported by signalbackup-tools [4].

[1] https://conversations.im
[2] https://gist.github.com/Terrance/4d9df2d550806fb038d18956fbe3b268
[3] https://core.telegram.org/import-export
[4] https://github.com/bepaald/signalbackup-tools
"""

from collections.abc import Sequence
import json
from pathlib import Path
import re
from shutil import rmtree
import sqlite3
from typing import Any


def main(
    con_db_path: Path,
    con_media_path: Path,
    output_path: Path,
    con_convs: Sequence[str],
    output_overwrite: bool,
):
    if output_overwrite:
        rmtree(output_path, ignore_errors=True)
    output_path.mkdir(parents=True)
    con_db = sqlite3.connect(con_db_path)
    con_cur = con_db.cursor()
    conv_params = ", ".join("?" * len(con_convs))
    con_cur.execute(
        f"SELECT uuid, name FROM conversations WHERE uuid IN ({conv_params})",
        con_convs,
    )
    names: dict[str, str] = dict(con_cur)
    chats: dict[str, dict[str, Any]] = {
        uuid: {
            "id": i,
            "name": names[uuid],
            "type": "personal_chat",
            "messages": [],
        }
        for i, uuid in enumerate(con_convs)
    }
    con_cur.execute(
        "SELECT conversationUuid, timeSent, body, relativeFilePath, status, reactions "
        f"FROM messages WHERE conversationUuid IN ({conv_params})",
        tuple(con_convs),
    )
    for uuid, sent_ms, body, attach, status, reacts in con_cur:
        chat = chats[uuid]
        delivery = None
        # https://codeberg.org/iNPUTmice/Conversations/src/tag/2.18.2/src/main/java/eu/siacs/conversations/entities/Message.java#L38-L45
        if status == 0:
            sender = chat["id"]
            recipient = -1
        else:
            sender = -1
            recipient = chat["id"]
            if status in (1, 2, 5, 6):
                delivery = "sending"
            elif status == 3:
                delivery = "failed"
            elif status == 7:
                delivery = "delivered"
            elif status == 8:
                delivery = "read"
        id_ = len(chat["messages"])
        msg = {
            "id": id_,
            "type": "message",
            "date_unixtime": sent_ms,
            "from_id": sender,
            "text_entities": [],
        }
        if delivery:
            msg["custom_delivery_reports"] = [{
                "status": delivery,
                "recipient": recipient,
                "timestamp": sent_ms,
            }]
        chat["messages"].append(msg)
        if attach:
            name = Path(attach).name
            ext = name.rsplit(".", 1)[1]
            type_ = None
            if ext in ("jpg", "png", "gif"):
                store = "photos"
                mime = "image/{}".format(ext)
                if ext == "png":
                    type_ = "sticker"
                elif ext == "gif":
                    type_ = "animation"
            elif ext == "mp4":
                store = "video_files"
                mime = "video/mp4"
                type_ = "video_file"
            elif ext == "pdf":
                store = "files"
                mime = "application/pdf"
            else:
                print("Bad extension:", name)
                continue
            relative = Path("chats", f"chat_{chat['id']}", store, name)
            key = "photo" if store == "photos" else "file"
            msg.update({
                key: str(relative),
                "mime_type": mime,
            })
            if type_:
                msg["media_type"] = type_
            try:
                found = next(con_media_path.rglob(name))
            except StopIteration:
                print("Missing:", name)
                continue
            absolute = output_path / relative
            absolute.parent.mkdir(parents=True, exist_ok=True)
            absolute.hardlink_to(found)
        else:
            geo = re.match(r"^geo:([0-9\.]+),([0-9\.]+)$", body)
            if geo:
                msg["location_information"] = {
                    "latitude": float(geo[1]),
                    "longitude": float(geo[2]),
                }
            else:
                msg["text_entities"].append({
                    "type": "plain",
                    "text": body,
                })
        if reacts:
            msg["custom_reactions"] = [
                {
                    "emoji": react["reaction"],
                    # Assume no reactions on own messages.
                    "author": recipient,
                    "timestamp": sent_ms + 15000,
                }
                for react in json.loads(reacts)
            ]
    with open(output_path / "result.json", "w") as fp:
        json.dump({"chats": {"list": list(chats.values())}}, fp, indent=2)


if __name__ == "__main__":

    from argparse import ArgumentParser

    parser = ArgumentParser(description=__doc__.strip())
    parser.add_argument(
        "-f", "--overwrite", action="store_true", help="clear existing target directory",
    )
    parser.add_argument(
        "database", type=Path, help="Conversations decrypted database file",
    )
    parser.add_argument(
        "media", type=Path, help="Conversations media directory",
    )
    parser.add_argument(
        "output", type=Path, help="target directory for new export",
    )
    parser.add_argument(
        "convs", nargs="+", metavar="UUID", help="selected conversations",
    )
    args = parser.parse_args()
    main(args.database, args.media, args.output, args.convs, args.overwrite)