210 lines
8.8 KiB
Python
210 lines
8.8 KiB
Python
"""
|
||
vault_utils.py — контексто-независимые утилиты для работы с Obsidian REST API.
|
||
Универсальный модуль без привязки к игровой специфике (фракциям, событиям и т.д.).
|
||
"""
|
||
|
||
import re
|
||
import yaml
|
||
import requests
|
||
import urllib.parse
|
||
from typing import Optional, Dict, Any
|
||
|
||
# ----------------------------------------------- Configuration ------------------------------------------------------------
|
||
REST_API_PORT = 27124
|
||
API_KEY = "31c4ee60e78cea8503d0ac076298a0da6f5cbc3992f4e4cdc97d84fcbf27248e"
|
||
BASE_URL = f"https://127.0.0.1:{REST_API_PORT}"
|
||
|
||
def _get_headers() -> Dict[str, str]:
|
||
"""Возвращает базовые заголовки для запросов к REST API."""
|
||
return {"Authorization": f"Bearer {API_KEY}", "Content-Type": "text/markdown"}
|
||
|
||
def _get_req_kwargs() -> Dict[str, Any]:
|
||
"""Возвращает параметры для отключения SSL и прокси."""
|
||
return {"verify": False, "proxies": {"http": None, "https": None}}
|
||
|
||
# ----------------------------------------------- Parsers ------------------------------------------------------------
|
||
def _parse_obsidian_content(content: str, path: str) -> Dict[str, Any]:
|
||
"""
|
||
Парсит контент из Markdown в структуру с Frontmatter.
|
||
"""
|
||
if content.startswith("---"):
|
||
parts = content.split("---", 2)
|
||
if len(parts) >= 3:
|
||
return {
|
||
"path": path,
|
||
"frontmatter": yaml.safe_load(parts[1]) or {},
|
||
"body": parts[2].strip(),
|
||
"name": path.split('/')[-1].replace('.md', '')
|
||
}
|
||
return {"path": path, "frontmatter": {}, "body": content, "name": path.split('/')[-1]}
|
||
|
||
# ----------------------------------------------- Core API ------------------------------------------------------------
|
||
def get_file(path: str) -> Dict[str, Any]:
|
||
"""Получает файл из Obsidian и возвращает распарсенный контент."""
|
||
encoded_path = urllib.parse.quote(path)
|
||
resp = requests.get(f"{BASE_URL}/vault/{encoded_path}", headers=_get_headers(), **_get_req_kwargs())
|
||
if resp.status_code == 200:
|
||
return _parse_obsidian_content(resp.text, path)
|
||
raise RuntimeError(f"Файл не найден: {path}")
|
||
|
||
def write_file(file_path: str, frontmatter: dict, body: str) -> bool:
|
||
"""Сохраняет frontmatter и тело файла в Obsidian vault."""
|
||
encoded_path = urllib.parse.quote(file_path)
|
||
fm = yaml.dump(frontmatter, allow_unicode=True)
|
||
content = f"---\n{fm}---\n\n{body}"
|
||
resp = requests.put(
|
||
f"{BASE_URL}/vault/{encoded_path}",
|
||
headers=_get_headers(),
|
||
data=content.encode('utf-8'),
|
||
**_get_req_kwargs()
|
||
)
|
||
return resp.status_code in [200, 201, 204]
|
||
|
||
def create_file(file_path: str, content: str) -> bool:
|
||
"""Создаёт новый файл в Obsidian vault с указанным контентом."""
|
||
encoded_path = urllib.parse.quote(file_path)
|
||
resp = requests.put(
|
||
f"{BASE_URL}/vault/{encoded_path}",
|
||
headers=_get_headers(),
|
||
data=content.encode('utf-8'),
|
||
**_get_req_kwargs()
|
||
)
|
||
return resp.status_code in [200, 201, 204]
|
||
|
||
def patch_frontmatter(path: str, target: str, value: Any, operation: str = "replace") -> bool:
|
||
"""Обновляет или добавляет поле во frontmatter файла."""
|
||
encoded_path = urllib.parse.quote(path)
|
||
headers = _get_headers()
|
||
headers.update({
|
||
"Operation": operation,
|
||
"Target-Type": "frontmatter",
|
||
"Target": target,
|
||
"Create-Target-If-Missing": "true",
|
||
"Content-Type": "application/json"
|
||
})
|
||
val = str(value)
|
||
resp = requests.patch(f"{BASE_URL}/vault/{encoded_path}", headers=headers, data=val.encode('utf-8'), **_get_req_kwargs())
|
||
return resp.status_code in [200, 204]
|
||
|
||
def search_by_filename(filename_query: str) -> list[str]:
|
||
"""
|
||
Выполняет поиск в Obsidian строго по названию файла (или части пути).
|
||
Возвращает список путей к найденным файлам.
|
||
"""
|
||
# Формируем JsonLogic запрос, который вернет true, если filename_query есть в пути
|
||
query = {
|
||
"glob": [f"*{filename_query}*.md", {"var": "path"}]
|
||
}
|
||
|
||
# Используем вашу готовую функцию для отправки JsonLogic запроса
|
||
raw_results = search_jsonlogic(query)
|
||
|
||
# Эндпоинт возвращает список словарей вида [{"filename": "...", "result": true}]
|
||
# Отфильтруем и вернем только пути (имена файлов)
|
||
return [item.get("filename") for item in raw_results if item.get("filename")]
|
||
|
||
'''def search_simple(query: str) -> list[dict]:
|
||
"""Выполняет простой текстовый поиск в Obsidian."""
|
||
resp = requests.post(
|
||
f"{BASE_URL}/search/simple/",
|
||
headers=_get_headers(),
|
||
params={"query": query},
|
||
**_get_req_kwargs()
|
||
)
|
||
if resp.status_code == 200:
|
||
return resp.json()
|
||
return []'''
|
||
|
||
def search_jsonlogic(query: dict) -> list[dict]:
|
||
"""Выполняет поиск по правилам JsonLogic в Obsidian."""
|
||
headers = _get_headers()
|
||
headers["Content-Type"] = "application/vnd.olrapi.jsonlogic+json"
|
||
resp = requests.post(
|
||
f"{BASE_URL}/search/",
|
||
headers=headers,
|
||
json=query,
|
||
**_get_req_kwargs()
|
||
)
|
||
if resp.status_code == 200:
|
||
return resp.json()
|
||
return []
|
||
|
||
def list_directory(path: str) -> dict:
|
||
"""Возвращает список файлов и папок в указанной директории."""
|
||
encoded_path = urllib.parse.quote(path) if path else ""
|
||
resp = requests.get(f"{BASE_URL}/vault/{encoded_path}", headers=_get_headers(), **_get_req_kwargs())
|
||
if resp.status_code == 200:
|
||
return resp.json()
|
||
return {"files": []}
|
||
|
||
# ----------------------------------------------- Utilities ------------------------------------------------------------
|
||
def normalize_wikilink(s: str) -> str:
|
||
"""Приводит wikilink к нижнему регистру без скобок и алиасов."""
|
||
if not s:
|
||
return ''
|
||
return re.sub(r'^\[\[|\]\]$', '', s).split('|')[0].strip().lower()
|
||
|
||
def resolve_wikilink(name: str) -> Optional[str]:
|
||
"""Ищет путь к файлу по имени wikilink через простой поиск."""
|
||
original_clean = re.sub(r'^\[\[|\]\]$', '', name).split('|')[0].strip()
|
||
results = search_by_filename(original_clean)
|
||
return results[0] if results else None
|
||
|
||
|
||
def get_recent_files(
|
||
folder: str,
|
||
tags: Optional[list[str]] = None,
|
||
max_count: Optional[int] = None
|
||
) -> list[dict]:
|
||
"""
|
||
Ищет файлы в папке, фильтрует по тегам и возвращает отсортированными по убыванию даты изменения (mtime).
|
||
Использует пакетную загрузку JsonLogic для минимизации запросов к API.
|
||
"""
|
||
conditions = [{"glob": [f"{folder}/*.md", {"var": "path"}]}]
|
||
|
||
if tags:
|
||
tag_conditions = [{"in": [tag.lstrip('#'), {"var": "tags"}]} for tag in tags]
|
||
conditions.append({"or": tag_conditions})
|
||
|
||
query = {
|
||
"if": [
|
||
{"and": conditions},
|
||
{"var": ""},
|
||
False
|
||
]
|
||
}
|
||
|
||
search_results = search_jsonlogic(query)
|
||
valid_results = [r for r in search_results if isinstance(r.get("result"), dict)]
|
||
|
||
valid_results.sort(
|
||
key=lambda x: x.get("result", {}).get("stat", {}).get("mtime", 0),
|
||
reverse=True
|
||
)
|
||
|
||
if max_count is not None:
|
||
valid_results = valid_results[:max_count]
|
||
|
||
files_data = []
|
||
for res in valid_results:
|
||
note_json = res.get("result", {})
|
||
file_path = res.get("filename", "")
|
||
files_data.append({
|
||
"path": file_path,
|
||
"name": file_path.split('/')[-1].replace('.md', ''),
|
||
"frontmatter": note_json.get("frontmatter", {}) or {},
|
||
"body": note_json.get("content", "")
|
||
})
|
||
|
||
return files_data
|
||
|
||
# ----------------------------------------------- Main ------------------------------------------------------------
|
||
if __name__ == "__main__":
|
||
import json
|
||
print("Проверка get_recent_files...")
|
||
# Ищем файлы в папке Фракции с тегом faction
|
||
files = get_recent_files("Фракции", tags=["faction"], max_count=3)
|
||
print(f"Найдено файлов: {len(files)}")
|
||
for f in files:
|
||
print(f"Путь: {f['path']}")
|
||
print(f"Фронтматтер keys: {list(f['frontmatter'].keys())}") |