Check a path against a fetched robots.txt

#!/usr/bin/env python3
'Check a path against a fetched robots.txt'
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='Check a path against a fetched robots.txt')
    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("--base-url",required=True)
    parser.add_argument("--path",required=True)
    parser.add_argument("--user-agent",default="ScriptageLearning")
    args = parser.parse_args()
    if not 0<args.timeout<=120: raise ValueError("Timeout must be greater than zero and at most 120.")
    base=urllib.parse.urlsplit(url(args.base_url))
    origin=urllib.parse.urlunsplit((base.scheme,base.netloc,"/","",""))
    robots_url=urllib.parse.urljoin(origin,"robots.txt")
    _,_,data=checked(robots_url,args.timeout)
    rules=urllib.robotparser.RobotFileParser()
    rules.set_url(robots_url)
    rules.parse(data.decode("utf-8-sig").splitlines())
    target=urllib.parse.urljoin(origin,args.path)
    if urllib.parse.urlsplit(target).netloc!=base.netloc: raise ValueError("Path must stay on the supplied origin.")
    write_json(args.output,{"robots_url":robots_url,"target":target,"user_agent":args.user_agent,"allowed":rules.can_fetch(args.user_agent,target)})

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