FilesExpand file tree

master

/

hook.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

91 lines (66 loc) · 2.92 KB

master

/

hook.py

Copy path

Top

File metadata and controls

• Code
• Blame

91 lines (66 loc) · 2.92 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

"""Allows custom commands to be run when an event is emitted by beets"""

from __future__ import annotations

import os

import shlex

import string

import subprocess

from typing import TYPE_CHECKING, Any

from beets.events import ALL_EVENTS

from beets.plugins import BeetsPlugin

if TYPE_CHECKING:

from beets.events import EventType

def _split_command(command: str) -> list[str]:

"""Split a hook command into argv, treating '\\' as a path separator on Windows."""

lex = shlex.shlex(command, posix=True)

lex.whitespace_split = True

lex.commenters = "" # shlex.split does this; raw shlex defaults to '#'

if os.name == "nt":

lex.escape = ""

return list(lex)

class BytesToStrFormatter(string.Formatter):

"""A variant of `string.Formatter` that converts `bytes` to `str`."""

def convert_field(self, value: Any, conversion: str | None) -> Any:

"""Converts the provided value given a conversion type.

This method decodes the converted value using the formatter's coding.

"""

converted = super().convert_field(value, conversion)

if isinstance(converted, bytes):

return os.fsdecode(converted)

return converted

class HookPlugin(BeetsPlugin):

"""Allows custom commands to be run when an event is emitted by beets"""

def __init__(self) -> None:

super().__init__()

self.config.add({"hooks": []})

hooks = self.config["hooks"].get(list)

for hook_index in range(len(hooks)):

hook = self.config["hooks"][hook_index]

hook_event: EventType = hook["event"].as_choice(choices=ALL_EVENTS)

hook_command = hook["command"].as_str()

self.create_and_register_hook(hook_event, hook_command)

def create_and_register_hook(self, event: EventType, command: str) -> None:

def hook_function(**kwargs) -> None:

if command is None or len(command) == 0:

self._log.error('invalid command "{}"', command)

return

# For backwards compatibility, use a string formatter that decodes

# bytes (in particular, paths) to strings.

formatter = BytesToStrFormatter()

command_pieces = [

formatter.format(piece, event=event, **kwargs)

for piece in _split_command(command)

]

self._log.debug(

'running command "{}" for event {}',

" ".join(command_pieces),

event,

)

try:

subprocess.check_call(command_pieces)

except subprocess.CalledProcessError as exc:

self._log.error(

"hook for {} exited with status {.returncode}", event, exc

)

except OSError as exc:

self._log.error("hook for {} failed: {}", event, exc)

self.register_listener(event, hook_function) # type: ignore[arg-type]