Files
SaraSort/Sortd.py
T
2026-10-09 14:16:50 +02:00

936 lines
44 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
import shutil
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Optional
import tomlkit
from tomlkit.items import AoT
def _Slashes(PathText: str) -> str:
"""Windows paths use "\\" (and configs may mix both); SaraSortd 7.5 names and compares with "/"."""
return PathText.replace("\\", "/")
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.parent.mkdir(parents=True, exist_ok=True) # configs can live outside their directory
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
PartialDownloadSuffixes = (".part", ".crdownload", ".download")
def IsPartialDownload(FilePath: str) -> bool:
"""SaraSortd 7.4's IsPartialDownload: still downloading (Firefox, Chromium,
Safari), or a file next to its own unfinished download."""
if os.path.basename(FilePath).lower().endswith(PartialDownloadSuffixes):
return True
return any(os.path.exists(FilePath + Suffix) for Suffix in PartialDownloadSuffixes)
def IsCatchAll(Pattern: str) -> bool:
"""SaraSortd 7.3's IsCatchAll: a pattern of only "*" takes any file, so it's a fallback."""
return Pattern != "" and set(Pattern) == {"*"}
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, Document: Optional[tomlkit.TOMLDocument] = None):
"""Document: a GlobalConf.toml that isn't on disk yet (an import being
checked); otherwise it's read from SortdConfigDir."""
self.Config = Config
self.ConfDir = Path(Config.SortdConfigDir).expanduser()
self.GlobalPath = Config.GlobalConfPath
if Document is not None:
self.Global = Document
elif not self.GlobalPath.exists():
raise SortdError(f"No GlobalConf.toml at {self.GlobalPath}")
else:
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 self.ResolveLookups(str(Found), ConfData) if Found is not None else Match.group(0)
return re.sub(f"{re.escape(Call[0])}(.*?){re.escape(Call[1])}", Replacer, String)
def ResolveLookups(self, String: str, ConfData: Optional[dict] = None, Depth: int = 0) -> str:
"""SaraSortd 7.3's ResolveLookups: %Name% lookups, also inside the values
they bring in (ImageDir = "%RootDir%/Images" used as "%ImageDir%").
Unknown names are left as they are."""
ConfData = self.Data if ConfData is None else ConfData
Call = self.Vars.get("VarCall", "%%")
if len(Call) < 2:
return String
def Replacer(Match: re.Match) -> str:
Found = GetConf(Match.group(1), ConfData) if Match.group(1) else None
if Found is None or Depth >= 10:
return Match.group(0)
return self.ResolveLookups(str(Found), ConfData, Depth + 1)
return re.sub(f"{re.escape(Call[0])}(.*?){re.escape(Call[1])}", Replacer, String)
def ParsePath(self, String: str) -> str:
"""SaraSortd 7.3's ParsePath: directory paths only get %Name% lookups and
the date and time variables, so "~/Stuff&Things" stays as it is; a
leading "~" is the home directory."""
Now = datetime.datetime.now()
V = self.Vars
for Key, Value in {
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"),
}.items():
if Key and Key in String:
String = String.replace(Key, Value)
return os.path.expanduser(self.ResolveLookups(String))
def ResolveDir(self, Raw: str) -> str:
return self.ParsePath(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.
Only when it parses back to the very same path: SaraSortd compares
ValidInputDirs as text, so RootDir = "/home/sarah/" must give
%RootDir%Downloads, not %RootDir%/Downloads (/home/sarah//Downloads)."""
Root = self.RootDir()
Call = self.Vars.get("VarCall", "%%")
if Root and len(Call) >= 2:
Root = Root.rstrip("/")
if PathText == Root or PathText.startswith(Root + "/"):
Lookup = f"{Call[0]}RootDir{Call[1]}"
for Short in (Lookup + PathText[len(Root):], Lookup + PathText[len(Root) + 1:]):
if self.ParsePath(Short) == PathText:
return Short
return PathText
def ConfFolder(self, PathText: str) -> Path:
"""SaraSortd's ConfFolder: relative paths count from GlobalConf.toml's folder."""
Parsed = self.ParsePath(str(PathText))
if os.path.isabs(Parsed):
return Path(os.path.normpath(Parsed))
return Path(os.path.normpath(self.GlobalPath.parent / Parsed))
def FlatName(self, DirPath: str) -> str:
"""SaraSortd's FlatName: "%RootDir%/Pictures/Cats" -> "Pictures_Cats".
Like SaraSortd 7.5, "\\" counts as "/" and a drive's ":" is dropped (Windows)."""
Root = _Slashes(self.ParsePath(str(self.Dirs.get("RootDir", "")))).rstrip("/")
DirPath = _Slashes(DirPath).rstrip("/")
if Root and DirPath.startswith(Root + "/"):
Relative = DirPath[len(Root) + 1:]
else:
Relative = DirPath.strip("/") or os.path.basename(DirPath)
return Relative.replace(":", "").replace("/", "_")
def CentralConf(self, DirPath: str, Kind: str) -> Optional[Path]:
"""SaraSortd 7.2: a config named in [DirConfs], or one in DirConfDir.
None when the config lives inside the directory."""
for Key, Value in (self.Data.get("DirConfs") or {}).items():
if _Slashes(self.ParsePath(str(Key))).rstrip("/") == _Slashes(DirPath).rstrip("/"):
if isinstance(Value, dict):
Value = Value.get(Kind)
if Value:
return self.ConfFolder(str(Value))
Folder = self.Dirs.get("DirConfDir")
if Folder:
Name = self.Parse(str(self.Names.get(NameKeys[Kind], ".@Config")), Parent=self.FlatName(DirPath)).lstrip(".")
return self.ConfFolder(str(Folder)) / f"{Name}.toml"
return None
def ConfFile(self, DirPath: str, Kind: str) -> Path:
"""A directory's own config: inside it (e.g. ~/Sorting/Output/.OutputConfig.toml),
or wherever DirConfDir / [DirConfs] put it."""
Central = self.CentralConf(DirPath, Kind)
if Central:
return Central
Name = self.Parse(str(self.Names.get(NameKeys[Kind], ".@Config")), Parent=os.path.basename(DirPath))
return Path(DirPath) / f"{Name}.toml"
def HistoryFile(self, DirPath: str) -> Optional[Path]:
"""SaraSortd's HistoryPath: next to a config kept outside the directory, or inside it."""
HistoryName = self.Names.get("FileHistoryName")
if not HistoryName:
return None
Central = self.CentralConf(DirPath, "In") or self.CentralConf(DirPath, "Out")
if Central:
return Central.parent / f"{self.Parse(str(HistoryName), Parent=self.FlatName(DirPath)).lstrip('.')}.toml"
return Path(DirPath) / f"{self.Parse(str(HistoryName), Parent=os.path.basename(DirPath))}.toml"
def SharesConf(self, DirPath: str) -> bool:
"""Its input and output config are one file (the default .@Config names do that)."""
return self.ConfFile(DirPath, "In") == self.ConfFile(DirPath, "Out")
def CentralFolder(self) -> Optional[Path]:
Folder = self.Dirs.get("DirConfDir")
return self.ConfFolder(str(Folder)) if Folder else None
def MoveDirConfs(self, NewFolder: str) -> int:
"""Sets DirConfDir ("" = inside each directory) and moves every existing
directory config and history file to match, so no rules are lost.
Directories named in [DirConfs] stay where they are. Returns how many
files moved. Raises SortdError (before moving anything) when a file is
already where one would go."""
NewDoc = tomlkit.parse(tomlkit.dumps(self.Global))
NewDoc.setdefault("DirectoryPaths", tomlkit.table())["DirConfDir"] = NewFolder
After = Setup(self.Config, NewDoc)
Folder = After.CentralFolder()
for InputPath in self.Inputs:
if Folder and Folder.is_relative_to(Path(InputPath)):
raise SortdError(f"{Folder} is inside the input {InputPath}; SaraSortd would sort its own configs away.")
Plan: dict[Path, Path] = {}
for DirPath in dict.fromkeys(self.Inputs + self.Outputs):
Kinds = (["In"] if DirPath in self.Inputs else []) + (["Out"] if DirPath in self.Outputs else [])
Pairs = [(self.ConfFile(DirPath, Kind), After.ConfFile(DirPath, Kind)) for Kind in Kinds]
Pairs.append((self.HistoryFile(DirPath), After.HistoryFile(DirPath)))
for Old, New in Pairs:
if Old and New and Old != New and Old.exists():
Plan[Old] = New
Taken = [New for New in Plan.values() if New.exists()]
if Taken:
raise SortdError(f"Not moved: {', '.join(str(New) for New in Taken)} already exist(s).")
for Old, New in Plan.items():
New.parent.mkdir(parents=True, exist_ok=True)
shutil.move(str(Old), str(New))
WriteToml(self.GlobalPath, NewDoc)
self.Global = NewDoc
self.Data = NewDoc.unwrap()
self.Dirs = self.Data.get("DirectoryPaths", {})
return len(Plan)
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."""
for Valid in GetConf("ValidInputDirs", OutConf) or []:
if fnmatch.fnmatchcase(InputPath, self.ParsePath(str(Valid))):
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
return any(fnmatch.fnmatchcase(InputPath, self.ParsePath(str(Valid))) 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 DirPath in self.Inputs and DirPath in self.Outputs and self.SharesConf(DirPath):
# 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()
Desired = {"NewFileName": Defaults["NewFileName"], **(EntryOverrides or {})}
Splits: list[str] = [] # patterns this input already gets, but named differently now
for Pattern in Patterns:
Current = next((Entry for Entry in Files if Entry.get("Pattern") == Pattern
and self._EntryAccepts(Entry.unwrap(), OutputPath, InputPath)), None)
if Current is not None:
if EntryOverrides and any(Current.get(Key) != Value for Key, Value in EntryOverrides.items()):
Splits.append(Pattern)
continue
# An entry other inputs use with the same pattern and naming can take this input too.
Reusable = next((Entry for Entry in Files if Entry.get("Pattern") == Pattern
and isinstance(Entry.get("ValidInputDirs"), list)
and all(Entry.get(Key) == Value for Key, Value in Desired.items())), None)
if Reusable is not None:
Reusable["ValidInputDirs"].append(self.Shorthand(InputPath))
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)
for Pattern in Splits:
Entries = (self.LoadDirConf(OutputPath, "Out") or {}).get("Files", [])
Index = next(Position for Position, Entry in enumerate(Entries) if Entry.get("Pattern") == Pattern
and self._EntryAccepts(Entry, OutputPath, InputPath))
self.EditRule(InputPath, OutputPath, Index, dict(EntryOverrides))
# 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 = InputPath in self.Outputs and self.SharesConf(InputPath)
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")
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.ParsePath(str(Own[Position])) == 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
def RuleInputs(self, OutputPath: str, Index: int) -> list[str]:
"""The inputs the [[Files]] entry at Index takes files from."""
OutConf = self.LoadDirConf(OutputPath, "Out") or {}
Entries = OutConf.get("Files", [])
if Index >= len(Entries):
return []
return [Input for Input in self.ConnectedInputs(OutputPath) if self._EntryAccepts(Entries[Index], OutputPath, Input)]
def EditRule(self, InputPath: str, OutputPath: str, Index: int, Changes: dict) -> Optional[str]:
"""Changes (Pattern, NewFileName, CaseSensitive, Overwrite) the [[Files]]
entry at Index for the arrow InputPath -> OutputPath. An entry other
inputs use too is split: this arrow gets its own changed copy just
before it (so it matches first), the others keep the original.
Returns a note when it couldn't be split."""
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]
Others = [Input for Input in self.RuleInputs(OutputPath, Index) if Input != InputPath]
Message = None
if Others:
Own = Entry.get("ValidInputDirs")
Remaining = None
if Own is None:
Remaining = [self.Shorthand(Other) for Other in Others]
else:
Remaining = [Valid for Valid in Own if self.ParsePath(str(Valid)) != InputPath]
Test = dict(Entry.unwrap(), ValidInputDirs=Remaining)
if self._EntryAccepts(Test, OutputPath, InputPath):
Remaining = None # a wildcard still matches this input: can't split it off
Message = (f"{Entry.get('Pattern')} reaches this input through a wildcard, so the change "
f"applies to {', '.join(os.path.basename(Other) for Other in Others)} too.")
if Remaining is not None:
Copy = tomlkit.table()
for Key, Value in Entry.unwrap().items():
Copy[Key] = Value
for Key, Value in Changes.items():
Copy[Key] = Value
Copy["ValidInputDirs"] = [self.Shorthand(InputPath)]
Entry["ValidInputDirs"] = Remaining
Files.insert(Index, Copy)
WriteToml(ConfPath, Document)
return None
for Key, Value in Changes.items():
Entry[Key] = Value
WriteToml(ConfPath, Document)
return Message
def DeleteDirectory(self, PathText: str) -> bool:
"""Takes a directory out of SaraSortd completely: out of InputDir and
OutputDir, out of other outputs' ValidInputDirs (entries left with no
input are removed), and its config and history files are deleted.
The directory and its files stay. True if GlobalConf changed."""
for OutputPath in dict.fromkeys(self.Outputs):
if OutputPath == PathText:
continue
ConfPath = self.ConfFile(OutputPath, "Out")
try:
Document = ReadToml(ConfPath)
except (OSError, tomlkit.exceptions.TOMLKitError):
continue
Changed = False
DirValid = Document.get("ValidInputDirs")
if isinstance(DirValid, list):
for Position in reversed(range(len(DirValid))):
if self.ParsePath(str(DirValid[Position])) == PathText:
del DirValid[Position]
Changed = True
Files = Document.get("Files")
for Index in reversed(range(len(Files) if Files is not None else 0)):
Own = Files[Index].get("ValidInputDirs")
if not isinstance(Own, list):
continue
Before = len(Own)
for Position in reversed(range(Before)):
if self.ParsePath(str(Own[Position])) == PathText:
del Own[Position]
if len(Own) != Before:
Changed = True
if len(Own) == 0:
del Files[Index]
if Changed:
WriteToml(ConfPath, Document)
GlobalChanged = self.RemoveDir(PathText)
Leftovers = {self.ConfFile(PathText, "In"), self.ConfFile(PathText, "Out"), self.HistoryFile(PathText)}
for FilePath in filter(None, Leftovers):
FilePath.unlink(missing_ok=True)
return GlobalChanged
# 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 IsPartialDownload(f"{InputPath}/{FileName}"):
return SimulationResult(["It's still downloading (.part, .crdownload or .download), so SaraSortd leaves it "
"alone until the download finishes."], None, "Skipped")
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")
Candidates = []
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
Candidates.append((OutputPath, OutConf))
# SaraSortd 7.3: specific patterns first, catch-alls ("*") only as a fallback.
for CatchAll in (False, True):
if CatchAll:
Steps.append("No specific pattern takes it; trying catch-alls (\"*\").")
for OutputPath, OutConf in Candidates:
for Entry in OutConf.get("Files", []):
Pattern = str(Entry.get("Pattern", ""))
if IsCatchAll(Pattern) != CatchAll or not self._EntryAccepts(Entry, OutputPath, InputPath):
continue
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")
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} uses a %Name% that GlobalConf.toml doesn't define."))
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:
if Missing != "Pattern" and Daemon and VersionAtLeast(Daemon, (7, 3)):
Found.append(("Info", f"{OutputPath}: [[Files]] #{Position + 1} has no {Missing}; SaraSortd fills it in from the template on its next start."))
else:
Found.append(("Error", f"{OutputPath}: [[Files]] #{Position + 1} has no {Missing}; SaraSortd will crash on it."))
if Daemon and not VersionAtLeast(Daemon, (7, 3)):
Found.append(("Warning", f"SaraSort shows how SaraSortd 7.4 sorts, and SaraSortd {Daemon} differs: a catch-all (\"*\") "
"output listed first takes every file, nested %Name%s don't resolve, ! & @ ~ ^ in "
"paths get swapped, and files still downloading get sorted. Update SaraSortd to 7.4."))
elif Daemon and not VersionAtLeast(Daemon, (7, 4)):
Found.append(("Warning", f"SaraSortd {Daemon} sorts files that are still downloading (.part, .crdownload, "
".download). Update SaraSortd to 7.4."))
Central = self.CentralFolder()
if (Central or self.Data.get("DirConfs")) and Daemon and not VersionAtLeast(Daemon, (7, 2)):
Found.append(("Error", f"Directory configs are kept outside their directories (DirConfDir / [DirConfs]), "
f"which SaraSortd {Daemon} doesn't know about. Update SaraSortd to 7.2."))
if Central:
for InputPath in Inputs:
if Central.is_relative_to(Path(InputPath)):
Found.append(("Error", f"DirConfDir ({Central}) is inside the input {InputPath}, so SaraSortd "
"would sort its own configs away."))
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."))
Order = {"Error": 0, "Warning": 1, "Info": 2}
return sorted(Found, key=lambda Item: Order[Item[0]])