mirror of
https://git.srhtnin.garden/sarah/SaraView.git
synced 2026-10-14 10:38:11 +02:00
471 lines
15 KiB
Python
471 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
from Core import NormalizeRelativePath, ResolvePath
|
|
from Explorer import ClassifyFile, ListDirectory, NoteFilePrefix
|
|
from Locks import IsPathAccessible
|
|
from Preview import ReadTextFile
|
|
|
|
|
|
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 _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
|
|
|
|
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:
|
|
FilePath = ResolvePath(RootName, RelativePath)
|
|
except ValueError:
|
|
return None
|
|
|
|
if not FilePath.is_file() or ClassifyFile(FilePath) not in PortalFileTypes:
|
|
return None
|
|
|
|
try:
|
|
return ReadTextFile(FilePath)
|
|
except (OSError, UnicodeDecodeError):
|
|
return None
|
|
|
|
|
|
def _FindNamedPortalInDirectory(
|
|
RootName: str, DirectoryRelativePath: str, Name: str, Username: str, UserGroups: list[str]
|
|
) -> str | None:
|
|
if not IsPathAccessible(RootName, DirectoryRelativePath, Username, UserGroups):
|
|
return None
|
|
|
|
try:
|
|
Entries = ListDirectory(RootName, DirectoryRelativePath, False, False)
|
|
except (ValueError, NotADirectoryError, FileNotFoundError):
|
|
return None
|
|
|
|
for Entry in Entries:
|
|
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:
|
|
TargetPath = ResolvePath(RootName, TargetRelativePath)
|
|
except ValueError:
|
|
return None
|
|
|
|
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]:
|
|
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(
|
|
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)
|