asyncio и асинхронное программирование в Python: синхронизация, продвинутые примеры, асинхронный парсер

class AdaptiveSemaphore:
    def __init__(self, initial=10, min_size=5, max_size=50):
        self.semaphore = asyncio.Semaphore(initial)
        self.current_size = initial
        self.min_size = min_size
        self.max_size = max_size
        self.response_times = []
    
    async def adjust_based_on_performance(self):
        if len(self.response_times) < 10:
            return
        
        avg_time = sum(self.response_times) / len(self.response_times)
        self.response_times.clear()
        
        if avg_time > 2.0 and self.current_size > self.min_size:
            # Медленно - уменьшаем нагрузку
            self.current_size -= 1
        elif avg_time < 0.5 and self.current_size < self.max_size:
            # Быстро - можем больше
            self.current_size += 1

async def priority_worker(queue: asyncio.PriorityQueue):
    while True:
        priority, task = await queue.get()
        try:
            await process_task(task)
        finally:
            queue.task_done()
 
# Добавление с приоритетом (меньше = важнее)
await queue.put((1, critical_task))   # Высокий приоритет
await queue.put((5, normal_task))     # Обычный
await queue.put((10, background_task)) # Низкий

class TaskGroup:
    def __init__(self):
        self.tasks = []
    
    def create_task(self, coro):
        task = asyncio.create_task(coro)
        self.tasks.append(task)
        return task
    
    async def cancel_all(self):
        for task in self.tasks:
            task.cancel()
        
        await asyncio.gather(*self.tasks, return_exceptions=True)

from functools import wraps
 
def async_lru_cache(maxsize=128):
    cache = {}
    
    def decorator(func):
        @wraps(func)
        async def wrapper(*args):
            if args in cache:
                return cache[args]
            
            result = await func(*args)
            
            if len(cache) >= maxsize:
                cache.pop(next(iter(cache)))
            
            cache[args] = result
            return result
        
        return wrapper
    return decorator
 
@async_lru_cache(maxsize=100)
async def expensive_api_call(param):
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/{param}") as resp:
            return await resp.json()

import asyncio
 
async def fetch_with_new_timeout(url):
    """Современный подход к таймаутам"""
    try:
        async with asyncio.timeout(5.0):
            async with aiohttp.ClientSession() as session:
                async with session.get(url) as response:
                    return await response.text()
    except asyncio.TimeoutError:
        print(f"Запрос к {url} превысил 5 секунд")
        return None

async def fetch_with_old_timeout(url):
    """Старый стиль таймаутов"""
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                result = await asyncio.wait_for(
                    response.text(),
                    timeout=5.0
                )
                return result
    except asyncio.TimeoutError:
        return None

async def cascading_timeouts():
    """Каскадные таймауты для сложных операций"""
    try:
        async with asyncio.timeout(10.0):  # Общий лимит 10 секунд
            # Первая операция - максимум 3 секунды
            async with asyncio.timeout(3.0):
                data = await fetch_initial_data()
            
            # Вторая операция - максимум 5 секунд
            async with asyncio.timeout(5.0):
                processed = await process_data(data)
            
            return processed
    except asyncio.TimeoutError:
        # Сработал один из таймаутов
        return None

class DatabaseTransaction:
    """Контекстный менеджер для транзакций"""
    
    def __init__(self, pool):
        self.pool = pool
        self.conn = None
    
    async def __aenter__(self):
        self.conn = await self.pool.acquire()
        await self.conn.execute('BEGIN')
        return self.conn
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        try:
            if exc_type is None:
                await self.conn.execute('COMMIT')
            else:
                await self.conn.execute('ROLLBACK')
        finally:
            await self.pool.release(self.conn)
        
        return False  # Не подавляем исключения
 
async def use_transaction(pool):
    async with DatabaseTransaction(pool) as conn:
        await conn.execute('INSERT INTO logs VALUES ($1)', 'entry')
        # При исключении автоматический rollback
        # При успехе - commit

async def process_batch_with_taskgroup(items):
    """Обработка батча с автоматической очисткой"""
    results = []
    
    async with asyncio.TaskGroup() as group:
        tasks = [
            group.create_task(process_item(item))
            for item in items
        ]
    
    # Сюда попадаем только когда все задачи завершены
    # Или при первом исключении (остальные отменены)
    return [task.result() for task in tasks]

