SaraTag/Updater.py

"""Bringing Books in line with their Lexicons: create missing Books, add new
tags, clean removed ones, reset values, rebuild, or repair Parent links.
Only directories that have a Lexicon are touched, so SaraTag never invents
tags for unrelated subdirectories."""

import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Iterable

from Books import FileForBook, InitialValue, Normalise, SerializeBook, ValidateValue
from Library import DataError, ReadToml, TagCore, WriteAtomically

# Name: (label, what it does). The names are what .SaraTagHistory.toml stores.
Modes: dict[str, tuple[str, str]] = {
    "Expand": ("Update all & add missing", "Creates missing Books and adds new tags. Keeps every value."),
    "Clean": ("Clean unused tags", "Like the above, then removes tags the Lexicon no longer has."),
    "Reset": ("Set all to defaults", "Sets every tag back to its default value."),
    "Rebuild": ("Regenerate Books", "Rewrites every Book from the Lexicon: defaults only, unknown tags dropped."),
    "RepairParents": ("Repair Book parents", "Only fixes each Book's Parent link (after moving or renaming directories)."),
}
DestructiveModes = {"Clean", "Reset", "Rebuild"}


@dataclass
class UpdateResult:
    RootPath: Path
    Mode: str
    DryRun: bool
    DirectoriesScanned: int = 0
    TaggedDirectories: int = 0
    FilesScanned: int = 0
    BooksCreated: int = 0
    BooksUpdated: int = 0
    TagsAdded: int = 0
    ValuesReset: int = 0
    TagsRemoved: int = 0
    ParentsRepaired: int = 0
    ChangedBooks: list[Path] = field(default_factory=list)
    OrphanedBooks: list[Path] = field(default_factory=list)
    InvalidValues: list[str] = field(default_factory=list)
    Errors: list[str] = field(default_factory=list)

    def Merge(self, Other: "UpdateResult") -> None:
        for Name in ("DirectoriesScanned", "TaggedDirectories", "FilesScanned", "BooksCreated", "BooksUpdated",
                     "TagsAdded", "ValuesReset", "TagsRemoved", "ParentsRepaired"):
            setattr(self, Name, getattr(self, Name) + getattr(Other, Name))
        for Name in ("ChangedBooks", "OrphanedBooks", "InvalidValues", "Errors"):
            getattr(self, Name).extend(getattr(Other, Name))

    def Summary(self) -> list[tuple[str, int]]:
        return [
            ("Directories scanned", self.DirectoriesScanned),
            ("Tagged directories", self.TaggedDirectories),
            ("Files scanned", self.FilesScanned),
            ("Books created", self.BooksCreated),
            ("Books updated", self.BooksUpdated),
            ("Tags added", self.TagsAdded),
            ("Values reset", self.ValuesReset),
            ("Tags removed", self.TagsRemoved),
            ("Parents repaired", self.ParentsRepaired),
            ("Orphaned Books", len(self.OrphanedBooks)),
            ("Invalid values", len(self.InvalidValues)),
            ("Errors", len(self.Errors)),
        ]


def _Directories(RootPath: Path) -> Iterable[Path]:
    """A Record and its non-hidden subdirectories."""
    for DirectoryText, Names, _ in os.walk(RootPath):
        Names[:] = sorted(Name for Name in Names if not Name.startswith("."))
        yield Path(DirectoryText)


def _OrdinaryFiles(Directory: Path) -> list[Path]:
    return sorted((Entry for Entry in Directory.iterdir() if Entry.is_file() and not Entry.name.startswith(".")),
                  key=lambda Entry: Entry.name.casefold())


def _Write(BookPath: Path, Metadata: dict[str, Any], Result: UpdateResult) -> bool:
    if Result.DryRun:
        return True
    try:
        WriteAtomically(BookPath, SerializeBook(Metadata))
        return True
    except OSError as Error:
        Result.Errors.append(f"Couldn't write {BookPath}: {Error}")
        return False


