Наблюдатель

from __future__ import annotations
from abc import ABC, abstractmethod


class PriceObserver(ABC):
    @abstractmethod
    def update(self, product: str, old_price: float,
               new_price: float) -> None: ...


class ProductCatalog:
    def __init__(self) -> None:
        self._prices: dict[str, float] = {}
        self._observers: list[PriceObserver] = []

    def subscribe(self, observer: PriceObserver) -> None:
        self._observers.append(observer)

    def unsubscribe(self, observer: PriceObserver) -> None:
        self._observers.remove(observer)

    def set_price(self, product: str, price: float) -> None:
        # TODO: сохранить цену, уведомить подписчиков если цена изменилась
        pass

    def get_price(self, product: str) -> float | None:
        return self._prices.get(product)

    def _notify(self, product: str, old_price: float,
                new_price: float) -> None:
        # TODO: оповестить всех подписчиков
        pass


class WishlistObserver(PriceObserver):
    """Покупатель с желаемой максимальной ценой для товаров."""

    def __init__(self, name: str) -> None:
        self.name = name
        self._wishlist: dict[str, float] = {}  # product -> max_price
        self.notifications: list[str] = []

    def add_to_wishlist(self, product: str, max_price: float) -> None:
        self._wishlist[product] = max_price

    def update(self, product: str, old_price: float,
               new_price: float) -> None:
        # TODO: уведомить если товар в списке и цена <= желаемой
        pass


class PriceAnalytics(PriceObserver):
    """Логирует все изменения цен и считает статистику."""

    def __init__(self) -> None:
        self.log: list[tuple[str, float, float]] = []

    def update(self, product: str, old_price: float,
               new_price: float) -> None:
        # TODO: записать изменение в лог
        pass

    def average_change(self) -> float:
        # TODO: вернуть среднее изменение цены в процентах
        pass


class DiscountManager(PriceObserver):
    """Автоматически создаёт скидку при росте цены более 10%."""

    def __init__(self) -> None:
        self.discounts: dict[str, float] = {}  # product -> discount%

    def update(self, product: str, old_price: float,
               new_price: float) -> None:
        # TODO: если рост > 10% — добавить скидку 5%
        pass


# Ожидаемый результат:
# catalog = ProductCatalog()
# alice = WishlistObserver("Alice")
# alice.add_to_wishlist("iPhone", max_price=800.0)
# analytics = PriceAnalytics()
# discounts = DiscountManager()
# catalog.subscribe(alice); catalog.subscribe(analytics); catalog.subscribe(discounts)
# catalog.set_price("iPhone", 1000.0)
# catalog.set_price("iPhone", 750.0)
# # [Alice] iPhone теперь стоит 750.0 — меньше вашего лимита 800.0!
# catalog.set_price("iPhone", 900.0)
# # DiscountManager: скидка 5% на iPhone (цена выросла на 20%)
# print(analytics.log)

from __future__ import annotations
from collections import defaultdict
from typing import Any, Callable


class EventEmitter:
    def __init__(self) -> None:
        self._listeners: dict[str, list[Callable]] = defaultdict(list)
        self._once: dict[str, set[Callable]] = defaultdict(set)

    def on(self, event: str, callback: Callable) -> "EventEmitter":
        # TODO: подписать callback на event, вернуть self для цепочки
        pass

    def once(self, event: str, callback: Callable) -> "EventEmitter":
        # TODO: подписать callback только на одно срабатывание
        pass

    def off(self, event: str, callback: Callable) -> "EventEmitter":
        # TODO: отписать callback от event
        pass

    def emit(self, event: str, *args: Any, **kwargs: Any) -> int:
        # TODO: вызвать все подписанные callback'и, вернуть кол-во вызовов
        #       одноразовые после вызова удалить
        pass

    def listener_count(self, event: str) -> int:
        return len(self._listeners.get(event, []))

    def event_names(self) -> list[str]:
        return list(self._listeners.keys())


# Ожидаемый результат:
# emitter = EventEmitter()
# log = []
# emitter.on("data", lambda x: log.append(f"A:{x}"))
# emitter.on("data", lambda x: log.append(f"B:{x}"))
# emitter.once("data", lambda x: log.append(f"ONCE:{x}"))
# emitter.emit("data", "hello")
# print(log)   # ["A:hello", "B:hello", "ONCE:hello"]
# emitter.emit("data", "world")
# print(log)   # ["A:hello", "B:hello", "ONCE:hello", "A:world", "B:world"]
# print(emitter.listener_count("data"))  # 2 (once уже удалён)

