От async/await к виртуальным потокам в Python

def task_one():
    print("Задача 1 начата")
    yield  # отдаём управление
    print("Задача 1 продолжается")
    yield
    print("Задача 1 завершена")
 
def task_two():
    print("Задача 2 начата")
    yield
    print("Задача 2 завершена")
 
# Простейший планировщик
tasks = [task_one(), task_two()]
while tasks:
    task = tasks.pop(0)
    try:
        next(task)
        tasks.append(task)  # возвращаем в очередь
    except StopIteration:
        pass  # задача завершена

# Было
def parse_data(text):
    return json.loads(text)
 
def process_file(path):
    with open(path) as f:
        return parse_data(f.read())
 
def main():
    result = process_file("data.json")
    print(result)
 
# Стало - parse_data теперь делает HTTP-запрос
async def parse_data(text):
    async with aiohttp.ClientSession() as session:
        # валидируем через внешний API
        await session.post("http://api.example.com/validate", json=text)
    return json.loads(text)
 
async def process_file(path):  # вынуждены сделать async
    async with aiofiles.open(path) as f:  # меняем библиотеку для файлов
        content = await f.read()
        return await parse_data(content)  # await везде
 
async def main():  # и тут тоже
    result = await process_file("data.json")
    print(result)
 
asyncio.run(main())  # event loop нужен теперь

# Корутина - явное управление через await
async def fetch_urls(urls):
    tasks = []
    for url in urls:
        tasks.append(fetch_one(url))  # создаём корутину
    return await asyncio.gather(*tasks)  # явно ждём всех
 
# Виртуальный поток - обычный код
def fetch_urls(urls):
    results = []
    with ThreadGroup() as group:
        for url in urls:
            results.append(group.spawn(lambda: fetch_one(url)))
    return [future.result() for future in results]

from gevent import monkey
monkey.patch_all()
 
# весь остальной код без изменений
from flask import Flask
app = Flask(__name__)
 
@app.route('/data')
def get_data():
    # обычный requests, но работает асинхронно
    response = requests.get('http://api.example.com/endpoint')
    return response.json()

import threading
 
# создание виртуального потока
vthread = threading.VirtualThread(target=fetch_data, args=(url,))
vthread.start()
result = vthread.join()
 
# или через контекстный менеджер для группы
with threading.VirtualThreadGroup() as group:
    for url in urls:
        group.spawn(fetch_data, url)

import asyncio
import aiohttp
from typing import List, Dict
 
async def fetch_user_data(session: aiohttp.ClientSession, user_id: int) -> Dict:
    # запрашиваем данные пользователя
    async with session.get(f'https://api.example.com/users/{user_id}') as response:
        user = await response.json()
    
    # параллельно грузим его посты и комментарии
    posts_task = session.get(f'https://api.example.com/users/{user_id}/posts')
    comments_task = session.get(f'https://api.example.com/users/{user_id}/comments')
    
    posts_response, comments_response = await asyncio.gather(posts_task, comments_task)
    
    return {
        'user': user,
        'posts': await posts_response.json(),
        'comments': await comments_response.json()
    }
 
async def aggregate_data(user_ids: List[int]) -> Dict[int, Dict]:
    async with aiohttp.ClientSession() as session:
        # создаём задачи для всех пользователей
        tasks = [fetch_user_data(session, uid) for uid in user_ids]
        results = await asyncio.gather(*tasks)
        
    return {uid: data for uid, data in zip(user_ids, results)}
 
# точка входа
result = asyncio.run(aggregate_data([1, 2, 3, 4, 5]))

import httpx
from typing import List, Dict
 
def fetch_user_data(user_id: int) -> Dict:
    # обычный блокирующий код
    user = httpx.get(f'https://api.example.com/users/{user_id}').json()
    
    # создаём дочерние потоки для параллельной загрузки
    with ThreadGroup() as group:
        posts_future = group.spawn(
            lambda: httpx.get(f'https://api.example.com/users/{user_id}/posts').json()
        )
        comments_future = group.spawn(
            lambda: httpx.get(f'https://api.example.com/users/{user_id}/comments').json()
        )
        
        return {
            'user': user,
            'posts': posts_future.result(),
            'comments': comments_future.result()
        }
 
def aggregate_data(user_ids: List[int]) -> Dict[int, Dict]:
    with ThreadGroup() as group:
        futures = [group.spawn(fetch_user_data, uid) for uid in user_ids]
        results = [f.result() for f in futures]
        
    return {uid: data for uid, data in zip(user_ids, results)}
 
# просто вызываем функцию
result = aggregate_data([1, 2, 3, 4, 5])

# async версия
import asyncio
import aiosqlite
import aioredis
 
async def get_product(product_id: int, redis: aioredis.Redis, db: aiosqlite.Connection):
    # пытаемся достать из кеша
    cached = await redis.get(f'product:{product_id}')
    if cached:
        return json.loads(cached)
    
    # если нет - идём в базу
    async with db.execute('SELECT * FROM products WHERE id = ?', (product_id,)) as cursor:
        row = await cursor.fetchone()
        
    if not row:
        return None
        
    product = dict(row)
    
    # сохраняем в кеш на 5 минут
    await redis.setex(f'product:{product_id}', 300, json.dumps(product))
    return product
 
async def get_batch(product_ids: List[int]):
    redis = await aioredis.create_redis_pool('redis://localhost')
    db = await aiosqlite.connect('products.db')
    
    try:
        tasks = [get_product(pid, redis, db) for pid in product_ids]
        return await asyncio.gather(*tasks)
    finally:
        await redis.close()
        await db.close()

import sqlite3
import redis
 
def get_product(product_id: int, redis_conn: redis.Redis, db_conn: sqlite3.Connection):
    # проверяем кеш - блокирующий вызов
    cached = redis_conn.get(f'product:{product_id}')
    if cached:
        return json.loads(cached)
    
    # запрос в базу - тоже блокируется
    cursor = db_conn.execute('SELECT * FROM products WHERE id = ?', (product_id,))
    row = cursor.fetchone()
    
    if not row:
        return None
    
    product = dict(row)
    redis_conn.setex(f'product:{product_id}', 300, json.dumps(product))
    return product
 
def get_batch(product_ids: List[int]):
    # соединения создаются обычным способом
    redis_conn = redis.Redis(host='localhost')
    db_conn = sqlite3.connect('products.db')
    
    try:
        with ThreadGroup() as group:
            futures = [
                group.spawn(get_product, pid, redis_conn, db_conn) 
                for pid in product_ids
            ]
            return [f.result() for f in futures]
    finally:
        redis_conn.close()
        db_conn.close()

# async - ловушки с исключениями
async def risky_fetch(url: str):
    async with aiohttp.ClientSession() as session:
        response = await session.get(url)
        if response.status != 200:
            raise ValueError(f"Плохой статус: {response.status}")
        return await response.json()
 
