"""Batched telemetry for the AI Segmentation plugin.

Design principles (do not deviate):
- Global opt-out: the shared TerraLab/telemetry_enabled QSettings key. When
  disabled, nothing is even queued. Fail-closed on read errors.
- Events batch in memory and flush once per generation cycle (or immediately
  for FLUSH_NOW milestones/failures). Batching collapses the power-user
  "hundreds of clicks per session" volume into one POST per cycle.
- Lifecycle events (NO_CONSENT_EVENTS) ship as soon as the plugin is activated;
  everything else additionally requires the user to have accepted the ToS.
  Pre-auth lifecycle events park in _pending_pre_auth until the first
  authenticated flush drains them.
- Relay model: plugin -> POST {base}/api/plugin/track -> our analytics relay. No analytics
  key in the plugin. Body shape {"events": [...]} matches the shared route.
- The typed per-event wrappers do NOT live here. They are in
  telemetry_session_events.py (lifecycle, install, pairing, account, manual,
  monetization, library), telemetry_run_events.py (everything an Automatic run
  and its review report) and telemetry_errors.py (plugin_error plus the slot
  and worker boundaries that catch an exception). This module is the transport:
  consent, batching, scrubbing, and the one POST.
- MAIN THREAD ONLY: flush() ends in QgsApplication.taskManager().addTask(),
  which is main-thread-only. Worker threads must only track(); the next
  main-thread flush ships the batch. A repeating main-thread timer flushes on
  its own, so a run whose events all come from worker threads does not have to
  wait for the dock to close. The batch is capped either way.
- Errors in telemetry never affect plugin functionality (fail silently).
- Payloads carry no paths, coordinates, layer names, urls or emails: every
  string leaving the machine goes through scrub_payload_value first.
- They are NOT anonymous, and nothing here should say they are. Each event
  carries a device_hash that is stable for the life of the install, and the
  batch is posted under the account's bearer key, so the relay can and does
  attach the account to it. track_auto_prompt_committed also sends what the
  user typed, verbatim. The Privacy card in the account dialog has to describe
  this honestly, and the public FAQ has to match it.

HTTP stack: QgsBlockingNetworkRequest (inside the flush QgsTask), so the
relay POST inherits QGIS proxy/TLS settings. No raw requests/urllib.
"""

from __future__ import annotations

import json
import platform
import sys
import threading
import time
import uuid

from qgis.core import (
    Qgis,
    QgsApplication,
    QgsBlockingNetworkRequest,
    QgsFeedback,
    QgsTask,
)
from qgis.PyQt.QtCore import QByteArray, QSettings, QThread, QTimer, QUrl
from qgis.PyQt.QtNetwork import QNetworkRequest

from . import transport_dials as _td
from .qt_compat import HttpStatusCodeAttribute, silent_task_flags
from .telemetry_events import FLUSH_NOW, NO_CONSENT_EVENTS, REGISTRY_VERSION

_TIMEOUT_MS = 5_000
_BATCH_MAX = 10
# The registered product id every event files under. The paid plan is a
# DIFFERENT id on the billing routes, and no event may arrive under that one:
# it is not a registered product on the analytics side, so anything filed there
# lands in no report at all.
TELEMETRY_PRODUCT_ID = "ai-segmentation"
_PENDING_PRE_AUTH_MAX = 50
_TELEMETRY_ENABLED_KEY = "TerraLab/telemetry_enabled"

# Ceiling on the events held in memory. The flush threshold above is a
# preference and flush() no-ops off the main thread, so this is the real bound.
_BATCH_HARD_MAX = 200
# Ceiling on one POST body. An error event carries a log tail, so a full batch
# of them would otherwise make a body of hundreds of KB.
_POST_MAX_BYTES = 128 * 1024
# How often the main thread ships what the worker threads queued.
_FLUSH_INTERVAL_S = 60
# Wait between the one POST and its single retry, spent in cancellable slices.
_RETRY_BACKOFF_S = 2.0

