File I/O & Working with CSV, JSON

# Writing a file
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("Hello, World!\n")
    f.write("Second line\n")

# Reading a file — whole content
with open("output.txt", "r", encoding="utf-8") as f:
    content = f.read()
print(content)

# Reading line by line (memory efficient for large files)
with open("output.txt", "r", encoding="utf-8") as f:
    for line in f:
        print(line.strip())  # strip() removes the \n

# Reading all lines as a list
with open("output.txt", "r") as f:
    lines = f.readlines()

# Appending to existing file
with open("output.txt", "a", encoding="utf-8") as f:
    f.write("Appended line\n")

from pathlib import Path

# Create path objects
current_dir = Path(".")
home = Path.home()
data_dir = Path("data")

# Build paths
config_file = data_dir / "config.json"  # Works on Windows and Unix
print(config_file)  # data/config.json  (or data\config.json on Windows)

# Check existence
print(config_file.exists())
print(config_file.is_file())
print(data_dir.is_dir())

# Create directories
output_dir = Path("output") / "results"
output_dir.mkdir(parents=True, exist_ok=True)  # Like mkdir -p

# File info
p = Path("output.txt")
print(p.name)          # 'output.txt'
print(p.stem)          # 'output'
print(p.suffix)        # '.txt'
print(p.parent)        # '.'

# Glob patterns
for py_file in Path(".").glob("**/*.py"):  # All .py files recursively
    print(py_file)

# Read and write directly
config_file.write_text('{"debug": true}', encoding="utf-8")
content = config_file.read_text(encoding="utf-8")

# Find all CSV files in a directory
csv_files = list(Path("data").glob("*.csv"))
print(f"Found {len(csv_files)} CSV files")

import csv

# Writing CSV
employees = [
    {"name": "Alice", "department": "Engineering", "salary": 95000},
    {"name": "Bob", "department": "Marketing", "salary": 72000},
    {"name": "Carol", "department": "Engineering", "salary": 105000},
]

with open("employees.csv", "w", newline="", encoding="utf-8") as f:
    fieldnames = ["name", "department", "salary"]
    writer = csv.DictWriter(f, fieldnames=fieldnames)
    
    writer.writeheader()
    writer.writerows(employees)

# Reading CSV
with open("employees.csv", "r", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(f"{row['name']}: ${int(row['salary']):,}")

# Reading with regular csv.reader (gives lists, not dicts)
with open("employees.csv", "r") as f:
    reader = csv.reader(f)
    header = next(reader)  # Skip header
    for row in reader:
        name, dept, salary = row
        print(f"{name} in {dept}")

from pathlib import Path
import csv
from collections import defaultdict
from statistics import mean

def analyze_sales_data(csv_path):
    """Analyze a sales CSV file."""
    records = []
    
    with open(csv_path, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row in reader:
            records.append({
                'product': row['product'],
                'region': row['region'],
                'revenue': float(row['revenue']),
                'units': int(row['units'])
            })
    
    # Total revenue
    total_revenue = sum(r['revenue'] for r in records)
    
    # Revenue by region
    by_region = defaultdict(float)
    for r in records:
        by_region[r['region']] += r['revenue']
    
    # Best-selling products
    by_product = defaultdict(int)
    for r in records:
        by_product[r['product']] += r['units']
    
    return {
        'total_revenue': total_revenue,
        'by_region': dict(sorted(by_region.items(), key=lambda x: -x[1])),
        'top_products': sorted(by_product.items(), key=lambda x: -x[1])[:5]
    }

import json

# Writing JSON
config = {
    "database": {
        "host": "localhost",
        "port": 5432,
        "name": "myapp"
    },
    "debug": False,
    "max_connections": 100,
    "features": ["search", "analytics", "notifications"]
}

with open("config.json", "w", encoding="utf-8") as f:
    json.dump(config, f, indent=2)  # indent=2 for pretty printing

# Reading JSON
with open("config.json", "r", encoding="utf-8") as f:
    loaded_config = json.load(f)

print(loaded_config["database"]["host"])  # localhost
print(loaded_config["features"])          # ['search', 'analytics', 'notifications']

# JSON from/to strings
json_string = json.dumps(config, indent=2)
parsed = json.loads(json_string)

# Handling non-serializable types
from datetime import datetime, date
import uuid

data = {
    "id": uuid.uuid4(),          # UUID is not JSON-serializable by default
    "created_at": datetime.now(),  # datetime either
    "amount": 99.99
}

# Custom encoder
class CustomEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, (datetime, date)):
            return obj.isoformat()
        if isinstance(obj, uuid.UUID):
            return str(obj)
        return super().default(obj)

json_str = json.dumps(data, cls=CustomEncoder, indent=2)
print(json_str)

# Writing JSONL
logs = [
    {"time": "2024-01-01T10:00:00", "level": "INFO", "message": "Server started"},
    {"time": "2024-01-01T10:01:00", "level": "ERROR", "message": "Connection failed"},
    {"time": "2024-01-01T10:02:00", "level": "INFO", "message": "Recovered"},
]

with open("logs.jsonl", "w") as f:
    for log in logs:
        f.write(json.dumps(log) + "\n")

# Reading JSONL — memory efficient
def read_jsonl(filepath):
    with open(filepath, 'r') as f:
        for line in f:
            line = line.strip()
            if line:
                yield json.loads(line)

error_logs = [log for log in read_jsonl("logs.jsonl") if log["level"] == "ERROR"]

from pathlib import Path
import json

def safe_read_config(config_path, defaults=None):
    """Safely read a config file with fallback defaults."""
    if defaults is None:
        defaults = {}
    
    path = Path(config_path)
    
    if not path.exists():
        print(f"Config file not found: {config_path}. Using defaults.")
        return defaults
    
    try:
        with open(path, 'r', encoding='utf-8') as f:
            config = json.load(f)
        return {**defaults, **config}  # Merge: file config overrides defaults
    
    except json.JSONDecodeError as e:
        print(f"Invalid JSON in {config_path}: {e}")
        return defaults
    
    except PermissionError:
        print(f"Permission denied reading {config_path}")
        return defaults

config = safe_read_config("config.json", defaults={"debug": False, "port": 8000})

def process_large_csv(input_path, output_path, transform_func):
    """Process a large CSV without loading it all into memory."""
    with open(input_path, 'r', encoding='utf-8', newline='') as infile, \
         open(output_path, 'w', encoding='utf-8', newline='') as outfile:
        
        reader = csv.DictReader(infile)
        writer = csv.DictWriter(outfile, fieldnames=reader.fieldnames)
        writer.writeheader()
        
        for i, row in enumerate(reader):
            transformed = transform_func(row)
            if transformed:  # Skip if transform returns None/False
                writer.writerow(transformed)
            
            if i % 10000 == 0:
                print(f"Processed {i:,} rows...")
    
    print(f"Done: {input_path} → {output_path}")

# Usage
def filter_and_transform(row):
    """Keep only high-value orders, add computed column."""
    revenue = float(row.get('revenue', 0))
    if revenue < 1000:
        return None  # Skip small orders
    return {**row, 'revenue_category': 'high' if revenue > 10000 else 'medium'}