json-transformer

JSON Transformer スキル

JSON データ構造を高度な操作で変換、操作、分析します。

指示

あなたは JSON 変換のエキスパートです。呼び出されたときは次のようにします。

1. JSON の解析と検証:

   • ファイル、文字列、API から JSON を解析する
   • JSON の構造とスキーマを検証する
   • 不正な形式の JSON を適切に処理する
   • JSON を整形して出力する
   • 一般的な JSON の問題を検出して修正する

2. データ構造の変換:

   • ネストされたオブジェクトや配列を再形成する
   • 構造をフラット化およびアンフラット化する
   • 特定のパスを抽出する (JSONPath、JMESPath)
   • JSON ドキュメントをマージおよび結合する
   • データをフィルタリングおよびマッピングする

3. 高度な操作:

   • JSON と他の形式 (CSV、YAML、XML) の間で変換する
   • 変換を適用する (jq スタイルの操作)
   • JSON データをクエリおよび検索する
   • JSON ドキュメントを差分比較する
   • スキーマから JSON を生成する

4. データ操作:

   • プロパティの追加、更新、削除
   • キーの名前変更
   • データ型の変換
   • 並べ替えと重複排除
   • 集計値の計算

使用例

@json-transformer data.json
@json-transformer --flatten
@json-transformer --path "users[*].email"
@json-transformer --merge file1.json file2.json
@json-transformer --to-csv data.json
@json-transformer --validate schema.json

基本的な JSON 操作

解析と書き込み

Python

import json

# Parse JSON string
data = json.loads('{"name": "John", "age": 30}')

# Parse from file
with open('data.json', 'r') as f:
    data = json.load(f)

# Write JSON to file
with open('output.json', 'w') as f:
    json.dump(data, f, indent=2)

# Pretty print
print(json.dumps(data, indent=2, sort_keys=True))

# Compact output
compact = json.dumps(data, separators=(',', ':'))

# Handle special types
from datetime import datetime
import decimal

def json_encoder(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    if isinstance(obj, decimal.Decimal):
        return float(obj)
    raise TypeError(f"Type {type(obj)} not serializable")

json.dumps(data, default=json_encoder)

JavaScript

// Parse JSON string
const data = JSON.parse('{"name": "John", "age": 30}');

// Parse from file (Node.js)
const fs = require('fs');
const data = JSON.parse(fs.readFileSync('data.json', 'utf8'));

// Write JSON to file
fs.writeFileSync('output.json', JSON.stringify(data, null, 2));

// Pretty print
console.log(JSON.stringify(data, null, 2));

// Custom serialization
const json = JSON.stringify(data, (key, value) => {
  if (value instanceof Date) {
    return value.toISOString();
  }
  return value;
}, 2);

jq (コマンドライン)

# Pretty print
cat data.json | jq '.'

# Compact output
cat data.json | jq -c '.'

# Sort keys
cat data.json | jq -S '.'

# Read from file, write to file
jq '.' input.json > output.json

検証

Python (jsonschema)

from jsonschema import validate, ValidationError

# Define schema
schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "number", "minimum": 0},
        "email": {"type": "string", "format": "email"}
    },
    "required": ["name", "email"]
}

# Validate data
data = {"name": "John", "email": "[email protected]", "age": 30}

try:
    validate(instance=data, schema=schema)
    print("Valid JSON")
except ValidationError as e:
    print(f"Invalid: {e.message}")

# Validate against JSON Schema draft
from jsonschema import Draft7Validator

validator = Draft7Validator(schema)
errors = list(validator.iter_errors(data))
for error in errors:
    print(f"Error at {'.'.join(str(p) for p in error.path)}: {error.message}")

JavaScript (ajv)

const Ajv = require('ajv');
const ajv = new Ajv();

const schema = {
  type: 'object',
  properties: {
    name: { type: 'string' },
    age: { type: 'number', minimum: 0 },
    email: { type: 'string', format: 'email' }
  },
  required: ['name', 'email']
};

const validate = ajv.compile(schema);

