Files & data

import mmap
import os

def read_tail_with_mmap(filepath, bytes_from_end=64):
    """Read the last bytes of a large file using a read-only mmap."""
    file_size = os.path.getsize(filepath)
    start = max(0, file_size - bytes_from_end)

    with open(filepath, "rb") as f:
        with mmap.mmap(f.fileno(), length=0, a…

import heapq


def merge_sorted_chunks(chunks, output_path):
    """Merge multiple sorted iterables into single sorted output file."""
    with open(output_path, "w") as out_f:
        # Open all chunk files
        handles = [open(chunk, "r") for chunk in chunks]
        try:
            # Heap of (value, index) tupl…

import re
from pathlib import Path

def parse_apache_line(line):
    pattern = r'^(\S+) (\S+) (\S+) \[([^\]]+)\] "(\S+) (\S+) (\S+)" (\d{3}) (\S+)'
    match = re.match(pattern, line)
    if not match:
        return None
    ip, ident, user, timestamp, method, path, protocol, status, size = match.groups()
    return …

import requests
from bs4 import BeautifulSoup

def scrape_headlines(url: str, selector: str) -> list:
    """
    Scrape headlines from a news website using Beautiful Soup.
    
    Args:
        url: The URL of the news website.
        selector: CSS selector for headline elements.
    
    Returns:
        List of h…

import csv
from pathlib import Path

def process_csv_in_chunks(file_path, chunk_size=1000):
    """Yield rows from a large CSV file in chunks without loading all into memory."""
    with open(file_path, 'r', newline='') as f:
        reader = csv.DictReader(f)
        chunk = []
        for row in reader:
            …

import os
import shutil
import sys
from pathlib import Path

def sync_folders(src: Path, dst: Path):
    """Sync src folder to dst folder, copying missing/updated files."""
    dst.mkdir(parents=True, exist_ok=True)

    for src_path in src.rglob("*"):
        relative = src_path.relative_to(src)
        dst_path = ds…

import fcntl
import os
import tempfile
import time

def acquire_exclusive_lock(filepath):
    fd = os.open(filepath, os.O_RDWR | os.O_CREAT)
    try:
        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        print(f"Exclusive lock acquired on {filepath}")
        time.sleep(1)  # Simulate work while holding the l…

from pathlib import Path
import tempfile
import os

def write_lines_safely(lines: list[str], filepath: str | Path) -> None:
    """Write lines to a text file atomically to avoid corruption."""
    path = Path(filepath)
    path.parent.mkdir(parents=True, exist_ok=True)
    
    fd, temp_path = tempfile.mkstemp(dir=str…

import csv

def join_csv(file1, file2, key, output="joined.csv"):
    # Read first CSV into dict keyed by the join column
    with open(file1, newline="") as f1:
        reader1 = csv.DictReader(f1)
        data1 = {row[key]: row for row in reader1}

    # Read second CSV and merge matching rows
    with open(file2, n…

```python
import csv
from pathlib import Path
from typing import Iterator, List

def read_parquet_like_columnar(csv_path: str, column_names: List[str], chunk_size: int = 2) -> Iterator[dict]:
    """Read CSV data in columnar chunks, similar to how parquet stores columns."""
    csv_file = Path(csv_path)
    with csv_f…

import requests
from bs4 import BeautifulSoup
import csv

url = "https://en.wikipedia.org/wiki/List_of_countries_by_GDP_(nominal)"
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')

tables = soup.find_all('table', {'class': 'wikitable'})

if tables:
    target_table = tables[2]
    rows =…

import time
from pathlib import Path

def tail_log(file_path, n=10, poll_interval=1.0, timeout=10):
    """
    Print the last n lines and follow new lines appended to a growing log file.
    """
    path = Path(file_path)
    # Read the last n lines from the current file
    with path.open("r", encoding="utf-8") as f…