def _Synchronise(Existing: dict[str, Any], Parent: str, Definitions: dict[str, dict[str, Any]], Mode: str,
                 Result: UpdateResult, FilePath: Path) -> tuple[dict[str, Any], bool]:
    """The Book's new contents and whether anything changed."""
    Updated = dict(Existing)
    if Updated.get("Parent") != Parent:
        Updated["Parent"] = Parent
        Result.ParentsRepaired += 1

    if Mode == "Rebuild":
        Rebuilt = {"Parent": Parent} | {Name: InitialValue(Definition) for Name, Definition in Definitions.items()}
        Result.TagsRemoved += len(set(Existing) - set(Rebuilt))
        Result.ValuesReset += sum(1 for Name in Definitions if Name in Existing and Existing[Name] != Rebuilt[Name])
        Result.TagsAdded += sum(1 for Name in Definitions if Name not in Existing)
        return Rebuilt, Rebuilt != Existing

    for Name, Definition in Definitions.items():
        if Definition.get("ForceDefault"):
            if "Default" not in Definition:
                Result.InvalidValues.append(f"{FilePath}: {Name} has ForceDefault but no Default.")
            elif Updated.get(Name) != Definition["Default"]:
                Result.ValuesReset += Name in Updated
                Result.TagsAdded += Name not in Updated
                Updated[Name] = Definition["Default"]
            continue
        if Mode == "Reset":
            Initial = InitialValue(Definition)
            if Updated.get(Name) != Initial:
                Result.ValuesReset += 1
                Updated[Name] = Initial
        elif Name not in Updated:
            Updated[Name] = InitialValue(Definition)
            Result.TagsAdded += 1

    if Mode == "Clean":
        Obsolete = [Key for Key in Updated if Key != "Parent" and Key not in Definitions]
        for Key in Obsolete:
            del Updated[Key]
        Result.TagsRemoved += len(Obsolete)

    for Name, Definition in Definitions.items():
        if Name in Updated:
            Problem = ValidateValue(Name, Updated[Name], Definition)
            if Problem:
                Result.InvalidValues.append(f"{FilePath}: {Problem}")
    return Updated, Updated != Existing


def UpdateDirectory(Directory: Path, Mode: str, DryRun: bool, LexiconPath: Path | None = None) -> UpdateResult:
    """One directory, with its own Lexicon (or LexiconPath, a Record's)."""
    Directory = Directory.resolve()
    Result = UpdateResult(Directory, Mode, DryRun, DirectoriesScanned=1)
    LexiconPath = LexiconPath or TagCore.GetLexiconPath(Directory)
    try:
        Files = _OrdinaryFiles(Directory)
    except OSError as Error:
        Result.Errors.append(f"Couldn't read {Directory}: {Error}")
        return Result
    Parent = TagCore.GetBookParentValue(Directory)

    if Mode == "RepairParents":
        for FilePath in Files:
            BookPath = TagCore.GetBookPath(FilePath)
            if not BookPath.is_file():
                continue
            Result.FilesScanned += 1
            try:
                Metadata = ReadToml(BookPath)
            except DataError as Error:
                Result.Errors.append(str(Error))
                continue
            if Metadata.get("Parent") == Parent:
                continue
            Metadata = {"Parent": Parent} | {Key: Value for Key, Value in Metadata.items() if Key != "Parent"}
            if _Write(BookPath, Metadata, Result):
                Result.ParentsRepaired += 1
                Result.BooksUpdated += 1
                Result.ChangedBooks.append(BookPath)
        return Result

    if not LexiconPath.is_file():
        return Result
    Result.TaggedDirectories = 1
    try:
        Definitions = {Name: Normalise(Definition) for Name, Definition in TagCore.ResolveLexiconTags(LexiconPath).items()}
    except (OSError, ValueError) as Error:
        Result.Errors.append(f"{LexiconPath.name}: {Error}")
        return Result

    for Entry in Directory.iterdir():
        if TagCore.IsBookPath(Entry):
            Owner = FileForBook(Entry)
            if Owner is not None and not Owner.exists():
                Result.OrphanedBooks.append(Entry)

    for FilePath in Files:
        Result.FilesScanned += 1
        BookPath = TagCore.GetBookPath(FilePath)
        if not BookPath.is_file():
            Metadata = {"Parent": Parent} | {Name: InitialValue(Definition) for Name, Definition in Definitions.items()}
            if _Write(BookPath, Metadata, Result):
                Result.BooksCreated += 1
                Result.ChangedBooks.append(BookPath)
            continue
        try:
            Existing = ReadToml(BookPath)
        except DataError as Error:
            Result.Errors.append(str(Error))
            continue
        Updated, Changed = _Synchronise(Existing, Parent, Definitions, Mode, Result, FilePath)
        if Changed and _Write(BookPath, Updated, Result):
            Result.BooksUpdated += 1
            Result.ChangedBooks.append(BookPath)
    return Result


def UpdateRecord(RootPath: Path, LexiconPath: Path | None, Mode: str, DryRun: bool,
                 Progress: Callable[[Path], None] | None = None) -> UpdateResult:
    """A Record and every subdirectory with its own Lexicon."""
    RootPath = RootPath.resolve()
    Result = UpdateResult(RootPath, Mode, DryRun)
    if not RootPath.is_dir():
        Result.Errors.append(f"Missing directory: {RootPath}")
        return Result
    for Directory in _Directories(RootPath):
        if Progress:
            Progress(Directory)
        Result.Merge(UpdateDirectory(Directory, Mode, DryRun, LexiconPath if Directory == RootPath else None))
    return Result