82 lines
2.3 KiB
Python
82 lines
2.3 KiB
Python
rot13_mapping = {}
|
|
for a, b, c, d in zip("anAN05", "mzMZ49", "naNA50", "zmZM94"):
|
|
rot13_mapping.update(dict((chr(k), chr(v))
|
|
for k, v in zip(range(ord(a), ord(b) + 1),
|
|
range(ord(c), ord(d) + 1))))
|
|
|
|
|
|
def right_now():
|
|
from datetime import datetime, timezone
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def nonsense_consistent(domain):
|
|
from random import Random
|
|
from string import ascii_lowercase
|
|
from zlib import crc32
|
|
rng = Random(crc32(domain.encode("utf-8")))
|
|
length = rng.choices((9, 10, 11, 12), (4, 5, 3, 2))[0]
|
|
return "".join(rng.choice(ascii_lowercase) for i in range(length))
|
|
|
|
|
|
def rot13(s):
|
|
return "".join(rot13_mapping.get(c, c) for c in s)
|
|
|
|
|
|
def concat_nonsense(domain):
|
|
return nonsense_consistent(domain) + "." + domain
|
|
|
|
|
|
def head(n, it):
|
|
res = []
|
|
try:
|
|
while len(res) < n:
|
|
res.append(next(it))
|
|
except StopIteration:
|
|
pass
|
|
return res
|
|
|
|
|
|
def taskize(item):
|
|
from types import CoroutineType
|
|
from asyncio import Task, create_task
|
|
|
|
if isinstance(item, CoroutineType):
|
|
assert not isinstance(item, Task) # TODO: paranoid?
|
|
item = create_task(item)
|
|
return item
|
|
|
|
|
|
def make_pooler(pool_size, finisher=None):
|
|
# TODO: write a less confusing interface
|
|
# that allows the code to be written more flatly.
|
|
# maybe like: async for done in apply(doit, [tuple_of_args]):
|
|
from asyncio import wait, FIRST_COMPLETED
|
|
|
|
pending = set()
|
|
|
|
async def pooler(item=None):
|
|
nonlocal pending
|
|
finish = item is None
|
|
if not finish:
|
|
pending.add(taskize(item))
|
|
desired_size = 0 if finish else pool_size - 1
|
|
while len(pending) > desired_size:
|
|
done, pending = await wait(pending, return_when=FIRST_COMPLETED)
|
|
finisher(done, pending)
|
|
|
|
return pooler
|
|
|
|
|
|
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)
|