135 lines
5.5 KiB
Python
135 lines
5.5 KiB
Python
"""
|
||
vault_utils.py — универсальные утилиты для работы с Obsidian vault через HTTP-прокси.
|
||
|
||
Содержит функции, переиспользуемые любыми модулями (фракции, время, события и т.д.).
|
||
|
||
Экспортирует:
|
||
vault_query(query_type, payload) — базовый вызов к бэкенду
|
||
write_file(path, frontmatter, body) — запись frontmatter и тела файла
|
||
create_file(path, content) — создание файла с содержимым
|
||
normalize_wikilink(s) — приведение wikilink к нормализованному виду
|
||
resolve_wikilink(name) — поиск пути к файлу по имени/алиасу
|
||
"""
|
||
|
||
import re
|
||
import requests as http_requests
|
||
from typing import Optional
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Настройки
|
||
# ---------------------------------------------------------------------------
|
||
|
||
BACKEND_URL = "http://localhost:5000/api"
|
||
VAULT_QUERY_TIMEOUT = 12 # секунд
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Базовый клиент
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def vault_query(query_type: str, payload: dict) -> any:
|
||
"""
|
||
Отправляет запрос к Obsidian vault через бэкенд-прокси.
|
||
|
||
Блокирующий вызов: ждёт пока Obsidian выполнит операцию.
|
||
Бросает RuntimeError при недоступности бэкенда или timeout.
|
||
|
||
Поля:
|
||
query_type: тип операции
|
||
payload: параметры операции
|
||
"""
|
||
try:
|
||
resp = http_requests.post(
|
||
f"{BACKEND_URL}/vault/query",
|
||
json={"type": query_type, "payload": payload},
|
||
timeout=VAULT_QUERY_TIMEOUT,
|
||
proxies={"http": None, "https": None}
|
||
)
|
||
except http_requests.exceptions.ConnectionError as e:
|
||
raise RuntimeError(
|
||
f"Бэкенд недоступен ({BACKEND_URL}). "
|
||
"Убедитесь что Flask-сервер запущен."
|
||
)
|
||
except http_requests.exceptions.Timeout:
|
||
raise RuntimeError(
|
||
f"vault_query timeout после {VAULT_QUERY_TIMEOUT}с. "
|
||
"Возможно Obsidian не подключён к WebSocket."
|
||
)
|
||
|
||
if resp.status_code == 504:
|
||
raise RuntimeError(
|
||
"Obsidian не ответил на запрос vault за 10 секунд. "
|
||
"Проверьте что плагин LLM Agent запущен и подключён."
|
||
)
|
||
|
||
if not resp.ok:
|
||
try:
|
||
err = resp.json().get("error", resp.text)
|
||
except Exception:
|
||
err = resp.text
|
||
raise RuntimeError(f"vault_query ошибка {resp.status_code}: {err}")
|
||
|
||
return resp.json()
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Запись данных (универсальные)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def write_file(file_path: str, frontmatter: dict, body: str) -> bool:
|
||
"""
|
||
Сохраняет frontmatter и тело файла в Obsidian vault.
|
||
|
||
file_path — путь внутри vault, например 'Фракции/Cult_of_the_Dragon.md'
|
||
frontmatter — словарь с данными (поля YAML)
|
||
body — содержимое файла после frontmatter
|
||
"""
|
||
result = vault_query("write_frontmatter", {
|
||
"path": file_path,
|
||
"frontmatter": frontmatter,
|
||
"body": body
|
||
})
|
||
if isinstance(result, dict):
|
||
return result.get("success", False)
|
||
return False
|
||
|
||
def create_file(file_path: str, content: str) -> bool:
|
||
"""
|
||
Создаёт файл в Obsidian vault (перезаписывает, если существует).
|
||
Промежуточные папки создаются автоматически.
|
||
"""
|
||
result = vault_query("create_file", {
|
||
"path": file_path,
|
||
"content": content
|
||
})
|
||
if isinstance(result, dict):
|
||
return result.get("success", False)
|
||
return False
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Нормализация и резолвинг wikilink
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def normalize_wikilink(s: str) -> str:
|
||
"""
|
||
Приводит wikilink к нижнему регистру без скобок и алиасов.
|
||
|
||
Примеры:
|
||
'[[Waterdeep]]' → 'waterdeep'
|
||
'[[Waterdeep|City]]' → 'waterdeep'
|
||
'Waterdeep' → 'waterdeep'
|
||
'' → ''
|
||
"""
|
||
if not s:
|
||
return ''
|
||
return re.sub(r'^\[\[|\]\]$', '', s).split('|')[0].strip().lower()
|
||
|
||
def resolve_wikilink(name: str) -> Optional[str]:
|
||
"""
|
||
Резолвит wikilink в путь к файлу через Obsidian metadataCache.
|
||
Знает об aliases.
|
||
|
||
Принимает '[[Waterdeep]]', '[[Waterdeep|City]]' или просто 'Waterdeep'.
|
||
Возвращает путь вида 'TTRPG/Locations/Waterdeep.md' или None.
|
||
"""
|
||
original_clean = re.sub(r'^\[\[|\]\]$', '', name).split('|')[0].strip()
|
||
result = vault_query("resolve_wikilink", {"name": original_clean})
|
||
return result if isinstance(result, str) else None |