scripts/chat_export_to_transcript.py — Chat Export to Transcript

#!/usr/bin/env python3
# By Zero Width Lab (zerowidthlab.com)
"""Convert official chat exports into uniform JSONL and readable transcripts.

By Zero Width Lab (zerowidthlab.com)
"""
from __future__ import annotations

import argparse
import errno
import csv
import hashlib
import io
import json
import os
import re
import stat
import sys
import unicodedata
import zipfile
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

__author__ = "Zero Width Lab"
VERSION = "0.1.1"
SIGNATURE = "By Zero Width Lab (zerowidthlab.com)"
MAX_INPUT_BYTES = 32 * 1024 * 1024
MAX_ZIP_MEMBERS = 100
CONTINUATION_PREFIX = "    | "


class CliError(Exception):
    pass


class Parser(argparse.ArgumentParser):
    def error(self, message: str) -> None:
        raise CliError(message)


def display_text(value: Any) -> str:
    parts = []
    for character in str(value):
        if character in {"\u2028", "\u2029"} or unicodedata.category(character).startswith("C"):
            number = ord(character)
            parts.append(f"\\u{number:04x}" if number <= 65535 else f"\\U{number:08x}")
        else:
            parts.append(character)
    return "".join(parts)


def require_platform() -> None:
    if (sys.platform not in {"linux", "darwin"} or os.name != "posix" or not hasattr(os, "O_DIRECTORY") or
            not hasattr(os, "O_NOFOLLOW") or os.open not in os.supports_dir_fd or
            os.mkdir not in os.supports_dir_fd or os.stat not in os.supports_dir_fd or
            os.stat not in os.supports_follow_symlinks or not hasattr(os, "O_NONBLOCK")):
        raise CliError("requires POSIX directory descriptors and no-follow APIs (macOS/Linux); Windows is unsupported")


def validate_path(path: Path) -> None:
    raw = str(path)
    if "\0" in raw or ".." in path.parts:
        raise CliError("path traversal and NUL are unsupported")


def open_directory(path: Path) -> int:
    validate_path(path)
    absolute = Path(os.path.abspath(path))
    descriptor = os.open("/", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
    try:
        for part in absolute.parts[1:]:
            try:
                next_descriptor = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                                          dir_fd=descriptor)
            except OSError as exc:
                if exc.errno in {errno.ENOTDIR, errno.ELOOP}:
                    raise CliError("a path component is a symlink; use the physical path (for example /private/tmp)") from exc
                raise
            os.close(descriptor)
            descriptor = next_descriptor
        return descriptor
    except BaseException:
        os.close(descriptor)
        raise


def path_kind(path: Path, directory: bool = False) -> bool:
    parent = open_directory(path.parent)
    try:
        try:
            metadata = os.stat(path.name, dir_fd=parent, follow_symlinks=False)
        except FileNotFoundError:
            return False
        if stat.S_ISLNK(metadata.st_mode):
            raise CliError(f"symlink input is unsupported: {display_text(path)}")
        return stat.S_ISDIR(metadata.st_mode) if directory else stat.S_ISREG(metadata.st_mode)
    finally:
        os.close(parent)


def directory_children(path: Path, directories: bool = False) -> list[Path]:
    descriptor = open_directory(path)
    try:
        children = []
        for name in sorted(os.listdir(descriptor)):
            metadata = os.stat(name, dir_fd=descriptor, follow_symlinks=False)
            if stat.S_ISLNK(metadata.st_mode):
                raise CliError(f"symlink in export is unsupported: {display_text(path / name)}")
            if not directories or stat.S_ISDIR(metadata.st_mode):
                children.append(path / name)
        return children
    finally:
        os.close(descriptor)


def reject_export_symlinks(root: Path) -> None:
    """Check export directory metadata without opening unrelated file contents."""
    stack = [root]
    while stack:
        directory = stack.pop()
        for child in directory_children(directory):
            if path_kind(child, directory=True):
                stack.append(child)


def identity(descriptor: int) -> list[int]:
    value = os.fstat(descriptor)
    return [value.st_dev, value.st_ino]


