Python으로 CSV를 JSON 변환하기

import json

# Python 딕셔너리로 표현된 단일 CSV 행
server_entry = {
    "hostname": "web-prod-03",
    "ip_address": "10.0.12.47",
    "port": 8080,
    "region": "eu-west-1"
}

# 딕셔너리를 JSON 문자열로 변환
json_string = json.dumps(server_entry)
print(json_string)
# {"hostname": "web-prod-03", "ip_address": "10.0.12.47", "port": 8080, "region": "eu-west-1"}
print(type(json_string))
# <class 'str'>

import json

server_entry = {
    "hostname": "web-prod-03",
    "ip_address": "10.0.12.47",
    "port": 8080,
    "region": "eu-west-1"
}

pretty_json = json.dumps(server_entry, indent=2)
print(pretty_json)
# {
#   "hostname": "web-prod-03",
#   "ip_address": "10.0.12.47",
#   "port": 8080,
#   "region": "eu-west-1"
# }

import json

warehouse_record = {
    "sku": "WH-9031",
    "location": "서울 창고 3호",
    "quantity": 240,
    "last_audit": "2026-03-10"
}

output = json.dumps(warehouse_record, indent=2, sort_keys=True, ensure_ascii=False)
print(output)
# {
#   "last_audit": "2026-03-10",
#   "location": "서울 창고 3호",
#   "quantity": 240,
#   "sku": "WH-9031"
# }

import csv
import json

# 1단계: CSV 행을 딕셔너리 리스트로 읽기
with open("inventory.csv", "r", encoding="utf-8") as csv_file:
    rows = list(csv.DictReader(csv_file))

# 2단계: 리스트를 JSON 파일로 쓰기
with open("inventory.json", "w", encoding="utf-8") as json_file:
    json.dump(rows, json_file, indent=2, ensure_ascii=False)

print(f"Converted {len(rows)} rows to inventory.json")

import csv
import json

with open("sensors.csv", "r", encoding="utf-8") as f:
    rows = list(csv.DictReader(f))

# 파일에 쓰는 대신 JSON을 문자열로 가져오기
json_payload = json.dumps(rows, indent=2)
print(json_payload)
# [
#   {
#     "sensor_id": "TMP-4401",
#     "location": "7번 빌딩 - 2층",
#     "reading": "22.4",
#     "unit": "celsius"
#   },
#   ...
# ]

import csv
import json

def coerce_types(row: dict) -> dict:
    """문자열 값을 적절한 Python 타입으로 변환합니다."""
    return {
        "sensor_id": row["sensor_id"],
        "location": row["location"],
        "temperature": float(row["temperature"]),
        "humidity": float(row["humidity"]),
        "battery_pct": int(row["battery_pct"]),
        "active": row["active"].lower() == "true",
    }

with open("sensor_readings.csv", "r", encoding="utf-8") as f:
    rows = [coerce_types(row) for row in csv.DictReader(f)]

print(json.dumps(rows[0], indent=2))
# {
#   "sensor_id": "TMP-4401",
#   "location": "7번 빌딩 - 2층",
#   "temperature": 22.4,
#   "humidity": 58.3,
#   "battery_pct": 87,
#   "active": true
# }

import json
from datetime import datetime
from decimal import Decimal
from uuid import UUID

def json_serial(obj):
    """비표준 타입을 위한 폴백 직렬화기."""
    if isinstance(obj, datetime):
        return obj.isoformat()
    if isinstance(obj, UUID):
        return str(obj)
    if isinstance(obj, Decimal):
        return float(obj)
    raise TypeError(f"Type {type(obj).__name__} is not JSON serializable")

transaction = {
    "txn_id": UUID("a1b2c3d4-e5f6-7890-abcd-ef1234567890"),
    "amount": Decimal("149.99"),
    "currency": "EUR",
    "processed_at": datetime(2026, 3, 15, 14, 30, 0),
    "gateway": "stripe",
}

print(json.dumps(transaction, indent=2, default=json_serial))
# {
#   "txn_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
#   "amount": 149.99,
#   "currency": "EUR",
#   "processed_at": "2026-03-15T14:30:00",
#   "gateway": "stripe"
# }

import json
from dataclasses import dataclass, asdict
from datetime import datetime

@dataclass
class ShipmentRecord:
    tracking_id: str
    origin: str
    destination: str
    weight_kg: float
    shipped_at: datetime