class AdaptiveTimeout:
    """Динамически адаптирующийся таймаут"""
    
    def __init__(self, base_timeout=5.0):
        self.base_timeout = base_timeout
        self.response_times = []
        self.max_samples = 50
    
    def record_time(self, duration):
        self.response_times.append(duration)
        if len(self.response_times) > self.max_samples:
            self.response_times.pop(0)
    
    def get_timeout(self):
        if not self.response_times:
            return self.base_timeout
        
        avg = sum(self.response_times) / len(self.response_times)
        # Таймаут = среднее * 2 + базовое
        return min(avg * 2 + self.base_timeout, 30.0)
 
adaptive = AdaptiveTimeout()
 
async def smart_fetch(url):
    timeout_value = adaptive.get_timeout()
    start = time.time()
    
    try:
        async with asyncio.timeout(timeout_value):
            result = await fetch_url(url)
        
        adaptive.record_time(time.time() - start)
        return result
    except asyncio.TimeoutError:
        # Не записываем в статистику - операция не завершилась
        return None

import asyncpg
import asyncio
 
class ConnectionPool:
    """Базовая реализация пула соединений"""
    
    def __init__(self, dsn, min_size=5, max_size=20):
        self.dsn = dsn
        self.min_size = min_size
        self.max_size = max_size
        self.pool = None
    
    async def initialize(self):
        """Создание пула при старте приложения"""
        self.pool = await asyncpg.create_pool(
            self.dsn,
            min_size=self.min_size,
            max_size=self.max_size,
            command_timeout=5.0
        )
    
    async def execute_query(self, query, *args):
        """Выполнение запроса через пул"""
        async with self.pool.acquire() as conn:
            return await conn.fetch(query, *args)
    
    async def close(self):
        """Закрытие всех соединений"""
        if self.pool:
            await self.pool.close()

async def configure_http_pool():
    """Настройка HTTP connection pool"""
    connector = aiohttp.TCPConnector(
        limit=100,              # Максимум соединений всего
        limit_per_host=20,      # На один хост
        ttl_dns_cache=300,      # DNS кеш на 5 минут
        keepalive_timeout=30    # Keep-alive соединений
    )
    
    session = aiohttp.ClientSession(connector=connector)
    return session

class ManagedResource:
    """Ресурс с гарантированной очисткой"""
    
    async def __aenter__(self):
        self.pool = await asyncpg.create_pool(DSN)
        self.session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.session.close()
        await self.pool.close()
        
        # Даем время на graceful shutdown
        await asyncio.sleep(0.250)
 
async def use_resources():
    """Использование с автоматической очисткой"""
    async with ManagedResource() as resources:
        # Работаем с пулом и сессией
        data = await resources.pool.fetch("SELECT * FROM users")
        response = await resources.session.get("https://api.example.com")
    
    # Здесь все уже закрыто

async def monitor_pool_health(pool):
    """Периодический мониторинг пула"""
    while True:
        total = pool.get_size()
        idle = pool.get_idle_size()
        active = total - idle
        
        utilization = (active / total) * 100 if total > 0 else 0
        
        print(f"Pool: {active}/{total} active, {utilization:.1f}% utilization")
        
        if utilization > 90:
            print("WARNING: Pool nearly exhausted!")
        
        await asyncio.sleep(10)

class Application:
    """Приложение с корректным завершением"""
    
    def __init__(self):
        self.pool = None
        self.shutdown_event = asyncio.Event()
    
    async def startup(self):
        self.pool = await asyncpg.create_pool(DSN)
    
    async def shutdown(self):
        """Graceful shutdown"""
        # Сигнал остановки фоновым задачам
        self.shutdown_event.set()
        
        # Ждем завершения активных операций
        await asyncio.sleep(1)
        
        # Закрываем пул
        await self.pool.close()
        
        # Финальная пауза для сетевых операций
        await asyncio.sleep(0.250)

import asyncio
 
counter = 0
 
async def increment_naive():
    """Небезопасный инкремент"""
    global counter
    
    # Читаем значение
    current = counter
    
    # Имитация асинхронной работы
    await asyncio.sleep(0.001)
    
    # Записываем увеличенное значение
    counter = current + 1
 
async def test_race_condition():
    """Демонстрация гонки"""
    global counter
    counter = 0
    
    # 100 корутин инкрементируют счетчик
    await asyncio.gather(*[increment_naive() for _ in range(100)])
    
    print(f"Ожидалось: 100, получили: {counter}")
 
asyncio.run(test_race_condition())
# Вывод: Ожидалось: 100, получили: 67 (или другое число < 100)