def read_file(path: Path) -> bytes:
    validate_path(path)
    try:
        parent = open_directory(path.parent)
        try:
            descriptor = os.open(path.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent)
        finally:
            os.close(parent)
        with os.fdopen(descriptor, "rb") as handle:
            metadata = os.fstat(handle.fileno())
            if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > MAX_INPUT_BYTES:
                raise CliError(f"input must be a regular file of at most {MAX_INPUT_BYTES} bytes: {display_text(path)}")
            data = handle.read(MAX_INPUT_BYTES + 1)
            if len(data) > MAX_INPUT_BYTES:
                raise CliError(f"input exceeds {MAX_INPUT_BYTES} bytes: {display_text(path)}")
            return data
    except OSError as exc:
        raise CliError(f"cannot read input: {display_text(path)}: {display_text(exc)}") from exc


def decoded(data: bytes, label: str) -> str:
    try:
        return data.decode("utf-8-sig")
    except UnicodeDecodeError as exc:
        raise CliError(f"invalid UTF-8 in {display_text(label)}") from exc


def json_file(path: Path) -> Any:
    try:
        return json.loads(decoded(read_file(path), str(path)))
    except (json.JSONDecodeError, RecursionError) as exc:
        raise CliError(f"invalid JSON in {display_text(path)}: {display_text(exc)}") from exc


def valid_timestamp(value: Any) -> str | None:
    if not isinstance(value, str) or not value.strip():
        return None
    if not re.fullmatch(
        r"(?:\d{4}-\d{2}-\d{2}|\d{8})(?:[T ](?:\d{2}:\d{2}(?::\d{2})?|\d{4}(?:\d{2})?)(?:[.,]\d+)?(?:Z|[+-]\d{2}:?\d{2})?)?",
        value.strip(), flags=re.ASCII):
        return None
    match = re.fullmatch(
        r"(\d{4})-?(\d{2})-?(\d{2})(?:[T ](\d{2}):?(\d{2})(?::?(\d{2}))?(?:[.,](\d+))?(Z|[+-]\d{2}:?\d{2})?)?",
        value.strip(), flags=re.ASCII)
    if not match:
        return None
    year, month, day, hour, minute, second, fraction, offset = match.groups()
    candidate = f"{year}-{month}-{day}"
    if hour is not None:
        candidate += f"T{hour}:{minute}:{second or '00'}"
        if fraction:
            candidate += "." + (fraction + "000000")[:6]
        if offset:
            candidate += "+00:00" if offset == "Z" else offset[:3] + ":" + offset[-2:]
    try:
        parsed = datetime.fromisoformat(candidate)
    except ValueError:
        return None
    return parsed.isoformat(timespec="seconds")


def validate_record_fields(record: dict[str, Any], label: str, fields: tuple[str, ...]) -> None:
    for field in fields:
        if field in record and record[field] is not None and not isinstance(record[field], (str, int)):
            raise CliError(f"{label}: {field} must be text or an integer")


def validate_text_fields(record: dict[str, Any], label: str, fields: tuple[str, ...]) -> None:
    for field in fields:
        if field in record and record[field] is not None and not isinstance(record[field], str):
            raise CliError(f"{label}: {field} must be text")


def message(platform: str, conversation: str, message_id: Any, timestamp: str,
            author: str, text: str, attachments: list[str] | None = None,
            reply_to: Any = None, system: bool = False) -> dict[str, Any]:
    return {"platform": platform, "conversation": conversation,
            "message_id": str(message_id) if message_id is not None else None,
            "timestamp": timestamp, "author": author, "text": text,
            "attachments": attachments or [],
            "reply_to": str(reply_to) if reply_to is not None else None,
            "system": system}


def wa_timestamp(date_text: str, time_text: str, day_first: bool) -> str | None:
    date_text = date_text.replace(".", "/").replace("-", "/")
    fields = date_text.split("/")
    if len(fields) != 3:
        return None
    try:
        if len(fields[0]) == 4:
            year, month, day = (int(item) for item in fields)
        else:
            first, second, year = (int(item) for item in fields)
            if year < 100:
                year += 2000
            month, day = (second, first) if day_first or first > 12 else (first, second)
        normalized = time_text.strip().replace("\u202f", " ").replace("\u00a0", " ")
        parsed_time = None
        for pattern in ("%I:%M:%S %p", "%I:%M %p", "%H:%M:%S", "%H:%M"):
            try:
                parsed_time = datetime.strptime(normalized.upper(), pattern).time()
                break
            except ValueError:
                continue
        if parsed_time is None:
            return None
        return datetime(year, month, day, parsed_time.hour, parsed_time.minute,
                        parsed_time.second).isoformat(timespec="seconds")
    except ValueError:
        return None


