respodns/respodns/util.py

rot13_mapping = {}
for a, b, c, d in zip("anAN05", "mzMZ49", "naNA50", "zmZM94"):
    rot13_mapping.update({chr(k): chr(v)
                          for k, v in zip(range(ord(a), ord(b) + 1),
                                          range(ord(c), ord(d) + 1))})


def rot13(s):
    return "".join(rot13_mapping.get(c, c) for c in s)


def right_now():
    from datetime import datetime, timezone
    return datetime.now(timezone.utc)


def nonsense_consistent(domain, weekly=True):
    from datetime import datetime, timezone
    from random import Random
    from string import ascii_lowercase
    from zlib import crc32
    if weekly:
        week = datetime.now(timezone.utc).isocalendar().week
        week_bytes = bytes((week, week, week, week))
        seed = crc32(week_bytes + domain.encode("utf-8"))
    else:
        seed = crc32(domain.encode("utf-8"))
    rng = Random(seed)
    length = rng.choices((9, 10, 11, 12), (4, 5, 3, 2))[0]
    return "".join(rng.choice(ascii_lowercase) for i in range(length))


def concat_nonsense(domain, weekly=True):
    return nonsense_consistent(domain, weekly=weekly) + "." + domain


def head(n, it):
    # TODO: maybe just do return [a for _, a in zip(range(n), it)]
    res = []
    try:
        while len(res) < n:
            res.append(next(it))
    except StopIteration:
        pass
    return res


class AttrCheck:
    """
    Inheriting AttrCheck prevents accidentally setting attributes
    that don't already exist.
    """
    def __setattr__(self, name, value):
        # NOTE: hasattr doesn't do what we want here. dir does.
        if name.startswith("_") or name in dir(self):
            super().__setattr__(name, value)
        else:
            raise AttributeError(name)


async def _release_later(sem, time=1):
    from asyncio import sleep

    await sleep(time)
    sem.release()


class LimitPerSecond:
    def __init__(self, limit):
        from asyncio import BoundedSemaphore

        if type(limit) is not int:
            raise ValueError("limit must be int")
        assert limit > 0, limit

        self.limit = limit
        self.tasks = []
        self.sem = BoundedSemaphore(limit)

    async def __aenter__(self):
        #if self.sem.locked:
        #    from sys import stderr
        #    print("THROTTLING", file=stderr)
        await self.sem.acquire()

    async def __aexit__(self, exc_type, exc_value, traceback):
        from asyncio import create_task

        task = create_task(_release_later(self.sem))
        self.tasks.append(task)

    async def finish(self):
        for task in self.tasks:
            await task