#!/usr/bin/env python3
"""Compare local documentation snapshots. Standard library only; no network or storage."""
from __future__ import annotations

import argparse
from dataclasses import dataclass, field
from difflib import SequenceMatcher
from html import unescape
from html.parser import HTMLParser
import json
import os
from pathlib import Path
import re
import signal
import stat
import sys

INPUT_LIMIT = 65_536
OUTPUT_LIMIT = 32_768
MAX_OUTPUT_LIMIT = 65_536
MAX_CHANGES = 100


class DocDeltaError(Exception):
    def __init__(self, code: str, input_name: str | None = None):
        super().__init__(code)
        self.code, self.input_name = code, input_name


@dataclass
class Unit:
    text: str
    start: int
    end: int
    heading_level: int = 0


@dataclass
class Section:
    heading: str | None
    level: int
    occurrence: int
    start: int
    end: int
    units: list[Unit] = field(default_factory=list)

    @property
    def key(self):
        return self.level, self.heading, self.occurrence


def markdown_units(text: str) -> list[Unit]:
    units, fence, code, code_start = [], None, [], 0
    for number, line in enumerate(text.splitlines(), 1):
        marker = re.match(r"^ {0,3}(`{3,}|~{3,})(.*)$", line)
        if fence is not None:
            code.append(line)
            if marker and marker[1][0] == fence[0] and len(marker[1]) >= fence[1] and not marker[2].strip():
                units.append(Unit("\n".join(code), code_start, number))
                fence, code = None, []
            continue
        if marker:
            fence, code, code_start = (marker[1][0], len(marker[1])), [line], number
            continue
        heading = re.match(r"^ {0,3}(#{1,6})(?:[ \t]+(.*)|[ \t]*)$", line)
        if heading:
            title = re.sub(r"[ \t]+#+[ \t]*$", "", heading[2] or "").strip()
            units.append(Unit(title, number, number, len(heading[1])))
        else:
            units.append(Unit(line, number, number))
    if code:
        units.append(Unit("\n".join(code), code_start, len(text.splitlines())))
    return units


class HTMLUnits(HTMLParser):
    BLOCKS = frozenset(("p", "div", "section", "article", "main", "header", "footer", "nav",
                        "aside", "li", "ul", "ol", "blockquote", "table", "tr", "td", "th", "dl", "dt", "dd", "form"))
    IGNORED = frozenset(("script", "style", "template"))
    URL_ATTRIBUTES = frozenset(("href", "src", "srcset", "action", "poster", "cite", "data"))

    def __init__(self):
        super().__init__(convert_charrefs=False)
        self.units, self.parts, self.skip, self.links = [], [], [], []
        self.start, self.end, self.heading_level, self.preserve = None, 0, 0, 0

    def append(self, text: str, protected: bool = False, end: int | None = None):
        line = self.getpos()[0]
        if self.start is None:
            self.start = line
        self.end = max(self.end, end if end is not None else line)
        self.parts.append((text, protected or self.preserve > 0))

    def flush(self, end: int | None = None, allow_empty: bool = False):
        preserved = []
        pieces = []
        for text, protected in self.parts:
            if protected:
                pieces.append("\0" + str(len(preserved)) + "\0")
                preserved.append(text)
            else:
                pieces.append(text)
        normalized = re.sub(r"\s+", " ", "".join(pieces)).strip()
        normalized = re.sub(r"\0([0-9]+)\0", lambda match: preserved[int(match[1])], normalized)
        if normalized or allow_empty:
            first = self.start if self.start is not None else self.getpos()[0]
            last = max(first, self.end, end or 0)
            self.units.append(Unit(normalized, first, last, self.heading_level))
        self.parts, self.start, self.end = [], None, 0

    def handle_starttag(self, tag, attrs):
        if self.skip:
            if tag in self.IGNORED:
                self.skip.append(tag)
            return
        if tag in self.IGNORED:
            self.skip.append(tag)
            return
        line = self.getpos()[0]
        heading = re.fullmatch(r"h([1-6])", tag)
        if heading:
            self.flush()
            self.heading_level, self.start = int(heading[1]), line
        elif tag in self.BLOCKS or tag == "pre":
            self.flush()
            self.start = line
        values = dict(attrs)
        if tag == "pre":
            self.preserve += 1
            self.append("<pre>", True)
        elif tag == "code":
            self.preserve += 1
            self.append("<code>", True)
        elif tag == "br":
            if self.preserve:
                self.append("\n", True)
            elif self.heading_level:
                self.append(" ")
            else:
                self.flush(end=line)
        elif tag == "hr":
            self.flush()
            self.append("<hr>", True)
            self.flush(end=line)
        elif tag == "a":
            target = values.get("href")
            self.links.append(target)
            if target is not None:
                self.append("[", True)
        elif tag == "img":
            self.append("[image " + (values.get("alt") or "") + "](" + (values.get("src") or "") + ")", True)
        targets = [(key, value) for key, value in attrs
                   if key in self.URL_ATTRIBUTES and value is not None
                   and not (tag == "a" and key == "href") and not (tag == "img" and key == "src")]
        if targets:
            self.append("<" + tag + " " + json.dumps(sorted(targets), ensure_ascii=False) + ">", True)

    def handle_endtag(self, tag):
        if self.skip:
            if tag == self.skip[-1]:
                self.skip.pop()
            return
        line = self.getpos()[0]
        if tag == "a" and self.links:
            target = self.links.pop()
            if target is not None:
                self.append("](" + target + ")", True)
        if tag in ("code", "pre"):
            self.append("</" + tag + ">", True)
            self.preserve = max(0, self.preserve - 1)
        if re.fullmatch(r"h[1-6]", tag):
            self.flush(end=line, allow_empty=True)
            self.heading_level = 0
        elif tag in self.BLOCKS or tag == "pre":
            self.flush(end=line)

    def handle_startendtag(self, tag, attrs):
        self.handle_starttag(tag, attrs)
        self.handle_endtag(tag)

    def handle_data(self, data):
        if not self.skip:
            self.append(data, end=self.getpos()[0] + data.count("\n"))

    def handle_entityref(self, name):
        if not self.skip:
            self.append(unescape("&" + name + ";"))

    def handle_charref(self, name):
        if not self.skip:
            self.append(unescape("&#" + name + ";"))


