from __future__ import annotations import hashlib import json import os import re import tempfile from datetime import datetime, timezone from pathlib import Path from typing import Any from claridoc.models import ValidationError _TAG_PATTERN = re.compile(r"<(?P[A-Z0-9_]+)>\s*(?P.*?)\s*", re.DOTALL) def read_json(path: str | Path) -> dict[str, Any]: file_path = Path(path) try: with file_path.open("r", encoding="utf-8") as handle: data = json.load(handle) except FileNotFoundError as exc: raise ValidationError(f"file not found: {file_path}") from exc except json.JSONDecodeError as exc: raise ValidationError(f"invalid JSON in {file_path}: line {exc.lineno}, column {exc.colno}: {exc.msg}") from exc if not isinstance(data, dict): raise ValidationError(f"top-level JSON value must be an object: {file_path}") return data def atomic_write_text(path: str | Path, content: str) -> Path: target = Path(path) target.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile( "w", encoding="utf-8", dir=target.parent, delete=False, newline="\n" ) as handle: handle.write(content) temp_name = handle.name os.replace(temp_name, target) return target def write_json(path: str | Path, data: Any) -> Path: return atomic_write_text(path, json.dumps(data, ensure_ascii=False, indent=2) + "\n") def extract_json_object(text: str) -> dict[str, Any]: stripped = text.strip() candidates = [stripped] fenced = re.findall(r"```(?:json)?\s*(\{.*?\})\s*```", stripped, flags=re.DOTALL | re.IGNORECASE) candidates.extend(fenced) first = stripped.find("{") last = stripped.rfind("}") if first >= 0 and last > first: candidates.append(stripped[first : last + 1]) errors: list[str] = [] for candidate in candidates: try: value = json.loads(candidate) except json.JSONDecodeError as exc: errors.append(exc.msg) continue if isinstance(value, dict): return value raise ValidationError("provider did not return a valid JSON object" + (f": {errors[-1]}" if errors else "")) def extract_tag(text: str, tag: str) -> str: for match in _TAG_PATTERN.finditer(text): if match.group("tag") == tag: return match.group("body").strip() raise ValidationError(f"missing tagged block: {tag}") def extract_tag_json(text: str, tag: str) -> dict[str, Any]: return extract_json_object(extract_tag(text, tag)) def utc_now_iso() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat() def sha256_file(path: str | Path) -> str: digest = hashlib.sha256() with Path(path).open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def slugify(text: str, fallback: str = "document") -> str: normalized = re.sub(r"[^0-9A-Za-z가-힣]+", "-", text.strip().lower()).strip("-") return normalized or fallback def word_count(text: str) -> int: without_code = re.sub(r"```.*?```", " ", text, flags=re.DOTALL) return len(re.findall(r"\b[\w가-힣]+\b", without_code, flags=re.UNICODE)) def line_number(text: str, index: int) -> int: return text.count("\n", 0, index) + 1 def normalize_heading(text: str) -> str: return re.sub(r"[^0-9a-z가-힣]+", "", text.casefold()) def strip_code_blocks(text: str) -> str: return re.sub(r"```.*?```", "", text, flags=re.DOTALL)