const data = { name: 'John', email: '[email protected]', age: 30 };

if (validate(data)) {
  console.log('Valid JSON');
} else {
  console.log('Invalid:', validate.errors);
}

データ抽出とクエリ

JSONPath クエリ

Python (jsonpath-ng)

from jsonpath_ng import jsonpath, parse

data = {
    "users": [
        {"name": "John", "age": 30, "email": "[email protected]"},
        {"name": "Jane", "age": 25, "email": "[email protected]"}
    ]
}

# Extract all user names
jsonpath_expr = parse('users[*].name')
names = [match.value for match in jsonpath_expr.find(data)]
# Result: ['John', 'Jane']

# Extract emails of users over 25
jsonpath_expr = parse('users[?(@.age > 25)].email')
emails = [match.value for match in jsonpath_expr.find(data)]

# Nested extraction
data = {
    "company": {
        "departments": [
            {
                "name": "Engineering",
                "employees": [
                    {"name": "Alice", "salary": 100000},
                    {"name": "Bob", "salary": 90000}
                ]
            }
        ]
    }
}

jsonpath_expr = parse('company.departments[*].employees[*].name')
names = [match.value for match in jsonpath_expr.find(data)]

jq

# Extract field
echo '{"name": "John", "age": 30}' | jq '.name'

# Extract from array
echo '[{"name": "John"}, {"name": "Jane"}]' | jq '.[].name'

# Filter array
echo '[{"name": "John", "age": 30}, {"name": "Jane", "age": 25}]' | \
  jq '.[] | select(.age > 25)'

# Extract nested fields
cat data.json | jq '.users[].email'

# Multiple fields
cat data.json | jq '.users[] | {name: .name, email: .email}'

# Conditional extraction
cat data.json | jq '.users[] | select(.age > 25) | .email'

JMESPath クエリ

Python (jmespath)

import jmespath

data = {
    "users": [
        {"name": "John", "age": 30, "tags": ["admin", "developer"]},
        {"name": "Jane", "age": 25, "tags": ["developer"]},
        {"name": "Bob", "age": 35, "tags": ["manager"]}
    ]
}

# Simple extraction
names = jmespath.search('users[*].name', data)
# Result: ['John', 'Jane', 'Bob']

# Filtering
admins = jmespath.search('users[?contains(tags, `admin`)]', data)

# Multiple conditions
senior_devs = jmespath.search(
    'users[?age > `28` && contains(tags, `developer`)]',
    data
)

# Projections
result = jmespath.search('users[*].{name: name, age: age}', data)

# Nested queries
data = {
    "departments": [
        {
            "name": "Engineering",
            "employees": [
                {"name": "Alice", "skills": ["Python", "Go"]},
                {"name": "Bob", "skills": ["JavaScript", "Python"]}
            ]
        }
    ]
}

python_devs = jmespath.search(
    'departments[*].employees[?contains(skills, `Python`)].name',
    data
)

データ変換

ネストされた JSON のフラット化

Python

def flatten_json(nested_json, parent_key='', sep='.'):
    """
    Flatten nested JSON structure
    """
    items = []

    for key, value in nested_json.items():
        new_key = f"{parent_key}{sep}{key}" if parent_key else key

        if isinstance(value, dict):
            items.extend(flatten_json(value, new_key, sep=sep).items())
        elif isinstance(value, list):
            for i, item in enumerate(value):
                if isinstance(item, dict):
                    items.extend(flatten_json(item, f"{new_key}[{i}]", sep=sep).items())
                else:
                    items.append((f"{new_key}[{i}]", item))
        else:
            items.append((new_key, value))

    return dict(items)

# Example
nested = {
    "user": {
        "name": "John",
        "address": {
            "city": "New York",
            "zip": "10001"
        },
        "tags": ["admin", "developer"]
    }
}

flat = flatten_json(nested)
# Result: {
#     'user.name': 'John',
#     'user.address.city': 'New York',
#     'user.address.zip': '10001',
#     'user.tags[0]': 'admin',
#     'user.tags[1]': 'developer'
# }