def html_units(text: str) -> list[Unit]:
    parser = HTMLUnits()
    parser.feed(text)
    parser.close()
    parser.flush()
    return parser.units


def sections_from_units(units: list[Unit]) -> list[Section]:
    sections, occurrences = [], {}
    current = Section(None, 0, 1, 1, 1)
    for unit in units:
        if unit.heading_level:
            if current.heading is not None or current.units:
                sections.append(current)
            base = unit.heading_level, unit.text
            occurrences[base] = occurrences.get(base, 0) + 1
            current = Section(unit.text, unit.heading_level, occurrences[base], unit.start, unit.end)
        else:
            if current.heading is None and not current.units:
                current.start = unit.start
            current.units.append(unit)
            current.end = max(current.end, unit.end)
    if current.heading is not None or current.units:
        sections.append(current)
    return sections


def span(units: list[Unit]):
    if not units:
        return None
    return {"startLine": units[0].start, "endLine": units[-1].end,
            "text": "\n".join(unit.text for unit in units)}


def section_anchor(section: Section | None):
    return None if section is None else {"startLine": section.start, "endLine": section.end}


def section_delta(old: Section | None, new: Section | None):
    if old is not None and new is not None:
        before, after = [u.text for u in old.units], [u.text for u in new.units]
        if before == after:
            return None
        edits = []
        for tag, a, b, c, d in SequenceMatcher(None, before, after, autojunk=False).get_opcodes():
            if tag != "equal":
                edits.append({"kind": {"replace": "changed", "delete": "removed", "insert": "added"}[tag],
                              "old": span(old.units[a:b]), "new": span(new.units[c:d])})
        kind, section = "changed", new
    else:
        kind, section = ("added", new) if new is not None else ("removed", old)
        body = section.units
        edits = [{"kind": kind, "old": span(body) if old is not None else None,
                  "new": span(body) if new is not None else None}] if body else []
    return {"kind": kind, "heading": section.heading, "level": section.level,
            "occurrence": section.occurrence, "old": section_anchor(old),
            "new": section_anchor(new), "edits": edits}


