feat: just did the stuff
This commit is contained in:
@@ -2,12 +2,30 @@ import logging
|
||||
import sys
|
||||
|
||||
from . import PKG_NAME, PKG_VERSION
|
||||
from .cache import Cache
|
||||
from .constants import LANGS
|
||||
from .factory import Factory
|
||||
from .logs import setup_logs
|
||||
from .params import parse_parameters
|
||||
from .tld_list import TLDList
|
||||
|
||||
|
||||
def main() -> int:
|
||||
params = parse_parameters(sys.argv[1:])
|
||||
setup_logs(params)
|
||||
logging.getLogger().info("%s %s", PKG_NAME, PKG_VERSION)
|
||||
with (
|
||||
Cache(filepath=params.cache_file, no_load=params.cache) as cache,
|
||||
TLDList(cache=cache, remote_uri=params.tld_list_uri) as tld_list,
|
||||
LANGS[params.lang] as word_list,
|
||||
):
|
||||
factory = Factory(tld_list=tld_list, word_list=word_list)
|
||||
logging.getLogger().info(
|
||||
"Searching %s domain%s", params.count, "s" if params.count > 1 else ""
|
||||
)
|
||||
for _ in range(params.count):
|
||||
if params.dig:
|
||||
logging.getLogger().info(factory.random_available_domain())
|
||||
else:
|
||||
logging.getLogger().info(factory.random_domain())
|
||||
return 0
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
import json
|
||||
import pathlib
|
||||
import typing
|
||||
|
||||
T = typing.TypeVar("T")
|
||||
|
||||
|
||||
class Cache:
|
||||
__slots__ = ["__data", "__filepath", "__no_load"]
|
||||
|
||||
def __init__(self, filepath: str, no_load: bool) -> None: # noqa: FBT001
|
||||
self.__filepath = pathlib.Path(filepath)
|
||||
self.__no_load = no_load
|
||||
self.__data: dict[str, typing.Any] | None = None
|
||||
|
||||
def load(self) -> None:
|
||||
if not self.__no_load and self.__filepath.exists():
|
||||
self.__data = json.loads(self.__filepath.read_text())
|
||||
else:
|
||||
self.__data = {}
|
||||
|
||||
def write(self) -> None:
|
||||
self.__filepath.write_text(json.dumps(self.__data))
|
||||
|
||||
def __enter__(self) -> typing.Self:
|
||||
self.load()
|
||||
return self
|
||||
|
||||
def __exit__(self, *_: object) -> None:
|
||||
self.write()
|
||||
|
||||
def has(self, key: str) -> bool:
|
||||
return self.__data is not None and key in self.__data
|
||||
|
||||
def get(self, key: str) -> typing.Any:
|
||||
if self.__data is not None and key in self.__data:
|
||||
return self.__data[key]
|
||||
return None
|
||||
|
||||
def set(self, key: str, value: typing.Any) -> None:
|
||||
if self.__data is not None:
|
||||
self.__data[key] = value
|
||||
self.write()
|
||||
|
||||
def get_or_set(self, key: str, callback: typing.Callable[[], T]) -> T:
|
||||
if not self.has(key):
|
||||
self.set(key, callback())
|
||||
return self.get(key)
|
||||
@@ -0,0 +1,6 @@
|
||||
from .word_list import WordList
|
||||
|
||||
LANGS: dict[str, WordList] = {
|
||||
"english": WordList("english", "english", "words_alpha.txt"),
|
||||
"french": WordList("french", "french", "francais.txt"),
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
import logging
|
||||
import random
|
||||
import shutil
|
||||
import subprocess
|
||||
import typing
|
||||
|
||||
if typing.TYPE_CHECKING:
|
||||
from .tld_list import TLDList
|
||||
from .word_list import WordList
|
||||
|
||||
|
||||
class FactoryError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class Factory:
|
||||
__slots__ = ["__found", "__tld_list", "__word_list", "logger", "logger"]
|
||||
|
||||
def __init__(self, tld_list: TLDList, word_list: WordList) -> None:
|
||||
self.logger: logging.Logger = logging.getLogger(self.__class__.__name__)
|
||||
self.__tld_list = tld_list
|
||||
self.__word_list = word_list
|
||||
self.__found: set[str] = set()
|
||||
|
||||
def __iter__(self) -> typing.Self:
|
||||
return self
|
||||
|
||||
def __next__(self) -> str:
|
||||
return self.random_domain()
|
||||
|
||||
@property
|
||||
def __tld_list_shuffled(self) -> typing.Iterable[str]:
|
||||
if self.__tld_list.tlds is None:
|
||||
msg = "TLD list is not loaded"
|
||||
raise FactoryError(msg)
|
||||
return random.sample(self.__tld_list.tlds, len(self.__tld_list.tlds))
|
||||
|
||||
def __is_domain_available(self, domain: str) -> bool:
|
||||
dig_bin = shutil.which("dig")
|
||||
if dig_bin is None:
|
||||
msg = "dig is not installed"
|
||||
raise FactoryError(msg)
|
||||
try:
|
||||
output = subprocess.check_output([dig_bin, domain, "+short"]) # noqa: S603
|
||||
return len(output) == 0
|
||||
except subprocess.CalledProcessError:
|
||||
return True
|
||||
|
||||
def random_domain(self) -> str:
|
||||
k = 10000
|
||||
while (k := k - 1) > 0 and (domain := self.__random_domain()) is None:
|
||||
pass
|
||||
if domain is None:
|
||||
raise StopIteration
|
||||
return domain
|
||||
|
||||
def random_available_domain(self) -> str:
|
||||
k = 10000
|
||||
while (k := k - 1) > 0 and not self.__is_domain_available(
|
||||
domain := self.random_domain()
|
||||
):
|
||||
pass
|
||||
if k <= 0:
|
||||
raise StopIteration
|
||||
return domain
|
||||
|
||||
def __random_domain(self) -> str | None:
|
||||
word = random.choice(self.__word_list.words) # noqa: S311
|
||||
for tld in self.__tld_list_shuffled:
|
||||
tld_escaped = tld.replace(".", "")
|
||||
if word.endswith(tld_escaped) and len(word) > len(tld_escaped):
|
||||
domain = word[: -len(tld_escaped)] + "." + tld
|
||||
if domain not in self.__found:
|
||||
self.__found.add(domain)
|
||||
return domain
|
||||
return None
|
||||
+110
-1
@@ -1,31 +1,140 @@
|
||||
import argparse
|
||||
import dataclasses
|
||||
import os
|
||||
import pathlib
|
||||
import tempfile
|
||||
import typing
|
||||
|
||||
from . import PKG_NAME
|
||||
from .constants import LANGS
|
||||
|
||||
|
||||
@dataclasses.dataclass(frozen=True, slots=True)
|
||||
class Parameters:
|
||||
debug: bool = False
|
||||
quiet: bool = False
|
||||
cache: bool = True
|
||||
count: int = 1
|
||||
dig: bool = True
|
||||
lang: str = "english"
|
||||
cache_file: str = str(pathlib.Path(tempfile.gettempdir()) / ".cool-domain-cache")
|
||||
tld_list_uri: str = "https://publicsuffix.org/list/public_suffix_list.dat"
|
||||
|
||||
@classmethod
|
||||
def from_namespace(cls, args: argparse.Namespace) -> Parameters:
|
||||
return Parameters(**vars(args))
|
||||
|
||||
|
||||
def __get_env_str(var: str, default: str) -> str:
|
||||
if (result := os.getenv(var)) is None:
|
||||
return default
|
||||
return result
|
||||
|
||||
|
||||
def __get_env_int(var: str, default: int) -> int:
|
||||
value = __get_env_str(var, str(default))
|
||||
if value.isdecimal():
|
||||
return int(value)
|
||||
return default
|
||||
|
||||
|
||||
def __add_arg_str(
|
||||
parser: argparse.ArgumentParser,
|
||||
*flags: str,
|
||||
env_var: str,
|
||||
default: str,
|
||||
help_txt: str,
|
||||
) -> None:
|
||||
parser.add_argument(
|
||||
*flags,
|
||||
metavar=env_var,
|
||||
default=__get_env_str(env_var, default),
|
||||
help=f"{help_txt} (default: {default})" if len(default) else help_txt,
|
||||
)
|
||||
|
||||
|
||||
def __add_arg_int(
|
||||
parser: argparse.ArgumentParser,
|
||||
*flags: str,
|
||||
env_var: str,
|
||||
default: int,
|
||||
help_txt: str,
|
||||
) -> None:
|
||||
parser.add_argument(
|
||||
*flags,
|
||||
type=int,
|
||||
metavar=env_var,
|
||||
default=__get_env_int(env_var, default),
|
||||
help=f"{help_txt} (default: {default})",
|
||||
)
|
||||
|
||||
|
||||
def __add_arg_bool(
|
||||
parser: argparse.ArgumentParser,
|
||||
*flags: str,
|
||||
default: bool,
|
||||
help_txt: str,
|
||||
dest: str | None = None,
|
||||
) -> None:
|
||||
parser.add_argument(
|
||||
*flags,
|
||||
action=argparse.BooleanOptionalAction,
|
||||
help=f"{help_txt} (default: {str(default).lower()})",
|
||||
default=default,
|
||||
dest=dest,
|
||||
)
|
||||
|
||||
|
||||
def parse_parameters(args: typing.Sequence[str]) -> Parameters:
|
||||
default_values = Parameters()
|
||||
parser = argparse.ArgumentParser(
|
||||
PKG_NAME,
|
||||
PKG_NAME.replace("_", "-"),
|
||||
suggest_on_error=True,
|
||||
)
|
||||
__add_arg_int(
|
||||
parser,
|
||||
"--count",
|
||||
env_var="DOMAIN_COUNT",
|
||||
default=default_values.count,
|
||||
help_txt="number of domains to generate",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--lang",
|
||||
default=default_values.lang,
|
||||
help="word list lang (default: english)",
|
||||
choices=LANGS.keys(),
|
||||
)
|
||||
__add_arg_bool(
|
||||
parser,
|
||||
"--dig",
|
||||
default=default_values.dig,
|
||||
help_txt="check availability with dig",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--debug", action=argparse.BooleanOptionalAction, default=default_values.debug
|
||||
)
|
||||
parser.add_argument(
|
||||
"--quiet", action=argparse.BooleanOptionalAction, default=default_values.quiet
|
||||
)
|
||||
__add_arg_bool(
|
||||
parser,
|
||||
"--cache",
|
||||
default=default_values.cache,
|
||||
help_txt="use cache",
|
||||
)
|
||||
__add_arg_str(
|
||||
parser,
|
||||
"--cache-file",
|
||||
env_var="CACHE_FILE",
|
||||
default=default_values.cache_file,
|
||||
help_txt="cache file location",
|
||||
)
|
||||
__add_arg_str(
|
||||
parser,
|
||||
"--tld-list-uri",
|
||||
env_var="TLD_LIST_URI",
|
||||
default=default_values.tld_list_uri,
|
||||
help_txt="Top Level Domains list",
|
||||
)
|
||||
parsed_args = parser.parse_args(args)
|
||||
return Parameters.from_namespace(parsed_args)
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
import http
|
||||
import logging
|
||||
import typing
|
||||
|
||||
import requests
|
||||
import unidecode
|
||||
|
||||
if typing.TYPE_CHECKING:
|
||||
from .cache import Cache
|
||||
|
||||
|
||||
class TLDError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class TLDList:
|
||||
__slots__ = ["__cache", "__remote_uri", "logger", "tlds"]
|
||||
|
||||
def __init__(self, cache: Cache, remote_uri: str) -> None:
|
||||
self.logger: logging.Logger = logging.getLogger(self.__class__.__name__)
|
||||
self.tlds: list[str] | None = None
|
||||
self.__cache = cache
|
||||
self.__remote_uri = remote_uri
|
||||
|
||||
def __load_remote(self) -> list[str]:
|
||||
response = requests.get(self.__remote_uri, timeout=5)
|
||||
if response.status_code != http.HTTPStatus.OK:
|
||||
msg = f"Failed to read {self.__remote_uri}."
|
||||
raise TLDError(msg)
|
||||
raw_lines = response.content.decode().split("\n")
|
||||
out_lines = []
|
||||
for line in raw_lines:
|
||||
stripped_line = line.strip()
|
||||
if (
|
||||
len(stripped_line)
|
||||
and not stripped_line.startswith("//")
|
||||
and not stripped_line.startswith("*")
|
||||
and unidecode.unidecode(stripped_line) == stripped_line
|
||||
):
|
||||
if "." in stripped_line:
|
||||
for other in out_lines:
|
||||
if stripped_line.endswith(other):
|
||||
break
|
||||
else:
|
||||
out_lines += [stripped_line]
|
||||
else:
|
||||
out_lines += [stripped_line]
|
||||
return out_lines
|
||||
|
||||
def load(self) -> None:
|
||||
self.tlds = self.__cache.get_or_set("tld_list", self.__load_remote)
|
||||
self.logger.info("Loaded %d TLD entries", len(self.tlds))
|
||||
|
||||
def __enter__(self) -> typing.Self:
|
||||
self.load()
|
||||
return self
|
||||
|
||||
def __exit__(self, *_: object) -> None:
|
||||
pass
|
||||
@@ -0,0 +1,35 @@
|
||||
import logging
|
||||
import pathlib
|
||||
import typing
|
||||
|
||||
import unidecode
|
||||
|
||||
|
||||
class WordList:
|
||||
__slots__ = ["__filepath", "__name", "logger", "words"]
|
||||
|
||||
def __init__(self, name: str, path: str, filename: str) -> None:
|
||||
self.logger: logging.Logger = logging.getLogger(self.__class__.__name__)
|
||||
self.__name = name
|
||||
self.__filepath = pathlib.Path(__file__).parent / ".." / path / filename
|
||||
self.words: list[str] = []
|
||||
|
||||
def open(self) -> None:
|
||||
with self.__filepath.open(encoding="utf-8") as file:
|
||||
self.words = list(
|
||||
filter(lambda v: len(v) > 0, map(self.__escape_word, file.readlines()))
|
||||
)
|
||||
self.logger.info("[%s] Loaded %d words", self.__name, len(self.words))
|
||||
|
||||
def __escape_word(self, word: str) -> str:
|
||||
return unidecode.unidecode(word.strip()).lower()
|
||||
|
||||
def close(self) -> None:
|
||||
pass
|
||||
|
||||
def __enter__(self) -> typing.Self:
|
||||
self.open()
|
||||
return self
|
||||
|
||||
def __exit__(self, *_: object) -> None:
|
||||
self.close()
|
||||
Reference in New Issue
Block a user