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
|
# -*- coding: utf-8 -*-
# Copyright 2018-2020 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.
"""Execute processes"""
from .common import PostProcessor
from .. import util
import subprocess
if util.WINDOWS:
def quote(s):
return '"' + s.replace('"', '\\"') + '"'
else:
from shlex import quote
class ExecPP(PostProcessor):
def __init__(self, job, options):
PostProcessor.__init__(self, job)
if options.get("async", False):
self._exec = self._exec_async
args = options["command"]
if isinstance(args, str):
self.args = args
execute = self.exec_string
else:
self.args = [util.Formatter(arg) for arg in args]
execute = self.exec_list
events = options.get("event")
if events is None:
events = ("after",)
if options.get("final"):
self.log.warning("'final' is deprecated, "
"use '\"event\": \"finalize\"' instead")
events = ("finalize",)
elif isinstance(events, str):
events = events.split(",")
for event in events:
job.hooks[event].append(execute)
def exec_list(self, pathfmt, status=None):
if status:
return
kwdict = pathfmt.kwdict
kwdict["_directory"] = pathfmt.realdirectory
kwdict["_filename"] = pathfmt.filename
kwdict["_path"] = pathfmt.realpath
args = [arg.format_map(kwdict) for arg in self.args]
self._exec(args, False)
def exec_string(self, pathfmt, status=None):
if status:
return
if status is None and pathfmt.realpath:
args = self.args.replace("{}", quote(pathfmt.realpath))
else:
args = self.args.replace("{}", quote(pathfmt.realdirectory))
self._exec(args, True)
def _exec(self, args, shell):
self.log.debug("Running '%s'", args)
retcode = subprocess.Popen(args, shell=shell).wait()
if retcode:
self.log.warning("'%s' returned with non-zero exit status (%d)",
args, retcode)
def _exec_async(self, args, shell):
self.log.debug("Running '%s'", args)
subprocess.Popen(args, shell=shell)
__postprocessor__ = ExecPP
|