def compare(before: str, after: str, before_format: str, after_format: str,
            max_changes: int = MAX_CHANGES):
    parser = {"markdown": markdown_units, "html": html_units}
    old = sections_from_units(parser[before_format](before))
    new = sections_from_units(parser[after_format](after))
    changes = []
    def add(a, b):
        delta = section_delta(a, b)
        if delta is not None:
            changes.append(delta)
            if len(changes) > max_changes or sum(len(x["edits"]) for x in changes) > max_changes:
                raise DocDeltaError("CHANGE_LIMIT")
    for tag, a, b, c, d in SequenceMatcher(None, [s.key for s in old], [s.key for s in new], autojunk=False).get_opcodes():
        if tag == "equal":
            for left, right in zip(old[a:b], new[c:d]):
                add(left, right)
        else:
            for section in old[a:b]:
                add(section, None)
            for section in new[c:d]:
                add(None, section)
    return {"schemaVersion": 1, "status": "ok", "changed": bool(changes),
            "formats": {"old": before_format, "new": after_format},
            "summary": {"addedSections": sum(x["kind"] == "added" for x in changes),
                        "removedSections": sum(x["kind"] == "removed" for x in changes),
                        "changedSections": sum(x["kind"] == "changed" for x in changes),
                        "edits": sum(len(x["edits"]) for x in changes)},
            "sections": changes}


def read_snapshot(path: str, label: str):
    try:
        fd = os.open(path, os.O_RDONLY | os.O_NONBLOCK)
        with os.fdopen(fd, "rb") as stream:
            info = os.fstat(stream.fileno())
            if not stat.S_ISREG(info.st_mode):
                raise DocDeltaError("REGULAR_FILE_REQUIRED", label)
            if info.st_size > INPUT_LIMIT:
                raise DocDeltaError("INPUT_TOO_LARGE", label)
            raw = stream.read(INPUT_LIMIT + 1)
    except OSError:
        raise DocDeltaError("INPUT_READ_FAILED", label) from None
    if len(raw) > INPUT_LIMIT:
        raise DocDeltaError("INPUT_TOO_LARGE", label)
    try:
        text = raw.decode("utf-8-sig", errors="strict")
    except UnicodeDecodeError:
        raise DocDeltaError("INVALID_UTF8", label) from None
    if "\0" in text:
        raise DocDeltaError("NUL_CHARACTER", label)
    return text


def bounded_integer(low, high):
    def parse(value):
        try:
            result = int(value)
        except ValueError:
            raise argparse.ArgumentTypeError("must be an integer") from None
        if not low <= result <= high:
            raise argparse.ArgumentTypeError(f"must be between {low} and {high}")
        return result
    return parse


def detect_format(path: str, selected: str):
    return selected if selected != "auto" else "html" if Path(path).suffix.lower() in (".html", ".htm") else "markdown"


def emit_error(error: DocDeltaError):
    result = {"schemaVersion": 1, "status": "error", "error": {"code": error.code}}
    if error.input_name:
        result["error"]["input"] = error.input_name
    print(json.dumps(result, separators=(",", ":")))
    return 2


class JSONArgumentParser(argparse.ArgumentParser):
    def error(self, message):
        raise DocDeltaError("INVALID_ARGUMENTS")


def main(argv=None):
    ap = JSONArgumentParser(description=__doc__)
    ap.add_argument("before")
    ap.add_argument("after")
    for name in ("before", "after"):
        ap.add_argument("--" + name + "-format", choices=("auto", "markdown", "html"), default="auto")
    ap.add_argument("--max-changes", type=bounded_integer(1, 256), default=MAX_CHANGES)
    ap.add_argument("--max-output-bytes", type=bounded_integer(256, MAX_OUTPUT_LIMIT), default=OUTPUT_LIMIT)
    try:
        args = ap.parse_args(argv)
    except DocDeltaError as error:
        return emit_error(error)
    previous = None
    if hasattr(signal, "SIGALRM"):
        previous = signal.getsignal(signal.SIGALRM)
        def timeout(*_):
            raise DocDeltaError("PROCESSING_LIMIT")
        signal.signal(signal.SIGALRM, timeout)
        signal.alarm(5)
    try:
        old = read_snapshot(args.before, "before")
        new = read_snapshot(args.after, "after")
        result = compare(old, new, detect_format(args.before, args.before_format),
                         detect_format(args.after, args.after_format), args.max_changes)
        encoded = (json.dumps(result, indent=2, ensure_ascii=False) + "\n").encode("utf-8")
        if len(encoded) > args.max_output_bytes:
            raise DocDeltaError("OUTPUT_LIMIT")
        if previous is not None:
            signal.alarm(0)  # Do not interrupt a bounded stdout write with a second JSON object.
        sys.stdout.buffer.write(encoded)
        return 0
    except DocDeltaError as error:
        return emit_error(error)
    finally:
        if previous is not None:
            signal.alarm(0)
            signal.signal(signal.SIGALRM, previous)


if __name__ == "__main__":
    raise SystemExit(main())