Cassowary
690f110bc5
- Clean up a ton of documentation.
- Make the modules import nicely.
- Add a cool logo to the command line tool
- Make the command-line tool use tqdm
- Make the command line tool load the metadata before processing the
files in a separate loop.
- Fix error handling in the command-line tool processing loops so they
work correctly (and jinja errors are more useful)
- Make command-line tool exit non-zero if there were errors.
- Fix load metadata to handle formats and errors better (and return {}
if it fails)
197 lines
6.6 KiB
Python
197 lines
6.6 KiB
Python
"""
|
|
Provides various utility functions that are exposed to the templates.
|
|
"""
|
|
import copy
|
|
import datetime
|
|
import glob
|
|
import itertools
|
|
import os
|
|
from typing import Callable, Dict, Iterable, List, Union, cast, Tuple
|
|
|
|
import jstyleson
|
|
|
|
import pytz
|
|
|
|
from .metadata import MetaTree
|
|
from .processchain import ProcessorChains
|
|
from .utils import deep_merge_dicts
|
|
|
|
|
|
def file_list(root: str, listcache: Dict) -> Callable:
|
|
"""
|
|
Return a function (memoized for the cache and root directory) which returns a list of files matching glob and
|
|
sorted as required.
|
|
"""
|
|
def get_file_list(
|
|
path_glob: Union[str, List[str], Tuple[str]],
|
|
*,
|
|
sort_order: str = "ctime",
|
|
reverse: bool = False,
|
|
limit: int = 0) -> Iterable:
|
|
stattable = cast(List, [])
|
|
if isinstance(path_glob, str):
|
|
path_glob = [path_glob]
|
|
for pglob in path_glob:
|
|
if pglob in listcache:
|
|
stattable.extend(listcache[pglob])
|
|
else:
|
|
for fil in glob.glob(os.path.join(root, pglob)):
|
|
if os.path.isdir(fil):
|
|
continue
|
|
if fil.endswith(".meta") or fil.endswith("~"):
|
|
continue
|
|
st = os.stat(fil)
|
|
stattable.append(
|
|
{
|
|
"file_path": os.path.relpath(fil, root),
|
|
"file_name": os.path.split(fil)[-1],
|
|
"mtime": st.st_mtime,
|
|
"ctime": st.st_ctime,
|
|
"size": st.st_size,
|
|
"ext": os.path.splitext(fil)[1],
|
|
}
|
|
)
|
|
listcache[pglob] = stattable
|
|
ret = sorted(stattable, key=lambda x: x[sort_order], reverse=reverse)
|
|
if limit > 0:
|
|
return itertools.islice(ret, limit)
|
|
return ret
|
|
|
|
return get_file_list
|
|
|
|
|
|
def file_list_hier(root: str, flist: Callable) -> Callable:
|
|
"""Return a function which, given a directory, will walk the directory and return the files within
|
|
it that match the glob passed."""
|
|
|
|
def get_file_list_hier(path: str, glob: str, *, sort_order: str = "ctime", reverse: bool = False) -> Iterable:
|
|
output = []
|
|
|
|
for pth in os.walk(os.path.join(root, path)):
|
|
output.extend(
|
|
flist(
|
|
os.path.join(os.path.relpath(os.path.realpath(pth[0]), root), glob),
|
|
sort_order=sort_order,
|
|
reverse=reverse,
|
|
)
|
|
)
|
|
|
|
return output
|
|
|
|
return get_file_list_hier
|
|
|
|
|
|
def file_name(root: str, metatree: MetaTree, processor_chains: ProcessorChains, namecache: Dict) -> Callable:
|
|
"""
|
|
Return a function (memoized for root directory, metatree and processor chains) which returns the output filename
|
|
given an input filename based on metadata and said processing chains.
|
|
"""
|
|
def get_file_name(file_name: str) -> Dict:
|
|
if file_name in namecache:
|
|
return namecache[file_name]
|
|
metadata = metatree.get_metadata(file_name)
|
|
chain = processor_chains.get_chain_for_filename(os.path.join(root, file_name), ctx=metadata)
|
|
namecache[file_name] = chain.output_filename
|
|
return namecache[file_name]
|
|
|
|
return get_file_name
|
|
|
|
|
|
def file_raw(root: str, contcache: Dict) -> Callable:
|
|
"""
|
|
Return a function (memoizedfor the root directory) which returns the raw content of a file.
|
|
"""
|
|
def get_raw(file_name: str) -> str:
|
|
if file_name in contcache:
|
|
return contcache[file_name]
|
|
with open(os.path.join(root, file_name), "r", encoding="utf-8") as f:
|
|
return f.read()
|
|
|
|
return get_raw
|
|
|
|
|
|
def file_json(root: str) -> Callable:
|
|
"""
|
|
Return a function (memoized for the root directory) which loads a file as json, merges it with an optional input dictionary
|
|
and returns.
|
|
"""
|
|
def get_json(file_name: str, parent: Dict = None) -> Dict:
|
|
outd = {}
|
|
if parent is not None:
|
|
outd = copy.deepcopy(parent)
|
|
|
|
with open(os.path.join(root, file_name), "r", encoding="utf-8") as f:
|
|
return deep_merge_dicts(outd, jstyleson.load(f))
|
|
|
|
return get_json
|
|
|
|
|
|
def file_heck(root: str) -> Callable:
|
|
"""
|
|
Return a function (memoized for the root directory) which loads a file as HECKFormat, merges it with an optional input
|
|
dictionary, and returns.
|
|
"""
|
|
def get_heck(file_name: str, parent: Dict = None) -> Dict:
|
|
outd = {}
|
|
if parent is not None:
|
|
outd = copy.deepcopy(parent)
|
|
|
|
with open(os.path.join(root, file_name), "r", encoding="utf-8") as f:
|
|
return deep_merge_dicts(outd, heckformat.parse.load(f).flatten_replace())
|
|
|
|
return get_heck
|
|
|
|
|
|
def file_content(root: str, metatree: MetaTree, processor_chains: ProcessorChains, contcache: Dict) -> Callable:
|
|
"""
|
|
Return a function (memoized for the root directory, metatree, and processor chains) which returns the post-processed
|
|
content of the input file.
|
|
"""
|
|
def get_file_content(file_name: str) -> Iterable:
|
|
if file_name in contcache:
|
|
return contcache[file_name]
|
|
metadata = metatree.get_metadata(file_name)
|
|
chain = processor_chains.get_chain_for_filename(os.path.join(root, file_name), ctx=metadata)
|
|
contcache[file_name] = chain.output
|
|
return str(chain.output)
|
|
|
|
return get_file_content
|
|
|
|
|
|
def file_metadata(metatree: MetaTree) -> Callable:
|
|
"""Returns a function (memoized for a metatree) which returns the meta data for a given file."""
|
|
def get_file_metadata(file_name: str) -> Dict:
|
|
return metatree.get_metadata(file_name)
|
|
|
|
return get_file_metadata
|
|
|
|
|
|
def containsone(needle: Iterable, haystack: Iterable):
|
|
"""
|
|
Returns true if at least one of the contents of needle is in haystack.
|
|
"""
|
|
for n in needle:
|
|
if n in haystack:
|
|
return True
|
|
|
|
return False
|
|
|
|
def time_iso8601(timezone: str) -> Callable:
|
|
"""Returns a function (memoized for a particular timezone) which formats a time as ISO8601 standard. """
|
|
tz = pytz.timezone(timezone)
|
|
|
|
def get_time_iso8601(time_t: Union[int, float]) -> str:
|
|
return datetime.datetime.fromtimestamp(time_t, tz).isoformat("T")
|
|
|
|
return get_time_iso8601
|
|
|
|
|
|
def date_iso8601(timezone: str) -> Callable:
|
|
"""Returns a function (memoized for a particular timezone) which formats a date as ISO8601 standard. """
|
|
tz = pytz.timezone(timezone)
|
|
|
|
def get_date_iso8601(time_t: Union[int, float]) -> str:
|
|
return datetime.datetime.fromtimestamp(time_t, tz).strftime("%Y-%m-%d")
|
|
|
|
return get_date_iso8601
|
