will
25592fa6c2
ci / test (push) Successful in 6m14s
History-free snapshot of the private development repository at 77a82eb. Code, docs and the public catalogue only: no plant or job knowledge layers, no client, job or producer names (tools/scrub.py clean against the widened term list; a second word-level grep clean). The intro deck is left out because its slide images cannot be checked for job drawings.
411 lines
15 KiB
Python
411 lines
15 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""One JSON line per button run -- the run log (`vibe.runlog`).
|
|
|
|
Owner requirement, 2026-09-03: "开发阶段，我做的一切，都留下 log，别我按完了
|
|
以后还得手动复制 report 给你。" Every button press must leave a machine-readable
|
|
record on disk, so the developer can read what happened without the owner
|
|
pasting reports by hand.
|
|
|
|
The stream is an append-only JSONL file, `runlog.jsonl`, inside a folder chosen
|
|
by `log_dir()`: $WILLTOOLS_RUNLOG_DIR when set, else %APPDATA%/WillTools/runlog,
|
|
else ~/WillTools/runlog. `record()` writes one JSON object per line; buttons
|
|
wrap themselves in `run(...)` so the outcome and any error land automatically,
|
|
and the module doubles as a CLI: `python lib/vibe/runlog.py [n]` prints the
|
|
last n records as compact one-liners. `tail(n)` is the same reader for the
|
|
developer's `tools/` helpers.
|
|
|
|
This module is engine-neutral on purpose -- IronPython 2.7 inside Revit *and*
|
|
host CPython 3. Nothing is imported from `vibe` at module scope (a bare
|
|
`vibe.runlog` import must work on a host that has no Revit at all, and
|
|
`vibe/__init__.py` pulls the Revit API in the moment it runs), so the tiny
|
|
py2/py3 text-type split that `vibe.compat` normally provides is repeated here
|
|
as a local `text_type` instead of imported. The rest is written in the
|
|
intersection of the two languages: no f-strings, no pathlib, `io.open`,
|
|
`'{0}'.format(...)`, and every string encoded to UTF-8 at the byte boundary.
|
|
|
|
One deliberate decision: `record()` never raises. A log that takes a button
|
|
down with it is worse than no log, so any failure to write is swallowed after
|
|
a best-effort one-line `runlog-failed` marker on stderr, and `record()`
|
|
returns None. Callers that need to know the write succeeded compare the
|
|
return value.
|
|
"""
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import traceback as _traceback
|
|
|
|
try: # IronPython 2.7
|
|
text_type = unicode # noqa: F821
|
|
except NameError: # CPython 3 / IronPython 3
|
|
text_type = str
|
|
|
|
|
|
LOG_DIR_ENV = 'WILLTOOLS_RUNLOG_DIR'
|
|
"""Environment variable naming the run-log folder, when the developer sets it."""
|
|
|
|
_FILE_NAME = 'runlog.jsonl'
|
|
_REPORT_TEXT_CAP = 20000
|
|
_TRUNCATED_MARKER = '...[truncated]'
|
|
|
|
|
|
# ------------------------------------------------------------------ location
|
|
|
|
def _home_dir():
|
|
"""The user's home folder, tolerantly, for the final fallback."""
|
|
try:
|
|
home = os.path.expanduser('~')
|
|
except Exception:
|
|
home = None
|
|
if home:
|
|
return home
|
|
# Last resort: a temp folder. Whatever record() gets for its base is only
|
|
# used once, to mkdir a folder and open one file -- and if even that fails
|
|
# record() swallows it and returns None.
|
|
try:
|
|
return os.environ.get('TEMP') or os.environ.get('TMP') or os.getcwd()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _base_dir():
|
|
"""The resolved folder name for the run log, never created here.
|
|
|
|
Order, matching the module contract: $WILLTOOLS_RUNLOG_DIR when set (a
|
|
developer or a test pointing the log somewhere specific), else
|
|
%APPDATA%/WillTools/runlog, else ~/WillTools/runlog.
|
|
"""
|
|
env = os.environ.get(LOG_DIR_ENV)
|
|
if env:
|
|
return env
|
|
appdata = os.environ.get('APPDATA')
|
|
if appdata:
|
|
return os.path.join(appdata, 'WillTools', 'runlog')
|
|
home = _home_dir()
|
|
if home:
|
|
return os.path.join(home, 'WillTools', 'runlog')
|
|
return None
|
|
|
|
|
|
def log_dir():
|
|
"""The run-log folder: $WILLTOOLS_RUNLOG_DIR, else %APPDATA%/WillTools/runlog,
|
|
else ~/WillTools/runlog.
|
|
|
|
The folder is resolved fresh on every call -- the resolution is a couple of
|
|
environment lookups, and a stale cache would make a test (or a developer
|
|
switching the log location) write to yesterday's folder. The folder itself
|
|
is created on demand: the first record() that needs it. Returns None when
|
|
no base folder can be resolved at all.
|
|
"""
|
|
return _base_dir()
|
|
|
|
|
|
def _log_path():
|
|
"""The full path of the run-log file, or None when no folder is resolvable.
|
|
|
|
Nothing is created here -- mkdir happens only inside record(), so asking
|
|
where the log lives never touches the disk.
|
|
"""
|
|
folder = log_dir()
|
|
if not folder:
|
|
return None
|
|
return os.path.join(folder, _FILE_NAME)
|
|
|
|
|
|
# ---------------------------------------------------------------- JSON help
|
|
|
|
def _safe_json_values(values):
|
|
"""A `values` dict with every value JSON-serialisable.
|
|
|
|
Buttons log whatever they happen to be holding -- ElementIds, tuples, units,
|
|
Decimals. `json.dumps` would raise on the first non-serialisable one and
|
|
the whole record would be lost, so each value is probed with a real
|
|
serialisation: a value that survives is kept as-is, one that raises is
|
|
flattened with str(), and if even that fails the slot becomes None.
|
|
"""
|
|
out = {}
|
|
if not values:
|
|
return out
|
|
for key, value in values.items():
|
|
try:
|
|
json.dumps(value)
|
|
out[key] = value
|
|
except Exception:
|
|
try:
|
|
out[key] = text_type(value)
|
|
except Exception:
|
|
out[key] = None
|
|
return out
|
|
|
|
|
|
def _clean_text(value, cap, marker):
|
|
"""One text field flattened for JSON: bytes decoded, capped at `cap`.
|
|
|
|
A report body can run to tens of thousands of characters; the log caps it
|
|
so a runaway report cannot balloon runlog.jsonl. Text longer than the cap
|
|
is cut at the cap and the marker is appended, exactly as the module
|
|
contract describes. `text_type(...)` rather than `str(...)`: under
|
|
IronPython 2.7 `str()` of a non-ASCII unicode tries an ASCII encode and
|
|
raises, where `unicode()` copies it untouched.
|
|
"""
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, bytes):
|
|
try:
|
|
value = value.decode('utf-8', 'replace')
|
|
except Exception:
|
|
value = repr(value)
|
|
try:
|
|
text = text_type(value)
|
|
except Exception:
|
|
text = repr(value)
|
|
if len(text) > cap:
|
|
return text[:cap] + marker
|
|
return text
|
|
|
|
|
|
def _write_failure(path, exc):
|
|
"""Best-effort one-line marker when the record could not be written."""
|
|
try:
|
|
sys.stderr.write('runlog-failed: could not write {0}: {1}\n'
|
|
.format(path or '<no log dir>', repr(exc)))
|
|
sys.stderr.flush()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# ------------------------------------------------------------------ record()
|
|
|
|
def record(button, doc_title=None, marks=None, outcome=None, report_text=None,
|
|
report_path=None, error=None, traceback_text=None, values=None,
|
|
extra=None):
|
|
"""Append one JSON line to <log_dir>/runlog.jsonl. Never raises.
|
|
|
|
Fields written: ts (ISO 8601), button, doc_title, marks (list), outcome,
|
|
report_path, report_text (capped at 20000 characters with a
|
|
'...[truncated]' marker), error, traceback, values (a dict,
|
|
JSON-serialised with default=str), extra (a dict). Missing fields are
|
|
written as None so a reader never has to guess which key was omitted.
|
|
|
|
Any failure -- a folder that cannot be made, a path that is really a file,
|
|
an unwritable stream, a value that refuses to flatten -- is swallowed after
|
|
a best-effort one-line 'runlog-failed' marker on stderr. Returns the path
|
|
written, or None.
|
|
"""
|
|
path = None
|
|
try:
|
|
path = _log_path()
|
|
if not path:
|
|
_write_failure(path, RuntimeError('no run-log folder resolvable'))
|
|
return None
|
|
folder = os.path.dirname(path)
|
|
if not os.path.isdir(folder):
|
|
os.makedirs(folder)
|
|
marks_list = None
|
|
if marks is not None:
|
|
try:
|
|
marks_list = list(marks)
|
|
except Exception:
|
|
marks_list = None
|
|
extra_dict = extra if isinstance(extra, dict) else None
|
|
entry = {
|
|
'ts': time.strftime('%Y-%m-%dT%H:%M:%S'),
|
|
'button': button,
|
|
'doc_title': doc_title,
|
|
'marks': marks_list,
|
|
'outcome': outcome,
|
|
'report_path': report_path,
|
|
'report_text': _clean_text(report_text, _REPORT_TEXT_CAP,
|
|
_TRUNCATED_MARKER),
|
|
'error': error,
|
|
'traceback': traceback_text,
|
|
'values': (_safe_json_values(values)
|
|
if isinstance(values, dict) else None),
|
|
'extra': extra_dict,
|
|
}
|
|
line = json.dumps(entry, ensure_ascii=False)
|
|
# json.dumps returns unicode on py2 whenever any character is
|
|
# non-ASCII and a plain str otherwise -- normalise both to bytes and
|
|
# append through a binary handle so the file is always UTF-8.
|
|
if not isinstance(line, bytes):
|
|
line = line.encode('utf-8')
|
|
with io.open(path, 'ab') as handle:
|
|
handle.write(line + b'\n')
|
|
return path
|
|
except Exception as exc:
|
|
_write_failure(path, exc)
|
|
return None
|
|
|
|
|
|
# ------------------------------------------------------------- run() context
|
|
|
|
class _Run(object):
|
|
"""The object `run()` yields: the body's window into the record.
|
|
|
|
The body sets rl.outcome, rl.report_text, rl.report_path, rl.values and
|
|
rl.extra; anything it appends to rl.lines is folded into report_text when
|
|
the body never set one itself. Everything is written by __exit__, never
|
|
here.
|
|
"""
|
|
|
|
def __init__(self, button, doc_title, marks):
|
|
self.button = button
|
|
self.doc_title = doc_title
|
|
self.marks = list(marks) if marks else None
|
|
self.outcome = None
|
|
self.report_text = None
|
|
self.report_path = None
|
|
self.values = None
|
|
self.extra = None
|
|
self.lines = []
|
|
self.path = None
|
|
|
|
|
|
def _folded_report(rl):
|
|
"""The body's report_text, or its collected `lines` when text is unset."""
|
|
if rl.report_text is not None:
|
|
return rl.report_text
|
|
if rl.lines:
|
|
return '\n'.join(rl.lines)
|
|
return None
|
|
|
|
|
|
class _RunContext(object):
|
|
"""Context manager behind `run()`; __exit__ does the writing."""
|
|
|
|
def __init__(self, button, doc_title, marks):
|
|
self.button = button
|
|
self.doc_title = doc_title
|
|
self.marks = marks
|
|
self.rl = _Run(button, doc_title, marks)
|
|
|
|
def __enter__(self):
|
|
return self.rl
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
rl = self.rl
|
|
if exc is not None:
|
|
rl.path = record(rl.button, rl.doc_title, rl.marks,
|
|
outcome='error',
|
|
report_text=_folded_report(rl),
|
|
report_path=rl.report_path,
|
|
error=repr(exc),
|
|
traceback_text=''.join(
|
|
_traceback.format_exception(exc_type, exc,
|
|
tb)),
|
|
values=rl.values, extra=rl.extra)
|
|
return False # re-raise: the button decides
|
|
rl.path = record(rl.button, rl.doc_title, rl.marks,
|
|
outcome=rl.outcome or 'ok',
|
|
report_text=_folded_report(rl),
|
|
report_path=rl.report_path,
|
|
values=rl.values, extra=rl.extra)
|
|
return False
|
|
|
|
|
|
def run(button, doc_title=None, marks=None):
|
|
"""A context manager recording one button run.
|
|
|
|
Usage:
|
|
|
|
with runlog.run('Panel Lift Check', doc_title, marks) as rl:
|
|
rl.report_text = ...
|
|
rl.values = {...}
|
|
|
|
The body may set rl.outcome, rl.report_text, rl.report_path, rl.values and
|
|
rl.extra. On normal exit the record is written with the body's fields,
|
|
defaulting the outcome to 'ok'. On an exception it records outcome='error'
|
|
with `error=repr(exc)` and the formatted traceback, then RE-RAISES, so the
|
|
button's own error handling still sees the exception. Anything the body
|
|
appends to rl.lines (a list of strings) is joined into report_text when
|
|
the body left report_text unset. The path of the record that was written
|
|
is left on rl.path afterwards (None when the write failed -- record()
|
|
never raises, so a logging failure must not look like a run failure).
|
|
"""
|
|
return _RunContext(button, doc_title, marks)
|
|
|
|
|
|
# --------------------------------------------------------------------- tail()
|
|
|
|
def _read_lines_backwards(path):
|
|
"""Yield the non-empty lines of a file, last line first.
|
|
|
|
Walks the file backwards byte by byte so reading the tail of a long log
|
|
costs O(tail), not O(log). A final line with no trailing newline and the
|
|
first line (which has no preceding newline) are both handled; a blank line
|
|
between records is skipped.
|
|
"""
|
|
with io.open(path, 'rb') as handle:
|
|
handle.seek(0, os.SEEK_END)
|
|
position = handle.tell()
|
|
buf = []
|
|
while position > 0:
|
|
position -= 1
|
|
handle.seek(position)
|
|
byte = handle.read(1)
|
|
if byte == b'\n':
|
|
if buf:
|
|
yield b''.join(reversed(buf))
|
|
buf = []
|
|
else:
|
|
buf.append(byte)
|
|
# Everything read before the scan hit byte 0: the first line, or the
|
|
# whole file when it never contained a newline at all.
|
|
if buf:
|
|
yield b''.join(reversed(buf))
|
|
|
|
|
|
def tail(n=20, path=None):
|
|
"""Read the last n records of a run log, newest last: a list of dicts.
|
|
|
|
Records are returned in written order (oldest of the n first). Lines that
|
|
fail to parse are skipped -- a torn tail from a crash mid-write is a fact
|
|
of an append-only log, not a reason for the reader to die. `path`
|
|
overrides the log location for the developer's helpers; the default is
|
|
the same folder record() writes to. An absent or unreadable log is an
|
|
empty list, never an exception.
|
|
"""
|
|
if path is None:
|
|
path = _log_path()
|
|
if not path or not os.path.isfile(path):
|
|
return []
|
|
records = []
|
|
try:
|
|
for raw in _read_lines_backwards(path):
|
|
if len(records) >= n:
|
|
break
|
|
if not raw.strip():
|
|
continue
|
|
try:
|
|
records.append(json.loads(raw.decode('utf-8')))
|
|
except Exception:
|
|
pass
|
|
records.reverse()
|
|
except Exception:
|
|
return []
|
|
return records
|
|
|
|
|
|
# ----------------------------------------------------------------------- CLI
|
|
|
|
def _cli(args):
|
|
"""Run `runlog.py [n]` and print the last n records as one-liners."""
|
|
count = 20
|
|
if args:
|
|
try:
|
|
count = int(args[0])
|
|
except (TypeError, ValueError):
|
|
count = 20
|
|
if count <= 0:
|
|
count = 20
|
|
for entry in tail(count):
|
|
print('{0} | {1} | {2} | {3} | {4} | {5}'.format(
|
|
entry.get('ts'), entry.get('button'), entry.get('outcome'),
|
|
entry.get('marks'), entry.get('report_path'),
|
|
entry.get('error')))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
_cli(sys.argv[1:])
|