def parse_whatsapp(data: bytes, conversation: str, day_first: bool) -> tuple[list[dict[str, Any]], list[str]]:
    text = decoded(data, conversation)
    records: list[dict[str, Any]] = []
    skipped: list[str] = []
    current: dict[str, Any] | None = None
    pattern = (r"^(?:\[)?(\d{1,4}[/.\-]\d{1,2}[/.\-]\d{1,4}),?\s+"
               r"(\d{1,2}:\d{2}(?::\d{2})?(?:\s*[AaPp]\.?[Mm]\.?)?)"
               r"(?:\])?\s*(?:-\s*)?(.*)$")
    for number, line in enumerate(text.splitlines(), 1):
        match = re.match(pattern, line)
        stamp = wa_timestamp(match.group(1), match.group(2).replace(".", ""), day_first) if match else None
        if stamp:
            payload = match.group(3).lstrip()
            author_match = re.match(r"^([^:]{1,200}):\s?(.*)$", payload)
            author = author_match.group(1) if author_match else "System"
            body = author_match.group(2) if author_match else payload
            attachments = []
            if "<Media omitted>" in body:
                attachments.append("Media omitted")
            attachment_match = re.search(r"([^\\/\s<>]+\.[A-Za-z0-9]{1,8})\s*\(file attached\)", body)
            if attachment_match:
                attachments.append(attachment_match.group(1))
            current = message("whatsapp", conversation, None, stamp, author, body,
                              attachments, system=author_match is None)
            records.append(current)
        elif match:
            skipped.append(f"{conversation}: line {number}: invalid date or time")
            current = None
        elif current is not None:
            current["text"] += "\n" + line
        elif line.strip():
            skipped.append(f"{conversation}: line {number}: unparseable")
    return records, skipped


def parse_whatsapp_path(path: Path, day_first: bool) -> tuple[list[dict[str, Any]], list[str]]:
    if path.suffix.lower() == ".txt":
        return parse_whatsapp(read_file(path), path.stem, day_first)
    try:
        with zipfile.ZipFile(io.BytesIO(read_file(path))) as archive:
            infos = archive.infolist()
            if len(infos) > MAX_ZIP_MEMBERS:
                raise CliError("WhatsApp ZIP has too many members")
            for item in infos:
                member = Path(item.filename)
                validate_path(member)
                if "%" in item.filename or "\\" in item.filename:
                    raise CliError("encoded or backslash ZIP member is unsupported")
                if member.is_absolute() or re.match(r"^[A-Za-z]:", item.filename):
                    raise CliError("absolute WhatsApp ZIP member is unsupported")
            candidates = [item for item in infos if item.filename.lower().endswith(".txt") and not item.is_dir()]
            if len(candidates) != 1:
                raise CliError("WhatsApp ZIP must contain exactly one chat TXT file")
            info = candidates[0]
            if info.file_size > MAX_INPUT_BYTES:
                raise CliError("WhatsApp TXT in ZIP exceeds size limit")
            with archive.open(info) as handle:
                data = handle.read(MAX_INPUT_BYTES + 1)
            if len(data) > MAX_INPUT_BYTES:
                raise CliError("WhatsApp TXT in ZIP exceeds size limit")
            return parse_whatsapp(data, Path(info.filename).stem, day_first)
    except (OSError, zipfile.BadZipFile, RuntimeError, NotImplementedError) as exc:
        raise CliError(f"cannot read WhatsApp ZIP: {display_text(exc)}") from exc