async def fetch_all(urls: List[str]):
    tasks = [risky_fetch(url) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # нужно вручную разбирать результаты
    valid_results = []
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Ошибка в {urls[i]}: {result}")
        else:
            valid_results.append(result)
    
    return valid_results

def risky_fetch(url: str):
    response = httpx.get(url)
    if response.status_code != 200:
        raise ValueError(f"Плохой статус: {response.status_code}")
    return response.json()
 
def fetch_all(urls: List[str]):
    results = []
    with ThreadGroup() as group:
        for url in urls:
            try:
                future = group.spawn(risky_fetch, url)
                results.append(future.result())
            except ValueError as e:
                print(f"Ошибка в {url}: {e}")
                # продолжаем с остальными
    
    return results

async def fetch_with_timeout(url: str, timeout: float):
    try:
        return await asyncio.wait_for(fetch_data(url), timeout=timeout)
    except asyncio.TimeoutError:
        return None  # или raise, или дефолтное значение

with ThreadGroup(timeout=5.0) as group:
    future = group.spawn(fetch_data, url)
    try:
        result = future.result()
    except TimeoutError:
        # вся группа отменилась по таймауту
        pass

import asyncio
import aiofiles
 
async def process_logs_async(log_files: list[str]):
    results = []
    async with asyncio.TaskGroup() as group:
        for log_file in log_files:
            # каждый файл читается в отдельном thread pool worker
            task = group.create_task(read_and_parse_async(log_file))
            results.append(task)
    
    return [t.result() for t in results]
 
async def read_and_parse_async(filepath: str):
    # под капотом это блокирующий open() в потоке
    async with aiofiles.open(filepath, 'r') as f:
        content = await f.read()
    
    # парсинг тоже CPU-bound, нужен executor
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(None, parse_log_content, content)
 
def parse_log_content(content: str):
    # настоящая работа происходит здесь
    return [line.split() for line in content.splitlines() if 'ERROR' in line]

def process_logs_vthread(log_files: list[str]):
    with ThreadGroup(max_workers=100) as group:
        futures = [
            group.spawn(read_and_parse_vthread, log_file) 
            for log_file in log_files
        ]
        return [f.result() for f in futures]
 
def read_and_parse_vthread(filepath: str):
    # обычное чтение файла
    with open(filepath, 'r') as f:
        content = f.read()
    
    # парсинг в том же потоке
    return [line.split() for line in content.splitlines() if 'ERROR' in line]

import asyncio
from asyncio import Queue
 
async def producer(queue: Queue, api_urls: list[str]):
    async with aiohttp.ClientSession() as session:
        for url in api_urls:
            data = await fetch_from_api(session, url)
            # отправляем каждый элемент в очередь
            for item in data['items']:
                await queue.put(item)
    
    # сигнализируем об окончании
    await queue.put(None)
 
async def consumer(queue: Queue, db_pool):
    batch = []
    while True:
        item = await queue.get()
        
        if item is None:  # sentinel value
            if batch:
                await save_batch_to_db(db_pool, batch)
            break
        
        batch.append(process_item(item))
        
        # сохраняем батчами по 100
        if len(batch) >= 100:
            await save_batch_to_db(db_pool, batch)
            batch = []
 
async def run_pipeline(api_urls: list[str]):
    queue = Queue(maxsize=1000)
    db_pool = await create_db_pool()
    
    producer_task = asyncio.create_task(producer(queue, api_urls))
    consumer_tasks = [
        asyncio.create_task(consumer(queue, db_pool)) 
        for _ in range(5)  # пять consumer потоков
    ]
    
    await producer_task
    # отправляем sentinel всем consumers
    for _ in consumer_tasks:
        await queue.put(None)
    
    await asyncio.gather(*consumer_tasks)

from queue import Queue
import threading
 
def producer(queue: Queue, api_urls: list[str]):
    for url in api_urls:
        data = fetch_from_api_sync(url)
        for item in data['items']:
            queue.put(item)
    
    # отправляем столько sentinel values, сколько consumers
    for _ in range(5):
        queue.put(None)
 
def consumer(queue: Queue, db_conn):
    batch = []
    while True:
        item = queue.get()
        
        if item is None:
            if batch:
                save_batch_to_db_sync(db_conn, batch)
            break
        
        batch.append(process_item(item))
        
        if len(batch) >= 100:
            save_batch_to_db_sync(db_conn, batch)
            batch = []
 
def run_pipeline(api_urls: list[str]):
    queue = Queue(maxsize=1000)
    db_conn = create_db_connection()
    
    with ThreadGroup() as group:
        # запускаем producer
        group.spawn(producer, queue, api_urls)
        
        # запускаем consumers
        for _ in range(5):
            group.spawn(consumer, queue, db_conn)

async def fetch_with_rate_limit(url: str, semaphore: asyncio.Semaphore):
    async with semaphore:  # ждём разрешения
        async with aiohttp.ClientSession() as session:
            return await session.get(url)
 
async def fetch_all_limited(urls: list[str], max_concurrent: int):
    semaphore = asyncio.Semaphore(max_concurrent)
    tasks = [fetch_with_rate_limit(url, semaphore) for url in urls]
    return await asyncio.gather(*tasks)

def fetch_all_limited(urls: list[str], max_concurrent: int):
    with ThreadGroup(max_concurrency=max_concurrent) as group:
        futures = [group.spawn(fetch_sync, url) for url in urls]
        return [f.result() for f in futures]

async def fetch_with_retry_async(url: str, max_attempts: int = 3):
    for attempt in range(max_attempts):
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as response:
                    response.raise_for_status()
                    return await response.json()
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt == max_attempts - 1:
                raise
            
            # экспоненциальная задержка
            delay = 2 ** attempt
            await asyncio.sleep(delay)

import time
import random
 
def fetch_with_retry_vthread(url: str, max_attempts: int = 3):
    for attempt in range(max_attempts):
        try:
            response = httpx.get(url, timeout=5.0)
            response.raise_for_status()
            return response.json()
        except httpx.HTTPError as e:
            if attempt == max_attempts - 1:
                raise
            
            # backoff с jitter
            delay = (2 ** attempt) + random.uniform(0, 1)
            time.sleep(delay)  # просто sleep, не await

import asyncpg
 
async def execute_queries_async(query_params: list[tuple]):
    # создаём pool асинхронно
    pool = await asyncpg.create_pool(
        'postgresql://user:pass@localhost/db',
        min_size=10,
        max_size=20
    )
    
    try:
        results = []
        for params in query_params:
            # берём соединение из пула
            async with pool.acquire() as conn:
                result = await conn.fetch('SELECT * FROM items WHERE category = $1', params[0])
                results.append(result)
        return results
    finally:
        await pool.close()

from psycopg2 import pool
 
# создаём пул один раз при старте
db_pool = pool.ThreadedConnectionPool(10, 20, 
    'postgresql://user:pass@localhost/db'
)
 
def execute_queries_vthread(query_params: list[tuple]):
    results = []
    for params in query_params:
        # берём соединение - блокируется если пул исчерпан
        conn = db_pool.getconn()
        try:
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM items WHERE category = %s', params)
            results.append(cursor.fetchall())
        finally:
            # возвращаем соединение в пул
            db_pool.putconn(conn)
    
    return results

import signal
 
shutdown_event = asyncio.Event()
 
async def handle_request(request_id: int):
    # проверяем shutdown на каждом await
    if shutdown_event.is_set():
        raise Exception("Server shutting down")
    
    data = await fetch_data_async()
    
    if shutdown_event.is_set():
        raise Exception("Server shutting down") 
    
    await process_data_async(data)
    return result
 
async def main():
    def signal_handler():
        shutdown_event.set()
    
    loop = asyncio.get_event_loop()
    loop.add_signal_handler(signal.SIGTERM, signal_handler)
    
    # запускаем обработку запросов
    tasks = [handle_request(i) for i in range(100)]
    
    try:
        await asyncio.gather(*tasks, return_exceptions=True)
    except KeyboardInterrupt:
        shutdown_event.set()
        # даём задачам время на завершение
        await asyncio.wait(tasks, timeout=30)

def handle_request(request_id: int):
    try:
        data = fetch_data_sync()
        process_data_sync(data)
        return result
    except ThreadCancelled:
        # делаем cleanup
        rollback_transaction()
        raise
 
def main():
    with ThreadGroup() as group:
        futures = [group.spawn(handle_request, i) for i in range(100)]
        
        # при получении SIGTERM ThreadGroup отменяет дочерние потоки
        def signal_handler(sig, frame):
            group.cancel()
        
        signal.signal(signal.SIGTERM, signal_handler)
        
        # ждём завершения с таймаутом
        try:
            results = [f.result(timeout=30) for f in futures]
        except TimeoutError:
            # принудительно завершаем
            pass

active_tasks = set()
 
async def tracked_task(task_id: int):
    task = asyncio.current_task()
    active_tasks.add(task)
    try:
        return await do_work(task_id)
    finally:
        active_tasks.remove(task)
 
# где-то в мониторинге
print(f"Active tasks: {len(active_tasks)}")
for task in active_tasks:
    print(f"  {task.get_name()}: {task.get_coro()}")

from concurrent.futures import ProcessPoolExecutor
 
def process_image_batch(image_paths):
    with ProcessPoolExecutor(max_workers=8) as executor:
        with ThreadGroup() as group:
            futures = []
            for path in image_paths:
                # виртуальный поток отправляет работу в процесс
                future = group.spawn(lambda p: executor.submit(resize_image, p).result(), path)
                futures.append(future)
            
            return [f.result() for f in futures]

# asyncpg требует переписывания запросов
async def get_users_async(min_age: int, pool):
    async with pool.acquire() as conn:
        # специфичный синтаксис с $1, $2
        rows = await conn.fetch(
            'SELECT id, name, email FROM users WHERE age >= $1 ORDER BY created_at DESC LIMIT $2',
            min_age, 100
        )
        return [dict(row) for row in rows]
 
# psycopg2 с виртуальными потоками использует стандартный DB-API
def get_users_vthread(min_age: int, conn):
    cursor = conn.cursor()
    # привычный синтаксис
    cursor.execute(
        'SELECT id, name, email FROM users WHERE age >= %s ORDER BY created_at DESC LIMIT %s',
        (min_age, 100)
    )
    return cursor.fetchall()

async def transfer_money_async(from_id: int, to_id: int, amount: float, pool):
    async with pool.acquire() as conn:
        async with conn.transaction():
            # списываем со счёта отправителя
            await conn.execute(
                'UPDATE accounts SET balance = balance - $1 WHERE user_id = $2',
                amount, from_id
            )
            # проверяем баланс
            balance = await conn.fetchval(
                'SELECT balance FROM accounts WHERE user_id = $1',
                from_id
            )
            if balance < 0:
                raise ValueError("Недостаточно средств")
            
            # зачисляем получателю
            await conn.execute(
                'UPDATE accounts SET balance = balance + $1 WHERE user_id = $2',
                amount, to_id
            )

def transfer_money_vthread(from_id: int, to_id: int, amount: float, conn):
    try:
        cursor = conn.cursor()
        cursor.execute(
            'UPDATE accounts SET balance = balance - %s WHERE user_id = %s',
            (amount, from_id)
        )
        cursor.execute(
            'SELECT balance FROM accounts WHERE user_id = %s',
            (from_id,)
        )
        balance = cursor.fetchone()[0]
        if balance < 0:
            raise ValueError("Недостаточно средств")
        
        cursor.execute(
            'UPDATE accounts SET balance = balance + %s WHERE user_id = %s',
            (amount, to_id)
        )
        conn.commit()
    except Exception:
        conn.rollback()
        raise

async def fetch_all_async(urls: list[str]):
    async with aiohttp.ClientSession() as session:
        tasks = [session.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [await r.json() for r in responses]
 
# 1000 URL
start = time.time()
results = asyncio.run(fetch_all_async(urls))
print(f"Async: {time.time() - start:.2f}s")

def fetch_all_vthread(urls: list[str]):
    with ThreadGroup(max_workers=1000) as group:
        futures = [group.spawn(lambda u: requests.get(u).json(), url) for url in urls]
        return [f.result() for f in futures]
 
start = time.time()
results = fetch_all_vthread(urls)
print(f"VThreads: {time.time() - start:.2f}s")

def process_response(data: dict) -> dict:
    # имитация реальной обработки
    result = {}
    for key, value in data.items():
        if isinstance(value, dict):
            result[key] = process_response(value)  # рекурсия
        elif isinstance(value, list):
            result[key] = [process_response(item) if isinstance(item, dict) else item 
                          for item in value]
        else:
            result[key] = str(value).upper()  # трансформация
    return result

async def fetch_and_process_async(urls: list[str]):
    loop = asyncio.get_event_loop()
    async with aiohttp.ClientSession() as session:
        tasks = []
        for url in urls:
            response = await session.get(url)
            data = await response.json()
            # CPU-работа в отдельном потоке
            task = loop.run_in_executor(None, process_response, data)
            tasks.append(task)
        return await asyncio.gather(*tasks)

def fetch_and_process_vthread(urls: list[str]):
    with ThreadGroup(max_workers=500) as group:
        def work(url):
            data = requests.get(url).json()
            return process_response(data)  # прямо здесь
        
        futures = [group.spawn(work, url) for url in urls]
        return [f.result() for f in futures]

async def process_logs_async(input_dir: Path, output_file: Path):
    error_lines = []
    files = list(input_dir.glob('*.log'))
    
    async def read_file(filepath):
        async with aiofiles.open(filepath, 'r') as f:
            content = await f.read()
            return [line for line in content.splitlines() if 'ERROR' in line]
    
    tasks = [read_file(f) for f in files]
    results = await asyncio.gather(*tasks)
    
    # собираем все строки
    for lines in results:
        error_lines.extend(lines)
    
    # пишем результат
    async with aiofiles.open(output_file, 'w') as f:
        await f.write('\n'.join(error_lines))

def process_logs_vthread(input_dir: Path, output_file: Path):
    error_lines = []
    files = list(input_dir.glob('*.log'))
    
    def read_file(filepath):
        with open(filepath, 'r') as f:
            content = f.read()
            return [line for line in content.splitlines() if 'ERROR' in line]
    
    with ThreadGroup(max_workers=200) as group:
        futures = [group.spawn(read_file, f) for f in files]
        for future in futures:
            error_lines.extend(future.result())
    
    with open(output_file, 'w') as f:
        f.write('\n'.join(error_lines))

semaphore = asyncio.Semaphore(100)
 
async def limited_request(url: str):
    async with semaphore:
        async with aiohttp.ClientSession() as session:
            return await session.get(url)

def limited_requests(urls: list[str]):
    with ThreadGroup(max_workers=100) as group:
        futures = [group.spawn(lambda u: requests.get(u), url) for url in urls]
        return [f.result() for f in futures]

from fastapi import FastAPI, BackgroundTasks
from crawler.core import CrawlerEngine
 
app = FastAPI()
crawler = CrawlerEngine(max_workers=100)
 
@app.post("/crawl/start")
async def start_crawl(sources: list[str], background_tasks: BackgroundTasks):
    # запускаем краулер асинхронно в фоне
    background_tasks.add_task(crawler.run, sources)
    return {"status": "started", "sources": len(sources)}

class CrawlerEngine:
    def __init__(self, max_workers: int = 100):
        self.max_workers = max_workers
        self.db_pool = psycopg2.pool.ThreadedConnectionPool(10, 20, DB_DSN)
        self.redis_client = redis.Redis(host='localhost')
        self.session = requests.Session()
        
    def run(self, sources: list[str]):
        """Главный метод запуска сбора"""
        with ThreadGroup(max_workers=self.max_workers) as group:
            futures = []
            for source in sources:
                # каждый источник обрабатывается в отдельном виртуальном потоке
                future = group.spawn(self._crawl_source, source)
                futures.append((source, future))
            
            # собираем результаты
            results = {}
            for source, future in futures:
                try:
                    result = future.result(timeout=300)  # максимум 5 минут на источник
                    results[source] = result
                except TimeoutError:
                    logger.warning(f"Источник {source} превысил таймаут")
                    results[source] = {"error": "timeout"}
                except Exception as e:
                    logger.error(f"Ошибка при обработке {source}: {e}")
                    results[source] = {"error": str(e)}
            
            return results

def _crawl_source(self, source: str) -> dict:
    """Обработка одного новостного источника"""
    # проверяем rate limit в Redis
    rate_key = f"rate_limit:{source}"
    last_crawl = self.redis_client.get(rate_key)
    
    if last_crawl:
        elapsed = time.time() - float(last_crawl)
        if elapsed < 60:  # минимум минута между запросами
            time.sleep(60 - elapsed)
    
    # загружаем главную страницу
    main_page = self._fetch_with_retry(source)
    if not main_page:
        return {"articles": 0, "errors": 1}
    
    # извлекаем ссылки на статьи
    article_urls = self._extract_article_urls(main_page, source)
    
    # скачиваем статьи параллельно с ограничением
    articles = []
    with ThreadGroup(max_concurrency=10) as article_group:
        for url in article_urls[:50]:  # максимум 50 статей за раз
            future = article_group.spawn(self._fetch_article, url)
            articles.append(future)
        
        # обрабатываем результаты
        parsed_articles = []
        for future in articles:
            try:
                article_data = future.result()
                if article_data:
                    parsed_articles.append(article_data)
            except Exception as e:
                logger.debug(f"Пропускаем статью: {e}")
    
    # сохраняем в базу батчем
    self._save_articles(parsed_articles)
    
    # обновляем rate limit
    self.redis_client.set(rate_key, time.time())
    
    return {"articles": len(parsed_articles), "errors": len(articles) - len(parsed_articles)}

def _fetch_with_retry(self, url: str, max_attempts: int = 3) -> str | None:
    """Загрузка с повторными попытками"""
    for attempt in range(max_attempts):
        try:
            response = self.session.get(
                url, 
                timeout=10,
                headers={'User-Agent': 'NewsBot/1.0'}
            )
            response.raise_for_status()
            return response.text
        except requests.RequestException as e:
            if attempt == max_attempts - 1:
                logger.error(f"Не удалось загрузить {url}: {e}")
                return None
            
            # экспоненциальная задержка с jitter
            delay = (2 ** attempt) + random.uniform(0, 1)
            time.sleep(delay)
    
    return None

def _extract_article_urls(self, html: str, base_url: str) -> list[str]:
    """Извлечение ссылок на статьи"""
    # CPU-bound операция в отдельном процессе
    with ProcessPoolExecutor(max_workers=4) as executor:
        future = executor.submit(parse_article_links, html, base_url)
        return future.result(timeout=30)
 
# в отдельном модуле для multiprocessing
def parse_article_links(html: str, base_url: str) -> list[str]:
    """Парсинг выполняется в отдельном процессе"""
    soup = BeautifulSoup(html, 'lxml')
    links = []
    
    for link in soup.find_all('a', href=True):
        url = urljoin(base_url, link['href'])
        # фильтруем по паттерну статей
        if re.match(r'.+/news/.+', url):
            links.append(url)
    
    return links

def _fetch_article(self, url: str) -> dict | None:
    """Загрузка и обработка одной статьи"""
    # проверяем, не обрабатывали ли уже
    url_hash = hashlib.md5(url.encode()).hexdigest()
    if self.redis_client.sismember('processed_urls', url_hash):
        return None
    
    # загружаем контент
    html = self._fetch_with_retry(url)
    if not html:
        return None
    
    # парсим в отдельном процессе
    with ProcessPoolExecutor() as executor:
        article_data = executor.submit(extract_article_content, html, url).result()
    
    # обогащаем метаданными через внешний API
    if article_data.get('title'):
        meta = self._enrich_metadata(article_data['title'])
        article_data.update(meta)
    
    # отмечаем как обработанную
    self.redis_client.sadd('processed_urls', url_hash)
    
    return article_data

def _save_articles(self, articles: list[dict]):
    """Сохранение статей в PostgreSQL и Elasticsearch"""
    if not articles:
        return
    
    conn = self.db_pool.getconn()
    try:
        cursor = conn.cursor()
        
        # батчевая вставка в PostgreSQL
        values = [
            (a['url'], a['title'], a['content'], a['published_at'], a.get('category'))
            for a in articles
        ]
        
        execute_batch(
            cursor,
            """INSERT INTO articles (url, title, content, published_at, category) 
               VALUES (%s, %s, %s, %s, %s)
               ON CONFLICT (url) DO UPDATE SET updated_at = NOW()""",
            values
        )
        
        conn.commit()
        
        # индексируем в Elasticsearch
        self._index_to_elasticsearch(articles)
        
    finally:
        self.db_pool.putconn(conn)

def _index_to_elasticsearch(self, articles: list[dict]):
    """Индексация статей для полнотекстового поиска"""
    from elasticsearch import Elasticsearch
    
    es = Elasticsearch(['http://localhost:9200'])
    
    # подготавливаем bulk операции
    actions = []
    for article in articles:
        actions.append({
            '_index': 'news',
            '_id': hashlib.md5(article['url'].encode()).hexdigest(),
            '_source': {
                'title': article['title'],
                'content': article['content'][:5000],  # ограничиваем размер
                'url': article['url'],
                'published_at': article.get('published_at'),
                'category': article.get('category'),
                'indexed_at': datetime.now().isoformat()
            }
        })
    
    # bulk вставка за один запрос
    from elasticsearch.helpers import bulk
    success, failed = bulk(es, actions, raise_on_error=False)
    
    if failed:
        logger.warning(f"Не удалось проиндексировать {len(failed)} статей")

def _enrich_metadata(self, title: str) -> dict:
    """Получение дополнительных метаданных через ML API"""
    try:
        response = requests.post(
            'http://ml-api:8080/classify',
            json={'text': title},
            timeout=5
        )
        
        if response.status_code == 200:
            data = response.json()
            return {
                'category': data.get('category'),
                'sentiment': data.get('sentiment'),
                'topics': data.get('topics', [])
            }
    except requests.RequestException as e:
        logger.debug(f"ML API недоступен: {e}")
    
    # возвращаем дефолты при ошибке
    return {'category': 'uncategorized', 'sentiment': 'neutral', 'topics': []}

def _record_metric(self, metric: str, value: int = 1):
    """Запись метрики в Redis"""
    key = f"metrics:{metric}:{datetime.now().strftime('%Y-%m-%d')}"
    self.redis_client.incrby(key, value)
    self.redis_client.expire(key, 86400 * 7)  # храним неделю
 
def _fetch_article(self, url: str) -> dict | None:
    start_time = time.time()
    
    try:
        # ... загрузка и обработка ...
        
        self._record_metric('articles_processed')
        return article_data
        
    except Exception as e:
        self._record_metric('articles_failed')
        raise
    finally:
        # записываем latency
        latency_ms = int((time.time() - start_time) * 1000)
        self._record_metric('article_latency_ms', latency_ms)

@app.get("/metrics")
async def get_metrics():
    """Текущие метрики системы"""
    today = datetime.now().strftime('%Y-%m-%d')
    
    redis_client = redis.Redis(host='localhost')
    
    return {
        'articles_processed': int(redis_client.get(f"metrics:articles_processed:{today}") or 0),
        'articles_failed': int(redis_client.get(f"metrics:articles_failed:{today}") or 0),
        'avg_latency_ms': int(redis_client.get(f"metrics:article_latency_ms:{today}") or 0) / 
                         max(int(redis_client.get(f"metrics:articles_processed:{today}") or 0), 1)
    }

import signal
import sys
 
class CrawlerEngine:
    def __init__(self, max_workers: int = 100):
        # ... инициализация ...
        self.shutdown_flag = False
        signal.signal(signal.SIGTERM, self._handle_shutdown)
        signal.signal(signal.SIGINT, self._handle_shutdown)
    
    def _handle_shutdown(self, signum, frame):
        """Обработчик сигнала остановки"""
        logger.info("Получен сигнал остановки, завершаем работу...")
        self.shutdown_flag = True
    
    def run(self, sources: list[str]):
        with ThreadGroup(max_workers=self.max_workers) as group:
            futures = []
            
            for source in sources:
                # проверяем флаг перед запуском нового потока
                if self.shutdown_flag:
                    break
                
                future = group.spawn(self._crawl_source, source)
                futures.append((source, future))
            
            # при shutdown даём 30 секунд на завершение активных задач
            timeout = 30 if self.shutdown_flag else 300
            
            results = {}
            for source, future in futures:
                try:
                    result = future.result(timeout=timeout)
                    results[source] = result
                except Exception as e:
                    results[source] = {"error": str(e)}
            
            return results

from pydantic import BaseSettings, Field
 
class CrawlerConfig(BaseSettings):
    max_workers: int = Field(100, description="Максимум параллельных потоков")
    db_pool_min: int = Field(10, description="Минимум соединений в пуле")
    db_pool_max: int = Field(20, description="Максимум соединений в пуле")
    redis_host: str = Field("localhost", description="Хост Redis")
    source_timeout: int = Field(300, description="Таймаут на источник в секундах")
    rate_limit_seconds: int = Field(60, description="Минимальный интервал между запросами")
    
    class Config:
        env_prefix = "CRAWLER_"
 
# использование
config = CrawlerConfig()
crawler = CrawlerEngine(max_workers=config.max_workers)

import click
 
@click.command()
@click.option('--sources-file', required=True, help='Файл со списком источников')
@click.option('--workers', default=100, help='Количество параллельных потоков')
def main(sources_file: str, workers: int):
    """Запуск краулера из командной строки"""
    with open(sources_file) as f:
        sources = [line.strip() for line in f if line.strip()]
    
    config = CrawlerConfig(max_workers=workers)
    crawler = CrawlerEngine(max_workers=config.max_workers)
    
    logger.info(f"Начинаем сбор с {len(sources)} источников")
    results = crawler.run(sources)
    
    total_articles = sum(r.get('articles', 0) for r in results.values())
    total_errors = sum(r.get('errors', 0) for r in results.values())
    
    logger.info(f"Собрано статей: {total_articles}, ошибок: {total_errors}")
 
if __name__ == '__main__':
    main()

import queue
import psycopg2
from contextlib import contextmanager
import threading
import time
 
class VirtualThreadConnectionPool:
    def __init__(self, dsn: str, min_size: int = 5, max_size: int = 20):
        self.dsn = dsn
        self.min_size = min_size
        self.max_size = max_size
        
        # очередь свободных соединений
        self._available = queue.Queue(maxsize=max_size)
        # счётчик всех созданных соединений
        self._total_connections = 0
        self._lock = threading.Lock()
        
        # метрики для мониторинга
        self._metrics = {
            'acquired': 0,
            'released': 0,
            'created': 0,
            'failed': 0,
            'wait_time_ms': []
        }
        
        # создаём минимальное количество соединений сразу
        for _ in range(min_size):
            self._create_connection()
    
    def _create_connection(self):
        """Создание нового соединения к базе"""
        try:
            conn = psycopg2.connect(self.dsn)
            conn.autocommit = False  # явное управление транзакциями
            self._available.put(conn)
            
            with self._lock:
                self._total_connections += 1
                self._metrics['created'] += 1
        except psycopg2.Error as e:
            with self._lock:
                self._metrics['failed'] += 1
            raise
    
    def acquire(self, timeout: float = 30) -> psycopg2.extensions.connection:
        """Получение соединения из пула"""
        start = time.time()
        
        try:
            # пытаемся взять из очереди доступных
            conn = self._available.get(timeout=timeout)
            
            # проверяем, живо ли соединение
            if self._is_connection_alive(conn):
                wait_time = int((time.time() - start) * 1000)
                with self._lock:
                    self._metrics['acquired'] += 1
                    self._metrics['wait_time_ms'].append(wait_time)
                return conn
            
            # соединение мёртвое, создаём новое
            conn.close()
            with self._lock:
                self._total_connections -= 1
            
            # рекурсивно пробуем ещё раз
            return self.acquire(timeout=timeout)
            
        except queue.Empty:
            # очередь пуста, можем ли создать новое соединение?
            with self._lock:
                if self._total_connections < self.max_size:
                    self._create_connection()
                    # новое соединение попало в очередь, берём рекурсивно
                    return self.acquire(timeout=timeout)
            
            # достигли максимума, виртуальный поток должен подождать
            raise TimeoutError("Не удалось получить соединение из пула")
    
    def _is_connection_alive(self, conn) -> bool:
        """Проверка что соединение активно"""
        try:
            # быстрая проверка через простой запрос
            cursor = conn.cursor()
            cursor.execute("SELECT 1")
            cursor.close()
            return True
        except psycopg2.Error:
            return False
    
    def release(self, conn):
        """Возврат соединения в пул"""
        try:
            # откатываем незавершённую транзакцию если есть
            if not conn.autocommit:
                conn.rollback()
            
            self._available.put_nowait(conn)
            with self._lock:
                self._metrics['released'] += 1
        except queue.Full:
            # очередь полна, закрываем лишнее соединение
            conn.close()
            with self._lock:
                self._total_connections -= 1
    
    @contextmanager
    def connection(self, timeout: float = 30):
        """Context manager для автоматического возврата"""
        conn = self.acquire(timeout=timeout)
        try:
            yield conn
        finally:
            self.release(conn)
    
    def get_stats(self) -> dict:
        """Текущая статистика пула"""
        with self._lock:
            avg_wait = (sum(self._metrics['wait_time_ms']) / 
                       len(self._metrics['wait_time_ms'])) if self._metrics['wait_time_ms'] else 0
            
            return {
                'total_connections': self._total_connections,
                'available': self._available.qsize(),
                'acquired': self._metrics['acquired'],
                'released': self._metrics['released'],
                'avg_wait_ms': avg_wait,
                'failed_creates': self._metrics['failed']
            }

class CrawlerEngine:
    def __init__(self):
        self.db_pool = VirtualThreadConnectionPool(
            dsn=DB_DSN,
            min_size=10,
            max_size=30
        )
    
    def _save_articles(self, articles: list[dict]):
        # context manager гарантирует возврат даже при исключении
        with self.db_pool.connection() as conn:
            cursor = conn.cursor()
            
            for article in articles:
                cursor.execute(
                    """INSERT INTO articles (url, title, content) 
                       VALUES (%s, %s, %s)""",
                    (article['url'], article['title'], article['content'])
                )
            
            conn.commit()

def close_all(self):
    """Закрытие всех соединений в пуле"""
    closed = 0
    
    # берём все доступные соединения и закрываем
    while not self._available.empty():
        try:
            conn = self._available.get_nowait()
            conn.close()
            closed += 1
        except queue.Empty:
            break
    
    with self._lock:
        self._total_connections = 0
    
    return closed

crawler/
├── __init__.py
├── main.py              # FastAPI приложение
├── engine.py            # Ядро краулера на виртуальных потоках
├── pool.py              # Кастомный connection pool
├── parsers.py           # Парсеры HTML в отдельных процессах
├── config.py            # Конфигурация через pydantic
└── requirements.txt

"""
Движок краулера на виртуальных потоках.
Каждый источник обрабатывается независимо, медленные не блокируют быстрые.
"""
import time
import random
import hashlib
import logging
from datetime import datetime
from concurrent.futures import ProcessPoolExecutor
from typing import Optional
from urllib.parse import urljoin
 
import requests
import redis
from greenlet import greenlet, getcurrent
from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk
 
from .pool import VirtualThreadConnectionPool
from .parsers import parse_article_links, extract_article_content
from .config import CrawlerConfig
 
logger = logging.getLogger(__name__)
 
 
class ThreadGroup:
    """
    Структурированная группа виртуальных потоков.
    При падении одного отменяются все остальные.
    """
    def __init__(self, max_concurrency: int = 100):
        self.max_concurrency = max_concurrency
        self.active_threads = []
        self.semaphore_count = 0
        self.failed = False
        
    def __enter__(self):
        return self
    
    def spawn(self, func, *args, **kwargs):
        """Создание дочернего виртуального потока с учётом лимита"""
        # блокируемся если достигнут лимит активных потоков
        while self.semaphore_count >= self.max_concurrency:
            time.sleep(0.001)  # отдаём управление планировщику
        
        self.semaphore_count += 1
        
        def wrapper():
            try:
                return func(*args, **kwargs)
            finally:
                self.semaphore_count -= 1
        
        # создаём и запускаем greenlet
        thread = greenlet(wrapper)
        self.active_threads.append(thread)
        
        # возвращаем future-подобный объект
        class Future:
            def __init__(self, thread):
                self.thread = thread
                self._result = None
                self._exception = None
                self._done = False
            
            def result(self, timeout=None):
                """Блокируется пока поток не завершится"""
                if not self._done:
                    start = time.time()
                    # запускаем greenlet и ждём завершения
                    try:
                        self._result = self.thread.switch()
                        self._done = True
                    except Exception as e:
                        self._exception = e
                        self._done = True
                        raise
                    
                    if timeout and (time.time() - start) > timeout:
                        raise TimeoutError()
                
                if self._exception:
                    raise self._exception
                return self._result
        
        return Future(thread)
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """Ждём завершения всех потоков при выходе"""
        for thread in self.active_threads:
            if not thread.dead:
                try:
                    thread.switch()
                except Exception as e:
                    logger.error(f"Ошибка в дочернем потоке: {e}")
 
 
class CrawlerEngine:
    """
    Главный движок краулера.
    Координирует загрузку, парсинг и сохранение данных.
    """
    def __init__(self, config: Optional[CrawlerConfig] = None):
        self.config = config or CrawlerConfig()
        
        # connection pool для PostgreSQL - потокобезопасный
        self.db_pool = VirtualThreadConnectionPool(
            dsn=self.config.database_url,
            min_size=self.config.db_pool_min,
            max_size=self.config.db_pool_max
        )
        
        # Redis для rate limiting и дедупликации
        self.redis_client = redis.Redis(
            host=self.config.redis_host,
            port=self.config.redis_port,
            decode_responses=False
        )
        
        # Elasticsearch для полнотекстового поиска
        self.es = Elasticsearch([self.config.elasticsearch_url])
        
        # HTTP session переиспользует соединения
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'NewsBot/1.0 (compatible; +https://example.com/bot)'
        })
        
        # ProcessPoolExecutor для CPU-bound операций
        self.process_pool = ProcessPoolExecutor(
            max_workers=self.config.parser_workers
        )
        
        logger.info(f"CrawlerEngine инициализирован: max_workers={self.config.max_workers}")
    
    def run(self, sources: list[str]) -> dict:
        """
        Главный метод запуска сбора со всех источников.
        Возвращает статистику по каждому источнику.
        """
        logger.info(f"Начинаем сбор с {len(sources)} источников")
        
        with ThreadGroup(max_concurrency=self.config.max_workers) as group:
            futures = []
            for source_url in sources:
                # каждый источник в отдельном виртуальном потоке
                future = group.spawn(self._crawl_source, source_url)
                futures.append((source_url, future))
            
            # собираем результаты с таймаутами
            results = {}
            for source_url, future in futures:
                try:
                    result = future.result(timeout=self.config.source_timeout)
                    results[source_url] = result
                except TimeoutError:
                    logger.warning(f"Источник {source_url} превысил таймаут")
                    results[source_url] = {"error": "timeout", "articles": 0}
                except Exception as e:
                    logger.error(f"Ошибка при обработке {source_url}: {e}", exc_info=True)
                    results[source_url] = {"error": str(e), "articles": 0}
        
        total_articles = sum(r.get('articles', 0) for r in results.values())
        logger.info(f"Сбор завершён: {total_articles} статей")
        
        return results
    
    def _crawl_source(self, source_url: str) -> dict:
        """Обработка одного новостного источника"""
        # проверяем rate limit через Redis
        rate_key = f"rate_limit:{hashlib.md5(source_url.encode()).hexdigest()}"
        last_crawl = self.redis_client.get(rate_key)
        
        if last_crawl:
            elapsed = time.time() - float(last_crawl)
            if elapsed < self.config.rate_limit_seconds:
                # спим до истечения лимита
                sleep_time = self.config.rate_limit_seconds - elapsed
                logger.debug(f"Rate limit для {source_url}: ждём {sleep_time:.1f}s")
                time.sleep(sleep_time)
        
        # загружаем главную страницу
        main_page_html = self._fetch_with_retry(source_url)
        if not main_page_html:
            return {"articles": 0, "errors": 1}
        
        # извлекаем ссылки на статьи в отдельном процессе (CPU-bound)
        try:
            article_urls = self.process_pool.submit(
                parse_article_links, 
                main_page_html, 
                source_url
            ).result(timeout=30)
        except Exception as e:
            logger.error(f"Ошибка парсинга ссылок для {source_url}: {e}")
            return {"articles": 0, "errors": 1}
        
        # ограничиваем количество статей за раз
        article_urls = article_urls[:self.config.max_articles_per_source]
        
        # скачиваем статьи параллельно с ограничением
        articles_data = []
        with ThreadGroup(max_concurrency=self.config.articles_concurrency) as article_group:
            article_futures = []
            for url in article_urls:
                future = article_group.spawn(self._fetch_article, url)
                article_futures.append(future)
            
            # собираем результаты
            for future in article_futures:
                try:
                    article = future.result()
                    if article:
                        articles_data.append(article)
                except Exception as e:
                    logger.debug(f"Пропускаем статью: {e}")
        
        # сохраняем батчем в базу и индекс
        if articles_data:
            self._save_articles(articles_data)
        
        # обновляем rate limit
        self.redis_client.set(rate_key, time.time())
        
        return {
            "articles": len(articles_data),
            "errors": len(article_urls) - len(articles_data)
        }
    
    def _fetch_with_retry(self, url: str, max_attempts: int = 3) -> Optional[str]:
        """
        Загрузка с повторными попытками и exponential backoff.
        Возвращает HTML или None при неудаче.
        """
        for attempt in range(max_attempts):
            try:
                response = self.session.get(
                    url,
                    timeout=self.config.request_timeout,
                    allow_redirects=True
                )
                response.raise_for_status()
                return response.text
                
            except requests.RequestException as e:
                if attempt == max_attempts - 1:
                    logger.error(f"Не удалось загрузить {url} после {max_attempts} попыток: {e}")
                    return None
                
                # экспоненциальная задержка с jitter
                delay = (2 ** attempt) + random.uniform(0, 1)
                logger.debug(f"Повтор для {url} через {delay:.1f}s")
                time.sleep(delay)
        
        return None
    
    def _fetch_article(self, url: str) -> Optional[dict]:
        """Загрузка и обработка одной статьи"""
        # проверяем дедупликацию
        url_hash = hashlib.md5(url.encode()).hexdigest()
        if self.redis_client.sismember('processed_urls', url_hash):
            return None
        
        # загружаем HTML
        html = self._fetch_with_retry(url)
        if not html:
            return None
        
        # парсим контент в отдельном процессе
        try:
            article_data = self.process_pool.submit(
                extract_article_content,
                html,
                url
            ).result(timeout=30)
        except Exception as e:
            logger.debug(f"Ошибка парсинга статьи {url}: {e}")
            return None
        
        if not article_data or not article_data.get('title'):
            return None
        
        # обогащаем метаданными
        article_data['category'] = self._classify_article(article_data.get('title', ''))
        article_data['collected_at'] = datetime.now().isoformat()
        
        # отмечаем как обработанную
        self.redis_client.sadd('processed_urls', url_hash)
        # TTL 30 дней для очистки старых
        self.redis_client.expire('processed_urls', 86400 * 30)
        
        return article_data
    
    def _classify_article(self, title: str) -> str:
        """
        Классификация статьи через внешний ML API.
        Фоллбэк на 'general' при недоступности API.
        """
        try:
            response = requests.post(
                self.config.ml_api_url,
                json={'text': title},
                timeout=5
            )
            if response.status_code == 200:
                return response.json().get('category', 'general')
        except requests.RequestException:
            pass
        
        return 'general'
    
    def _save_articles(self, articles: list[dict]):
        """Сохранение статей в PostgreSQL и Elasticsearch"""
        # получаем соединение из пула
        with self.db_pool.connection() as conn:
            cursor = conn.cursor()
            
            # батчевая вставка в PostgreSQL
            for article in articles:
                cursor.execute(
                    """
                    INSERT INTO articles (url, title, content, published_at, category, collected_at)
                    VALUES (%s, %s, %s, %s, %s, %s)
                    ON CONFLICT (url) DO UPDATE 
                    SET updated_at = NOW(), content = EXCLUDED.content
                    """,
                    (
                        article['url'],
                        article['title'],
                        article.get('content', ''),
                        article.get('published_at'),
                        article.get('category', 'general'),
                        article.get('collected_at')
                    )
                )
            
            conn.commit()
        
        # индексируем в Elasticsearch для полнотекстового поиска
        self._index_to_elasticsearch(articles)
        
        logger.info(f"Сохранено {len(articles)} статей")
    
    def _index_to_elasticsearch(self, articles: list[dict]):
        """Bulk индексация в Elasticsearch"""
        actions = []
        for article in articles:
            doc_id = hashlib.md5(article['url'].encode()).hexdigest()
            actions.append({
                '_index': 'news',
                '_id': doc_id,
                '_source': {
                    'title': article['title'],
                    'content': article.get('content', '')[:10000],  # лимит размера
                    'url': article['url'],
                    'published_at': article.get('published_at'),
                    'category': article.get('category'),
                    'indexed_at': datetime.now().isoformat()
                }
            })
        
        if actions:
            success, failed = bulk(self.es, actions, raise_on_error=False)
            if failed:
                logger.warning(f"Не удалось проиндексировать {len(failed)} документов")
    
    def shutdown(self):
        """Корректное завершение работы"""
        logger.info("Завершаем работу краулера...")
        self.process_pool.shutdown(wait=True)
        self.db_pool.close_all()
        self.session.close()
        logger.info("Краулер остановлен")