# How much telemetry travels is a server dial: batch size, transport timeout,
# and a per-event sample rate for throttling one noisy event during an
# incident. All three default to exactly what ships.
#
# WHAT the plugin is allowed to send is NOT a dial and must never become one.
# The opt-out default, NO_CONSENT_EVENTS and the scrubbing patterns are
# privacy commitments published in the FAQ; a server that could widen them
# could collect, from an installed plugin, something the user never agreed to.
# Volume can be tuned remotely. Scope cannot.


def _batch_max() -> int:
    """Events queued before a flush goes out."""
    try:
        from .server_dials import dial_in_range

        return int(dial_in_range("telemetry.batch_max", _BATCH_MAX, 1, 200))
    except Exception:  # noqa: BLE001 -- telemetry must never break a caller
        return _BATCH_MAX


def _timeout_ms() -> int:
    """Transport timeout on one relay call.

    The ceiling is deliberately low. This timeout is what a flush at unload
    waits out before it gives up, so a high one served by mistake would hold
    the teardown for that long, and no batch of usage events is worth that.
    """
    try:
        from .server_dials import dial_in_range

        return int(dial_in_range("telemetry.timeout_ms", _TIMEOUT_MS, 500, 10_000))
    except Exception:  # noqa: BLE001 -- telemetry must never break a caller
        return _TIMEOUT_MS


def _event_sampled_in(event: str, urgent: bool = False) -> bool:
    """Whether this occurrence of ``event`` is kept.

    ``telemetry.sample_rates`` is a ``{event_name: rate}`` map, every event
    defaulting to 1.0 (keep everything, today's behaviour). A rate below one
    drops that share of occurrences at random, which is how one noisy event
    gets throttled during an incident without shipping a release. Only a
    number in [0, 1] counts; anything else keeps the event.

    An event that ships immediately reads its own map,
    ``telemetry.urgent_sample_rates``. An incident is exactly what makes a
    failure or a milestone noisy, so those need a brake too, and a separate map
    means reaching for it can never thin the batched funnel by accident.

    A dropped event is never queued, so this cannot be used to collect more.
    """
    try:
        from .server_dials import read_value

        key = "telemetry.urgent_sample_rates" if urgent else "telemetry.sample_rates"
        rates = read_value(key)
        if not isinstance(rates, dict):
            return True
        rate = rates.get(event)
        if not isinstance(rate, (int, float)) or isinstance(rate, bool):
            return True
        if not 0.0 <= rate < 1.0:
            return True
        import random  # noqa: PLC0415 -- only needed when a rate is served

        return random.random() < rate  # nosec B311 -- sampling, not security
    except Exception:  # noqa: BLE001 -- telemetry must never break a caller
        return True


# Guards _batch / _pending_pre_auth / _inflight: track() can run on a worker
# thread while the main thread flushes, so the list mutations must not race.
_lock = threading.Lock()
_batch: list[dict] = []
_pending_pre_auth: list[dict] = []
_inflight: list = []
_session_id = uuid.uuid4().hex
# The periodic main-thread flush, created once on the first main-thread track().
_flush_timer = None

# Most-recent Automatic run correlation id (the id the server archives each
# billed run under). Kept so the error report can quote it for support; None
# until the first run this process. Not telemetry state, just a breadcrumb.
_last_run_id: str | None = None

# The plugin version and relay base URL come from static bundled files that
# never change during a session; memoize them so track()/flush() do not re-read
# metadata.txt (and .env.local) from disk on every event.
_plugin_version_cache: str | None = None

# The opt-out, held for a moment. See is_telemetry_enabled.
_enabled_cache: bool | None = None
_enabled_cache_at: float = 0.0
_ENABLED_CACHE_S = 5.0


# --- Opt-out --------------------------------------------------------------


