Converts Google Takeout JSONL Files to JSON

google_takeout_jsonl_converter.py

#!/usr/bin/env python

import argparse
import csv
import io
import json
import os
import sys


def deep_parse(value: object) -> object:
    if isinstance(value, dict):
        return {k: deep_parse(v) for k, v in value.items()}
    if isinstance(value, list):
        return [deep_parse(item) for item in value]
    if isinstance(value, str):
        try:
            parsed = json.loads(value)
        except (json.JSONDecodeError, ValueError):
            return value
        # Strings that decode to a container get recursed into; other JSON
        # primitives (bool, int, float, null, or a de-quoted string) are kept as-is.
        if isinstance(parsed, (dict, list)):
            return deep_parse(parsed)
        return parsed
    return value


def parse_jsonl_line(line: str) -> object:
    line = line.strip()
    if not line:
        return None
    row = next(csv.reader(io.StringIO(line)))
    if not row:
        return None
    # CSV unquoting turns "" -> "; remaining \\" sequences are escaped quotes
    inner = row[0].replace('\\\\"', '\\"')
    return deep_parse(json.loads(inner))


def convert_jsonl_to_json(jsonl_path: str, out_path: str) -> None:
    records = []
    with open(jsonl_path, encoding="utf-8") as f:
        for lineno, line in enumerate(f, 1):
            if not line.strip():
                continue
            try:
                record = parse_jsonl_line(line)
                if record is not None:
                    records.append(record)
            except Exception as e:
                print(f"  Warning: {jsonl_path}:{lineno}: {e}", file=sys.stderr)

    os.makedirs(os.path.dirname(out_path) or ".", exist_ok=True)
    with open(out_path, "w", encoding="utf-8") as f:
        json.dump(records, f, indent=2, ensure_ascii=False)
    print(f"{jsonl_path} -> {out_path} ({len(records)} records)")


def main() -> None:
    parser = argparse.ArgumentParser(description="Convert JSONL files to JSON.")
    parser.add_argument("root", nargs="?", default=".", help="Source directory (default: .)")
    parser.add_argument("-o", "--output", default=None, help="Output directory (default: alongside source files)")
    args = parser.parse_args()

    if not os.path.isdir(args.root):
        print(f"Error: {args.root} is not a directory", file=sys.stderr)
        sys.exit(1)

    root = os.path.normpath(args.root)

    for dirpath, _, filenames in os.walk(root):
        for filename in sorted(filenames):
            if not filename.endswith(".jsonl"):
                continue
            jsonl_path = os.path.join(dirpath, filename)
            stem = os.path.splitext(filename)[0] + ".json"
            if args.output:
                rel = os.path.relpath(dirpath, root)
                out_path = os.path.normpath(os.path.join(args.output, rel, stem))
            else:
                out_path = os.path.join(dirpath, stem)
            convert_jsonl_to_json(jsonl_path, out_path)


if __name__ == "__main__":
    main()