def parse_discord(root: Path) -> tuple[list[dict[str, Any]], list[str]]:
    messages: list[dict[str, Any]] = []
    skipped: list[str] = []
    seen_channels: dict[str, str] = {}
    channels_root = root / "messages"
    if not path_kind(channels_root, directory=True):
        raise CliError("Discord export needs messages/<channel>/messages.json or messages.csv")
    for channel_dir in directory_children(channels_root, directories=True):
        if not path_kind(channel_dir, directory=True):
            continue
        json_path, csv_path = channel_dir / "messages.json", channel_dir / "messages.csv"
        if not path_kind(json_path) and not path_kind(csv_path):
            continue
        channel_meta = channel_dir / "channel.json"
        conversation = channel_dir.name
        channel_name = None
        if not path_kind(channel_meta):
            skipped.append(f"{channel_dir.name}: channel.json missing")
        else:
            metadata = json_file(channel_meta)
            if isinstance(metadata, dict):
                validate_record_fields(metadata, "channel.json", ("id",))
                validate_text_fields(metadata, "channel.json", ("name",))
                conversation = str(metadata.get("id") or conversation)
                channel_name = metadata.get("name")
            else:
                raise CliError(f"{channel_dir.name}: channel.json expected object")
        if conversation in seen_channels:
            raise CliError(f"duplicate Discord channel ID {conversation}: folders {seen_channels[conversation]} and {channel_dir.name}")
        seen_channels[conversation] = channel_dir.name
        if path_kind(json_path):
            payload = json_file(json_path)
            if not isinstance(payload, list):
                skipped.append(f"{channel_dir.name}/messages.json: expected array")
                continue
        else:
            try:
                payload = list(csv.DictReader(io.StringIO(decoded(read_file(csv_path), str(csv_path)))))
            except csv.Error as exc:
                skipped.append(f"{channel_dir.name}/messages.csv: {display_text(exc)}")
                continue
        for index, record in enumerate(payload, 1):
            if not isinstance(record, dict):
                skipped.append(f"{channel_dir.name}: record {index}: expected object")
                continue
            validate_record_fields(record, f"{channel_dir.name}: record {index}", ("ID", "id", "reply_to"))
            if "system" in record and not isinstance(record["system"], bool):
                raise CliError("Discord system field must be boolean")
            stamp = valid_timestamp(record.get("Timestamp") or record.get("timestamp"))
            content = record.get("Contents", record.get("content", ""))
            if stamp is None or not isinstance(content, str):
                skipped.append(f"{channel_dir.name}: record {index}: invalid timestamp or content")
                continue
            attachments_raw = record.get("Attachments", record.get("attachments", []))
            if isinstance(attachments_raw, str):
                attachments = [item.strip() for item in attachments_raw.split(",") if item.strip()]
            elif isinstance(attachments_raw, list):
                attachments = []
                for attachment in attachments_raw:
                    value = attachment.get("filename") or attachment.get("url") if isinstance(attachment, dict) else attachment
                    if not isinstance(value, str):
                        raise CliError(f"{channel_dir.name}: record {index}: invalid attachment")
                    attachments.append(value)
            else:
                raise CliError(f"{channel_dir.name}: record {index}: attachments expected string or array")
            author = record.get("Author", record.get("author", "Unknown"))
            if isinstance(author, dict):
                validate_text_fields(author, "Discord author", ("name", "username"))
                author = author.get("name") or author.get("username") or "Unknown"
            if not isinstance(author, str):
                raise CliError("Discord author must be a string or name object")
            messages.append(message("discord", conversation, record.get("ID", record.get("id")),
                                    stamp, str(author), content, attachments,
                                    record.get("reply_to"), bool(record.get("system", False))))
            messages[-1]["channel_name"] = channel_name
    if not messages and not skipped:
        raise CliError("no Discord message files found")
    return messages, skipped


def telegram_text(value: Any) -> str | None:
    if isinstance(value, str):
        return value
    if isinstance(value, list):
        parts = []
        for item in value:
            if isinstance(item, str):
                parts.append(item)
            elif isinstance(item, dict) and isinstance(item.get("text"), str):
                parts.append(item["text"])
            else:
                return None
        return "".join(parts)
    return None