JavaScript

function flattenJSON(obj, prefix = '', result = {}) {
  for (const [key, value] of Object.entries(obj)) {
    const newKey = prefix ? `${prefix}.${key}` : key;

    if (value && typeof value === 'object' && !Array.isArray(value)) {
      flattenJSON(value, newKey, result);
    } else if (Array.isArray(value)) {
      value.forEach((item, index) => {
        if (typeof item === 'object') {
          flattenJSON(item, `${newKey}[${index}]`, result);
        } else {
          result[`${newKey}[${index}]`] = item;
        }
      });
    } else {
      result[newKey] = value;
    }
  }

  return result;
}

JSON のアンフラット化

def unflatten_json(flat_json, sep='.'):
    """
    Unflatten a flattened JSON structure
    """
    result = {}

    for key, value in flat_json.items():
        parts = key.split(sep)
        current = result

        for i, part in enumerate(parts[:-1]):
            # Handle array notation
            if '[' in part:
                array_key, index = part.split('[')
                index = int(index.rstrip(']'))

                if array_key not in current:
                    current[array_key] = []

                # Extend array if needed
                while len(current[array_key]) <= index:
                    current[array_key].append({})

                current = current[array_key][index]
            else:
                if part not in current:
                    current[part] = {}
                current = current[part]

        # Set the final value
        final_key = parts[-1]
        if '[' in final_key:
            array_key, index = final_key.split('[')
            index = int(index.rstrip(']'))

            if array_key not in current:
                current[array_key] = []

            while len(current[array_key]) <= index:
                current[array_key].append(None)

            current[array_key][index] = value
        else:
            current[final_key] = value

    return result

JSON のマージ

Python

def deep_merge(dict1, dict2):
    """
    Deep merge two dictionaries
    """
    result = dict1.copy()

    for key, value in dict2.items():
        if key in result and isinstance(result[key], dict) and isinstance(value, dict):
            result[key] = deep_merge(result[key], value)
        else:
            result[key] = value

    return result

# Example
base = {
    "user": {"name": "John", "age": 30},
    "settings": {"theme": "dark"}
}

override = {
    "user": {"age": 31, "email": "[email protected]"},
    "settings": {"language": "en"}
}

merged = deep_merge(base, override)
# Result: {
#     'user': {'name': 'John', 'age': 31, 'email': '[email protected]'},
#     'settings': {'theme': 'dark', 'language': 'en'}
# }

jq

# Merge two JSON files
jq -s '.[0] * .[1]' file1.json file2.json

# Deep merge
jq -s 'reduce .[] as $item ({}; . * $item)' file1.json file2.json

キーの変換

def transform_keys(obj, transform_fn):
    """
    Transform all keys in JSON structure
    """
    if isinstance(obj, dict):
        return {transform_fn(k): transform_keys(v, transform_fn) for k, v in obj.items()}
    elif isinstance(obj, list):
        return [transform_keys(item, transform_fn) for item in obj]
    else:
        return obj

# Convert to snake_case
import re

def to_snake_case(text):
    s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', text)
    return re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1).lower()

data = {
    "firstName": "John",
    "lastName": "Doe",
    "userInfo": {
        "emailAddress": "[email protected]"
    }
}

snake_case_data = transform_keys(data, to_snake_case)
# Result: {
#     'first_name': 'John',
#     'last_name': 'Doe',
#     'user_info': {'email_address': '[email protected]'}
# }

# Convert to camelCase
def to_camel_case(text):
    components = text.split('_')
    return components[0] + ''.join(x.title() for x in components[1:])

形式の変換

JSON から CSV

Python

import json
import csv
import pandas as pd

# Using pandas (recommended)
data = [
    {"name": "John", "age": 30, "email": "[email protected]"},
    {"name": "Jane", "age": 25, "email": "[email protected]"}
]

df = pd.DataFrame(data)
df.to_csv('output.csv', index=False)

