Версия: 0.0.1 Автор: Semykin Vladislav vladislav_semykin01@mail.ru
- Описание
- Цель проекта
- Основные задачи
- Архитектура и алгоритмы
- Быстрый старт
- Тестирование
- Документация
Collector - это универсальная система сбора и обработки файлов, поддерживающая три интерфейса взаимодействия:
- CLI (Command Line Interface) - командная строка для автоматизации и скриптов
- REST API - HTTP API для интеграции с другими системами
- GUI (Graphical User Interface) - графический интерфейс для интерактивной работы
Система предназначена для эффективного сбора файлов из указанных директорий с применением фильтрации по шаблонам, выполнения операций копирования или перемещения, создания архивов и сбора системной информации.
Основная цель проекта - предоставить универсальное, производительное и надежное решение для:
- Автоматизации сбора файлов - массовый сбор файлов по заданным критериям
- Организации данных - структурированное перемещение и копирование файлов
- Архивирования - создание архивов в различных форматах (ZIP, TAR, 7Z)
- Мониторинга систем - сбор информации о системе для диагностики
- Интеграции - предоставление API для интеграции с другими системами
Система выполняет следующие задачи:
- Рекурсивный обход директорий с поддержкой множественных исходных путей
- Фильтрация файлов по шаблонам (glob и regex)
- Обработка больших объемов файлов (тысячи и десятки тысяч файлов)
- Копирование - копирование файлов с сохранением структуры директорий
- Перемещение - перемещение файлов с удалением исходных
- Перемещение с удалением - перемещение с гарантированным удалением исходных файлов
- Создание архивов в форматах ZIP, TAR, 7Z
- Поддержка сжатия (gzip, bzip2, xz для TAR)
- Сохранение структуры директорий в архиве
- Информация о процессоре, памяти, дисках
- Информация о сетевых интерфейсах
- Информация о запущенных процессах
- Экспорт в JSON формат
- Отслеживание прогресса обработки в реальном времени
- WebSocket уведомления для API
- Callback-функции для интеграции
Применение: Операции с файлами (копирование, перемещение)
Реализация: FileOperationStrategy с реализациями CopyStrategy, MoveStrategy, MoveRemoveStrategy
Преимущества:
- Легкое добавление новых стратегий операций
- Инкапсуляция алгоритмов операций
- Возможность динамической смены стратегии
Применение: Построение конфигурации сбора
Реализация: CollectionConfigBuilder
Преимущества:
- Пошаговое построение сложных объектов
- Гибкость в настройке параметров
- Валидация на этапе построения
Применение: Отслеживание прогресса обработки
Реализация: ProgressTracker с подпиской на callback-функции
Преимущества:
- Разделение логики обработки и уведомлений
- Множественные подписчики на события
- Асинхронные уведомления
Применение: Параллельная обработка файлов
Реализация: WorkerPool с динамическим количеством потоков
Преимущества:
- Эффективное использование ресурсов CPU
- Масштабируемость под нагрузку
- Контроль параллелизма
Алгоритм: Рекурсивный обход с использованием Path.rglob()
Временная сложность: O(N), где N - общее количество файлов и директорий
Пространственная сложность: O(D), где D - глубина дерева директорий (для стека рекурсии)
Амортизационная сложность: O(1) на файл (благодаря оптимизации Python pathlib)
Реализация:
for filepath in source_path.rglob("*"):
if filepath.is_file():
all_files.append(filepath)Алгоритм: Линейный проход с кэшированием результатов
Временная сложность:
- Без кэша: O(N × M × P), где N - количество файлов, M - количество шаблонов, P - сложность шаблона
- С кэшем: O(N × M) в худшем случае, O(N) в среднем (при попадании в кэш)
Пространственная сложность: O(N × M) для кэша результатов
Амортизационная сложность: O(1) на файл при использовании кэша
Реализация:
def match(self, filepath: Path, pattern_config: PatternConfig) -> bool:
cache_key = f"{filepath}:{pattern_config.pattern}:{pattern_config.pattern_type}"
if cache_key in self._cache:
return self._cache[cache_key] # O(1) - кэш попадание
# O(P) - сложность зависит от типа шаблона
if pattern_config.pattern_type == "regex":
result = self._match_regex(pattern_config.pattern, filepath)
else:
result = self._match_glob(pattern_config.pattern, filepath)
self._cache[cache_key] = result
return resultТипы шаблонов:
- Glob: O(1) - простая проверка имени файла через
fnmatch - Regex: O(P), где P - длина строки пути (используется
re.search)
Алгоритм: Разделение файлов на батчи и распределение по потокам
Временная сложность: O(N / W), где N - количество файлов, W - количество потоков
Пространственная сложность: O(N) для хранения списка файлов
Амортизационная сложность: O(1) на файл при оптимальном количестве потоков
Оптимальное количество потоков:
def _calculate_optimal_workers(self, total_files: int) -> int:
return min(
os.cpu_count() or 4, # Количество ядер CPU
max(1, total_files // 100), # Один поток на 100 файлов
MAX_WORKERS, # Максимум 32 потока
)Сложность батчинга: O(N) для разделения на батчи
Алгоритм: Thread-local счетчики с периодической синхронизацией
Временная сложность:
- Инкремент: O(1) - thread-local операция
- Flush: O(1) - атомарная операция обновления общего счетчика
- Callback уведомления: O(C), где C - количество callback-функций
Пространственная сложность: O(T), где T - количество потоков (для thread-local хранилища)
Амортизационная сложность: O(1) на инкремент благодаря батчингу
Оптимизация:
- Thread-local счетчики снижают количество операций блокировки в 333-500 раз
- Batch updates минимизируют contention
- Callbacks выполняются вне блокировки
Детали оптимизации см. в: docs/OPTIMIZATION_EVOLUTION.md
Алгоритм: Поиск наибольшего общего префикса путей
Временная сложность: O(N × L), где N - количество путей, L - средняя длина пути
Пространственная сложность: O(N × L) для хранения разрешенных путей
Амортизационная сложность: O(L) на путь
Реализация:
def _find_common_base(filepaths: List[Path], source_paths: List[Path]) -> Path:
# Разрешение путей: O(N × L)
resolved_paths = [Path(p).resolve() for p in source_paths]
# Поиск общего префикса: O(N × L)
common_parts = None
for filepath in filepaths[:10]: # Ограничение для производительности
# Поиск относительного пути: O(L)
relative = filepath.resolve().relative_to(source_path)
# Обновление общего префикса: O(L)
...Структура: Dict[str, bool] - словарь для хранения результатов сопоставления
Операции:
- Поиск: O(1) в среднем, O(N) в худшем случае (коллизии хэша)
- Вставка: O(1) в среднем, O(N) в худшем случае
Использование: Кэширование результатов сопоставления файлов с шаблонами для избежания повторных вычислений
Структура: threading.local() - изолированное хранилище для каждого потока
Операции:
- Доступ: O(1)
- Изоляция: Гарантирована на уровне Python
Использование: Хранение счетчиков прогресса для каждого потока без синхронизации
Структура: List[threading.Thread] - список потоков-обработчиков
Операции:
- Создание потока: O(1)
- Запуск потока: O(1)
- Ожидание завершения: O(T), где T - время выполнения задачи
Использование: Управление параллельной обработкой файлов
- Python 3.7 или выше
- pip (менеджер пакетов Python)
git clone <repository-url>
cd collectorWindows:
python -m venv venv
.\venv\Scripts\Activate.ps1Linux/macOS:
python3 -m venv venv
source venv/bin/activateБазовые зависимости (для использования):
pip install -e .С зависимостями для разработки:
pip install -e ".[dev]"Обновление зависимостей:
pip install -e ".[dev]" --upgradeФайл pyproject.toml содержит конфигурацию проекта:
[project] - метаданные проекта:
name- имя пакетаversion- версияdependencies- основные зависимостиrequires-python- минимальная версия Python
[project.optional-dependencies] - опциональные зависимости:
dev- зависимости для разработки (black, flake8, mypy, pytest и др.)
[project.scripts] - точки входа:
log-collector-cli- CLI интерфейсlog-collector-api- API серверlog-collector-gui- GUI приложение
[tool.black] - настройки форматирования кода:
line-length = 120- максимальная длина строкиtarget-version = ['py37']- целевая версия Python
[tool.flake8] - настройки линтера:
max-line-length = 120- максимальная длина строкиmax-complexity = 20- максимальная цикломатическая сложность
[tool.pytest.ini_options] - настройки тестирования:
timeout = 60- таймаут для тестовmarkers- маркеры для категоризации тестов
[tool.mypy] - настройки статической проверки типов:
python_version = "3.8"- версия Python для проверкиwarn_return_any = true- предупреждения о возврате Any
Базовое использование:
log-collector-cli --source-paths /path/to/source --target-path /path/to/targetС фильтрацией по шаблону:
log-collector-cli \
--source-paths /var/log \
--target-path /backup/logs \
--patterns "*.log" "*.txt" \
--pattern-type globС созданием архива:
log-collector-cli \
--source-paths /var/log \
--target-path /backup/logs \
--create-archive \
--archive-format zipПодробнее см.: docs/CLI_GUIDE.md
Запуск API сервера:
log-collector-apiИли через uvicorn:
uvicorn src.api.main:app --host 0.0.0.0 --port 8000Базовое использование:
curl -X POST "http://localhost:8000/api/v1/collect" \
-H "Content-Type: application/json" \
-d '{
"source_paths": ["/var/log"],
"target_path": "/backup/logs",
"patterns": [{"pattern": "*.log", "pattern_type": "glob"}]
}'Подробнее см.: docs/API_GUIDE.md
Запуск GUI:
log-collector-guiВсе тесты:
pytest .Конкретный файл:
pytest tests/test_file_filter.pyКонкретный тест:
pytest tests/test_file_filter.py::test_match_regexС маркерами:
pytest -m unit # Только unit тесты
pytest -m "not slow" # Исключить медленные тесты
pytest -m integration # Только интеграционные тестыС подробным выводом:
pytest -v # Verbose режим
pytest -vv # Очень подробный вывод
pytest -x # Остановка на первой ошибкеС покрытием кода:
pytest --cov=src --cov-report=htmlТолько упавшие тесты:
pytest --lf # Last failedПараллельное выполнение:
pytest -n auto # Требует pytest-xdist====================================================================================== test session starts ======================================================================================
platform win32 -- Python 3.8.10, pytest-8.3.5, pluggy-1.5.0
rootdir: C:\Users\vladislavsemykin\Downloads\collector
configfile: pyproject.toml
plugins: anyio-4.5.2, cov-5.0.0
collected 218 items
tests\test_api.py ........... [ 5%]
tests\test_archiver.py ............. [ 11%]
tests\test_cli.py ............. [ 16%]
tests\test_collection_service.py .............. [ 23%]
tests\test_exception_wrapper.py ......... [ 27%]
tests\test_file_filter.py .............................. [ 41%]
tests\test_file_operations.py ............... [ 48%]
tests\test_path_sanitizer.py ......................... [ 59%]
tests\test_production_logs.py ............... [ 66%]
tests\test_progress_tracker.py ............. [ 72%]
tests\test_security.py ................... [ 81%]
tests\test_validator.py ...................... [ 91%]
tests\test_worker_pool.py ................... [100%]
======================================================================================= warnings summary ========================================================================================
venv38\lib\site-packages\_pytest\config\__init__.py:1441
c:\users\vladislavsemykin\downloads\collector\venv38\lib\site-packages\_pytest\config\__init__.py:1441: PytestConfigWarning: Unknown config option: timeout
self._warn_or_fail_if_strict(f"Unknown config option: {key}\n")
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
===================================================================================== slowest 10 durations ======================================================================================
3.66s call tests/test_production_logs.py::TestProductionLogsArchiving::test_archive_production_logs_tar_gz
2.82s call tests/test_api.py::TestAPIRateLimiting::test_rate_limiting
2.16s call tests/test_production_logs.py::TestProductionLogsFullWorkflow::test_full_workflow_with_system_info
1.66s call tests/test_collection_service.py::TestCollectionServicePCInfoCollector::test_collect_with_system_info
1.59s call tests/test_production_logs.py::TestProductionLogsFullWorkflow::test_full_workflow_collect_and_archive
1.18s call tests/test_production_logs.py::TestProductionLogsArchiving::test_archive_production_logs_with_progress
1.13s call tests/test_production_logs.py::TestProductionLogsCollection::test_collect_production_logs_copy_mode
1.12s call tests/test_production_logs.py::TestProductionLogsArchiving::test_archive_production_logs_zip
1.11s call tests/test_security.py::TestRateLimiting::test_rate_limiting_resets_after_window
0.73s call tests/test_production_logs.py::TestProductionLogsCollection::test_collect_production_logs_move_mode
================================================================================ 218 passed, 1 warning in 29.29s ================================================================================
Интерпретация результатов:
- 218 passed - все тесты прошли успешно
- 1 warning - одно предупреждение (неизвестная опция конфигурации)
- 29.29s - общее время выполнения
- slowest 10 durations - список самых медленных тестов
Unit тесты - тестирование отдельных компонентов:
test_file_filter.py- фильтрация файловtest_progress_tracker.py- отслеживание прогрессаtest_validator.py- валидация конфигурации
Integration тесты - тестирование взаимодействия компонентов:
test_collection_service.py- сервис сбораtest_worker_pool.py- пул потоковtest_api.py- API эндпоинты
Security тесты - тестирование безопасности:
test_security.py- проверка безопасности путей, rate limiting
Подробная документация находится в директории docs/:
docs/OPTIMIZATION_EVOLUTION.md- эволюция оптимизаций системы отслеживания прогрессаdocs/API_GUIDE.md- подробное руководство по REST APIdocs/CLI_GUIDE.md- подробное руководство по CLI интерфейсу
- Python 3.7+ - основной язык программирования
- FastAPI - веб-фреймворк для REST API
- Uvicorn - ASGI сервер
- Pydantic - валидация данных
- PyJWT - JWT аутентификация
- PyYAML - работа с YAML конфигурациями
- psutil - сбор системной информации
- py7zr - поддержка формата 7Z
MIT License
Конец документа