from __future__ import annotations
from typing import Generic, TypeVar, Callable

T = TypeVar("T")


class ReactiveProperty(Generic[T]):
    def __init__(self, initial: T) -> None:
        self._value = initial
        self._subscribers: list[Callable[[T, T], None]] = []

    @property
    def value(self) -> T:
        return self._value

    @value.setter
    def value(self, new_val: T) -> None:
        # TODO: установить значение и уведомить подписчиков
        #       если значение изменилось (old_val, new_val)
        pass

    def subscribe(self, callback: Callable[[T, T], None]) -> None:
        # TODO: добавить подписчика
        pass

    def unsubscribe(self, callback: Callable[[T, T], None]) -> None:
        self._subscribers.remove(callback)


class FormValidator:
    def __init__(self, model: "RegistrationModel") -> None:
        self.errors: dict[str, str] = {}
        # TODO: подписаться на изменения полей модели

    def _validate_username(self, old: str, new: str) -> None:
        # TODO: длина >= 3, только буквы и цифры
        pass

    def _validate_email(self, old: str, new: str) -> None:
        # TODO: должен содержать '@' и '.'
        pass

    def _validate_password(self, old: str, new: str) -> None:
        # TODO: длина >= 8
        pass

    @property
    def is_valid(self) -> bool:
        return len(self.errors) == 0


class LivePreview:
    def __init__(self, model: "RegistrationModel") -> None:
        self.preview = ""
        # TODO: подписаться на изменения полей, обновлять preview

    def _update(self, old: str, new: str) -> None:
        # TODO: собрать превью из текущих значений модели
        pass


class RegistrationModel:
    def __init__(self) -> None:
        self.username: ReactiveProperty[str] = ReactiveProperty("")
        self.email: ReactiveProperty[str] = ReactiveProperty("")
        self.password: ReactiveProperty[str] = ReactiveProperty("")


# Ожидаемый результат:
# model = RegistrationModel()
# validator = FormValidator(model)
# preview = LivePreview(model)
# model.username.value = "Al"
# print(validator.errors)   # {"username": "слишком короткое"}
# model.username.value = "Alice"
# print(validator.errors)   # {}
# model.email.value = "bad-email"
# print(validator.is_valid) # False
# model.email.value = "[email protected]"
# model.password.value = "secret123"
# print(validator.is_valid) # True
# print(preview.preview)    # "Alice | [email protected]"

from __future__ import annotations
from abc import ABC, abstractmethod
from collections import deque
from dataclasses import dataclass


@dataclass
class ServerMetrics:
    cpu_percent: float
    ram_percent: float
    disk_percent: float
    timestamp: str = ""


class MetricsObserver(ABC):
    @abstractmethod
    def update(self, metrics: ServerMetrics) -> None: ...


class ServerMonitor:
    def __init__(self) -> None:
        self._observers: list[MetricsObserver] = []
        self._current: ServerMetrics | None = None

    def subscribe(self, observer: MetricsObserver) -> None:
        self._observers.append(observer)

    def unsubscribe(self, observer: MetricsObserver) -> None:
        self._observers.remove(observer)

    def report_metrics(self, metrics: ServerMetrics) -> None:
        # TODO: сохранить метрики и оповестить подписчиков
        pass

    def _notify(self) -> None:
        # TODO: оповестить всех подписчиков текущими метриками
        pass


class AlertSystem(MetricsObserver):
    def __init__(self, thresholds: dict[str, float]) -> None:
        # thresholds: {"cpu": 80.0, "ram": 90.0, "disk": 95.0}
        self.thresholds = thresholds
        self.active_alerts: list[str] = []

    def update(self, metrics: ServerMetrics) -> None:
        # TODO: проверить каждую метрику против порога,
        #       добавить в active_alerts если превышен
        pass


class MetricsDashboard(MetricsObserver):
    def __init__(self, history_size: int = 10) -> None:
        self._history: deque[ServerMetrics] = deque(maxlen=history_size)

    def update(self, metrics: ServerMetrics) -> None:
        # TODO: добавить метрики в историю
        pass

    def average_cpu(self) -> float:
        # TODO: средний CPU за всю историю
        pass

    def get_history(self) -> list[ServerMetrics]:
        return list(self._history)