# Using csv module
with open('output.csv', 'w', newline='') as csvfile:
    if data:
        writer = csv.DictWriter(csvfile, fieldnames=data[0].keys())
        writer.writeheader()
        writer.writerows(data)

# Handle nested JSON
def flatten_for_csv(data):
    """Flatten nested JSON for CSV export"""
    if isinstance(data, list):
        return [flatten_json(item) for item in data]
    return flatten_json(data)

flattened = flatten_for_csv(data)
pd.DataFrame(flattened).to_csv('output.csv', index=False)

jq

# Convert JSON array to CSV
cat data.json | jq -r '.[] | [.name, .age, .email] | @csv'

# With headers
cat data.json | jq -r '["name", "age", "email"], (.[] | [.name, .age, .email]) | @csv'

JSON から YAML

Python

import json
import yaml

# JSON to YAML
with open('data.json', 'r') as json_file:
    data = json.load(json_file)

with open('data.yaml', 'w') as yaml_file:
    yaml.dump(data, yaml_file, default_flow_style=False)

# YAML to JSON
with open('data.yaml', 'r') as yaml_file:
    data = yaml.safe_load(yaml_file)

with open('data.json', 'w') as json_file:
    json.dump(data, json_file, indent=2)

JSON から XML

Python

import json
import xml.etree.ElementTree as ET

def json_to_xml(json_obj, root_name='root'):
    """Convert JSON to XML"""

    def build_xml(parent, obj):
        if isinstance(obj, dict):
            for key, val in obj.items():
                elem = ET.SubElement(parent, key)
                build_xml(elem, val)
        elif isinstance(obj, list):
            for item in obj:
                elem = ET.SubElement(parent, 'item')
                build_xml(elem, item)
        else:
            parent.text = str(obj)

    root = ET.Element(root_name)
    build_xml(root, json_obj)

    return ET.tostring(root, encoding='unicode')

# Example
data = {"user": {"name": "John", "age": 30}}
xml_string = json_to_xml(data)

高度な変換

jq スタイルの変換

Python (pyjq)

import pyjq

data = {
    "users": [
        {"name": "John", "age": 30, "city": "New York"},
        {"name": "Jane", "age": 25, "city": "San Francisco"},
        {"name": "Bob", "age": 35, "city": "New York"}
    ]
}

# Select and transform
result = pyjq.all('.users[] | {name, age}', data)

# Filter and group
result = pyjq.all('group_by(.city) | map({city: .[0].city, count: length})', data)

# Complex transformation
result = pyjq.all('''
    .users
    | map(select(.age > 25))
    | sort_by(.age)
    | reverse
''', data)

jq の例

# Map over array
echo '[1,2,3,4,5]' | jq 'map(. * 2)'

# Filter and transform
cat users.json | jq '.users | map(select(.age > 25) | {name, email})'

# Group by field
cat data.json | jq 'group_by(.category) | map({category: .[0].category, count: length})'

# Calculate sum
cat orders.json | jq '[.[] | .amount] | add'

# Create new structure
cat users.json | jq '{
  total: length,
  users: [.[] | {name, email}],
  avgAge: ([.[] | .age] | add / length)
}'

# Conditional logic
cat data.json | jq '.[] | if .status == "active" then .name else empty end'

複雑な再構築

def restructure_json(data):
    """
    Example: Transform flat user records into hierarchical structure
    """
    # Input: [
    #   {"userId": 1, "name": "John", "orderId": 101, "product": "A"},
    #   {"userId": 1, "name": "John", "orderId": 102, "product": "B"},
    #   {"userId": 2, "name": "Jane", "orderId": 103, "product": "C"}
    # ]

    # Output: [
    #   {
    #     "userId": 1,
    #     "name": "John",
    #     "orders": [
    #       {"orderId": 101, "product": "A"},
    #       {"orderId": 102, "product": "B"}
    #     ]
    #   },
    #   {
    #     "userId": 2,
    #     "name": "Jane",
    #     "orders": [{"orderId": 103, "product": "C"}]
    #   }
    # ]

    from collections import defaultdict

    users = defaultdict(lambda: {"orders": []})

    for record in data:
        user_id = record["userId"]

        if "name" not in users[user_id]:
            users[user_id]["userId"] = user_id
            users[user_id]["name"] = record["name"]

        users[user_id]["orders"].append({
            "orderId": record["orderId"],
            "product": record["product"]
        })

    return list(users.values())

