1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

import usocket
# 目标服务器信息
#host = "10.1.1.75"
#port = 8084
#url = "/devs/?cid=xxx"
# timeout =1
def HttpGet(host,port,url,timeout):
    try:
        # 创建 socket 连接
        s = usocket.socket(usocket.AF_INET, usocket.SOCK_STREAM)
        addr = usocket.getaddrinfo(host, port)[0][-1]
        s.connect(addr)
        # 构建 GET 请求
        request = "GET {url} HTTP/1.1\r\nHost: {host}\r\nUser-Agent: myUrequests\r\n\r\n".format(url=url, host=host)
        s.send(request) # 发送请求
        response = b"" # 接收响应
        s.settimeout(timeout)  # 设置超时时间为x秒
        while True:
            try:
                data = s.recv(4096)
                if not data:
                    break
                response += data
            except OSError:
                break
        # 关闭 socket 连接
        s.close()
        # 解析响应数据
        headers, body = response.split(b'\r\n\r\n', 1)
        content_length = 0
        for header in headers.split(b'\r\n'):
            if header.startswith(b'Content-Length:'):
                content_length = int(header.split(b':')[1])
                break
        return body[:content_length].decode()
    except:
        print("http get错误")
        return ""
        
def HttpsGet(host, port, url, timeout):
    import ussl
    import select
    import time
    # 创建 socket 连接
    s = usocket.socket(usocket.AF_INET, usocket.SOCK_STREAM)
    addr = usocket.getaddrinfo(host, port)[0][-1]
    s.connect(addr)

    # 使用 SSL 加密连接
    s = ussl.wrap_socket(s, server_hostname=host)

    # 设置超时时间
    s.setblocking(False)

    # 构建 GET 请求
    request = "GET {url} HTTP/1.1\r\nHost: {host}\r\nUser-Agent: myUrequests\r\n\r\n".format(url=url, host=host)
    s.write(request.encode())  # 发送请求

    # 等待响应数据
    response = b""
    start_time = time.time()
    while True:
        ready = select.select([s], [], [], 0.1)  # 每次等待 0.1 秒钟
        if ready[0]:
            try:
                data = s.read(4096)
                if data:
                    response += data
                else:
                    break
            except OSError:
                break
        
        # 检查超时
        elapsed_time = time.time() - start_time
        if elapsed_time >= timeout:
            break

    # 关闭 socket 连接
    s.close()

    # 解析响应数据
    headers, body = response.split(b'\r\n\r\n', 1)
    content_length = 0
    for header in headers.split(b'\r\n'):
        if header.startswith(b'Content-Length:'):
            content_length = int(header.split(b':')[1])
            break

    return body[:content_length].decode()

def AutoHttpGet(url, timeout):
    try:
        # 解析URL
        parts = url.split('://')
        scheme = parts[0]
        remaining = parts[1].split('/', 1)
        host_port = remaining[0].split(':')
        if len(host_port) == 1: #默认端口
            host = host_port[0]
            port = 443 if scheme == 'https' else 80
        else:#自定义端口
            host = host_port[0]
            port = int(host_port[1])
        path = '/' + remaining[1] if len(remaining) > 1 else '/'
        query = ''
        if '?' in path:
            path, query = path.split('?', 1)
            query = '?' + query
        print(scheme,host, port, path + query)
        if scheme == 'https':
            return HttpsGet(host, port, path + query, timeout)
        else:
            return HttpGet(host, port, path + query, timeout)
    except:
        print("自动http/s get错误")
        return ""