Variables: one engine for server and client, and spans for live editors

Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
sarahandClaude Opus 5.5 committed 2026-10-13 17:30:41 +02:00
1 parent c7a7596e99
commit 6d15fb1c53
3 files changed
+535 -441

No files matched your search

+18 -1
View File
@@ -87,7 +87,7 @@ from Users import (
UserColourOptions, UserColourOptions,
VerifyLogin, VerifyLogin,
) )
from VarFunctions import ResolvePortals from VarFunctions import ResolvePortals, ResolveSpans
from pydantic import BaseModel from pydantic import BaseModel
@@ -614,6 +614,23 @@ def GetFileEndpoint(
return StreamBinaryFile(FilePath, request.headers.get("range")) return StreamBinaryFile(FilePath, request.headers.get("range"))
class SpansRequest(BaseModel):
Content: str
# Where each variable is in the text being edited, and what it resolves to
# (VarEngine.ResolveSpans): the live editors show the results in place.
@Application.post("/api/file/spans")
def PostFileSpansEndpoint(
Root: str, Path: str, Body: SpansRequest, CurrentUser: dict = Depends(GetCurrentUserOrGuest)
) -> dict:
RequirePathAccessible(Root, Path, CurrentUser)
if len(Body.Content) > 2_000_000:
raise HTTPException(status_code=413, detail="Too long to resolve.")
Spans = ResolveSpans(Root, Path, Body.Content, CurrentUser["Username"], CurrentUser.get("Groups", []))
return {"Spans": Spans}
@Application.get("/api/file/resolved") @Application.get("/api/file/resolved")
def GetResolvedFileEndpoint( def GetResolvedFileEndpoint(
Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUserOrGuest) Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUserOrGuest)
+482
View File
@@ -0,0 +1,482 @@
"""SaraView's variables (portals, definitions, loops), resolved: the one
engine both the server (VarFunctions.py) and the desktop client (local
folders) use, so a note reads the same everywhere.
It never touches files itself: an Access object reads them, so the server's
can check locks and the client's can read a folder on disk. Paths are
relative to the root, like the server's.
ResolvePortals gives the note with everything filled in (the rendered view);
ResolveSpans gives where each variable is in the note and what it becomes,
for the live editors to show results in place of the code."""
from __future__ import annotations
import re
from typing import Optional, Protocol
NoteFilePrefix = ".sv_"
class Access(Protocol):
def ReadText(self, RelativePath: str) -> Optional[str]:
"""A Markdown or text file's text, or None (missing, another type, or not allowed)."""
def List(self, DirectoryPath: str) -> Optional[list[dict]]:
"""The entries ({"Name", "IsDirectory", "FileType"}) you're allowed to
see in a directory, or None if it isn't one or you can't."""
def IsDirectory(self, RelativePath: str) -> bool: ...
def NormalizeRelativePath(BaseDirectory: str, RawPath: str) -> str:
Combined = f"{BaseDirectory}/{RawPath}" if BaseDirectory else RawPath
Parts: list[str] = []
for Part in Combined.split("/"):
if Part in ("", "."):
continue
if Part == "..":
if Parts:
Parts.pop()
continue
Parts.append(Part)
return "/".join(Parts)
PortalFileTypes = {"markdown", "text"}
LoopSources = ("children", "siblings", "parents")
MarkerPattern = re.compile(
r"\{(?:(?P<CrossPath>[^\s{}']+)\s+)?\*\*\*(?P<CrossName>[^\s*{}]+)"
r"(?:\s+(?P<CrossPipe>head|tail)\s+(?P<CrossPipeCount>\d+))?"
r"(?:\s*\\\s*(?P<CrossClip>[^{}]*))?\}"
r"|\{(?P<PipePath>[^\s{}']+)\s+(?P<PipeMode>head|tail)\s+(?P<PipeCount>\d+)"
r"(?:\s*\\\s*(?P<PipeClip>[^{}]*))?\}"
r"|\{(?P<LinePath>[^\s{}']+)\s+'(?P<LineNumber>\d+)'"
r"(?:\s*\\\s*(?P<LineClip>[^{}]*))?\}"
r"|\{\*\*(?P<LineName>[^\s*{}]+)\}"
r"|\{\*(?P<BlockName>[^\s*{}]+)\}"
)
LoopMarkerPattern = re.compile(
r"\{(?P<LoopSource>children|siblings|parents)\s+%\s+(?P<LoopVar>[A-Za-z_][A-Za-z0-9_]*)\}"
r"|\{(?P<CloseSource>children|siblings|parents)\s+%\}"
r"|\{(?:(?P<LoopCrossPath>[^\s{}']+)\s+)?\*\*\*(?P<LoopCrossName>[^\s*{}]+)"
r"(?:\s+(?P<LoopCrossPipe>head|tail)\s+(?P<LoopCrossPipeCount>\d+))?"
r"\s+%\s+(?P<LoopCrossVar>[A-Za-z_][A-Za-z0-9_]*)\}"
r"|\{\*\*\*(?P<CloseCrossName>[^\s*{}]+)\s+%\}"
)
def _CompileClipPattern(Pattern: str) -> re.Pattern:
RegexParts = []
LastIndex = len(Pattern) - 1
for Index, Char in enumerate(Pattern):
IsLast = Index == LastIndex
if Char == "!":
RegexParts.append("(.*)" if IsLast else "(.*?)")
elif Char == "?":
RegexParts.append(".*" if IsLast else ".*?")
else:
RegexParts.append(re.escape(Char))
if not Pattern or Pattern[-1] not in "!?":
RegexParts.append(".*")
return re.compile("".join(RegexParts), re.DOTALL)
def _ApplyClip(Value: str, Pattern: str) -> str:
Pattern = Pattern.strip()
if not Pattern:
return Value
Match = _CompileClipPattern(Pattern).match(Value)
if Match is None:
return Value
return "".join(Match.groups())
def _LineBounds(Content: str, Position: int) -> tuple[int, int]:
LineStart = Content.rfind("\n", 0, Position) + 1
LineEnd = Content.find("\n", Position)
if LineEnd == -1:
LineEnd = len(Content)
return LineStart, LineEnd
def _FindLocalDefinitions(Content: str) -> dict[str, str]:
Matches = list(MarkerPattern.finditer(Content))
LineDefinitions: dict[str, str] = {}
BlockOccurrences: dict[str, list[re.Match]] = {}
for Match in Matches:
if Match.group("LineName"):
Name = Match.group("LineName")
if Name not in LineDefinitions:
LineStart, LineEnd = _LineBounds(Content, Match.start())
Line = Content[LineStart:LineEnd]
Captured = (
Line[: Match.start() - LineStart] + Line[Match.end() - LineStart :]
).strip()
LineDefinitions[Name] = _ResolveLocalPortals(Captured)
elif Match.group("BlockName"):
Name = Match.group("BlockName")
BlockOccurrences.setdefault(Name, []).append(Match)
BlockDefinitions: dict[str, str] = {}
for Name, Occurrences in BlockOccurrences.items():
Open = Occurrences[0]
if len(Occurrences) >= 2:
Close = Occurrences[1]
Captured = Content[Open.end() : Close.start()].strip("\n")
else:
Captured = Content[Open.end() :].lstrip("\n")
BlockDefinitions[Name] = _ResolveLocalPortals(Captured)
return {**LineDefinitions, **BlockDefinitions}
def _ResolveLocalPortals(Content: str) -> str:
Matches = list(MarkerPattern.finditer(Content))
Definitions = _FindLocalDefinitions(Content)
LineSeenCount: dict[str, int] = {}
BlockSeenCount: dict[str, int] = {}
Output: list[str] = []
Cursor = 0
for Match in Matches:
Replacement = None
if Match.group("LineName"):
Name = Match.group("LineName")
Occurrence = LineSeenCount.get(Name, 0)
LineSeenCount[Name] = Occurrence + 1
Replacement = "" if Occurrence == 0 else Definitions.get(Name, "")
elif Match.group("BlockName"):
Name = Match.group("BlockName")
Occurrence = BlockSeenCount.get(Name, 0)
BlockSeenCount[Name] = Occurrence + 1
Replacement = "" if Occurrence in (0, 1) else Definitions.get(Name, "")
if Replacement is not None:
Output.append(Content[Cursor : Match.start()])
Output.append(Replacement)
Cursor = Match.end()
Output.append(Content[Cursor:])
return "".join(Output)
def _ListEntryNames(Files: Access, DirectoryPath: str, Exclude: str | None) -> list[str]:
Entries = Files.List(DirectoryPath)
return [Entry["Name"] for Entry in Entries or [] if Entry["Name"] != Exclude]
def _ComputeLoopLists(Files: Access, RelativePath: str) -> dict[str, list[str]]:
FileName = RelativePath.rsplit("/", 1)[-1] if RelativePath else ""
ParentDirectory = RelativePath.rsplit("/", 1)[0] if "/" in RelativePath else ""
ParentBaseName = ParentDirectory.rsplit("/", 1)[-1] if ParentDirectory else ""
IsDirectoryNote = FileName.startswith(f"{NoteFilePrefix}{ParentBaseName}.")
Children = _ListEntryNames(Files, ParentDirectory, FileName) if IsDirectoryNote else []
Siblings = _ListEntryNames(Files, ParentDirectory, FileName)
Parents = [Segment for Segment in ParentDirectory.split("/") if Segment]
return {"children": Children, "siblings": Siblings, "parents": Parents}
def _FindLoopClose(Matches: list[re.Match], StartIndex: int, IsFixed: bool, Key: str) -> re.Match | None:
GroupName = "CloseSource" if IsFixed else "CloseCrossName"
for Candidate in Matches[StartIndex:]:
if Candidate.group(GroupName) == Key:
return Candidate
return None
def _ResolveLoops(
Content: str,
Lists: dict[str, list[str]],
Files: Access,
RefererDirectory: str,
) -> str:
Matches = list(LoopMarkerPattern.finditer(Content))
Output: list[str] = []
Cursor = 0
Index = 0
while Index < len(Matches):
Match = Matches[Index]
IsFixed = bool(Match.group("LoopSource"))
IsCross = bool(Match.group("LoopCrossName"))
if not IsFixed and not IsCross:
Index += 1
continue
if IsFixed:
Key = Match.group("LoopSource")
LoopVar = Match.group("LoopVar")
Items = Lists.get(Key, [])
else:
Key = Match.group("LoopCrossName")
LoopVar = Match.group("LoopCrossVar")
PortalContent = _ResolveCrossFileReference(
Files, RefererDirectory, Match.group("LoopCrossPath"), Key
)
if PortalContent is None:
Items = []
elif Match.group("LoopCrossPipe"):
Items = _SlicePipeLines(
PortalContent, Match.group("LoopCrossPipe"), int(Match.group("LoopCrossPipeCount"))
)
else:
Items = PortalContent.splitlines()
CloseMatch = _FindLoopClose(Matches, Index + 1, IsFixed, Key)
BodyEnd = CloseMatch.start() if CloseMatch is not None else len(Content)
BodyTemplate = Content[Match.end() : BodyEnd]
EscapedVar = re.escape(LoopVar)
VariablePattern = re.compile(
r"\{%" + EscapedVar + r"\s*\\\s*(?P<Clip>[^{}]*)\}"
r"|%" + EscapedVar + r"\b"
)
RenderedItems = []
for Item in Items:
ItemRendered = VariablePattern.sub(
lambda VariableMatch, CurrentItem=Item: (
_ApplyClip(CurrentItem, VariableMatch.group("Clip"))
if VariableMatch.group("Clip") is not None
else CurrentItem
),
BodyTemplate,
)
RenderedItems.append(ItemRendered)
Rendered = "\n".join(RenderedItems)
Output.append(Content[Cursor : Match.start()])
Output.append(Rendered)
Cursor = CloseMatch.end() if CloseMatch is not None else len(Content)
if CloseMatch is not None:
Index = Matches.index(CloseMatch) + 1
else:
break
Output.append(Content[Cursor:])
return "".join(Output)
def _FindNamedPortalInDirectory(Files: Access, DirectoryPath: str, Name: str) -> str | None:
for Entry in Files.List(DirectoryPath) or []:
if Entry["IsDirectory"] or Entry["FileType"] not in PortalFileTypes:
continue
EntryPath = f"{DirectoryPath}/{Entry['Name']}" if DirectoryPath else Entry["Name"]
Content = Files.ReadText(EntryPath)
if Content is None:
continue
Definitions = _FindLocalDefinitions(Content)
if Name in Definitions:
return Definitions[Name]
return None
def _ResolveCrossFileReference(
Files: Access, RefererDirectory: str, PathText: str | None, Name: str
) -> str | None:
if not PathText:
return _FindNamedPortalInDirectory(Files, RefererDirectory, Name)
TargetRelativePath = NormalizeRelativePath(RefererDirectory, PathText)
if Files.IsDirectory(TargetRelativePath):
return _FindNamedPortalInDirectory(Files, TargetRelativePath, Name)
Content = Files.ReadText(TargetRelativePath)
if Content is None:
return None
return _FindLocalDefinitions(Content).get(Name)
def _SlicePipeLines(Text: str, Mode: str, Count: int) -> list[str]:
Lines = Text.splitlines()
return Lines[:Count] if Mode == "head" else (Lines[-Count:] if Count > 0 else [])
def _ApplyPipe(Text: str, Mode: str, Count: int) -> str:
return "\n".join(_SlicePipeLines(Text, Mode, Count))
def _ResolveWholeFilePipe(Files: Access, RefererDirectory: str, PathText: str, Mode: str, Count: int) -> str | None:
Content = Files.ReadText(NormalizeRelativePath(RefererDirectory, PathText))
if Content is None:
return None
return _ApplyPipe(_ResolveLocalPortals(Content), Mode, Count)
def _ResolveLineNumberReference(Files: Access, RefererDirectory: str, PathText: str, LineNumber: int) -> str | None:
Content = Files.ReadText(NormalizeRelativePath(RefererDirectory, PathText))
if Content is None:
return None
Lines = Content.splitlines()
Index = LineNumber - 1
if 0 <= Index < len(Lines):
return Lines[Index]
return None
def ResolvePortals(RelativePath: str, Content: str, Files: Access) -> str:
"""The note with its variables filled in."""
RefererDirectory = RelativePath.rsplit("/", 1)[0] if "/" in RelativePath else ""
LoopLists = _ComputeLoopLists(Files, RelativePath)
WithLoopsResolved = _ResolveLoops(Content, LoopLists, Files, RefererDirectory)
LocallyResolved = _ResolveLocalPortals(WithLoopsResolved)
Matches = list(MarkerPattern.finditer(LocallyResolved))
Output: list[str] = []
Cursor = 0
for Match in Matches:
Replacement = None
if Match.group("CrossName"):
Replacement = _ResolveCrossFileReference(
Files, RefererDirectory, Match.group("CrossPath"), Match.group("CrossName")
)
if Replacement is not None and Match.group("CrossPipe"):
Replacement = _ApplyPipe(
Replacement, Match.group("CrossPipe"), int(Match.group("CrossPipeCount"))
)
if Replacement is not None and Match.group("CrossClip"):
Replacement = _ApplyClip(Replacement, Match.group("CrossClip"))
elif Match.group("PipeMode"):
Replacement = _ResolveWholeFilePipe(
Files, RefererDirectory, Match.group("PipePath"), Match.group("PipeMode"), int(Match.group("PipeCount"))
)
if Replacement is not None and Match.group("PipeClip"):
Replacement = _ApplyClip(Replacement, Match.group("PipeClip"))
elif Match.group("LineNumber"):
Replacement = _ResolveLineNumberReference(
Files, RefererDirectory, Match.group("LinePath"), int(Match.group("LineNumber"))
)
if Replacement is not None and Match.group("LineClip"):
Replacement = _ApplyClip(Replacement, Match.group("LineClip"))
if Replacement is not None:
Output.append(LocallyResolved[Cursor : Match.start()])
Output.append(Replacement)
Cursor = Match.end()
Output.append(LocallyResolved[Cursor:])
return "".join(Output)
# --- Where each variable is, and what it becomes (the live editors) --------
_Token = re.compile("\ue000(\\d+)\ue001|\ue002(\\d+)\ue003")
def _TopLevelVariables(Content: str) -> tuple[list[tuple[int, int]], list[tuple[int, int]]]:
"""(Variables, Definitions). Variables: each loop as a whole (its opening
to its closing marker), and every other variable outside the loops.
Definitions: the markers that only define ({**name} the first time,
{*name} the first two), which always resolve to nothing."""
Found: list[tuple[int, int]] = []
Matches = list(LoopMarkerPattern.finditer(Content))
Index = 0
while Index < len(Matches):
Match = Matches[Index]
IsFixed = bool(Match.group("LoopSource"))
if not IsFixed and not Match.group("LoopCrossName"):
Index += 1
continue
Key = Match.group("LoopSource") if IsFixed else Match.group("LoopCrossName")
Close = _FindLoopClose(Matches, Index + 1, IsFixed, Key)
Found.append((Match.start(), Close.end() if Close is not None else len(Content)))
if Close is None:
break
Index = Matches.index(Close) + 1
Definitions: list[tuple[int, int]] = []
Seen: dict[tuple[str, str], int] = {}
for Match in MarkerPattern.finditer(Content):
if any(Start <= Match.start() < End for Start, End in Found):
continue
Kind = "Line" if Match.group("LineName") else "Block" if Match.group("BlockName") else None
if Kind is not None:
Key = (Kind, Match.group(f"{Kind}Name"))
Seen[Key] = Seen.get(Key, 0) + 1
if Seen[Key] <= (1 if Kind == "Line" else 2):
Definitions.append((Match.start(), Match.end()))
continue
Found.append((Match.start(), Match.end()))
return sorted(Found), Definitions
def ResolveSpans(RelativePath: str, Content: str, Files: Access) -> list[dict]:
"""[{"From", "To", "Text"}]: each variable's place in Content (character
offsets) and what it resolves to ("" for a definition's own marker).
Ones that stay as written (a portal to a missing file) are left out.
Each variable is wrapped in private-use marker characters, the note is
resolved as usual, and the markers say which result is whose. Results
copied into others (a definition used later) carry their markers along,
nested, and are skipped."""
Variables, Definitions = _TopLevelVariables(Content)
# Definitions aren't marked: markers beside them would stop the resolver
# trimming the lines around them.
Spans = [{"From": Start, "To": End, "Text": ""} for Start, End in Definitions]
if not Variables:
return Spans
Parts: list[str] = []
Cursor = 0
for Number, (Start, End) in enumerate(Variables):
Parts += [Content[Cursor:Start], f"\ue000{Number}\ue001", Content[Start:End], f"\ue002{Number}\ue003"]
Cursor = End
Parts.append(Content[Cursor:])
Resolved = ResolvePortals(RelativePath, "".join(Parts), Files)
Results: dict[int, str] = {}
Open: list[tuple[int, int]] = [] # (number, where its text starts)
for Token in _Token.finditer(Resolved):
if Token.group(1) is not None:
Open.append((int(Token.group(1)), Token.end()))
continue
Number = int(Token.group(2))
if all(Number != Waiting for Waiting, _ in Open):
continue # a stray close, copied in with a definition's text
while Open[-1][0] != Number:
Open.pop() # stray opens, likewise
_, TextStart = Open.pop()
if not Open and Number not in Results:
Results[Number] = _Token.sub("", Resolved[TextStart:Token.start()])
Spans += [
{"From": Start, "To": End, "Text": Results[Number]}
for Number, (Start, End) in enumerate(Variables)
if Number in Results and Results[Number] != Content[Start:End]
]
return sorted(Spans, key=lambda Span: Span["From"])
+35 -440
View File
@@ -1,470 +1,65 @@
"""The server's side of SaraView's variables: VarEngine.py (shared with the
desktop client) with file access that follows the configured roots and the
locks, so a portal can't show what you couldn't open yourself."""
from __future__ import annotations from __future__ import annotations
import re from Core import ResolvePath
from Explorer import ClassifyFile, ListDirectory
from Core import NormalizeRelativePath, ResolvePath
from Explorer import ClassifyFile, ListDirectory, NoteFilePrefix
from Locks import IsPathAccessible from Locks import IsPathAccessible
from Preview import ReadTextFile from Preview import ReadTextFile
from VarEngine import MarkerPattern, PortalFileTypes # MarkerPattern: Graph.py finds links with it
from VarEngine import ResolvePortals as _ResolvePortals
from VarEngine import ResolveSpans as _ResolveSpans
__all__ = ["MarkerPattern", "ResolvePortals", "ResolveSpans"]
PortalFileTypes = {"markdown", "text"} class ServerAccess:
LoopSources = ("children", "siblings", "parents") def __init__(self, RootName: str, Username: str, UserGroups: list[str]):
self.RootName = RootName
self.Username = Username
self.UserGroups = UserGroups
MarkerPattern = re.compile( def _Allowed(self, RelativePath: str) -> bool:
r"\{(?:(?P<CrossPath>[^\s{}']+)\s+)?\*\*\*(?P<CrossName>[^\s*{}]+)" return IsPathAccessible(self.RootName, RelativePath, self.Username, self.UserGroups)
r"(?:\s+(?P<CrossPipe>head|tail)\s+(?P<CrossPipeCount>\d+))?"
r"(?:\s*\\\s*(?P<CrossClip>[^{}]*))?\}"
r"|\{(?P<PipePath>[^\s{}']+)\s+(?P<PipeMode>head|tail)\s+(?P<PipeCount>\d+)"
r"(?:\s*\\\s*(?P<PipeClip>[^{}]*))?\}"
r"|\{(?P<LinePath>[^\s{}']+)\s+'(?P<LineNumber>\d+)'"
r"(?:\s*\\\s*(?P<LineClip>[^{}]*))?\}"
r"|\{\*\*(?P<LineName>[^\s*{}]+)\}"
r"|\{\*(?P<BlockName>[^\s*{}]+)\}"
)
LoopMarkerPattern = re.compile(
r"\{(?P<LoopSource>children|siblings|parents)\s+%\s+(?P<LoopVar>[A-Za-z_][A-Za-z0-9_]*)\}"
r"|\{(?P<CloseSource>children|siblings|parents)\s+%\}"
r"|\{(?:(?P<LoopCrossPath>[^\s{}']+)\s+)?\*\*\*(?P<LoopCrossName>[^\s*{}]+)"
r"(?:\s+(?P<LoopCrossPipe>head|tail)\s+(?P<LoopCrossPipeCount>\d+))?"
r"\s+%\s+(?P<LoopCrossVar>[A-Za-z_][A-Za-z0-9_]*)\}"
r"|\{\*\*\*(?P<CloseCrossName>[^\s*{}]+)\s+%\}"
)
def _CompileClipPattern(Pattern: str) -> re.Pattern:
RegexParts = []
LastIndex = len(Pattern) - 1
for Index, Char in enumerate(Pattern):
IsLast = Index == LastIndex
if Char == "!":
RegexParts.append("(.*)" if IsLast else "(.*?)")
elif Char == "?":
RegexParts.append(".*" if IsLast else ".*?")
else:
RegexParts.append(re.escape(Char))
if not Pattern or Pattern[-1] not in "!?":
RegexParts.append(".*")
return re.compile("".join(RegexParts), re.DOTALL)
def _ApplyClip(Value: str, Pattern: str) -> str:
Pattern = Pattern.strip()
if not Pattern:
return Value
Match = _CompileClipPattern(Pattern).match(Value)
if Match is None:
return Value
return "".join(Match.groups())
def _LineBounds(Content: str, Position: int) -> tuple[int, int]:
LineStart = Content.rfind("\n", 0, Position) + 1
LineEnd = Content.find("\n", Position)
if LineEnd == -1:
LineEnd = len(Content)
return LineStart, LineEnd
def _FindLocalDefinitions(Content: str) -> dict[str, str]:
Matches = list(MarkerPattern.finditer(Content))
LineDefinitions: dict[str, str] = {}
BlockOccurrences: dict[str, list[re.Match]] = {}
for Match in Matches:
if Match.group("LineName"):
Name = Match.group("LineName")
if Name not in LineDefinitions:
LineStart, LineEnd = _LineBounds(Content, Match.start())
Line = Content[LineStart:LineEnd]
Captured = (
Line[: Match.start() - LineStart] + Line[Match.end() - LineStart :]
).strip()
LineDefinitions[Name] = _ResolveLocalPortals(Captured)
elif Match.group("BlockName"):
Name = Match.group("BlockName")
BlockOccurrences.setdefault(Name, []).append(Match)
BlockDefinitions: dict[str, str] = {}
for Name, Occurrences in BlockOccurrences.items():
Open = Occurrences[0]
if len(Occurrences) >= 2:
Close = Occurrences[1]
Captured = Content[Open.end() : Close.start()].strip("\n")
else:
Captured = Content[Open.end() :].lstrip("\n")
BlockDefinitions[Name] = _ResolveLocalPortals(Captured)
return {**LineDefinitions, **BlockDefinitions}
def _ResolveLocalPortals(Content: str) -> str:
Matches = list(MarkerPattern.finditer(Content))
Definitions = _FindLocalDefinitions(Content)
LineSeenCount: dict[str, int] = {}
BlockSeenCount: dict[str, int] = {}
Output: list[str] = []
Cursor = 0
for Match in Matches:
Replacement = None
if Match.group("LineName"):
Name = Match.group("LineName")
Occurrence = LineSeenCount.get(Name, 0)
LineSeenCount[Name] = Occurrence + 1
Replacement = "" if Occurrence == 0 else Definitions.get(Name, "")
elif Match.group("BlockName"):
Name = Match.group("BlockName")
Occurrence = BlockSeenCount.get(Name, 0)
BlockSeenCount[Name] = Occurrence + 1
Replacement = "" if Occurrence in (0, 1) else Definitions.get(Name, "")
if Replacement is not None:
Output.append(Content[Cursor : Match.start()])
Output.append(Replacement)
Cursor = Match.end()
Output.append(Content[Cursor:])
return "".join(Output)
def _ListAccessibleEntryNames(
RootName: str, DirectoryRelativePath: str, Username: str, UserGroups: list[str], Exclude: str | None
) -> list[str]:
if not IsPathAccessible(RootName, DirectoryRelativePath, Username, UserGroups):
return []
try:
Entries = ListDirectory(RootName, DirectoryRelativePath, False, False)
except (ValueError, NotADirectoryError, FileNotFoundError):
return []
Names: list[str] = []
for Entry in Entries:
if Entry["Name"] == Exclude:
continue
EntryPath = (
f"{DirectoryRelativePath}/{Entry['Name']}" if DirectoryRelativePath else Entry["Name"]
)
if IsPathAccessible(RootName, EntryPath, Username, UserGroups):
Names.append(Entry["Name"])
return Names
def _ComputeLoopLists(
RootName: str, RelativePath: str, Username: str, UserGroups: list[str]
) -> dict[str, list[str]]:
FileName = RelativePath.rsplit("/", 1)[-1] if RelativePath else ""
ParentDirectory = RelativePath.rsplit("/", 1)[0] if "/" in RelativePath else ""
ParentBaseName = ParentDirectory.rsplit("/", 1)[-1] if ParentDirectory else ""
IsDirectoryNote = FileName.startswith(f"{NoteFilePrefix}{ParentBaseName}.")
Children = (
_ListAccessibleEntryNames(RootName, ParentDirectory, Username, UserGroups, FileName)
if IsDirectoryNote
else []
)
Siblings = _ListAccessibleEntryNames(RootName, ParentDirectory, Username, UserGroups, FileName)
Parents = [Segment for Segment in ParentDirectory.split("/") if Segment]
return {"children": Children, "siblings": Siblings, "parents": Parents}
def _FindLoopClose(Matches: list[re.Match], StartIndex: int, IsFixed: bool, Key: str) -> re.Match | None:
GroupName = "CloseSource" if IsFixed else "CloseCrossName"
for Candidate in Matches[StartIndex:]:
if Candidate.group(GroupName) == Key:
return Candidate
def ReadText(self, RelativePath: str) -> str | None:
if not self._Allowed(RelativePath):
return None return None
def _ResolveLoops(
Content: str,
Lists: dict[str, list[str]],
RootName: str,
RefererDirectory: str,
Username: str,
UserGroups: list[str],
) -> str:
Matches = list(LoopMarkerPattern.finditer(Content))
Output: list[str] = []
Cursor = 0
Index = 0
while Index < len(Matches):
Match = Matches[Index]
IsFixed = bool(Match.group("LoopSource"))
IsCross = bool(Match.group("LoopCrossName"))
if not IsFixed and not IsCross:
Index += 1
continue
if IsFixed:
Key = Match.group("LoopSource")
LoopVar = Match.group("LoopVar")
Items = Lists.get(Key, [])
else:
Key = Match.group("LoopCrossName")
LoopVar = Match.group("LoopCrossVar")
PortalContent = _ResolveCrossFileReference(
RootName, RefererDirectory, Match.group("LoopCrossPath"), Key, Username, UserGroups
)
if PortalContent is None:
Items = []
elif Match.group("LoopCrossPipe"):
Items = _SlicePipeLines(
PortalContent, Match.group("LoopCrossPipe"), int(Match.group("LoopCrossPipeCount"))
)
else:
Items = PortalContent.splitlines()
CloseMatch = _FindLoopClose(Matches, Index + 1, IsFixed, Key)
BodyEnd = CloseMatch.start() if CloseMatch is not None else len(Content)
BodyTemplate = Content[Match.end() : BodyEnd]
EscapedVar = re.escape(LoopVar)
VariablePattern = re.compile(
r"\{%" + EscapedVar + r"\s*\\\s*(?P<Clip>[^{}]*)\}"
r"|%" + EscapedVar + r"\b"
)
RenderedItems = []
for Item in Items:
ItemRendered = VariablePattern.sub(
lambda VariableMatch, CurrentItem=Item: (
_ApplyClip(CurrentItem, VariableMatch.group("Clip"))
if VariableMatch.group("Clip") is not None
else CurrentItem
),
BodyTemplate,
)
RenderedItems.append(ItemRendered)
Rendered = "\n".join(RenderedItems)
Output.append(Content[Cursor : Match.start()])
Output.append(Rendered)
Cursor = CloseMatch.end() if CloseMatch is not None else len(Content)
if CloseMatch is not None:
Index = Matches.index(CloseMatch) + 1
else:
break
Output.append(Content[Cursor:])
return "".join(Output)
def _ReadAccessiblePortalSource(
RootName: str, RelativePath: str, Username: str, UserGroups: list[str]
) -> str | None:
if not IsPathAccessible(RootName, RelativePath, Username, UserGroups):
return None
try: try:
FilePath = ResolvePath(RootName, RelativePath) FilePath = ResolvePath(self.RootName, RelativePath)
except ValueError: except ValueError:
return None return None
if not FilePath.is_file() or ClassifyFile(FilePath) not in PortalFileTypes: if not FilePath.is_file() or ClassifyFile(FilePath) not in PortalFileTypes:
return None return None
try: try:
return ReadTextFile(FilePath) return ReadTextFile(FilePath)
except (OSError, UnicodeDecodeError): except (OSError, UnicodeDecodeError):
return None return None
def List(self, DirectoryPath: str) -> list[dict] | None:
def _FindNamedPortalInDirectory( if not self._Allowed(DirectoryPath):
RootName: str, DirectoryRelativePath: str, Name: str, Username: str, UserGroups: list[str]
) -> str | None:
if not IsPathAccessible(RootName, DirectoryRelativePath, Username, UserGroups):
return None return None
try: try:
Entries = ListDirectory(RootName, DirectoryRelativePath, False, False) Entries = ListDirectory(self.RootName, DirectoryPath, False, False)
except (ValueError, NotADirectoryError, FileNotFoundError): except (ValueError, NotADirectoryError, FileNotFoundError):
return None return None
return [
Entry for Entry in Entries
if self._Allowed(f"{DirectoryPath}/{Entry['Name']}" if DirectoryPath else Entry["Name"])
]
for Entry in Entries: def IsDirectory(self, RelativePath: str) -> bool:
if Entry["IsDirectory"] or Entry["FileType"] not in PortalFileTypes:
continue
EntryPath = (
f"{DirectoryRelativePath}/{Entry['Name']}" if DirectoryRelativePath else Entry["Name"]
)
Content = _ReadAccessiblePortalSource(RootName, EntryPath, Username, UserGroups)
if Content is None:
continue
Definitions = _FindLocalDefinitions(Content)
if Name in Definitions:
return Definitions[Name]
return None
def _ResolveCrossFileReference(
RootName: str,
RefererDirectory: str,
PathText: str | None,
Name: str,
Username: str,
UserGroups: list[str],
) -> str | None:
if not PathText:
return _FindNamedPortalInDirectory(RootName, RefererDirectory, Name, Username, UserGroups)
TargetRelativePath = NormalizeRelativePath(RefererDirectory, PathText)
try: try:
TargetPath = ResolvePath(RootName, TargetRelativePath) return ResolvePath(self.RootName, RelativePath).is_dir()
except ValueError: except ValueError:
return None return False
if TargetPath.is_dir():
return _FindNamedPortalInDirectory(
RootName, TargetRelativePath, Name, Username, UserGroups
)
Content = _ReadAccessiblePortalSource(RootName, TargetRelativePath, Username, UserGroups)
if Content is None:
return None
return _FindLocalDefinitions(Content).get(Name)
def _SlicePipeLines(Text: str, Mode: str, Count: int) -> list[str]: def ResolvePortals(RootName: str, RelativePath: str, Content: str, Username: str, UserGroups: list[str]) -> str:
Lines = Text.splitlines() return _ResolvePortals(RelativePath, Content, ServerAccess(RootName, Username, UserGroups))
return Lines[:Count] if Mode == "head" else (Lines[-Count:] if Count > 0 else [])
def _ApplyPipe(Text: str, Mode: str, Count: int) -> str: def ResolveSpans(RootName: str, RelativePath: str, Content: str, Username: str, UserGroups: list[str]) -> list[dict]:
return "\n".join(_SlicePipeLines(Text, Mode, Count)) return _ResolveSpans(RelativePath, Content, ServerAccess(RootName, Username, UserGroups))
def _ResolveWholeFilePipe(
RootName: str,
RefererDirectory: str,
PathText: str,
Mode: str,
Count: int,
Username: str,
UserGroups: list[str],
) -> str | None:
TargetRelativePath = NormalizeRelativePath(RefererDirectory, PathText)
Content = _ReadAccessiblePortalSource(RootName, TargetRelativePath, Username, UserGroups)
if Content is None:
return None
return _ApplyPipe(_ResolveLocalPortals(Content), Mode, Count)
def _ResolveLineNumberReference(
RootName: str,
RefererDirectory: str,
PathText: str,
LineNumber: int,
Username: str,
UserGroups: list[str],
) -> str | None:
TargetRelativePath = NormalizeRelativePath(RefererDirectory, PathText)
Content = _ReadAccessiblePortalSource(RootName, TargetRelativePath, Username, UserGroups)
if Content is None:
return None
Lines = Content.splitlines()
Index = LineNumber - 1
if 0 <= Index < len(Lines):
return Lines[Index]
return None
def ResolvePortals(
RootName: str, RelativePath: str, Content: str, Username: str, UserGroups: list[str]
) -> str:
RefererDirectory = RelativePath.rsplit("/", 1)[0] if "/" in RelativePath else ""
LoopLists = _ComputeLoopLists(RootName, RelativePath, Username, UserGroups)
WithLoopsResolved = _ResolveLoops(
Content, LoopLists, RootName, RefererDirectory, Username, UserGroups
)
LocallyResolved = _ResolveLocalPortals(WithLoopsResolved)
Matches = list(MarkerPattern.finditer(LocallyResolved))
Output: list[str] = []
Cursor = 0
for Match in Matches:
Replacement = None
if Match.group("CrossName"):
Replacement = _ResolveCrossFileReference(
RootName,
RefererDirectory,
Match.group("CrossPath"),
Match.group("CrossName"),
Username,
UserGroups,
)
if Replacement is not None and Match.group("CrossPipe"):
Replacement = _ApplyPipe(
Replacement, Match.group("CrossPipe"), int(Match.group("CrossPipeCount"))
)
if Replacement is not None and Match.group("CrossClip"):
Replacement = _ApplyClip(Replacement, Match.group("CrossClip"))
elif Match.group("PipeMode"):
Replacement = _ResolveWholeFilePipe(
RootName,
RefererDirectory,
Match.group("PipePath"),
Match.group("PipeMode"),
int(Match.group("PipeCount")),
Username,
UserGroups,
)
if Replacement is not None and Match.group("PipeClip"):
Replacement = _ApplyClip(Replacement, Match.group("PipeClip"))
elif Match.group("LineNumber"):
Replacement = _ResolveLineNumberReference(
RootName,
RefererDirectory,
Match.group("LinePath"),
int(Match.group("LineNumber")),
Username,
UserGroups,
)
if Replacement is not None and Match.group("LineClip"):
Replacement = _ApplyClip(Replacement, Match.group("LineClip"))
if Replacement is not None:
Output.append(LocallyResolved[Cursor : Match.start()])
Output.append(Replacement)
Cursor = Match.end()
Output.append(LocallyResolved[Cursor:])
return "".join(Output)