Индексация данных в файловой системе с помощью Python
Основной подход к индексации файлов
Эффективное решение с pathlib
Для создания индекса всех файлов в каталоге и его подпапках рекомендуется использовать модуль pathlib. Он предоставляет удобный интерфейс для работы с путями и встроенную рекурсивную итерацию. Функция ниже возвращает словарь с числовыми ключами и полными путями к файлам.
from pathlib import Path
def build_index(path: str) -> dict:
index = {}
for idx, file in enumerate(Path(path).rglob('*'), start=1):
if file.is_file():
index[idx] = str(file.resolve())
return index
# Пример использования
print(build_index('/tmp/test'))ввод программ на python (ввод данных в программе python)
Такой подход позволяет быстро получить пронумерованный список файлов. Преимущество: автоматическая обработка путей, кроссплатформенность, читаемый код.
Возможные проблемы и их решение
- При очень большом количестве файлов (миллионы) rglob может замедлиться. Альтернатива - использовать os.scandir с явным управлением рекурсией.
- Ошибки доступа к отдельным папкам приводят к исключению PermissionError. Рекомендуется обернуть вызов в try-except или использовать ignore-функции.
- Символические ссылки могут зациклиться. В pathlib по умолчанию они разрешаются, но можно отключить через параметр follow_symlinks=False.
Как пронумеровать файлы только в текущей папке без рекурсии?
import os
def index_flat(path: str) -> dict:
files = [f for f in os.listdir(path) if os.path.isfile(os.path.join(path, f))]
return {i: os.path.join(path, f) for i, f in enumerate(files, start=1)}Python file io (ввод-вывод файлов в python)
Этот вариант прост, но не затрагивает подпапки. Используется, когда нужен только список файлов в одном каталоге.
Ошибка: os.listdir не различает файлы и папки, поэтому требуется дополнительная проверка isfile. При работе с большими папками os.listdir может потреблять много памяти, лучше использовать os.scandir.
Как построить индекс с фильтрацией по расширению?
from pathlib import Path
def index_filtered(path: str, extension: str) -> dict:
index = {}
pattern = f'**/*{extension}'
for idx, file in enumerate(Path(path).glob(pattern), start=1):
if file.is_file():
index[idx] = str(file.resolve())
return indexPython temp files (временные файлы в python)
Фильтрация происходит на этапе обхода, что сокращает объём обрабатываемых данных. Для нескольких расширений можно комбинировать через or.
Важно: glob.iglob может не найти файлы, если расширение указано неверно (регистр символов в Windows не учитывается, в Linux - да). Рекомендуется приводить к нижнему регистру.
Как сохранить индекс для последующего быстрого поиска?
import sqlite3
def save_index_to_db(path: str, db_path: str):
conn = sqlite3.connect(db_path)
c = conn.cursor()
c.execute('CREATE TABLE IF NOT EXISTS index_files (id INTEGER PRIMARY KEY, path TEXT)')
for idx, file in enumerate(Path(path).rglob('*'), start=1):
if file.is_file():
c.execute('INSERT INTO index_files (id, path) VALUES (?, ?)', (idx, str(file.resolve())))
conn.commit()
conn.close()Хранение индекса в SQLite позволяет выполнять произвольные запросы (по части пути, по id) без повторного обхода файловой системы. Это удобно при частом обращении к одному и тому же каталогу.
Проблема: база данных может быстро расти. Для миллионов записей требуется индексирование полей. Также при изменении файловой структуры индекс устаревает - нужна периодическая переиндексация или отслеживание через inotify.
Расширенные примеры индексации файлов
Ниже приведены более сложные сценарии, которые могут потребоваться в реальных проектах.
Индексация с метаданными
Помимо пути, часто нужно хранить размер, дату изменения и тип файла. Используем os.stat для получения атрибутов.
from pathlib import Path
import os, time
def index_with_metadata(base: str) -> list:
records = []
for file in Path(base).rglob('*'):
if file.is_file():
stat = file.stat()
records.append({
'path': str(file),
'size': stat.st_size,
'mtime': time.ctime(stat.st_mtime),
'permissions': oct(stat.st_mode)[-3:]
})
return records
result = index_with_metadata('/tmp/test')
print(result[:2])[{'path': '/tmp/test/file1.txt', 'size': 1024, 'mtime': 'Mon Mar 10 12:00:00 2025', 'permissions': '644'},
{'path': '/tmp/test/sub/file2.py', 'size': 2048, 'mtime': 'Mon Mar 10 12:05:00 2025', 'permissions': '755'}]Параллельная индексация с multiprocessing
Для ускорения обхода нескольких корневых каталогов можно распараллелить работу.
from multiprocessing import Pool
from pathlib import Path
def index_single(root: str) -> dict:
idx = {}
for i, f in enumerate(Path(root).rglob('*'), start=1):
if f.is_file():
idx[i] = str(f)
return idx
def parallel_index(roots: list) -> dict:
with Pool() as pool:
results = pool.map(index_single, roots)
combined = {}
counter = 1
for res in results:
for k, v in res.items():
combined[counter] = v
counter += 1
return combined
print(parallel_index(['/tmp/a', '/tmp/b'])){1: '/tmp/a/f1.txt', 2: '/tmp/a/f2.txt', 3: '/tmp/b/g1.txt', ...}Инвертированный индекс для полнотекстового поиска
Этот метод создаёт словарь, где каждому слову сопоставлен список файлов, в которых оно встречается. Подходит для поиска по содержимому.
from collections import defaultdict
import re
from pathlib import Path
def build_inverted_index(dir_path: str) -> dict:
index = defaultdict(list)
for file in Path(dir_path).rglob('*.txt'):
if file.is_file():
with open(file, 'r', encoding='utf-8', errors='ignore') as f:
words = set(re.findall(r'\w+', f.read().lower()))
for word in words:
index[word].append(str(file))
return dict(index)
inv = build_inverted_index('/tmp/docs')
print('собака' in inv)True
Примечание: такой индекс может быть большим. Для реальных приложений лучше использовать специализированные библиотеки, например Whoosh или Elasticsearch.
Индекс с помощью Pandas DataFrame
Если требуется дальнейший анализ (статистика размеров, группировка по папкам), удобно представить индекс как DataFrame.
import pandas as pd
from pathlib import Path
def index_to_dataframe(base: str) -> pd.DataFrame:
data = []
for file in Path(base).rglob('*'):
if file.is_file():
stat = file.stat()
data.append({
'id': len(data) + 1,
'path': str(file),
'size': stat.st_size,
'parent': str(file.parent)
})
return pd.DataFrame(data)
df = index_to_dataframe('/tmp/test')
print(df.describe())id size count 5.000000 5.000000 mean 3.000000 1536.000000 std 1.581139 820.487000 min 1.000000 512.000000 25% 2.000000 1024.000000 50% 3.000000 1536.000000 75% 4.000000 2048.000000 max 5.000000 2560.000000
DataFrame позволяет легко фильтровать, сортировать и визуализировать данные.