whois.py

import re
import socket
import requests
import json
from concurrent.futures import ThreadPoolExecutor
from django.core.serializers.json import DjangoJSONEncoder


def convert_idna(n):
    try:
        domain = n.encode('idna').decode('utf-8')
        return domain
    except TypeError:
        domain = n.decode('utf-8').encode('idna').decode('utf-8')
        return domain
    except AttributeError:
        domain = n.decode('utf-8').encode('idna').decode('utf-8')
        return domain
    except UnicodeError:
        pass


def get_tld(n):
    domain = convert_idna(n).split('.')[-1]
    return domain


def get_length(n):
    domain = n.split('.')[0].__len__()
    return domain


class FindWhoIsServer(object):

    def __init__(self, domain):
        self.datas = {}
        self.domain = domain
        self.url = "http://www.iana.org/domains/root/db/{}.html".format(convert_idna(get_tld(self.domain)))

    def GetWhoIsServer(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            headers = {
                "User-Agent": 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3578.80 Safari/537.36',
                "Pragma": 'no-cache',
                "Host": 'www.iana.org'}
            r = requests.get(self.url, headers=headers, timeout=5)
            if r is not False:
                whois_server = re.findall('''<b>WHOIS Server:</b> (.*)''', r.text)
                registry_server = re.findall('''<b>URL for registration services:</b> <a[^>]+href=["'](.*?)["']''',
                                             r.text)
            if not whois_server:
                try:
                    server = "whois.nic.{domain}".format(domain=convert_idna(get_tld(self.domain)))
                    ip = socket.gethostbyname(server)
                    if ip != server:
                        whois_server = server
                except Exception:
                    return ''
            try:
                s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                s.connect((whois_server[0], 43))
                s.send(('domain ' + self.domain + '\r\n').encode())
                response = b""
                while True:
                    d = s.recv(80192)
                    response += d
                    if not d:
                        break
                s.close()
            except Exception:
                return ''

            self.datas["domain_name"] = self.domain
            self.datas["server"] = whois_server
            self.datas["registry"] = registry_server
            self.datas["domain_length"] = get_length(self.domain)
            self.datas["tld"] = get_tld(self.domain)
            self.datas["whois"] = response.decode('ISO-8859-1')
            return json.dumps(self.datas,
                              sort_keys=True,
                              indent=2,
                              cls=DjangoJSONEncoder)


print(FindWhoIsServer('google.com').GetWhoIsServer())