Test an HTTP conditional request

#!/usr/bin/env python3
'Test an HTTP conditional request'
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='Test an HTTP conditional request')
    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("--etag")
    parser.add_argument("--modified-since")
    args = parser.parse_args()
    if not 0<args.timeout<=120: raise ValueError("Timeout must be greater than zero and at most 120.")
    headers={}
    if args.etag: headers["If-None-Match"]=args.etag
    if args.modified_since: headers["If-Modified-Since"]=args.modified_since
    if not headers: raise ValueError("Provide --etag or --modified-since.")
    status,final,response,_=request(args.url,args.timeout,headers=headers,limit=0)
    write_json(args.output,{"url":final,"status":status,"not_modified":status==304,"etag":response.get("etag"),"last_modified":response.get("last-modified")})

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