551 lines
21 KiB
Python
551 lines
21 KiB
Python
"""
|
||
faction_utils.py — утилиты для работы с фракциями через Obsidian vault.
|
||
|
||
Все операции с файлами (.md) выполняются через HTTP-эндпоинт бэкенда
|
||
/api/vault/query, который проксирует запрос в Obsidian через WebSocket.
|
||
Obsidian отвечает через vault_query_response.
|
||
|
||
Экспортирует:
|
||
vault_query(query_type, payload) — базовый вызов
|
||
get_all_factions(folder) — список всех фракций
|
||
get_faction(path) — данные одной фракции
|
||
get_location(wikilink_or_path) — данные локации
|
||
get_travel_distance(from_loc, to_loc) — расстояние в милях
|
||
get_turns_to_travel(from_loc, to_loc) — ходов на перемещение
|
||
are_within_one_move(loc_a, loc_b) — в пределах одного хода?
|
||
get_assets_in_location(faction, loc) — активы фракции в локации
|
||
get_factions_in_location(factions, loc) — фракции в локации
|
||
get_recent_events(folder, tags, limit) — последние события
|
||
write_faction(path, frontmatter, body) — сохранить фракцию
|
||
create_event_file(path, content) — создать файл события
|
||
normalize_wikilink(s) — '[[Waterdeep]]' → 'waterdeep'
|
||
resolve_wikilink(name) — wikilink → путь к файлу
|
||
roll_dice(notation) — '2d6+1' → int
|
||
roll_attribute_check(a, d, extra_a, extra_d) — WWN attribute check
|
||
"""
|
||
|
||
import re
|
||
import math
|
||
import random
|
||
import requests as http_requests
|
||
from typing import Optional
|
||
|
||
from wwn_mechanics import calc_hp_max
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
scripts_path = Path(__file__).resolve().parent.parent
|
||
if str(scripts_path) not in sys.path:
|
||
sys.path.insert(0, str(scripts_path))
|
||
|
||
from vault_utils import (
|
||
vault_query,
|
||
write_file as write_faction, # алиас для обратной совместимости
|
||
create_file as create_event_file, # алиас для обратной совместимости
|
||
normalize_wikilink,
|
||
resolve_wikilink,
|
||
)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Настройки
|
||
# ---------------------------------------------------------------------------
|
||
|
||
BACKEND_URL = "http://localhost:5000/api"
|
||
VAULT_QUERY_TIMEOUT = 12 # секунд
|
||
|
||
# По WWN: один faction turn ≈ 1 месяц ≈ 100 миль перемещения
|
||
WWN_MILES_PER_TURN = 100
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Фракции
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def get_all_factions(
|
||
factions_folder: str = "Фракции"
|
||
) -> list[dict]:
|
||
"""
|
||
Возвращает список всех фракций из папки.
|
||
|
||
Каждый элемент:
|
||
{
|
||
"path": "Фракции/Cult_of_the_Dragon.md",
|
||
"name": "Cult_of_the_Dragon",
|
||
"frontmatter": { ... }
|
||
}
|
||
"""
|
||
result = vault_query("get_all_factions", {
|
||
"folder": factions_folder,
|
||
"tag_filter": "faction"
|
||
})
|
||
if isinstance(result, list):
|
||
factions = result
|
||
elif isinstance(result, dict):
|
||
factions = result.get("data", [])
|
||
else:
|
||
factions = []
|
||
|
||
for f in factions:
|
||
f["frontmatter"] = normalize_faction_frontmatter(f.get("frontmatter", {}))
|
||
return factions
|
||
|
||
|
||
def get_faction(file_path: str) -> dict:
|
||
"""
|
||
Возвращает данные одной фракции: frontmatter + body.
|
||
|
||
Возвращает:
|
||
{
|
||
"path": "...",
|
||
"name": "...",
|
||
"frontmatter": { ... },
|
||
"body": "# Лор фракции..."
|
||
}
|
||
"""
|
||
return vault_query("get_frontmatter", {"path": file_path})
|
||
|
||
|
||
def get_active_factions(
|
||
factions_folder: str = "Фракции"
|
||
) -> list[dict]:
|
||
"""Возвращает только активные фракции (state != 'defeated')."""
|
||
all_factions = get_all_factions(factions_folder)
|
||
return [
|
||
f for f in all_factions
|
||
if f.get("frontmatter", {}).get("state") != "defeated"
|
||
]
|
||
|
||
|
||
def get_factions_in_location(all_factions: list[dict], locations_wikilinks: list[str] = None) -> list[dict]:
|
||
"""
|
||
Возвращает список фракций, присутствующих в указанных локациях.
|
||
Если locations_wikilinks пуст или None — возвращает все фракции.
|
||
"""
|
||
if not locations_wikilinks:
|
||
return all_factions
|
||
|
||
# Если locations_wikilinks — строка, преобразуем её в список
|
||
if isinstance(locations_wikilinks, str):
|
||
locations_wikilinks = [locations_wikilinks]
|
||
|
||
targets = [normalize_wikilink(loc) for loc in locations_wikilinks if loc]
|
||
if not targets:
|
||
return all_factions
|
||
|
||
result = []
|
||
for faction in all_factions:
|
||
fm = faction.get("frontmatter", {})
|
||
|
||
# Проверка списка locations фракции
|
||
locs = fm.get("locations", [])
|
||
if isinstance(locs, str): locs = [locs]
|
||
normalized_locs = [normalize_wikilink(l) for l in locs if isinstance(l, str)]
|
||
|
||
# Проверка локаций активов
|
||
asset_locs = [
|
||
normalize_wikilink(a.get("location", ""))
|
||
for a in fm.get("assets", []) if isinstance(a, dict)
|
||
]
|
||
|
||
# Если фракция есть хотя бы в одной из целевых локаций
|
||
if any(t in normalized_locs or t in asset_locs for t in targets):
|
||
result.append(faction)
|
||
|
||
return result
|
||
|
||
def get_assets_in_location(
|
||
faction_data: dict,
|
||
locations_wikilinks: list[str] = None
|
||
) -> list[dict]:
|
||
"""
|
||
Возвращает активы фракции в указанных локациях.
|
||
Если список локаций пуст — возвращает ВСЕ активы фракции.
|
||
"""
|
||
fm = faction_data.get("frontmatter", faction_data)
|
||
assets = fm.get("assets", [])
|
||
|
||
if not locations_wikilinks:
|
||
return [a for a in assets if isinstance(a, dict)]
|
||
|
||
targets = [normalize_wikilink(loc) for loc in locations_wikilinks if loc]
|
||
if not targets:
|
||
return [a for a in assets if isinstance(a, dict)]
|
||
|
||
return [
|
||
a for a in assets
|
||
if isinstance(a, dict) and normalize_wikilink(a.get("location", "")) in targets
|
||
]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Локации и расстояния
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def get_location(wikilink_or_path: str) -> dict:
|
||
"""
|
||
Возвращает данные локации: frontmatter + body.
|
||
|
||
Принимает '[[Waterdeep]]' или 'TTRPG/Locations/Waterdeep.md'.
|
||
|
||
frontmatter локации содержит поле travel-distance:
|
||
travel-distance:
|
||
"[[Neverwinter]]": 300
|
||
"[[Baldur's Gate]]": 250
|
||
"""
|
||
if wikilink_or_path.startswith("[[") or not wikilink_or_path.endswith(".md"):
|
||
path = resolve_wikilink(wikilink_or_path)
|
||
if not path:
|
||
raise ValueError(
|
||
f"Локация не найдена в vault: '{wikilink_or_path}'. "
|
||
"Проверьте что файл локации существует."
|
||
)
|
||
else:
|
||
path = wikilink_or_path
|
||
|
||
return vault_query("get_frontmatter", {"path": path})
|
||
|
||
|
||
def get_travel_distance(
|
||
from_location: str,
|
||
to_location: str
|
||
) -> Optional[int]:
|
||
"""
|
||
Возвращает расстояние в милях между двумя локациями.
|
||
|
||
Читает поле travel-distance из файла from_location.
|
||
Если нет прямой связи — пробует обратный маршрут.
|
||
Возвращает None если расстояние не задано.
|
||
|
||
Формат поля в Location .md:
|
||
travel-distance:
|
||
"[[Neverwinter]]": 300
|
||
"[[Baldur's Gate]]": 250
|
||
"[[Daggerford]]": 50
|
||
"""
|
||
to_normalized = normalize_wikilink(to_location)
|
||
|
||
# Прямой маршрут
|
||
try:
|
||
loc_data = get_location(from_location)
|
||
distances: dict = loc_data.get("frontmatter", {}).get("travel-distance", {})
|
||
for key, dist in distances.items():
|
||
if normalize_wikilink(str(key)) == to_normalized:
|
||
return int(dist)
|
||
except (ValueError, RuntimeError):
|
||
pass
|
||
|
||
# Обратный маршрут (граф ненаправленный)
|
||
from_normalized = normalize_wikilink(from_location)
|
||
try:
|
||
loc_data_reverse = get_location(to_location)
|
||
distances_reverse: dict = loc_data_reverse.get("frontmatter", {}).get("travel-distance", {})
|
||
for key, dist in distances_reverse.items():
|
||
if normalize_wikilink(str(key)) == from_normalized:
|
||
return int(dist)
|
||
except (ValueError, RuntimeError):
|
||
pass
|
||
|
||
return None
|
||
|
||
|
||
def get_turns_to_travel(
|
||
from_location: str,
|
||
to_location: str,
|
||
miles_per_turn: int = WWN_MILES_PER_TURN
|
||
) -> Optional[int]:
|
||
"""
|
||
Возвращает количество faction turns для перемещения актива.
|
||
|
||
По WWN: один ход ≈ 100 миль. Округление вверх.
|
||
Возвращает None если расстояние не задано.
|
||
"""
|
||
dist = get_travel_distance(from_location, to_location)
|
||
if dist is None:
|
||
return None
|
||
return math.ceil(dist / miles_per_turn)
|
||
|
||
|
||
def are_within_one_move(
|
||
location_a: str,
|
||
location_b: str,
|
||
miles_per_turn: int = WWN_MILES_PER_TURN
|
||
) -> bool:
|
||
"""
|
||
Проверяет находятся ли две локации в пределах одного faction turn move.
|
||
|
||
Используется для проверки радиуса спецспособностей
|
||
(Informers, Vigilant Agents, Omniscient Seers и т.д.).
|
||
|
||
Если расстояние неизвестно — возвращает True (safe default,
|
||
лучше разрешить чем молча заблокировать).
|
||
"""
|
||
dist = get_travel_distance(location_a, location_b)
|
||
if dist is None:
|
||
return True
|
||
return dist <= miles_per_turn
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# События
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def get_recent_events(
|
||
events_folder: str = "События",
|
||
tags_filter: Optional[list[str]] = None,
|
||
max_count: int = 5
|
||
) -> list[dict]:
|
||
"""
|
||
Возвращает последние N событий из папки events_folder.
|
||
|
||
tags_filter — список тегов (хотя бы один должен совпасть).
|
||
None — без фильтрации.
|
||
|
||
Каждый элемент:
|
||
{
|
||
"path": "События/2026-06-22_Cult_Attack.md",
|
||
"name": "2026-06-22_Cult_Attack",
|
||
"frontmatter": { ... }
|
||
}
|
||
"""
|
||
result = vault_query("get_recent_files", {
|
||
"folder": events_folder,
|
||
"limit": max_count,
|
||
"tags": tags_filter or []
|
||
})
|
||
if isinstance(result, list):
|
||
return result
|
||
return []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Кубики (WWN-механика)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def roll_dice(notation: str) -> int:
|
||
"""
|
||
Бросает кубики по нотации и возвращает результат.
|
||
|
||
Поддерживает: '1d8', '2d6', '1d10+3', '2d6+2', '1d8-1'.
|
||
Минимальный результат: 1.
|
||
"""
|
||
notation = notation.strip().lower()
|
||
|
||
match = re.fullmatch(r'(\d+)d(\d+)([+-]\d+)?', notation)
|
||
if not match:
|
||
raise ValueError(
|
||
f"Неверный формат нотации: '{notation}'. "
|
||
"Ожидается: '2d6', '1d8+2', '1d10-1'."
|
||
)
|
||
|
||
num_dice = int(match.group(1))
|
||
die_sides = int(match.group(2))
|
||
modifier = int(match.group(3)) if match.group(3) else 0
|
||
|
||
total = sum(random.randint(1, die_sides) for _ in range(num_dice)) + modifier
|
||
return max(1, total)
|
||
|
||
|
||
def roll_attribute_check(
|
||
attacker_attr: int,
|
||
defender_attr: int,
|
||
attacker_extra_die: bool = False,
|
||
defender_extra_die: bool = False,
|
||
) -> tuple[int, int, bool]:
|
||
"""
|
||
Выполняет attribute check по правилам WWN.
|
||
|
||
Оба бросают 1d10 + атрибут.
|
||
Атакующий побеждает только при СТРОГО большем результате (ничья = защитник).
|
||
|
||
attacker_extra_die — True если faction tag даёт extra die
|
||
(Machiavellian/Martial/Rich): бросаются 2d10, берётся лучший.
|
||
defender_extra_die — аналогично для защитника.
|
||
|
||
Возвращает: (attacker_total, defender_total, attacker_wins: bool)
|
||
"""
|
||
def _roll(attr: int, extra: bool) -> int:
|
||
if extra:
|
||
return max(random.randint(1, 10), random.randint(1, 10)) + attr
|
||
return random.randint(1, 10) + attr
|
||
|
||
a_total = _roll(attacker_attr, attacker_extra_die)
|
||
d_total = _roll(defender_attr, defender_extra_die)
|
||
return a_total, d_total, a_total > d_total
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Тест
|
||
# ---------------------------------------------------------------------------
|
||
|
||
if __name__ == "__main__":
|
||
print("=== Тест faction_utils.py ===\n")
|
||
|
||
print("--- roll_dice ---")
|
||
for notation in ["1d8", "2d6", "1d10+3", "1d6-1", "3d4+2"]:
|
||
print(f" {notation} → {roll_dice(notation)}")
|
||
|
||
print("\n--- normalize_wikilink ---")
|
||
for s in [
|
||
"[[Waterdeep]]",
|
||
"[[Waterdeep|City of Splendors]]",
|
||
"Waterdeep",
|
||
"[[baldur's gate]]",
|
||
"",
|
||
]:
|
||
print(f" '{s}' → '{normalize_wikilink(s)}'")
|
||
|
||
print("\n--- roll_attribute_check (Force 5 vs Force 3, 10 бросков) ---")
|
||
wins = sum(1 for _ in range(10) if roll_attribute_check(5, 3)[2])
|
||
print(f" Атакующий выиграл {wins}/10")
|
||
|
||
print("\n--- roll_attribute_check с extra die (Cunning 5 vs Cunning 5) ---")
|
||
wins_extra = sum(1 for _ in range(10) if roll_attribute_check(5, 5, attacker_extra_die=True)[2])
|
||
print(f" С extra die: {wins_extra}/10 (ожидаем > 5)")
|
||
|
||
print("\n--- vault_query (требует запущенный бэкенд + Obsidian) ---")
|
||
try:
|
||
factions = get_all_factions("Фракции")
|
||
print(f" Найдено фракций: {len(factions)}")
|
||
if factions:
|
||
f = factions[0]
|
||
fm = f.get("frontmatter", {})
|
||
print(f" Первая: {f['name']} | HP: {fm.get('hp')} | Treasure: {fm.get('treasure')}")
|
||
except RuntimeError as e:
|
||
print(f" Пропущен (бэкенд недоступен): {e}")
|
||
|
||
print("\n--- get_travel_distance (требует бэкенд + Obsidian) ---")
|
||
try:
|
||
dist = get_travel_distance("[[Waterdeep]]", "[[Neverwinter]]")
|
||
turns = get_turns_to_travel("[[Waterdeep]]", "[[Neverwinter]]")
|
||
in_range = are_within_one_move("[[Waterdeep]]", "[[Daggerford]]")
|
||
print(f" Waterdeep → Neverwinter: {dist} миль, {turns} ходов")
|
||
print(f" Waterdeep → Daggerford ≤100 миль: {in_range}")
|
||
except RuntimeError as e:
|
||
print(f" Пропущен (бэкенд недоступен): {e}")
|
||
|
||
print("\n=== Готово ===")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Нормализация фракции: дефолты для отсутствующих полей
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# Дефолтные механические значения WWN для фракции
|
||
FACTION_DEFAULTS: dict = {
|
||
"force": 1,
|
||
"wealth": 1,
|
||
"cunning": 1,
|
||
"magic": "None",
|
||
"treasure": 0,
|
||
"xp": 0,
|
||
"state": "active",
|
||
"locations": [],
|
||
"current_goal": "",
|
||
"faction_tags": [],
|
||
"tags_listen": [],
|
||
"dw_category": "",
|
||
"impulse": "",
|
||
"enemies": [], # явные враги (имена файлов фракций)
|
||
"allies": [], # явные союзники (имена файлов фракций)
|
||
}
|
||
|
||
|
||
def _clean_list_field(value) -> list:
|
||
"""
|
||
Приводит поле-список к чистому виду:
|
||
- None → []
|
||
- строка → [строка] (если не пустая)
|
||
- список → без пустых строк "" и None
|
||
Используется для locations, motivations, assets и т.п.
|
||
"""
|
||
if value is None:
|
||
return []
|
||
if isinstance(value, str):
|
||
return [value] if value.strip() else []
|
||
if isinstance(value, list):
|
||
cleaned = []
|
||
for item in value:
|
||
if item is None:
|
||
continue
|
||
if isinstance(item, str) and not item.strip():
|
||
continue # пустая строка-заглушка "- ''"
|
||
cleaned.append(item)
|
||
return cleaned
|
||
return []
|
||
|
||
|
||
def _clean_assets(fm: dict) -> list[dict]:
|
||
"""
|
||
Оставляет только валидные активы-объекты с непустым name.
|
||
Отбрасывает строки-заглушки ('- ""') и мусор.
|
||
Для каждого актива подставляет дефолты недостающих полей.
|
||
"""
|
||
raw = _clean_list_field(fm.get("assets"))
|
||
result = []
|
||
for a in raw:
|
||
if not isinstance(a, dict) or not a.get("name"):
|
||
continue
|
||
# Дефолты полей актива
|
||
is_boi = a.get("is_base_of_influence", False)
|
||
a.setdefault("is_base_of_influence", is_boi)
|
||
a.setdefault("location", "")
|
||
a.setdefault("stealthed", False)
|
||
a.setdefault("subtle", False)
|
||
a.setdefault("hp_max", a.get("hp", 1))
|
||
a.setdefault("hp", a.get("hp_max", 1))
|
||
a.setdefault("cost", 0)
|
||
if not is_boi:
|
||
a.setdefault("attribute", "Force")
|
||
a.setdefault("tier_required", 1)
|
||
result.append(a)
|
||
return result
|
||
|
||
|
||
def normalize_faction_frontmatter(fm: dict) -> dict:
|
||
"""
|
||
Приводит frontmatter фракции к формату, ожидаемому механикой WWN.
|
||
|
||
- Подставляет дефолты для отсутствующих скалярных полей.
|
||
- Чистит списочные поля (locations, motivations, assets...).
|
||
- Если основная 'location' не задана, но есть непустой 'locations',
|
||
берёт первый элемент как основную локацию.
|
||
- Пересчитывает hp_max по атрибутам, если не задан;
|
||
подставляет hp = hp_max, если hp не задан.
|
||
|
||
Модифицирует fm на месте и возвращает его же.
|
||
"""
|
||
# Скалярные дефолты
|
||
for key, default in FACTION_DEFAULTS.items():
|
||
if fm.get(key) is None:
|
||
fm[key] = default.copy() if isinstance(default,
|
||
(list, dict)) else default
|
||
|
||
# Списочные поля (нарративные) — чистим от пустых заглушек
|
||
for list_field in ("locations", "motivations", "headquarters", "leaders",
|
||
"enemies"):
|
||
fm[list_field] = _clean_list_field(fm.get(list_field))
|
||
|
||
# tags_listen / faction_tags тоже чистим
|
||
fm["faction_tags"] = _clean_list_field(fm.get("faction_tags"))
|
||
fm["tags_listen"] = _clean_list_field(fm.get("tags_listen"))
|
||
|
||
# Активы
|
||
fm["assets"] = _clean_assets(fm)
|
||
|
||
# HP: считаем hp_max по атрибутам, если не задан
|
||
if fm.get("hp_max") is None:
|
||
fm["hp_max"] = calc_hp_max(
|
||
fm.get("force", 1),
|
||
fm.get("wealth", 1),
|
||
fm.get("cunning", 1),
|
||
)
|
||
if fm.get("hp") is None:
|
||
fm["hp"] = fm["hp_max"]
|
||
|
||
# Гарантируем наличие и чистоту enemies/allies
|
||
fm.setdefault("enemies", [])
|
||
fm.setdefault("allies", [])
|
||
fm["enemies"] = [str(x).strip() for x in fm["enemies"] if x]
|
||
fm["allies"] = [str(x).strip() for x in fm["allies"] if x]
|
||
|
||
return fm
|