112 lines
3.5 KiB
Python
112 lines
3.5 KiB
Python
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<tag>[A-Z0-9_]+)>\s*(?P<body>.*?)\s*</(?P=tag)>", 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)
|