mirror of
https://git.srhtnin.garden/sarah/SaraSort.git
synced 2026-10-13 18:28:18 +02:00
Flowchart of inputs/outputs with editable arrows, service page (status, controls, autostart, version, logs), config editor (form + raw), templates, tools (warnings, filename tester, failed files) and settings with SaraFem-compatible export/import. AppImage build and a SaraSuite manifest in Packaging/. Co-Authored-By: Claude Opus 5.5 <[email protected]>
674 lines
30 KiB
Python
674 lines
30 KiB
Python
"""SaraSortd's configs as SaraSort sees them: which directories are inputs and
|
|
outputs, which file patterns go from where to where, and the edits the
|
|
flowchart and templates make.
|
|
|
|
SaraSort never sorts anything itself. To show what SaraSortd will really do,
|
|
Parse, GetConf and Simulate follow SaraSortd.py's own Parse, GetConf and
|
|
DecideNewPath, quirks included (e.g. Parse also swaps variable characters
|
|
inside paths). All writes go through tomlkit, so comments and layout in the
|
|
files survive; SaraSortd itself rewrites directory configs with `toml`, which
|
|
drops comments, but GlobalConf.toml and the templates keep theirs.
|
|
"""
|
|
|
|
import datetime
|
|
import fnmatch
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
import tomlkit
|
|
from tomlkit.items import AoT
|
|
|
|
from Core import AppConfig
|
|
|
|
TemplateNames = {"In": "InputDirConf.toml", "Out": "OutputDirConf.toml"}
|
|
NameKeys = {"In": "InDirConfName", "Out": "OutDirConfName"}
|
|
# 0/1 switches in SaraSortd's configs, shown as switches in the editor.
|
|
SwitchKeys = {"SafeMode", "SilentMode", "DotFiles", "Logging", "OverwriteDirConf", "DeleteOrg", "SortDirs", "CaseSensitive", "Overwrite"}
|
|
|
|
|
|
class SortdError(Exception):
|
|
"""GlobalConf.toml is missing or unreadable."""
|
|
|
|
|
|
def ReadToml(FilePath: Path) -> tomlkit.TOMLDocument:
|
|
return tomlkit.parse(FilePath.read_text(encoding="utf-8"))
|
|
|
|
|
|
def WriteToml(FilePath: Path, Document: tomlkit.TOMLDocument) -> None:
|
|
FilePath.write_text(tomlkit.dumps(Document), encoding="utf-8")
|
|
|
|
|
|
def GetConf(Parameter: str, Data: dict) -> Any:
|
|
"""SaraSortd's GetConf: a top-level key, or one a level down inside a
|
|
table or array of tables."""
|
|
if Parameter in Data:
|
|
return Data[Parameter]
|
|
for Value in Data.values():
|
|
if isinstance(Value, dict):
|
|
if Parameter in Value:
|
|
return Value[Parameter]
|
|
elif isinstance(Value, list):
|
|
for Item in Value:
|
|
if isinstance(Item, dict) and Parameter in Item:
|
|
return Item[Parameter]
|
|
return None
|
|
|
|
|
|
def MajorVersion(Version: str) -> str:
|
|
return str(Version).split(".")[0]
|
|
|
|
|
|
def VersionAtLeast(Version: str, Wanted: tuple[int, int]) -> bool:
|
|
try:
|
|
Parts = tuple(int(Part) for Part in str(Version).split(".")[:2])
|
|
except ValueError:
|
|
return False
|
|
return Parts >= Wanted
|
|
|
|
|
|
@dataclass
|
|
class FileRule:
|
|
"""One [[Files]] entry of an output directory that applies to an arrow."""
|
|
|
|
Index: int # position in the output's Files array
|
|
Pattern: str
|
|
NewFileName: str
|
|
OwnInputs: bool # True: the entry has its own ValidInputDirs (7.1)
|
|
|
|
|
|
@dataclass
|
|
class Edge:
|
|
Input: str
|
|
Output: str
|
|
Rules: list[FileRule] = field(default_factory=list)
|
|
|
|
@property
|
|
def Patterns(self) -> list[str]:
|
|
return [Rule.Pattern for Rule in self.Rules]
|
|
|
|
|
|
@dataclass
|
|
class DirNode:
|
|
Path: str # resolved, as SaraSortd sees it
|
|
IsInput: bool = False
|
|
IsOutput: bool = False
|
|
|
|
@property
|
|
def Kind(self) -> str:
|
|
if self.IsInput and self.IsOutput:
|
|
return "Both"
|
|
return "Input" if self.IsInput else "Output"
|
|
|
|
|
|
@dataclass
|
|
class SimulationResult:
|
|
Steps: list[str]
|
|
Destination: Optional[str] # full new path, or None
|
|
Outcome: str # "Sorted", "Failed", "Skipped"
|
|
|
|
|
|
class Setup:
|
|
"""GlobalConf.toml plus everything reachable from it. Cheap to reload:
|
|
pages just make a new one after each change."""
|
|
|
|
def __init__(self, Config: AppConfig):
|
|
self.Config = Config
|
|
self.ConfDir = Path(Config.SortdConfigDir).expanduser()
|
|
self.GlobalPath = Config.GlobalConfPath
|
|
if not self.GlobalPath.exists():
|
|
raise SortdError(f"No GlobalConf.toml at {self.GlobalPath}")
|
|
try:
|
|
self.Global = ReadToml(self.GlobalPath)
|
|
except tomlkit.exceptions.TOMLKitError as Error:
|
|
raise SortdError(f"GlobalConf.toml can't be read: {Error}") from Error
|
|
self.Data = self.Global.unwrap()
|
|
self.Vars = self.Data.get("Variables", {})
|
|
self.Dirs = self.Data.get("DirectoryPaths", {})
|
|
self.Names = self.Data.get("Names", {})
|
|
|
|
# Parsing, as SaraSortd.py does it
|
|
|
|
def Parse(
|
|
self, String: str, ConfData: Optional[dict] = None, NextNum=None, NextChar=None, Parent=None,
|
|
OrgFile: Optional[str] = None, VarCall=None, Now: Optional[datetime.datetime] = None,
|
|
) -> str:
|
|
"""SaraSortd's Parse. ConfData is the file %Name% lookups read from
|
|
(GlobalConf unless given, like SaraSortd's Path argument)."""
|
|
if not isinstance(String, str):
|
|
return String
|
|
ConfData = self.Data if ConfData is None else ConfData
|
|
Now = Now or datetime.datetime.now()
|
|
OrgFileName, OrgFileType = os.path.splitext(OrgFile) if OrgFile is not None else (None, None)
|
|
V = self.Vars
|
|
VarValues = {
|
|
V.get("NextNum"): str(NextNum),
|
|
V.get("NextChar"): str(NextChar),
|
|
V.get("Parent"): str(Parent),
|
|
V.get("OrgFileName"): str(OrgFileName),
|
|
V.get("OrgFileType"): str(OrgFileType),
|
|
V.get("Year"): str(Now.year),
|
|
V.get("Month"): str(Now.month).rjust(2, "0"),
|
|
V.get("Day"): str(Now.day).rjust(2, "0"),
|
|
V.get("Hour"): str(Now.hour).rjust(2, "0"),
|
|
V.get("Minute"): str(Now.minute).rjust(2, "0"),
|
|
V.get("Second"): str(Now.second).rjust(2, "0"),
|
|
V.get("VarCall"): str(VarCall),
|
|
}
|
|
for Key, Value in VarValues.items():
|
|
if Key and Key in String:
|
|
String = String.replace(Key, Value)
|
|
|
|
Call = V.get("VarCall", "%%")
|
|
if len(Call) < 2:
|
|
return String
|
|
|
|
def Replacer(Match: re.Match) -> str:
|
|
Content = Match.group(1)
|
|
if not Content:
|
|
return str(VarCall) if VarCall is not None else ""
|
|
Found = GetConf(Content, ConfData)
|
|
return str(Found) if Found is not None else Match.group(0)
|
|
|
|
return re.sub(f"{re.escape(Call[0])}(.*?){re.escape(Call[1])}", Replacer, String)
|
|
|
|
def ResolveDir(self, Raw: str) -> str:
|
|
return self.Parse(Raw)
|
|
|
|
def DirList(self, Key: str) -> list[str]:
|
|
Value = self.Dirs.get(Key, [])
|
|
return [str(Item) for Item in (Value if isinstance(Value, list) else [Value])]
|
|
|
|
@property
|
|
def Inputs(self) -> list[str]:
|
|
return [self.ResolveDir(Raw) for Raw in self.DirList("InputDir")]
|
|
|
|
@property
|
|
def Outputs(self) -> list[str]:
|
|
return [self.ResolveDir(Raw) for Raw in self.DirList("OutputDir")]
|
|
|
|
def SpecialDir(self, Key: str) -> Optional[Path]:
|
|
Raw = self.Dirs.get(Key)
|
|
return Path(self.ResolveDir(str(Raw))) if Raw else None
|
|
|
|
def RootDir(self) -> Optional[str]:
|
|
Root = self.Dirs.get("RootDir")
|
|
return None if not Root or Root == "Unset" else str(Root)
|
|
|
|
def Shorthand(self, PathText: str) -> str:
|
|
"""How a path is written into GlobalConf: %RootDir%/... when under RootDir."""
|
|
Root = self.RootDir()
|
|
Call = self.Vars.get("VarCall", "%%")
|
|
if Root and len(Call) >= 2:
|
|
Root = Root.rstrip("/")
|
|
if PathText == Root or PathText.startswith(Root + "/"):
|
|
return f"{Call[0]}RootDir{Call[1]}{PathText[len(Root):]}"
|
|
return PathText
|
|
|
|
def ConfFile(self, DirPath: str, Kind: str) -> Path:
|
|
"""A directory's own config, e.g. ~/Sorting/Output/.OutputConfig.toml."""
|
|
Name = self.Parse(str(self.Names.get(NameKeys[Kind], ".@Config")), Parent=os.path.basename(DirPath))
|
|
return Path(DirPath) / f"{Name}.toml"
|
|
|
|
def SharedConfName(self) -> bool:
|
|
"""Input and output configs use the same file name (both default to .@Config)."""
|
|
return self.Names.get("InDirConfName") == self.Names.get("OutDirConfName")
|
|
|
|
def TemplatePath(self, Kind: str) -> Path:
|
|
return self.ConfDir / TemplateNames[Kind]
|
|
|
|
def LoadDirConf(self, DirPath: str, Kind: str, UseTemplate: bool = True) -> Optional[dict]:
|
|
"""A directory's config. When it has none yet, the template it will
|
|
get from SaraSortd on its next start (unless UseTemplate is off)."""
|
|
ConfPath = self.ConfFile(DirPath, Kind)
|
|
try:
|
|
return ReadToml(ConfPath).unwrap()
|
|
except (OSError, tomlkit.exceptions.TOMLKitError):
|
|
pass
|
|
if not UseTemplate or ConfPath.exists():
|
|
return None
|
|
try:
|
|
return self._NewDirConf(DirPath, Kind).unwrap()
|
|
except (OSError, tomlkit.exceptions.TOMLKitError):
|
|
return None
|
|
|
|
def DaemonVersion(self) -> Optional[str]:
|
|
try:
|
|
Match = re.search(r'# Version = "([^"]+)"', self.Config.SortdScriptPath.read_text())
|
|
except OSError:
|
|
return None
|
|
return Match.group(1) if Match else None
|
|
|
|
# The graph
|
|
|
|
def Nodes(self) -> dict[str, DirNode]:
|
|
Nodes: dict[str, DirNode] = {}
|
|
for DirPath in self.Inputs:
|
|
Nodes.setdefault(DirPath, DirNode(DirPath)).IsInput = True
|
|
for DirPath in self.Outputs:
|
|
Nodes.setdefault(DirPath, DirNode(DirPath)).IsOutput = True
|
|
return Nodes
|
|
|
|
def _DirAccepts(self, OutConf: dict, OutputPath: str, InputPath: str) -> bool:
|
|
"""The output directory's own ValidInputDirs, as DecideNewPath checks it."""
|
|
Parent = GetConf("ParentDir", OutConf)
|
|
for Valid in GetConf("ValidInputDirs", OutConf) or []:
|
|
if fnmatch.fnmatchcase(InputPath, self.Parse(str(Valid), Parent=Parent)):
|
|
return True
|
|
return False
|
|
|
|
def _EntryAccepts(self, Entry: dict, OutputPath: str, InputPath: str) -> bool:
|
|
"""SaraSortd 7.1's FileInputValid."""
|
|
Own = Entry.get("ValidInputDirs")
|
|
if Own is None:
|
|
return True
|
|
Parent = os.path.basename(OutputPath)
|
|
return any(fnmatch.fnmatchcase(InputPath, self.Parse(str(Valid), Parent=Parent)) for Valid in Own)
|
|
|
|
def Edges(self) -> list[Edge]:
|
|
Edges = []
|
|
for OutputPath in dict.fromkeys(self.Outputs):
|
|
OutConf = self.LoadDirConf(OutputPath, "Out")
|
|
if OutConf is None:
|
|
continue
|
|
Entries = OutConf.get("Files", [])
|
|
for InputPath in dict.fromkeys(self.Inputs):
|
|
if not self._DirAccepts(OutConf, OutputPath, InputPath):
|
|
continue
|
|
Found = Edge(InputPath, OutputPath)
|
|
for Index, Entry in enumerate(Entries):
|
|
if "Pattern" not in Entry or not self._EntryAccepts(Entry, OutputPath, InputPath):
|
|
continue
|
|
Found.Rules.append(FileRule(
|
|
Index, str(Entry["Pattern"]), str(Entry.get("NewFileName", "")), "ValidInputDirs" in Entry,
|
|
))
|
|
if Found.Rules:
|
|
Edges.append(Found)
|
|
return Edges
|
|
|
|
# Edits
|
|
|
|
def _SaveGlobal(self) -> None:
|
|
WriteToml(self.GlobalPath, self.Global)
|
|
self.Data = self.Global.unwrap()
|
|
self.Dirs = self.Data.get("DirectoryPaths", {})
|
|
|
|
def _AddToList(self, Key: str, PathText: str) -> bool:
|
|
"""Adds a directory to InputDir/OutputDir. True if GlobalConf changed."""
|
|
if PathText in (self.Inputs if Key == "InputDir" else self.Outputs):
|
|
return False
|
|
Section = self.Global.setdefault("DirectoryPaths", tomlkit.table())
|
|
Current = Section.get(Key)
|
|
if Current is None:
|
|
Section[Key] = [self.Shorthand(PathText)]
|
|
elif isinstance(Current, list):
|
|
Current.append(self.Shorthand(PathText))
|
|
else: # a single string; SaraSortd expects a list
|
|
Section[Key] = [str(Current), self.Shorthand(PathText)]
|
|
self._SaveGlobal()
|
|
return True
|
|
|
|
def AddDirectory(self, PathText: str, Keys: list[str]) -> bool:
|
|
"""Adds a directory as an input and/or output ("InputDir"/"OutputDir")
|
|
and writes its config from the template. True if GlobalConf changed."""
|
|
Changed = False
|
|
for Key in Keys:
|
|
Changed = self._AddToList(Key, PathText) or Changed
|
|
for Key in Keys:
|
|
Kind = "In" if Key == "InputDir" else "Out"
|
|
Document, IsNew = self.EnsureDirConf(PathText, Kind)
|
|
WriteToml(self.ConfFile(PathText, Kind), Document)
|
|
return Changed
|
|
|
|
def RemoveDir(self, PathText: str) -> bool:
|
|
"""Takes a directory out of InputDir and OutputDir. Leaves the
|
|
directory and its config on disk. True if GlobalConf changed."""
|
|
Changed = False
|
|
Section = self.Global.get("DirectoryPaths", {})
|
|
for Key in ("InputDir", "OutputDir"):
|
|
Current = Section.get(Key)
|
|
if not isinstance(Current, list):
|
|
continue
|
|
for Index in reversed(range(len(Current))):
|
|
if self.ResolveDir(str(Current[Index])) == PathText:
|
|
del Current[Index]
|
|
Changed = True
|
|
if Changed:
|
|
self._SaveGlobal()
|
|
return Changed
|
|
|
|
def _NewDirConf(self, DirPath: str, Kind: str) -> tomlkit.TOMLDocument:
|
|
"""What SaraSortd's Dir() would write: the template, with ParentDir and
|
|
Title set."""
|
|
Name = os.path.basename(DirPath)
|
|
Document = ReadToml(self.TemplatePath(Kind))
|
|
Document["ParentDir"] = Name
|
|
Document["Title"] = f"{Name} Config"
|
|
return Document
|
|
|
|
def EnsureDirConf(self, DirPath: str, Kind: str) -> tuple[tomlkit.TOMLDocument, bool]:
|
|
"""The directory's config, made from the template when missing.
|
|
Returns (document, IsNew). The directory itself is made too."""
|
|
Path(DirPath).mkdir(parents=True, exist_ok=True)
|
|
ConfPath = self.ConfFile(DirPath, Kind)
|
|
if ConfPath.exists():
|
|
Document = ReadToml(ConfPath)
|
|
IsNew = False
|
|
else:
|
|
Document = self._NewDirConf(DirPath, Kind)
|
|
IsNew = True
|
|
if self.SharedConfName() and DirPath in self.Inputs and DirPath in self.Outputs:
|
|
# One file for both roles: make sure it has the keys both read.
|
|
Other = "In" if Kind == "Out" else "Out"
|
|
OtherTemplate = ReadToml(self.TemplatePath(Other))
|
|
for Key, Value in OtherTemplate.items():
|
|
if Key != "Files" and Key not in Document:
|
|
Document[Key] = Value
|
|
# Its [[Files]] are read as output rules too, which need more keys.
|
|
Defaults = self.TemplateFileDefaults()
|
|
for Entry in Document.get("Files", []):
|
|
for Key, Value in Defaults.items():
|
|
if Key not in Entry:
|
|
Entry[Key] = Value
|
|
return Document, IsNew
|
|
|
|
def TemplateFileDefaults(self) -> dict:
|
|
"""The first [[Files]] entry of OutputDirConf.toml: the defaults for
|
|
every new entry (NewFileName, NextNum, ...)."""
|
|
try:
|
|
Entries = ReadToml(self.TemplatePath("Out")).unwrap().get("Files", [])
|
|
except (OSError, tomlkit.exceptions.TOMLKitError):
|
|
Entries = []
|
|
Defaults = dict(Entries[0]) if Entries else {}
|
|
Defaults.setdefault("NewFileName", "~^")
|
|
Defaults.setdefault("NextNum", "1")
|
|
Defaults.setdefault("NextChar", "A")
|
|
Defaults.setdefault("CaseSensitive", 0)
|
|
Defaults.setdefault("Overwrite", 1)
|
|
Defaults.pop("ValidInputDirs", None)
|
|
return Defaults
|
|
|
|
def Connect(
|
|
self, InputPath: str, OutputPath: str, Patterns: list[str],
|
|
EntryOverrides: Optional[dict] = None, OutputSettings: Optional[dict] = None,
|
|
InputSettings: Optional[dict] = None,
|
|
) -> bool:
|
|
"""Makes InputPath send Patterns to OutputPath. Each pattern becomes a
|
|
[[Files]] entry in the output's config (defaults from OutputDirConf.toml)
|
|
with its own ValidInputDirs, so other inputs aren't affected. Returns
|
|
True if GlobalConf.toml changed (SaraSortd needs a restart)."""
|
|
Patterns = [Pattern.strip() for Pattern in Patterns if Pattern.strip()]
|
|
Changed = self._AddToList("InputDir", InputPath)
|
|
Changed = self._AddToList("OutputDir", OutputPath) or Changed
|
|
|
|
# The output side.
|
|
OutDoc, IsNew = self.EnsureDirConf(OutputPath, "Out")
|
|
if IsNew and "Files" in OutDoc:
|
|
# The template's catch-all entry would take every file from every
|
|
# input; the new entries replace it.
|
|
del OutDoc["Files"]
|
|
for Key, Value in (OutputSettings or {}).items():
|
|
OutDoc[Key] = Value
|
|
OutData = OutDoc.unwrap()
|
|
if not self._DirAccepts(OutData, OutputPath, InputPath):
|
|
Valid = OutDoc.get("ValidInputDirs")
|
|
if isinstance(Valid, list):
|
|
Valid.append(self.Shorthand(InputPath))
|
|
else:
|
|
OutDoc["ValidInputDirs"] = [self.Shorthand(InputPath)]
|
|
if "Files" not in OutDoc:
|
|
OutDoc["Files"] = AoT([])
|
|
Files = OutDoc["Files"]
|
|
Defaults = self.TemplateFileDefaults()
|
|
for Pattern in Patterns:
|
|
Existing = next((Entry for Entry in Files if Entry.get("Pattern") == Pattern), None)
|
|
if Existing is not None:
|
|
Own = Existing.get("ValidInputDirs")
|
|
if Own is not None and not self._EntryAccepts(Existing.unwrap(), OutputPath, InputPath):
|
|
Own.append(self.Shorthand(InputPath))
|
|
for Key, Value in (EntryOverrides or {}).items():
|
|
Existing[Key] = Value
|
|
continue
|
|
Entry = tomlkit.table()
|
|
Entry["Pattern"] = Pattern
|
|
for Key, Value in Defaults.items():
|
|
if Key != "Pattern":
|
|
Entry[Key] = Value
|
|
for Key, Value in (EntryOverrides or {}).items():
|
|
Entry[Key] = Value
|
|
Entry["ValidInputDirs"] = [self.Shorthand(InputPath)]
|
|
Files.append(Entry)
|
|
WriteToml(self.ConfFile(OutputPath, "Out"), OutDoc)
|
|
|
|
# The input side: it has to let these files through at all.
|
|
InDoc, IsNew = self.EnsureDirConf(InputPath, "In")
|
|
for Key, Value in (InputSettings or {}).items():
|
|
InDoc[Key] = Value
|
|
if "Files" not in InDoc:
|
|
InDoc["Files"] = AoT([])
|
|
InPatterns = [str(Entry.get("Pattern")) for Entry in InDoc["Files"]]
|
|
# When the input is also an output and both use one config file, its
|
|
# [[Files]] are output rules too: an input-only entry (no NewFileName)
|
|
# would crash SaraSortd, so leave them alone.
|
|
SharedFile = self.SharedConfName() and InputPath in self.Outputs
|
|
if "*" not in InPatterns and not SharedFile:
|
|
for Pattern in Patterns:
|
|
if Pattern not in InPatterns:
|
|
Entry = tomlkit.table()
|
|
Entry["Pattern"] = Pattern
|
|
Entry["CaseSensitive"] = 0
|
|
InDoc["Files"].append(Entry)
|
|
WriteToml(self.ConfFile(InputPath, "In"), InDoc)
|
|
return Changed
|
|
|
|
def ConnectedInputs(self, OutputPath: str) -> list[str]:
|
|
OutConf = self.LoadDirConf(OutputPath, "Out") or {}
|
|
return [Input for Input in dict.fromkeys(self.Inputs) if self._DirAccepts(OutConf, OutputPath, Input)]
|
|
|
|
def RemoveRule(self, InputPath: str, OutputPath: str, Index: int) -> Optional[str]:
|
|
"""Stops the [[Files]] entry at Index taking files from InputPath, and
|
|
removes the entry once no input is left. Returns a message when that
|
|
can't be done exactly (a wildcard ValidInputDirs entry)."""
|
|
ConfPath = self.ConfFile(OutputPath, "Out")
|
|
Document = ReadToml(ConfPath)
|
|
Files = Document.get("Files")
|
|
if Files is None or Index >= len(Files):
|
|
return "That rule no longer exists."
|
|
Entry = Files[Index]
|
|
Own = Entry.get("ValidInputDirs")
|
|
Parent = os.path.basename(OutputPath)
|
|
if Own is None:
|
|
# It took files from every input: list the others explicitly.
|
|
Others = [Other for Other in self.ConnectedInputs(OutputPath) if Other != InputPath]
|
|
Entry["ValidInputDirs"] = [self.Shorthand(Other) for Other in Others]
|
|
Own = Entry["ValidInputDirs"]
|
|
else:
|
|
for Position in reversed(range(len(Own))):
|
|
if self.Parse(str(Own[Position]), Parent=Parent) == InputPath:
|
|
del Own[Position]
|
|
Message = None
|
|
if self._EntryAccepts(Entry.unwrap(), OutputPath, InputPath):
|
|
Message = f"{Entry.get('Pattern')} still matches through a wildcard in its ValidInputDirs; edit it in Configs."
|
|
if len(Own) == 0:
|
|
del Files[Index]
|
|
WriteToml(ConfPath, Document)
|
|
return Message
|
|
|
|
def Disconnect(self, InputPath: str, OutputPath: str) -> list[str]:
|
|
Messages = []
|
|
for FoundEdge in self.Edges():
|
|
if FoundEdge.Input == InputPath and FoundEdge.Output == OutputPath:
|
|
for Rule in sorted(FoundEdge.Rules, key=lambda Rule: Rule.Index, reverse=True):
|
|
Message = self.RemoveRule(InputPath, OutputPath, Rule.Index)
|
|
if Message:
|
|
Messages.append(Message)
|
|
return Messages
|
|
|
|
# What would happen to a file
|
|
|
|
def Simulate(self, InputPath: str, FileName: str) -> SimulationResult:
|
|
"""Follows Main() and DecideNewPath() for one file, without touching
|
|
anything (NextNum/NextChar aren't incremented)."""
|
|
Steps: list[str] = []
|
|
if FileName.startswith(".") and self.Data.get("DotFiles", 0) == 0:
|
|
return SimulationResult(["Dot files are skipped (DotFiles = 0)."], None, "Skipped")
|
|
InConf = self.LoadDirConf(InputPath, "In")
|
|
if InConf is None:
|
|
return SimulationResult(["The input's config can't be read."], None, "Skipped")
|
|
Parent = os.path.basename(InputPath)
|
|
Matched = False
|
|
for Entry in InConf.get("Files", []):
|
|
Pattern = self.Parse(str(Entry.get("Pattern", "")), Parent=Parent)
|
|
if Entry.get("CaseSensitive", 0) == 1:
|
|
Matched = fnmatch.fnmatchcase(FileName, Pattern)
|
|
else:
|
|
Matched = fnmatch.fnmatchcase(FileName.lower(), Pattern.lower())
|
|
if Matched:
|
|
Steps.append(f"The input lets it through ({Entry.get('Pattern')}).")
|
|
break
|
|
if not Matched:
|
|
Steps.append("None of the input's own patterns match, so it stays where it is.")
|
|
return SimulationResult(Steps, None, "Skipped")
|
|
|
|
for OutputPath in self.Outputs:
|
|
OutConf = self.LoadDirConf(OutputPath, "Out")
|
|
if OutConf is None:
|
|
Steps.append(f"{OutputPath}: its config can't be read, skipped.")
|
|
continue
|
|
if not self._DirAccepts(OutConf, OutputPath, InputPath):
|
|
Steps.append(f"{OutputPath}: its ValidInputDirs don't include this input.")
|
|
continue
|
|
Limit = OutConf.get("FileLimit", 0) or 0
|
|
if Limit > 0 and os.path.isdir(OutputPath):
|
|
Count = sum(
|
|
1 for Name in os.listdir(OutputPath)
|
|
if not Name.startswith(".") and os.path.isfile(os.path.join(OutputPath, Name))
|
|
)
|
|
if Count >= Limit:
|
|
Steps.append(f"{OutputPath}: full ({Count}/{Limit} files).")
|
|
continue
|
|
for Entry in OutConf.get("Files", []):
|
|
if not self._EntryAccepts(Entry, OutputPath, InputPath):
|
|
continue
|
|
Pattern = str(Entry.get("Pattern", ""))
|
|
if Entry.get("CaseSensitive", 0) == 1:
|
|
Match = fnmatch.fnmatchcase(FileName, Pattern)
|
|
else:
|
|
Match = fnmatch.fnmatch(FileName.lower(), Pattern.lower())
|
|
if not Match:
|
|
continue
|
|
NewName = self.PreviewName(str(Entry.get("NewFileName", "~^")), FileName, OutputPath, Entry, OutConf)
|
|
if Entry.get("Overwrite", 1) == 0 and os.path.exists(os.path.join(OutputPath, NewName)):
|
|
Steps.append(f"{OutputPath}: {Pattern} matches, but {NewName} exists and Overwrite = 0.")
|
|
continue
|
|
Action = "moved" if OutConf.get("DeleteOrg", 1) == 1 else "copied"
|
|
Steps.append(f"{OutputPath}: {Pattern} matches, so it's {Action} and named {NewName}.")
|
|
return SimulationResult(Steps, f"{OutputPath}/{NewName}", "Sorted")
|
|
Steps.append(f"{OutputPath}: no pattern matches.")
|
|
Failed = self.SpecialDir("FailedDir")
|
|
Steps.append(f"No output takes it, so it goes to {Failed}.")
|
|
return SimulationResult(Steps, f"{Failed}/{FileName}" if Failed else None, "Failed")
|
|
|
|
def PreviewName(
|
|
self, NewFileName: str, FileName: str, OutputPath: str, Entry: Optional[dict] = None,
|
|
OutConf: Optional[dict] = None,
|
|
) -> str:
|
|
Entry = Entry or {}
|
|
NextChar = str(Entry.get("NextChar", "A"))
|
|
# Init() pads NextChar to the longest run of the NextChar variable.
|
|
Character = self.Vars.get("NextChar", "&")
|
|
Runs = re.findall(f"{re.escape(Character)}+", NewFileName) if Character else []
|
|
Width = max((len(Run) for Run in Runs), default=0)
|
|
if Width:
|
|
NextChar = NextChar.upper().rjust(Width, "A")
|
|
return self.Parse(
|
|
NewFileName, ConfData=OutConf if OutConf is not None else {}, OrgFile=FileName,
|
|
NextNum=Entry.get("NextNum", "1"), NextChar=NextChar, Parent=os.path.basename(OutputPath),
|
|
)
|
|
|
|
# Problems worth showing
|
|
|
|
def Warnings(self) -> list[tuple[str, str]]:
|
|
"""(Level, Text) pairs; Level is "Error", "Warning" or "Info"."""
|
|
Found: list[tuple[str, str]] = []
|
|
|
|
def Walk(Data: Any, Where: str) -> None:
|
|
if isinstance(Data, dict):
|
|
for Key, Value in Data.items():
|
|
Walk(Value, f"{Where}{Key}" if not Where else f"{Where}.{Key}")
|
|
elif Data is None or Data == "Unset":
|
|
Found.append(("Error", f"GlobalConf.toml: {Where} is Unset. With SafeMode = 1 SaraSortd won't start."))
|
|
|
|
Walk(self.Data, "")
|
|
|
|
Daemon = self.DaemonVersion()
|
|
ConfVersion = str(self.Data.get("Version", "?"))
|
|
if Daemon is None:
|
|
Found.append(("Warning", f"Can't find SaraSortd.py at {self.Config.SortdScriptPath} (see Settings)."))
|
|
elif MajorVersion(Daemon) != MajorVersion(ConfVersion):
|
|
Found.append(("Error", f"GlobalConf.toml is version {ConfVersion}, but SaraSortd is {Daemon}: the first number differs, so the config won't work."))
|
|
|
|
Call = self.Vars.get("VarCall", "%%")
|
|
if len(Call) >= 2:
|
|
Leftover = re.compile(f"{re.escape(Call[0])}[^{re.escape(Call[1])}]+{re.escape(Call[1])}")
|
|
for Key in ("InputDir", "OutputDir"):
|
|
for Raw in self.DirList(Key):
|
|
if Leftover.search(self.ResolveDir(Raw)):
|
|
Found.append(("Error", f"{Raw} doesn't resolve (SaraSortd only replaces one level of variables)."))
|
|
|
|
Inputs = list(dict.fromkeys(self.Inputs))
|
|
Outputs = list(dict.fromkeys(self.Outputs))
|
|
Edges = self.Edges()
|
|
UsesOwnInputs = False
|
|
for DirPath in dict.fromkeys(Inputs + Outputs):
|
|
if not os.path.isdir(DirPath):
|
|
Found.append(("Info", f"{DirPath} doesn't exist yet; SaraSortd makes it when it starts."))
|
|
for Kind in (["In"] if DirPath in Inputs else []) + (["Out"] if DirPath in Outputs else []):
|
|
if os.path.isdir(DirPath) and not self.ConfFile(DirPath, Kind).exists():
|
|
Found.append(("Info", f"{DirPath} has no {'input' if Kind == 'In' else 'output'} config yet; SaraSortd copies the template when it starts."))
|
|
|
|
for InputPath in Inputs:
|
|
if not any(FoundEdge.Input == InputPath for FoundEdge in Edges):
|
|
Found.append(("Warning", f"{InputPath} has no outputs: every file it lets through goes to FailedDir."))
|
|
for OutputPath in Outputs:
|
|
OutConf = self.LoadDirConf(OutputPath, "Out")
|
|
if OutConf is None:
|
|
continue
|
|
if not any(FoundEdge.Output == OutputPath for FoundEdge in Edges):
|
|
Found.append(("Warning", f"{OutputPath} can't receive anything: no input matches its ValidInputDirs and patterns."))
|
|
if "*" in [str(Valid) for Valid in OutConf.get("ValidInputDirs", [])] and any(
|
|
"ValidInputDirs" not in Entry for Entry in OutConf.get("Files", [])
|
|
):
|
|
Found.append(("Info", f"{OutputPath} takes files from every input (ValidInputDirs = [\"*\"])."))
|
|
Entries = OutConf.get("Files", [])
|
|
for Position, Entry in enumerate(Entries):
|
|
UsesOwnInputs = UsesOwnInputs or "ValidInputDirs" in Entry
|
|
for Missing in ("Pattern", "NewFileName", "NextNum", "NextChar", "CaseSensitive", "Overwrite"):
|
|
if Missing not in Entry:
|
|
Found.append(("Error", f"{OutputPath}: [[Files]] #{Position + 1} has no {Missing}; SaraSortd will crash on it."))
|
|
if Entry.get("Pattern") == "*" and "ValidInputDirs" not in Entry and Position < len(Entries) - 1:
|
|
Found.append(("Warning", f"{OutputPath}: [[Files]] #{Position + 1} is a catch-all \"*\", so the entries after it never get a file."))
|
|
for InputPath in Inputs:
|
|
# SaraSortd tries outputs in OutputDir order and stops at the first match.
|
|
CatchAll = None
|
|
for FoundEdge in sorted(
|
|
(FoundEdge for FoundEdge in Edges if FoundEdge.Input == InputPath),
|
|
key=lambda FoundEdge: Outputs.index(FoundEdge.Output),
|
|
):
|
|
if CatchAll:
|
|
Found.append(("Warning", f"{CatchAll} takes every file from {InputPath} first, so {FoundEdge.Output} never gets any ({', '.join(FoundEdge.Patterns)})."))
|
|
elif "*" in FoundEdge.Patterns:
|
|
CatchAll = FoundEdge.Output
|
|
if UsesOwnInputs and Daemon and not VersionAtLeast(Daemon, (7, 1)):
|
|
Found.append(("Error", f"Some [[Files]] entries have their own ValidInputDirs, which SaraSortd {Daemon} ignores. Update SaraSortd to 7.1."))
|
|
|
|
for Raw in self.DirList("InputDir") + self.DirList("OutputDir"):
|
|
Odd = [Char for Char in (self.Vars.get(Key) for Key in ("NextNum", "NextChar", "Parent", "OrgFileName", "OrgFileType")) if Char and Char in Raw]
|
|
if Odd:
|
|
Found.append(("Warning", f"{Raw} contains {', '.join(Odd)}, which SaraSortd swaps for a variable."))
|
|
|
|
Order = {"Error": 0, "Warning": 1, "Info": 2}
|
|
return sorted(Found, key=lambda Item: Order[Item[0]])
|