Please enable Javascript to view the contents

webdog 的二次蜕变

 ·  ☕  2 分钟

用了三天左右时间重构了 ‘webdog’,简单的整理后开源了 -> WebHunt

命令行的设计

首先是对命令行的设计,之前没有注意到这些细节,导致命令行的程序很难用,网上可以搜索一些”命令行的设计“文章或者项目看看。

命令行处理使用 click 模块,将其分为了 managescan 两个组:

具体如下:

manage:

Options:
  -d, --directory TEXT  Components directory, default ./components
  -l, --list            List components
  --pull                Pull custom components from remote database
  --pull_webanalyzer    Pull custom components from remote database
  --sync                Synchronize to remote database
  --sync-updating       Update existing components when synchronizing to
                        remote database

  --host TEXT           MySQL database host
  --port INTEGER        MySQL database port
  --db TEXT             MySQL database name
  --user TEXT           MySQL database user
  --passwd TEXT         MySQL database password
  --search TEXT         Search component name
  -v, --verbose         Output detailed debugging information
  --help                Show this message and exit.

scan:

Options:
  -u, --url TEXT             Target  [required]
  -d, --directory TEXT       Components directory, default ./components
  -a, --aggression           Open aggression mode
  -U, --user-agent TEXT      Custom user agent
  -H, --header TEXT          Pass custom header LINE to serve
  --disallow-redirect        Disallow redirect
  -c, --component TEXT       Specify component
  -t, --max-threads INTEGER  Set the maximum number of threads
  --proxy TEXT               Set proxy is like: '[HTTP/SOCKS4/SOCKS5]/[usernam
                             e]@[password]/[addr]:[port]'

  --proxy_rdns               Proxy uses rdns
  -v, --verbose              Output detailed debugging information
  --help                     Show this message and exit.

简单解决命令交互终端下的彩色输出

这样就不用安装第三方模块

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
# 定义 bcolors
class bcolors:
    HEADER = '\033[95m'
    OKBLUE = '\033[94m'
    OKGREEN = '\033[92m'
    WARNING = '\033[93m'
    FAIL = '\033[91m'
    ENDC = '\033[0m'
    BOLD = '\033[1m'
    UNDERLINE = '\033[4m'

# 配合 click.echo 使用
click.echo("%s*Count: %s%s" %
                   (bcolors.OKGREEN, " ".join("%s: %d" % (k, v) for k, v in count.items()), bcolors.ENDC))

尽可能的 Pythonic

  • 类方法的使用
1
2
3
4
5
6
7
8
...
class ComponentType(enum.Enum):
    others = 0
    ...

    @classmethod
    def all_kinds(cls) -> List:
        return [c.name for c in cls]
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
class Component:
    def __init__(self, info: Dict):
        self._info = info
    ...
    @classmethod
    def make(cls, path: str):
        """Make a instance of 'Component' from JSON file
        """
        try:
            with open(path, 'r', encoding="utf-8") as f:
                return cls(json.load(f))
        except Exception as err:
            logger.debug("'%s' make error: %s", cls.__name__, err)
        return None
  • 属性缓存装饰器

属性缓存,获取该属性时将该方法执行结果写入 obj.__dict__

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
class cached_property(object):
    ''' A property that is only computed once per instance and then replaces
        itself with an ordinary attribute. Deleting the attribute resets the
        property. '''

    def __init__(self, func):
        self.__doc__ = getattr(func, '__doc__')
        self.func = func

    def __get__(self, obj, cls):
        if obj is None:
            return self
        value = obj.__dict__[self.func.__name__] = self.func(obj)
        return value
  • 线程安全的属性装饰器
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
class synchronized_property(property):
    ''' A property that is thread safety.
    '''

    def __init__(self, func):
        self.__doc__ = getattr(func, '__doc__')
        self.func = func
        self.lock = threading.Lock()

    def __get__(self, obj, cls):
        if obj is None:
            return self
        with self.lock:
            return self.func(obj)
  • 使用 Mixin 类

例如 RequestManagerMixin 是个具有自己历史的线程安全请求管理器…

1
2
3
4
5
6
7
8
class RequestManagerMixin:
    """This is a thread safe request manager with its own history
    """
    request_manager_history = {}
    request_manager_lock = threading.Lock()

    def request(self, url: str, **kwargs) -> Optional[Dict]:
        ...

感觉代码质量比之前的项目好了不少 🤔。

目录