lock = asyncio.Lock()
counter = 0
 
async def increment_safe():
    """Безопасный инкремент с блокировкой"""
    global counter
    
    async with lock:
        current = counter
        await asyncio.sleep(0.001)
        counter = current + 1
 
async def test_with_lock():
    global counter
    counter = 0
    
    await asyncio.gather(*[increment_safe() for _ in range(100)])
    
    print(f"С блокировкой: {counter}")
    # Вывод: С блокировкой: 100

# Блокировка избыточна - нет await внутри
async with lock:
    counter += 1  # Атомарно и без блокировки
 
# Блокировка нужна - есть await
async with lock:
    current = counter
    await asyncio.sleep(0.001)  # Тут управление может уйти
    counter = current + 1

semaphore = asyncio.Semaphore(3)  # Максимум 3 одновременно
 
async def limited_access(task_id):
    async with semaphore:
        print(f"Задача {task_id} вошла")
        await asyncio.sleep(1)
        print(f"Задача {task_id} вышла")
 
async def test_semaphore():
    # 10 задач, но одновременно работают только 3
    await asyncio.gather(*[limited_access(i) for i in range(10)])

async def waiter(event: asyncio.Event, name: str):
    """Ждет сигнала"""
    print(f"{name} ждет...")
    await event.wait()
    print(f"{name} получил сигнал и продолжил работу")
 
async def signaler(event: asyncio.Event):
    """Отправляет сигнал"""
    await asyncio.sleep(2)
    print("Отправка сигнала всем ожидающим")
    event.set()
 
async def test_event():
    event = asyncio.Event()
    
    await asyncio.gather(
        waiter(event, "Корутина 1"),
        waiter(event, "Корутина 2"),
        signaler(event)
    )

condition = asyncio.Condition()
items = []
 
async def producer(item):
    async with condition:
        items.append(item)
        print(f"Произведен: {item}")
        condition.notify()  # Будим одного ожидающего
 
async def consumer(name):
    async with condition:
        while not items:
            await condition.wait()  # Спим пока нет элементов
        
        item = items.pop(0)
        print(f"{name} потребил: {item}")

# Плохо - потенциальный дедлок
async def task_a(lock1, lock2):
    async with lock1:
        await asyncio.sleep(0.1)
        async with lock2:  # Ждем lock2
            pass
 
async def task_b(lock1, lock2):
    async with lock2:
        await asyncio.sleep(0.1)
        async with lock1:  # Ждем lock1
            pass
 
# Хорошо - фиксированный порядок
async def safe_task(lock1, lock2):
    # Всегда сначала lock1, потом lock2
    async with lock1:
        async with lock2:
            pass

import asyncio
 
async def inner_function():
    await asyncio.sleep(0.1)
    raise ValueError("Ошибка в глубине стека")
 
async def middle_layer():
    await inner_function()
 
async def outer_layer():
    tasks = [middle_layer() for _ in range(3)]
    await asyncio.gather(*tasks)
 
try:
    asyncio.run(outer_layer())
except ValueError as e:
    import traceback
    traceback.print_exc()
    # Вывод: 15+ строк вложенных вызовов через asyncio internals

# Включение debug режима
asyncio.run(main(), debug=True)
 
# Или глобально для всех loop'ов
import asyncio
asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())
loop = asyncio.new_event_loop()
loop.set_debug(True)
asyncio.set_event_loop(loop)

import aiomonitor
 
async def main():
    # Ваш код здесь
    pass
 
with aiomonitor.start_monitor(loop):
    loop.run_until_complete(main())
 
# В другом терминале:
# telnet localhost 50101
# > ps  # список задач
# > cancel <task_id>  # отменить задачу

import asyncio
import logging
 
logger = logging.getLogger(__name__)
 
async def debuggable_function(request_id):
    logger.info(f"[{request_id}] Начало обработки")
    
    await asyncio.sleep(1)
    
    logger.info(f"[{request_id}] Завершение")

# Запуск с профилированием
# py-spy record -o profile.svg -- python script.py

import pytest
 
@pytest.mark.asyncio
async def test_async_function():
    result = await some_coroutine()
    assert result == expected_value

import asyncio
import aiohttp
import asyncpg
from asyncio import Queue
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
import logging
 
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s'
)
logger = logging.getLogger(__name__)
 
@dataclass
class Article:
    """Структура для хранения данных статьи"""
    url: str
    title: str
    content: str
    published_at: Optional[datetime]
    fetched_at: datetime
    
