Parth
e7d75c3b05
* Rewrite comments in Google developer documentation style Rewrite the comments and docstrings across the backend core modules so they read plainly. The previous prose was accurate but dense and figurative, which made it slow to skim. Applies the Google developer documentation style guide: short sentences, active voice, present tense, American spelling, and no metaphors, idioms, or rhetorical asides. Replaces em-dash chains with separate sentences.
113 lines
4.6 KiB
Python
113 lines
4.6 KiB
Python
"""Runs an adventure's enabled scripts through a turn's hook points, applying
|
|
state and story-card mutations back to the database after each hook."""
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .. import models
|
|
from ..context import history as context_history
|
|
from .engine import run_hook
|
|
|
|
MAX_STORY_CARDS = 5000 # AI Dungeon's per-adventure sanity cap
|
|
|
|
|
|
class ScriptPipeline:
|
|
def __init__(self, adventure: models.Adventure, db: Session):
|
|
self.adventure = adventure
|
|
self.db = db
|
|
self.logs: list[str] = []
|
|
self.errors: list[str] = []
|
|
|
|
@property
|
|
def message(self) -> str | None:
|
|
state = self.adventure.script_state
|
|
msg = state.get("message") if isinstance(state, dict) else None
|
|
return msg if isinstance(msg, str) and msg.strip() else None
|
|
|
|
def _history(self) -> list[dict]:
|
|
# Read the path rather than `adventure.actions`. That collection holds
|
|
# every branch's actions, and this is the documented history API a user
|
|
# script reads. Giving a script the siblings of the turn it is running on
|
|
# would be the same bug as building a prompt from them, and visible to
|
|
# the user.
|
|
#
|
|
# `story_actions` also drops rows with blank text, which
|
|
# `adventure.actions` kept, so this array is shorter than it was for an
|
|
# adventure that has any such rows. `info.actionCount` counts the same
|
|
# way. That is intended. A row with no text is this app's bookkeeping, it
|
|
# has no counterpart in the AI Dungeon history a ported script was
|
|
# written against, and the prompt has never included one. A script keyed
|
|
# on every N actions lands on different turns than it did before phase
|
|
# 14, and no reading of this is compatible with both.
|
|
return [
|
|
{"text": a.text, "rawText": a.text, "type": a.type}
|
|
for a in context_history.story_actions(self.adventure)
|
|
]
|
|
|
|
def _cards(self) -> list[dict]:
|
|
return [
|
|
{"id": c.id, "keys": c.keys, "entry": c.entry, "type": c.type}
|
|
for c in self.adventure.story_cards
|
|
]
|
|
|
|
def _info(self) -> dict:
|
|
return {
|
|
"actionCount": context_history.count(self.adventure),
|
|
"characterNames": [],
|
|
"memoryLength": len(self.adventure.memory),
|
|
"maxChars": 0,
|
|
}
|
|
|
|
def _apply_cards(self, returned: list) -> None:
|
|
existing = {c.id: c for c in self.adventure.story_cards}
|
|
seen_ids = set()
|
|
added = 0
|
|
for item in returned:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
card_id = item.get("id")
|
|
keys = str(item.get("keys") or "")
|
|
entry = str(item.get("entry") or "")
|
|
card_type = str(item.get("type") or "")
|
|
if card_id in existing:
|
|
seen_ids.add(card_id)
|
|
card = existing[card_id]
|
|
card.keys, card.entry, card.type = keys, entry, card_type
|
|
elif len(existing) + added < MAX_STORY_CARDS:
|
|
self.db.add(
|
|
models.StoryCard(
|
|
adventure_id=self.adventure.id,
|
|
keys=keys, entry=entry, type=card_type,
|
|
)
|
|
)
|
|
added += 1
|
|
for card_id, card in existing.items():
|
|
if card_id not in seen_ids:
|
|
self.db.delete(card)
|
|
|
|
def run(self, hook: str, text: str) -> tuple[str, bool]:
|
|
"""Chain `hook` across all enabled scripts. Returns (text, stop)."""
|
|
state = self.adventure.script_state if isinstance(self.adventure.script_state, dict) else {}
|
|
for script in self.adventure.scripts:
|
|
hook_js = getattr(script, f"{hook}_js")
|
|
if not script.enabled or not hook_js.strip():
|
|
continue
|
|
result = run_hook(
|
|
script.library_js, hook_js, text, state,
|
|
self._history(), self._cards(), self._info(),
|
|
)
|
|
if result.error:
|
|
self.errors.append(f"{script.name} ({hook}): {result.error}")
|
|
continue # a broken script never breaks the turn
|
|
self.logs.extend(f"[{script.name}/{hook}] {line}" for line in result.logs)
|
|
self._apply_cards(result.story_cards)
|
|
state = result.state
|
|
self.adventure.script_state = state
|
|
self.db.commit()
|
|
text = result.text
|
|
if result.stop:
|
|
return text, True
|
|
return text, False
|
|
|
|
def report(self) -> dict:
|
|
return {"logs": self.logs, "errors": self.errors, "message": self.message}
|