""" 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())}")