class NewsParser:
    """Асинхронный парсер новостей"""
    
    def __init__(
        self, 
        db_pool: asyncpg.Pool,
        max_workers: int = 10,
        batch_size: int = 50
    ):
        self.db_pool = db_pool
        self.max_workers = max_workers
        self.batch_size = batch_size
        self.queue = Queue(maxsize=1000)
        self.results_queue = Queue()
        self.stats = {
            'processed': 0,
            'errors': 0,
            'start_time': None
        }
        
    async def fetch_page(
        self, 
        session: aiohttp.ClientSession, 
        url: str
    ) -> Optional[str]:
        """Загрузка HTML страницы с обработкой ошибок"""
        try:
            async with asyncio.timeout(10.0):
                async with session.get(url) as response:
                    if response.status == 200:
                        return await response.text()
                    else:
                        logger.warning(
                            f"URL {url} вернул статус {response.status}"
                        )
                        return None
        except asyncio.TimeoutError:
            logger.error(f"Таймаут при загрузке {url}")
            return None
        except aiohttp.ClientError as e:
            logger.error(f"Ошибка сети для {url}: {e}")
            return None
            
    def parse_article(self, html: str, url: str) -> Optional[Article]:
        """Извлечение данных из HTML"""
        # Упрощенный парсинг - в реальности используйте BeautifulSoup
        try:
            # Имитация парсинга
            if '<title>' not in html:
                return None
                
            title_start = html.find('<title>') + 7
            title_end = html.find('</title>')
            title = html[title_start:title_end] if title_end > title_start else "Без названия"
            
            # Извлечение первых 500 символов как контент
            content = html[:500]
            
            return Article(
                url=url,
                title=title,
                content=content,
                published_at=None,
                fetched_at=datetime.now()
            )
        except Exception as e:
            logger.error(f"Ошибка парсинга {url}: {e}")
            return None
            
    async def worker(
        self, 
        worker_id: int, 
        session: aiohttp.ClientSession
    ):
        """Воркер для обработки URL из очереди"""
        logger.info(f"Воркер {worker_id} запущен")
        
        while True:
            url = await self.queue.get()
            
            if url is None:  # Сигнал завершения
                self.queue.task_done()
                break
                
            try:
                # Загружаем страницу
                html = await self.fetch_page(session, url)
                
                if html:
                    # Парсим контент
                    article = self.parse_article(html, url)
                    
                    if article:
                        await self.results_queue.put(article)
                        self.stats['processed'] += 1
                    else:
                        self.stats['errors'] += 1
                else:
                    self.stats['errors'] += 1
                    
            except Exception as e:
                logger.error(f"Воркер {worker_id} упал на {url}: {e}")
                self.stats['errors'] += 1
            finally:
                self.queue.task_done()
                
        logger.info(f"Воркер {worker_id} завершен")
        
    async def storage_worker(self):
        """Батчевая запись результатов в базу"""
        batch = []
        
        while True:
            try:
                article = await asyncio.wait_for(
                    self.results_queue.get(),
                    timeout=2.0
                )
                
                if article is None:  # Сигнал завершения
                    break
                    
                batch.append(article)
                
                if len(batch) >= self.batch_size:
                    await self.save_batch(batch)
                    batch.clear()
                    
            except asyncio.TimeoutError:
                # Таймаут - сохраняем накопленное
                if batch:
                    await self.save_batch(batch)
                    batch.clear()
                    
        # Сохраняем остаток
        if batch:
            await self.save_batch(batch)
            
    async def save_batch(self, articles: list[Article]):
        """Сохранение пачки статей в базу"""
        try:
            async with self.db_pool.acquire() as conn:
                await conn.executemany(
                    '''
                    INSERT INTO articles (url, title, content, fetched_at)
                    VALUES ($1, $2, $3, $4)
                    ON CONFLICT (url) DO UPDATE 
                    SET title = EXCLUDED.title,
                        content = EXCLUDED.content,
                        fetched_at = EXCLUDED.fetched_at
                    ''',
                    [
                        (a.url, a.title, a.content, a.fetched_at)
                        for a in articles
                    ]
                )
            logger.info(f"Сохранено {len(articles)} статей")
        except Exception as e:
            logger.error(f"Ошибка сохранения в базу: {e}")
            
    async def producer(self, urls: list[str]):
        """Загрузка URL в очередь"""
        for url in urls:
            await self.queue.put(url)
            
        # Отправляем сигналы завершения воркерам
        for _ in range(self.max_workers):
            await self.queue.put(None)
            
    async def monitor_progress(self):
        """Мониторинг прогресса выполнения"""
        while True:
            await asyncio.sleep(5)
            
            elapsed = (datetime.now() - self.stats['start_time']).total_seconds()
            rate = self.stats['processed'] / elapsed if elapsed > 0 else 0
            
            logger.info(
                f"Обработано: {self.stats['processed']}, "
                f"Ошибок: {self.stats['errors']}, "
                f"Скорость: {rate:.1f} стр/сек"
            )
            
    async def run(self, urls: list[str]):
        """Запуск парсера"""
        self.stats['start_time'] = datetime.now()
        logger.info(f"Запуск парсера для {len(urls)} URL")
        
        # Настройка HTTP сессии
        connector = aiohttp.TCPConnector(
            limit=100,
            limit_per_host=20
        )
        
        async with aiohttp.ClientSession(connector=connector) as session:
            # Запуск компонентов
            workers = [
                asyncio.create_task(self.worker(i, session))
                for i in range(self.max_workers)
            ]
            
            storage = asyncio.create_task(self.storage_worker())
            monitor = asyncio.create_task(self.monitor_progress())
            producer_task = asyncio.create_task(self.producer(urls))
            
            # Ждем завершения producer
            await producer_task
            
            # Ждем обработки всей очереди
            await self.queue.join()
            
            # Ждем завершения воркеров
            await asyncio.gather(*workers)
            
            # Останавливаем storage
            await self.results_queue.put(None)
            await storage
            
            # Останавливаем монитор
            monitor.cancel()
            try:
                await monitor
            except asyncio.CancelledError:
                pass
                
        elapsed = (datetime.now() - self.stats['start_time']).total_seconds()
        logger.info(
            f"Парсинг завершен за {elapsed:.1f}с. "
            f"Обработано: {self.stats['processed']}, "
            f"Ошибок: {self.stats['errors']}"
        )
 
