HeadHunter Technical: Python, ретраи и надежность

Подсказки

• Сначала частоты

  Используйте dict: freq[ch] = freq.get(ch, 0) + 1.

• Сортировка

  Сортируйте пары (символ, частота) по ключу (-частота, символ).

Идея решения

Эталонный код

def find_top_k(s: str, k: int) -> list[str]:
    freq: dict[str, int] = {}
    for ch in s:
        freq[ch] = freq.get(ch, 0) + 1

    ordered = sorted(freq.items(), key=lambda item: (-item[1], item[0]))
    return [ch for ch, _ in ordered[:k]]

Подсказки

• Три уровня функций

  retry(on, num_attempts) возвращает decorator, decorator(func) возвращает wrapper.

• Последняя попытка

  Если это последняя попытка, внутри except нужно снова поднять исключение через raise.

• Метаданные

  Используйте @wraps(func), иначе имя функции станет wrapper.

Идея решения

Эталонный код

from functools import wraps


def retry(on: type[Exception], num_attempts: int = 1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(num_attempts):
                try:
                    return func(*args, **kwargs)
                except on:
                    if attempt == num_attempts - 1:
                        raise
        return wrapper
    return decorator


def run_retry_case(outcomes: list[str], retry_on: str, num_attempts: int) -> dict:
    exception_types = {
        'KeyError': KeyError,
        'ValueError': ValueError,
        'RuntimeError': RuntimeError,
    }
    state = {'calls': 0, 'index': 0}

    def flaky():
        state['calls'] += 1
        outcome = outcomes[state['index']]
        state['index'] += 1
        if outcome.startswith('ok:'):
            return outcome[3:]
        raise exception_types[outcome](outcome)

    wrapped = retry(exception_types[retry_on], num_attempts)(flaky)
    try:
        value = wrapped()
        return {'status': 'ok', 'value': value, 'calls': state['calls'], 'name': getattr(wrapped, '__name__', '')}
    except Exception as exc:
        return {'status': 'error', 'error': type(exc).__name__, 'calls': state['calls'], 'name': getattr(wrapped, '__name__', '')}

Подсказки

• Очередь событий

  Нужна структура, из которой удобно удалять самые старые записи.

• Не пересчитывайте сумму

  Поддерживайте текущую сумму и вычитайте значение, когда удаляете старую запись.

• Граница окна

  Событие с timestamp ровно now() - 300 еще находится в последних 5 минутах.

Идея решения

Эталонный код

from collections import deque


class MovingAvg:
    def __init__(self):
        self.data = deque()
        self.cur_sum = 0

    def record(self, value: int) -> None:
        ts = self.now()
        self.data.append((ts, value))
        self.cur_sum += value
        self._clean_old(ts)

    def mean(self) -> float:
        ts = self.now()
        self._clean_old(ts)
        if not self.data:
            return 0.0
        return self.cur_sum / len(self.data)

    def _clean_old(self, ts: int) -> None:
        while self.data and ts - self.data[0][0] > 300:
            _, value = self.data.popleft()
            self.cur_sum -= value

    def now(self) -> int:
        raise NotImplementedError


def moving_average_results(events: list[list]) -> list[float]:
    obj = MovingAvg()
    current_time = 0

    def fake_now():
        return current_time

    obj.now = fake_now
    result = []

    for event in events:
        if event[0] == 'record':
            current_time = event[1]
            obj.record(event[2])
        elif event[0] == 'mean':
            current_time = event[1]
            result.append(obj.mean())

    return result