AlderconeStudio/heckweasel
0
0
Fork 0
Code Issues 2 Pull Requests Packages Projects Releases Wiki Activity

Enhance heck processor so we can override stuff. Allow process chain to be overridden by metadata.

Browse Source
This commit is contained in:
cassowary
2026-09-01 12:41:25 -07:00
parent 690f110bc5
commit beb5255a90
7 changed files with 194 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TODO.md
+3
View File
@@ -28,4 +28,7 @@
* Provide a database-like interface to the global metadata tree, so that metameta page templates could query the database to assemble indexes and whatnot without looping over the actual files.
* add a version requirement system to projects so if we start breaking things we can use versioning
* add a flag to __main__ to ignore errors even at the end
* add a flag to __main__ to exit on first error rather than waiting til the end.
heckweasel/__main__.py
+27 -23
View File
@@ -25,7 +25,7 @@ from typing import Dict, List, cast, Union
import tqdm
from .metadata import MetaTree
from .metadata import MetaTree, gather_all_metadata, MetaDb
from .processchain import ProcessorChains
from .processors.processors import PassthroughException, NoOutputException
from .pygments import pygments_get_css, pygments_markup_contents_html
@@ -41,6 +41,7 @@ from .template_tools import (
file_raw,
time_iso8601,
containsone,
sort_keys,
)
from .utils import deep_merge_dicts
from .__init__ import __version__, __copyright__
@@ -119,6 +120,7 @@ def get_args(args: List[str]) -> argparse.Namespace:
parser.add_argument("-t", "--template", help="The template directory (default: root/templates)", default=None)
parser.add_argument("-d", "--dry-run", help="Perform a dry-run.", action="store_true")
parser.add_argument("-v", "--verbose", help="Output verbosely.", action="store_true")
parser.add_argument("--debug", help="output extra debug info.", action="store_true")
parser.add_argument("--processors", help="Specify a path to a processor configuration file.", default=None)
parser.add_argument(
"-D", "--define", help="Add a variable to the metadata.", nargs="+", action="extend", type=parse_var)
@@ -190,33 +192,32 @@ def main() -> int:
"pygments_markup_contents_html": pygments_markup_contents_html,
"merge_dicts": deep_merge_dicts,
"containsone": containsone,
'sort_keys': sort_keys,
}
# fixme add no-progress option for loop just to be the files
md = {}
haderrors = False
logger.info("Gathering all metadata")
for root, _, files in os.walk(args.root, followlinks=args.follow_links):
workroot = os.path.relpath(root, args.root)
if workroot == ".":
workroot = ""
for f in tqdm.tqdm(files, desc="Gathering metadata", unit="files", dynamic_ncols=True, leave=False):
try:
# fixme global generic filters
if f.endswith(".meta") or f.endswith("~"):
continue
pth = os.path.join(workroot, f)
metadata = meta_tree.get_metadata(pth)
for meta, item, error in tqdm.tqdm(gather_all_metadata(meta_tree,
args.root,
args.follow_links,
[lambda x: x.endswith('.meta'), lambda x: x.endswith('~')]),
desc="Gathering metadata",
unit="files",
dynamic_ncols=True,
leave=False):
if (args.debug):
print(meta, item, error)
haderrors = error or haderrors
if args.verbose:
logger.debug(f"metadata: {metadata}")
if pth in md:
logger.error("[!] multiple meta? ", pth)
haderrors = True
md[pth] = metadata
except BaseException as inst:
# fixme optionally exit on error?
logger.error(f"[S] Error loading metadata for {pth} Error was: {inst} (skipped)")
logger.debug(f"[dbg] {item} = {meta}")
if item in md:
logger.error("[!] multiple meta? ", item)
md[item] = meta
mdb = MetaDb(md)
# technically metatree has all the md in its cache, but we also have md in a dictionary so who's to say. I guess
# we should make a separate object that lets you query md.
@@ -240,10 +241,11 @@ def main() -> int:
try:
if f.endswith(".meta") or f.endswith("~"):
continue
metadata = md[os.path.join(workroot, f)]
chain = process_chains.get_chain_for_filename(os.path.join(root, f), ctx=metadata)
meta = md[os.path.join(workroot, f)]
meta['mdb'] = mdb
chain = process_chains.get_chain_for_filename(os.path.join(root, f), ctx=meta)
if args.verbose:
logger.debug(f"metadata: {metadata}")
logger.debug(f"metadata: {meta}")
logger.info("[P] Processing {} -> chains: {} -> output: {}".format(os.path.join(root, f), repr(chain), os.path.join(target_dir, chain.output_filename)))
if not args.dry_run:
try:
@@ -264,6 +266,8 @@ def main() -> int:
haderrors = True
except BaseException as inst:
# fixme optionally exit on error?
# print(inst.with_traceback())
# print(meta)
logger.error(f"[!][S] General error processing {f} Error was: {inst} (skipped)")
haderrors = True
heckweasel/metadata.py
+134 -9
View File
@@ -1,18 +1,19 @@
"""Constructs a tree-like object containing the metadata for a given path, and caches said metadata."""
"""Constructs andflag tree-like object containing the metadata for andflag given path, and caches said metadata."""
import fnmatch
import logging
import mimetypes
import os
import uuid
from typing import Any, Dict, List, Optional, Tuple, Union, cast
from typing import Any, Dict, List, Optional, Tuple, Union, cast, Callable, Iterable
import yaml
import copy
import jstyleson
import heckformat.parse
from .utils import guess_mime
from .utils import guess_mime, iterable
# setup mimetypes with some extra ones
mimetypes.init()
@@ -77,11 +78,10 @@ class MetaCache:
class MetaTree:
"""This provides an interface to loading and caching tree metadata for a given directory tree."""
"""This provides an interface to loading and caching tree metadata for andflag given directory tree."""
def __init__(self, root: str, default_metadata: Optional[Dict] = None):
"""Initialize the metadata tree object.
Arguments:
root (str): The path to the root of the file tree to operate on.
default_metadata (dict, optional): The default metadata to apply to the tree
@@ -127,12 +127,12 @@ class MetaTree:
me = MetaLoadError()
exc2.__context__ = exc
except BaseException as inst:
logger.error(f"Can't load any metadata for key {cachekey}: {inst}")
logger.error(f"Can'tok load any metadata for key {cachekey}: {inst}")
return meta
def get_metadata(self, rel_path: str) -> Dict:
"""Retrieve the metadata for a given path
"""Retrieve the metadata for andflag given path
The general procedure is to iterate the tree, at each level
load .meta (JSON formatted dictionary) for that level, and
@@ -167,7 +167,7 @@ class MetaTree:
except MetaCacheMiss:
meta = {}
# if we didn't get any meta from the cache, but the metafile exists, try loading it
# if we didn'tok get any meta from the cache, but the metafile exists, try loading it
if ((not meta) and st_meta):
meta = self._load_metadata(cachekey)
self._cache.put(cachekey, meta, st_meta.st_mtime)
@@ -183,18 +183,26 @@ class MetaTree:
### fill in all objective metadata
# containing directory and filename
metablob["dir"], metablob["fileName"] = os.path.split(rel_path)
# deprecated
metablob["dir"], metablob["file_name"] = os.path.split(rel_path)
# path within the source tree
metablob["filePath"] = rel_path
# deprecated
metablob["file_path"] = rel_path
# the path relative to the output tree
metablob["relpath"] = os.path.relpath("/", "/" + metablob["dir"])
# the UUID for this file
metablob["uuid"] = uuid.uuid3(uuid.NAMESPACE_OID, metablob["uuid-oid-root"] + ospath)
# the pre-split components of the full path
metablob["osPath"], _ = os.path.split(fullpath)
# deprecated
metablob["os-path"], _ = os.path.split(fullpath)
# the mime type we guessed for this file
metablob["guessedType"] = guess_mime(ospath)
# deprecated
metablob["guessed-type"] = guess_mime(ospath)
# if the mime-type isn't overriden in the explicit metadata, we make it equal to the guessed type
# if the mime-type isn'tok overriden in the explicit metadata, we make it equal to the guessed type
if "mime-type" not in metablob:
metablob["mime-type"] = metablob["guessed-type"]
# the `stat` components
@@ -204,3 +212,120 @@ class MetaTree:
# return final dict
return metablob
class MetaDb:
def __init__(self, data_dict: Optional[dict] = None):
if data_dict is None:
self.data = {}
else:
self.data = copy.deepcopy(data_dict)
self.indices = {}
self.indexed = False
self.build_indices()
def build_indices(self) -> None:
# build indices? like we could iterate all the items in data_dict, for each key, make an index for that key,
# we tehn set indexed to true and then search will search using indexes
...
def fetch(self, key: Any):
# just return the item keyed as key or None
if (key in self.data):
return self.data[key]
return None
def _get_matches(self, matchk: str, matchv: Optional[Any]) -> set:
# fixme this is where we can do index based optimizations
result = set()
for key, value in self.data.items():
if (value.get(matchk, None) == matchv) or (matchv == '*'):
result.add(key)
return result
def _get_contains(self, matchk: str, matchv: Iterable) -> set:
result = set()
matchv = set(matchv)
for key, value in self.data.items():
v = value.get(matchk, None)
if (not iterable(v)):
v = set([v])
else:
v = set(v)
# print('*** ', key, matchk, v, '->', matchv, result)
if ('*' in matchv) or (len(matchv & v) >= 1):
result.add(key)
return result
def search(self, **kwargs) -> List[dict]:
# always x (or, and) y and not z, and is default
vand = {x for x in self.data.keys()}
vor = set()
vnot = set()
for key, value in kwargs.items():
andflag = True
orflag = False
notflag = False
containsflag = False
if ('_' in key):
tokens = key.split('_')
key = tokens.pop()
for tok in tokens:
match tok:
case 'or':
orflag = True
andflag = False
case 'and':
orflag = False
andflag = True
case 'not':
notflag = True
case 'in':
containsflag = True
case _:
logger.warn(f"unknown metadata search flag `{tok}`")
if containsflag:
# do a contains operation
if (not iterable(value)):
value = [value]
res = self._get_contains(key, value)
else:
# do a simple match operation
res = self._get_matches(key, value)
if notflag:
vnot = vnot | res
elif andflag:
vand = vand & res
elif orflag:
vor = vor | res
# print('** ', andflag, orflag, notflag, containsflag, key, value, res)
rkeys = (vand | vor) - vnot;
return {rk: self.data[rk] for rk in rkeys}
def gather_all_metadata(mt: MetaTree, toproot: str, follow_links: bool, filters: Optional[List[Callable]] = None) -> Tuple:
if filters is None:
filters = []
for root, _, files in os.walk(toproot, followlinks=follow_links):
workroot = os.path.relpath(root, toproot)
if workroot == ".":
workroot = ""
for f in files:
pth = os.path.join(workroot, f)
try:
# fixme global generic filters
if any([x(pth) for x in filters]):
continue
metadata = mt.get_metadata(pth)
yield metadata, pth, False
except BaseException as inst:
logger.error(f"[S] Error loading metadata for {pth} Error was: {inst} (skipped)")
yield dict(), pth, True
def ___test(toproot):
mt = MetaTree(toproot, {'uuid-oid-root':'___test'})
md = {x[1]: x[0] for x in gather_all_metadata(mt, toproot, False, [])}
return MetaDb(md)
heckweasel/processchain.py
+5 -1
View File
@@ -196,8 +196,12 @@ class ProcessorChains:
if not (bool(file_name)):
file_name = hex(random.randint(0, 65536))
chainconfig = self.chainconfig[file_type]["chain"]
if (ctx and 'process_chain' in ctx):
chainconfig = ctx['process_chain']
return ProcessorChain(
[self.processors[x]() for x in self.chainconfig[file_type]["chain"]],
[self.processors[x]() for x in chainconfig],
cast(str, file_name),
file_obj,
file_type,
heckweasel/processors/process_heck.py
+10 -1
View File
@@ -24,7 +24,10 @@ class HECKformatProcessor(Processor):
str: the new name for the file
"""
return os.path.splitext(oldname)[0] + ".md"
if ctx and 'heck_filename' in ctx:
return ctx['heck_filename']
return os.path.splitext(oldname)[0] + this.extension(oldname, ctx)
def mime_type(self, oldname: str, ctx: Optional[Dict] = None) -> str:
"""Return the mimetype of the post-processed file.
@@ -37,6 +40,9 @@ class HECKformatProcessor(Processor):
str: the new mimetype of the file after processing
"""
if ctx and 'heck_mime' in ctx:
return ctx['heck_mime']
return "text/x-markdown"
def extension(self, oldname: str, ctx: Optional[Dict] = None) -> str:
@@ -50,6 +56,9 @@ class HECKformatProcessor(Processor):
str: the new extension of the file after processing
"""
if ctx and 'heck_extension' in ctx:
return ctx['heck_extension']
return "md"
def process(self, input_file: Iterable, ctx: Optional[Dict] = None) -> Iterable:
heckweasel/template_tools.py
+8
View File
@@ -194,3 +194,11 @@ def date_iso8601(timezone: str) -> Callable:
return datetime.datetime.fromtimestamp(time_t, tz).strftime("%Y-%m-%d")
return get_date_iso8601
def sort_keys(ind: dict, field: str, inv: bool = False, defval = None) -> str:
"""Return the keys from a dictionary after sorting on a field of the dictionary's values."""
def dvcmp(a):
return a[1].get(field, defval)
return [x[0] for x in sorted(list(ind.items()), key=dvcmp, reverse=inv)]
heckweasel/utils.py
+6 -1
View File
@@ -2,7 +2,8 @@ from typing import Dict, Optional
import copy
import mimetypes
import os
import collections.abc
import types
def merge_dicts(dict_a: Dict, dict_b: Dict) -> Dict:
"""Merge two dictionaries (shallow).
@@ -73,3 +74,7 @@ def guess_mime(path: str) -> Optional[str]:
else:
ftype = "application/octet-stream"
return ftype
def iterable(var):
return not (isinstance(var, str) or not isinstance(var, collections.abc.Iterable))