"""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, 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 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() 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 = 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 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: Parent = os.path.basename(OutputPath) Remaining = [Valid for Valid in Own if self.Parse(str(Valid), Parent=Parent) != 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): Parent = Document.get("ParentDir") for Position in reversed(range(len(DirValid))): if self.Parse(str(DirValid[Position]), Parent=Parent) == PathText: del DirValid[Position] Changed = True Files = Document.get("Files") Parent = os.path.basename(OutputPath) 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.Parse(str(Own[Position]), Parent=Parent) == 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) Parent = os.path.basename(PathText) Leftovers = {self.ConfFile(PathText, "In"), self.ConfFile(PathText, "Out")} HistoryName = self.Names.get("FileHistoryName") if HistoryName: Leftovers.add(Path(PathText) / f"{self.Parse(str(HistoryName), Parent=Parent)}.toml") for FilePath in 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 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]])