"""
HTTP API для управления краулером.
FastAPI остаётся async - обрабатывает запросы админов,
реальная работа делегируется виртуальным потокам.
"""
from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel
import logging
 
from .engine import CrawlerEngine
from .config import CrawlerConfig
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
app = FastAPI(title="News Crawler API")
config = CrawlerConfig()
crawler = CrawlerEngine(config)
 
 
class CrawlRequest(BaseModel):
    sources: list[str]
 
 
@app.post("/crawl/start")
async def start_crawl(request: CrawlRequest, background_tasks: BackgroundTasks):
    """Запуск сбора новостей в фоновом режиме"""
    if not request.sources:
        raise HTTPException(status_code=400, detail="Список источников пуст")
    
    # запускаем в background task - не блокирует ответ
    background_tasks.add_task(crawler.run, request.sources)
    
    return {
        "status": "started",
        "sources_count": len(request.sources)
    }
 
 
@app.get("/stats")
async def get_stats():
    """Статистика пула соединений"""
    return crawler.db_pool.get_stats()
 
 
@app.on_event("shutdown")
async def shutdown_event():
    """Корректное завершение при остановке приложения"""
    crawler.shutdown()

from pydantic import BaseSettings, Field
 
 
class CrawlerConfig(BaseSettings):
    """Настройки через environment variables"""
    max_workers: int = Field(100, description="Максимум параллельных источников")
    articles_concurrency: int = Field(10, description="Параллельных статей на источник")
    max_articles_per_source: int = Field(50, description="Лимит статей с одного источника")
    
    database_url: str = Field("postgresql://user:pass@localhost/news")
    db_pool_min: int = Field(10)
    db_pool_max: int = Field(30)
    
    redis_host: str = Field("localhost")
    redis_port: int = Field(6379)
    
    elasticsearch_url: str = Field("http://localhost:9200")
    
    ml_api_url: str = Field("http://ml-api:8080/classify")
    
    request_timeout: int = Field(10, description="Таймаут HTTP-запросов в секундах")
    source_timeout: int = Field(300, description="Таймаут на обработку источника")
    rate_limit_seconds: int = Field(60, description="Минимальный интервал между запросами")
    
    parser_workers: int = Field(4, description="Процессов для парсинга")
    
    class Config:
        env_prefix = "CRAWLER_"