def is_telemetry_enabled() -> bool:
    """Whether usage telemetry is enabled. Opt-out: defaults to True.

    Reads the shared TerraLab/telemetry_enabled QSettings key (shared with
    AI Edit so the user opts out once). Fail-closed: if the preference cannot be
    read, do NOT send (privacy over a data point).

    Read at most once every few seconds and held in between, because every
    event asks. Short rather than for the session: the key is shared, so the
    sibling plugin can turn it off while this one is loaded, and that answer
    has to reach us without a restart."""
    global _enabled_cache, _enabled_cache_at
    now = time.monotonic()
    if _enabled_cache is not None and (now - _enabled_cache_at) < _ENABLED_CACHE_S:
        return _enabled_cache
    try:
        value = bool(QSettings().value(_TELEMETRY_ENABLED_KEY, True, type=bool))
    except Exception:  # nosec B110
        value = False
    _enabled_cache, _enabled_cache_at = value, now
    return value


def _forget_enabled_cache() -> None:
    """Drop the held preference so the next read goes to QSettings."""
    global _enabled_cache
    _enabled_cache = None


def set_telemetry_enabled(enabled: bool) -> None:
    """Persist the global telemetry opt-out flag (shared across TerraLab plugins).

    Turning it OFF also drops what is still queued. The queues are already
    empty on the normal path, since the opt-out wrapper drops them before it
    records the choice; this is the backstop for any other caller. The drop
    runs even when the write failed, because an unwritten preference is the
    fail-closed case."""
    try:
        QSettings().setValue(_TELEMETRY_ENABLED_KEY, bool(enabled))
    except Exception:  # nosec B110
        pass
    _forget_enabled_cache()
    if not enabled:
        drop_queued_events()


def drop_queued_events() -> None:
    """Throw away everything still queued, sent nowhere.

    The opt-out path calls this BEFORE it records the choice, so the events
    behind the record cannot ride out with it: they were queued under a
    consent the user is in the middle of withdrawing.

    A batch already handed to the task manager is queued too, just one step
    further along, so cancel those as well: the cancel reaches the blocking
    POST through the task's own feedback.
    """
    with _lock:
        _batch.clear()
        _pending_pre_auth.clear()
        pending_tasks = list(_inflight)
        _inflight.clear()
    for task in pending_tasks:
        try:
            task.cancel()
        except Exception:  # nosec B110 - telemetry must never break the plugin
            pass


def new_session() -> None:
    """Rotate the per-session id. Call on dock open so events group by session."""
    global _session_id
    _session_id = uuid.uuid4().hex


def current_session_id() -> str:
    """The id every event carries until new_session() rotates it. Read by the
    error path, which reports one occurrence of a crash per session."""
    return _session_id


def set_last_run_id(run_id: str | None) -> None:
    """Remember the most recent Automatic run's correlation id so a later error
    report can quote it. Best-effort breadcrumb, never raises."""
    global _last_run_id
    _last_run_id = run_id or None


def get_last_run_id() -> str | None:
    """The most recent Automatic run id this process, or None if no run yet."""
    return _last_run_id


# --- Payload helpers ------------------------------------------------------


def _base_properties() -> dict:
    """Properties shared by every event (computed once per call, cheap)."""
    try:
        qgis_version = Qgis.QGIS_VERSION
    except Exception:
        qgis_version = "unknown"
    props = {
        "product_id": TELEMETRY_PRODUCT_ID,
        "plugin_version": _read_plugin_version(),
        "os": platform.system(),
        "os_version": platform.release(),
        "arch": platform.machine(),
        "python_version": f"{sys.version_info.major}.{sys.version_info.minor}",
        "qgis_version": qgis_version,
        "session_id": _session_id,
        "registry_version": REGISTRY_VERSION,
    }
    try:
        from .device_id import get_device_hash
        props["device_hash"] = get_device_hash()
    except Exception:  # nosec B110
        pass
    return props


