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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
|
# -*- coding: utf-8 -*-
# Copyright 2023 Mike Fährmann
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
""" """
import re
import logging
import operator
from . import util, exception
def parse(actionspec):
if isinstance(actionspec, dict):
actionspec = actionspec.items()
actions = {}
actions[logging.DEBUG] = actions_d = []
actions[logging.INFO] = actions_i = []
actions[logging.WARNING] = actions_w = []
actions[logging.ERROR] = actions_e = []
for event, spec in actionspec:
level, _, pattern = event.partition(":")
type, _, args = spec.partition(" ")
action = (re.compile(pattern).search, ACTIONS[type](args))
level = level.strip()
if not level or level == "*":
actions_d.append(action)
actions_i.append(action)
actions_w.append(action)
actions_e.append(action)
else:
actions[_level_to_int(level)].append(action)
return actions
def _level_to_int(level):
try:
return logging._nameToLevel[level]
except KeyError:
return int(level)
def action_print(opts):
def _print(_):
print(opts)
return _print
def action_status(opts):
op, value = re.match(r"\s*([&|^=])=?\s*(\d+)", opts).groups()
op = {
"&": operator.and_,
"|": operator.or_,
"^": operator.xor,
"=": lambda x, y: y,
}[op]
value = int(value)
def _status(args):
args["job"].status = op(args["job"].status, value)
return _status
def action_level(opts):
level = _level_to_int(opts.lstrip(" ~="))
def _level(args):
args["level"] = level
return _level
def action_wait(opts):
def _wait(args):
input("Press Enter to continue")
return _wait
def action_restart(opts):
return util.raises(exception.RestartExtraction)
def action_exit(opts):
try:
opts = int(opts)
except ValueError:
pass
def _exit(args):
raise SystemExit(opts)
return _exit
ACTIONS = {
"print" : action_print,
"status" : action_status,
"level" : action_level,
"restart": action_restart,
"wait" : action_wait,
"exit" : action_exit,
}
|
