notebook vb

MedicusWithClaudeSelects/db_bridge_vm.py

@@ -0,0 +1,78 @@
+"""db_bridge_vm.py – VM strana souborového mostu k Medicusu.
+
+Použití z Linuxu:
+    from db_bridge_vm import query
+    rows, columns = query("SELECT COUNT(*) FROM KAR")
+    print(columns, rows)
+"""
+import json, time, os, uuid
+
+BRIDGE_DIR = os.path.dirname(os.path.abspath(__file__))
+REQUEST    = os.path.join(BRIDGE_DIR, 'query_request.json')
+RESPONSE   = os.path.join(BRIDGE_DIR, 'query_response.json')
+
+TIMEOUT_SEC = 30
+POLL_SEC    = 0.3
+
+
+def query(sql, params=None, timeout=TIMEOUT_SEC):
+    """Pošle SQL dotaz přes souborový most a vrátí (rows, columns).
+
+    Raises:
+        TimeoutError – watchdog neodpověděl do timeout sekund
+        RuntimeError – Firebird vrátil chybu
+    """
+    # Smaž případnou starou response
+    if os.path.exists(RESPONSE):
+        os.remove(RESPONSE)
+
+    req_id = uuid.uuid4().hex
+    req = {'id': req_id, 'sql': sql, 'params': params or []}
+
+    with open(REQUEST, 'w', encoding='utf-8') as f:
+        json.dump(req, f, ensure_ascii=False)
+
+    # Čekej na odpověď
+    waited = 0.0
+    while waited < timeout:
+        time.sleep(POLL_SEC)
+        waited += POLL_SEC
+        if os.path.exists(RESPONSE):
+            with open(RESPONSE, 'r', encoding='utf-8') as f:
+                resp = json.load(f)
+            os.remove(RESPONSE)
+            if resp.get('status') == 'error':
+                raise RuntimeError(f"DB chyba: {resp.get('error')}")
+            return resp.get('rows', []), resp.get('columns', [])
+
+    # Timeout – smaž request aby watchdog nezpracoval zastaralý dotaz
+    if os.path.exists(REQUEST):
+        os.remove(REQUEST)
+    raise TimeoutError(f'Watchdog neodpověděl do {timeout}s – běží db_bridge_windows.py?')
+
+
+def query_print(sql, params=None):
+    """Spustí dotaz a vypíše výsledek přehledně."""
+    rows, cols = query(sql, params)
+    if not cols:
+        print('(žádné sloupce)')
+        return rows, cols
+    col_w = [max(len(str(c)), max((len(str(r[i])) for r in rows), default=0))
+             for i, c in enumerate(cols)]
+    sep = '+' + '+'.join('-' * (w + 2) for w in col_w) + '+'
+    fmt = '|' + '|'.join(f' {{:<{w}}} ' for w in col_w) + '|'
+    print(sep)
+    print(fmt.format(*cols))
+    print(sep)
+    for row in rows:
+        print(fmt.format(*[str(v) if v is not None else 'NULL' for v in row]))
+    print(sep)
+    print(f'{len(rows)} řádků')
+    return rows, cols
+
+
+if __name__ == '__main__':
+    # Rychlý test
+    print('Testuji spojení...')
+    rows, cols = query('SELECT COUNT(*) AS POCET FROM KAR')
+    print(f'OK – pacientů v KAR: {rows[0][0]}')