"""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]])