Python-এ JSON থেকে CSV — DictWriter ও pandas উদাহরণ

import json
import csv

# নমুনা JSON ডেটা — অর্ডার অবজেক্টের একটি অ্যারে
json_string = """
[
  {"order_id": "ord_91a3", "product": "Wireless Keyboard", "quantity": 2, "unit_price": 74.99},
  {"order_id": "ord_b7f2", "product": "USB-C Hub", "quantity": 1, "unit_price": 34.50},
  {"order_id": "ord_c4e8", "product": "Monitor Stand", "quantity": 3, "unit_price": 29.95}
]
"""

records = json.loads(json_string)

with open("orders.csv", "w", newline="", encoding="utf-8") as csvfile:
    writer = csv.DictWriter(csvfile, fieldnames=records[0].keys())
    writer.writeheader()
    writer.writerows(records)

# orders.csv:
# order_id,product,quantity,unit_price
# ord_91a3,Wireless Keyboard,2,74.99
# ord_b7f2,USB-C Hub,1,34.50
# ord_c4e8,Monitor Stand,3,29.95

import json
import csv

with open("server_metrics.json", encoding="utf-8") as jf:
    metrics = json.load(jf)  # ফাইল অবজেক্টের জন্য json.load()

# স্পষ্ট fieldnames কলামের ক্রম নিয়ন্ত্রণ করে
columns = ["timestamp", "hostname", "cpu_percent", "memory_mb", "disk_io_ops"]

with open("server_metrics.csv", "w", newline="", encoding="utf-8") as cf:
    writer = csv.DictWriter(cf, fieldnames=columns, extrasaction="ignore")
    writer.writeheader()
    writer.writerows(metrics)

# শুধুমাত্র পাঁচটি নির্দিষ্ট কলাম ঠিক সেই ক্রমে আসবে

import json
import csv
from datetime import datetime, timezone
from decimal import Decimal
from uuid import UUID

# Python টাইপ সহ পার্স করা API রেসপন্সের সিমুলেশন
transactions = [
    {
        "txn_id": UUID("a1b2c3d4-e5f6-7890-abcd-ef1234567890"),
        "created_at": datetime(2026, 3, 15, 9, 30, 0, tzinfo=timezone.utc),
        "amount": Decimal("1249.99"),
        "currency": "USD",
        "merchant": "CloudHost Inc.",
    },
    {
        "txn_id": UUID("b2c3d4e5-f6a7-8901-bcde-f12345678901"),
        "created_at": datetime(2026, 3, 15, 14, 12, 0, tzinfo=timezone.utc),
        "amount": Decimal("87.50"),
        "currency": "EUR",
        "merchant": "DataSync GmbH",
    },
]

def prepare_row(record: dict) -> dict:
    """অ-স্ট্রিং টাইপগুলো CSV-বান্ধব স্ট্রিংয়ে রূপান্তর করুন।"""
    return {
        "txn_id": str(record["txn_id"]),
        "created_at": record["created_at"].isoformat(),
        "amount": f"{record['amount']:.2f}",
        "currency": record["currency"],
        "merchant": record["merchant"],
    }

