ইমপোর্ট ও এক্সপোর্ট: CSV, JSON, Excel আর Parquet

import csv

new_customers = [
    {"customer_id": 9,  "name": "Rumana Begum",    "email": "rumana@example.com", "city": "Barishal", "joined_on": "2026-06-15", "address": "House 12, Road 5, Dhanmondi"},
    {"customer_id": 10, "name": "তানিয়া সুলতানা",    "email": "tania@example.com",  "city": "Dhaka",    "joined_on": "2026-06-18", "address": "Shop \"Tech Corner\", Level 3"},
    {"customer_id": 11, "name": "Arif Hasan",      "email": "nadia@example.com",  "city": "Sylhet",   "joined_on": "2026-06-20", "address": "Zindabazar"},
    {"customer_id": 12, "name": "Jui Akter",       "email": "jui@example.com",    "city": "Khulna",   "joined_on": "2026-13-01", "address": "Sonadanga"},
    {"customer_id": 13, "name": "",                "email": "blank@example.com",  "city": "Dhaka",    "joined_on": "2026-06-22", "address": "Mirpur 10"},
    {"customer_id": 14, "name": "Sabbir Khan",     "email": "sabbir@example.com", "city": "",         "joined_on": "2026-06-25", "address": "Agrabad"},
]
with open("new_customers.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=list(new_customers[0]))
    writer.writeheader()
    writer.writerows(new_customers)

with open("new_customers.csv", encoding="utf-8") as f:
    print(f.read())

customer_id,name,email,city,joined_on,address
9,Rumana Begum,rumana@example.com,Barishal,2026-06-15,"House 12, Road 5, Dhanmondi"
10,তানিয়া সুলতানা,tania@example.com,Dhaka,2026-06-18,"Shop ""Tech Corner"", Level 3"
11,Arif Hasan,nadia@example.com,Sylhet,2026-06-20,Zindabazar
12,Jui Akter,jui@example.com,Khulna,2026-13-01,Sonadanga
13,,blank@example.com,Dhaka,2026-06-22,Mirpur 10
14,Sabbir Khan,sabbir@example.com,,2026-06-25,Agrabad

with open("new_customers.csv", encoding="utf-8") as f:
    line = f.readlines()[1]
print(len(line.split(",")), "pieces with split(',')")
print(len(next(csv.reader([line]))), "fields with csv.reader")

8 pieces with split(',')
6 fields with csv.reader

city = "ঢাকা"
data = city.encode("utf-8")
print(len(city), "characters,", len(data), "bytes")
print(data.decode("cp1252", errors="replace"))         # ভুল এনকোডিংয়ে পড়া

try:
    with open("new_customers.csv", encoding="ascii") as f:
        f.read()
except UnicodeDecodeError as exc:
    print("UnicodeDecodeError:", exc)

4 characters, 12 bytes
ঢাকা
UnicodeDecodeError: 'ascii' codec can't decode byte 0xe0 in position 135: ordinal not in range(128)

import sqlite3
from datetime import date

con = sqlite3.connect("shop.db")
con.execute("PRAGMA foreign_keys = ON")

def problem_with(row):
    """সারিটা কেন অবৈধ তা ফেরত দেয়, ঠিক থাকলে None।"""
    if not row["name"].strip():
        return "name is empty"
    if "@" not in row["email"]:
        return "email is invalid"
    try:
        date.fromisoformat(row["joined_on"])
    except ValueError:
        return "joined_on is not a valid date"
    return None

loaded, quarantined = 0, []
with open("new_customers.csv", newline="", encoding="utf-8") as f, con:
    for row in csv.DictReader(f):
        reason = problem_with(row)
        if reason is None:
            try:
                con.execute(
                    "INSERT INTO customers (customer_id, name, email, city, joined_on) VALUES (?, ?, ?, ?, ?)",
                    (int(row["customer_id"]), row["name"], row["email"], row["city"] or None, row["joined_on"]),
                )
                loaded += 1
                continue
            except sqlite3.IntegrityError as exc:
                reason = f"database refused: {exc}"
        quarantined.append({**row, "reason": reason})

with open("customers_quarantine.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=list(quarantined[0]))
    writer.writeheader()
    writer.writerows(quarantined)

print(loaded, "loaded,", len(quarantined), "quarantined")
for q in quarantined:
    print(q["customer_id"], "-", q["reason"])

3 loaded, 3 quarantined
11 - database refused: UNIQUE constraint failed: customers.email
12 - joined_on is not a valid date
13 - name is empty

import pandas as pd

staging = pd.read_csv("new_customers.csv", dtype=str, keep_default_na=False)
staging.to_sql("staging_customers", con, if_exists="replace", index=False)

with con:
    cur = con.execute("""
        INSERT INTO customers (customer_id, name, email, city, joined_on)
        SELECT CAST(customer_id AS INTEGER), name, email, NULLIF(city, ''), joined_on
        FROM staging_customers s
        WHERE name <> ''
          AND date(joined_on) = joined_on                            -- a real date
          AND NOT EXISTS (SELECT 1 FROM customers c
                          WHERE c.email = s.email OR c.customer_id = CAST(s.customer_id AS INTEGER))
    """)
print(cur.rowcount, "new rows (the earlier import already loaded the valid ones)")
print(pd.read_sql("SELECT customer_id, name, city FROM customers WHERE customer_id >= 9 ORDER BY customer_id", con))

0 new rows (the earlier import already loaded the valid ones)
   customer_id             name      city
0            9     Rumana Begum  Barishal
1           10  তানিয়া সুলতানা     Dhaka
2           14      Sabbir Khan       NaN

import os
import psycopg2

pg = psycopg2.connect(host=os.environ["PGHOST"], port=os.environ["PGPORT"], user=os.environ["PGUSER"],
                      password=os.environ["PGPASSWORD"], dbname=os.environ["PGDATABASE"])
with pg, pg.cursor() as cur:
    cur.execute("CREATE TABLE staging_customers (customer_id text, name text, email text, "
                "city text, joined_on text, address text)")
    with open("new_customers.csv", encoding="utf-8") as f:
        cur.copy_expert("COPY staging_customers FROM STDIN WITH (FORMAT csv, HEADER true)", f)
    cur.execute("SELECT customer_id, name, address FROM staging_customers ORDER BY customer_id::int LIMIT 2")
    for row in cur.fetchall():
        print(row)
pg.close()

('9', 'Rumana Begum', 'House 12, Road 5, Dhanmondi')
('10', 'তানিয়া সুলতানা', 'Shop "Tech Corner", Level 3')

-- MySQL
LOAD DATA LOCAL INFILE 'new_customers.csv'
INTO TABLE staging_customers
CHARACTER SET utf8mb4
FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '"'
LINES TERMINATED BY '\r\n'
IGNORE 1 LINES;

import hashlib
import json

def export_query(con, sql, params, path):
    """কোয়েরির সারিগুলো একটা UTF-8 CSV আর একটা JSON ম্যানিফেস্টে লেখে; ম্যানিফেস্টটা ফেরত দেয়।"""
    cur = con.execute(sql, params)
    rows = 0
    with open(path, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow([d[0] for d in cur.description])
        for row in cur:
            writer.writerow(row)
            rows += 1
    with open(path, "rb") as f:
        digest = hashlib.sha256(f.read()).hexdigest()
    manifest = {"file": path, "rows": rows, "sha256": digest[:16]}
    with open(path + ".manifest.json", "w", encoding="utf-8") as f:
        json.dump(manifest, f)
    return manifest

print(export_query(
    con,
    "SELECT order_id, customer_id, order_date, status FROM orders WHERE order_date >= ? ORDER BY order_id",
    ("2026-05-01",),
    "orders_2026-05.csv",
))

{'file': 'orders_2026-05.csv', 'rows': 4, 'sha256': '5993311f728321ff'}

orders = pd.read_sql("SELECT order_id, customer_id, order_date, status FROM orders ORDER BY order_id", con)
products = pd.read_sql("SELECT product_id, name, price FROM products ORDER BY product_id", con)

with pd.ExcelWriter("shop_report.xlsx") as xl:
    orders.to_excel(xl, sheet_name="orders", index=False)
    products.to_excel(xl, sheet_name="products", index=False)

sheets = pd.read_excel("shop_report.xlsx", sheet_name=None)
print({name: df.shape for name, df in sheets.items()})

{'orders': (14, 4), 'products': (11, 3)}

from openpyxl import load_workbook
from openpyxl.styles import Font

wb = load_workbook("shop_report.xlsx")
ws = wb["products"]
for cell in ws[1]:
    cell.font = Font(bold=True)                    # হেডারের সারি বোল্ড
ws.column_dimensions["B"].width = 30
ws.freeze_panes = "A2"                             # স্ক্রল করলেও হেডার দেখা যাবে
wb.save("shop_report.xlsx")

for row in ws.iter_rows(min_row=1, max_row=3, values_only=True):
    print(row)

('product_id', 'name', 'price')
(1, 'Python Crash Course', 1200)
(2, 'Hands-On Machine Learning', 2500)

api_response = {
    "orders": [
        {"order_id": 6, "customer": {"id": 2, "name": "Tanvir Ahmed"},
         "items": [{"sku": "AI-BOOT", "qty": 1, "price": 15000}]},
        {"order_id": 9, "customer": {"id": 6, "name": "Imran Hossain"},
         "items": [{"sku": "KEYB-MECH", "qty": 1, "price": 4050},
                   {"sku": "MOUSE-WL", "qty": 1, "price": 900}]},
    ]
}
with open("orders_api.json", "w", encoding="utf-8") as f:
    json.dump(api_response, f)

with open("orders_api.json", encoding="utf-8") as f:
    payload = json.load(f)
items = pd.json_normalize(payload["orders"], record_path="items",
                          meta=["order_id", ["customer", "name"]])
print(items)

         sku  qty  price order_id  customer.name
0    AI-BOOT    1  15000        6   Tanvir Ahmed
1  KEYB-MECH    1   4050        9  Imran Hossain
2   MOUSE-WL    1    900        9  Imran Hossain

examples = pd.DataFrame({
    "prompt": ["Where does customer 10 live?", "What is the name of customer 10?"],
    "answer": ["Dhaka", "তানিয়া সুলতানা"],
})
print(json.dumps({"answer": "তানিয়া"}))                         # ডিফল্ট: এস্কেপ করা
print(json.dumps({"answer": "তানিয়া"}, ensure_ascii=False))    # পড়া যায় এমন UTF-8

examples.to_json("examples.jsonl", orient="records", lines=True, force_ascii=False)
with open("examples.jsonl", encoding="utf-8") as f:
    print(f.read().strip())
print(pd.read_json("examples.jsonl", lines=True).shape)

{"answer": "\u09a4\u09be\u09a8\u09bf\u09af\u09bc\u09be"}
{"answer": "তানিয়া"}
{"prompt":"Where does customer 10 live?","answer":"Dhaka"}
{"prompt":"What is the name of customer 10?","answer":"তানিয়া সুলতানা"}
(2, 2)

import xml.etree.ElementTree as ET

xml_text = """<?xml version="1.0" encoding="UTF-8"?>
<price_feed supplier="Gadget House">
  <item sku="KEYB-MECH"><name>Mechanical Keyboard</name><price>4300</price></item>
  <item sku="MOUSE-WL"><name>Wireless Mouse</name><price>850</price></item>
</price_feed>"""
with open("price_feed.xml", "w", encoding="utf-8") as f:
    f.write(xml_text)

root = ET.parse("price_feed.xml").getroot()
print(root.get("supplier"))
for item in root.iter("item"):
    print(item.get("sku"), item.findtext("name"), int(item.findtext("price")))

print(pd.read_xml("price_feed.xml", xpath=".//item", parser="etree"))

Gadget House
KEYB-MECH Mechanical Keyboard 4300
MOUSE-WL Wireless Mouse 850
         sku                 name  price
0  KEYB-MECH  Mechanical Keyboard   4300
1   MOUSE-WL       Wireless Mouse    850

সারিভিত্তিক ফরম্যাট (CSV)              কলামভিত্তিক ফরম্যাট (Parquet)
1,2026-01-12,chat,312                  event_id:   1, 2, 3, ...        (int64)
2,2026-01-12,view,45                   day:        2026-01-12, ...     (date)
3,2026-01-13,chat,280                  kind:       chat, view, ...     (আলাদা মান কম: ভালো কম্প্রেস হয়)
...                                    seconds:    312, 45, 280, ...   (int64)

import numpy as np

rng = np.random.default_rng(0)
n = 200_000
events = pd.DataFrame({
    "event_id": np.arange(1, n + 1),
    "day": pd.to_datetime("2026-01-01") + pd.to_timedelta(rng.integers(0, 180, n), unit="D"),
    "kind": rng.choice(["view", "search", "add_to_cart", "chat"], n),
    "seconds": rng.integers(1, 600, n),
})
events.to_csv("events.csv", index=False)
events.to_parquet("events.parquet", index=False)

for path in ["events.csv", "events.parquet"]:
    print(f"{path:15} {os.path.getsize(path) / 1_000_000:5.1f} MB")

back_csv = pd.read_csv("events.csv")
back_parquet = pd.read_parquet("events.parquet", columns=["day", "seconds"])   # শুধু দুটো কলাম পড়া
print("day from CSV:", back_csv["day"].dtype, "| day from Parquet:", back_parquet["day"].dtype)

events.csv        5.7 MB
events.parquet    1.6 MB
day from CSV: str | day from Parquet: datetime64[us]

totals = {}
for chunk in pd.read_csv("events.csv", usecols=["kind", "seconds"], chunksize=50_000):
    for kind, s in chunk.groupby("kind")["seconds"].sum().items():
        totals[kind] = totals.get(kind, 0) + int(s)
print(totals)

def stream_rows(path):
    """প্রতিটা CSV সারির জন্য একটা dict দেয়: ফাইল যত বড়ই হোক, মেমরি একই।"""
    with open(path, newline="", encoding="utf-8") as f:
        yield from csv.DictReader(f)

long_chats = sum(1 for row in stream_rows("events.csv") if row["kind"] == "chat" and int(row["seconds"]) > 500)
print(long_chats, "chats longer than 500 seconds")

{'add_to_cart': 14949553, 'chat': 15001659, 'search': 15148999, 'view': 14898307}
8295 chats longer than 500 seconds

# ১. সহজ
pd.read_sql("SELECT * FROM products ORDER BY product_id", con).to_csv(
    "products_for_excel.csv", index=False, encoding="utf-8-sig")
with open("products_for_excel.csv", "rb") as f:
    print(f.read(3))

# ২. মাঝারি
lines = pd.read_sql("""
    SELECT o.order_id, o.order_date, oi.product_id, oi.quantity, oi.unit_price
    FROM orders o JOIN order_items oi ON oi.order_id = o.order_id
    WHERE o.status = 'delivered' ORDER BY o.order_id, oi.product_id""", con)
nested = [
    {"order_id": int(oid), "order_date": g["order_date"].iloc[0],
     "items": g[["product_id", "quantity", "unit_price"]].to_dict("records")}
    for oid, g in lines.groupby("order_id")
]
with open("delivered.json", "w", encoding="utf-8") as f:
    json.dump(nested, f, default=int)
with open("delivered.json", encoding="utf-8") as f:
    flat = pd.json_normalize(json.load(f), record_path="items", meta=["order_id", "order_date"])
print(len(flat), len(lines), len(flat) == len(lines))

# ৩. কঠিন
def load_csv(con, path, table, required):
    loaded, rejected = 0, []
    with open(path, newline="", encoding="utf-8") as f, con:
        reader = csv.DictReader(f)
        cols = reader.fieldnames
        sql = f"INSERT INTO {table} ({', '.join(cols)}) VALUES ({', '.join('?' * len(cols))})"
        for row in reader:
            empty = [c for c in required if not row[c].strip()]
            if empty:
                rejected.append({**row, "reason": f"empty: {', '.join(empty)}"})
                continue
            try:
                con.execute(sql, [row[c] or None for c in cols])
                loaded += 1
            except sqlite3.Error as exc:
                rejected.append({**row, "reason": str(exc)})
    if rejected:
        with open(path + ".rejected.csv", "w", newline="", encoding="utf-8") as f:
            w = csv.DictWriter(f, fieldnames=list(rejected[0]))
            w.writeheader()
            w.writerows(rejected)
    return loaded, len(rejected)

con.execute("CREATE TABLE price_changes (product_id INTEGER, new_price INTEGER CHECK (new_price > 0), "
            "FOREIGN KEY (product_id) REFERENCES products (product_id))")
with open("price_changes.csv", "w", newline="", encoding="utf-8") as f:
    f.write("product_id,new_price\n3,4300\n99,500\n4,0\n")
print(load_csv(con, "price_changes.csv", "price_changes", required=["product_id", "new_price"]))
with open("price_changes.csv.rejected.csv", encoding="utf-8") as f:
    print(f.read().strip())