def parse_telegram(path: Path) -> tuple[list[dict[str, Any]], list[str]]:
    if path.name != "result.json":
        raise CliError("Telegram Desktop JSON export must be result.json; HTML is unsupported")
    payload = json_file(path)
    if not isinstance(payload, dict) or not isinstance(payload.get("messages"), list):
        raise CliError("Telegram result.json needs a messages array")
    validate_text_fields(payload, "result.json", ("name",))
    conversation = str(payload.get("name") or "Telegram chat")
    messages = []
    skipped = []
    for index, item in enumerate(payload["messages"], 1):
        if not isinstance(item, dict):
            skipped.append(f"result.json: record {index}: expected object")
            continue
        validate_record_fields(item, f"result.json: record {index}", ("id", "reply_to_message_id"))
        validate_text_fields(item, f"result.json: record {index}", ("from", "actor", "type", "file", "photo"))
        stamp = valid_timestamp(item.get("date"))
        body = telegram_text(item.get("text", ""))
        if stamp is None or body is None:
            skipped.append(f"result.json: record {index}: invalid date or text")
            continue
        attachments = [str(item[key]) for key in ("file", "photo") if isinstance(item.get(key), str)]
        messages.append(message("telegram", conversation, item.get("id"), stamp,
                                str(item.get("from") or item.get("actor") or "System"), body,
                                attachments, item.get("reply_to_message_id"),
                                item.get("type") != "message"))
    return messages, skipped


def slack_timestamp(value: Any) -> str | None:
    try:
        number = float(value)
        return datetime.fromtimestamp(number, timezone.utc).isoformat(timespec="seconds")
    except (TypeError, ValueError, OverflowError, OSError):
        return None


def parse_slack(root: Path) -> tuple[list[dict[str, Any]], list[str]]:
    user_path = root / "users.json"
    if not path_kind(user_path):
        raise CliError("Slack export needs users.json")
    users_raw = json_file(user_path)
    if not isinstance(users_raw, list):
        raise CliError("Slack users.json must be an array")
    users = {}
    for item in users_raw:
        if not isinstance(item, dict) or not isinstance(item.get("id"), str):
            raise CliError("Slack users.json entry needs a string id")
        if "profile" in item and not isinstance(item["profile"], dict):
            raise CliError("Slack user profile must be an object")
        if isinstance(item, dict) and isinstance(item.get("id"), str):
            profile = item.get("profile") if isinstance(item.get("profile"), dict) else {}
            for values, fields in ((profile, ("display_name", "real_name")), (item, ("name",))):
                for field in fields:
                    if field in values and values[field] is not None and not isinstance(values[field], str):
                        raise CliError(f"Slack user {field} must be a string")
            users[item["id"]] = str(profile.get("display_name") or profile.get("real_name") or item.get("name") or item["id"])
    channels = directory_children(root, directories=True)
    legacy = root / "channels"
    if legacy in channels and not any(file.suffix == ".json" for file in directory_children(legacy)):
        channels.remove(legacy)
        channels.extend(directory_children(legacy, directories=True))
    if len({channel.name for channel in channels}) != len(channels):
        raise CliError("duplicate Slack channel folders across root and channels/ layouts")
    messages = []
    skipped = []
    for channel in channels:
        if not path_kind(channel, directory=True):
            continue
        for file in directory_children(channel):
            if file.suffix != ".json":
                continue
            payload = json_file(file)
            if not isinstance(payload, list):
                skipped.append(f"{channel.name}/{file.name}: expected array")
                continue
            for index, item in enumerate(payload, 1):
                if not isinstance(item, dict):
                    skipped.append(f"{channel.name}/{file.name}: record {index}: expected object")
                    continue
                validate_text_fields(item, f"{channel.name}/{file.name}: record {index}", ("user", "username", "client_msg_id", "thread_ts", "subtype"))
                stamp = slack_timestamp(item.get("ts"))
                body = item.get("text", "")
                if stamp is None or not isinstance(body, str):
                    skipped.append(f"{channel.name}/{file.name}: record {index}: invalid ts or text")
                    continue
                attachments_raw = item.get("files", [])
                if not isinstance(attachments_raw, list) or any(not isinstance(value, dict) for value in attachments_raw):
                    raise CliError(f"{channel.name}/{file.name}: record {index}: files must be an array of objects")
                attachments = []
                for value in attachments_raw:
                    label = value.get("name") or value.get("id")
                    if not isinstance(label, str):
                        raise CliError(f"{channel.name}/{file.name}: record {index}: file needs a string name or id")
                    attachments.append(label)
                user = str(item.get("user") or item.get("username") or "System")
                messages.append(message("slack", channel.name, item.get("client_msg_id") or item.get("ts"),
                                        stamp, users.get(user, user), body, attachments,
                                        item.get("thread_ts") if item.get("thread_ts") != item.get("ts") else None,
                                        item.get("subtype") not in (None, "")))
    if not messages and not skipped:
        raise CliError("no Slack channel files found")
    return messages, skipped