def _read_plugin_version() -> str:
    global _plugin_version_cache
    if _plugin_version_cache is not None:
        return _plugin_version_cache
    import os
    version = "unknown"
    try:
        plugin_dir = os.path.dirname(
            os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        )
        metadata_path = os.path.join(plugin_dir, "metadata.txt")
        with open(metadata_path, encoding="utf-8") as f:
            for line in f:
                if line.startswith("version="):
                    version = line.strip().split("=", 1)[1]
                    break
    except Exception:
        pass  # nosec B110
    # Cache only a real read: a transient failure ("unknown") is retried next
    # call rather than pinned for the whole session.
    if version != "unknown":
        _plugin_version_cache = version
    return version


def _build_base_url() -> str:
    """The address the relay POST goes to. The same reader the client uses, so
    the two can never end up on different backends."""
    from .env_local import terralab_base_url

    return terralab_base_url()


def _get_auth_header() -> dict | None:
    """Return {'Authorization': 'Bearer <key>'} if the plugin is activated."""
    try:
        from .activation_manager import get_auth_header, is_plugin_activated
        if not is_plugin_activated():
            return None
        hdr = get_auth_header()
        if hdr and hdr.get("Authorization"):
            return hdr
    except Exception:
        pass  # nosec B110
    return None


def _has_consent() -> bool:
    """Non-lifecycle events additionally require ToS acceptance (raw message
    fields can carry path fragments; the ToS is the user's data gate).

    Precedence, highest first: the global telemetry opt-out refuses everything
    and is read on its own before this; then the ToS lock, which is permanent
    once the user has run with consent; then the tick box, which defaults to
    on. So a locked ToS answers yes whatever the box says, and the opt-out
    answers no whatever either says.
    """
    try:
        from .activation_manager import has_tos_accepted, has_tos_locked
        return bool(has_tos_accepted() or has_tos_locked())
    except Exception:
        return False


def on_main_thread() -> bool:
    try:
        app = QgsApplication.instance()
        return app is not None and QThread.currentThread() == app.thread()
    except Exception:
        return False


# --- Background flush task ------------------------------------------------


class _TelemetryFlushTask(QgsTask):
    """Sends one batch. Failures swallowed: telemetry must never break the plugin."""

    def __init__(self, events: list, auth: dict):
        super().__init__("AI Segmentation telemetry flush", silent_task_flags())
        self._events = events
        self._auth = auth
        # Carries the cancel into the blocking POST. Without it the transport
        # runs to its own timeout whatever the task manager asks, and the last
        # flush of a session is exactly the one a user is waiting on.
        self._feedback = QgsFeedback()

    def cancel(self) -> None:
        try:
            self._feedback.cancel()
        except Exception:  # nosec B110 - telemetry must never break the plugin
            pass
        super().cancel()

    def run(self) -> bool:
        if self.isCanceled():
            return False
        # One retry with a short backoff covers a transient network blip without
        # a disk queue; a hard-offline session still loses the batch (accepted).
        if not self._post() and not self.isCanceled():
            self._wait_before_retry(_td.telemetry_retry_backoff_s(_RETRY_BACKOFF_S))
            if self.isCanceled():
                return False
            self._post()
        return True

    def _wait_before_retry(self, seconds: float) -> None:
        """Back off in slices, so a cancel is honoured while it waits."""
        end = time.monotonic() + seconds
        while time.monotonic() < end:
            if self.isCanceled():
                return
            time.sleep(0.1)

    def _post(self) -> bool:
        try:
            payload = json.dumps({"events": self._events}).encode("utf-8")
            url = f"{_build_base_url().rstrip('/')}/api/plugin/track"
            from .server_dials import cleartext_remote_url

            if cleartext_remote_url(url):
                # The batch travels under the account's bearer key. Reported as
                # done, not as a failure: a retry would send it the same way.
                return True
            req = QNetworkRequest(QUrl(url))
            req.setRawHeader(b"Content-Type", b"application/json")
            if hasattr(req, "setTransferTimeout"):
                req.setTransferTimeout(_timeout_ms())
            for k, v in self._auth.items():
                req.setRawHeader(k.encode("utf-8"), v.encode("utf-8"))
            blocker = QgsBlockingNetworkRequest()
            try:
                err = blocker.post(req, QByteArray(payload), False, self._feedback)
            except TypeError:
                # Bindings without the feedback argument: the POST then runs to
                # the transport timeout, which is why that dial stays short.
                err = blocker.post(req, QByteArray(payload))
            # ErrorCode 0 = NoError. Ignoring the result made run()'s single
            # retry dead code: a failed batch returned True and was dropped.
            if int(err) != 0:
                return False
            # A transport-level NoError still covers a 4xx/5xx response: the
            # POST reached the relay but it rejected the batch. A 5xx, and a
            # rate limit, are worth run()'s single retry. Any other 4xx is the
            # relay refusing this body, and it would refuse the same body a
            # second time, so that one counts as done and the batch is dropped.
            # An unreadable status (None) stays a success, since the transport
            # itself succeeded.
            status = self._http_status(blocker)
            if status is None or status < 400:
                return True
            return not (status >= 500 or status == 429)
        except Exception:
            return False  # nosec B110 - telemetry must never break the plugin

    @staticmethod
    def _http_status(blocker) -> int | None:
        """HTTP status of the reply, or None. Never raises (Qt can hand back a
        non-numeric attribute that int() would choke on)."""
        if HttpStatusCodeAttribute is None:
            return None
        try:
            reply = blocker.reply()
            if reply is None:
                return None
            attr = reply.attribute(HttpStatusCodeAttribute)
        except (RuntimeError, AttributeError):
            return None
        if attr is None:
            return None
        try:
            return int(attr)
        except (TypeError, ValueError):
            return None

    def finished(self, result: bool) -> None:
        return