async def setup_database():
    """Инициализация базы данных"""
    pool = await asyncpg.create_pool(
        'postgresql://user:password@localhost/news',
        min_size=5,
        max_size=20
    )
    
    async with pool.acquire() as conn:
        await conn.execute('''
            CREATE TABLE IF NOT EXISTS articles (
                id SERIAL PRIMARY KEY,
                url TEXT UNIQUE NOT NULL,
                title TEXT NOT NULL,
                content TEXT,
                fetched_at TIMESTAMP NOT NULL
            )
        ''')
        
    return pool
 
async def main():
    """Точка входа"""
    # Список URL для парсинга
    urls = [
        f"https://example.com/news/article-{i}"
        for i in range(500)
    ]
    
    # Инициализация базы
    pool = await setup_database()
    
    try:
        # Запуск парсера
        parser = NewsParser(pool, max_workers=20, batch_size=50)
        await parser.run(urls)
    finally:
        await pool.close()
 
if __name__ == "__main__":
    asyncio.run(main())

class AdaptiveRateLimiter:
    """Динамический контроль скорости запросов"""
    
    def __init__(self, initial_rate=10):
        self.max_rate = initial_rate
        self.current_tokens = initial_rate
        self.last_update = datetime.now()
        self.lock = asyncio.Lock()
        self.error_streak = 0
        
    async def acquire(self):
        """Получение разрешения на запрос"""
        async with self.lock:
            now = datetime.now()
            elapsed = (now - self.last_update).total_seconds()
            
            # Пополняем токены
            self.current_tokens = min(
                self.max_rate,
                self.current_tokens + elapsed * self.max_rate
            )
            self.last_update = now
            
            # Ждем если токенов нет
            if self.current_tokens < 1:
                wait_time = (1 - self.current_tokens) / self.max_rate
                await asyncio.sleep(wait_time)
                self.current_tokens = 1
                
            self.current_tokens -= 1
            
    def report_error(self):
        """Сообщение об ошибке - снижаем скорость"""
        self.error_streak += 1
        if self.error_streak >= 3 and self.max_rate > 1:
            self.max_rate *= 0.7
            logger.warning(
                f"Снижаем rate до {self.max_rate:.1f} запросов/сек"
            )
            
    def report_success(self):
        """Успешный запрос - можем ускориться"""
        self.error_streak = 0
        if self.max_rate < 20:
            self.max_rate *= 1.05