"""
Модуль парсинга HTML.
Функции здесь вызываются в отдельных процессах через ProcessPoolExecutor,
поэтому они должны быть top-level и не зависеть от состояния приложения.
"""
import re
from urllib.parse import urljoin, urlparse
from datetime import datetime
from typing import Optional
 
from bs4 import BeautifulSoup
 
 
def parse_article_links(html: str, base_url: str) -> list[str]:
    """
    Извлечение ссылок на статьи из главной страницы.
    CPU-bound операция - BeautifulSoup держит GIL.
    """
    soup = BeautifulSoup(html, 'lxml')
    links = set()
    
    # ищем ссылки в типичных контейнерах новостей
    for container in soup.find_all(['article', 'div'], class_=re.compile(r'article|news|post')):
        for link in container.find_all('a', href=True):
            url = urljoin(base_url, link['href'])
            
            # фильтруем по паттернам URL статей
            if _looks_like_article_url(url):
                links.add(url)
    
    # если не нашли через контейнеры, берём все подходящие ссылки
    if not links:
        for link in soup.find_all('a', href=True):
            url = urljoin(base_url, link['href'])
            if _looks_like_article_url(url):
                links.add(url)
    
    return list(links)
 
 
def _looks_like_article_url(url: str) -> bool:
    """Эвристика определения что URL ведёт на статью"""
    parsed = urlparse(url)
    path = parsed.path.lower()
    
    # исключаем служебные страницы
    if any(x in path for x in ['/tag/', '/category/', '/page/', '/author/', 'wp-admin', 'wp-login']):
        return False
    
    # включаем типичные паттерны новостей
    article_patterns = [
        r'/news/',
        r'/\d{4}/\d{2}/',  # датированные URL
        r'/article/',
        r'/post/',
        r'-\d+\.html$',  # ID в конце
    ]
    
    return any(re.search(pattern, path) for pattern in article_patterns)
 
 