with open("transactions.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=["txn_id", "created_at", "amount", "currency", "merchant"])
    writer.writeheader()
    for txn in transactions:
        writer.writerow(prepare_row(txn))

# transactions.csv:
# txn_id,created_at,amount,currency,merchant
# a1b2c3d4-e5f6-7890-abcd-ef1234567890,2026-03-15T09:30:00+00:00,1249.99,USD,CloudHost Inc.
# b2c3d4e5-f6a7-8901-bcde-f12345678901,2026-03-15T14:12:00+00:00,87.50,EUR,DataSync GmbH

import pandas as pd

# JSON অ্যারে সরাসরি DataFrame-এ পড়ুন
df = pd.read_json("warehouse_inventory.json")

# CSV-তে লিখুন — index=False স্বয়ংক্রিয়-তৈরি সারি নম্বর প্রতিরোধ করে
df.to_csv("warehouse_inventory.csv", index=False)

# ব্যস। দুই লাইন। pandas স্বয়ংক্রিয়ভাবে কলামের টাইপ অনুমান করে।

import json
import pandas as pd

api_response = """
[
  {
    "order_id": "ord_91a3",
    "placed_at": "2026-03-15T09:30:00Z",
    "customer": {
      "name": "Sarah Chen",
      "email": "s.chen@example.com",
      "tier": "premium"
    },
    "shipping": {
      "method": "express",
      "address": {
        "city": "Portland",
        "state": "OR",
        "zip": "97201"
      }
    },
    "total": 299.95
  },
  {
    "order_id": "ord_b7f2",
    "placed_at": "2026-03-15T14:12:00Z",
    "customer": {
      "name": "James Park",
      "email": "j.park@example.com",
      "tier": "standard"
    },
    "shipping": {
      "method": "standard",
      "address": {
        "city": "Austin",
        "state": "TX",
        "zip": "73301"
      }
    },
    "total": 87.50
  }
]
"""

orders = json.loads(api_response)

# json_normalize নেস্টেড dict ফ্ল্যাটেন করে — sep ডিলিমিটার নিয়ন্ত্রণ করে
df = pd.json_normalize(orders, sep="_")
df.to_csv("flat_orders.csv", index=False)

# ফলাফলের কলামসমূহ:
# order_id, placed_at, customer_name, customer_email, customer_tier,
# shipping_method, shipping_address_city, shipping_address_state,
# shipping_address_zip, total

import pandas as pd

df = pd.read_json("telemetry_events.json")

# স্পষ্ট এনকোডিং ও অনুপস্থিত মান হ্যান্ডেলিং সহ TSV আউটপুট
df.to_csv(
    "telemetry_events.tsv",
    sep="\t",
    index=False,
    encoding="utf-8",
    na_rep="NULL",
    columns=["event_id", "timestamp", "source", "severity", "message"],
)

# শেল স্ক্রিপ্টে পাইপিংয়ের জন্য stdout-এ লিখুন
print(df.to_csv(index=False))

# স্ট্রিং হিসেবে ফেরত নিন (কোনো ফাইল লেখা হয় না)
csv_string = df.to_csv(index=False)
print(len(csv_string), "অক্ষর")

import json
import csv
import sys

def json_file_to_csv(input_path: str, output_path: str) -> int:
    """অবজেক্টের অ্যারে ধারণকারী JSON ফাইল CSV-তে রূপান্তর করুন।
    লেখা সারির সংখ্যা ফেরত দেয়।
    """
    try:
        with open(input_path, encoding="utf-8") as jf:
            data = json.load(jf)
    except FileNotFoundError:
        print(f"Error: {input_path} not found", file=sys.stderr)
        return 0
    except json.JSONDecodeError as exc:
        print(f"Error: invalid JSON in {input_path}: {exc.msg} at line {exc.lineno}", file=sys.stderr)
        return 0

    if not isinstance(data, list) or not data:
        print(f"Error: expected a non-empty JSON array in {input_path}", file=sys.stderr)
        return 0

    # সমস্ত রেকর্ড জুড়ে অনন্য key সংগ্রহ করুন — অসামঞ্জস্যপূর্ণ স্কিমা হ্যান্ডেল করে
    all_keys: list[str] = []
    seen: set[str] = set()
    for record in data:
        for key in record:
            if key not in seen:
                all_keys.append(key)
                seen.add(key)

    with open(output_path, "w", newline="", encoding="utf-8") as cf:
        writer = csv.DictWriter(cf, fieldnames=all_keys, restval="", extrasaction="ignore")
        writer.writeheader()
        writer.writerows(data)

    return len(data)

rows = json_file_to_csv("deploy_logs.json", "deploy_logs.csv")
print(f"Wrote {rows} rows to deploy_logs.csv")

import json
import csv
import urllib.request
import urllib.error

def api_response_to_csv(url: str, output_path: str) -> int:
    """REST API এন্ডপয়েন্ট থেকে JSON ফেচ করে CSV হিসেবে লিখুন।"""
    try:
        req = urllib.request.Request(url, headers={"Accept": "application/json"})
        with urllib.request.urlopen(req, timeout=30) as resp:
            if resp.status != 200:
                print(f"Error: API returned status {resp.status}")
                return 0
            body = resp.read().decode("utf-8")
    except urllib.error.URLError as exc:
        print(f"Error: could not reach {url}: {exc.reason}")
        return 0

    try:
        records = json.loads(body)
    except json.JSONDecodeError as exc:
        print(f"Error: API returned invalid JSON: {exc.msg}")
        return 0

    if not isinstance(records, list) or not records:
        print("Error: expected a non-empty JSON array from the API")
        return 0

    with open(output_path, "w", newline="", encoding="utf-8") as cf:
        writer = csv.DictWriter(cf, fieldnames=records[0].keys())
        writer.writeheader()
        writer.writerows(records)

    return len(records)

rows = api_response_to_csv(
    "https://api.internal.example.com/v2/deployments?status=completed",
    "completed_deployments.csv",
)
print(f"Exported {rows} deployments to CSV")

# Python one-liner: stdin থেকে JSON পড়ে, stdout-এ CSV লেখে
cat orders.json | python3 -c "
import json, csv, sys
data = json.load(sys.stdin)
w = csv.DictWriter(sys.stdout, fieldnames=data[0].keys())
w.writeheader()
w.writerows(data)
"

# ফাইলে আউটপুট সংরক্ষণ করুন
cat orders.json | python3 -c "
import json, csv, sys
data = json.load(sys.stdin)
w = csv.DictWriter(sys.stdout, fieldnames=data[0].keys())
w.writeheader()
w.writerows(data)
" > orders.csv

# json2csv.py হিসেবে সংরক্ষণ করুন এবং চালান: python3 json2csv.py input.json -o output.csv
python3 -c "
import json, csv, argparse, sys

parser = argparse.ArgumentParser(description='Convert JSON array to CSV')
parser.add_argument('input', help='Path to JSON file')
parser.add_argument('-o', '--output', default=None, help='Output CSV path (default: stdout)')
parser.add_argument('-d', '--delimiter', default=',', help='CSV delimiter')
args = parser.parse_args()

with open(args.input) as f:
    data = json.load(f)

out = open(args.output, 'w', newline='') if args.output else sys.stdout
writer = csv.DictWriter(out, fieldnames=data[0].keys(), delimiter=args.delimiter)
writer.writeheader()
writer.writerows(data)
if args.output:
    out.close()
    print(f'Wrote {len(data)} rows to {args.output}', file=sys.stderr)
" "$@"

# csvkit ইনস্টল করুন: pip install csvkit

# jq ফিল্ড নির্বাচন ও ফ্ল্যাটেন করে, in2csv CSV ফরম্যাটিং হ্যান্ডেল করে
cat api_response.json | jq '[.[] | {id: .order_id, customer: .customer.name, total}]' | in2csv -f json > orders.csv

# Miller (mlr) JSON-থেকে-CSV-র আরেকটি বিকল্প
mlr --json2csv cat orders.json > orders.csv

pip install pyarrow

import pandas as pd

# pyarrow ইঞ্জিন দিয়ে JSON পড়ুন (বড় ফাইলে দ্রুত পার্সিং)
df = pd.read_json("sensor_readings.json", engine="pyarrow")

# to_csv-এর engine প্যারামিটার নেই, কিন্তু পড়া ও লেখার মাঝে
# DataFrame অপারেশনগুলো pyarrow-এর কলামার লেআউট থেকে উপকৃত হয়
df.to_csv("sensor_readings.csv", index=False)

# সত্যিকারের বড় এক্সপোর্টের জন্য, CSV-র পরিবর্তে Parquet-এ লেখার কথা ভাবুন
# — বাইনারি ফরম্যাট, ৫-১০ গুণ ছোট, টাইপ সংরক্ষণ করে
df.to_parquet("sensor_readings.parquet", engine="pyarrow")

pip install rich

import json
from rich.console import Console
from rich.table import Table

json_string = """
[
  {"hostname": "web-prod-1", "cpu_percent": 72.3, "memory_mb": 3840, "uptime_hours": 720},
  {"hostname": "web-prod-2", "cpu_percent": 45.1, "memory_mb": 2560, "uptime_hours": 168},
  {"hostname": "db-replica-1", "cpu_percent": 91.7, "memory_mb": 7680, "uptime_hours": 2160}
]
"""

records = json.loads(json_string)
console = Console()

table = Table(title="Server Metrics Preview", show_lines=True)
for key in records[0]:
    table.add_column(key, style="cyan" if key == "hostname" else "white")

for row in records:
    table.add_row(*[str(v) for v in row.values()])

console.print(table)
# টার্মিনালে বর্ডার সহ রঙিন হাইলাইটেড টেবিল রেন্ডার করে

pip install ijson

import ijson
import csv

def stream_json_to_csv(json_path: str, csv_path: str) -> int:
    """সম্পূর্ণ মেমরিতে লোড না করে বড় JSON অ্যারে CSV-তে রূপান্তর করুন।"""
    with open(json_path, "rb") as jf, open(csv_path, "w", newline="", encoding="utf-8") as cf:
        # ijson.items শীর্ষ-স্তরের অ্যারের প্রতিটি উপাদান একটি করে yield করে
        records = ijson.items(jf, "item")

        first_record = next(records)
        fieldnames = list(first_record.keys())

        writer = csv.DictWriter(cf, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerow(first_record)

        count = 1
        for record in records:
            writer.writerow(record)
            count += 1

    return count

rows = stream_json_to_csv("clickstream_2026_03.json", "clickstream_2026_03.csv")
print(f"Streamed {rows} records to CSV")

import json
import csv

def ndjson_to_csv(ndjson_path: str, csv_path: str) -> int:
    """নিউলাইন-ডিলিমিটেড JSON ফাইল একটি লাইন করে CSV-তে রূপান্তর করুন।"""
    with open(ndjson_path, encoding="utf-8") as nf:
        first_line = nf.readline()
        first_record = json.loads(first_line)
        fieldnames = list(first_record.keys())

        with open(csv_path, "w", newline="", encoding="utf-8") as cf:
            writer = csv.DictWriter(cf, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerow(first_record)

            count = 1
            for line in nf:
                line = line.strip()
                if not line:
                    continue
                try:
                    record = json.loads(line)
                    writer.writerow(record)
                    count += 1
                except json.JSONDecodeError:
                    continue  # ত্রুটিপূর্ণ লাইন এড়িয়ে যান

    return count

rows = ndjson_to_csv("access_log.ndjson", "access_log.csv")
print(f"Converted {rows} log entries to CSV")