Coverage for texgit / run.py: 89%
149 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-02-22 02:50 +0000
1"""Process a LaTeX aux file."""
2import argparse
3from os.path import dirname, getsize
4from typing import Final, Generator
6from pycommons.io.arguments import make_argparser, make_epilog
7from pycommons.io.console import logger
8from pycommons.io.path import Path, directory_path, write_lines
9from pycommons.strings.string_tools import escape, unescape
11from texgit.repository.git_manager import GitPath
12from texgit.repository.process_manager import ProcessManager
13from texgit.version import __version__
15#: the header for git file requests
16REQUEST_GIT_FILE: Final[str] = r"\@texgit@gitFile"
17#: the header for argument file requests
18REQUEST_ARG_FILE: Final[str] = r"\@texgit@argFile"
19#: the header for process result requests
20REQUEST_PROCESS: Final[str] = r"\@texgit@process"
21#: the forbidden line marker that needs to be purged
22FORBIDDEN_LINE: Final[str] = r"\@texgit@needsTexgitPass"
24#: the replacements
25__REPL: Final[dict[str, str]] = {
26 r"\\": "\\", r"\{": "{", "{{": "{", r"\}": "}", "}}": "}", r"\ ": " ",
27}
30def __get_request(line: str) -> list[str | None] | None:
31 r"""
32 Get the repository request, if any.
34 :param line: the line
35 :return: the request, composed of the request function, the repository
36 (if any), the path (if any), and the optional command; or `None` if
37 no request was found.
39 >>> print(__get_request(""))
40 None
41 >>> print(__get_request(r"\hello"))
42 None
43 >>> print(__get_request(r"\@texgit@gitFile{x}{y}{}"))
44 ['\\@texgit@gitFile', 'x', 'y', None]
45 >>> print(__get_request(r"\@texgit@process{x}{y}{python3 --version}"))
46 ['\\@texgit@process', 'x', 'y', 'python3', '--version']
47 >>> print(__get_request(r"\@texgit@gitFile{x}{y}{a}"))
48 ['\\@texgit@gitFile', 'x', 'y', 'a']
49 >>> print(__get_request(r"\@texgit@gitFile{x}{y}{a b}"))
50 ['\\@texgit@gitFile', 'x', 'y', 'a', 'b']
51 >>> print(__get_request(r"\@texgit@gitFile{x}{y}{a\ b}"))
52 ['\\@texgit@gitFile', 'x', 'y', 'a b']
53 >>> print(__get_request(r"\@texgit@gitFile{x{{y}{y}{a\ b}"))
54 ['\\@texgit@gitFile', 'x{y', 'y', 'a b']
55 >>> print(__get_request(r"\@texgit@gitFile{x\{y}{y}{a\ b}"))
56 ['\\@texgit@gitFile', 'x{y', 'y', 'a b']
57 >>> print(__get_request(r"\@texgit@gitFile{x\{y}{}}y}{a\ b}"))
58 ['\\@texgit@gitFile', 'x{y', '}y', 'a b']
59 >>> print(__get_request(r"\@texgit@gitFile{x\{y}{}}y}{a\ \\b}"))
60 ['\\@texgit@gitFile', 'x{y', '}y', 'a \\b']
61 >>> print(__get_request(r"\@texgit@gitFile {x\{y}{}}y}{a\ \\b}"))
62 ['\\@texgit@gitFile', 'x{y', '}y', 'a \\b']
63 >>> print(__get_request(
64 ... r" \@texgit@gitFile { x\{y}{ }}y }{ a\ \\b } "))
65 ['\\@texgit@gitFile', 'x{y', '}y', 'a \\b']
66 >>> print(__get_request(
67 ... r" \@texgit@argFile { x\{y}{ }}y }{ a\ \\b } {xx} {y }"))
68 ['\\@texgit@argFile', 'x{y', '}y', 'a \\b', 'xx', 'y']
69 """
70 use_line = str.strip(line)
71 if str.__len__(use_line) >= 67108864:
72 raise ValueError(f"line is {len(use_line)} characters long?")
74 request: Final[str | None] = REQUEST_GIT_FILE if str.startswith(
75 use_line, REQUEST_GIT_FILE) else (REQUEST_ARG_FILE if str.startswith(
76 use_line, REQUEST_ARG_FILE) else (
77 REQUEST_PROCESS if str.startswith(use_line, REQUEST_PROCESS)
78 else None))
79 if request is None:
80 return None
81 use_line = str.strip(use_line[str.__len__(request):])
82 if (str.__len__(use_line) <= 0) or (use_line[0] != "{"):
83 raise ValueError(
84 f"rest line={use_line!r} for {request!r} in {line!r}.")
86 # find markers for search-replacing problematic chars
87 use_line, esc = escape(use_line, __REPL.keys())
89 # Now we collect all the arguments
90 command: list[str | None] = [request]
91 idx_0: int = 1
92 while True:
93 idx_1: int = use_line.find("}", idx_0)
94 if idx_1 < idx_0:
95 raise ValueError(f"Found {{ but no }} in {line!r}?")
96 arg: str = str.strip(use_line[idx_0:idx_1])
97 if arg:
98 for argi in str.split(arg):
99 argj = str.strip(argi)
100 if argj:
101 argj = unescape(argj, esc)
102 for k, v in __REPL.items():
103 argj = str.replace(argj, k, v)
104 command.append(argj)
105 else:
106 command.append(None)
107 idx_0 = use_line.find("{", idx_1 + 1)
108 if idx_0 <= idx_1:
109 break
110 idx_0 += 1
111 return command
114#: the response header for the path
115RESPONSE_PATH: Final[str] = "@texgit@path@"
116#: the response header for the file name
117RESPONSE_NAME: Final[str] = "@texgit@name@"
118#: the response header for the escaped file name
119RESPONSE_ESCAPED_NAME: Final[str] = "@texgit@escName@"
120#: the response header for the url
121RESPONSE_URL: Final[str] = "@texgit@url@"
123#: the command start A
124__CMD_0A: Final[str] = r"\expandafter\xdef\csname "
125#: the command start B
126__CMD_0B: Final[str] = r"\expandafter\gdef\csname "
127#: the command middle
128__CMD_1: Final[str] = r"\endcsname{"
129#: the command end
130__CMD_2: Final[str] = r"}%"
133def __make_response(prefix: str, name: str, value: str,
134 xdef: bool = True) -> str:
135 """
136 Make a response command.
138 :param prefix: the prefix
139 :param value: the value
140 :param xdef: do we do xdef?
141 :return: the result
143 >>> print(__make_response(RESPONSE_PATH,
144 ... "lst:test", "./git/12.txt").replace(chr(92), "x"))
145 xexpandafterxxdefxcsname @texgit@path@lst:testxendcsname{./git/12.txt}%
147 >>> print(__make_response(RESPONSE_PATH,
148 ... "lst:test", "./git/12.txt", False).replace(chr(92), "x"))
149 xexpandafterxgdefxcsname @texgit@path@lst:testxendcsname{./git/12.txt}%
150 """
151 return (f"{__CMD_0A if xdef else __CMD_0B}{str.strip(prefix)}"
152 f"{str.strip(name)}{__CMD_1}{value}{str.strip(__CMD_2)}")
155def __make_path_response(name: str, path: Path, base_dir: Path,
156 basename: str | None = None)\
157 -> Generator[str, None, None]:
158 r"""
159 Make the path response commands.
161 :param name: the name prefix
162 :param path: the file path
163 :param basename: the basename
164 :param base_dir: the base directory
166 >>> from pycommons.io.temp import temp_dir
167 >>> with temp_dir() as td:
168 ... list(__make_path_response("x", td.resolve_inside("yy"), td, None))
169 ['\\expandafter\\xdef\\csname @texgit@path@x\\endcsname{yy}%']
170 >>> with temp_dir() as td:
171 ... v = list(__make_path_response("x", td.resolve_inside("yy"), td,
172 ... "bla y_x"))
173 >>> v[0]
174 '\\expandafter\\xdef\\csname @texgit@path@x\\endcsname{yy}%'
175 >>> v[1]
176 '\\expandafter\\xdef\\csname @texgit@name@x\\endcsname{bla y_x}%'
177 >>> v[2]
178 '\\expandafter\\gdef\\csname @texgit@escName@x\\endcsname{bla~y\\_x}%'
179 """
180 yield __make_response(RESPONSE_PATH, name, path.relative_to(base_dir))
181 if basename is not None:
182 yield __make_response(RESPONSE_NAME, name, basename)
183 yield __make_response(
184 RESPONSE_ESCAPED_NAME, name,
185 basename.replace("$", "\\$").replace("_", "\\_").replace(
186 " ", "~"), False) # this one must be gdef!
189def cmd_git_file(base_dir: Path, pm: ProcessManager,
190 command: list[str | None]) -> Generator[str, None, None]:
191 """
192 Get a file from `git`, maybe post-processed.
194 :param base_dir: the base directory
195 :param pm: the process manager
196 :param command: the command
197 :return: the answer
198 """
199 name: Final[str] = str.strip(command[1])
200 repo_url: Final[str] = str.strip(command[2])
201 relative_file: Final[str] = str.strip(command[3])
202 cmd: list[str] | None = command[4:]
203 ll: Final[int] = list.__len__(cmd)
204 if ll == 0 or ((ll == 1) and (cmd[0] is None)):
205 cmd = None
206 gp: Final[GitPath] = pm.get_git_file(
207 repo_url, relative_file, name, cmd)
208 yield from __make_path_response(
209 name, gp.path, base_dir, gp.basename)
210 if gp.url:
211 yield __make_response(RESPONSE_URL, name, gp.url)
214def cmd_arg_file(base_dir: Path, pm: ProcessManager,
215 command: list[str | None]) -> Generator[str, None, None]:
216 """
217 Get an argument file.
219 :param base_dir: the base directory
220 :param pm: the process manager
221 :param command: the command
222 :return: the answer
223 """
224 name: Final[str] = str.strip(command[1])
225 prefix: Final[str | None] = command[2]
226 suffix: Final[str | None] = command[3]
227 af: Final[Path] = pm.get_argument_file(name, prefix, suffix)[0]
228 yield from __make_path_response(name, af, base_dir, af.basename())
231def cmd_exec(base_dir: Path, pm: ProcessManager,
232 command: list[str | None]) -> Generator[str, None, None]:
233 """
234 Execute a command and capture the output.
236 :param base_dir: the base directory
237 :param pm: the process manager
238 :param command: the command
239 :return: the answer
240 """
241 name: Final[str] = str.strip(command[1])
242 repo_url: Final[str | None] = command[2]
243 relative_dir: Final[str | None] = command[3]
244 cmd: list[str] = command[4:]
245 yield from __make_path_response(
246 name, pm.get_output(name, cmd, repo_url, relative_dir), base_dir)
249def run(aux_arg: str, repo_dir_arg: str = "__git__") -> None:
250 """
251 Execute the `texgit` tool.
253 This tool loads an LaTeX `aux` file, processes all file loading requests,
254 and flushes the produced file paths back to the `aux` file.
256 :param aux_arg: the `aux` file argument
257 :param repo_dir_arg: the repository directory argument
258 """
259 aux_file: Path = Path(aux_arg)
260 if not aux_file.is_file():
261 aux_file = Path(f"{aux_arg}.aux")
262 if not aux_file.is_file():
263 raise ValueError(f"aux argument {aux_arg!r} does not identify a file "
264 f"and neither does {aux_file!r}")
265 logger(f"Using aux file {aux_file!r}.")
267 if getsize(aux_file) <= 0:
268 logger(f"aux file {aux_file!r} is empty. Nothing to do. Exiting.")
269 return
270 lines: Final[list[str]] = list(aux_file.open_for_read())
271 lenlines: Final[int] = len(lines)
272 if lenlines <= 0:
273 logger(f"aux file {aux_file!r} contains no lines. "
274 "Nothing to do. Exiting.")
275 else:
276 logger(f"Loaded {lenlines} lines from aux file {aux_file!r}.")
278 base_dir: Final[Path] = directory_path(dirname(aux_file))
279 logger(f"The base directory is {base_dir!r}.")
281 pm: ProcessManager | None = None
282 append: list[str] = []
283 stripped_lines: list[str] = list(map(str.strip, lines))
284 deleted: int = 0 # the number of lines deleted
286 try:
287 resolved: int = 0
288 for idx, line in enumerate(stripped_lines):
289 if line.startswith(FORBIDDEN_LINE):
290 del lines[idx - deleted]
291 deleted += 1
293 request: list[str | None] | None = __get_request(line)
294 if request is None:
295 continue
297 if pm is None:
298 git_dir: Path = base_dir.resolve_inside(repo_dir_arg)
299 logger(f"The repository directory is {git_dir!r}.")
300 pm = ProcessManager(git_dir)
302 func = str.strip(request[0])
303 if func == REQUEST_GIT_FILE:
304 append.extend(cmd_git_file(base_dir, pm, request))
305 elif func == REQUEST_ARG_FILE:
306 append.extend(cmd_arg_file(base_dir, pm, request))
307 elif func == REQUEST_PROCESS:
308 append.extend(cmd_exec(base_dir, pm, request))
309 else:
310 raise ValueError(f"Invalid command {func} in line {line!r}.")
311 resolved += 1
312 finally:
313 if pm is not None:
314 pm.close()
315 del pm
317 if (len(append) <= 0) and (deleted <= 0):
318 logger("No file requests or deletion markers found. Nothing to do.")
319 return
321 logger(f"Found and resolved {resolved} file requests.")
322 for app in map(str.strip, append): # make the texgit invocation idempotent
323 if app and (app not in stripped_lines):
324 lines.append(app)
325 with aux_file.open_for_write() as wd:
326 write_lines(lines, wd)
327 logger(f"Finished flushing {len(lines)} lines to aux file {aux_file!r}.")
330# Execute the texgit tool
331if __name__ == "__main__":
332 parser: Final[argparse.ArgumentParser] = make_argparser(
333 __file__, "Execute the texgit Tool.",
334 make_epilog(
335 "Download and provide local paths for "
336 "files from git repositories and execute programs.",
337 2023, 2025, "Thomas Weise",
338 url="https://thomasweise.github.io/texgit_py",
339 email="tweise@hfuu.edu.cn, tweise@ustc.edu.cn"),
340 __version__)
341 parser.add_argument(
342 "aux", help="the aux file to process", type=str, default="")
343 parser.add_argument(
344 "--repoDir", help="the directory to use for caching output",
345 type=str, default="__git__", nargs="?")
346 args: Final[argparse.Namespace] = parser.parse_args()
348 run(args.aux.strip(), args.repoDir.strip())
349 logger("All done.")