Trace HTTP redirects step by step

#!/usr/bin/env python3
'Trace HTTP redirects step by step'
import argparse
import json
from pathlib import Path

def write_json(path, value):
    data = json.dumps(value, indent=2, ensure_ascii=False, allow_nan=False) + "\n"
    with Path(path).open("x", encoding="utf-8", newline="") as handle:
        handle.write(data)

import urllib.request
import urllib.error
import urllib.parse
import urllib.robotparser
import socket
import ssl
import time
import re
import ipaddress
import queue
import threading
from datetime import datetime, timezone

def url(value):
    parts=urllib.parse.urlsplit(value)
    if parts.scheme not in {"http","https"} or not parts.hostname or parts.username is not None or parts.password is not None:
        raise ValueError("Use an http(s) URL without embedded credentials.")
    return value
def request(address,timeout,method="GET",headers=None,limit=1024*1024,opener=None):
    url(address)
    if not 0<timeout<=120: raise ValueError("Timeout must be 0..120 seconds, excluding zero.")
    req=urllib.request.Request(address,method=method,headers={"User-Agent":"ScriptageLearning/1.0",**(headers or {})})
    try: response=(opener.open(req,timeout=timeout) if opener else urllib.request.urlopen(req,timeout=timeout))
    except urllib.error.HTTPError as exc: response=exc
    with response:
        data=response.read(limit+1) if limit else b""
        if len(data)>limit: raise ValueError("Response exceeds the byte limit.")
        return response.status,response.geturl(),{k.lower():v for k,v in response.headers.items()},data
def checked(address,timeout,**kwargs):
    status,final,headers,data=request(address,timeout,**kwargs)
    if not 200<=status<300: raise ValueError("HTTP status "+str(status))
    return final,headers,data
def urls(path):
    result=[line.strip() for line in Path(path).read_text(encoding="utf-8").splitlines() if line.strip()]
    if len(result)>100: raise ValueError("At most 100 URLs per run.")
    return result
def resolve(host,port,timeout):
    result=queue.Queue()
    def work():
        try: result.put((True,socket.getaddrinfo(host,port,type=socket.SOCK_STREAM)))
        except OSError as exc: result.put((False,str(exc)))
    threading.Thread(target=work,daemon=True).start()
    try: ok,value=result.get(timeout=timeout)
    except queue.Empty: raise ValueError("DNS lookup timed out.")
    if not ok: raise ValueError(value)
    return value
def connect(host,port,timeout):
    last=None
    for family,kind,proto,_,address in resolve(host,port,timeout):
        connection=socket.socket(family,kind,proto)
        connection.settimeout(timeout)
        try:
            connection.connect(address)
            return connection
        except OSError as exc:
            last=exc
            connection.close()
    raise OSError(str(last))

def main():
    parser = argparse.ArgumentParser(description='Trace HTTP redirects step by step')
    parser.add_argument("--output",required=True,help="New output file")
    parser.add_argument("--timeout",type=float,default=10,help="Per-operation timeout in seconds")
    parser.add_argument("--url",required=True)
    parser.add_argument("--max-hops",type=int,default=8)
    args = parser.parse_args()
    if not 0<args.timeout<=120: raise ValueError("Timeout must be greater than zero and at most 120.")
    if not 1<=args.max_hops<=20: raise ValueError("Max hops must be 1..20.")
    class NoRedirect(urllib.request.HTTPRedirectHandler):
        def redirect_request(self,req,fp,code,msg,headers,newurl): return None
    opener=urllib.request.build_opener(NoRedirect)
    address=args.url
    seen=set()
    trace=[]
    for _ in range(args.max_hops):
        if address in seen: raise ValueError("Redirect loop detected.")
        seen.add(address)
        status,_,headers,_=request(address,args.timeout,method="HEAD",limit=0,opener=opener)
        trace.append({"url":address,"status":status,"location":headers.get("location")})
        if status not in {301,302,303,307,308} or "location" not in headers: break
        address=urllib.parse.urljoin(address,headers["location"])
    else: trace.append({"stopped":"hop limit"})
    write_json(args.output,trace)

if __name__ == "__main__":
    try:
        main()
    except (OSError, ValueError, KeyError, TypeError, ArithmeticError, IndexError, LookupError) as exc:
        raise SystemExit(str(exc))