Compose: spelling and grammar check with LanguageTool
A self-hosted LanguageTool (dev: docker/docker-compose.yml, started by `make up`) checks what the user writes in the compose window, through POST /api/spellcheck — the browser never talks to the engine directly. - Marks are CSS highlights over the text, never written into the message: red wavy for spelling, blue for grammar/style. Clicking one shows the message, up to five suggestions (applied as typed, so Ctrl+Z undoes them) and Ignore. The popup, like the link popup, opens above its anchor when there is no room below. - Checked after a pause in typing; the signature and quoted original are skipped; marks on words just edited drop at once; a check overtaken by a newer one, by turning checking off, or by closing the window is discarded. - With "auto", text is checked in chunks of lines, each in its own detected language — LanguageTool settles on one language per request, so a mixed Polish/English message had every English word flagged as a Polish typo. - Settings > Appearance: on/off and language (Auto-detect or any of the engine's languages); the compose window's More menu overrides it for one message. The browser's own spellcheck steps aside while ours runs, and comes back when it is off or the server has no engine (503 from the endpoint). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -0,0 +1,187 @@
|
||||
"""Spelling and grammar check for the compose window, backed by LanguageTool.
|
||||
|
||||
The browser sends the text it wants checked here, never to LanguageTool
|
||||
itself: the engine stays on the internal network, only signed-in users reach
|
||||
it, and nothing leaves the server. When no engine is configured (or it is
|
||||
down), the check answers 503 and the compose window simply goes without.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
from collections import Counter
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
from fastapi import APIRouter, Depends
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from app.api.errors import raise_api_error
|
||||
from app.config import get_settings
|
||||
from app.identity.deps import CurrentUser, require_csrf
|
||||
|
||||
router = APIRouter(prefix="/spellcheck", tags=["compose"])
|
||||
|
||||
_TIMEOUT = 10.0
|
||||
# A message body is checked as a whole; past this it is a document, not mail.
|
||||
MAX_TEXT = 20_000
|
||||
_MAX_SUGGESTIONS = 6
|
||||
# With "auto", LanguageTool settles on ONE language per request, so a message
|
||||
# mixing Polish and English paragraphs would have every English word flagged as
|
||||
# a Polish typo. Auto-detected text is therefore checked in chunks, each in its
|
||||
# own language. A chunk is at least this many characters of consecutive lines:
|
||||
# a greeting alone ("Witam,") is too short to detect reliably, so it rides with
|
||||
# the lines after it.
|
||||
_MIN_CHUNK = 60
|
||||
_PARALLEL = 4
|
||||
|
||||
|
||||
class CheckIn(BaseModel):
|
||||
text: str = Field(max_length=MAX_TEXT)
|
||||
# "auto" or a LanguageTool code ("pl-PL", "en-GB", "de-DE" ...).
|
||||
language: str = Field(default="auto", max_length=16)
|
||||
|
||||
|
||||
class Issue(BaseModel):
|
||||
offset: int
|
||||
length: int
|
||||
message: str
|
||||
short_message: str
|
||||
replacements: list[str]
|
||||
# spelling | grammar | style: what the underline colour follows.
|
||||
kind: str
|
||||
rule_id: str
|
||||
|
||||
|
||||
class CheckOut(BaseModel):
|
||||
language: str # the language checked (detected, with "auto")
|
||||
language_name: str
|
||||
issues: list[Issue]
|
||||
|
||||
|
||||
class LanguageOut(BaseModel):
|
||||
code: str
|
||||
name: str
|
||||
|
||||
|
||||
def _base() -> str:
|
||||
url = get_settings().languagetool_url.rstrip("/")
|
||||
if not url:
|
||||
raise_api_error(503, "spellcheck_unavailable", "spell checking is not set up")
|
||||
return url
|
||||
|
||||
|
||||
def _kind(issue_type: str) -> str:
|
||||
if issue_type == "misspelling":
|
||||
return "spelling"
|
||||
if issue_type in ("style", "typographical", "whitespace", "register", "locale-violation"):
|
||||
return "style"
|
||||
return "grammar"
|
||||
|
||||
|
||||
def chunks(text: str) -> list[tuple[int, str]]:
|
||||
"""Split ``text`` into (offset, chunk) runs of whole lines, each at least
|
||||
_MIN_CHUNK characters where the text allows; a blank line always ends one."""
|
||||
out: list[tuple[int, str]] = []
|
||||
start = pos = 0
|
||||
for line in text.split("\n"):
|
||||
end = pos + len(line)
|
||||
if not line.strip():
|
||||
if text[start:pos].strip():
|
||||
# Up to the line break before the blank line, not including it.
|
||||
out.append((start, text[start:pos].rstrip("\n")))
|
||||
start = end + 1
|
||||
elif end - start >= _MIN_CHUNK:
|
||||
out.append((start, text[start:end]))
|
||||
start = end + 1
|
||||
pos = end + 1
|
||||
if text[start:].strip():
|
||||
out.append((start, text[start:]))
|
||||
return out
|
||||
|
||||
|
||||
async def _check_one(
|
||||
client: httpx.AsyncClient, base: str, text: str, language: str
|
||||
) -> dict[str, Any]:
|
||||
resp = await client.post(f"{base}/v2/check", data={"text": text, "language": language})
|
||||
resp.raise_for_status()
|
||||
data: dict[str, Any] = resp.json()
|
||||
return data
|
||||
|
||||
|
||||
def _issues(data: dict[str, Any], shift: int) -> list[Issue]:
|
||||
return [
|
||||
Issue(
|
||||
offset=int(m["offset"]) + shift,
|
||||
length=int(m["length"]),
|
||||
message=str(m.get("message", "")),
|
||||
short_message=str(m.get("shortMessage", "")),
|
||||
replacements=[str(r["value"]) for r in (m.get("replacements") or [])][
|
||||
:_MAX_SUGGESTIONS
|
||||
],
|
||||
kind=_kind(str((m.get("rule") or {}).get("issueType", ""))),
|
||||
rule_id=str((m.get("rule") or {}).get("id", "")),
|
||||
)
|
||||
for m in data.get("matches") or []
|
||||
]
|
||||
|
||||
|
||||
def _detected(data: dict[str, Any]) -> tuple[str, str]:
|
||||
lang = data.get("language") or {}
|
||||
detected = lang.get("detectedLanguage") or lang
|
||||
return str(detected.get("code", "")), str(detected.get("name", ""))
|
||||
|
||||
|
||||
@router.post("", response_model=CheckOut, dependencies=[Depends(require_csrf)])
|
||||
async def check(body: CheckIn, user: CurrentUser) -> CheckOut:
|
||||
base = _base()
|
||||
if not body.text.strip():
|
||||
return CheckOut(language=body.language, language_name="", issues=[])
|
||||
parts = chunks(body.text) if body.language == "auto" else [(0, body.text)]
|
||||
gate = asyncio.Semaphore(_PARALLEL)
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=_TIMEOUT) as client:
|
||||
|
||||
async def one(text: str) -> dict[str, Any]:
|
||||
async with gate:
|
||||
return await _check_one(client, base, text, body.language)
|
||||
|
||||
results = await asyncio.gather(*(one(text) for _, text in parts))
|
||||
except (httpx.HTTPError, ValueError):
|
||||
raise_api_error(503, "spellcheck_unavailable", "the spell checker is not answering")
|
||||
|
||||
issues = [
|
||||
i for (shift, _), data in zip(parts, results, strict=True) for i in _issues(data, shift)
|
||||
]
|
||||
# The language reported is the one most of the text was written in.
|
||||
weight: Counter[tuple[str, str]] = Counter()
|
||||
for (_, text), data in zip(parts, results, strict=True):
|
||||
weight[_detected(data)] += len(text)
|
||||
(code, name), _ = weight.most_common(1)[0]
|
||||
return CheckOut(language=code or body.language, language_name=name, issues=issues)
|
||||
|
||||
|
||||
_languages_cache: tuple[float, list[LanguageOut]] | None = None
|
||||
|
||||
|
||||
@router.get("/languages", response_model=list[LanguageOut])
|
||||
async def languages(user: CurrentUser) -> list[LanguageOut]:
|
||||
"""What the engine can check, for the language picker in Settings."""
|
||||
global _languages_cache
|
||||
if _languages_cache and time.monotonic() - _languages_cache[0] < 3600:
|
||||
return _languages_cache[1]
|
||||
base = _base()
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=_TIMEOUT) as client:
|
||||
resp = await client.get(f"{base}/v2/languages")
|
||||
resp.raise_for_status()
|
||||
rows: list[dict[str, Any]] = resp.json()
|
||||
except (httpx.HTTPError, ValueError):
|
||||
raise_api_error(503, "spellcheck_unavailable", "the spell checker is not answering")
|
||||
seen: dict[str, str] = {}
|
||||
for row in rows:
|
||||
seen.setdefault(str(row["longCode"]), str(row["name"]))
|
||||
out = sorted((LanguageOut(code=c, name=n) for c, n in seen.items()), key=lambda x: x.name)
|
||||
_languages_cache = (time.monotonic(), out)
|
||||
return out
|
||||
Reference in New Issue
Block a user