def detect_format(path: Path) -> str:
    if path_kind(path):
        if path.name == "result.json":
            return "telegram"
        if path.suffix.lower() in {".txt", ".zip"}:
            return "whatsapp"
    elif path_kind(path, directory=True):
        if path_kind(path / "users.json"):
            return "slack"
        if path_kind(path / "messages", directory=True):
            return "discord"
    raise CliError("cannot detect official export format; pass --format")


def transcript(messages: list[dict[str, Any]]) -> str:
    lines = []
    if messages and messages[0]["platform"] == "discord":
        first = messages[0]
        label = first.get("channel_name") or first["conversation"]
        lines.append(f"Discord channel: {display_text(label)} (ID: {display_text(first['conversation'])})")
    for item in messages:
        body_lines = item["text"].splitlines() or [""]
        body_lines = [display_text(line) for line in body_lines]
        extras = "".join(f"  [attachment: {display_text(name)}]" for name in item["attachments"])
        body_lines[-1] += extras
        lines.append(f"{display_text(item['timestamp'])}  {display_text(item['author'])}: {body_lines[0]}")
        lines.extend(CONTINUATION_PREFIX + line for line in body_lines[1:])
    return "\n".join(lines) + ("\n" if lines else "")


def render_files(messages: list[dict[str, Any]]) -> dict[str, bytes]:
    result = {"messages.jsonl": ("".join(json.dumps(item, ensure_ascii=True, sort_keys=True) + "\n" for item in messages)).encode("utf-8")}
    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for item in messages:
        grouped[item["conversation"]].append(item)
    used = set()
    for conversation in sorted(grouped):
        base = re.sub(r"[^A-Za-z0-9._-]+", "-", conversation).strip(".-_")[:60] or "conversation"
        suffix = hashlib.sha256(conversation.encode("utf-8")).hexdigest()[:8]
        name = f"transcripts/{base}-{suffix}.txt"
        if name in used:
            raise CliError("transcript filename collision")
        used.add(name)
        result[name] = transcript(grouped[conversation]).encode("utf-8")
    return result


def output_plan(path: Path) -> dict[str, Any]:
    validate_path(path)
    parent = open_directory(path.parent)
    try:
        parent_identity = identity(parent)
        try:
            descriptor = os.open(path.name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=parent)
        except FileNotFoundError:
            destination_identity = None
        else:
            try:
                if os.listdir(descriptor):
                    raise CliError("output directory must be new or empty")
                destination_identity = identity(descriptor)
            finally:
                os.close(descriptor)
        return {"destination": str(path.resolve(strict=False)),
                "parent_identity": parent_identity, "directory_identity": destination_identity,
                "create_destination": destination_identity is None}
    finally:
        os.close(parent)


def plan_digest(files: dict[str, bytes], destination: Path, approved: dict[str, Any] | None = None) -> str:
    plan = dict(approved) if approved is not None else output_plan(destination)
    plan["create_directories"] = (["."] if plan["create_destination"] else []) + (["transcripts"] if any(name.startswith("transcripts/") for name in files) else [])
    plan["files"] = [{"path": name, "size": len(files[name]),
                      "sha256": hashlib.sha256(files[name]).hexdigest()} for name in sorted(files)]
    return hashlib.sha256(json.dumps(plan, sort_keys=True, separators=(",", ":")).encode()).hexdigest()


def prepare_output(path: Path) -> None:
    output_plan(path)


