Структуры данных collections в Python: оптимизация кода и скорость

from collections import Counter, defaultdict, OrderedDict, namedtuple, deque

from collections import Counter

# Подсчёт символов в строке
c = Counter("абракадабра")
print(c) # Counter({'а': 5, 'б': 2, 'р': 2, 'к': 1, 'д': 1})

# Подсчёт элементов в списке
languages = ["python", "java", "python", "javascript", "python", "java"]
lang_counter = Counter(languages)
print(lang_counter) # Counter({'python': 3, 'java': 2, 'javascript': 1})

from collections import Counter
import re

def analyze_text(text):
# Разбиваем текст на слова, игнорируя пунктуацию и регистр
words = re.findall(r'\b\w+\b', text.lower())

# Подсчитываем частоту слов
word_counts = Counter(words)

# Получаем 5 самых популярных слов
top_words = word_counts.most_common(5)

# Получаем общее количество уникальных слов
unique_words = len(word_counts)

return {
"top_words": top_words,
"total_words": len(words),
"unique_words": unique_words
}

sample_text = """
Python — высокоуровневый язык программирования общего назначения. 
Python ориентирован на повышение производительности разработчика и 
читаемости кода. Python поддерживает несколько парадигм программирования.
"""

result = analyze_text(sample_text)
print(f"Топ-5 слов: {result['top_words']}")
print(f"Всего слов: {result['total_words']}")
print(f"Уникальных слов: {result['unique_words']}")

from collections import defaultdict

# Словарь со списками по умолчанию
grouped_data = defaultdict(list)

# Добавляем элементы без необходимости проверять существование ключа
for name, group in [("Алексей", "Администраторы"), 
("Мария", "Пользователи"), 
("Иван", "Администраторы")]:
grouped_data[group].append(name)

print(grouped_data)
# defaultdict(<class 'list'>, {'Администраторы': ['Алексей', 'Иван'], 'Пользователи': ['Мария']})

# Без defaultdict
grouped_data = {}

for name, group in [("Алексей", "Администраторы"), 
("Мария", "Пользователи"), 
("Иван", "Администраторы")]:
if group not in grouped_data:
grouped_data[group] = []
grouped_data[group].append(name)

# Словарь с целыми числами (0) по умолчанию
word_count = defaultdict(int)
for word in ["яблоко", "груша", "яблоко", "ананас", "груша", "яблоко"]:
word_count[word] += 1
print(word_count) # defaultdict(<class 'int'>, {'яблоко': 3, 'груша': 2, 'ананас': 1})

# Словарь со словарями по умолчанию
nested = defaultdict(dict)
nested["сервер1"]["cpu"] = 80
nested["сервер1"]["ram"] = 70
nested["сервер2"]["cpu"] = 60
print(nested)

# Собственная фабрика значений
def default_factory():
return {"count": 0, "sum": 0}

stats = defaultdict(default_factory)
data = [("A", 10), ("B", 20), ("A", 30), ("C", 40)]

for key, value in data:
stats[key]["count"] += 1
stats[key]["sum"] += value

for key, values in stats.items():
values["average"] = values["sum"] / values["count"]

print(stats)

from collections import OrderedDict

# Создание OrderedDict
od = OrderedDict()
od['a'] = 1
od['b'] = 2
od['c'] = 3
print(od) # OrderedDict([('a', 1), ('b', 2), ('c', 3)])

# Перемещение элемента в конец
od.move_to_end('a')
print(od) # OrderedDict([('b', 2), ('c', 3), ('a', 1)])

# Перемещение элемента в начало
od.move_to_end('a', last=False)
print(od) # OrderedDict([('a', 1), ('b', 2), ('c', 3)])

# Удаление и возврат последней пары (key, value)
key, value = od.popitem()
print(key, value) # c 3
print(od) # OrderedDict([('a', 1), ('b', 2)])

from collections import deque

# Создание deque из итерируемого объекта
d = deque([1, 2, 3, 4, 5])
print(d) # deque([1, 2, 3, 4, 5])