配列操作

import json

def unique_by_key(array, key):
    """Remove duplicates based on key"""
    seen = set()
    result = []

    for item in array:
        value = item.get(key)
        if value not in seen:
            seen.add(value)
            result.append(item)

    return result

def sort_by_key(array, key, reverse=False):
    """Sort array by key"""
    return sorted(array, key=lambda x: x.get(key, ''), reverse=reverse)

def group_by_key(array, key):
    """Group array elements by key"""
    from collections import defaultdict

    groups = defaultdict(list)
    for item in array:
        groups[item.get(key)].append(item)

    return dict(groups)

# Example usage
users = [
    {"name": "John", "age": 30, "city": "New York"},
    {"name": "Jane", "age": 25, "city": "San Francisco"},
    {"name": "Bob", "age": 35, "city": "New York"},
    {"name": "Alice", "age": 28, "city": "San Francisco"}
]

# Sort by age
sorted_users = sort_by_key(users, 'age')

# Group by city
by_city = group_by_key(users, 'city')

JSON の差分と比較

import json
from deepdiff import DeepDiff

def json_diff(obj1, obj2):
    """Compare two JSON objects and return differences"""
    diff = DeepDiff(obj1, obj2, ignore_order=True)
    return diff

# Example
old = {
    "name": "John",
    "age": 30,
    "addresses": [{"city": "New York"}]
}

new = {
    "name": "John",
    "age": 31,
    "addresses": [{"city": "San Francisco"}]
}

diff = json_diff(old, new)
print(json.dumps(diff, indent=2))

# Manual diff
def simple_diff(obj1, obj2, path=""):
    """Simple diff implementation"""
    diffs = []

    if type(obj1) != type(obj2):
        diffs.append(f"{path}: type changed from {type(obj1)} to {type(obj2)}")
        return diffs

    if isinstance(obj1, dict):
        all_keys = set(obj1.keys()) | set(obj2.keys())

        for key in all_keys:
            new_path = f"{path}.{key}" if path else key

            if key not in obj1:
                diffs.append(f"{new_path}: added")
            elif key not in obj2:
                diffs.append(f"{new_path}: removed")
            elif obj1[key] != obj2[key]:
                diffs.extend(simple_diff(obj1[key], obj2[key], new_path))

    elif isinstance(obj1, list):
        if len(obj1) != len(obj2):
            diffs.append(f"{path}: length changed from {len(obj1)} to {len(obj2)}")

        for i, (item1, item2) in enumerate(zip(obj1, obj2)):
            diffs.extend(simple_diff(item1, item2, f"{path}[{i}]"))

    elif obj1 != obj2:
        diffs.append(f"{path}: changed from {obj1} to {obj2}")

    return diffs

スキーマ生成

def generate_schema(data, name="root"):
    """
    Generate JSON Schema from data
    """
    if isinstance(data, dict):
        properties = {}
        required = []

        for key, value in data.items():
            properties[key] = generate_schema(value, key)
            if value is not None:
                required.append(key)

        schema = {
            "type": "object",
            "properties": properties
        }

        if required:
            schema["required"] = required

        return schema

    elif isinstance(data, list):
        if data:
            return {
                "type": "array",
                "items": generate_schema(data[0], name)
            }
        return {"type": "array"}

    elif isinstance(data, bool):
        return {"type": "boolean"}

    elif isinstance(data, int):
        return {"type": "integer"}

    elif isinstance(data, float):
        return {"type": "number"}

    elif isinstance(data, str):
        return {"type": "string"}

    elif data is None:
        return {"type": "null"}

    return {}

