Python File I/O & JSON: Reading, Writing, and Parsing Data

# Opening files - ALWAYS use context manager (with)
with open('file.txt', 'r') as f:
    content = f.read()

# Read modes:
# 'r' - Read (default)
# 'w' - Write (overwrites)
# 'a' - Append (adds to end)
# 'x' - Exclusive creation (fails if exists)
# 'b' - Binary mode (for images, etc.)

# Read entire file
with open('config.txt', 'r') as f:
    content = f.read()
    print(content)

# Read line by line (memory efficient for large files)
with open('large_log.txt', 'r') as f:
    for line in f:
        print(line.strip())

# Read all lines into list
with open('data.txt', 'r') as f:
    lines = f.readlines()

# Write to file
with open('output.txt', 'w') as f:
    f.write('Hello, World!\n')
    f.writelines(['Line 1\n', 'Line 2\n'])

# Append to file
with open('log.txt', 'a') as f:
    f.write('New log entry\n')

# Check if file exists
import os
if os.path.exists('config.txt'):
    print('File exists')

# Delete file
import os
os.remove('old_file.txt')

# Rename file
os.rename('old.txt', 'new.txt')

from pathlib import Path

# Create Path objects (works on Windows, Linux, Mac)
config_path = Path('/etc/myapp/config.yaml')
home_dir = Path.home()
current_dir = Path.cwd()

# Join paths (uses correct separator for OS)
log_file = Path('/var/log') / 'myapp' / 'app.log'

# Check properties
if config_path.exists():
    print(f'Size: {config_path.stat().st_size} bytes')
    print(f'Is file: {config_path.is_file()}')
    print(f'Is directory: {config_path.is_dir()}')

# Read/write with pathlib
content = Path('file.txt').read_text()
Path('output.txt').write_text('Hello, World!')

# Iterate through directory
for file in Path('/var/log').glob('*.log'):
    print(f'Found log: {file.name}')

# Create directories
Path('/tmp/myapp/data').mkdir(parents=True, exist_ok=True)

# Get file parts
path = Path('/home/user/data/config.yaml')
print(f'Name: {path.name}')           # config.yaml
print(f'Stem: {path.stem}')           # config
print(f'Suffix: {path.suffix}')       # .yaml
print(f'Parent: {path.parent}')       # /home/user/data

import json

# Sample data
data = {
    'name': 'web-server',
    'port': 8080,
    'enabled': True,
    'tags': ['production', 'web'],
    'config': {
        'timeout': 30,
        'retries': 3
    }
}

# Write JSON to file
with open('config.json', 'w') as f:
    json.dump(data, f, indent=2)

# Read JSON from file
with open('config.json', 'r') as f:
    loaded_data = json.load(f)
    print(loaded_data['name'])

# Convert string to JSON
json_string = '{"name": "server", "port": 8080}'
parsed = json.loads(json_string)

# Convert Python object to JSON string
json_output = json.dumps(data, indent=2)

# Handle custom objects
def datetime_handler(obj):
    if hasattr(obj, 'isoformat'):
        return obj.isoformat()
    raise TypeError(f'Object of type {type(obj)} is not JSON serializable')

from datetime import datetime
now = datetime.now()
data_with_date = {'timestamp': now, 'value': 100}
json_str = json.dumps(data_with_date, default=datetime_handler)
print(json_str)

import csv

# Write CSV file
data = [
    ['Name', 'CPU', 'Memory', 'Status'],
    ['web-01', 45, 60, 'running'],
    ['web-02', 52, 65, 'running'],
    ['db-01', 38, 80, 'degraded']
]

with open('servers.csv', 'w', newline='') as f:
    writer = csv.writer(f)
    writer.writerows(data)

# Read CSV file
with open('servers.csv', 'r') as f:
    reader = csv.reader(f)
    for row in reader:
        print(f'Row: {row}')

# DictReader (uses header row as keys)
with open('servers.csv', 'r') as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(f"{row['Name']}: {row['CPU']}% CPU")

# DictWriter
with open('output.csv', 'w', newline='') as f:
    fieldnames = ['Name', 'CPU', 'Memory', 'Status']
    writer = csv.DictWriter(f, fieldnames=fieldnames)
    writer.writeheader()
    writer.writerow({'Name': 'cache-01', 'CPU': 25, 'Memory': 45, 'Status': 'running'})

# Read CSV with custom delimiter (e.g., TSV)
with open('data.tsv', 'r') as f:
    reader = csv.reader(f, delimiter='\t')
    for row in reader:
        print(row)

# Memory-efficient: read line by line
def process_large_log(log_file, error_pattern):
    errors = []
    with open(log_file, 'r') as f:
        for line in f:
            if error_pattern in line:
                errors.append(line.strip())
                # Don't store too many
                if len(errors) > 1000:
                    process_errors(errors)
                    errors = []
    return errors

# Using generators for even better efficiency
def read_large_file(file_path):
    with open(file_path, 'r') as f:
        for line in f:
            yield line.strip()