async def smart_worker(
    self, 
    worker_id: int, 
    session: aiohttp.ClientSession,
    rate_limiter: AdaptiveRateLimiter
):
    """Воркер с адаптивным rate limiting"""
    
    while True:
        url = await self.queue.get()
        
        if url is None:
            self.queue.task_done()
            break
            
        try:
            # Ждем разрешения от rate limiter
            await rate_limiter.acquire()
            
            html = await self.fetch_page(session, url)
            
            if html:
                article = self.parse_article(html, url)
                if article:
                    await self.results_queue.put(article)
                    self.stats['processed'] += 1
                    rate_limiter.report_success()
                else:
                    self.stats['errors'] += 1
            else:
                self.stats['errors'] += 1
                rate_limiter.report_error()
                
        except Exception as e:
            logger.error(f"Воркер {worker_id} ошибка на {url}: {e}")
            self.stats['errors'] += 1
            rate_limiter.report_error()
        finally:
            self.queue.task_done()

class SimpleCache:
    """Кеш с TTL для результатов"""
    
    def __init__(self, ttl_seconds=3600):
        self.cache = {}
        self.ttl = ttl_seconds
        self.lock = asyncio.Lock()
        
    async def get(self, key: str) -> Optional[Article]:
        async with self.lock:
            if key in self.cache:
                article, timestamp = self.cache[key]
                age = (datetime.now() - timestamp).total_seconds()
                
                if age < self.ttl:
                    return article
                else:
                    del self.cache[key]
                    
        return None
        
    async def set(self, key: str, value: Article):
        async with self.lock:
            self.cache[key] = (value, datetime.now())
            
            # Очистка старых записей
            if len(self.cache) > 10000:
                await self._cleanup()
                
    async def _cleanup(self):
        """Удаление устаревших записей"""
        now = datetime.now()
        expired = [
            k for k, (_, ts) in self.cache.items()
            if (now - ts).total_seconds() > self.ttl
        ]
        
        for key in expired:
            del self.cache[key]

async def fetch_with_retry(
    self,
    session: aiohttp.ClientSession,
    url: str,
    max_attempts: int = 3
) -> Optional[str]:
    """Загрузка с повторными попытками"""
    
    for attempt in range(max_attempts):
        try:
            async with asyncio.timeout(10.0):
                async with session.get(url) as response:
                    if response.status == 200:
                        return await response.text()
                    elif response.status == 429:
                        # Rate limit - ждем подольше
                        delay = 2 ** (attempt + 2)
                        logger.warning(
                            f"Rate limit на {url}, жду {delay}с"
                        )
                        await asyncio.sleep(delay)
                        continue
                    else:
                        return None
                        
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt == max_attempts - 1:
                logger.error(f"Финальная неудача для {url}: {e}")
                return None
                
            delay = 2 ** attempt
            logger.info(f"Повтор {attempt + 1} для {url} через {delay}с")
            await asyncio.sleep(delay)
            
    return None

class GracefulShutdown:
    """Менеджер корректного завершения"""
    
    def __init__(self):
        self.shutdown_event = asyncio.Event()
        
    def signal_shutdown(self):
        """Сигнал остановки"""
        logger.info("Получен сигнал остановки")
        self.shutdown_event.set()
        
    async def wait_for_shutdown(self):
        """Ожидание сигнала"""
        await self.shutdown_event.wait()
        
async def run_with_shutdown(self, urls: list[str]):
    """Запуск с поддержкой graceful shutdown"""
    shutdown = GracefulShutdown()
    
    # Регистрируем обработчики сигналов
    import signal
    
    def handle_signal(sig, frame):
        shutdown.signal_shutdown()
        
    signal.signal(signal.SIGTERM, handle_signal)
    signal.signal(signal.SIGINT, handle_signal)
    
    try:
        connector = aiohttp.TCPConnector(limit=100)
        
        async with aiohttp.ClientSession(connector=connector) as session:
            rate_limiter = AdaptiveRateLimiter(initial_rate=10)
            cache = SimpleCache(ttl_seconds=3600)
            
            # Запуск воркеров
            workers = [
                asyncio.create_task(
                    self.enhanced_worker(i, session, rate_limiter, cache)
                )
                for i in range(self.max_workers)
            ]
            
            storage = asyncio.create_task(self.storage_worker())
            monitor = asyncio.create_task(self.monitor_progress())
            
            # Producer в фоне
            producer = asyncio.create_task(self.producer(urls))
            
            # Ждем либо завершения, либо сигнала
            done, pending = await asyncio.wait(
                [producer, shutdown.wait_for_shutdown()],
                return_when=asyncio.FIRST_COMPLETED
            )
            
            if shutdown.shutdown_event.is_set():
                logger.info("Корректное завершение начато")
                
                # Отменяем producer если еще работает
                if producer in pending:
                    producer.cancel()
                    
                # Отправляем сигналы завершения
                for _ in workers:
                    await self.queue.put(None)
                    
            # Ждем завершения очереди
            await self.queue.join()
            
            # Ждем воркеров
            await asyncio.gather(*workers, return_exceptions=True)
            
            # Завершаем storage и monitor
            await self.results_queue.put(None)
            await storage
            
            monitor.cancel()
            try:
                await monitor
            except asyncio.CancelledError:
                pass
                
    except Exception as e:
        logger.error(f"Критическая ошибка: {e}")
        raise

