64 lines
2.6 KiB
Python
64 lines
2.6 KiB
Python
"""Project model-facing timestamps after saving the original query evidence."""
|
|
from datetime import datetime, timedelta, timezone
|
|
import re
|
|
from typing import Any
|
|
|
|
DISPLAY_TIME_ZONE = "Asia/Shanghai"
|
|
DISPLAY_OFFSET = timezone(timedelta(hours=8))
|
|
_INSTANT = re.compile(
|
|
r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?P<fraction>\.\d+)?(?:Z|[+-]\d{2}:\d{2})"
|
|
)
|
|
_PROSE_FIELDS = frozenset({
|
|
"title", "summary", "detail", "description", "notice", "label", "message",
|
|
"current_conclusion", "conclusion_boundary", "recommendations", "pending_items",
|
|
"missing_data", "required_data", "impact", "scope", "fallback_text", "text",
|
|
})
|
|
_OPAQUE_FIELDS = frozenset({"id", "cursor", "url", "uri", "path", "sha256", "instructions", "code", "sql"})
|
|
_OPAQUE_SUFFIXES = ("_id", "_ids", "_ref", "_refs", "_cursor", "_digest", "_sha256", "_url", "_uri", "_path")
|
|
|
|
|
|
def _opaque(key: str) -> bool:
|
|
return key in _OPAQUE_FIELDS or key.endswith(_OPAQUE_SUFFIXES)
|
|
|
|
|
|
def _display_instant(value: str) -> str:
|
|
match = _INSTANT.fullmatch(value)
|
|
if match is None:
|
|
return value
|
|
try:
|
|
local = datetime.fromisoformat(value).astimezone(DISPLAY_OFFSET)
|
|
except (ValueError, OverflowError):
|
|
return value
|
|
fraction = match.group("fraction") or ""
|
|
return (f"{local.year:04d}-{local.month:02d}-{local.day:02d}T"
|
|
f"{local.hour:02d}:{local.minute:02d}:{local.second:02d}{fraction}+08:00")
|
|
|
|
|
|
def display_time_text(value: str) -> str:
|
|
"""Convert explicit instants in prose while retaining links and quoted code."""
|
|
parts = re.split(r"(https?://\S+|tjwater://\S+|`[^`]*`)", value)
|
|
for index in range(0, len(parts), 2):
|
|
parts[index] = re.sub(
|
|
r"(?<![A-Za-z0-9_/+:.-])" + _INSTANT.pattern + r"(?![A-Za-z0-9_:+/-])",
|
|
lambda match: _display_instant(match.group()), parts[index],
|
|
)
|
|
return "".join(parts)
|
|
|
|
|
|
def display_query_result(value: Any, *, field: str = "") -> Any:
|
|
"""Keep instants, fractions and identifiers intact; change display offsets."""
|
|
if _opaque(field):
|
|
return value
|
|
if isinstance(value, dict):
|
|
path = value.get("selected_path") or []
|
|
opaque_selection = isinstance(path, list) and bool(path) and _opaque(str(path[-1]))
|
|
return {key: item if key == "value" and opaque_selection else display_query_result(item, field=key)
|
|
for key, item in value.items()}
|
|
if isinstance(value, list):
|
|
return [display_query_result(item, field=field) for item in value]
|
|
if isinstance(value, str):
|
|
if field in _PROSE_FIELDS:
|
|
return display_time_text(value)
|
|
return _display_instant(value) if len(value) <= 128 else value
|
|
return value
|