diff --git a/Backend/Main.py b/Backend/Main.py index 5321d52..479e03f 100644 --- a/Backend/Main.py +++ b/Backend/Main.py @@ -87,7 +87,7 @@ from Users import ( UserColourOptions, VerifyLogin, ) -from VarFunctions import ResolvePortals +from VarFunctions import ResolvePortals, ResolveSpans from pydantic import BaseModel @@ -614,6 +614,23 @@ def GetFileEndpoint( 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") def GetResolvedFileEndpoint( Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUserOrGuest) diff --git a/Backend/VarEngine.py b/Backend/VarEngine.py new file mode 100644 index 0000000..8d13644 --- /dev/null +++ b/Backend/VarEngine.py @@ -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[^\s{}']+)\s+)?\*\*\*(?P[^\s*{}]+)" + r"(?:\s+(?Phead|tail)\s+(?P\d+))?" + r"(?:\s*\\\s*(?P[^{}]*))?\}" + r"|\{(?P[^\s{}']+)\s+(?Phead|tail)\s+(?P\d+)" + r"(?:\s*\\\s*(?P[^{}]*))?\}" + r"|\{(?P[^\s{}']+)\s+'(?P\d+)'" + r"(?:\s*\\\s*(?P[^{}]*))?\}" + r"|\{\*\*(?P[^\s*{}]+)\}" + r"|\{\*(?P[^\s*{}]+)\}" +) + +LoopMarkerPattern = re.compile( + r"\{(?Pchildren|siblings|parents)\s+%\s+(?P[A-Za-z_][A-Za-z0-9_]*)\}" + r"|\{(?Pchildren|siblings|parents)\s+%\}" + r"|\{(?:(?P[^\s{}']+)\s+)?\*\*\*(?P[^\s*{}]+)" + r"(?:\s+(?Phead|tail)\s+(?P\d+))?" + r"\s+%\s+(?P[A-Za-z_][A-Za-z0-9_]*)\}" + r"|\{\*\*\*(?P[^\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[^{}]*)\}" + 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"]) diff --git a/Backend/VarFunctions.py b/Backend/VarFunctions.py index e9e7159..6411822 100644 --- a/Backend/VarFunctions.py +++ b/Backend/VarFunctions.py @@ -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 -import re - -from Core import NormalizeRelativePath, ResolvePath -from Explorer import ClassifyFile, ListDirectory, NoteFilePrefix +from Core import ResolvePath +from Explorer import ClassifyFile, ListDirectory from Locks import IsPathAccessible from Preview import ReadTextFile - - -PortalFileTypes = {"markdown", "text"} -LoopSources = ("children", "siblings", "parents") - -MarkerPattern = re.compile( - r"\{(?:(?P[^\s{}']+)\s+)?\*\*\*(?P[^\s*{}]+)" - r"(?:\s+(?Phead|tail)\s+(?P\d+))?" - r"(?:\s*\\\s*(?P[^{}]*))?\}" - r"|\{(?P[^\s{}']+)\s+(?Phead|tail)\s+(?P\d+)" - r"(?:\s*\\\s*(?P[^{}]*))?\}" - r"|\{(?P[^\s{}']+)\s+'(?P\d+)'" - r"(?:\s*\\\s*(?P[^{}]*))?\}" - r"|\{\*\*(?P[^\s*{}]+)\}" - r"|\{\*(?P[^\s*{}]+)\}" -) - -LoopMarkerPattern = re.compile( - r"\{(?Pchildren|siblings|parents)\s+%\s+(?P[A-Za-z_][A-Za-z0-9_]*)\}" - r"|\{(?Pchildren|siblings|parents)\s+%\}" - r"|\{(?:(?P[^\s{}']+)\s+)?\*\*\*(?P[^\s*{}]+)" - r"(?:\s+(?Phead|tail)\s+(?P\d+))?" - r"\s+%\s+(?P[A-Za-z_][A-Za-z0-9_]*)\}" - r"|\{\*\*\*(?P[^\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[^{}]*)\}" - 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) +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"] + + +class ServerAccess: + def __init__(self, RootName: str, Username: str, UserGroups: list[str]): + self.RootName = RootName + self.Username = Username + self.UserGroups = UserGroups + + def _Allowed(self, RelativePath: str) -> bool: + return IsPathAccessible(self.RootName, RelativePath, self.Username, self.UserGroups) + + def ReadText(self, RelativePath: str) -> str | None: + if not self._Allowed(RelativePath): + return None + try: + FilePath = ResolvePath(self.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 List(self, DirectoryPath: str) -> list[dict] | None: + if not self._Allowed(DirectoryPath): + return None + try: + Entries = ListDirectory(self.RootName, DirectoryPath, False, False) + except (ValueError, NotADirectoryError, FileNotFoundError): + return None + return [ + Entry for Entry in Entries + if self._Allowed(f"{DirectoryPath}/{Entry['Name']}" if DirectoryPath else Entry["Name"]) + ] + + def IsDirectory(self, RelativePath: str) -> bool: + try: + return ResolvePath(self.RootName, RelativePath).is_dir() + except ValueError: + return False + + +def ResolvePortals(RootName: str, RelativePath: str, Content: str, Username: str, UserGroups: list[str]) -> str: + return _ResolvePortals(RelativePath, Content, ServerAccess(RootName, Username, UserGroups)) + + +def ResolveSpans(RootName: str, RelativePath: str, Content: str, Username: str, UserGroups: list[str]) -> list[dict]: + return _ResolveSpans(RelativePath, Content, ServerAccess(RootName, Username, UserGroups))