# Use it
for line in read_large_file('huge_log.txt'):
    if 'ERROR' in line:
        print(line)

# Process in chunks
def process_in_chunks(file_path, chunk_size=1024*1024):  # 1MB chunks
    with open(file_path, 'rb') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            # Process chunk
            process_chunk(chunk)

# Use tempfile for temporary data
import tempfile
with tempfile.NamedTemporaryFile(mode='w', delete=False) as tmp:
    tmp.write('Temporary data')
    tmp_path = tmp.name
# Use tmp_path...
os.unlink(tmp_path)  # Clean up

#!/usr/bin/env python3
"""
log_config_manager.py - Parse logs and manage configuration files
"""

import json
import csv
import re
from pathlib import Path
from datetime import datetime
from collections import defaultdict, Counter

class LogConfigManager:
    """Manage log files and configuration"""
    
    def __init__(self, config_path=None):
        self.config = self.load_config(config_path)
        self.log_stats = defaultdict(Counter)
    
    def load_config(self, config_path):
        """Load configuration from JSON file"""
        if config_path and Path(config_path).exists():
            with open(config_path, 'r') as f:
                return json.load(f)
        
        # Default configuration
        return {
            'log_patterns': {
                'error': r'ERROR|CRITICAL|FATAL',
                'warning': r'WARNING|WARN',
                'info': r'INFO',
                'debug': r'DEBUG'
            },
            'output': {
                'summary': 'summary.json',
                'errors_csv': 'errors.csv'
            }
        }
    
    def save_config(self, config_path):
        """Save configuration to JSON file"""
        with open(config_path, 'w') as f:
            json.dump(self.config, f, indent=2)
        print(f"✓ Config saved to {config_path}")
    
    def parse_log_file(self, log_path):
        """Parse log file and extract statistics"""
        if not Path(log_path).exists():
            print(f"Error: {log_path} not found")
            return
        
        print(f"Parsing {log_path}...")
        
        with open(log_path, 'r') as f:
            for line_num, line in enumerate(f, 1):
                # Count by log level
                for level, pattern in self.config['log_patterns'].items():
                    if re.search(pattern, line, re.IGNORECASE):
                        self.log_stats[level]['count'] += 1
                        
                        # Extract timestamp if present
                        timestamp_match = re.search(r'\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}', line)
                        if timestamp_match:
                            hour = timestamp_match.group(0)[11:13]
                            self.log_stats[level][f'hour_{hour}'] += 1
                        
                        # Store sample errors
                        if level == 'error' and self.log_stats[level]['samples'] < 10:
                            if 'samples' not in self.log_stats[level]:
                                self.log_stats[level]['samples'] = 0
                            if self.log_stats[level]['samples'] < 10:
                                self.log_stats[level][f'sample_{self.log_stats[level]["samples"]}'] = line.strip()
                                self.log_stats[level]['samples'] += 1
                        break
        
        print(f"✓ Parsed {line_num} lines")
    
    def generate_report(self):
        """Generate summary report"""
        report = {
            'timestamp': datetime.now().isoformat(),
            'summary': {}
        }
        
        total = 0
        for level, stats in self.log_stats.items():
            count = stats.get('count', 0)
            total += count
            report['summary'][level] = {
                'count': count,
                'percentage': (count / total * 100) if total > 0 else 0
            }
        
        report['total_lines_processed'] = total
        return report
    
    def export_errors_to_csv(self):
        """Export errors to CSV file"""
        error_samples = []
        for key, value in self.log_stats.get('error', {}).items():
            if key.startswith('sample_'):
                error_samples.append({'error': value})
        
        if error_samples:
            output_path = self.config['output']['errors_csv']
            with open(output_path, 'w', newline='') as f:
                writer = csv.DictWriter(f, fieldnames=['error'])
                writer.writeheader()
                writer.writerows(error_samples)
            print(f"✓ Errors exported to {output_path}")
    
    def save_summary(self):
        """Save summary to JSON"""
        report = self.generate_report()
        output_path = self.config['output']['summary']
        with open(output_path, 'w') as f:
            json.dump(report, f, indent=2)
        print(f"✓ Summary saved to {output_path}")
        
        # Print summary to console
        print("\n=== Log Analysis Summary ===")
        for level, stats in report['summary'].items():
            print(f"{level.upper():10}: {stats['count']:6} ({stats['percentage']:.1f}%)")


def main():
    import argparse
    
    parser = argparse.ArgumentParser(description='Log and Config Manager')
    parser.add_argument('log_file', help='Path to log file to parse')
    parser.add_argument('--config', '-c', help='Path to config JSON file')
    parser.add_argument('--export-errors', action='store_true', help='Export errors to CSV')
    
    args = parser.parse_args()
    
    manager = LogConfigManager(args.config)
    manager.parse_log_file(args.log_file)
    manager.save_summary()
    
    if args.export_errors:
        manager.export_errors_to_csv()

if __name__ == '__main__':
    main()