# Example
sample_data = {
    "name": "John",
    "age": 30,
    "email": "[email protected]",
    "active": True,
    "tags": ["developer", "admin"],
    "address": {
        "city": "New York",
        "zip": "10001"
    }
}

schema = generate_schema(sample_data)
print(json.dumps(schema, indent=2))

ユーティリティ関数

色付き整形出力

from pygments import highlight
from pygments.lexers import JsonLexer
from pygments.formatters import TerminalFormatter

def pretty_print_json(data):
    """Print JSON with syntax highlighting"""
    json_str = json.dumps(data, indent=2, sort_keys=True)
    print(highlight(json_str, JsonLexer(), TerminalFormatter()))

デフォルト値による安全なアクセス

def safe_get(data, path, default=None):
    """
    Safely get nested value from JSON
    path: "user.address.city" or ["user", "address", "city"]
    """
    if isinstance(path, str):
        path = path.split('.')

    current = data
    for key in path:
        if isinstance(current, dict):
            current = current.get(key)
        elif isinstance(current, list) and key.isdigit():
            index = int(key)
            current = current[index] if 0 <= index < len(current) else None
        else:
            return default

        if current is None:
            return default

    return current

# Example
data = {"user": {"address": {"city": "New York"}}}
city = safe_get(data, "user.address.city")  # "New York"
country = safe_get(data, "user.address.country", "Unknown")  # "Unknown"

コマンドラインツール

jq の使用

# Format JSON
cat messy.json | jq '.'

# Extract specific fields
cat data.json | jq '.users[] | {name, email}'

# Filter arrays
cat data.json | jq '.[] | select(.age > 30)'

# Transform keys to lowercase
cat data.json | jq 'with_entries(.key |= ascii_downcase)'

# Merge multiple JSON files
jq -s 'add' file1.json file2.json file3.json

# Convert to CSV
cat data.json | jq -r '.[] | [.name, .age, .email] | @csv'

Python の使用 (コマンドライン)

# Pretty print
python -m json.tool input.json

# Compact output
python -c "import json; print(json.dumps(json.load(open('data.json')), separators=(',',':')))"

# Extract field
python -c "import json; data=json.load(open('data.json')); print(data['users'][0]['name'])"

ベストプラクティス

1. 処理の前に必ず JSON を検証する
2. API コントラクトにはスキーマ検証を使用する
3. エラーを適切に処理する (不正な形式の JSON)
4. 適切なライブラリを使用する (jq、jmespath、jsonpath)
5. 変換中はデータ型を保持する
6. 複雑な変換を文書化する
7. スキーマ定義にはバージョン管理を使用する
8. エッジケースで変換をテストする
9. 大きなファイルではメモリ使用量を考慮する
10. 非常に大きな JSON にはストリーミングパーサーを使用する

一般的なパターン

API レスポンスの変換

def transform_api_response(response):
    """Transform API response to application format"""
    return {
        "users": [
            {
                "id": user["userId"],
                "name": f"{user['firstName']} {user['lastName']}",
                "email": user["emailAddress"],
                "active": user["status"] == "active"
            }
            for user in response.get("data", {}).get("users", [])
        ],
        "pagination": {
            "page": response.get("page", 1),
            "total": response.get("totalResults", 0)
        }
    }

設定のマージ

def merge_configs(base_config, user_config):
    """Merge user configuration with base configuration"""
    result = deep_merge(base_config, user_config)

    # Validate required fields
    required = ["database", "api_key"]
    for field in required:
        if field not in result:
            raise ValueError(f"Missing required field: {field}")

    return result

注意事項

• 常にエッジケース (null、空の配列、存在しないキー) を処理してください
• 目的に応じて適切なツールを使用してください (CLI には jq、データサイエンスには pandas)
• 大きな JSON ファイルではパフォーマンスを考慮してください
• 本番環境ではスキーマを検証してください
• 可能な場合は変換を冪等に保ってください
• 期待される JSON 構造を文書化してください
• 型安全性のために TypeScript/JSON Schema を使用してください