def _drop_inflight(task: _TelemetryFlushTask) -> None:
    with _lock:
        try:
            _inflight.remove(task)
        except ValueError:
            pass


# --- Core API -------------------------------------------------------------


def track(event: str, properties: dict | None = None, flush_now: bool = False) -> None:
    """Queue an event. Global opt-out short-circuits before anything is queued.

    Ships immediately when flush_now is True, the event is a FLUSH_NOW milestone,
    or the batch is full; otherwise it waits for the next flush()."""
    if not is_telemetry_enabled():
        return
    urgent = flush_now or event in FLUSH_NOW
    if not _event_sampled_in(event, urgent):
        return
    try:
        evt = {
            "event": event,
            "properties": {**_base_properties(), **(properties or {})},
        }
        # Pinned AFTER the merge. A caller that passes a subscription row
        # through (the paid plan carries its own product id on the billing
        # routes) would otherwise file its event under an id the analytics
        # side does not know.
        evt["properties"]["product_id"] = TELEMETRY_PRODUCT_ID
    except Exception:  # nosec B110
        return
    with _lock:
        _batch.append(evt)
        _trim_batch_locked()
        should_flush = urgent or len(_batch) >= _batch_max()
    _arm_flush_timer()
    if should_flush:
        flush()


def _trim_batch_locked() -> None:
    """Hold the batch under its ceiling. Call with _lock held.

    A worker thread can track for the length of a whole run while flush()
    no-ops off the main thread, so the list has to have an end. The oldest
    ordinary events go first and the milestones stay, since those are what the
    funnel is measured on; a flood of milestones alone still gets cut from the
    oldest, or the bound would not be one."""
    drop = len(_batch) - _td.telemetry_batch_hard_max(_BATCH_HARD_MAX)
    if drop <= 0:
        return
    kept: list[dict] = []
    for evt in _batch:
        if drop > 0 and evt.get("event") not in FLUSH_NOW:
            drop -= 1
            continue
        kept.append(evt)
    if drop > 0:
        kept = kept[drop:]
    _batch[:] = kept


def _on_flush_timer() -> None:
    """Timer slot. Swallows everything: a telemetry failure must never reach
    the event loop."""
    try:
        flush()
    except Exception:  # nosec B110
        pass


