cgcardona / muse public
tag.py python
107 lines 3.4 KB
8d5137ed fix(security): full surface hardening — validation, path containment, p… Gabriel Cardona <cgcardona@gmail.com> 10h ago
1 """muse tag — attach and query semantic tags on commits.
2
3 Usage::
4
5 muse tag add emotion:joyful <commit> — tag a commit
6 muse tag list — list all tags in the repo
7 muse tag list <commit> — list tags on a specific commit
8 muse tag remove <tag> <commit> — remove a tag
9
10 Tag conventions::
11
12 emotion:* — emotional character (emotion:melancholic, emotion:tense)
13 section:* — song section (section:verse, section:chorus)
14 stage:* — production stage (stage:rough-mix, stage:master)
15 key:* — musical key (key:Am, key:Eb)
16 tempo:* — tempo annotation (tempo:120bpm)
17 ref:* — reference track (ref:beatles)
18 """
19
20 from __future__ import annotations
21
22 import json
23 import logging
24 import pathlib
25 import uuid
26
27 import typer
28
29 from muse.core.errors import ExitCode
30 from muse.core.repo import require_repo
31 from muse.core.store import (
32 TagRecord,
33 get_all_tags,
34 get_tags_for_commit,
35 resolve_commit_ref,
36 write_tag,
37 )
38 from muse.core.validation import sanitize_display
39
40 logger = logging.getLogger(__name__)
41
42 app = typer.Typer()
43 add_app = typer.Typer()
44 list_app = typer.Typer()
45 remove_app = typer.Typer()
46
47 app.add_typer(add_app, name="add", help="Attach a tag to a commit.")
48 app.add_typer(list_app, name="list", help="List tags.")
49 app.add_typer(remove_app, name="remove", help="Remove a tag from a commit.")
50
51
52 def _read_branch(root: pathlib.Path) -> str:
53 head_ref = (root / ".muse" / "HEAD").read_text().strip()
54 return head_ref.removeprefix("refs/heads/").strip()
55
56
57 def _read_repo_id(root: pathlib.Path) -> str:
58 return str(json.loads((root / ".muse" / "repo.json").read_text())["repo_id"])
59
60
61 @add_app.callback(invoke_without_command=True)
62 def add(
63 ctx: typer.Context,
64 tag_name: str = typer.Argument(..., help="Tag string (e.g. emotion:joyful)."),
65 ref: str | None = typer.Argument(None, help="Commit ID or branch (default: HEAD)."),
66 ) -> None:
67 """Attach a tag to a commit."""
68 root = require_repo()
69 repo_id = _read_repo_id(root)
70 branch = _read_branch(root)
71
72 commit = resolve_commit_ref(root, repo_id, branch, ref)
73 if commit is None:
74 typer.echo(f"❌ Commit '{ref}' not found.")
75 raise typer.Exit(code=ExitCode.USER_ERROR)
76
77 write_tag(root, TagRecord(
78 tag_id=str(uuid.uuid4()),
79 repo_id=repo_id,
80 commit_id=commit.commit_id,
81 tag=tag_name,
82 ))
83 typer.echo(f"Tagged {commit.commit_id[:8]} with '{sanitize_display(tag_name)}'")
84
85
86 @list_app.callback(invoke_without_command=True)
87 def list_tags(
88 ctx: typer.Context,
89 ref: str | None = typer.Argument(None, help="Commit ID to list tags for (default: all)."),
90 ) -> None:
91 """List tags."""
92 root = require_repo()
93 repo_id = _read_repo_id(root)
94 branch = _read_branch(root)
95
96 if ref:
97 commit = resolve_commit_ref(root, repo_id, branch, ref)
98 if commit is None:
99 typer.echo(f"❌ Commit '{ref}' not found.")
100 raise typer.Exit(code=ExitCode.USER_ERROR)
101 tags = get_tags_for_commit(root, repo_id, commit.commit_id)
102 for t in sorted(tags, key=lambda x: x.tag):
103 typer.echo(f"{t.commit_id[:8]} {sanitize_display(t.tag)}")
104 else:
105 tags = get_all_tags(root, repo_id)
106 for t in sorted(tags, key=lambda x: (x.tag, x.commit_id)):
107 typer.echo(f"{t.commit_id[:8]} {sanitize_display(t.tag)}")