1 files changed, 186 insertions, 0 deletions
diff --git a/src/debputy/lsp/lsp_debian_changelog.py b/src/debputy/lsp/lsp_debian_changelog.py
new file mode 100644
index 0000000..3ec0b4d
--- /dev/null
+++ b/src/debputy/lsp/lsp_debian_changelog.py
@@ -0,0 +1,186 @@
+import sys
+from typing import (
+    Union,
+    List,
+    Dict,
+    Iterator,
+    Optional,
+    Iterable,
+)
+
+from lsprotocol.types import (
+    Diagnostic,
+    DidOpenTextDocumentParams,
+    DidChangeTextDocumentParams,
+    TEXT_DOCUMENT_DID_OPEN,
+    TEXT_DOCUMENT_DID_CHANGE,
+    TEXT_DOCUMENT_WILL_SAVE_WAIT_UNTIL,
+    TEXT_DOCUMENT_CODE_ACTION,
+    TEXT_DOCUMENT_DID_CLOSE,
+    DidCloseTextDocumentParams,
+    Range,
+    Position,
+    DiagnosticSeverity,
+)
+
+from debputy.lsp.lsp_features import lsp_diagnostics, lsp_standard_handler
+from debputy.lsp.quickfixes import (
+    provide_standard_quickfixes_from_diagnostics,
+)
+from debputy.lsp.spellchecking import spellcheck_line
+from debputy.lsp.text_util import (
+    on_save_trim_end_of_line_whitespace,
+    LintCapablePositionCodec,
+)
+
+try:
+    from debian._deb822_repro.locatable import Position as TEPosition, Ranage as TERange
+
+    from pygls.server import LanguageServer
+    from pygls.workspace import TextDocument
+except ImportError:
+    pass
+
+
+# Same as Lintian
+_MAXIMUM_WIDTH: int = 82
+_LANGUAGE_IDS = [
+    "debian/changelog",
+    # emacs's name
+    "debian-changelog",
+    # vim's name
+    "debchangelog",
+]
+
+DOCUMENT_VERSION_TABLE: Dict[str, int] = {}
+
+
+def register_dch_lsp(ls: "LanguageServer") -> None:
+    ls.feature(TEXT_DOCUMENT_DID_OPEN)(_diagnostics_debian_changelog)
+    ls.feature(TEXT_DOCUMENT_DID_CHANGE)(_diagnostics_debian_changelog)
+    ls.feature(TEXT_DOCUMENT_DID_CLOSE)(_handle_close)
+    ls.feature(TEXT_DOCUMENT_CODE_ACTION)(
+        ls.thread()(provide_standard_quickfixes_from_diagnostics)
+    )
+    ls.feature(TEXT_DOCUMENT_WILL_SAVE_WAIT_UNTIL)(on_save_trim_end_of_line_whitespace)
+
+
+def _handle_close(
+    ls: "LanguageServer",
+    params: DidCloseTextDocumentParams,
+) -> None:
+    try:
+        del DOCUMENT_VERSION_TABLE[params.text_document.uri]
+    except KeyError:
+        pass
+
+
+def is_doc_at_version(uri: str, version: int) -> bool:
+    dv = DOCUMENT_VERSION_TABLE.get(uri)
+    return dv == version
+
+
+lsp_standard_handler(_LANGUAGE_IDS, TEXT_DOCUMENT_CODE_ACTION)
+lsp_standard_handler(_LANGUAGE_IDS, TEXT_DOCUMENT_WILL_SAVE_WAIT_UNTIL)
+
+
+@lsp_diagnostics(_LANGUAGE_IDS)
+def _diagnostics_debian_changelog(
+    ls: "LanguageServer",
+    params: Union[DidOpenTextDocumentParams, DidChangeTextDocumentParams],
+) -> Iterable[List[Diagnostic]]:
+    doc_uri = params.text_document.uri
+    doc = ls.workspace.get_text_document(doc_uri)
+    lines = doc.lines
+    max_words = 1_000
+    delta_update_size = 10
+    max_lines_between_update = 10
+    scanner = _scan_debian_changelog_for_diagnostics(
+        lines,
+        doc.position_codec,
+        delta_update_size,
+        max_words,
+        max_lines_between_update,
+    )
+
+    yield from scanner
+
+
+def _scan_debian_changelog_for_diagnostics(
+    lines: List[str],
+    position_codec: LintCapablePositionCodec,
+    delta_update_size: int,
+    max_words: int,
+    max_lines_between_update: int,
+    *,
+    max_line_length: int = _MAXIMUM_WIDTH,
+) -> Iterator[List[Diagnostic]]:
+    diagnostics = []
+    diagnostics_at_last_update = 0
+    lines_since_last_update = 0
+    for line_no, line in enumerate(lines):
+        orig_line = line
+        line = line.rstrip()
+        if not line:
+            continue
+        if not line.startswith("  "):
+            continue
+        # minus 1 for newline
+        orig_line_len = len(orig_line) - 1
+        if orig_line_len > max_line_length:
+            range_server_units = Range(
+                Position(
+                    line_no,
+                    max_line_length,
+                ),
+                Position(
+                    line_no,
+                    orig_line_len,
+                ),
+            )
+            diagnostics.append(
+                Diagnostic(
+                    position_codec.range_to_client_units(lines, range_server_units),
+                    f"Line exceeds {max_line_length} characters",
+                    severity=DiagnosticSeverity.Hint,
+                    source="debputy",
+                )
+            )
+        if len(line) > 3 and line[2] == "[" and line[-1] == "]":
+            # Do not spell check [ X ] as X is usually a name
+            continue
+        lines_since_last_update += 1
+        if max_words > 0:
+            typos = list(spellcheck_line(lines, position_codec, line_no, line))
+            new_diagnostics = len(typos)
+            max_words -= new_diagnostics
+            diagnostics.extend(typos)
+
+        current_diagnostics_len = len(diagnostics)
+        if (
+            lines_since_last_update >= max_lines_between_update
+            or current_diagnostics_len - diagnostics_at_last_update > delta_update_size
+        ):
+            diagnostics_at_last_update = current_diagnostics_len
+            lines_since_last_update = 0
+
+            yield diagnostics
+    if not diagnostics or diagnostics_at_last_update != len(diagnostics):
+        yield diagnostics
+
+
+def _lint_debian_changelog(
+    _doc_reference: str,
+    _path: str,
+    lines: List[str],
+    position_codec: LintCapablePositionCodec,
+) -> Optional[List[Diagnostic]]:
+    limits = sys.maxsize
+    scanner = _scan_debian_changelog_for_diagnostics(
+        lines,
+        position_codec,
+        limits,
+        limits,
+        limits,
+    )
+    return next(iter(scanner), None)