be-my-anchor/tools/fix_multiline_json.py

#!/usr/bin/env python3
"""
Tool: fix_multiline_json.py
Description:
  Fixes JSON files that contain invalid raw multiline strings (e.g. pasted articles
  or multiline text in "content" or "title" fields) without altering any content,
  newlines, spaces, or HTML tags.

Usage:
  python3 tools/fix_multiline_json.py <file1.json> [file2.json ...]
  python3 tools/fix_multiline_json.py test/data/
  python3 tools/fix_multiline_json.py --check <file.json>
"""

import sys
import os
import glob
import json
import re
from typing import List, Dict, Any, Optional


def is_valid_json(text: str) -> bool:
    try:
        json.loads(text)
        return True
    except Exception:
        return False


def parse_multiline_json_array(text: str) -> List[Dict[str, Any]]:
    """
    Parses a loosely formatted JSON array of objects where string fields
    may contain raw unescaped newlines.
    """
    lines = text.splitlines(keepends=True)
    n = len(lines)
    items: List[Dict[str, Any]] = []
    i = 0
    current_obj: Optional[Dict[str, Any]] = None

    while i < n:
        line = lines[i]
        stripped = line.strip()

        # Start of a new object
        if stripped == "{" and current_obj is None:
            current_obj = {}
            i += 1
            continue

        # End of current object
        if stripped in ("},", "}") and current_obj is not None:
            items.append(current_obj)
            current_obj = None
            i += 1
            continue

        # Field detection: look for `"key":` pattern
        if current_obj is not None:
            key_match = re.match(r'^\s*"([^"]+)"\s*:\s*(.*)', line)
            if key_match:
                key = key_match.group(1)
                val_start = key_match.group(2).lstrip()

                # Case 1: String value
                if val_start.startswith('"'):
                    val_after_quote = val_start[1:]
                    # Check if single-line closed: e.g. "value", or "value"
                    # But be careful if it ends with ", or "
                    if val_after_quote.endswith(',\n') or val_after_quote.endswith(',\r\n'):
                        trimmed = val_after_quote.rstrip('\r\n')[:-1].rstrip()
                    else:
                        trimmed = val_after_quote.rstrip('\r\n')

                    if trimmed.endswith('"'):
                        # Single line string
                        current_obj[key] = trimmed[:-1]
                        i += 1
                        continue
                    else:
                        # Multiline string starting on this line
                        val_lines = [val_after_quote]
                        i += 1
                        while i < n:
                            next_line = lines[i]
                            next_stripped = next_line.strip()
                            # Check if next line is closing of object or next key
                            is_obj_end = next_stripped in ("},", "}")
                            is_next_key = bool(re.match(r'^\s*"[^"]+"\s*:\s*', next_line))

                            if is_obj_end or is_next_key:
                                # The previous line had the closing quote
                                last = val_lines[-1]
                                last_stripped = last.rstrip('\r\n')
                                if last_stripped.endswith(','):
                                    last_stripped = last_stripped[:-1].rstrip()
                                if last_stripped.endswith('"'):
                                    val_lines[-1] = last_stripped[:-1]
                                break
                            else:
                                val_lines.append(next_line)
                                i += 1

                        current_obj[key] = "".join(val_lines)
                        continue

                # Case 2: Number, boolean, null
                else:
                    raw_val = val_start.rstrip('\r\n')
                    if raw_val.endswith(','):
                        raw_val = raw_val[:-1].strip()
                    try:
                        current_obj[key] = json.loads(raw_val)
                    except Exception:
                        current_obj[key] = raw_val
                    i += 1
                    continue

        i += 1

    return items


def process_file(filepath: str, check_only: bool = False) -> bool:
    print(f"Checking: {filepath}")
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            raw_text = f.read()
    except Exception as e:
        print(f"  [ERROR] Failed to read {filepath}: {e}")
        return False

    if is_valid_json(raw_text):
        print(f"  [OK] Already valid JSON syntax.")
        return True

    print(f"  [FIXING] Detected multiline or invalid JSON syntax...")
    try:
        parsed_items = parse_multiline_json_array(raw_text)
        if not parsed_items:
            print(f"  [WARNING] Could not parse any objects from {filepath}.")
            return False

        formatted_json = json.dumps(parsed_items, ensure_ascii=False, indent=4) + "\n"

        # Verify round-trip syntax
        test_load = json.loads(formatted_json)
        assert len(test_load) == len(parsed_items)

        if check_only:
            print(f"  [DRY-RUN] Can successfully repair {len(parsed_items)} items.")
            return True

        with open(filepath, "w", encoding="utf-8") as f:
            f.write(formatted_json)

        print(f"  [SUCCESS] Repaired and formatted {len(parsed_items)} items.")
        return True
    except Exception as e:
        print(f"  [ERROR] Failed repairing {filepath}: {e}")
        return False


def main():
    if len(sys.argv) < 2:
        print("Usage: python3 tools/fix_multiline_json.py [--check] <path-to-json-or-directory> [...]")
        sys.exit(1)

    args = sys.argv[1:]
    check_only = False
    if "--check" in args:
        check_only = True
        args.remove("--check")

    files_to_process = []
    for arg in args:
        if os.path.isdir(arg):
            files_to_process.extend(glob.glob(os.path.join(arg, "**/*.json"), recursive=True))
        elif os.path.isfile(arg):
            files_to_process.append(arg)
        else:
            matches = glob.glob(arg)
            if matches:
                files_to_process.extend(matches)
            else:
                print(f"[WARN] No files matched: {arg}")

    if not files_to_process:
        print("No JSON files found to process.")
        sys.exit(0)

    success_count = 0
    for file_path in files_to_process:
        if process_file(file_path, check_only=check_only):
            success_count += 1

    print(f"\nCompleted: {success_count}/{len(files_to_process)} files processed successfully.")


if __name__ == "__main__":
    main()