class AutoScaler(MetricsObserver):
    def __init__(self, cpu_threshold: float = 75.0) -> None:
        self.instances = 1
        self.cpu_threshold = cpu_threshold
        self.scale_events: list[str] = []

    def update(self, metrics: ServerMetrics) -> None:
        # TODO: если CPU > порога — увеличить instances на 1,
        #       если CPU < порога / 2 и instances > 1 — уменьшить
        pass


# Ожидаемый результат:
# monitor = ServerMonitor()
# alert = AlertSystem({"cpu": 80.0, "ram": 90.0, "disk": 95.0})
# dash = MetricsDashboard(history_size=5)
# scaler = AutoScaler(cpu_threshold=70.0)
# monitor.subscribe(alert); monitor.subscribe(dash); monitor.subscribe(scaler)
# monitor.report_metrics(ServerMetrics(cpu_percent=85.0, ram_percent=60.0, disk_percent=50.0))
# print(alert.active_alerts)   # ["CPU превышен: 85.0% > 80.0%"]
# print(scaler.instances)      # 2
# monitor.report_metrics(ServerMetrics(cpu_percent=20.0, ram_percent=40.0, disk_percent=50.0))
# print(scaler.instances)      # 1
# print(dash.average_cpu())    # 52.5

from __future__ import annotations
from typing import Generic, TypeVar, Callable, Iterator

T = TypeVar("T")


class CollectionEvent(Generic[T]):
    def __init__(self, event_type: str, item: T | None = None,
                 index: int = -1, old_item: T | None = None) -> None:
        self.event_type = event_type  # "added", "removed", "changed", "cleared"
        self.item = item
        self.index = index
        self.old_item = old_item


class ObservableCollection(Generic[T]):
    def __init__(self) -> None:
        self._items: list[T] = []
        self._listeners: dict[str, list[Callable[[CollectionEvent[T]], None]]] = {
            "added": [], "removed": [], "changed": [], "cleared": []
        }

    def on(self, event: str,
           callback: Callable[[CollectionEvent[T]], None]) -> None:
        # TODO: подписать на событие
        pass

    def off(self, event: str,
            callback: Callable[[CollectionEvent[T]], None]) -> None:
        self._listeners[event].remove(callback)

    def _emit(self, event: CollectionEvent[T]) -> None:
        # TODO: уведомить подписчиков на event_type
        pass

    def append(self, item: T) -> None:
        # TODO: добавить элемент и эмитить "added"
        pass

    def remove(self, item: T) -> None:
        # TODO: удалить элемент и эмитить "removed"
        pass

    def __setitem__(self, index: int, value: T) -> None:
        # TODO: заменить элемент и эмитить "changed" (old_item = предыдущий)
        pass

    def clear(self) -> None:
        # TODO: очистить и эмитить "cleared"
        pass

    def __iter__(self) -> Iterator[T]:
        return iter(self._items)

    def __len__(self) -> int:
        return len(self._items)

    def __getitem__(self, index: int) -> T:
        return self._items[index]


class FilteredView(Generic[T]):
    """Живой отфильтрованный вид коллекции."""

    def __init__(self, source: ObservableCollection[T],
                 predicate: Callable[[T], bool]) -> None:
        self._source = source
        self._predicate = predicate
        self._view: list[T] = []
        # TODO: инициализировать _view из source, подписаться на события

    def _rebuild(self, _: CollectionEvent[T]) -> None:
        # TODO: пересобрать _view по предикату из _source
        pass

    def __iter__(self) -> Iterator[T]:
        return iter(self._view)

    def __len__(self) -> int:
        return len(self._view)


# Ожидаемый результат:
# numbers = ObservableCollection()
# evens = FilteredView(numbers, lambda x: x % 2 == 0)
# numbers.append(1); numbers.append(2); numbers.append(3); numbers.append(4)
# print(list(evens))   # [2, 4]
# numbers.remove(2)
# print(list(evens))   # [4]
# numbers.append(6)
# print(list(evens))   # [4, 6]
# log = []
# numbers.on("added", lambda e: log.append(f"added:{e.item}"))
# numbers.append(8)
# print(log)           # ["added:8"]