def extract_article_content(html: str, url: str) -> Optional[dict]:
    """
    Извлечение содержимого статьи из HTML.
    Возвращает dict с title, content, published_at или None.
    """
    soup = BeautifulSoup(html, 'lxml')
    
    # извлекаем заголовок
    title = _extract_title(soup)
    if not title:
        return None
    
    # извлекаем основной текст
    content = _extract_content(soup)
    
    # пытаемся найти дату публикации
    published_at = _extract_date(soup)
    
    return {
        'url': url,
        'title': title,
        'content': content,
        'published_at': published_at
    }
 
 
def _extract_title(soup: BeautifulSoup) -> Optional[str]:
    """Извлечение заголовка статьи"""
    # пробуем Open Graph
    og_title = soup.find('meta', property='og:title')
    if og_title and og_title.get('content'):
        return og_title['content'].strip()
    
    # пробуем тег title
    title_tag = soup.find('title')
    if title_tag:
        title = title_tag.get_text().strip()
        # убираем название сайта
        title = re.split(r'[|–—-]', title)[0].strip()
        if title:
            return title
    
    # пробуем h1
    h1 = soup.find('h1')
    if h1:
        return h1.get_text().strip()
    
    return None
 
 
def _extract_content(soup: BeautifulSoup) -> str:
    """Извлечение основного текста статьи"""
    # удаляем мусор
    for tag in soup.find_all(['script', 'style', 'nav', 'footer', 'aside']):
        tag.decompose()
    
    # ищем основной контейнер контента
    content_containers = soup.find_all(
        ['article', 'div'],
        class_=re.compile(r'content|article|post-body|entry-content', re.I)
    )
    
    if content_containers:
        # берём самый большой контейнер
        main_container = max(content_containers, key=lambda x: len(x.get_text()))
        paragraphs = main_container.find_all('p')
    else:
        # фоллбэк на все параграфы
        paragraphs = soup.find_all('p')
    
    # собираем текст, фильтруя короткие параграфы
    content_parts = []
    for p in paragraphs:
        text = p.get_text().strip()
        if len(text) > 50:  # только содержательные параграфы
            content_parts.append(text)
    
    return '\n\n'.join(content_parts)
 
 