# Добавление элементов
d.append(6) # Добавить справа
d.appendleft(0) # Добавить слева
print(d) # deque([0, 1, 2, 3, 4, 5, 6])

# Удаление элементов
right = d.pop() # Удалить справа
left = d.popleft() # Удалить слева
print(f"Удалено: справа = {right}, слева = {left}")
print(d) # deque([1, 2, 3, 4, 5])

# Вращение deque
d.rotate(2) # Вращение вправо на 2 позиции
print(d) # deque([4, 5, 1, 2, 3])
d.rotate(-2) # Вращение влево на 2 позиции
print(d) # deque([1, 2, 3, 4, 5])

# Ограничение размера deque
limited_deque = deque(maxlen=3)
for i in range(5):
limited_deque.append(i)
print(limited_deque)
# deque([0], maxlen=3)
# deque([0, 1], maxlen=3)
# deque([0, 1, 2], maxlen=3)
# deque([1, 2, 3], maxlen=3)
# deque([2, 3, 4], maxlen=3)

from collections import deque
import random

def moving_average(iterable, n=3):
"""Вычисляет скользящее среднее с окном размера n."""
if n < 1:
raise ValueError("Размер окна должен быть положительным")

window = deque(maxlen=n)
for x in iterable:
window.append(x)
if len(window) == n: # Окно заполнено
yield sum(window) / n

# Генерируем случайные данные (например, цены акций)
prices = [random.uniform(100, 200) for _ in range(10)]
print("Исходные данные:", prices)

# Вычисляем скользящее среднее с окном 3
averages = list(moving_average(prices, 3))
print("Скользящее среднее (окно 3):", averages)

from collections import Counter, defaultdict
import re

def analyze_logs(log_file):
# Шаблон для парсинга строк лога (упрощенный)
pattern = r'(\d+\.\d+\.\d+\.\d+) .* "(GET|POST) ([^"]*)" (\d+)'

urls = []
ips = []
statuses = []

with open(log_file, 'r') as f:
for line in f:
match = re.search(pattern, line)
if match:
ip, method, url, status = match.groups()
ips.append(ip)
urls.append(url)
statuses.append(status)

# Анализ с использованием Counter
top_urls = Counter(urls).most_common(5)
top_ips = Counter(ips).most_common(5)
status_distribution = Counter(statuses)

# Группировка URL по статусам с использованием defaultdict
url_by_status = defaultdict(list)
for url, status in zip(urls, statuses):
url_by_status[status].append(url)

# Находим URL с ошибками (4xx, 5xx)
error_urls = {status: Counter(urls).most_common(3) 
for status, urls in url_by_status.items() 
if status.startswith(('4', '5'))}

return {
"top_urls": top_urls,
"top_ips": top_ips,
"status_distribution": status_distribution,
"error_urls": error_urls
}

# Использование
# results = analyze_logs("access.log")
# print("Топ-5 URL:", results["top_urls"])
# print("Топ-5 IP:", results["top_ips"])
# print("Распределение статусов:", results["status_distribution"])
# print("URL с ошибками:", results["error_urls"])

from collections import OrderedDict

class LRUCache:
def __init__(self, capacity):
self.capacity = capacity
self.cache = OrderedDict()

def get(self, key):
if key not in self.cache:
return -1

# Перемещаем элемент в конец, обновляя порядок "недавности"
self.cache.move_to_end(key)
return self.cache[key]

def put(self, key, value):
# Если ключ уже существует, обновляем значение и перемещаем в конец
if key in self.cache:
self.cache.move_to_end(key)
# Если кэш заполнен, удаляем самый старый элемент
elif len(self.cache) >= self.capacity:
self.cache.popitem(last=False)

self.cache[key] = value

# Пример использования
cache = LRUCache(3)
cache.put(1, 'one')
cache.put(2, 'two')
cache.put(3, 'three')
print(cache.cache) # OrderedDict([(1, 'one'), (2, 'two'), (3, 'three')])