def json_serial(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    raise TypeError(f"Not serializable: {type(obj).__name__}")

shipment = ShipmentRecord(
    tracking_id="SHP-9827",
    origin="인천",
    destination="서울",
    weight_kg=1240.5,
    shipped_at=datetime(2026, 3, 12, 8, 0, 0),
)

print(json.dumps(asdict(shipment), indent=2, default=json_serial))
# {
#   "tracking_id": "SHP-9827",
#   "origin": "인천",
#   "destination": "서울",
#   "weight_kg": 1240.5,
#   "shipped_at": "2026-03-12T08:00:00"
# }

import csv
import json

# 데이터베이스 내보내기에서 가져온 탭 구분 파일
with open("user_sessions.tsv", "r", encoding="utf-8") as f:
    reader = csv.DictReader(f, delimiter="\t")
    sessions = list(reader)

print(json.dumps(sessions[:2], indent=2))
# [
#   {
#     "session_id": "sess_8f2a91bc",
#     "user_id": "usr_4421",
#     "started_at": "2026-03-15T09:12:00Z",
#     "duration_sec": "342",
#     "pages_viewed": "7"
#   },
#   {
#     "session_id": "sess_3c7d44ef",
#     "user_id": "usr_1187",
#     "started_at": "2026-03-15T09:14:22Z",
#     "duration_sec": "128",
#     "pages_viewed": "3"
#   }
# ]

import csv
import json
import sys

def csv_to_json_file(csv_path: str, json_path: str) -> int:
    """CSV 파일을 JSON으로 변환합니다. 작성된 행 수를 반환합니다."""
    try:
        with open(csv_path, "r", encoding="utf-8") as f:
            rows = list(csv.DictReader(f))
    except FileNotFoundError:
        print(f"Error: {csv_path} not found", file=sys.stderr)
        sys.exit(1)
    except csv.Error as e:
        print(f"CSV parse error in {csv_path}: {e}", file=sys.stderr)
        sys.exit(1)

    with open(json_path, "w", encoding="utf-8") as f:
        json.dump(rows, f, indent=2, ensure_ascii=False)

    return len(rows)

count = csv_to_json_file("fleet_vehicles.csv", "fleet_vehicles.json")
print(f"Wrote {count} records to fleet_vehicles.json")

import csv
import io
import json
import urllib.request

def fetch_csv_as_json(url: str) -> str:
    """URL에서 CSV를 가져와 JSON 문자열로 반환합니다."""
    try:
        with urllib.request.urlopen(url, timeout=10) as resp:
            raw = resp.read().decode("utf-8")
    except urllib.error.URLError as e:
        raise RuntimeError(f"Failed to fetch {url}: {e}")

    reader = csv.DictReader(io.StringIO(raw))
    rows = list(reader)

    if not rows:
        raise ValueError("CSV response was empty or had no data rows")

    return json.dumps(rows, indent=2, ensure_ascii=False)

# 예시: CSV를 반환하는 내보내기 엔드포인트
try:
    result = fetch_csv_as_json("https://reports.internal/api/v2/daily-metrics.csv")
    print(result)
except (RuntimeError, ValueError) as e:
    print(f"Error: {e}")

import json

json_string = json.dumps({"order_id": "ORD-7291", "total": 129.99})

# 다시 파싱하여 유효한 JSON인지 확인
try:
    parsed = json.loads(json_string)
    print(f"Valid JSON with {len(parsed)} keys")
except json.JSONDecodeError as e:
    print(f"Invalid JSON: {e}")
# Valid JSON with 2 keys

python3 -c "
import csv, json, sys
rows = list(csv.DictReader(sys.stdin))
json.dump(rows, sys.stdout, indent=2)
" < inventory.csv > inventory.json

python3 -c "import csv,json,sys; print(json.dumps(list(csv.DictReader(sys.stdin))))" < data.csv | python3 -m json.tool

python3 -c "import csv,json,sys; json.dump(list(csv.DictReader(sys.stdin)),sys.stdout)" < report.csv | jq .

pip install orjson

import csv
import orjson

with open("telemetry_events.csv", "r", encoding="utf-8") as f:
    rows = list(csv.DictReader(f))

# orjson.dumps()는 str이 아닌 bytes를 반환합니다
json_bytes = orjson.dumps(rows, option=orjson.OPT_INDENT_2)

with open("telemetry_events.json", "wb") as f:  # 참고: bytes를 위해 "wb"
    f.write(json_bytes)

print(f"Wrote {len(rows)} events ({len(json_bytes)} bytes)")

pip install rich

import csv
import json
from rich.console import Console
from rich.syntax import Syntax

console = Console()

with open("deployment_log.csv", "r", encoding="utf-8") as f:
    rows = list(csv.DictReader(f))

json_output = json.dumps(rows[:3], indent=2, ensure_ascii=False)
syntax = Syntax(json_output, "json", theme="monokai", line_numbers=True)
console.print(syntax)

import csv
import json

def csv_to_ndjson(csv_path: str, ndjson_path: str) -> int:
    """CSV를 NDJSON으로 변환합니다. 한 번에 한 행씩 처리합니다."""
    count = 0
    with open(csv_path, "r", encoding="utf-8") as infile, \
         open(ndjson_path, "w", encoding="utf-8") as outfile:
        for row in csv.DictReader(infile):
            outfile.write(json.dumps(row, ensure_ascii=False) + "\n")
            count += 1
    return count

rows_written = csv_to_ndjson("access_log.csv", "access_log.ndjson")
print(f"Wrote {rows_written} lines to access_log.ndjson")
# 각 줄은 독립적인 JSON 객체입니다:
# {"timestamp":"2026-03-15T09:12:00Z","method":"GET","path":"/api/v2/orders","status":"200"}
# {"timestamp":"2026-03-15T09:12:01Z","method":"POST","path":"/api/v2/payments","status":"201"}

import ijson  # pip install ijson

def count_high_value_orders(json_path: str, threshold: float) -> int:
    """전체 파일을 로드하지 않고 임계값 이상의 주문을 셉니다."""
    count = 0
    with open(json_path, "rb") as f:
        for item in ijson.items(f, "item"):
            if float(item.get("total", 0)) > threshold:
                count += 1
    return count

# 일정한 메모리 사용량으로 2 GB JSON 파일 처리
high_value = count_high_value_orders("all_orders.json", 500.0)
print(f"Found {high_value} orders above $500")