data_studio/utils.py

from __future__ import annotations

import hashlib
import re
from datetime import datetime, timezone
from typing import Iterable


SPACE_RE = re.compile(r"\s+")
SENTENCE_SPLIT_RE = re.compile(r"(?<=[.!?।])\s+")


def utc_now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def utc_now() -> datetime:
    return datetime.now(timezone.utc)


def normalize_text(text: str) -> str:
    normalized = text.replace("\u00a0", " ").strip()
    return SPACE_RE.sub(" ", normalized)


def normalize_multiline_text(text: str) -> str:
    normalized = text.replace("\u00a0", " ").replace("\r\n", "\n").replace("\r", "\n")
    lines = [normalize_text(line) for line in normalized.split("\n")]
    kept = [line for line in lines if line]
    return "\n".join(kept)


def split_document_lines(text: str) -> list[str]:
    normalized = normalize_multiline_text(text)
    if not normalized:
        return [""]
    if "\n" in normalized:
        return normalized.split("\n")

    segments = [normalize_text(part) for part in SENTENCE_SPLIT_RE.split(normalized)]
    kept = [segment for segment in segments if segment]
    return kept or [normalized]


def content_hash(parts: Iterable[str]) -> str:
    joined = "\n".join(part for part in parts if part)
    return hashlib.sha256(joined.encode("utf-8")).hexdigest()


def split_csv_tags(raw: str) -> list[str]:
    items = [item.strip() for item in raw.split(",")]
    return [item for item in items if item]