feat(app): limit scope of notes with one or more filters (#13)

* style: rename `VaultMetadata.add_metadata` to `VaultMetadata.index_metadata`

* refactor(vault): refactor filtering notes

* fix(application): improve usage display

* fix(application): improve colors of questions

* feat(application): limit the scope of notes to be processed with one or more filters

* build(deps): update identify

src/obsidian_metadata/models/__init__.py

@@ -8,7 +8,7 @@ from obsidian_metadata.models.metadata import (
     VaultMetadata,
 )
 from obsidian_metadata.models.notes import Note
-from obsidian_metadata.models.vault import Vault
+from obsidian_metadata.models.vault import Vault, VaultFilter
 
 from obsidian_metadata.models.application import Application  # isort: skip
 
@@ -23,4 +23,5 @@ __all__ = [
     "Patterns",
     "Vault",
     "VaultMetadata",
+    "VaultFilter",
 ]

src/obsidian_metadata/models/application.py

@@ -5,10 +5,12 @@ from typing import Any
 
 import questionary
 from rich import print
-from textwrap import dedent
+from rich import box
+from rich.console import Console
+from rich.table import Table
 from obsidian_metadata._config import VaultConfig
 from obsidian_metadata._utils.alerts import logger as log
-from obsidian_metadata.models import Patterns, Vault
+from obsidian_metadata.models import Patterns, Vault, VaultFilter
 from obsidian_metadata._utils import alerts
 from obsidian_metadata.models.questions import Questions
 from obsidian_metadata.models.enums import MetadataType
@@ -28,10 +30,11 @@ class Application:
         self.config = config
         self.dry_run = dry_run
         self.questions = Questions()
+        self.filters: list[VaultFilter] = []
 
     def application_main(self) -> None:
         """Questions for the main application."""
-        self.load_vault()
+        self._load_vault()
 
         while True:
             self.vault.info()
@@ -65,12 +68,9 @@ class Application:
 
     def application_add_metadata(self) -> None:
         """Add metadata."""
-        help_text = """
-        USAGE    | Add Metadata
-                   [dim]Add new metadata to your vault. Currently only supports
-                   adding to the frontmatter of a note.[/]
-        """
-        print(dedent(help_text))
+        alerts.usage(
+            "Add new metadata to your vault. Currently only supports adding to the frontmatter of a note."
+        )
 
         area = self.questions.ask_area()
         match area:
@@ -103,41 +103,109 @@ class Application:
 
     def application_filter(self) -> None:
         """Filter notes."""
-        help_text = """
-        USAGE    | Filter Notes
-                   [dim]Enter a regex to filter notes by path. This allows you to
-                   specify a subset of notes to update. Leave empty to include
-                   all markdown files.[/]
-        """
-        print(dedent(help_text))
+        alerts.usage("Limit the scope of notes to be processed with one or more filters.")
 
         choices = [
-            {"name": "Apply regex filter", "value": "apply_filter"},
+            {"name": "Apply new regex path filter", "value": "apply_path_filter"},
+            {"name": "Apply new metadata filter", "value": "apply_metadata_filter"},
+            {"name": "Apply new in-text tag filter", "value": "apply_tag_filter"},
+            {"name": "List and clear filters", "value": "list_filters"},
             {"name": "List notes in scope", "value": "list_notes"},
             questionary.Separator(),
             {"name": "Back", "value": "back"},
         ]
         while True:
             match self.questions.ask_selection(choices=choices, question="Select an action"):
-                case "apply_filter":
+                case "apply_path_filter":
 
-                    path_filter = self.questions.ask_filter_path()
-                    if path_filter is None:
+                    path = self.questions.ask_filter_path()
+                    if path is None or path == "":
                         return
 
-                    if path_filter == "":
-                        path_filter = None
+                    self.filters.append(VaultFilter(path_filter=path))
+                    self._load_vault()
 
-                    self.load_vault(path_filter=path_filter)
+                case "apply_metadata_filter":
+                    key = self.questions.ask_existing_key()
+                    if key is None:
+                        return
 
-                    total_notes = self.vault.num_notes() + self.vault.num_excluded_notes()
-
-                    if path_filter is None:
-                        alerts.success(f"Loaded all {total_notes} total notes")
+                    questions2 = Questions(vault=self.vault, key=key)
+                    value = questions2.ask_existing_value(
+                        question="Enter the value for the metadata filter",
+                    )
+                    if value is None:
+                        return
+                    if value == "":
+                        self.filters.append(VaultFilter(key_filter=key))
                     else:
-                        alerts.success(
-                            f"Loaded {self.vault.num_notes()} notes from {total_notes} total notes"
-                        )
+                        self.filters.append(VaultFilter(key_filter=key, value_filter=value))
+                    self._load_vault()
+
+                case "apply_tag_filter":
+                    tag = self.questions.ask_existing_inline_tag()
+                    if tag is None or tag == "":
+                        return
+
+                    self.filters.append(VaultFilter(tag_filter=tag))
+                    self._load_vault()
+
+                case "list_filters":
+                    if len(self.filters) == 0:
+                        alerts.notice("No filters have been applied")
+                        return
+
+                    print("")
+                    table = Table(
+                        "Opt",
+                        "Filter",
+                        "Type",
+                        title="Current Filters",
+                        show_header=False,
+                        box=box.HORIZONTALS,
+                    )
+                    for _n, filter in enumerate(self.filters, start=1):
+                        if filter.path_filter is not None:
+                            table.add_row(
+                                str(_n),
+                                f"Path regex: [tan bold]{filter.path_filter}",
+                                end_section=bool(_n == len(self.filters)),
+                            )
+                        elif filter.tag_filter is not None:
+                            table.add_row(
+                                str(_n),
+                                f"Tag filter: [tan bold]{filter.tag_filter}",
+                                end_section=bool(_n == len(self.filters)),
+                            )
+                        elif filter.key_filter is not None and filter.value_filter is None:
+                            table.add_row(
+                                str(_n),
+                                f"Key filter: [tan bold]{filter.key_filter}",
+                                end_section=bool(_n == len(self.filters)),
+                            )
+                        elif filter.key_filter is not None and filter.value_filter is not None:
+                            table.add_row(
+                                str(_n),
+                                f"Key/Value : [tan bold]{filter.key_filter}={filter.value_filter}",
+                                end_section=bool(_n == len(self.filters)),
+                            )
+                    table.add_row(f"{len(self.filters) + 1}", "Clear All")
+                    table.add_row(f"{len(self.filters) + 2}", "Return to Main Menu")
+                    Console().print(table)
+
+                    num = self.questions.ask_number(
+                        question="Enter the number of the filter to clear"
+                    )
+                    if num is None:
+                        return
+                    if int(num) <= len(self.filters):
+                        self.filters.pop(int(num) - 1)
+                        self._load_vault()
+                        return
+                    if int(num) == len(self.filters) + 1:
+                        self.filters = []
+                        self._load_vault()
+                        return
 
                 case "list_notes":
                     self.vault.list_editable_notes()
@@ -147,12 +215,9 @@ class Application:
 
     def application_inspect_metadata(self) -> None:
         """View metadata."""
-        help_text = """
-        USAGE    | View Metadata
-                   [dim]Inspect the metadata in your vault. Note, uncommitted
-                   changes will be reflected in these reports[/]
-        """
-        print(dedent(help_text))
+        alerts.usage(
+            "Inspect the metadata in your vault. Note, uncommitted changes will be reflected in these reports"
+        )
 
         choices = [
             {"name": "View all metadata", "value": "all_metadata"},
@@ -168,11 +233,7 @@ class Application:
 
     def application_vault(self) -> None:
         """Vault actions."""
-        help_text = """
-        USAGE    | Vault Actions
-                   [dim]Create or delete a backup of your vault.[/]
-        """
-        print(dedent(help_text))
+        alerts.usage("Create or delete a backup of your vault.")
 
         choices = [
             {"name": "Backup vault", "value": "backup_vault"},
@@ -191,12 +252,7 @@ class Application:
                     return
 
     def application_delete_metadata(self) -> None:
-        help_text = """
-        USAGE    | Delete Metadata
-                   [dim]Delete either a key and all associated values,
-                   or a specific value.[/]
-        """
-        print(dedent(help_text))
+        alerts.usage("Delete either a key and all associated values, or a specific value.")
 
         choices = [
             {"name": "Delete key", "value": "delete_key"},
@@ -219,11 +275,7 @@ class Application:
 
     def application_rename_metadata(self) -> None:
         """Rename metadata."""
-        help_text = """
-        USAGE    | Rename Metadata
-                   [dim]Select the type of metadata to rename.[/]
-        """
-        print(dedent(help_text))
+        alerts.usage("Select the type of metadata to rename.")
 
         choices = [
             {"name": "Rename key", "value": "rename_key"},
@@ -324,14 +376,17 @@ class Application:
 
         return
 
-    def load_vault(self, path_filter: str = None) -> None:
-        """Load the vault.
+    def _load_vault(self) -> None:
+        """Load the vault."""
 
-        Args:
-            path_filter (str, optional): Regex to filter notes by path.
-        """
-        self.vault: Vault = Vault(config=self.config, dry_run=self.dry_run, path_filter=path_filter)
-        log.info(f"Indexed {self.vault.num_notes()} notes from {self.vault.vault_path}")
+        if len(self.filters) == 0:
+            self.vault: Vault = Vault(config=self.config, dry_run=self.dry_run)
+        else:
+            self.vault = Vault(config=self.config, dry_run=self.dry_run, filters=self.filters)
+
+        alerts.success(
+            f"Loaded {len(self.vault.notes_in_scope)} notes from {len(self.vault.all_notes)} total notes"
+        )
         self.questions = Questions(vault=self.vault)
 
     def rename_key(self) -> None:

src/obsidian_metadata/models/metadata.py

@@ -31,8 +31,8 @@ class VaultMetadata:
         """Representation of all metadata."""
         return str(self.dict)
 
-    def add_metadata(self, metadata: dict[str, list[str]]) -> None:
-        """Add metadata to the vault. Takes a dictionary as input and merges it with the existing metadata.  Does not overwrite existing keys.
+    def index_metadata(self, metadata: dict[str, list[str]]) -> None:
+        """Index pre-existing metadata in the vault. Takes a dictionary as input and merges it with the existing metadata.  Does not overwrite existing keys.
 
         Args:
             metadata (dict): Metadata to add.

src/obsidian_metadata/models/questions.py

@@ -64,13 +64,13 @@ class Questions:
         """
         self.style = questionary.Style(
             [
-                ("qmark", "fg:#808080 bold"),
-                ("question", "bold"),
+                ("qmark", "fg:#729fcf bold"),
+                ("question", "fg:#729fcf bold"),
                 ("separator", "fg:#808080"),
                 ("instruction", "fg:#808080"),
-                ("highlighted", "fg:#c0c0c0 bold reverse"),
+                ("highlighted", "fg:#729fcf bold underline"),
                 ("text", ""),
-                ("pointer", "bold"),
+                ("pointer", "fg:#729fcf bold"),
             ]
         )
         self.vault = vault
@@ -174,6 +174,20 @@ class Questions:
 
         return True
 
+    def _validate_number(self, text: str) -> bool | str:
+        """Validate a number.
+
+        Args:
+            text (str): The number to validate.
+
+        Returns:
+            bool | str: True if the number is valid, otherwise a string with the error message.
+        """
+        if not text.isdigit():
+            return "Must be an integer"
+
+        return True
+
     def _validate_valid_vault_regex(self, text: str) -> bool | str:
         """Validates a valid regex.
 
@@ -202,8 +216,8 @@ class Questions:
         Returns:
             bool | str: True if the value is valid, otherwise a string with the error message.
         """
-        if len(text) < 1:
-            return "Value cannot be empty"
+        if len(text) == 0:
+            return True
 
         if self.key is not None and not self.vault.metadata.contains(self.key, text):
             return f"{self.key}:{text} does not exist"
@@ -408,6 +422,19 @@ class Questions:
             question, validate=self._validate_new_value, style=self.style, qmark="INPUT    |"
         ).ask()
 
+    def ask_number(self, question: str = "Enter a number") -> int:
+        """Ask the user for a number.
+
+        Args:
+            question (str, optional): The question to ask. Defaults to "Enter a number".
+
+        Returns:
+            int: A number.
+        """
+        return questionary.text(
+            question, validate=self._validate_number, style=self.style, qmark="INPUT    |"
+        ).ask()
+
     def ask_selection(
         self, choices: list[Any], question: str = "Select an option"
     ) -> Any:  # pragma: no cover

src/obsidian_metadata/models/vault.py

@@ -2,6 +2,7 @@
 
 import re
 import shutil
+from dataclasses import dataclass
 from pathlib import Path
 
 import rich.repr
@@ -17,6 +18,16 @@ from obsidian_metadata._utils.alerts import logger as log
 from obsidian_metadata.models import MetadataType, Note, VaultMetadata
 
 
+@dataclass
+class VaultFilter:
+    """Vault filters."""
+
+    path_filter: str = None
+    key_filter: str = None
+    value_filter: str = None
+    tag_filter: str = None
+
+
 @rich.repr.auto
 class Vault:
     """Representation of the Obsidian vault.
@@ -28,7 +39,12 @@ class Vault:
         notes (list[Note]): List of all notes in the vault.
     """
 
-    def __init__(self, config: VaultConfig, dry_run: bool = False, path_filter: str = None):
+    def __init__(
+        self,
+        config: VaultConfig,
+        dry_run: bool = False,
+        filters: list[VaultFilter] = [],
+    ):
         self.vault_path: Path = config.path
         self.dry_run: bool = dry_run
         self.backup_path: Path = self.vault_path.parent / f"{self.vault_path.name}.bak"
@@ -37,8 +53,8 @@ class Vault:
         for p in config.exclude_paths:
             self.exclude_paths.append(Path(self.vault_path / p))
 
-        self.path_filter = path_filter
-        self.note_paths = self._find_markdown_notes(path_filter)
+        self.filters = filters
+        self.all_note_paths = self._find_markdown_notes()
 
         with Progress(
             SpinnerColumn(),
@@ -46,9 +62,10 @@ class Vault:
             transient=True,
         ) as progress:
             progress.add_task(description="Processing notes...", total=None)
-            self.notes: list[Note] = [
-                Note(note_path=p, dry_run=self.dry_run) for p in self.note_paths
+            self.all_notes: list[Note] = [
+                Note(note_path=p, dry_run=self.dry_run) for p in self.all_note_paths
             ]
+            self.notes_in_scope = self._filter_notes()
 
         self._rebuild_vault_metadata()
 
@@ -57,33 +74,54 @@ class Vault:
         yield "vault_path", self.vault_path
         yield "dry_run", self.dry_run
         yield "backup_path", self.backup_path
-        yield "num_notes", self.num_notes()
+        yield "num_notes", len(self.all_notes)
+        yield "num_notes_in_scope", len(self.notes_in_scope)
         yield "exclude_paths", self.exclude_paths
 
-    def _find_markdown_notes(self, path_filter: str = None) -> list[Path]:
-        """Build list of all markdown files in the vault.
+    def _filter_notes(self) -> list[Note]:
+        """Filter notes by path and metadata using the filters defined in self.filters.
 
-        Args:
-            path_filter (str, optional): Regex to filter notes by path.
+        Returns:
+            list[Note]: List of notes matching the filters.
+        """
+        notes_list = self.all_notes.copy()
+
+        for _filter in self.filters:
+            if _filter.path_filter is not None:
+                notes_list = [
+                    n
+                    for n in notes_list
+                    if re.search(_filter.path_filter, str(n.note_path.relative_to(self.vault_path)))
+                ]
+
+            if _filter.tag_filter is not None:
+                notes_list = [n for n in notes_list if n.contains_inline_tag(_filter.tag_filter)]
+
+            if _filter.key_filter is not None and _filter.value_filter is not None:
+                notes_list = [
+                    n
+                    for n in notes_list
+                    if n.contains_metadata(_filter.key_filter, _filter.value_filter)
+                ]
+            if _filter.key_filter is not None and _filter.value_filter is None:
+                notes_list = [n for n in notes_list if n.contains_metadata(_filter.key_filter)]
+
+        return notes_list
+
+    def _find_markdown_notes(self) -> list[Path]:
+        """Build list of all markdown files in the vault.
 
         Returns:
             list[Path]: List of paths to all matching files in the vault.
 
         """
-        notes_list = [
+        return [
             p.resolve()
             for p in self.vault_path.glob("**/*")
             if p.suffix in [".md", ".MD", ".markdown", ".MARKDOWN"]
             and not any(item in p.parents for item in self.exclude_paths)
         ]
 
-        if path_filter is not None:
-            notes_list = [
-                p for p in notes_list if re.search(path_filter, str(p.relative_to(self.vault_path)))
-            ]
-
-        return notes_list
-
     def _rebuild_vault_metadata(self) -> None:
         """Rebuild vault metadata."""
         self.metadata = VaultMetadata()
@@ -93,10 +131,12 @@ class Vault:
             transient=True,
         ) as progress:
             progress.add_task(description="Processing notes...", total=None)
-            for _note in self.notes:
-                self.metadata.add_metadata(_note.frontmatter.dict)
-                self.metadata.add_metadata(_note.inline_metadata.dict)
-                self.metadata.add_metadata({_note.inline_tags.metadata_key: _note.inline_tags.list})
+            for _note in self.notes_in_scope:
+                self.metadata.index_metadata(_note.frontmatter.dict)
+                self.metadata.index_metadata(_note.inline_metadata.dict)
+                self.metadata.index_metadata(
+                    {_note.inline_tags.metadata_key: _note.inline_tags.list}
+                )
 
     def add_metadata(self, area: MetadataType, key: str, value: str | list[str] = None) -> int:
         """Add metadata to all notes in the vault.
@@ -111,7 +151,7 @@ class Vault:
         """
         num_changed = 0
 
-        for _note in self.notes:
+        for _note in self.notes_in_scope:
             if _note.add_metadata(area, key, value):
                 num_changed += 1
 
@@ -153,7 +193,7 @@ class Vault:
         Returns:
             bool: True if tag is found in vault.
         """
-        return any(_note.contains_inline_tag(tag) for _note in self.notes)
+        return any(_note.contains_inline_tag(tag) for _note in self.notes_in_scope)
 
     def contains_metadata(self, key: str, value: str = None, is_regex: bool = False) -> bool:
         """Check if vault contains the given metadata.
@@ -193,7 +233,7 @@ class Vault:
         """
         num_changed = 0
 
-        for _note in self.notes:
+        for _note in self.notes_in_scope:
             if _note.delete_inline_tag(tag):
                 num_changed += 1
 
@@ -214,7 +254,7 @@ class Vault:
         """
         num_changed = 0
 
-        for _note in self.notes:
+        for _note in self.notes_in_scope:
             if _note.delete_metadata(key, value):
                 num_changed += 1
 
@@ -230,7 +270,7 @@ class Vault:
             list[Note]: List of notes that have changes.
         """
         changed_notes = []
-        for _note in self.notes:
+        for _note in self.notes_in_scope:
             if _note.has_changes():
                 changed_notes.append(_note)
 
@@ -245,36 +285,23 @@ class Vault:
             table.add_row("Backup path", str(self.backup_path))
         else:
             table.add_row("Backup", "None")
-        table.add_row("Notes in scope", str(self.num_notes()))
+        table.add_row("Notes in scope", str(len(self.notes_in_scope)))
         table.add_row("Notes excluded from scope", str(self.num_excluded_notes()))
-        table.add_row("Active path filter", str(self.path_filter))
-        table.add_row("Notes with updates", str(len(self.get_changed_notes())))
+        table.add_row("Active filters", str(len(self.filters)))
+        table.add_row("Notes with changes", str(len(self.get_changed_notes())))
 
         Console().print(table)
 
     def list_editable_notes(self) -> None:
         """Print a list of notes within the scope that are being edited."""
         table = Table(title="Notes in current scope", show_header=False, box=box.HORIZONTALS)
-        for _n, _note in enumerate(self.notes, start=1):
+        for _n, _note in enumerate(self.notes_in_scope, start=1):
             table.add_row(str(_n), str(_note.note_path.relative_to(self.vault_path)))
         Console().print(table)
 
     def num_excluded_notes(self) -> int:
         """Count number of excluded notes."""
-        excluded_notes = [
-            p.resolve()
-            for p in self.vault_path.glob("**/*")
-            if p.suffix in [".md", ".MD", ".markdown", ".MARKDOWN"] and p not in self.note_paths
-        ]
-        return len(excluded_notes)
-
-    def num_notes(self) -> int:
-        """Number of notes in the vault.
-
-        Returns:
-            int: Number of notes in the vault.
-        """
-        return len(self.notes)
+        return len(self.all_notes) - len(self.notes_in_scope)
 
     def rename_metadata(self, key: str, value_1: str, value_2: str = None) -> int:
         """Renames a key or key-value pair in the note's metadata.
@@ -291,7 +318,7 @@ class Vault:
         """
         num_changed = 0
 
-        for _note in self.notes:
+        for _note in self.notes_in_scope:
             if _note.rename_metadata(key, value_1, value_2):
                 num_changed += 1
 
@@ -312,7 +339,7 @@ class Vault:
         """
         num_changed = 0
 
-        for _note in self.notes:
+        for _note in self.notes_in_scope:
             if _note.rename_inline_tag(old_tag, new_tag):
                 num_changed += 1
 
@@ -325,6 +352,6 @@ class Vault:
         """Write changes to the vault."""
         log.debug("Writing changes to vault...")
         if self.dry_run is False:
-            for _note in self.notes:
+            for _note in self.notes_in_scope:
                 log.trace(f"writing to {_note.note_path}")
                 _note.write()