def write_files(path: Path, files: dict[str, bytes], approved: dict[str, Any]) -> None:
    parent = open_directory(path.parent)
    destination = None
    transcripts_dir = None
    try:
        if identity(parent) != approved["parent_identity"] or str(path.resolve(strict=False)) != approved["destination"]:
            raise CliError("output parent changed after approval")
        if approved["create_destination"]:
            os.mkdir(path.name, mode=0o700, dir_fd=parent)
        destination = os.open(path.name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=parent)
        if approved["directory_identity"] is not None and identity(destination) != approved["directory_identity"]:
            raise CliError("output directory changed after approval")
        if os.listdir(destination):
            raise CliError("output directory must be new or empty")
        if any(name.startswith("transcripts/") for name in files):
            os.mkdir("transcripts", mode=0o700, dir_fd=destination)
            transcripts_dir = os.open("transcripts", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=destination)
        for name in sorted(files):
            target_dir = transcripts_dir if name.startswith("transcripts/") else destination
            descriptor = os.open(Path(name).name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
                                 0o600, dir_fd=target_dir)
            with os.fdopen(descriptor, "wb") as handle:
                handle.write(files[name])
    except OSError as exc:
        raise CliError(f"cannot create output: {display_text(exc)}; any partial output is retained") from exc
    finally:
        if transcripts_dir is not None:
            os.close(transcripts_dir)
        if destination is not None:
            os.close(destination)
        os.close(parent)


def run(argv: list[str] | None = None) -> int:
    parser = Parser(description="Convert official chat exports to transcripts", epilog=SIGNATURE)
    parser.add_argument("--version", action="version", version=VERSION)
    sub = parser.add_subparsers(dest="command", required=True)
    convert = sub.add_parser("convert", help="convert an official export", epilog=SIGNATURE)
    convert.add_argument("input", type=Path)
    convert.add_argument("--out", type=Path, required=True)
    convert.add_argument("--format", choices=("whatsapp", "discord", "telegram", "slack"))
    convert.add_argument("--day-first", action="store_true", help="interpret ambiguous WhatsApp dates as day/month/year")
    convert.add_argument("--dry-run", action="store_true")
    convert.add_argument("--approve", help="SHA-256 digest from the matching dry run")
    convert.add_argument("--allow-partial", action="store_true")
    args = parser.parse_args(argv)
    require_platform()
    validate_path(args.input)
    validate_path(args.out)
    source = args.input
    format_name = args.format or detect_format(source)
    if format_name in {"discord", "slack"}:
        reject_export_symlinks(source)
    if format_name == "whatsapp":
        if not path_kind(source):
            raise CliError("WhatsApp input must be a .txt or .zip file")
        messages, skipped = parse_whatsapp_path(source, args.day_first)
    elif format_name == "telegram":
        messages, skipped = parse_telegram(source)
    elif format_name == "discord":
        messages, skipped = parse_discord(source)
    else:
        messages, skipped = parse_slack(source)
    if not messages and not skipped:
        raise CliError("no valid messages found")
    files = render_files(messages)
    prepare_output(args.out)
    approved = output_plan(args.out)
    digest = plan_digest(files, args.out, approved)
    print(f"status: {'incomplete' if skipped else 'complete'}")
    print(f"messages: {len(messages)}")
    print(f"skipped: {len(skipped)}")
    for item in skipped:
        print(f"  skipped: {display_text(item)}")
    for name in sorted(files):
        print(f"  file: {display_text(args.out / name)} ({len(files[name])} bytes)")
    if approved["create_destination"]:
        print(f"  create directory: {display_text(args.out)}")
    if any(name.startswith("transcripts/") for name in files):
        print(f"  create directory: {display_text(args.out / 'transcripts')}")
    print(f"approval digest: {digest}")
    if not args.dry_run:
        if args.approve != digest:
            raise CliError("live create requires --approve with the matching dry-run digest")
        if skipped and not args.allow_partial:
            raise CliError("incomplete export; use --allow-partial to write valid records")
        write_files(args.out, files, approved)
    return 2 if skipped and not args.allow_partial else 0


def main() -> int:
    try:
        return run()
    except CliError as exc:
        print(f"error: {display_text(exc)}", file=sys.stderr)
        return 2
    except (OSError, ValueError, csv.Error) as exc:
        print(f"error: {display_text(exc)}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    sys.exit(main())

License: MIT (read it).

Back to the listing