# Обращение к существующему элементу перемещает его в конец
cache.get(1)
print(cache.cache) # OrderedDict([(2, 'two'), (3, 'three'), (1, 'one')])

# Добавление нового элемента при заполненном кэше вытесняет старейший
cache.put(4, 'four')
print(cache.cache) # OrderedDict([(3, 'three'), (1, 'one'), (4, 'four')])

from collections import defaultdict, namedtuple
import csv
from datetime import datetime

# Определяем структуру записи о продаже
Sale = namedtuple('Sale', ['date', 'product', 'category', 'price', 'quantity'])

def analyze_sales(csv_file):
# Группировка продаж по категориям
sales_by_category = defaultdict(list)

with open(csv_file, 'r') as f:
reader = csv.reader(f)
next(reader) # Пропускаем заголовок

for row in reader:
date = datetime.strptime(row[0], '%Y-%m-%d')
product = row[1]
category = row[2]
price = float(row[3])
quantity = int(row[4])

sale = Sale(date, product, category, price, quantity)
sales_by_category[category].append(sale)

# Статистика по категориям
category_stats = {}
for category, sales in sales_by_category.items():
total_revenue = sum(sale.price * sale.quantity for sale in sales)
avg_price = sum(sale.price for sale in sales) / len(sales)
total_quantity = sum(sale.quantity for sale in sales)

# Находим бестселлеры в каждой категории
product_sales = defaultdict(int)
for sale in sales:
product_sales[sale.product] += sale.quantity

bestsellers = sorted(product_sales.items(), key=lambda x: x[1], reverse=True)[:3]

category_stats[category] = {
'total_revenue': total_revenue,
'avg_price': avg_price,
'total_quantity': total_quantity,
'bestsellers': bestsellers
}

return category_stats

# Пример использования
# stats = analyze_sales("sales.csv")
# for category, data in stats.items():
# print(f"\nКатегория: {category}")
# print(f"Общая выручка: ${data['total_revenue']:.2f}")
# print(f"Средняя цена: ${data['avg_price']:.2f}")
# print(f"Общее количество: {data['total_quantity']}")
# print("Бестселлеры:")
# for product, qty in data['bestsellers']:
# print(f" – {product}: {qty} шт.")

from collections import defaultdict, deque

class Graph:
def __init__(self):
# Используем defaultdict для представления списков смежности
self.edges = defaultdict(list)

def add_edge(self, u, v, directed=False):
"""Добавляет ребро между вершинами u и v."""
self.edges[u].append(v)
if not directed:
self.edges[v].append(u)

def bfs(self, start):
"""Поиск в ширину от начальной вершины."""
visited = set()
queue = deque([start])
visited.add(start)
result = []

while queue:
vertex = queue.popleft()
result.append(vertex)

# Обрабатываем всех соседей
for neighbor in self.edges[vertex]:
if neighbor not in visited:
visited.add(neighbor)
queue.append(neighbor)

return result

def dfs(self, start):
"""Поиск в глубину от начальной вершины."""
visited = set()
result = []

def dfs_recursive(vertex):
visited.add(vertex)
result.append(vertex)

for neighbor in self.edges[vertex]:
if neighbor not in visited:
dfs_recursive(neighbor)

dfs_recursive(start)
return result

def shortest_path(self, start, end):
"""Находит кратчайший путь между вершинами start и end."""
if start == end:
return [start]

visited = {start}
queue = deque([(start, [start])])

while queue:
vertex, path = queue.popleft()

for neighbor in self.edges[vertex]:
if neighbor == end:
return path + [neighbor]

if neighbor not in visited:
visited.add(neighbor)
queue.append((neighbor, path + [neighbor]))

return None # Путь не найден

# Пример использования
g = Graph()
g.add_edge('A', 'B')
g.add_edge('A', 'C')
g.add_edge('B', 'D')
g.add_edge('B', 'E')
g.add_edge('C', 'F')
g.add_edge('E', 'F')

print("BFS от A:", g.bfs('A'))
print("DFS от A:", g.dfs('A'))
print("Кратчайший путь от A до F:", g.shortest_path('A', 'F'))