252 lines
11 KiB
Python
252 lines
11 KiB
Python
def categorize_status(status):
|
|
s = (status or "").strip().lower()
|
|
if s in SUCCESS_STATUSES:
|
|
return "success"
|
|
if s in FAILED_STATUSES:
|
|
return "failed"
|
|
return "attention"
|
|
|
|
|
|
# ── Progress heuristic: map printed markers -> step index + % ─────────────────
|
|
# (run_multi is one long call, so we infer progress from the lines it prints.)
|
|
STEP_MARKERS = [
|
|
("multi-company / multi-location", 0, 5),
|
|
("[input] login sheet", 1, 12),
|
|
("########## company", 2, 25),
|
|
("[choose company]", 2, 32),
|
|
("practice home", 3, 45),
|
|
("clearinghouse", 4, 55),
|
|
("unreviewed", 5, 62),
|
|
("posting", 5, 70),
|
|
("processed successfully", 5, 74),
|
|
("test-notposted", 5, 74),
|
|
("medicalperformer", 6, 82), # ERA -> EOB (PDF) phase
|
|
("era import ->", 6, 84),
|
|
("print-to-pdf", 6, 84),
|
|
("[all done]", 7, 100),
|
|
]
|
|
|
|
STEPS = [
|
|
"Initializing",
|
|
"Reading Excel",
|
|
"Login to Tebra",
|
|
"Select company",
|
|
"Open Clearinghouse",
|
|
"Post payments",
|
|
"ERA → EOB conversion",
|
|
"Completed",
|
|
]
|
|
|
|
|
|
# ── Friendly log translation ──────────────────────────────────────────────────
|
|
# The on-screen log is read by medical-billing users, not developers, so raw
|
|
# automation lines are translated to plain English here. Every RAW line is
|
|
# still stored in full in the BatchLog table (per batch) for debugging.
|
|
#
|
|
# Rules are (marker-substring, friendly-message, type) matched on the
|
|
# lowercased line, first match wins. friendly-message None = keep the line off
|
|
# the screen. Unmatched lines are hidden too, UNLESS they are errors — a real
|
|
# failure must never be invisible.
|
|
# IMPORTANT: markers are matched by SUBSTRING against the lowercased line, first
|
|
# match wins, so they MUST be specific enough not to catch unrelated lines. e.g.
|
|
# a bare "clearinghouse" wrongly matched "Total unreviewed clearinghouse reports:
|
|
# 10852" and produced a duplicate "Opening Clearinghouse" — always anchor on the
|
|
# distinct "[step N]" / "[medicalperformer] <verb>" prefix instead of a bare word.
|
|
_FRIENDLY_RULES = [
|
|
("multi-company / multi-location", "Starting the automation…", "info"),
|
|
("[input] login sheet", "Reading the Excel input file…", "info"),
|
|
("[choose company] screen", "Selecting the company in Tebra…", "info"),
|
|
("[login] done", "Logged in to Tebra.", "success"),
|
|
("[step 3] confirmed", "Practice opened.", "info"),
|
|
("[step 4+5] encounters", "Opening Clearinghouse Reports…", "info"),
|
|
("[step 6] checking", "Searching for unreviewed ERAs…", "info"),
|
|
("[step 9] downloading + posting", "Checking for ERAs to post…", "info"),
|
|
("no more unprocessed era rows", "No more ERAs to post here.", "info"),
|
|
("grid still not found", "The ERA list did not load — moving on.", "error"),
|
|
("grid not found", "The ERA list is slow to load — retrying…", "warn"),
|
|
("[step 9] downloaded:", "ERA file downloaded.", "info"),
|
|
("download failed", "Could not download the ERA file — continuing.", "warn"),
|
|
("didn't open", "The ERA is slow to open — trying again…", "warn"),
|
|
("still could not open view", "Could not open the ERA — marked as Error.", "error"),
|
|
("clicking 'post payment", "Clicking Post Payment…", "info"),
|
|
("post payment button not found", "Post Payment button not found.", "error"),
|
|
("-> era errors screen", "Tebra reported ERA Errors for this payment — cancelling it.", "warn"),
|
|
("-> payment screen", "Payment screen opened.", "info"),
|
|
("post-payment retry", "The payment screen didn't load — trying again…", "warn"),
|
|
("unrecognized screen after post payment", "Unexpected screen after Post Payment — retrying…", "warn"),
|
|
("batch # field / reference # not ready", "Batch # field not ready — retrying…", "warn"),
|
|
("batch# field or reference# value not found", "Batch # field not found — marked as Error.", "error"),
|
|
("filling batch #", "Entering the Batch number…", "info"),
|
|
("test mode — would have posted", "Test mode: payment ready — clicking Cancel instead of Save.", "warn"),
|
|
("save button not found", "Save button element not found.", "error"),
|
|
("[step 9] posted era", "Payment saved — ERA posted.", "success"),
|
|
("done posting era payments", "Finished posting for this practice.", "info"),
|
|
("[medicalperformer] launching", "Converting ERA files to EOB PDFs…", "info"),
|
|
("[medicalperformer] no .txt", "No ERA files to convert.", "info"),
|
|
("[done] no-location company complete", "Company finished.", "success"),
|
|
# The final line is emitted by Python (_run_automation) from the real run
|
|
# summary — hide the raw markers so a run that FAILED can never print a
|
|
# hardcoded green "completed".
|
|
("[run summary]", None, "info"),
|
|
("[all done]", None, "info"),
|
|
]
|
|
|
|
# Dynamic rules that pull a value out of the line (checked before the static
|
|
# list). Each is (compiled-regex, lambda match -> (msg, type)).
|
|
_FRIENDLY_DYNAMIC = [
|
|
(re.compile(r"#+\s*COMPANY\s+(\d+)/(\d+):\s*(.+?)\s*#*\s*$", re.I),
|
|
lambda m: (f"Working on company {m.group(1)} of {m.group(2)}: "
|
|
f"{m.group(3)}", "info")),
|
|
(re.compile(r"PROCESSING LOCATION:\s*(.+)", re.I),
|
|
lambda m: (f"Working on location: {m.group(1).strip()}", "info")),
|
|
# Login failure — make it clearly visible (the COM/network error the user
|
|
# sees when Tebra pops "connection error" and they click OK).
|
|
(re.compile(r"\[login\] FAILED for company '(.+?)':", re.I),
|
|
lambda m: (f"Could not log in to Tebra for {m.group(1)} — connection or "
|
|
f"Tebra error. This company was not processed.", "error")),
|
|
(re.compile(r"Unreviewed reports total\s*:\s*(\d+)", re.I),
|
|
lambda m: (f"Found {m.group(1)} unreviewed report(s) to check.", "info")),
|
|
(re.compile(r"\[Step 9\] Opening ERA row id=(\S+)", re.I),
|
|
lambda m: (f"Opening ERA {m.group(1)}…", "info")),
|
|
(re.compile(r"Check ID=(\S*)\s+Date=(\S*)\s+Method=.*Amount=(\S*)", re.I),
|
|
lambda m: (f"Check {m.group(1) or '—'} · Date {m.group(2) or '—'} · "
|
|
f"Amount {m.group(3) or '—'}", "info")),
|
|
(re.compile(r"\[Step 9\] Skipping\s+\S+:\s*(.+)", re.I),
|
|
lambda m: (f"Skipped — {m.group(1).strip()}", "warn")),
|
|
(re.compile(r"\[Done\] Location '(.+?)' complete", re.I),
|
|
lambda m: (f"Location finished: {m.group(1)}", "success")),
|
|
]
|
|
|
|
|
|
# Error-level lines that are pure developer noise — the friendly login-error
|
|
# line and the Python run-summary already tell the user what failed, so these
|
|
# raw dumps are kept in the BatchLog table but never shown on screen.
|
|
_ERROR_NOISE = (
|
|
"traceback (most recent call last)",
|
|
"during handling of the above",
|
|
"another exception occurred",
|
|
"comerror",
|
|
"[error] company '", # per-company COM dump — summary covers it
|
|
)
|
|
|
|
|
|
def _is_error_noise(low_line):
|
|
return any(marker in low_line for marker in _ERROR_NOISE)
|
|
|
|
|
|
def friendly_log_line(line):
|
|
"""Translate one raw automation line for the user-facing log.
|
|
Returns (message, type) to display, or None to keep it off the screen."""
|
|
for rx, fmt in _FRIENDLY_DYNAMIC:
|
|
m = rx.search(line)
|
|
if m:
|
|
return fmt(m)
|
|
low = line.lower()
|
|
for marker, msg, ltype in _FRIENDLY_RULES:
|
|
if marker in low:
|
|
return (msg, ltype) if msg else None
|
|
return None
|
|
|
|
|
|
class _StreamToBridge:
|
|
"""A stdout/stderr replacement that tees to the real stream AND forwards
|
|
every completed line to the UI as a log event."""
|
|
|
|
def __init__(self, bridge, original, level="info"):
|
|
self.bridge = bridge
|
|
self.original = original
|
|
self.level = level
|
|
self._buf = ""
|
|
|
|
def write(self, text):
|
|
try:
|
|
if self.original:
|
|
self.original.write(text)
|
|
except Exception:
|
|
pass
|
|
self._buf += text
|
|
while "\n" in self._buf:
|
|
line, self._buf = self._buf.split("\n", 1)
|
|
self._emit_line(line)
|
|
|
|
def _emit_line(self, line):
|
|
line = line.rstrip("\r")
|
|
if not line.strip():
|
|
return
|
|
self.bridge._on_automation_line(line, self.level)
|
|
|
|
def flush(self):
|
|
try:
|
|
if self.original:
|
|
self.original.flush()
|
|
except Exception:
|
|
pass
|
|
|
|
def isatty(self):
|
|
return False
|
|
|
|
|
|
class Bridge(QObject):
|
|
"""Exposed to JavaScript as 'pyBridge'."""
|
|
|
|
js_message = pyqtSignal(str)
|
|
|
|
def __init__(self, window):
|
|
super().__init__()
|
|
self.window = window
|
|
self.username = "User"
|
|
self._max_pct = 0
|
|
self._cur_step = 0
|
|
self.batch_id = None # set on a fresh Trigger or from a resume check
|
|
self._batch_log = None # BatchLogWriter while a run is active
|
|
self._run_log_file = None # plain-text run log (DB-independent) while running
|
|
self._last_friendly = None # dedupe consecutive identical friendly lines
|
|
self._auto_proc = None # the separate automation process (bug-094) while running
|
|
|
|
# ── Python -> JS ────────────────────────────────────────────────────────────
|
|
def send_to_js(self, event, data):
|
|
self.js_message.emit(json.dumps({"event": event, "data": data}))
|
|
|
|
def _log(self, msg, type="info"):
|
|
self.send_to_js("log", {"msg": msg, "type": type})
|
|
|
|
def _step(self, label, pct):
|
|
self.send_to_js("step", {"label": label, "pct": pct})
|
|
|
|
def _batch_chip(self, batch_id, live=False):
|
|
"""Persistent Run-tab chip showing the batch this run is tracked under.
|
|
`batch_id` None/0 CLEARS it. A stale number is worse than none, so every
|
|
site that assigns self.batch_id must call this in the same breath."""
|
|
self.send_to_js("batch_state", {
|
|
"batch_id": str(batch_id) if batch_id else "",
|
|
"live": bool(live)})
|
|
|
|
# ── JS -> Python ────────────────────────────────────────────────────────────
|
|
@pyqtSlot(str)
|
|
def log(self, msg):
|
|
print(f"[JS] {msg}")
|
|
|
|
@pyqtSlot(bool)
|
|
def set_dark(self, on):
|
|
"""Persist the dark-mode choice so the login page (a separate window)
|
|
can pick it up too."""
|
|
try:
|
|
from src.ui.prefs import set_dark
|
|
set_dark(on)
|
|
except Exception as e:
|
|
print(f"[prefs] could not save dark mode: {e}")
|
|
|
|
@pyqtSlot()
|
|
def logout(self):
|
|
"""JS clicked Logout — defer to the next event-loop tick so we don't
|
|
run a nested dialog from inside the web-channel callback."""
|
|
from PyQt6.QtCore import QTimer
|
|
QTimer.singleShot(0, self.window.do_logout)
|
|
|
|
@pyqtSlot()
|
|
def browse_file(self):
|
|
path, _ = QFileDialog.getOpenFileName(
|
|
self.window, "Select Input Excel File",
|
|
os.path.expanduser("~"),
|
|
"Excel files (*.xlsx *.xlsm *.xls)"
|