SaraView/Backend/Portals.py

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"|\{(?P<PipePath>[^\s{}']+)\s+(?P<PipeMode>head|tail)\s+(?P<PipeCount>\d+)\}"
    r"|\{(?P<LinePath>[^\s{}']+)\s+'(?P<LineNumber>\d+)'\}"
    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+%\}"
)


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 _ResolveLoops(Content: str, Lists: dict[str, list[str]]) -> str:
    Matches = list(LoopMarkerPattern.finditer(Content))
    Output: list[str] = []
    Cursor = 0
    Index = 0

    while Index < len(Matches):
        Match = Matches[Index]

        if not Match.group("LoopSource"):
            Index += 1
            continue

        Source = Match.group("LoopSource")
        LoopVar = Match.group("LoopVar")

        CloseMatch = None
        for Candidate in Matches[Index + 1 :]:
            if Candidate.group("CloseSource") == Source:
                CloseMatch = Candidate
                break

        BodyEnd = CloseMatch.start() if CloseMatch is not None else len(Content)
        BodyTemplate = Content[Match.end() : BodyEnd]
        Items = Lists.get(Source, [])
        VariablePattern = re.compile(rf"%{re.escape(LoopVar)}\b")

        Rendered = "\n".join(VariablePattern.sub(Item, BodyTemplate) for Item in Items)

        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 _ApplyPipe(Text: str, Mode: str, Count: int) -> str:
    Lines = Text.splitlines()
    Selected = Lines[:Count] if Mode == "head" else Lines[-Count:] if Count > 0 else []
    return "\n".join(Selected)


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:
    LoopLists = _ComputeLoopLists(RootName, RelativePath, Username, UserGroups)
    WithLoopsResolved = _ResolveLoops(Content, LoopLists)
    LocallyResolved = _ResolveLocalPortals(WithLoopsResolved)
    RefererDirectory = RelativePath.rsplit("/", 1)[0] if "/" in RelativePath else ""

    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"))
                )
        elif Match.group("PipeMode"):
            Replacement = _ResolveWholeFilePipe(
                RootName,
                RefererDirectory,
                Match.group("PipePath"),
                Match.group("PipeMode"),
                int(Match.group("PipeCount")),
                Username,
                UserGroups,
            )
        elif Match.group("LineNumber"):
            Replacement = _ResolveLineNumberReference(
                RootName,
                RefererDirectory,
                Match.group("LinePath"),
                int(Match.group("LineNumber")),
                Username,
                UserGroups,
            )

        if Replacement is not None:
            Output.append(LocallyResolved[Cursor : Match.start()])
            Output.append(Replacement)
            Cursor = Match.end()

    Output.append(LocallyResolved[Cursor:])
    return "".join(Output)