from compression import gzip

import concurrent.futures

from pathlib import Path

import html.parser

import functools

import argparse

import json

import sys

import re

IGNORED_ID_RE = re.compile(

r"""

index-\d+

| id\d+

| [_a-z]+_\d+

""",

re.VERBOSE,

)

class Redirect(Exception): # noqa: N818 Exception should be named with an Error suffix

def __init__(self, redirect_to):

self.redirect_to = redirect_to

class IDGatherer(html.parser.HTMLParser):

def __init__(self, ids):

super().__init__()

self.__ids = ids

def handle_starttag(self, tag, attrs):

if tag == "meta":

# Redirects are done with a meta tag:

# <meta http-equiv="refresh" content="0; url=../builtins/stdtypes.html"/>

dattr = dict(attrs)

if dattr.get("http-equiv") == "refresh":

content = dattr.get("content", "")

if content.startswith("0; url="):

redirect_to = content[7:]

raise Redirect(redirect_to)

for name, value in attrs:

if name == 'id':

if not IGNORED_ID_RE.fullmatch(value):

self.__ids.add(value)

@functools.cache

def get_ids_from_file(path):

ids = set()

gatherer = IDGatherer(ids)

with path.open(encoding='utf-8') as file:

while chunk := file.read(4096):

gatherer.feed(chunk)

return ids

def get_ids_including_redirects(path):

# Only try 6 redirects, to avoid accidental endless loops

for _ in range(6):

try:

return get_ids_from_file(path)

except Redirect as r:

path = (path.parent / r.redirect_to).resolve()

continue

else:

raise RuntimeError("Apparent infinite redirects")

def gather_ids(htmldir, *, verbose_print):

if not htmldir.joinpath('objects.inv').exists():

raise ValueError(f'{htmldir!r} is not a Sphinx HTML output directory')

if sys._is_gil_enabled:

pool = concurrent.futures.ProcessPoolExecutor()

else:

pool = concurrent.futures.ThreadPoolExecutor()

tasks = {}

for path in htmldir.glob('**/*.html'):

relative_path = path.relative_to(htmldir)

if '_static' in relative_path.parts:

continue

if 'whatsnew' in relative_path.parts:

continue

tasks[relative_path] = pool.submit(

get_ids_including_redirects, path=path

)

ids_by_page = {}

for relative_path, future in tasks.items():

verbose_print(relative_path)

ids = future.result()

ids_by_page[str(relative_path)] = ids

verbose_print(f' - {len(ids)} ids found')

common = set.intersection(*ids_by_page.values())

verbose_print(f'Filtering out {len(common)} common ids')

for key, page_ids in ids_by_page.items():

ids_by_page[key] = sorted(page_ids - common)

return ids_by_page

def do_check(baseline, checked, excluded, *, verbose_print):

successful = True

for name, baseline_ids in sorted(baseline.items()):

try:

checked_ids = checked[name]

except KeyError:

if (name, '(page missing)') not in excluded:

successful = False

print(f'{name}: (page missing)')

print()

else:

missing_ids = set(baseline_ids) - set(checked_ids)

if missing_ids:

missing_ids = {

a

for a in missing_ids

if not IGNORED_ID_RE.fullmatch(a)

and (name, a) not in excluded

}

if missing_ids:

successful = False

for missing_id in sorted(missing_ids):

print(f'{name}: {missing_id}')

print()

return successful

def main(argv):

parser = argparse.ArgumentParser()

parser.add_argument(

'-v',

'--verbose',

action='store_true',

help='print out more information',

)

subparsers = parser.add_subparsers(dest='command', required=True)

collect = subparsers.add_parser(

'collect', help='collect IDs from a set of HTML files'

)

collect.add_argument(

'htmldir', type=Path, help='directory with HTML documentation'

)

collect.add_argument(

'-o',

'--outfile',

help='File to save the result in; default <htmldir>/html-ids.json.gz',

)

check = subparsers.add_parser('check', help='check two archives of IDs')

check.add_argument(

'baseline_file', type=Path, help='file with baseline IDs'

)

check.add_argument('checked_file', type=Path, help='file with checked IDs')

check.add_argument(

'-x',

'--exclude-file',

type=Path,

help='file with IDs to exclude from the check',

)

args = parser.parse_args(argv[1:])

if args.verbose:

verbose_print = functools.partial(print, file=sys.stderr)

else:

def verbose_print(*args, **kwargs):

"""do nothing"""

if args.command == 'collect':

ids = gather_ids(args.htmldir, verbose_print=verbose_print)

if args.outfile is None:

args.outfile = args.htmldir / 'html-ids.json.gz'

with gzip.open(args.outfile, 'wt', encoding='utf-8') as zfile:

json.dump({'ids_by_page': ids}, zfile)

if args.command == 'check':

with gzip.open(args.baseline_file) as zfile:

baseline = json.load(zfile)['ids_by_page']

with gzip.open(args.checked_file) as zfile:

checked = json.load(zfile)['ids_by_page']

excluded = set()

if args.exclude_file:

with open(args.exclude_file, encoding='utf-8') as file:

for line in file:

line = line.strip()

if line and not line.startswith('#'):

name, sep, excluded_id = line.partition(':')

if sep:

excluded.add((name.strip(), excluded_id.strip()))

if do_check(baseline, checked, excluded, verbose_print=verbose_print):

verbose_print('All OK')

else:

sys.stdout.flush()

print(

'ERROR: Removed IDs found',

'The above HTML IDs were removed from the documentation, '

+ 'resulting in broken links. Please add them back.',

sep='\n',

file=sys.stderr,

)

if args.exclude_file:

print(f'Alternatively, add them to {args.exclude_file}.')

sys.exit(1)

if __name__ == '__main__':

main(sys.argv)