m/obsidian-metadata
1
0
Fork 0
mirror of https://github.com/natelandau/obsidian-metadata.git synced 2025-11-18 09:53:40 -05:00
Code Issues Packages Projects Releases Wiki Activity

feat: bulk update metadata from a CSV file

Browse Source
This commit is contained in:
Nathaniel Landau
2023-03-20 00:16:19 -04:00
parent 593dbc3b55
commit d636fb2672
14 changed files with 521 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/obsidian_metadata/_utils/alerts.py
View File

@@ -87,13 +87,16 @@ def info(msg: str) -> None:
console.print(f"INFO | {msg}")
def usage(msg: str, width: int = 80) -> None:
def usage(msg: str, width: int = None) -> None:
"""Print a usage message without using logging.
Args:
msg: Message to print
width (optional): Width of the message
"""
if width is None:
width = console.width - 15
for _n, line in enumerate(wrap(msg, width=width)):
if _n == 0:
console.print(f"[dim]USAGE | {line}")

69
src/obsidian_metadata/cli.py
View File

@@ -91,74 +91,7 @@ def main(
[bold underline]Configuration:[/]
Configuration is specified in a configuration file. On First run, this file will be created at [tan]~/.{0}.env[/]. Any options specified on the command line will override the configuration file.
[bold underline]Usage:[/]
[tan]Obsidian-metadata[/] provides a menu of sub-commands.
[bold underline]Vault Actions[/]
Create or delete a backup of your vault.
• Backup: Create a backup of the vault.
• Delete Backup: Delete a backup of the vault.
[bold underline]Inspect Metadata[/]
Inspect the metadata in your vault.
• View all metadata in the vault
• View all frontmatter
• View all inline metadata
• View all inline tags
• Export all metadata to CSV or JSON file
[bold underline]Filter Notes in Scope[/]
Limit the scope of notes to be processed with one or more filters.
• Path filter (regex): Limit scope based on the path or filename
• Metadata filter: Limit scope based on a key or key/value pair
• Tag filter: Limit scope based on an in-text tag
• List and clear filters: List all current filters and clear one or all
• List notes in scope: List notes that will be processed.
[bold underline]Add Metadata[/]
Add new metadata to your vault.
• Add new metadata to the frontmatter
• Add new inline metadata - Set `insert_location` in the config to
control where the new metadata is inserted. (Default: Bottom)
• Add new inline tag - Set `insert_location` in the config to
control where the new tag is inserted. (Default: Bottom)
[bold underline]Rename Metadata[/]
Rename either a key and all associated values, a specific value within a key. or an in-text tag.
• Rename a key
• Rename a value
• rename an inline tag
[bold underline]Delete Metadata[/]
Delete either a key and all associated values, or a specific value.
• Delete a key and associated values
• Delete a value from a key
• Delete an inline tag
[bold underline]Move Inline Metadata[/]
Move inline metadata to a specified location with a note
• Move to Top - Move all inline metadata beneath the frontmatter
• Move to After Title* - Move all inline metadata beneath the first markdown header
• Move to Bottom - Move all inline metadata to the bottom of the note
[bold underline]Transpose Metadata[/]
Move metadata from inline to frontmatter or the reverse. When transposing to inline metadata,
the `insert location` value in the config file will specify where in the
note it will be inserted.
• Transpose all metadata - Moves all frontmatter to inline
metadata, or the reverse
• Transpose key - Transposes a specific key and all it's values
• Transpose value - Transpose a specific key:value pair
[bold underline]Review Changes[/]
Prior to committing changes, review all changes that will be made.
• View a diff of the changes that will be made
[bold underline]Commit Changes[/]
Write the changes to disk. This step is not undoable.
• Commit changes to the vault
Full usage information is available at https://github.com/natelandau/obsidian-metadata
"""
# Instantiate logger

105
src/obsidian_metadata/models/application.py
View File

@@ -1,6 +1,7 @@
"""Questions for the cli."""
import csv
from pathlib import Path
from typing import Any
@@ -53,8 +54,12 @@ class Application:
match self.questions.ask_application_main():
case "vault_actions":
self.application_vault()
case "export_metadata":
self.application_export_metadata()
case "inspect_metadata":
self.application_inspect_metadata()
case "import_from_csv":
self.application_import_csv()
case "filter_notes":
self.application_filter()
case "add_metadata":
@@ -124,6 +129,7 @@ class Application:
alerts.usage("Delete either a key and all associated values, or a specific value.")
choices = [
questionary.Separator(),
{"name": "Delete inline tag", "value": "delete_inline_tag"},
{"name": "Delete key", "value": "delete_key"},
{"name": "Delete value", "value": "delete_value"},
@@ -147,6 +153,7 @@ class Application:
alerts.usage("Select the type of metadata to rename.")
choices = [
questionary.Separator(),
{"name": "Rename inline tag", "value": "rename_inline_tag"},
{"name": "Rename key", "value": "rename_key"},
{"name": "Rename value", "value": "rename_value"},
@@ -170,6 +177,7 @@ class Application:
alerts.usage("Limit the scope of notes to be processed with one or more filters.")
choices = [
questionary.Separator(),
{"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"},
@@ -276,6 +284,82 @@ class Application:
case _:
return
def application_import_csv(self) -> None:
"""Import CSV for bulk changes to metadata."""
alerts.usage(
"Import CSV to make build changes to metadata. The CSV must have the following columns: path, type, key, value. Where type is one of 'frontmatter', 'inline_metadata', or 'tag'. Note: this will not create new notes."
)
path = self.questions.ask_path(question="Enter path to a CSV file", valid_file=True)
if path is None:
return
csv_path = Path(path).expanduser()
if "csv" not in csv_path.suffix.lower():
alerts.error("File must be a CSV file")
return
csv_dict: dict[str, Any] = {}
with csv_path.open("r") as csv_file:
csv_reader = csv.DictReader(csv_file, delimiter=",")
for row in csv_reader:
if row["path"] not in csv_dict:
csv_dict[row["path"]] = []
csv_dict[row["path"]].append(
{"type": row["type"], "key": row["key"], "value": row["value"]}
)
num_changed = self.vault.update_from_dict(csv_dict)
if num_changed == 0:
alerts.warning("No notes were changed")
return
alerts.success(f"Rewrote metadata for {num_changed} notes.")
def application_export_metadata(self) -> None:
"""Export metadata to various formats."""
alerts.usage(
"Export the metadata in your vault. Note, uncommitted changes will be reflected in these files. The notes csv export can be used as template for importing bulk changes"
)
choices = [
questionary.Separator(),
{"name": "Metadata by type to CSV", "value": "export_csv"},
{"name": "Metadata by type to JSON", "value": "export_json"},
{
"name": "Metadata by note to CSV [Bulk import template]",
"value": "export_notes_csv",
},
questionary.Separator(),
{"name": "Back", "value": "back"},
]
while True:
match self.questions.ask_selection(choices=choices, question="Export format"):
case "export_csv":
path = self.questions.ask_path(question="Enter a path for the CSV file")
if path is None:
return
self.vault.export_metadata(path=path, export_format="csv")
alerts.success(f"CSV written to {path}")
case "export_json":
path = self.questions.ask_path(question="Enter a path for the JSON file")
if path is None:
return
self.vault.export_metadata(path=path, export_format="json")
alerts.success(f"JSON written to {path}")
case "export_notes_csv":
path = self.questions.ask_path(question="Enter a path for the CSV file")
if path is None:
return
self.vault.export_notes_to_csv(path=path)
alerts.success(f"CSV written to {path}")
return
case _:
return
def application_inspect_metadata(self) -> None:
"""View metadata."""
alerts.usage(
@@ -283,19 +367,17 @@ class Application:
)
choices = [
questionary.Separator(),
{"name": "View all frontmatter", "value": "all_frontmatter"},
{"name": "View all inline metadata", "value": "all_inline"},
{"name": "View all inline tags", "value": "all_tags"},
{"name": "View all keys", "value": "all_keys"},
{"name": "View all metadata", "value": "all_metadata"},
questionary.Separator(),
{"name": "Write all metadata to CSV", "value": "export_csv"},
{"name": "Write all metadata to JSON file", "value": "export_json"},
questionary.Separator(),
{"name": "Back", "value": "back"},
]
while True:
match self.questions.ask_selection(choices=choices, question="Select a vault action"):
match self.questions.ask_selection(choices=choices, question="Select an action"):
case "all_metadata":
console.print("")
self.vault.metadata.print_metadata(area=MetadataType.ALL)
@@ -316,18 +398,6 @@ class Application:
console.print("")
self.vault.metadata.print_metadata(area=MetadataType.TAGS)
console.print("")
case "export_csv":
path = self.questions.ask_path(question="Enter a path for the CSV file")
if path is None:
return
self.vault.export_metadata(path=path, export_format="csv")
alerts.success(f"Metadata written to {path}")
case "export_json":
path = self.questions.ask_path(question="Enter a path for the JSON file")
if path is None:
return
self.vault.export_metadata(path=path, export_format="json")
alerts.success(f"Metadata written to {path}")
case _:
return
@@ -342,6 +412,7 @@ class Application:
alerts.usage(" 2. Move the location of inline metadata within a note.")
choices = [
questionary.Separator(),
{"name": "Move inline metadata to top of note", "value": "move_to_top"},
{
"name": "Move inline metadata beneath the first header",
@@ -374,6 +445,7 @@ class Application:
alerts.usage("Create or delete a backup of your vault.")
choices = [
questionary.Separator(),
{"name": "Backup vault", "value": "backup_vault"},
{"name": "Delete vault backup", "value": "delete_backup"},
questionary.Separator(),
@@ -564,6 +636,7 @@ class Application:
alerts.info(f"Found {len(changed_notes)} changed notes in the vault")
choices: list[dict[str, Any] | questionary.Separator] = []
choices.append(questionary.Separator())
for n, note in enumerate(changed_notes, start=1):
_selection = {
"name": f"{n}: {note.note_path.relative_to(self.vault.vault_path)}",

7
src/obsidian_metadata/models/metadata.py
View File

@@ -245,6 +245,9 @@ class Frontmatter:
except Exception as e: # noqa: BLE001
raise AttributeError(e) from e
if frontmatter is None or frontmatter == [None]:
return {}
for k in frontmatter:
if frontmatter[k] is None:
frontmatter[k] = []
@@ -326,6 +329,10 @@ class Frontmatter:
return False
def delete_all(self) -> None:
"""Delete all Frontmatter from the note."""
self.dict = {}
def has_changes(self) -> bool:
"""Check if the frontmatter has changes.

11
src/obsidian_metadata/models/notes.py
View File

@@ -190,6 +190,17 @@ class Note:
return False
def delete_all_metadata(self) -> None:
"""Delete all metadata from the note. Removes all frontmatter and inline metadata and tags from the body of the note and from the associated metadata objects."""
for key in self.inline_metadata.dict:
self.delete_metadata(key=key, area=MetadataType.INLINE)
for tag in self.inline_tags.list:
self.delete_inline_tag(tag=tag)
self.frontmatter.delete_all()
self.write_frontmatter()
def delete_inline_tag(self, tag: str) -> bool:
"""Delete an inline tag from the `inline_tags` attribute AND removes the tag from the text of the note if it exists.

34
src/obsidian_metadata/models/questions.py
View File

@@ -200,6 +200,23 @@ class Questions:
return True
def _validate_path_is_file(self, text: str) -> bool | str:
"""Validate a path is a file.
Args:
text (str): The path to validate.
Returns:
bool | str: True if the path is valid, otherwise a string with the error message.
"""
path_to_validate: Path = Path(text).expanduser().resolve()
if not path_to_validate.exists():
return f"Path does not exist: {path_to_validate}"
if not path_to_validate.is_file():
return f"Path is not a file: {path_to_validate}"
return True
def _validate_valid_vault_regex(self, text: str) -> bool | str:
"""Validate a valid regex.
@@ -276,9 +293,11 @@ class Questions:
choices=[
questionary.Separator("-------------------------------"),
{"name": "Vault Actions", "value": "vault_actions"},
{"name": "Export Metadata", "value": "export_metadata"},
{"name": "Inspect Metadata", "value": "inspect_metadata"},
{"name": "Filter Notes in Scope", "value": "filter_notes"},
questionary.Separator("-------------------------------"),
{"name": "Bulk changes from imported CSV", "value": "import_from_csv"},
{"name": "Add Metadata", "value": "add_metadata"},
{"name": "Delete Metadata", "value": "delete_metadata"},
{"name": "Rename Metadata", "value": "rename_metadata"},
@@ -475,15 +494,27 @@ class Questions:
question, validate=self._validate_number, style=self.style, qmark="INPUT |"
).ask()
def ask_path(self, question: str = "Enter a path") -> str: # pragma: no cover
def ask_path(
self, question: str = "Enter a path", valid_file: bool = False
) -> str: # pragma: no cover
"""Ask the user for a path.
Args:
question (str, optional): The question to ask. Defaults to "Enter a path".
valid_file (bool, optional): Whether the path should be a valid file. Defaults to False.
Returns:
str: A path.
"""
if valid_file:
return questionary.path(
question,
only_directories=False,
style=self.style,
validate=self._validate_path_is_file,
qmark="INPUT |",
).ask()
return questionary.path(question, style=self.style, qmark="INPUT |").ask()
def ask_selection(
@@ -498,7 +529,6 @@ class Questions:
Returns:
any: The selected item value.
"""
choices.insert(0, questionary.Separator())
return questionary.select(
question,
choices=choices,

91
src/obsidian_metadata/models/vault.py
View File

@@ -6,6 +6,7 @@ import re
import shutil
from dataclasses import dataclass
from pathlib import Path
from typing import Any
import rich.repr
import typer
@@ -360,6 +361,44 @@ class Vault:
with export_file.open(mode="w", encoding="UTF8") as f:
json.dump(dict_to_dump, f, indent=4, ensure_ascii=False, sort_keys=True)
def export_notes_to_csv(self, path: str) -> None:
"""Export notes and their associated metadata to a csv file. This is useful as a template for importing metadata changes to a vault.
Args:
path (str): Path to write csv file to.
"""
export_file = Path(path).expanduser().resolve()
if not export_file.parent.exists():
alerts.error(f"Path does not exist: {export_file.parent}")
raise typer.Exit(code=1)
with export_file.open(mode="w", encoding="UTF8") as f:
writer = csv.writer(f)
writer.writerow(["path", "type", "key", "value"])
for _note in self.all_notes:
for key, value in _note.frontmatter.dict.items():
for v in value:
writer.writerow(
[_note.note_path.relative_to(self.vault_path), "frontmatter", key, v]
)
for key, value in _note.inline_metadata.dict.items():
for v in value:
writer.writerow(
[
_note.note_path.relative_to(self.vault_path),
"inline_metadata",
key,
v,
]
)
for tag in _note.inline_tags.list:
writer.writerow(
[_note.note_path.relative_to(self.vault_path), "tag", "", f"{tag}"]
)
def get_changed_notes(self) -> list[Note]:
"""Return a list of notes that have changes.
@@ -510,3 +549,55 @@ class Vault:
self._rebuild_vault_metadata()
return num_changed
def update_from_dict(self, dictionary: dict[str, Any]) -> int:
"""Update note metadata from a dictionary. This is a destructive operation. All metadata in the specified notes not in the dictionary will be removed.
Requires a dictionary with the note path as the key and a dictionary of metadata as the value. Each key must have a list of associated dictionaries in the following format:
{
'type': 'frontmatter|inline_metadata|tag',
'key': 'string',
'value': 'string'
}
Args:
dictionary (dict[str, Any]): Dictionary to update metadata from.
Returns:
int: Number of notes that had metadata updated.
"""
num_changed = 0
for _note in self.all_notes:
path = _note.note_path.relative_to(self.vault_path)
if str(path) in dictionary:
log.debug(f"Updating metadata for {path}")
num_changed += 1
_note.delete_all_metadata()
for row in dictionary[str(path)]:
if row["type"].lower() == "frontmatter":
_note.add_metadata(
area=MetadataType.FRONTMATTER, key=row["key"], value=row["value"]
)
if row["type"].lower() == "inline_metadata":
_note.add_metadata(
area=MetadataType.INLINE,
key=row["key"],
value=row["value"],
location=self.insert_location,
)
if row["type"].lower() == "tag" or row["type"].lower() == "tags":
console.print(f"Adding tag {row['value']}")
_note.add_metadata(
area=MetadataType.TAGS,
value=row["value"],
location=self.insert_location,
)
if num_changed > 0:
self._rebuild_vault_metadata()
return num_changed