def _extract_date(soup: BeautifulSoup) -> Optional[str]:
    """Извлечение даты публикации"""
    # пробуем meta теги
    date_meta = soup.find('meta', property=re.compile(r'published|date', re.I))
    if date_meta and date_meta.get('content'):
        return date_meta['content']
    
    # пробуем time тег
    time_tag = soup.find('time')
    if time_tag:
        # сначала атрибут datetime
        if time_tag.get('datetime'):
            return time_tag['datetime']
        # потом текст внутри
        return time_tag.get_text().strip()
    
    return None

fastapi==0.104.1
uvicorn[standard]==0.24.0
pydantic==2.5.0
pydantic-settings==2.1.0
 
# HTTP клиенты
requests==2.31.0
httpx==0.25.1
 
# Парсинг
beautifulsoup4==4.12.2
lxml==4.9.3
 
# Базы данных
psycopg2-binary==2.9.9
redis==5.0.1
elasticsearch==8.11.0
 
# Виртуальные потоки
greenlet==3.0.1
 
# Утилиты
python-dotenv==1.0.0

"""
Консольный интерфейс для запуска краулера.
Полезен для cron jobs и ручного запуска.
"""
import sys
import logging
from pathlib import Path
 
import click
 
from .engine import CrawlerEngine
from .config import CrawlerConfig
 
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
 
 
@click.group()
def cli():
    """News Crawler CLI"""
    pass
 
 
