267 lines
12 KiB
Python
267 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Injects pseudonym-based matching into every Diagnostik-App Shiny app.R.
|
|
|
|
Usage (run from the "formr" folder that directly contains all the app
|
|
subfolders, next to server.py / get_pseudo.R / pseudonyme.db):
|
|
|
|
python inject_pseudonym.py # dry run: reports what WOULD happen
|
|
python inject_pseudonym.py --apply # actually writes the changes
|
|
|
|
Finds every "<app-folder>/app.R" itself (no shell globbing needed, so this
|
|
works the same from Windows cmd.exe as from bash). Idempotent (marker-based -
|
|
safe to rerun), makes a .bak next to each file it actually changes, and
|
|
leaves a file completely untouched if any of the required anchors can't be
|
|
found unambiguously - such files are reported with a reason instead of being
|
|
guessed at.
|
|
"""
|
|
import re
|
|
import shutil
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
MARKER = "# --- pseudonym-support-injection v1 ---"
|
|
|
|
|
|
def inject(path, write=True):
|
|
path = str(path)
|
|
raw = open(path, encoding="utf-8", newline="").read()
|
|
uses_crlf = raw.count("\r\n") > raw.count("\n") - raw.count("\r\n")
|
|
text = raw.replace("\r\n", "\n")
|
|
if MARKER in text:
|
|
return "skip: already patched", text
|
|
|
|
orig = text
|
|
warnings = []
|
|
notes = []
|
|
|
|
# 1. UI field: a pseudonym textInput with a small italic hint instead of
|
|
# a bulky bold label. If textInput("chiffre", ...) is directly preceded
|
|
# by a simple single-purpose `div(style = "...",` wrapper (the common
|
|
# shape), clone a sibling div so both fields sit side by side in the
|
|
# same flex row instead of stacking. Otherwise fall back to a plain
|
|
# inline sibling insert (functional, just possibly stacked) and note it.
|
|
label_block = (
|
|
'tagList(\n'
|
|
'{i} "Pseudonym",\n'
|
|
'{i} tags$span(style = "font-weight: normal; font-style: italic; '
|
|
'font-size: 0.78em; color: #888; margin-left: 4px; white-space: nowrap;",\n'
|
|
'{i} "optional, hat Vorrang vor Chiffre")\n'
|
|
'{i} )'
|
|
)
|
|
|
|
m_wrapped = re.search(
|
|
r'^([ \t]*)div\(style\s*=\s*"[^"]*"\s*,\s*\n([ \t]*)textInput\("chiffre"',
|
|
text, re.MULTILINE,
|
|
)
|
|
if m_wrapped:
|
|
outer_indent, inner_indent = m_wrapped.group(1), m_wrapped.group(2)
|
|
label = label_block.format(i=inner_indent)
|
|
insertion = (
|
|
f'{outer_indent}div(style = "min-width: 360px; white-space: nowrap;",\n'
|
|
f'{inner_indent}textInput("pseudonym",\n'
|
|
f'{inner_indent} label = {label},\n'
|
|
f'{inner_indent} placeholder = "optional", width = "340px")\n'
|
|
f'{outer_indent}),\n'
|
|
)
|
|
text = text[: m_wrapped.start()] + insertion + text[m_wrapped.start() :]
|
|
else:
|
|
m = re.search(r'^([ \t]*)textInput\("chiffre"', text, re.MULTILINE)
|
|
if not m:
|
|
warnings.append('UI: textInput("chiffre" not found')
|
|
else:
|
|
indent = m.group(1)
|
|
label = label_block.format(i=indent)
|
|
insertion = (
|
|
f'{indent}textInput("pseudonym",\n'
|
|
f'{indent} label = {label},\n'
|
|
f'{indent} placeholder = "optional", width = "340px"),\n'
|
|
)
|
|
text = text[: m.start()] + insertion + text[m.start() :]
|
|
notes.append(
|
|
"UI: no simple preceding div(style=...) wrapper found around "
|
|
"textInput(\"chiffre\" - pseudonym field inserted inline; layout "
|
|
"may need a manual look (may stack instead of sitting side by side)"
|
|
)
|
|
|
|
# 2. URL prefill: new observe() block right after the server function
|
|
# signature, independent of/in addition to the existing chiffre one.
|
|
m = re.search(r"server\s*=\s*function\(input,\s*output,\s*session\)\s*\{\n", text)
|
|
if not m:
|
|
warnings.append("server = function(input, output, session) { not found")
|
|
else:
|
|
block = (
|
|
f" {MARKER}\n"
|
|
" observe({\n"
|
|
" query = parseQueryString(session$clientData$url_search)\n"
|
|
" if (!is.null(query$pseudonym) && nchar(trimws(query$pseudonym)) > 0) {\n"
|
|
' updateTextInput(session, "pseudonym", value = trimws(query$pseudonym))\n'
|
|
" }\n"
|
|
" })\n\n"
|
|
)
|
|
idx = m.end()
|
|
text = text[:idx] + block + text[idx:]
|
|
|
|
# Some apps run their chiffre-empty/format checks inside a dedicated
|
|
# top-level helper function (e.g. desii_validiere_chiffre(),
|
|
# validiere_chiffre()) defined OUTSIDE server() - such a function has no
|
|
# `input` in scope. Blindly rewriting a check there would compile fine
|
|
# but throw "object 'input' not found" at runtime on every single click,
|
|
# for every patient, whether pseudonym is used or not. So items 3 and 4
|
|
# only ever touch occurrences that are reachable from inside server()
|
|
# (i.e. at/after the server function's opening brace); anything found
|
|
# only earlier in the file is left completely untouched and reported as
|
|
# a note instead - pseudonym-only entry (no chiffre at all) just won't
|
|
# work for that one app, but the chiffre+pseudonym disambiguation (the
|
|
# actual bug this whole thing fixes) is unaffected either way.
|
|
m_anchor = re.search(r"server\s*=\s*function\(input,\s*output,\s*session\)\s*\{", text)
|
|
reactive_start = m_anchor.start() if m_anchor else None
|
|
|
|
def patch_reachable(text, target, replacement):
|
|
matches = list(re.finditer(re.escape(target), text))
|
|
reachable = [
|
|
mm for mm in matches
|
|
if reactive_start is not None and mm.start() >= reactive_start
|
|
]
|
|
for mm in reversed(reachable):
|
|
text = text[: mm.start()] + replacement + text[mm.end() :]
|
|
return text, len(matches), len(reachable)
|
|
|
|
# 3. Relax the "chiffre leer" gate - phrasing varies (some apps check
|
|
# `chiffre`, others read into `chiffre_roh` first).
|
|
empty_gate_variants = [
|
|
("nchar(chiffre) == 0",
|
|
"(nchar(trimws(input$pseudonym)) == 0 && nchar(chiffre) == 0)"),
|
|
("nchar(trimws(chiffre_roh)) == 0",
|
|
"(nchar(trimws(input$pseudonym)) == 0 && nchar(trimws(chiffre_roh)) == 0)"),
|
|
('trimws(chiffre_roh) == ""',
|
|
'(nchar(trimws(input$pseudonym)) == 0 && trimws(chiffre_roh) == "")'),
|
|
]
|
|
any_empty_gate_patched = False
|
|
for target, replacement in empty_gate_variants:
|
|
text, total, reachable = patch_reachable(text, target, replacement)
|
|
if reachable:
|
|
any_empty_gate_patched = True
|
|
if not any_empty_gate_patched:
|
|
notes.append(
|
|
"no reachable chiffre-empty gate found (the format-check gate below "
|
|
"usually already covers an empty chiffre too, so this is often fine)"
|
|
)
|
|
|
|
# 4. Relax the chiffre-format gate.
|
|
target = 'grepl("^[A-Z][0-9]{6}$", chiffre)'
|
|
replacement = '(nchar(trimws(input$pseudonym)) > 0 || grepl("^[A-Z][0-9]{6}$", chiffre))'
|
|
text, total, reachable = patch_reachable(text, target, replacement)
|
|
if reachable == 0:
|
|
if total > 0:
|
|
notes.append(
|
|
"chiffre-format check only exists inside a top-level helper function "
|
|
"with no access to the pseudonym input - pseudonym-only entry (no "
|
|
"chiffre at all) won't work for this app, but chiffre+pseudonym "
|
|
"disambiguation (the main fix) is unaffected"
|
|
)
|
|
else:
|
|
warnings.append(f'gate not found: "{target}"')
|
|
|
|
# 5. Backfill `chiffre` from the pseudonym right after the pseudonym
|
|
# script is sourced (pseudo now exists in .GlobalEnv - guaranteed
|
|
# name, set by the one shared get_pseudo.R).
|
|
target = "source(PFAD_PSEUDONYM_SKRIPT, local = FALSE)"
|
|
occurrences = [mm.start() for mm in re.finditer(re.escape(target), text)]
|
|
if len(occurrences) != 1:
|
|
warnings.append(
|
|
f'expected exactly 1 occurrence of source(PFAD_PSEUDONYM_SKRIPT...), found {len(occurrences)}'
|
|
)
|
|
else:
|
|
pos = occurrences[0]
|
|
line_start = text.rfind("\n", 0, pos) + 1
|
|
indent = re.match(r"[ \t]*", text[line_start:pos]).group(0)
|
|
end = pos + len(target)
|
|
backfill = (
|
|
f"\n{indent}if (nchar(trimws(input$pseudonym)) > 0) {{\n"
|
|
f'{indent} .pw_wert = trimws(input$pseudonym)\n'
|
|
f'{indent} .pw_tab = get("pseudo", envir = .GlobalEnv)\n'
|
|
f"{indent} .pw_treffer = .pw_tab[.pw_tab$pseudonym == .pw_wert, ]\n"
|
|
f"{indent} if (nrow(.pw_treffer) > 0) chiffre = toupper(trimws(.pw_treffer$chiffre[1]))\n"
|
|
f"{indent}}}"
|
|
)
|
|
text = text[:end] + backfill + text[end:]
|
|
|
|
# 6. Override the final session-id(s) with the exact pseudonym, so an
|
|
# ambiguous chiffre (multiple pseudonyms) doesn't silently pick the
|
|
# wrong / most-recent entry.
|
|
sid_pattern = re.compile(
|
|
r"^([ \t]*)([A-Za-z_][A-Za-z0-9_.]*)\s*=\s*"
|
|
r"(unique\(as\.character\(\w+\$pseudonym\)\)|unique\(\w+\$pseudonym\)|\w+\$pseudonym\[1\])[ \t]*$",
|
|
re.MULTILINE,
|
|
)
|
|
matches = list(sid_pattern.finditer(text))
|
|
if len(matches) == 0:
|
|
warnings.append("session-id assignment (X = unique(Y$pseudonym) / Y$pseudonym[1]) not found")
|
|
else:
|
|
for mm in reversed(matches):
|
|
indent, varname = mm.group(1), mm.group(2)
|
|
override = f"\n{indent}if (nchar(trimws(input$pseudonym)) > 0) {varname} = trimws(input$pseudonym)"
|
|
text = text[: mm.end()] + override + text[mm.end() :]
|
|
|
|
if warnings:
|
|
return "skip: " + "; ".join(warnings), orig
|
|
|
|
out = text.replace("\n", "\r\n") if uses_crlf else text
|
|
|
|
if write:
|
|
shutil.copy(path, path + ".bak")
|
|
with open(path, "w", encoding="utf-8", newline="") as f:
|
|
f.write(out)
|
|
|
|
status = "patched"
|
|
if notes:
|
|
status += " (note: " + "; ".join(notes) + ")"
|
|
return status, out
|
|
|
|
|
|
def main():
|
|
args = sys.argv[1:]
|
|
apply_changes = "--apply" in args
|
|
args = [a for a in args if a != "--apply"]
|
|
|
|
if args:
|
|
paths = [Path(a) for a in args]
|
|
else:
|
|
paths = sorted(Path(".").glob("*/app.R"))
|
|
|
|
if not paths:
|
|
print('No "*/app.R" files found in the current directory. '
|
|
"Run this from the folder that directly contains all the app "
|
|
"subfolders (next to server.py / get_pseudo.R / pseudonyme.db), "
|
|
"or pass explicit paths.")
|
|
sys.exit(1)
|
|
|
|
print(f"{'DRY RUN (no files will be changed - pass --apply to write)' if not apply_changes else 'APPLYING CHANGES'}")
|
|
print(f"{len(paths)} app(s) found.\n")
|
|
|
|
counts = {"patched": 0, "patched-with-note": 0, "skip-already-patched": 0, "skip": 0}
|
|
for p in paths:
|
|
try:
|
|
status, _ = inject(p, write=apply_changes)
|
|
except Exception as e:
|
|
status = f"skip: UNEXPECTED ERROR ({type(e).__name__}: {e}) - file left untouched, please report this"
|
|
if status.startswith("skip: already patched"):
|
|
counts["skip-already-patched"] += 1
|
|
elif status.startswith("skip:"):
|
|
counts["skip"] += 1
|
|
elif "(note:" in status:
|
|
counts["patched-with-note"] += 1
|
|
else:
|
|
counts["patched"] += 1
|
|
print(f"{p}: {status}")
|
|
|
|
print(
|
|
f"\n{counts['patched']} clean, {counts['patched-with-note']} patched-with-note, "
|
|
f"{counts['skip']} skipped, {counts['skip-already-patched']} already patched."
|
|
)
|
|
if not apply_changes:
|
|
print("This was a dry run - nothing was written. Rerun with --apply once you're happy.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|