91 lines
3.3 KiB
Python
91 lines
3.3 KiB
Python
"""db_bridge_windows.py – Windows watchdog pro dotazy z Linux VM.
|
||
|
||
Spusť jednou na Windows:
|
||
python db_bridge_windows.py
|
||
|
||
Skript sleduje soubor query_request.json ve stejné složce.
|
||
Jakmile ho najde, spustí SQL dotaz proti Medicusu a zapíše výsledek
|
||
do query_response.json. Pak čeká na další dotaz.
|
||
|
||
Ukonči: Ctrl+C
|
||
"""
|
||
import fdb, json, time, os, traceback, datetime
|
||
|
||
# ── Konfigurace ───────────────────────────────────────────────────────────────
|
||
DSN = r'localhost:c:\medicus 3\data\medicus.fdb'
|
||
USER = 'SYSDBA'
|
||
PASSWORD = 'masterkey'
|
||
CHARSET = 'win1250'
|
||
|
||
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')
|
||
POLL_SEC = 0.5
|
||
|
||
# ── Pomocné funkce ────────────────────────────────────────────────────────────
|
||
|
||
def serialize(val):
|
||
"""Převede Python hodnoty na JSON-serializovatelné typy."""
|
||
if isinstance(val, (datetime.date, datetime.datetime)):
|
||
return val.isoformat()
|
||
if isinstance(val, datetime.time):
|
||
return val.isoformat()
|
||
if isinstance(val, bytes):
|
||
return f'<bytes len={len(val)}>'
|
||
return val
|
||
|
||
|
||
def run_query(sql, params=None):
|
||
conn = fdb.connect(dsn=DSN, user=USER, password=PASSWORD, charset=CHARSET)
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute(sql, params or [])
|
||
columns = [d[0] for d in cur.description] if cur.description else []
|
||
rows = [[serialize(v) for v in row] for row in cur.fetchall()]
|
||
return {'status': 'ok', 'columns': columns, 'rows': rows, 'error': None}
|
||
except Exception as e:
|
||
return {'status': 'error', 'columns': [], 'rows': [], 'error': str(e)}
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
# ── Hlavní smyčka ─────────────────────────────────────────────────────────────
|
||
print(f'DB Bridge spuštěn. Sleduju: {REQUEST}')
|
||
print('Ukončení: Ctrl+C\n')
|
||
|
||
while True:
|
||
try:
|
||
if os.path.exists(REQUEST):
|
||
print(f'[{datetime.datetime.now().strftime("%H:%M:%S")}] Přijat dotaz...')
|
||
|
||
with open(REQUEST, 'r', encoding='utf-8') as f:
|
||
req = json.load(f)
|
||
|
||
os.remove(REQUEST)
|
||
|
||
sql = req.get('sql', '')
|
||
params = req.get('params', [])
|
||
req_id = req.get('id', '')
|
||
|
||
result = run_query(sql, params)
|
||
result['id'] = req_id
|
||
result['sql'] = sql
|
||
|
||
with open(RESPONSE, 'w', encoding='utf-8') as f:
|
||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||
|
||
if result['status'] == 'ok':
|
||
print(f' → OK, {len(result["rows"])} řádků')
|
||
else:
|
||
print(f' → CHYBA: {result["error"]}')
|
||
|
||
time.sleep(POLL_SEC)
|
||
|
||
except KeyboardInterrupt:
|
||
print('\nDB Bridge ukončen.')
|
||
break
|
||
except Exception as e:
|
||
print(f'Neočekávaná chyba: {e}')
|
||
traceback.print_exc()
|
||
time.sleep(2)
|