def _arm_flush_timer() -> None:
    """Start the periodic main-thread flush, once per process.

    flush() no-ops off the main thread, and the plugin calls it at unload and
    at dock close only. Without this, events a worker queued during a run wait
    for one of those two moments and die with the session if it ends first.
    Armed from track() because that is the first main-thread call telemetry
    makes: Qt refuses to start a timer from another thread."""
    global _flush_timer
    if _flush_timer is not None or not on_main_thread():
        return
    try:
        timer = QTimer()
        timer.setInterval(_td.telemetry_flush_interval_s(_FLUSH_INTERVAL_S) * 1000)
        timer.timeout.connect(_on_flush_timer)
        timer.start()
        _flush_timer = timer
    except Exception:  # nosec B110
        _flush_timer = None


def stop_flush_timer() -> None:
    """Stop the periodic flush. Called from the plugin unload.

    A QTimer with no parent outlives the plugin that armed it, so without this
    it keeps firing into a module the user has just disabled. The next
    main-thread track() arms a fresh one, so a re-enable is unaffected.
    """
    global _flush_timer
    timer = _flush_timer
    _flush_timer = None
    if timer is None:
        return
    try:
        timer.stop()
        timer.timeout.disconnect(_on_flush_timer)
    except Exception:  # nosec B110
        pass
    try:
        timer.deleteLater()
    except Exception:  # nosec B110
        pass


def _split_for_post(events: list[dict]) -> list[list[dict]]:
    """Cut one flush into bodies the relay will accept.

    An error event carries a log tail, so a batch of them is worth hundreds of
    KB and one POST would carry the lot. Each chunk stays under the byte cap;
    an event that busts the cap on its own still travels, alone."""
    chunks: list[list[dict]] = []
    current: list[dict] = []
    size = 0
    post_max = _td.telemetry_post_max_bytes(_POST_MAX_BYTES)
    for evt in events:
        try:
            evt_bytes = len(json.dumps(evt).encode("utf-8"))
        except Exception:  # noqa: BLE001 -- unmeasurable: give it its own POST
            evt_bytes = post_max
        if current and size + evt_bytes > post_max:
            chunks.append(current)
            current = []
            size = 0
        current.append(evt)
        size += evt_bytes
    if current:
        chunks.append(current)
    return chunks


def flush() -> None:
    """Ship the queued batch. MAIN THREAD ONLY (no-ops off it). Lifecycle events
    ship pre-consent; everything else requires consent. Pre-auth lifecycle events
    park in _pending_pre_auth until the first authenticated flush."""
    if not on_main_thread():
        return
    # The opt-out is re-read here, not only in track(). The pre-auth queue can
    # hold events from before a sign-in that happened after the user turned
    # telemetry off, and nothing queued under an earlier consent may leave once
    # the setting says no.
    if not is_telemetry_enabled():
        with _lock:
            _batch.clear()
            _pending_pre_auth.clear()
        return
    with _lock:
        if not _batch and not _pending_pre_auth:
            return
    # Resolved BEFORE the lock is taken again. Both reach into the auth manager
    # and QgsSettings, and a worker thread holds this same lock in track(), so
    # doing them under it puts a GUI-thread wait behind a worker every minute.
    # The lock then only drains the two lists. The check above is repeated
    # below because a worker can only add to the batch in between, never empty
    # it.
    auth = _get_auth_header()
    consented = _has_consent()
    with _lock:
        if not _batch and not _pending_pre_auth:
            return
        if not auth:
            pre_auth_max = _td.telemetry_pending_pre_auth_max(_PENDING_PRE_AUTH_MAX)
            for evt in _batch:
                if evt["event"] in NO_CONSENT_EVENTS and len(_pending_pre_auth) < pre_auth_max:
                    _pending_pre_auth.append(evt)
            _batch.clear()
            return
        events_to_send = list(_pending_pre_auth) + [
            e for e in _batch
            if consented or e["event"] in NO_CONSENT_EVENTS
        ]
        _batch.clear()
        _pending_pre_auth.clear()
    if not events_to_send:
        return
    # Sizing the chunks means serializing every event, so it happens off the
    # lock: a worker thread must not wait on it to queue one more.
    tasks = [_TelemetryFlushTask(chunk, auth) for chunk in _split_for_post(events_to_send)]
    with _lock:
        _inflight.extend(tasks)
    for task in tasks:
        try:
            task.taskCompleted.connect(lambda t=task: _drop_inflight(t))
            task.taskTerminated.connect(lambda t=task: _drop_inflight(t))
        except Exception:  # nosec B110
            pass
        QgsApplication.taskManager().addTask(task)


