Эффективность работы разработчика часто измеряется не количеством написаных строк, а скоростью решения задач. Готовые сниппеты значительно ускоряют разработку, помогают избежать типичных ошибок и держать код в соотвествии с лучшими практиками. Использование проверенных решений также обеспечивает единообразие кодовой базы, что особенно важно при командной работе.
В Python множество типичных задач можно решить стандартными средствами языка. Но даже опытные разработчики не всегда помнят все нюансы синтаксиса или самые эффективные подходы к решению определенных проблем. В этой статье я собрал 50 наиболее полезных сниппетов Python, которые регулярно применяю в своих проектах. Некоторые из них могут показаться очевидными для опытных питонистов, но я уверен, что даже они найдут здесь что-то новое или вспомнят забытое. Каждый пример сопровождается пояснениями и контекстом применения, чтобы вы понимали не только "как", но и "почему".
Работа со строками и текстом
Строки - наверное, самый часто используемый тип данных в Python. На первый взгляд, работа с ними кажется тривиальной, но часто приходится прибегать к нетривиальным приемам. Я собрал наиболее полезные сниппеты, которые используются чаще всего.
Переворот строки
Начнем с базовой операции - переворота строки. В отличие от многих языков, в Python это делается элегантно через слайсы:
| Python | 1
2
3
4
5
6
| def reverse_string(s):
return s[::-1]
# Пример использования
original = "Python"
reversed_str = reverse_string(original) # "nohtyP" |
|
[::-1] я использую почти каждый день. Он работает значительно быстрее, чем итерационные методы типа ''.join(reversed(s)), что доказано бенчмарками. Кстати, тот же метод применим и к спискам.
Проверка на палиндром
Палиндром - это строка, которая читается одинаково в обоих направлениях. Используя предыдущий сниппет, можно легко проверить это свойство:
| Python | 1
2
3
4
5
6
7
8
| def is_palindrome(s):
# Предварительная обработка: приведение к нижнему регистру и удаление пробелов
s = s.lower().replace(" ", "")
return s == s[::-1]
# Примеры
print(is_palindrome("А роза упала на лапу Азора")) # True
print(is_palindrome("Python")) # False |
|
Подсчет вхождений подстроки
Часто нужно узнать, сколько раз определенная подстрока встречается в тексте:
| Python | 1
2
3
4
5
6
| def count_substring(text, sub):
return text.count(sub)
# Пример
text = "Python это мощный язык. Python используется везде."
count = count_substring(text, "Python") # 2 |
|
count() очень эффективен, так как реализован на C уровне. Однако он чувствителен к регистру. Если нужен поиск без учета регистра, можно сделать так:
| Python | 1
2
| def count_substring_case_insensitive(text, sub):
return text.lower().count(sub.lower()) |
|
Замена подстроки
Замена текста - еще одна распространенная операция:
| Python | 1
2
3
4
5
6
| def replace_substring(text, old, new):
return text.replace(old, new)
# Пример
text = "Python - интерпретируемый язык"
new_text = replace_substring(text, "интерпретируемый", "динамический") |
|
re:
| Python | 1
2
3
4
5
6
7
8
9
| import re
def replace_with_regex(text, pattern, replacement):
return re.sub(pattern, replacement, text)
# Пример: заменяем все цифры на X
text = "Моя версия Python 3.9 работает на Windows 10"
masked = replace_with_regex(text, r'\d', 'X')
# "Моя версия Python X.X работает на Windows XX" |
|
Разбиение строки на список
Превращение строки в список - базовая операция для дальнейшей обработки данных:
| Python | 1
2
3
4
5
6
| def split_string(text, delimiter=" "):
return text.split(delimiter)
# Пример
text = "Python,Java,C++,JavaScript"
languages = split_string(text, ",") # ["Python", "Java", "C++", "JavaScript"] |
|
re.split():
| Python | 1
2
3
4
5
6
7
8
9
10
| import re
def split_by_multiple_delimiters(text, delimiters):
pattern = '|'.join(map(re.escape, delimiters))
return re.split(pattern, text)
# Пример
text = "Python;Java,C++:JavaScript"
languages = split_by_multiple_delimiters(text, [";", ",", ":"])
# ["Python", "Java", "C++", "JavaScript"] |
|
Форматирование строк с f-строками
F-строки появились в Python 3.6 и стали моим любимым способом форматирования:
| Python | 1
2
3
4
5
6
| def format_with_fstring(name, age):
return f"Меня зовут {name} и мне {age} лет."
# Пример
message = format_with_fstring("Алексей", 30)
# "Меня зовут Алексей и мне 30 лет." |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
| items = ["яблоки", "груши", "апельсины"]
quantities = [5, 3, 2]
# Вывод таблицы с выравниванием и вычислениями
for i, (item, qty) in enumerate(zip(items, quantities)):
print(f"{i+1}. {item:<10} x {qty:>2} = {qty * 10:>3} руб.")
# Вывод:
# 1. яблоки x 5 = 50 руб.
# 2. груши x 3 = 30 руб.
# 3. апельсины x 2 = 20 руб. |
|
:<10 для выравнивания по левому краю, :>2 для выравнивания по правому) и встроенные вычисления.
Проверка начала и конца строки
Часто в коде приходится проверять, начинается или заканчивается ли строка определенной подстрокой. Вместо ручного сравнения срезов, в Python есть встроенные методы:
| Python | 1
2
3
4
5
6
7
8
| def check_string_boundaries(text, prefix, suffix):
starts_with = text.startswith(prefix)
ends_with = text.endswith(suffix)
return starts_with, ends_with
# Пример
text = "requirements.txt"
is_req_file = check_string_boundaries(text, "", ".txt")[1] # True |
|
Объединение строк
Для объединения списка строк в одну я предпочитаю метод join(), который работает намного эффективнее, чем конкатенация с помощью оператора +:
| Python | 1
2
3
4
5
6
7
8
9
10
| def join_strings(strings, delimiter):
return delimiter.join(strings)
# Пример
words = ["Python", "это", "удобно"]
sentence = join_strings(words, " ") # "Python это удобно"
# Создание CSV строки
data = ["имя", "возраст", "город"]
csv_header = join_strings(data, ",") # "имя,возраст,город" |
|
join и конкатенацией через + может достигать порядка и более, поскольку строки в Python неизменяемы, и при каждой конкатенации создается новый объект.
Удаление лишних пробелов
Данные, полученные от пользователя или из файлов, часто содержат лишние пробелы:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
| def clean_string(text):
return text.strip()
# Более полезная версия для очистки всех пробельных символов внутри строки
def normalize_whitespace(text):
import re
return re.sub(r'\s+', ' ', text).strip()
# Примеры
user_input = " Python программирование "
clean = clean_string(user_input) # "Python программирование"
normalized = normalize_whitespace(user_input) # "Python программирование" |
|
Выравнивание текста
Для создания читаемых таблиц или форматированного вывода удобно использовать методы выравнивания:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
| def align_text(text, width, alignment='left'):
if alignment == 'left':
return text.ljust(width)
elif alignment == 'right':
return text.rjust(width)
elif alignment == 'center':
return text.center(width)
return text
# Пример: создание простой таблицы
header = align_text("Имя", 15) + align_text("Возраст", 10, 'right')
print(header)
print("-" * 25)
print(align_text("Алексей", 15) + align_text("30", 10, 'right'))
print(align_text("Екатерина", 15) + align_text("25", 10, 'right'))
# Вывод:
# Имя Возраст
# -------------------------
# Алексей 30
# Екатерина 25 |
|
tabulate или prettytable, но для простых случаев встроенных методов вполне достаточно.
Нахождение самых частых триграмм, содержащих букву А
Здравствуйте, нужна помощь с решением задачи из егэ на языке программирования python, вот условие:... Найти сумму 5 самых больших и 5 самых маленьких элементов списка
Две список целых чисел, состоящий из 30 элементов.Найти сумму 5 самых больших и 5 самых маленьких... Выведите произведение трёх самых больших элементов последовательности и трёх самых маленьких элементов последовательно
Пользователь вводит последовательность из 12 чисел, переверните
последовательность в обратном... Нахождение 10 наиболее частых слов на web странице
Ребят, выручайте, через 4 дня защита. Указывает на ошибку в последней строке. Питон версии 3.2 что...
Манипуляции со списками и словарями
Списки и словари — это фундаментальные структуры данных в Python, с которыми я работаю ежедневно. Они предоставляют широкие возможности для хранения и обработки информации. Давайте рассмотрим наиболее полезные операции с этими структурами.
Список с условием (List Comprehension)
List comprehension — один из самых мощных инструментов Python, позволяющий создавать списки одной строкой кода:
| Python | 1
2
3
4
5
6
| def filtered_squares():
# Получаем квадраты только четных чисел
squares = [x**2 for x in range(10) if x % 2 == 0]
return squares
# Результат: [0, 4, 16, 36, 64] |
|
Сглаживание вложенного списка
Вложенные списки — частое явление, но иногда нужно "сгладить" их до одного уровня:
| Python | 1
2
3
4
5
6
7
| def flatten_list(nested_list):
# Простой вариант для списка с одним уровнем вложенности
return [item for sublist in nested_list for item in sublist]
# Пример
nested = [[1, 2], [3, 4], [5]]
flat = flatten_list(nested) # [1, 2, 3, 4, 5] |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
| def deep_flatten(items):
result = []
for item in items:
if isinstance(item, list):
result.extend(deep_flatten(item))
else:
result.append(item)
return result
# Пример
multi_level = [1, [2, [3, 4], 5], 6]
flat = deep_flatten(multi_level) # [1, 2, 3, 4, 5, 6] |
|
Удаление дубликатов из списка
Превращение списка в множество (set) и обратно — самый быстрый способ избавиться от дубликатов:
| Python | 1
2
3
4
5
6
| def remove_duplicates(items):
return list(set(items))
# Пример
duplicate_list = [1, 2, 2, 3, 4, 3, 5]
unique_items = remove_duplicates(duplicate_list) # [1, 2, 3, 4, 5] |
|
| Python | 1
2
3
4
5
6
7
| def remove_duplicates_preserve_order(items):
seen = set()
return [x for x in items if not (x in seen or seen.add(x))]
# Пример
ordered_no_dupes = remove_duplicates_preserve_order([1, 2, 2, 3, 4, 3, 5])
# [1, 2, 3, 4, 5] |
|
seen.add(x) заключается в том, что он возвращает None, поэтому условие срабатывает только когда элемент встречается впервые.
Переворот списка
Аналогично строкам, списки можно перевернуть с помощью среза:
| Python | 1
2
3
4
5
6
| def reverse_list(items):
return items[::-1]
# Пример
original = [1, 2, 3, 4, 5]
reversed_list = reverse_list(original) # [5, 4, 3, 2, 1] |
|
reverse(), но он изменяет список на месте:
| Python | 1
2
| items = [1, 2, 3, 4, 5]
items.reverse() # items теперь [5, 4, 3, 2, 1] |
|
Сортировка списка словарей по ключу
При работе с данными часто требуется сортировка списка объектов по определенному полю:
| Python | 1
2
3
4
5
6
7
8
9
10
11
| def sort_dicts_by_key(dict_list, key_name):
return sorted(dict_list, key=lambda x: x[key_name])
# Пример
users = [
{'name': 'Петр', 'age': 30},
{'name': 'Иван', 'age': 25},
{'name': 'Мария', 'age': 28}
]
sorted_by_age = sort_dicts_by_key(users, 'age')
# [{'name': 'Иван', 'age': 25}, {'name': 'Мария', 'age': 28}, {'name': 'Петр', 'age': 30}] |
|
itemgetter из модуля operator:
| Python | 1
2
3
4
| from operator import itemgetter
# Сортировка по нескольким ключам
sorted_users = sorted(users, key=itemgetter('age', 'name')) |
|
Объединение словарей
До Python 3.9 объединить словари можно было несколькими способами:
| Python | 1
2
3
4
5
6
7
8
9
| def merge_dicts(dict1, dict2):
result = dict1.copy() # Создаем копию, чтобы не изменять оригинал
result.update(dict2)
return result
# Пример
user_info = {'name': 'Анна', 'age': 28}
additional_info = {'city': 'Москва', 'email': 'anna@example.com'}
complete_info = merge_dicts(user_info, additional_info) |
|
| для объединения словарей:
| Python | 1
2
| # Python 3.9+
complete_info = user_info | additional_info |
|
Инвертирование словаря
Иногда нужно поменять местами ключи и значения:
| Python | 1
2
3
4
5
6
| def invert_dict(d):
return {v: k for k, v in d.items()}
# Пример
country_code = {'Россия': 'RU', 'США': 'US', 'Германия': 'DE'}
code_country = invert_dict(country_code) # {'RU': 'Россия', 'US': 'США', 'DE': 'Германия'} |
|
| Python | 1
2
3
4
5
6
7
8
9
10
| def invert_dict_with_lists(d):
result = {}
for k, v in d.items():
result.setdefault(v, []).append(k)
return result
# Пример
fruit_colors = {'яблоко': 'красный', 'банан': 'желтый', 'вишня': 'красный'}
colors_fruits = invert_dict_with_lists(fruit_colors)
# {'красный': ['яблоко', 'вишня'], 'желтый': ['банан']} |
|
setdefault() создает ключ со значением по умолчанию, если его еще нет в словаре, и возвращает значение этого ключа.
Создание словаря из двух списков
Часто возникает задача создать словарь из двух отдельных списков, где один содержит ключи, а другой - значения:
| Python | 1
2
3
4
5
6
7
| def dict_from_lists(keys, values):
return dict(zip(keys, values))
# Пример
names = ['Иван', 'Мария', 'Алексей']
ages = [25, 30, 28]
name_to_age = dict_from_lists(names, ages) # {'Иван': 25, 'Мария': 30, 'Алексей': 28} |
|
zip() создает итератор, который объединяет элементы из нескольких итерируемых объектов. Затем конструктор dict() превращает пары в словарь. Если списки имеют разную длину, результирующий словарь будет ограничен длиной самого короткого списка.
Подсчет частоты элементов
Для анализа данных часто нужно подсчитать, сколько раз каждый элемент встречается в коллекции:
| Python | 1
2
3
4
5
6
7
8
9
10
11
| from collections import Counter
def count_frequencies(items):
return Counter(items)
# Пример
words = ['apple', 'orange', 'apple', 'banana', 'orange', 'apple']
word_counts = count_frequencies(words) # Counter({'apple': 3, 'orange': 2, 'banana': 1})
# Получение наиболее часто встречающихся элементов
most_common = word_counts.most_common(2) # [('apple', 3), ('orange', 2)] |
|
Counter - невероятно полезный класс, который я применяю почти в каждом проекте обработки данных. Он наследуется от dict, поэтому с ним можно работать как с обычным словарем, но с дополнительными методами.
Сортировка словаря по значению
Словари в Python сохраняют порядок добавления ключей (с версии 3.7), но иногда требуется отсортировать их по значениям:
| Python | 1
2
3
4
5
6
7
8
9
10
11
| def sort_dict_by_value(d):
return dict(sorted(d.items(), key=lambda item: item[1]))
# Пример
scores = {'Иван': 85, 'Мария': 92, 'Петр': 78, 'Анна': 95}
sorted_scores = sort_dict_by_value(scores)
[H2]{'Петр': 78, 'Иван': 85, 'Мария': 92, 'Анна': 95}[/H2]
# Сортировка по убыванию
sorted_desc = dict(sorted(scores.items(), key=lambda item: item[1], reverse=True))
# {'Анна': 95, 'Мария': 92, 'Иван': 85, 'Петр': 78} |
|
Фильтрация словаря
Отфильтровать словарь по определенному условию можно с помощью словарного включения (dict comprehension):
| Python | 1
2
3
4
5
6
7
8
9
10
| def filter_dict(d, condition):
return {k: v for k, v in d.items() if condition(k, v)}
# Пример: оставляем только элементы со значением > 80
high_scores = filter_dict(scores, lambda k, v: v > 80)
[H2]{'Иван': 85, 'Мария': 92, 'Анна': 95}[/H2]
# Фильтрация по ключу
starts_with_a = filter_dict(scores, lambda k, v: k.startswith('А'))
# {'Анна': 95} |
|
Генераторы и итераторы для ленивых вычислений
Генераторы в Python - одна из тех функций, которые кардинально меняют подход к обработке данных. Я убедился в этом, когда столкнулся с задачей обработки логов размером в несколько гигабайт - обычные списки просто не справлялись с нагрузкой, а генераторы спасли ситуацию. Суть генераторов в том, что они вычисляют значения "на лету", не храня всю последовательность в памяти. Это идеально для работы с большими наборами данных.
Создание простого генератора
| Python | 1
2
3
4
5
6
7
8
| def squares_generator(n):
for i in range(n):
yield i**2
# Использование генератора
gen = squares_generator(5)
print(next(gen)) # 0
print(next(gen)) # 1 |
|
yield приостанавливает выполнение функции и возвращает значение. При следующем вызове функция продолжит работу с того места, где остановилась. Это очень эффективно с точки зрения использования памяти.
Генераторные выражения
Компактная альтернатива функциям-генераторам:
| Python | 1
2
3
4
5
6
7
8
9
| # Генераторное выражение (отличается от списковых включений круглыми скобками)
squares = (x**2 for x in range(1000000))
# Никаких проблем с памятью, даже для больших диапазонов
for i, square in enumerate(squares):
if i < 5:
print(square)
else:
break |
|
Цепочки генераторов
Генераторы можно объединять в цепочки для создания конвейеров обработки данных:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| def read_large_file(file_path):
with open(file_path, 'r') as file:
for line in file:
yield line.strip()
def grep(lines, pattern):
for line in lines:
if pattern in line:
yield line
def process_logs(file_path, pattern):
lines = read_large_file(file_path)
matches = grep(lines, pattern)
for match in matches:
# Обработка происходит построчно, без загрузки всего файла
yield f"FOUND: {match}"
# Использование:
for result in process_logs("huge_log.txt", "ERROR"):
print(result)
# Прервем после первых 10 ошибок
if some_counter > 10:
break |
|
Бесконечные последовательности
Генераторы позволяют создавать бесконечные последовательности без риска исчерпания памяти:
| Python | 1
2
3
4
5
6
7
8
9
10
| def infinite_sequence():
num = 0
while True:
yield num
num += 1
# Безопасное использование бесконечного генератора
counter = infinite_sequence()
for _ in range(5):
print(next(counter)) # 0, 1, 2, 3, 4 |
|
Собственные итерируемые объекты
Создание класса, который ведет себя как итератор:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| class Fibonacci:
def __init__(self, limit):
self.limit = limit
self.a, self.b = 0, 1
self.count = 0
def __iter__(self):
# Сбрасываем состояние при новой итерации
self.a, self.b = 0, 1
self.count = 0
return self
def __next__(self):
if self.count >= self.limit:
raise StopIteration
result = self.a
self.a, self.b = self.b, self.a + self.b
self.count += 1
return result
# Использование
for num in Fibonacci(8):
print(num, end=" ") # 0 1 1 2 3 5 8 13 |
|
Ленивая фильтрация с itertools
Модуль itertools предоставляет множество инструментов для работы с итераторами:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
| import itertools
# Объединение итераторов
combined = itertools.chain(range(3), range(10, 13))
print(list(combined)) # [0, 1, 2, 10, 11, 12]
# Выбор первых N элементов из бесконечного генератора
limited = itertools.islice(infinite_sequence(), 5)
print(list(limited)) # [0, 1, 2, 3, 4]
# Цикличный итератор
cycled = itertools.cycle(['A', 'B', 'C'])
for _ in range(5):
print(next(cycled), end=" ") # A B C A B |
|
itertools для решения сложных задач обработки данных. Эта библиотека - настоящая сокровищница для работы с итераторами. Ленивые вычисления с генераторами стали для меня незаменимым инструментом при работе с большими данными. Они позволяют создавать чистый, эффективный код, который работает даже с ограничеными ресурсами. И главное - использование генераторов часто делает код более понятным, выражая намерение программиста яснее, чем сложные циклы с промежуточными переменными.
Файловые операции и обработка данных
Работа с файлами - неотъемлемая часть большинства проектов. За годы работы я собрал коллекцию надежных сниппетов для эффективной обработки файлов разных форматов. Давайте начнем с базовых операций и перейдем к более специализированным.
Безопасное чтение файла
Всегда используйте контекстный менеджер with для работы с файлами - он гарантирует закрытие файла даже при возникновении исключений:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
| def read_file(file_path):
try:
with open(file_path, 'r', encoding='utf-8') as file:
return file.read()
except FileNotFoundError:
print(f"Файл {file_path} не найден")
return None
except Exception as e:
print(f"Ошибка при чтении файла: {e}")
return None
# Пример использования
content = read_file("example.txt") |
|
encoding='utf-8', чтобы избежать проблем с символами, особенно при работе на разных платформах. Однажды потратил целый день на отладку, пока не обнаружил, что на Windows и Linux файлы открывались с разной кодировкой по умолчанию.
Построчное чтение больших файлов
Для больших файлов нецелесообразно загружать весь контент в память:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| def process_large_file(file_path):
try:
with open(file_path, 'r', encoding='utf-8') as file:
for line in file:
# Обработка каждой строки
processed_line = line.strip().upper()
# Здесь может быть любая логика
yield processed_line
except Exception as e:
print(f"Ошибка обработки файла: {e}")
# Пример использования
for processed_line in process_large_file("large_log.txt"):
if "ERROR" in processed_line:
print(processed_line) |
|
Запись и добавление в файл
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
| def write_to_file(file_path, content, mode='w'):
try:
with open(file_path, mode, encoding='utf-8') as file:
file.write(content)
return True
except Exception as e:
print(f"Ошибка при записи в файл: {e}")
return False
# Запись в новый файл
write_to_file("output.txt", "Привет, мир!")
# Добавление в существующий файл
write_to_file("output.txt", "\nВторая строка", mode='a') |
|
mode контролирует режим открытия файла: 'w' для перезаписи, 'a' для добавления в конец.
Работа с CSV файлами
CSV - один из самых распространенных форматов для обмена данными:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
| import csv
def read_csv(file_path, delimiter=','):
results = []
try:
with open(file_path, 'r', encoding='utf-8', newline='') as file:
reader = csv.DictReader(file, delimiter=delimiter)
for row in reader:
results.append(dict(row))
return results
except Exception as e:
print(f"Ошибка при чтении CSV: {e}")
return []
def write_csv(file_path, data, fieldnames, delimiter=','):
try:
with open(file_path, 'w', encoding='utf-8', newline='') as file:
writer = csv.DictWriter(file, fieldnames=fieldnames, delimiter=delimiter)
writer.writeheader()
writer.writerows(data)
return True
except Exception as e:
print(f"Ошибка при записи CSV: {e}")
return False
# Пример использования
users = read_csv("users.csv")
# Добавление нового пользователя
users.append({"name": "Иван", "age": "30", "city": "Москва"})
# Запись обновленных данных
write_csv("users_updated.csv", users, ["name", "age", "city"]) |
|
newline='' при работе с CSV, это предотвращает проблемы с переносами строк на разных платформах.
Обработка JSON данных
JSON стал стандартом для API и конфигурационных файлов:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
| import json
def read_json(file_path):
try:
with open(file_path, 'r', encoding='utf-8') as file:
return json.load(file)
except Exception as e:
print(f"Ошибка при чтении JSON: {e}")
return None
def write_json(file_path, data, indent=4):
try:
with open(file_path, 'w', encoding='utf-8') as file:
json.dump(data, file, ensure_ascii=False, indent=indent)
return True
except Exception as e:
print(f"Ошибка при записи JSON: {e}")
return False
# Пример использования
config = read_json("config.json")
if config:
# Изменение конфигурации
config["debug_mode"] = True
write_json("config_updated.json", config) |
|
ensure_ascii=False - он позволяет корректно сохранять Unicode символы, включая кириллицу, без экранирования.
Управление путями к файлам
Модуль pathlib значительно упрощает работу с путями к файлам:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| from pathlib import Path
def process_directory(directory_path, extension=".txt"):
dir_path = Path(directory_path)
if not dir_path.exists() or not dir_path.is_dir():
print(f"Директория {directory_path} не существует")
return []
# Поиск файлов с указанным расширением
matching_files = list(dir_path.glob(f"*{extension}"))
return matching_files
# Пример: обработка всех .log файлов в директории
log_files = process_directory("/var/log", ".log")
for log_file in log_files:
print(f"Обрабатываю {log_file.name}, размер: {log_file.stat().st_size} байт") |
|
pathlib появился в Python 3.4 и стал одним из моих любимых модулей. Он заменяет множество функций из os.path и предоставляет более интуитивный интерфейс.
Асинхронная обработка файлов с использованием aiofiles
Когда я начал писать асинхронный код на Python, то быстро столкнулся с проблемой: стандартные операции с файлами блокируют выполнение программы. В асинхронном мире это неприемлемо. На помощь приходит библиотека aiofiles, которая предоставляет асинхронный интерфейс для операций ввода-вывода.
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| import asyncio
import aiofiles
async def read_file_async(file_path):
try:
async with aiofiles.open(file_path, 'r', encoding='utf-8') as file:
return await file.read()
except Exception as e:
print(f"Ошибка при чтении файла: {e}")
return None
async def write_file_async(file_path, content):
try:
async with aiofiles.open(file_path, 'w', encoding='utf-8') as file:
await file.write(content)
return True
except Exception as e:
print(f"Ошибка при записи в файл: {e}")
return False |
|
async/await. Однако реальная мощь проявляется при параллельной обработке множества файлов:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
| async def process_multiple_files(file_paths):
tasks = [read_file_async(path) for path in file_paths]
results = await asyncio.gather(*tasks)
return dict(zip(file_paths, results))
# Использование
async def main():
files_to_process = ["file1.txt", "file2.txt", "file3.txt"]
results = await process_multiple_files(files_to_process)
print(f"Обработано {len(results)} файлов")
# Запуск асинхронной функции
asyncio.run(main()) |
|
aiofiles не делает сами дисковые операции быстрее. Он просто позволяет вашему коду не простаивать во время ожидания ввода-вывода, занимаясь другими задачами. Для действительно параллельной записи/чтения физических файлов всё ещё требуется многопроцессорный подход из-за ограничений GIL в Python.
При массовой асинхронной записи я столкнулся с проблемой перегрузки диска. Решение - добавить семафор для ограничения количества одновременных операций:
| Python | 1
2
3
4
5
6
7
8
9
| async def controlled_file_processing(file_paths, max_concurrent=10):
semaphore = asyncio.Semaphore(max_concurrent)
async def process_with_limit(path):
async with semaphore:
return await read_file_async(path)
tasks = [process_with_limit(path) for path in file_paths]
return await asyncio.gather(*tasks) |
|
Мониторинг изменений в файловой системе через watchdog
Часто возникает необходимость отслеживать изменения в файлах или директориях - например, для автоматической обработки загруженных логов или перезагрузки конфигурации. Библиотека watchdog предоставляет кроссплатформенное решение этой задачи, избавляя от необходимости постоянно опрашивать файловую систему. Вот базовый пример мониторинга директории:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
| import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class MyHandler(FileSystemEventHandler):
def on_modified(self, event):
if not event.is_directory:
print(f"Файл изменен: {event.src_path}")
def on_created(self, event):
if not event.is_directory:
print(f"Создан новый файл: {event.src_path}")
def watch_directory(path):
event_handler = MyHandler()
observer = Observer()
observer.schedule(event_handler, path, recursive=True)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
# Использование
watch_directory("/path/to/monitor") |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
| from watchdog.events import PatternMatchingEventHandler
class LogFileHandler(PatternMatchingEventHandler):
patterns = ["*.log"]
def on_created(self, event):
print(f"Новый лог-файл: {event.src_path}")
# Здесь можно добавить автоматическую обработку
process_new_log(event.src_path)
def process_new_log(file_path):
# Логика обработки нового лог-файла
pass |
|
watchdog я сталкивался с проблемой: в некоторых системах (особенно в сетевых файловых системах) событие модификации может вызываться несколько раз для одного изменения. Решение - добавить дебаунсинг:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| import time
class DebouncedHandler(FileSystemEventHandler):
def __init__(self, delay=1.0):
self.last_modified = {}
self.delay = delay
def on_modified(self, event):
if not event.is_directory:
current_time = time.time()
last_time = self.last_modified.get(event.src_path, 0)
if (current_time - last_time) > self.delay:
print(f"Файл изменен: {event.src_path}")
self.last_modified[event.src_path] = current_time |
|
Параллельные вычисления с multiprocessing и concurrent.futures
Python из-за GIL (Global Interpreter Lock) не может эффективно использовать многопоточность для CPU-интенсивных задач. Но это не значит, что мы лишены возможности распараллеливать вычисления. Для таких случаев у нас есть модули multiprocessing и concurrent.futures, позволяющие задействовать все ядра процессора.
Я долго сопротивлялся использованию многопроцессорности в своих проектах, опасаясь сложности, но однажды мне пришлось обрабатывать массив из миллиона изображений. Последовательная обработка заняла бы дни, а с параллельным подходом управился за несколько часов.
Базовое использование multiprocessing
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
| from multiprocessing import Pool
def process_item(item):
# Какая-то вычислительно сложная операция
return item * item
def parallel_processing(items, processes=None):
with Pool(processes) as pool:
results = pool.map(process_item, items)
return results
# Пример использования
data = list(range(1000))
result = parallel_processing(data, processes=4) |
|
Pool создает группу рабочих процессов и распределяет задачи между ними. Параметр processes определяет количество процессов - если не указан, используется количество, равное числу ядер CPU.
Продвинутый подход с concurrent.futures
Модуль concurrent.futures, появившийся в Python 3.2, предоставляет более высокоуровневый интерфейс:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
| import concurrent.futures
import math
def is_prime(n):
if n < 2:
return False
for i in range(2, int(math.sqrt(n)) + 1):
if n % i == 0:
return False
return True
def find_primes(numbers):
primes = []
# ProcessPoolExecutor для CPU-bound задач
with concurrent.futures.ProcessPoolExecutor() as executor:
# Запускаем задачи асинхронно
future_to_number = {executor.submit(is_prime, num): num for num in numbers}
# Обрабатываем результаты по мере их готовности
for future in concurrent.futures.as_completed(future_to_number):
number = future_to_number[future]
try:
if future.result():
primes.append(number)
except Exception as e:
print(f"Ошибка при обработке {number}: {e}")
return sorted(primes)
# Проверяем числа от 1 до 100000
large_range = range(1, 100001)
primes = find_primes(large_range)
print(f"Найдено {len(primes)} простых чисел") |
|
1. Обработка результатов по мере готовности через as_completed().
2. Более удобная обработка исключений.
3. Простое переключение между процессами и потоками (для IO-bound задач достаточно заменить .ProcessPoolExecutor на ThreadPoolExecutor).
Передача дополнительных аргументов
Часто функции обработки нужны дополнительные параметры:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| def process_with_config(item, threshold, multiplier):
if item > threshold:
return item * multiplier
return item
def parallel_map_with_args(func, items, processes=None, **kwargs):
with concurrent.futures.ProcessPoolExecutor(max_workers=processes) as executor:
# Используем partial для фиксации дополнительных аргументов
from functools import partial
fn = partial(func, **kwargs)
return list(executor.map(fn, items))
# Пример использования
result = parallel_map_with_args(process_with_config, range(100),
threshold=50, multiplier=2) |
|
dill или cloudpickle.
Эти инструменты параллельных вычислений стали неотъемлемой частью моего арсенала при работе с большими объемами данных. Конечно, добавление параллелизма усложняет код и отладку, но производительность часто стоит этих усилий, особенно когда речь идет о сокращении времени выполнения с часов до минут.
Сетевые запросы и API
Современная разработка уже немыслима без взаимодействия с внешними сервисами через API. За свою карьеру я перепробовал кучу подходов к работе с сетью в Python - от стандартного urllib до специализированных клиентских библиотек. В итоге сформировался набор шаблонов, которые использую регулярно.
Базовые HTTP запросы с requests
Библиотека requests - настоящая находка для разработчика. Она делает HTTP-запросы интуитивно понятными:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| import requests
def get_data(url, params=None, headers=None):
try:
response = requests.get(url, params=params, headers=headers, timeout=10)
response.raise_for_status() # Вызовет исключение при статусе 4XX/5XX
return response.json()
except requests.exceptions.RequestException as e:
print(f"Ошибка запроса: {e}")
return None
# Пример использования
api_url = "https://api.example.com/data"
params = {"limit": 10, "offset": 0}
headers = {"Authorization": "Bearer your_token_here"}
data = get_data(api_url, params, headers) |
|
timeout, чтобы избежать зависания программы при проблемах с сетью. Однажды потерял два часа, выясняя, почему скрипт "завис" - оказалось, что без таймаута запрос к недоступному серверу просто ждал вечно.
POST запросы и отправка данных
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
| def post_data(url, json_data=None, headers=None):
try:
response = requests.post(url, json=json_data, headers=headers, timeout=10)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"Ошибка при отправке данных: {e}")
return None
# Пример: создание нового пользователя
user_data = {"name": "Иван", "email": "ivan@example.com"}
result = post_data("https://api.example.com/users", json_data=user_data) |
|
json= вместо data=. Он автоматически сериализует Python-словарь в JSON и устанавливает правильный заголовок Content-Type.
Асинхронные запросы с aiohttp
Когда нужно выполнить множество запросов одновременно, асинхронный подход значительно ускоряет работу:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
| import asyncio
import aiohttp
async def fetch_url(session, url):
try:
async with session.get(url, timeout=10) as response:
if response.status == 200:
return await response.json()
return None
except Exception as e:
print(f"Ошибка при запросе {url}: {e}")
return None
async def fetch_multiple_urls(urls):
async with aiohttp.ClientSession() as session:
tasks = [fetch_url(session, url) for url in urls]
results = await asyncio.gather(*tasks)
return results
# Использование
urls = [
"https://api.example.com/data/1",
"https://api.example.com/data/2",
"https://api.example.com/data/3"
]
data = asyncio.run(fetch_multiple_urls(urls)) |
|
Обработка ошибок и повторные попытки
API могут быть нестабильными, поэтому всегда стоит реализовать механизм повторных попыток:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| def get_with_retry(url, max_retries=3, backoff_factor=0.5):
import time
from requests.exceptions import RequestException
retries = 0
while retries < max_retries:
try:
response = requests.get(url, timeout=10)
response.raise_for_status()
return response.json()
except RequestException as e:
wait = backoff_factor * (2 ** retries)
print(f"Попытка {retries+1} неудачна, ожидание {wait:.2f}c: {e}")
time.sleep(wait)
retries += 1
print(f"Не удалось получить данные после {max_retries} попыток")
return None |
|
backoff_factor * (2 ** retries)) помогает избежать перегрузки сервера и дает ему время восстановиться.
Загрузка файлов
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
| def download_file(url, local_path):
try:
with requests.get(url, stream=True, timeout=30) as response:
response.raise_for_status()
with open(local_path, 'wb') as file:
for chunk in response.iter_content(chunk_size=8192):
file.write(chunk)
return True
except Exception as e:
print(f"Ошибка при загрузке файла: {e}")
return False
# Пример: загрузка изображения
success = download_file("https://example.com/image.jpg", "local_image.jpg") |
|
stream=True и использование iter_content() критически важны при загрузке больших файлов, так как позволяют избежать загрузки всего файла в память.
Эти сниппеты стали основой для большинства моих проектов, взаимодействующих с внешними API. Они достаточно универсальны и легко адаптируются под конкретные нужды, будь то работа с RESTful API, GraphQL или веб-скрапинг.
Алгоритмические задачи
В работе программиста периодически возникают классические алгоритмические задачи, которые уже имеют проверенные временем решения. Я заметил, что многие разработчики часто изобретают велосипед вместо того, чтобы использовать оптимизированные алгоритмы. Поделюсь пятью наиболее полезными сниппетами, которые не раз спасали меня в проектах.
Вычисление факториала
Классическая задача, которая встречается во многих статистических и математических расчетах:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| import math
# Стандартный способ через библиотеку math
def factorial(n):
return math.factorial(n)
# Рекурсивная реализация (не рекомендуется для больших чисел)
def factorial_recursive(n):
if n <= 1:
return 1
return n * factorial_recursive(n - 1)
# Оптимизированная итеративная реализация
def factorial_iterative(n):
result = 1
for i in range(2, n + 1):
result *= i
return result |
|
Последовательность Фибоначчи
Еще одна классическая последовательность, часто используемая в алгоритмах:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| def fibonacci(n):
a, b = 0, 1
for _ in range(n):
yield a
a, b = b, a + b
# Получение n-го числа Фибоначчи эффективным способом
def fibonacci_n(n):
if n <= 0:
return 0
a, b = 0, 1
for _ in range(n - 1):
a, b = b, a + b
return b
# Пример использования
print(list(fibonacci(10))) # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
print(fibonacci_n(10)) # 55 |
|
Проверка на простое число
Оптимизированный алгоритм для проверки простоты числа:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| def is_prime(n):
if n <= 1:
return False
if n <= 3:
return True
if n % 2 == 0 or n % 3 == 0:
return False
# Проверяем только числа вида 6k ± 1 до sqrt(n)
i = 5
while i * i <= n:
if n % i == 0 or n % (i + 2) == 0:
return False
i += 6
return True |
|
Наибольший общий делитель
Алгоритм Евклида - элегантное решение для нахождения НОД:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| def gcd(a, b):
while b:
a, b = b, a % b
return a
# С использованием встроенной функции из модуля math
import math
print(math.gcd(48, 18)) # 6
# Расширенный алгоритм Евклида (возвращает НОД и коэффициенты Безу)
def extended_gcd(a, b):
if a == 0:
return b, 0, 1
gcd, x1, y1 = extended_gcd(b % a, a)
x = y1 - (b // a) * x1
y = x1
return gcd, x, y |
|
Эффективный поиск в массиве
Бинарный поиск - один из фундаментальных алгоритмов, который работает намного эффективнее линейного поиска на отсортированных данных:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
| def binary_search(arr, target):
left, right = 0, len(arr) - 1
while left <= right:
mid = left + (right - left) // 2 # Избегаем переполнения
if arr[mid] == target:
return mid
elif arr[mid] < target:
left = mid + 1
else:
right = mid - 1
return -1 # Элемент не найден |
|
left + (right - left) // 2 вместо обычной (left + right) // 2 защищает от переполнения при работе с большими массивами. Такая мелочь однажды спасла меня от трудноуловимого бага в промышленном коде.
Утилитарные функции
В арсенале каждого разработчика должны быть универсальные помощники - утилитарные функции, которые можно быстро подключить к проекту для решения типовых задач. Я постоянно использую несколько таких инструментов, которые существенно упрощают разработку и отладку.
Измерение времени выполнения кода
Отладка производительности - частая головная боль. Простой декоратор для измерения времени выполнения функций спасает в таких ситуациях:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| import time
import functools
def timer(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
print(f"Функция {func.__name__} выполнилась за {end_time - start_time:.4f} секунд")
return result
return wrapper
# Пример использования
@timer
def slow_function(n):
time.sleep(n) # Имитация долгой работы
return n * n
result = slow_function(2) # Функция slow_function выполнилась за 2.0021 секунд |
|
Проверка типов данных
При отладке или работе с незнакомым API бывает полезно узнать тип и структуру объекта:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| def inspect_object(obj, name="объект"):
obj_type = type(obj).__name__
print(f"{name} имеет тип: {obj_type}")
if hasattr(obj, "__len__"):
print(f"Длина: {len(obj)}")
if isinstance(obj, dict):
print("Ключи словаря:")
for key in obj.keys():
print(f" - {key}: {type(obj[key]).__name__}")
elif isinstance(obj, (list, tuple)):
print(f"Первые 3 элемента (из {len(obj)}):")
for i, item in enumerate(obj[:3]):
print(f" [{i}] {item} ({type(item).__name__})")
# Пример
data = {"users": [{"name": "Иван", "age": 30}, {"name": "Анна"}], "status": "ok"}
inspect_object(data, "response") |
|
Безопасное получение вложенных значений
Доступ к вложенным атрибутам объектов или вложенным ключам словарей часто приводит к ошибкам:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
| def safe_get(obj, path, default=None):
keys = path.split('.') if isinstance(path, str) else path
for key in keys:
try:
obj = obj[key] if isinstance(obj, dict) else getattr(obj, key)
except (KeyError, AttributeError, TypeError, IndexError):
return default
return obj
# Примеры
data = {"user": {"profile": {"name": "Иван", "contacts": {"email": "ivan@example.com"}}}}
email = safe_get(data, "user.profile.contacts.email") # "ivan@example.com"
phone = safe_get(data, "user.profile.contacts.phone", "Не указан") # "Не указан" |
|
Анализ производительности популярных подходов с бенчмарками
Когда я начал серьезно заниматься оптимизацией кода, то понял, что интуитивные представления о производительности нередко бывают ошибочными. Python - язык с множеством способов решения одной задачи, и выбор правильного подхода может дать огромный прирост производительности. Давайте сравним некоторые распространенные операции с помощью простых бенчмарков.
| Python | 1
2
3
4
5
6
7
8
| import timeit
import random
import statistics
def benchmark(func, setup="pass", number=10000):
times = timeit.repeat(func, setup, number=number, repeat=5)
print(f"{func:<50} {min(times):.6f} сек.")
return min(times) |
|
| Python | 1
2
3
| concat_plus = benchmark("s = ''; [s + str(i) for i in range(1000)]")
concat_join = benchmark("''.join(str(i) for i in range(1000))")
concat_fstring = benchmark("''.join(f'{i}' for i in range(1000))") |
|
join обычно в 10-15 раз быстрее, чем последовательное использование оператора +. Причина проста - строки в Python неизменяемы, и каждая операция + создает новый объект.
Еще одно интересное сравнение - различные способы фильтрации списка:
| Python | 1
2
3
4
5
6
7
8
9
10
11
| data = "random.sample(range(1000000), 10000)"
filter_loop = benchmark(
"[x for x in data if x % 2 == 0]",
setup=f"import random; data = {data}"
)
filter_built_in = benchmark(
"list(filter(lambda x: x % 2 == 0, data))",
setup=f"import random; data = {data}"
) |
|
filter() с lambda-функцией. Однако этот разрыв может меняться в зависимости от сложности фильтрующей функции. Особенно впечатляет разница при доступе к элементам словаря:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
| setup = """
data = {str(i): i for i in range(10000)}
keys = list(data.keys())
import random
random_keys = random.sample(keys, 1000)
"""
get_try_except = benchmark(
"""
result = []
for key in random_keys:
try:
result.append(data[key])
except KeyError:
result.append(None)
""", setup=setup
)
get_if_in = benchmark(
"""
result = []
for key in random_keys:
if key in data:
result.append(data[key])
else:
result.append(None)
""", setup=setup
)
get_method = benchmark(
"""
result = [data.get(key) for key in random_keys]
""", setup=setup
) |
|
dict.get() обычно работает примерно на 30-40% быстрее, чем явная проверка через if key in dict, и примерно в 2-3 раза быстрее, чем использование блока try-except.
Но не всегда стандартные методы лучше. Например, при сортировке списков с пользовательским ключом, использование operator.itemgetter() может быть на 10-15% быстрее lambda-функций:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| setup = """
import random
data = [{'id': i, 'value': random.random()} for i in range(10000)]
from operator import itemgetter
"""
sort_lambda = benchmark(
"sorted(data, key=lambda x: x['value'])",
setup=setup
)
sort_itemgetter = benchmark(
"sorted(data, key=itemgetter('value'))",
setup=setup
) |
|
Интеграция всех техник в финальном демонстрационном приложении для анализа логов
Объединим изученные техники в реальном приложении для анализа лог-файлов. Я решил создать такое приложение после того, как в одном из проектов пришлось вручную просматривать гигабайты логов в поисках причины сбоя. Потратив день на рутину, понял - такую работу должен делать скрипт. Наше приложение будет автоматически мониторить директорию, находить новые лог-файлы, анализировать их содержимое и генерировать отчеты с полезной информацией. Начнем с основной структуры:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
| import asyncio
import re
import os
import time
from pathlib import Path
from collections import Counter
from concurrent.futures import ProcessPoolExecutor
from functools import partial
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import aiofiles
# Настройки
LOG_DIR = Path("/var/log")
REPORT_DIR = Path("./reports")
ERROR_PATTERNS = [
r"ERROR.*",
r"CRITICAL.*",
r"EXCEPTION.*",
r"Failed to.*"
] |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
| class LogFileHandler(FileSystemEventHandler):
def __init__(self, processor):
self.processor = processor
self.processing_files = set()
def on_created(self, event):
if not event.is_directory and event.src_path.endswith('.log'):
file_path = Path(event.src_path)
if file_path not in self.processing_files:
self.processing_files.add(file_path)
asyncio.create_task(self.processor.process_file(file_path))
print(f"Обнаружен новый лог-файл: {file_path}") |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
| class LogAnalyzer:
def __init__(self):
self.executor = ProcessPoolExecutor()
async def process_file(self, file_path):
# Используем aiofiles для асинхронного чтения
try:
lines = []
async with aiofiles.open(file_path, 'r') as f:
async for line in f:
lines.append(line.strip())
# Распределяем анализ по процессам
loop = asyncio.get_event_loop()
batch_size = max(1, len(lines) // os.cpu_count())
batches = [lines[i:i+batch_size] for i in range(0, len(lines), batch_size)]
analyze_func = partial(self.analyze_batch, patterns=ERROR_PATTERNS)
results = await loop.run_in_executor(
self.executor,
lambda: list(map(analyze_func, batches))
)
# Объединяем результаты
all_errors = []
ip_addresses = Counter()
for errors, ips in results:
all_errors.extend(errors)
for ip, count in ips.items():
ip_addresses[ip] += count
# Создаем отчет
await self.generate_report(file_path, all_errors, ip_addresses)
except Exception as e:
print(f"Ошибка при обработке {file_path}: {e}")
def analyze_batch(self, lines, patterns):
errors = []
ip_addresses = Counter()
ip_pattern = re.compile(r'\b(?:\d{1,3}\.){3}\d{1,3}\b')
error_patterns = [re.compile(pattern) for pattern in patterns]
for line in lines:
# Ищем IP-адреса
ips = ip_pattern.findall(line)
for ip in ips:
ip_addresses[ip] += 1
# Ищем ошибки
for pattern in error_patterns:
if pattern.search(line):
errors.append(line)
break
return errors, ip_addresses
async def generate_report(self, file_path, errors, ip_addresses):
report_name = REPORT_DIR / f"report_{file_path.name}_{int(time.time())}.txt"
async with aiofiles.open(report_name, 'w') as f:
await f.write(f"Отчет по файлу: {file_path}\n")
await f.write(f"Время анализа: {time.ctime()}\n\n")
await f.write(f"Найдено ошибок: {len(errors)}\n")
if errors:
await f.write("Примеры ошибок:\n")
for error in errors[:10]:
await f.write(f" - {error}\n")
await f.write("\nТоп IP-адресов:\n")
for ip, count in ip_addresses.most_common(10):
await f.write(f" - {ip}: {count} раз\n") |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
| async def main():
# Создаем директорию для отчетов, если её нет
REPORT_DIR.mkdir(exist_ok=True)
# Инициализируем анализатор
analyzer = LogAnalyzer()
# Запускаем наблюдатель за директорией с логами
event_handler = LogFileHandler(analyzer)
observer = Observer()
observer.schedule(event_handler, str(LOG_DIR), recursive=True)
observer.start()
print(f"Мониторинг директории {LOG_DIR} запущен...")
try:
# Также проверим существующие лог-файлы
for file_path in LOG_DIR.glob("*.log"):
await analyzer.process_file(file_path)
# Продолжаем работу бесконечно
while True:
await asyncio.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
if __name__ == "__main__":
asyncio.run(main()) |
|
Заключение с рекомендациями по адаптации под конкретные проекты
Собрав внушительную коллекцию Python-сниппетов, хочу поделиться несколькими мыслями о том, как эффективно адаптировать их под ваши конкретные задачи. Ведь прямое копирование редко бывает идеальным решением.
Прежде всего, при интеграции любого сниппета в свой проект, потратьте время на его понимание. Я не раз обжигался, бездумно вставляя чужой код. Помню случай, когда вставил "оптимизированную" функцию обработки CSV, не заметив, что она использует устаревший API - в результате получил странные ошибки, на отладку которых ушло больше времени, чем если бы написал всё с нуля.
Организуйте свою личную библиотеку сниппетов. Я храню полезные куски кода в специальном репозитории с подробными комментариями и примерами использования. Когда встречаю интересное решение, сразу добавляю его туда, предварительно протестировав. Такой подход значительно ускоряет разработку новых проектов.
Вывести список наиболее частых пар слов, употребленных в предложении, введенном с клавиатуры в консоли
Здравствуйте, форумчане. Задача примерно следующая (из заголовка темы): В консоли с клавиатуры... Нахождение 10 наиболее частых слов на web странице
Добрый день. Есть программа для нахождения 10 наиболее частых слов на web странице. Помогите... Пояснения экспертов, Лекции и др. Разъяснения Важных деталей или Частых проблем
Хотелось бы выкладывать тут ссылки на добротные, ёмкие, детальные или короткие но полезные лекции \... Найти среднее количество трудозатрат на добычу каждой полезных ископаемых во всем объеме почвы в каждом пласте
Задача: Найти среднее количество трудозатрат на добычу каждой полезных ископаемых во всем объеме... Российская госкорпорация занимается реорганизацией добычи полезных ископаемых в Западной Африке
Российская госкорпорация занимается реорганизацией добычи полезных ископаемых в Западной Африке.... Поиск полезных ископаемых
Здравствуйте. Не могу решить задачу, прошу помощи, дорогие форумчане
Поиск полезных ископаемых
... Python 3 для самых маленьких или простые задачи
Друзья, вот развлекаюсь с Питоном 3...
>>> print int('11111111', 2)
File "<stdin>", line 1
... Напишите программу для расчета по двум формулам. Подготовьте не менее пяти тестовых примеров
Напишите программу для расчета по двум формулам. Подготовьте не менее пяти тестовых примеров.... Вычислить матрицу вероятностей встречи пикселей из примеров изображений для каждой цифры отдельно
Собрать свои примеры для цифр от 0 до 9. Изображения должны быть высотой 10 и шириной 15 пикселей в... Для проверки правильности решения примеров и выставления оценки за их решения
Всем привет!
Помогите написать прогу:
Написать программу для проверки правильности решения... Модуль turtle, 20 примеров!
Всем привет!
На модуль turtle , на его атрибутах нужно сделать примерно 20 разных примеров с... Вывод случайных примеров
разработайте универсальную функцию, которая будет выводить на
экран 15 (или N) случайных примеров...
|