from abc import ABC, abstractmethod
from typing import Protocol, Optional
import asyncio
 
class ArticleRepository(ABC):
    """Абстракция для хранения статей"""
    
    @abstractmethod
    async def save_batch(self, articles: list['Article']) -> None:
        """Сохранение пачки статей"""
        pass
    
    @abstractmethod
    async def get_by_url(self, url: str) -> Optional['Article']:
        """Получение статьи по URL"""
        pass
 
class PostgresRepository(ArticleRepository):
    """Реализация хранилища через PostgreSQL"""
    
    def __init__(self, pool: asyncpg.Pool):
        self.pool = pool
    
    async def save_batch(self, articles: list['Article']) -> None:
        async with self.pool.acquire() as conn:
            await conn.executemany(
                '''
                INSERT INTO articles (url, title, content, fetched_at)
                VALUES ($1, $2, $3, $4)
                ON CONFLICT (url) DO UPDATE 
                SET title = EXCLUDED.title,
                    content = EXCLUDED.content,
                    fetched_at = EXCLUDED.fetched_at
                ''',
                [(a.url, a.title, a.content, a.fetched_at) for a in articles]
            )
    
    async def get_by_url(self, url: str) -> Optional['Article']:
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow(
                'SELECT * FROM articles WHERE url = $1', 
                url
            )
            return Article.from_row(row) if row else None

class Parser(Protocol):
    """Интерфейс парсера"""
    
    def can_parse(self, url: str) -> bool:
        """Проверка, может ли парсер обработать URL"""
        ...
    
    async def parse(self, html: str, url: str) -> Optional[Article]:
        """Извлечение данных из HTML"""
        ...
 
class GenericParser:
    """Базовый парсер для обычных сайтов"""
    
    def can_parse(self, url: str) -> bool:
        return True  # Обрабатывает всё по умолчанию
    
    async def parse(self, html: str, url: str) -> Optional[Article]:
        from bs4 import BeautifulSoup
        
        soup = BeautifulSoup(html, 'html.parser')
        title = soup.find('title')
        
        if not title:
            return None
        
        return Article(
            url=url,
            title=title.get_text().strip(),
            content=soup.get_text()[:500],
            published_at=None,
            fetched_at=datetime.now()
        )
 
class HabrParser:
    """Специализированный парсер для Habr"""
    
    def can_parse(self, url: str) -> bool:
        return 'habr.com' in url
    
    async def parse(self, html: str, url: str) -> Optional[Article]:
        from bs4 import BeautifulSoup
        
        soup = BeautifulSoup(html, 'html.parser')
        
        # Специфичные селекторы для Habr
        title_elem = soup.select_one('.tm-article-snippet__title')
        content_elem = soup.select_one('.tm-article-body')
        
        if not title_elem:
            return None
        
        return Article(
            url=url,
            title=title_elem.get_text().strip(),
            content=content_elem.get_text()[:500] if content_elem else "",
            published_at=None,
            fetched_at=datetime.now()
        )
 
class ParserRegistry:
    """Реестр парсеров с выбором подходящего"""
    
    def __init__(self):
        self.parsers: list[Parser] = []
    
    def register(self, parser: Parser) -> None:
        self.parsers.append(parser)
    
    def get_parser(self, url: str) -> Parser:
        for parser in self.parsers:
            if parser.can_parse(url):
                return parser
        
        # Fallback на базовый парсер
        return GenericParser()

from enum import Enum
 
class TaskPriority(Enum):
    LOW = 3
    NORMAL = 2
    HIGH = 1
 
