11. Работа с файлами

Чтение и запись текстовых файлов

# Запись файла
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("Hello, World!\n")
    f.write("Привет, мир!\n")

# Дозапись в конец файла
with open("output.txt", "a", encoding="utf-8") as f:
    f.write("Ещё одна строка\n")

# Чтение всего файла
with open("output.txt", "r", encoding="utf-8") as f:
    content = f.read()
    print(content)

# Чтение построчно (эффективно для больших файлов)
with open("output.txt", "r", encoding="utf-8") as f:
    for line in f:
        print(line.strip())  # strip() убирает \n

# Чтение всех строк в список
with open("output.txt", "r", encoding="utf-8") as f:
    lines = f.readlines()  # ["Hello, World!\n", ...]
    lines = [line.strip() for line in f]  # без \n

pathlib — современный способ

from pathlib import Path

# Создание путей
base = Path("data")
file = base / "users.json"
nested = Path.home() / ".config" / "app" / "settings.json"

# Проверки
file.exists()      # существует ли
file.is_file()     # это файл?
file.is_dir()      # это папка?
file.suffix        # ".json"
file.stem          # "users"
file.name          # "users.json"
file.parent        # Path("data")

# Создание директорий
nested.parent.mkdir(parents=True, exist_ok=True)

# Чтение и запись
text = file.read_text(encoding="utf-8")
file.write_text('{"users": []}', encoding="utf-8")

bytes_data = file.read_bytes()
file.write_bytes(b"binary data")

# Поиск файлов
python_files = list(Path(".").glob("**/*.py"))  # рекурсивно
json_files = list(Path("data").glob("*.json"))  # только в data/

# Переименование и удаление
file.rename("data/users_backup.json")
file.unlink()       # удалить файл
Path("dir").rmdir() # удалить пустую папку

import shutil
shutil.rmtree("dir")          # удалить папку со содержимым
shutil.copy("src.txt", "dst.txt")  # копировать файл
shutil.copytree("src/", "dst/")    # копировать папку

JSON файлы

import json
from pathlib import Path

# Запись JSON
data = {
    "users": [
        {"id": 1, "name": "Яша", "role": "admin"},
        {"id": 2, "name": "Анна", "role": "user"},
    ],
    "total": 2
}

path = Path("users.json")
path.write_text(
    json.dumps(data, ensure_ascii=False, indent=2),
    encoding="utf-8"
)

# Чтение JSON
loaded = json.loads(path.read_text(encoding="utf-8"))

# Или через файловый объект
with open("users.json", "r", encoding="utf-8") as f:
    data = json.load(f)

# Кастомная сериализация для datetime
from datetime import datetime
import json

class DateTimeEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime):
            return obj.isoformat()
        return super().default(obj)

event = {"name": "Deploy", "at": datetime.now()}
json.dumps(event, cls=DateTimeEncoder)

CSV файлы

import csv
from pathlib import Path

# Чтение CSV
with open("users.csv", "r", newline="", encoding="utf-8") as f:
    reader = csv.DictReader(f)  # заголовки → ключи словаря
    users = list(reader)

# Запись CSV
users = [
    {"name": "Яша", "age": "2", "role": "admin"},
    {"name": "Анна", "age": "25", "role": "user"},
]

with open("output.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=["name", "age", "role"])
    writer.writeheader()
    writer.writerows(users)

Конфигурационные файлы

# TOML (Python 3.11+) — для pyproject.toml и конфигов
import tomllib  # Python 3.11+
# или: pip install tomli

with open("pyproject.toml", "rb") as f:
    config = tomllib.load(f)

# YAML — популярный формат конфигурации
# pip install pyyaml
import yaml

with open("config.yaml", "r") as f:
    config = yaml.safe_load(f)

# Пример config.yaml:
"""
database:
  host: localhost
  port: 5432
  name: mydb
server:
  port: 8000
  debug: true
"""

# .env файлы — переменные окружения
# pip install python-dotenv
from dotenv import load_dotenv
import os

load_dotenv(".env")  # читает .env
DATABASE_URL = os.environ["DATABASE_URL"]
SECRET_KEY = os.getenv("SECRET_KEY", "default-key")

Практический пример: обработка данных

import csv
import json
from pathlib import Path
from datetime import datetime
from collections import defaultdict

def process_sales(input_file: str, output_dir: str) -> dict:
    """
    Обрабатывает CSV с продажами и создаёт отчёты.

    Returns:
        Словарь со статистикой
    """
    output = Path(output_dir)
    output.mkdir(exist_ok=True)

    sales = []
    with open(input_file, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row in reader:
            sales.append({
                "date": datetime.strptime(row["date"], "%Y-%m-%d"),
                "product": row["product"],
                "amount": float(row["amount"]),
                "region": row["region"]
            })

    # Агрегация по продуктам
    by_product = defaultdict(float)
    for sale in sales:
        by_product[sale["product"]] += sale["amount"]

    # Топ-10 продуктов
    top_products = sorted(by_product.items(), key=lambda x: x[1], reverse=True)[:10]

    # Сохранить отчёт
    report = {
        "generated_at": datetime.now().isoformat(),
        "total_sales": sum(s["amount"] for s in sales),
        "total_records": len(sales),
        "top_products": [{"product": p, "revenue": r} for p, r in top_products]
    }

    (output / "report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2),
        encoding="utf-8"
    )

    return report

# Использование
result = process_sales("sales.csv", "reports/")
print(f"Всего продаж: {result['total_sales']:,.2f}₽")

Контекстный менеджер

# Свой контекстный менеджер (класс)
class FileTransaction:
    """Безопасная запись файла с откатом."""

    def __init__(self, filepath: str):
        self.filepath = Path(filepath)
        self.backup = None

    def __enter__(self):
        if self.filepath.exists():
            self.backup = self.filepath.read_bytes()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None and self.backup is not None:
            self.filepath.write_bytes(self.backup)  # откат
            print(f"Ошибка! Файл восстановлен.")
        self.backup = None
        return False  # не подавляем исключение

with FileTransaction("important.json") as ft:
    Path("important.json").write_text('{"broken": json}')  # ошибка при чтении
    data = json.loads(Path("important.json").read_text())   # вот здесь упадёт

# contextlib — более простой способ
from contextlib import contextmanager

@contextmanager
def temp_directory():
    import tempfile, shutil
    tmpdir = Path(tempfile.mkdtemp())
    try:
        yield tmpdir
    finally:
        shutil.rmtree(tmpdir)

with temp_directory() as tmpdir:
    (tmpdir / "test.txt").write_text("временный файл")
    # после выхода из блока директория удаляется

Задание

# Задание 1: Парсинг логов
# Напиши функцию parse_logs(log_file: str) -> dict
# Формат лога: "2024-01-15 10:30:00 ERROR Something went wrong"
# Верни: {"ERROR": 5, "WARNING": 3, "INFO": 100}

# Задание 2: Слияние JSON файлов
# Функция merge_configs(files: list[str]) -> dict
# Читает несколько JSON файлов и объединяет их (глубокое слияние)
# Более поздние файлы перезаписывают более ранние

# Задание 3: CSV обработка
# Дан файл students.csv: name,grade_math,grade_physics,grade_cs
# Напиши функцию analyze_students(file: str) -> dict которая вернёт:
# - average_by_subject: средний балл по каждому предмету
# - top_students: 3 лучших студента (по среднему)
# - failing: студенты с хотя бы одной оценкой < 60

В следующем уроке — обработка ошибок!