# --- Payload scrubbing (kept as-is) ---------------------------------------


_COORD_PATTERN = None
_URL_PATTERN = None
_EMAIL_PATTERN = None
_PATH_PATTERN = None

# One path segment: no whitespace, no separator, no quote and no bracket. The
# colon is out as well, so a trailing "file.py: cannot open" keeps the colon
# that separates the name from the sentence after it.
_PATH_SEGMENT = r"[^\s\\/'\"<>,;()=:]"


def scrub_payload_value(value: str) -> str:
    """Strip file paths, coordinate tuples, URLs and email addresses from
    telemetry strings.

    Applied defensively to any string leaving the machine. We already call
    log_scrub.anonymize_paths for filesystem paths, but this pass also catches
    leftover coordinate-like artefacts (crop bounds, click tuples, bbox
    extents) plus URLs/emails: the unhandled-error catch-all forwards raw
    third-party exception text, which can embed a host or an address, and the
    telemetry contract is no URLs and no emails, so redact rather than trust
    the source.

    Paths travel as "<path>", whole. anonymize_paths only replaces the home
    prefix, so a system path, a Windows install path and every folder the user
    named under their home came through intact. A folder tree is named by the
    user and often after their client, and the contract is no file paths at
    all. The error class words are what make an event useful, and those stay:
    the run ends at the first segment that carries a space, which is where a
    path stops and a sentence starts.
    """
    import re as _re
    global _COORD_PATTERN, _URL_PATTERN, _EMAIL_PATTERN, _PATH_PATTERN
    if _COORD_PATTERN is None:
        _COORD_PATTERN = _re.compile(
            r"(?:[-+]?\d+(?:\.\d+)?)(?:\s*,\s*[-+]?\d+(?:\.\d+)?){1,}"
        )
        _URL_PATTERN = _re.compile(r"[a-zA-Z][a-zA-Z0-9+.-]*://[^\s'\"]+")
        _EMAIL_PATTERN = _re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b")
        _PATH_PATTERN = _re.compile(
            # A root: the marker anonymize_paths writes over the home
            # directory, a drive letter, a "~", a UNC prefix, or a separator
            # that starts a token. The lookbehinds keep it off ordinary prose,
            # where a slash sits between two word characters (3/4, km/h).
            r"(?:<USER>"
            r"|(?<![\w.])[A-Za-z]:(?=[\\/])"
            r"|~(?=[\\/])"
            r"|\\\\(?=" + _PATH_SEGMENT + r")"
            r"|(?<![\w.:)\]])/(?=" + _PATH_SEGMENT + r"))"
            # Everything up to the last separator (folder names hold spaces,
            # so the run crosses them), then a final segment that holds none.
            r"(?:[^\r\n'\"<>,;()=]*[\\/])?" + _PATH_SEGMENT + r"*"
        )
    # core-to-core import, deliberately NOT wrapped in try/except: a broken
    # scrubber must fail loudly, never silently ship unscrubbed paths.
    from .log_scrub import anonymize_paths
    value = anonymize_paths(value)
    value = _URL_PATTERN.sub("<URL>", value or "")
    value = _EMAIL_PATTERN.sub("<EMAIL>", value)
    # After the URL pass, so a scheme-bearing address is already gone and its
    # path half cannot be read as a local one.
    value = _PATH_PATTERN.sub("<path>", value)
    return _COORD_PATTERN.sub("<COORDS>", value)