@cli.command()
@click.option('--sources', '-s', required=True, type=click.Path(exists=True),
              help='Файл со списком URL источников (по одному на строку)')
@click.option('--workers', '-w', default=100, help='Количество параллельных потоков')
def crawl(sources: str, workers: int):
    """Запуск сбора новостей"""
    # читаем источники из файла
    with open(sources) as f:
        source_urls = [line.strip() for line in f if line.strip() and not line.startswith('#')]
    
    if not source_urls:
        click.echo("Файл источников пуст", err=True)
        sys.exit(1)
    
    config = CrawlerConfig(max_workers=workers)
    crawler = CrawlerEngine(config)
    
    click.echo(f"Начинаем сбор с {len(source_urls)} источников...")
    
    try:
        results = crawler.run(source_urls)
        
        # выводим статистику
        total_articles = sum(r.get('articles', 0) for r in results.values())
        total_errors = sum(r.get('errors', 0) for r in results.values())
        
        click.echo(f"\nРезультаты:")
        click.echo(f"  Собрано статей: {total_articles}")
        click.echo(f"  Ошибок: {total_errors}")
        
        # детальная статистика по источникам
        for url, result in results.items():
            if result.get('error'):
                click.echo(f"  {url}: {result['error']}", err=True)
            else:
                click.echo(f"  {url}: {result['articles']} статей")
    
    finally:
        crawler.shutdown()
 
 