class Task(ABC):
    """Базовая задача"""
    
    def __init__(self, priority: TaskPriority = TaskPriority.NORMAL):
        self.priority = priority
        self.created_at = datetime.now()
    
    @abstractmethod
    async def execute(self, context: 'ExecutionContext') -> None:
        """Выполнение задачи"""
        pass
    
    def __lt__(self, other: 'Task') -> bool:
        # Для сортировки в PriorityQueue
        return self.priority.value < other.priority.value
 
class ParseUrlTask(Task):
    """Задача парсинга URL"""
    
    def __init__(self, url: str, priority: TaskPriority = TaskPriority.NORMAL):
        super().__init__(priority)
        self.url = url
    
    async def execute(self, context: 'ExecutionContext') -> None:
        html = await context.fetcher.fetch(self.url)
        
        if not html:
            return
        
        parser = context.parser_registry.get_parser(self.url)
        article = await parser.parse(html, self.url)
        
        if article:
            await context.repository.save_batch([article])
 
class BatchProcessTask(Task):
    """Задача массовой обработки"""
    
    def __init__(self, urls: list[str]):
        super().__init__(TaskPriority.LOW)
        self.urls = urls
    
    async def execute(self, context: 'ExecutionContext') -> None:
        tasks = [ParseUrlTask(url, TaskPriority.LOW) for url in self.urls]
        
        for task in tasks:
            await context.task_queue.put(task)

class WorkerObserver(ABC):
    """Наблюдатель за работой воркеров"""
    
    @abstractmethod
    async def on_task_started(self, task: Task) -> None:
        pass
    
    @abstractmethod
    async def on_task_completed(self, task: Task, duration: float) -> None:
        pass
    
    @abstractmethod
    async def on_task_failed(self, task: Task, error: Exception) -> None:
        pass
 
class MetricsCollector(WorkerObserver):
    """Сборщик метрик"""
    
    def __init__(self):
        self.completed = 0
        self.failed = 0
        self.total_duration = 0.0
        self.lock = asyncio.Lock()
    
    async def on_task_started(self, task: Task) -> None:
        pass  # Можно логировать начало
    
    async def on_task_completed(self, task: Task, duration: float) -> None:
        async with self.lock:
            self.completed += 1
            self.total_duration += duration
    
    async def on_task_failed(self, task: Task, error: Exception) -> None:
        async with self.lock:
            self.failed += 1
    
    def get_stats(self) -> dict:
        return {
            'completed': self.completed,
            'failed': self.failed,
            'avg_duration': self.total_duration / self.completed if self.completed > 0 else 0
        }
 
class AlertingObserver(WorkerObserver):
    """Отправка алертов при проблемах"""
    
    def __init__(self, error_threshold: int = 10):
        self.error_threshold = error_threshold
        self.recent_errors = 0
    
    async def on_task_started(self, task: Task) -> None:
        pass
    
    async def on_task_completed(self, task: Task, duration: float) -> None:
        self.recent_errors = max(0, self.recent_errors - 1)
    
    async def on_task_failed(self, task: Task, error: Exception) -> None:
        self.recent_errors += 1
        
        if self.recent_errors >= self.error_threshold:
            await self.send_alert(
                f"Высокий уровень ошибок: {self.recent_errors}"
            )
    
    async def send_alert(self, message: str) -> None:
        # Реальная отправка в Slack, email и т.д.
        logger.critical(f"ALERT: {message}")

class ServiceContainer:
    """Контейнер зависимостей"""
    
    def __init__(self):
        self._services = {}
        self._factories = {}
    
    def register(self, interface: type, implementation: type):
        """Регистрация реализации для интерфейса"""
        self._factories[interface] = implementation
    
    def register_instance(self, interface: type, instance):
        """Регистрация готового экземпляра"""
        self._services[interface] = instance
    
    def get(self, interface: type):
        """Получение сервиса"""
        if interface in self._services:
            return self._services[interface]
        
        if interface in self._factories:
            factory = self._factories[interface]
            instance = factory(self)
            self._services[interface] = instance
            return instance
        
        raise ValueError(f"Сервис {interface} не зарегистрирован")
 
class WorkerFactory:
    """Фабрика для создания воркеров"""
    
    def __init__(self, container: ServiceContainer):
        self.container = container
    
    async def create_worker(self, worker_id: int):
        """Создание сконфигурированного воркера"""
        return Worker(
            worker_id=worker_id,
            fetcher=self.container.get(HttpFetcher),
            parser_registry=self.container.get(ParserRegistry),
            repository=self.container.get(ArticleRepository),
            observers=self.container.get(list[WorkerObserver])
        )