Какие знаешь способы отслеживания узлов?

Способы отслеживания узлов в Python

Отслеживание узлов применяется в различных контекстах — от веб-скрейпинга и парсинга DOM до отслеживания состояния сетевых узлов и графовых структур. Расскажу о наиболее практичных подходах.

1. DOM-навигация в веб-парсинге

При работе с BeautifulSoup, Selenium или Playwright отслеживание узлов означает поиск и навигацию по элементам страницы:

from bs4 import BeautifulSoup
import requests

response = requests.get("https://example.com")
soup = BeautifulSoup(response.content, "html.parser")

# Отслеживание узлов по селектору
header = soup.select_one("header")
links = header.find_all("a", class_="nav-link")

for link in links:
    print(link.get("href"), link.text)

# XPath для более сложных селекторов
from lxml import etree
tree = etree.HTML(response.content)
nodes = tree.xpath("//div[@class=content]//p")

2. Отслеживание состояния узлов в сетях

Для мониторинга сетевых узлов используются инструменты вроде ping, SSH проверок, метрик и health-checks:

import asyncio
import subprocess
from dataclasses import dataclass

@dataclass
class NodeStatus:
    host: str
    is_alive: bool
    latency: float | None

async def check_node_health(host: str) -> NodeStatus:
    try:
        result = subprocess.run(
            ["ping", "-c", "1", host],
            capture_output=True,
            timeout=5
        )
        is_alive = result.returncode == 0
        return NodeStatus(host=host, is_alive=is_alive, latency=None)
    except subprocess.TimeoutExpired:
        return NodeStatus(host=host, is_alive=False, latency=None)

async def monitor_cluster(nodes: list[str]):
    tasks = [check_node_health(node) for node in nodes]
    statuses = await asyncio.gather(*tasks)
    return statuses

3. Граф-структуры и дерево зависимостей

Для отслеживания узлов в графах используются алгоритмы обхода и анализа:

from collections import defaultdict, deque

class Graph:
    def __init__(self):
        self.graph = defaultdict(list)
        self.visited = set()
    
    def add_edge(self, u, v):
        self.graph[u].append(v)
    
    # BFS — поиск в ширину
    def bfs_traverse(self, start):
        queue = deque([start])
        visited = {start}
        
        while queue:
            node = queue.popleft()
            print(f"Visiting node: {node}")
            
            for neighbor in self.graph[node]:
                if neighbor not in visited:
                    visited.add(neighbor)
                    queue.append(neighbor)
    
    # DFS — поиск в глубину
    def dfs_traverse(self, node, visited=None):
        if visited is None:
            visited = set()
        
        visited.add(node)
        print(f"Visiting node: {node}")
        
        for neighbor in self.graph[node]:
            if neighbor not in visited:
                self.dfs_traverse(neighbor, visited)

4. Отслеживание в AST (Abstract Syntax Tree)

Для анализа Python кода можно отслеживать узлы синтаксического дерева:

import ast

class NodeVisitor(ast.NodeVisitor):
    def visit_FunctionDef(self, node):
        print(f"Found function: {node.name}")
        print(f"Arguments: {[arg.arg for arg in node.args.args]}")
        self.generic_visit(node)
    
    def visit_Call(self, node):
        if isinstance(node.func, ast.Name):
            print(f"Function call: {node.func.id}")
        self.generic_visit(node)

code = """
def greet(name):
    print(f"Hello, {name}")

greet("World")
"""

tree = ast.parse(code)
visitor = NodeVisitor()
visitor.visit(tree)

5. Кэширование и отслеживание узлов в cache

Отслеживание узлов в кэшах с использованием LRU (Least Recently Used):

from functools import lru_cache

class LRUCache:
    def __init__(self, capacity: int):
        self.cache = {}
        self.capacity = capacity
        self.access_order = []
    
    def get(self, key: str):
        if key in self.cache:
            # Переместить в конец (недавно использованный)
            self.access_order.remove(key)
            self.access_order.append(key)
            return self.cache[key]
        return None
    
    def put(self, key: str, value: any):
        if key in self.cache:
            self.access_order.remove(key)
        elif len(self.cache) >= self.capacity:
            # Удалить самый старый узел
            oldest = self.access_order.pop(0)
            del self.cache[oldest]
        
        self.cache[key] = value
        self.access_order.append(key)

Best Practices

• Используй BFS для поиска кратчайшего пути до узла
• Используй DFS для проверки связности графа
• Кэшируй результаты проверок состояния узлов
• Логируй изменения состояния критических узлов
• Устанавливай таймауты при проверке доступности узлов

Выбор метода отслеживания зависит от контекста: веб-скрейпинг, мониторинг инфраструктуры, анализ графов или синтаксических деревьев.