@cli.command()
def test_connection():
    """Проверка подключения к базам данных"""
    config = CrawlerConfig()
    click.echo("Проверяем подключения...")
    
    # PostgreSQL
    try:
        from .pool import VirtualThreadConnectionPool
        pool = VirtualThreadConnectionPool(config.database_url, min_size=1, max_size=1)
        with pool.connection() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT version()")
            version = cursor.fetchone()[0]
            click.echo(f"PostgreSQL: {version[:50]}...")
        pool.close_all()
    except Exception as e:
        click.echo(f"PostgreSQL: {e}", err=True)
    
    # Redis
    try:
        import redis
        r = redis.Redis(host=config.redis_host, port=config.redis_port)
        r.ping()
        info = r.info('server')
        click.echo(f"Redis: version {info['redis_version']}")
    except Exception as e:
        click.echo(f"Redis: {e}", err=True)
    
    # Elasticsearch
    try:
        from elasticsearch import Elasticsearch
        es = Elasticsearch([config.elasticsearch_url])
        info = es.info()
        click.echo(f"Elasticsearch: version {info['version']['number']}")
    except Exception as e:
        click.echo(f"Elasticsearch: {e}", err=True)
 
 
if __name__ == '__main__':
    cli()

https://example-news.com
https://another-news-site.com
https://tech.news-aggregator.org
# комментарии игнорируются
https://local-news-portal.com