storage.py - master - LKastornova/Python_lesson

storage.py
254 строки10 KB
@lvkastor
Initial commit: Finance Manager 2026 application
11 янв 2026, 19:15
11 янв 2026, 19:15209a1c3
"""
Модуль для работы с хранением данных приложения.

Обеспечивает сохранение и загрузку финансовых операций в различных форматах:
- CSV (основной формат хранения)
- JSON (для импорта/экспорта)

Модуль обрабатывает ошибки ввода/вывода и обеспечивает целостность данных.
"""

import csv
import json
from datetime import datetime
from models import FinancialOperation
import os

def ensure_data_directory():
    """
    Создает директорию data, если она не существует.
    
    Гарантирует, что директория для хранения файлов с данными существует.
    Если директория не существует, создает ее.
    
    Пример:
        >>> ensure_data_directory()
        # Создает папку 'data' в текущей директории
    """
    if not os.path.exists('data'):
        os.makedirs('data')

def load_operations(filename='data/operations.csv'):
    """
    Загружает операции из CSV файла.
    
    Args:
        filename: Путь к CSV файлу (по умолчанию 'data/operations.csv')
    
    Returns:
        list: Список объектов FinancialOperation, загруженных из файла.
              Если файл не существует, возвращает пустой список.
    
    Raises:
        IOError: Если возникла ошибка при чтении файла (кроме FileNotFoundError)
        ValueError: Если данные в файле некорректны
    
    Пример:
        >>> operations = load_operations()
        >>> print(f"Загружено {len(operations)} операций")
        Загружено 25 операций
    """
    ensure_data_directory()
    operations = []
    
    try:
        with open(filename, mode='r', newline='', encoding='utf-8') as file:
            reader = csv.DictReader(file)
            for row in reader:
                try:
                    # Обработка пустых значений
                    subcategory = row['subcategory'] if row['subcategory'] else None
                    comment = row['comment'] if row['comment'] else None
                    
                    # Парсим дату
                    date_str = row['date']
                    try:
                        date_obj = datetime.strptime(date_str, '%Y-%m-%d %H:%M:%S')
                    except ValueError:
                        # Если время отсутствует, добавляем 00:00:00
                        date_obj = datetime.strptime(date_str, '%Y-%m-%d')
                    
                    operation = FinancialOperation(
                        operation_type=row['operation_type'],
                        amount=float(row['amount']),
                        category=row['category'],
                        subcategory=subcategory,
                        date=date_obj,
                        comment=comment
                    )
                    operations.append(operation)
                except (ValueError, KeyError) as e:
                    print(f"Ошибка при обработке строки: {row}. Ошибка: {e}")
                    continue
    except FileNotFoundError:
        # Файл не существует, создадим его при первом сохранении
        pass
    except Exception as e:
        print(f"Ошибка при загрузке файла: {e}")
    
    return operations

def save_operations(operations, filename='data/operations.csv'):
    """
    Сохраняет операции в CSV файл.
    
    Args:
        operations: Список объектов FinancialOperation для сохранения
        filename: Путь к CSV файлу (по умолчанию 'data/operations.csv')
    
    Returns:
        bool: True если сохранение успешно, иначе False
    
    Raises:
        IOError: Если возникла ошибка при записи в файл
        Exception: Если произошла другая ошибка при сохранении
    
    Пример:
        >>> operations = [FinancialOperation('income', 1000, 'работа')]
        >>> save_operations(operations)
        True
    """
    ensure_data_directory()
    
    try:
        with open(filename, mode='w', newline='', encoding='utf-8') as file:
            fieldnames = ['operation_type', 'amount', 'category', 'subcategory', 'date', 'comment']
            writer = csv.DictWriter(file, fieldnames=fieldnames)
            writer.writeheader()
            for operation in operations:
                writer.writerow({
                    'operation_type': operation.operation_type,
                    'amount': operation.amount,
                    'category': operation.category,
                    'subcategory': operation.subcategory or '',
                    'date': operation.date.strftime('%Y-%m-%d %H:%M:%S'),
                    'comment': operation.comment or ''
                })
        return True
    except Exception as e:
        raise Exception(f"Не удалось сохранить данные: {str(e)}")

def export_to_csv(operations, filename):
    """
    Экспортирует операции в CSV файл.
    
    Args:
        operations: Список объектов FinancialOperation для экспорта
        filename: Путь к файлу для сохранения
    
    Returns:
        bool: True если экспорт успешен
    
    Raises:
        Exception: Если не удалось экспортировать данные
    
    Примечание:
        Функция-обертка над save_operations для экспорта в произвольный файл.
    """
    save_operations(operations, filename)

def export_to_json(operations, filename):
    """
    Экспортирует операции в JSON файл.
    
    Args:
        operations: Список объектов FinancialOperation для экспорта
        filename: Путь к JSON файлу для сохранения
    
    Returns:
        bool: True если экспорт успешен
    
    Raises:
        Exception: Если не удалось экспортировать данные
    
    Пример:
        >>> export_to_json(operations, 'backup.json')
        True
    """
    ensure_data_directory()
    
    try:
        data = []
        for op in operations:
            data.append({
                'operation_type': op.operation_type,
                'amount': op.amount,
                'category': op.category,
                'subcategory': op.subcategory,
                'date': op.date.strftime('%Y-%m-%d %H:%M:%S'),
                'comment': op.comment
            })
        
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        return True
    except Exception as e:
        raise Exception(f"Не удалось экспортировать в JSON: {str(e)}")

def import_from_csv(filename):
    """
    Импортирует операции из CSV файла.
    
    Args:
        filename: Путь к CSV файлу для импорта
    
    Returns:
        list: Список объектов FinancialOperation, загруженных из файла
    
    Raises:
        Exception: Если не удалось импортировать данные
    
    Примечание:
        Функция-обертка над load_operations для импорта из произвольного файла.
    """
    return load_operations(filename)

def import_from_json(filename):
    """
    Импортирует операции из JSON файла.
    
    Args:
        filename: Путь к JSON файлу для импорта
    
    Returns:
        list: Список объектов FinancialOperation, загруженных из файла
    
    Raises:
        Exception: Если не удалось импортировать данные
    
    Пример:
        >>> imported_ops = import_from_json('backup.json')
        >>> print(f"Импортировано {len(imported_ops)} операций")
        Импортировано 15 операций
    """
    operations = []
    
    try:
        with open(filename, 'r', encoding='utf-8') as f:
            data = json.load(f)
            
        for item in data:
            try:
                # Парсим дату
                date_str = item['date']
                try:
                    date_obj = datetime.strptime(date_str, '%Y-%m-%d %H:%M:%S')
                except ValueError:
                    date_obj = datetime.strptime(date_str, '%Y-%m-%d')
                
                operation = FinancialOperation(
                    operation_type=item['operation_type'],
                    amount=float(item['amount']),
                    category=item['category'],
                    subcategory=item.get('subcategory'),
                    date=date_obj,
                    comment=item.get('comment')
                )
                operations.append(operation)
            except (ValueError, KeyError) as e:
                print(f"Ошибка при обработке элемента: {item}. Ошибка: {e}")
                continue
    except Exception as e:
        raise Exception(f"Не удалось импортировать из JSON: {str(e)}")
    
    return operations