Coverage for tarjinja/_cli.py: 38%
184 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-05 14:24 +0000
1import functools
2import json
3import os
4import string
5import subprocess
6import sys
7import tempfile
8from logging import DEBUG, INFO, basicConfig, getLogger
10import click
11import requests
12import yaml
14from .choice import detect_input, detect_output, filter_items, input_items, output_items
15from .iface import Pipeline
16from .multifilter import MultiFilter
17from .version import VERSION
19log = getLogger(__name__)
22@click.version_option(version=VERSION, prog_name="tarjinja")
23@click.group(invoke_without_command=True)
24@click.pass_context
25def cli(ctx):
26 if ctx.invoked_subcommand is None:
27 print(ctx.get_help())
30def set_verbose(flag):
31 fmt = "%(asctime)s %(levelname)s %(message)s"
32 if flag:
33 basicConfig(level=DEBUG, format=fmt)
34 else:
35 basicConfig(level=INFO, format=fmt)
38_cli_option = [
39 click.option("--verbose/--no-verbose"),
40]
42_value_option = [
43 click.option("--value-from", type=click.File("r")),
44 click.option("--value", default="{}", type=str),
45 click.option("--gitconfig/--no-gitconfig"),
46 click.option("--github-user/--no-github-user"),
47]
49_inout_option = [
50 click.option("--input-args", default="{}", type=str),
51 click.argument("input", type=click.Path(), required=True),
52 click.argument("output", type=click.Path(), required=True),
53]
56def multi_options(decs):
57 def deco(f):
58 for dec in reversed(decs):
59 f = dec(f)
60 return f
62 return deco
65def cli_option(func):
66 @functools.wraps(func)
67 def wrap(verbose, *args, **kwargs):
68 set_verbose(verbose)
69 return func(*args, **kwargs)
71 return multi_options(_cli_option)(wrap)
74def value_option(func):
75 @functools.wraps(func)
76 def wrap(value, value_from, gitconfig, github_user, *args, **kwargs):
77 vals = {}
78 if value_from:
79 vals.update(yaml.load(value_from, Loader=yaml.FullLoader))
80 if value:
81 vals.update(json.loads(value))
82 if gitconfig:
83 p = subprocess.run(
84 ["git", "config", "-l"],
85 check=False,
86 stdout=subprocess.PIPE,
87 encoding="UTF-8",
88 stdin=subprocess.DEVNULL,
89 )
90 for line in p.stdout.split("\n"):
91 if "=" not in line:
92 continue
93 k, v = line.strip().split("=", 1)
94 v = v.strip()
95 if not k.startswith("user."):
96 continue
97 if v in ["true", "false"] + list(string.digits):
98 continue
99 k = "git_" + k.replace(".", "_")
100 if len(v) != 0:
101 vals[k] = v
102 if github_user:
103 p = subprocess.run(
104 ["hub", "api", "user"],
105 check=False,
106 stdout=subprocess.PIPE,
107 encoding="UTF-8",
108 stdin=subprocess.DEVNULL,
109 )
110 data = json.loads(p.stdout)
111 for k, v in data.items():
112 if isinstance(v, str) and v != "":
113 vals["github_" + k] = v
114 return func(*args, value=vals, **kwargs)
116 return multi_options(_value_option)(wrap)
119def inout_option(func):
120 @functools.wraps(func)
121 def wrap(input, output, input_args, *args, **kwargs):
122 if "://" in input and ".git" not in input:
123 tmpd = tempfile.TemporaryDirectory()
124 tmpfn = os.path.join(tmpd.name, os.path.basename(input))
125 with open(tmpfn, "wb") as ofp:
126 ofp.write(requests.get(input).content)
127 input = tmpfn
128 return func(
129 *args,
130 input=input,
131 output=output,
132 input_args=json.loads(input_args),
133 **kwargs,
134 )
136 return multi_options(_inout_option)(wrap)
139def do_pipe(
140 in_type,
141 out_type,
142 input,
143 output,
144 filter_type,
145 vals,
146 thru=None,
147 notag=False,
148 input_args=None,
149):
150 if input_args is None:
151 input_args = {}
152 log.debug(
153 "input: %s (%s), output: %s (%s), filter: %s, input_args: %s",
154 input,
155 in_type,
156 output,
157 out_type,
158 filter_type,
159 input_args,
160 )
161 if isinstance(in_type, str):
162 input_val = dict(input_items()).get(in_type)(input, **input_args)
163 else:
164 input_val = in_type(input, **input_args)
165 if isinstance(out_type, str):
166 output_val = dict(output_items()).get(out_type)(output)
167 else:
168 output_val = out_type(output)
169 if isinstance(filter_type, (list, tuple)):
170 if len(filter_type) == 1:
171 filter_val = dict(filter_items()).get(filter_type[0])()
172 else:
173 filter_val = MultiFilter()
174 for f in filter_type:
175 filter_val.add_filter(dict(filter_items()).get(f)())
176 else:
177 filter_val = dict(filter_items()).get(filter_type)()
178 if notag:
179 filter_val.tag_escape = {}
180 pipeline = Pipeline(input_val, filter_val, output_val, thru)
181 pipeline.render(vals)
184@cli.command()
185@cli_option
186@value_option
187@click.option("--out-type", type=click.Choice(dict(output_items())), required=True)
188@click.option("--in-type", type=click.Choice(dict(input_items())), required=True)
189@click.option("--filter-type", type=click.Choice(dict(filter_items())), multiple=True)
190@click.option("--input-args", type=str, default="{}")
191@click.option("--thru", type=str, default=None)
192@click.argument("input", type=click.Path(), required=True)
193@click.argument("output", type=click.Path(), required=True)
194def copy(in_type, out_type, filter_type, input, output, value, thru, input_args):
195 iarg = json.loads(input_args)
196 log.debug("input_args: %s", iarg)
197 do_pipe(in_type, out_type, input, output, filter_type, value, thru, False, iarg)
200@cli.command()
201@cli_option
202@value_option
203@click.option("--filter-type", type=click.Choice(dict(filter_items())), default="Jinja")
204@inout_option
205def tarc(output, input, input_args, value, filter_type):
206 out_type = detect_output(output, "Tar")
207 in_type = detect_input(input, "Dir")
208 do_pipe(in_type, out_type, input, output, filter_type, value, input_args=input_args)
211@cli.command()
212@cli_option
213@click.option("--filter-type", type=click.Choice(dict(filter_items())), default="Jinja")
214@value_option
215@click.option("--verbose/--no-verbose")
216@inout_option
217def tarx(output, input, input_args, value, filter_type):
218 out_type = detect_output(output, "Dir")
219 in_type = detect_input(input, "Tar")
220 do_pipe(in_type, out_type, input, output, filter_type, value, input_args=input_args)
223@cli.command()
224@cli_option
225@click.option("--filter-type", type=click.Choice(dict(filter_items())), default="Jinja")
226@value_option
227@click.option("--dry/--no-dry")
228@click.option("--skiptag/--no-skiptag", default=False)
229@inout_option
230def rsync(output, input, input_args, value, filter_type, dry, skiptag):
231 log.info("input %s, output %s", input, output)
232 in_type = detect_input(input, "Single")
233 if dry:
234 out_type = "List"
235 else:
236 out_type = detect_output(output, "Dir")
237 do_pipe(
238 in_type,
239 out_type,
240 input,
241 output,
242 filter_type,
243 value,
244 None,
245 skiptag,
246 input_args=input_args,
247 )
250@cli.command()
251@cli_option
252@value_option
253@click.option("--filter-type", type=click.Choice(dict(filter_items())), default="Jinja")
254@click.option("--dry/--no-dry")
255@click.argument("input", type=click.Path())
256@click.argument("output", type=click.File("w"), default=sys.stdout)
257def var_names(input, output, value, filter_type, dry):
258 in_type = detect_input(input, "Single")
259 input_val = dict(input_items()).get(in_type)(input)
260 flt = dict(filter_items()).get(filter_type)()
261 assert hasattr(flt, "var_names")
262 res = set()
263 for fnpat, mode, ts in input_val.walk():
264 res.update(flt.var_names(fnpat))
265 content = input_val.readfile(fnpat)
266 res.update(flt.var_names(content))
267 log.debug("%s", res)
268 if dry:
269 json.dump(list(filter(lambda f: f not in value, res)), fp=output)
270 return
271 vars = {}
272 for i in res:
273 if i not in value:
274 vars[i] = click.prompt(i, type=str)
275 json.dump(vars, fp=output)
278@cli.command()
279@cli_option
280@value_option
281def dump_value(value):
282 json.dump(value, sys.stdout)
285if __name__ == "__main__": 285 ↛ 286line 285 didn't jump to line 286 because the condition on line 285 was never true
286 cli()