Unlock a world of possibilities! Login now and discover the exclusive benefits awaiting you.
Hello,
I have been trying to compile a list of current stream and app permissions for all users directly from the PG DB of Qlik Sense. The permissions were set through "Custom properties".
So far, I have identified the following tables with the relevant information needed to get the desired result:
I have tried various ways to try and join them (in Qlik Sense script) but the CustomPropertyDefinitions table is throwing me off in its relation to SystemRules.
Can anyone tell me how to join these tables or if I am missing a bridge table of some sort?
Is it better to try and fetch the data through an SQL query instead?
Thank you.
Edit: Added table name.
Hello,
I (Claude AI actually), made a python script that retrieves the app and stream access privileges for users with a license.
What you need:
Note: This code contains a snippet to connect to an OracleDB as to retrieve "Section Access" data.
"""
Comprehensive Qlik Sense access audit report.
Combines four sources into one CSV (one row per User x Stream x App):
1. QRS REST -- users (with CPs + license type), streams (with CPs), apps
2. Qlik Engine (websocket) -- per-app load script, scanned for "section access"
3. Oracle DB -- YOUR_SECTION_ACCESS table, pivoted into per-facility flags
4. Admin role check -- flag users who are "App User" license but have any admin role
TLS NOTE: Qlik's self-generated root cert lacks the keyUsage extension and is
rejected by modern OpenSSL. We rely on the mutual client-certificate auth and
skip server-cert verification. Set VERIFY_TLS = True only if you've replaced
the Qlik root with a compliant one.
Output columns:
UserDirectory, UserId, UserName, UserInactive,
LicenseType, IsAdmin, AppUserButAdmin,
StreamName, AppName, AppPublished, AppUsesSectionAccess,
FACILITY_<n>... (one True/False column per facility seen in the Oracle table,
plus FACILITY_ALL for users with FACILITY_NO='*')
"""
import csv
import json
import ssl
import urllib3
from collections import defaultdict
import requests
import websocket
import oracledb
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# =========================
# CONFIG
# =========================
SERVER = "myqlikserver.company.com"
XRFKEY = "abcdefghij123nop"
USER_DIR = "company"
USER_ID = "qlikserviceuser"
CLIENT_CERT = ("client.pem", "client_key.pem")
# TLS: see module docstring. Set to a CA bundle path only if you have a
# RFC-5280-compliant Qlik root cert. Otherwise leave False.
VERIFY_TLS = False
# Engine WSS uses the same client cert pair in your environment
ENGINE_CERTFILE = "client.pem"
ENGINE_KEYFILE = "client_key.pem"
# Oracle connection (YOUR_SECTION_ACCESS table)
ORACLE_USER = "audit_reader"
ORACLE_PASSWORD = "REPLACE_ME"
ORACLE_DSN = "oracle-host:1521/SERVICE"
ORACLE_TABLE = "FACILITY_SECTION_ACCESS"
# Roles considered "admin" for the AppUserButAdmin check.
ADMIN_ROLES = {
"RootAdmin", "ContentAdmin", "DeploymentAdmin",
"AuditAdmin", "SecurityAdmin",
}
QRS_BASE = f"https://{SERVER}:4242/qrs"
OUT_CSV = "qlik_full_audit.csv"
# =========================
# HTTP HELPERS
# =========================
def qrs_headers():
return {
"X-Qlik-XrfKey": XRFKEY,
"X-Qlik-User": f"UserDirectory={USER_DIR};UserId={USER_ID}",
"Content-Type": "application/json",
"Accept": "application/json",
}
def qrs_get(path):
url = f"{QRS_BASE}{path}?xrfkey={XRFKEY}"
r = requests.get(url, headers=qrs_headers(),
cert=CLIENT_CERT, verify=VERIFY_TLS, timeout=180)
r.raise_for_status()
return r.json()
# =========================
# 1. QRS: users / streams / apps / licenses
# =========================
def cp_set(entity):
return {
(cp["definition"]["name"], cp["value"])
for cp in (entity.get("customProperties") or [])
}
def fetch_qrs_data():
print("QRS: users ...")
users = qrs_get("/user/full")
print(f" {len(users)}")
print("QRS: streams ...")
streams = qrs_get("/stream/full")
print(f" {len(streams)}")
print("QRS: apps ...")
apps = qrs_get("/app/full")
print(f" {len(apps)}")
print("QRS: license assignments ...")
license_by_user = {}
for endpoint, label in [
("/license/professionalaccesstype/full", "Professional"),
("/license/analyzeraccesstype/full", "Analyzer"),
("/license/useraccesstype/full", "User"),
("/license/loginaccessusage/full", "Login"),
]:
try:
for la in qrs_get(endpoint):
uid = (la.get("user") or {}).get("id")
if uid:
license_by_user[uid] = label
except requests.HTTPError as e:
print(f" {endpoint}: skipped ({e})")
print(f" {len(license_by_user)} users with assigned licenses")
return users, streams, apps, license_by_user
# =========================
# 2. Engine: section-access detection per app
# =========================
def detect_section_access(app_ids):
"""Return {app_id: True / False / None}. None = could not open / no script."""
out = {}
# Build SSL options once. CERT_NONE matches VERIFY_TLS=False above.
if VERIFY_TLS:
sslopt = {
"certfile": ENGINE_CERTFILE,
"keyfile": ENGINE_KEYFILE,
"ca_certs": VERIFY_TLS,
"cert_reqs": ssl.CERT_REQUIRED,
}
else:
sslopt = {
"certfile": ENGINE_CERTFILE,
"keyfile": ENGINE_KEYFILE,
"cert_reqs": ssl.CERT_NONE,
}
for i, app_id in enumerate(app_ids, 1):
try:
ws = websocket.create_connection(
f"wss://{SERVER}:4747/app/",
sslopt=sslopt,
header=[f"X-Qlik-User: UserDirectory={USER_DIR};UserId={USER_ID}"],
timeout=60,
)
ws.recv() # OnConnected
ws.send(json.dumps({
"jsonrpc": "2.0", "id": 1, "method": "OpenDoc",
"handle": -1, "params": [app_id],
}))
msg = None
while True:
msg = json.loads(ws.recv())
if msg.get("id") == 1:
break
if "result" not in msg:
out[app_id] = None
ws.close()
continue
handle = msg["result"]["qReturn"]["qHandle"]
ws.send(json.dumps({
"jsonrpc": "2.0", "id": 2, "method": "GetScript",
"handle": handle, "params": {},
}))
while True:
msg = json.loads(ws.recv())
if msg.get("id") == 2:
break
if "result" in msg:
script = msg["result"].get("qScript") or ""
out[app_id] = "section access" in script.lower()
else:
out[app_id] = None
ws.close()
except Exception as e:
print(f" engine: {app_id} ERROR {e}")
out[app_id] = None
if i % 25 == 0:
print(f" scanned {i}/{len(app_ids)} apps")
return out
# =========================
# 3. Oracle: facility section access
# =========================
def fetch_facility_access():
"""
Returns:
facilities_by_user: { 'userdir\\userid' (lowercased) :
{ 'access': 'ADMIN'/'USER', 'facilities': set([...]) } }
all_facilities: sorted list of distinct facility numbers (without '*')
"""
print(f"Oracle: reading {ORACLE_TABLE} ...")
with oracledb.connect(user=ORACLE_USER, password=ORACLE_PASSWORD,
dsn=ORACLE_DSN) as conn:
cur = conn.cursor()
cur.execute(f'SELECT "ACCESS", "USERID", "FACILITY_NO" FROM {ORACLE_TABLE}')
rows = cur.fetchall()
facilities_by_user = defaultdict(lambda: {"access": None, "facilities": set()})
all_facilities = set()
for access, userid, fac in rows:
if userid is None:
continue
key = userid.strip().lower()
fac_s = (str(fac).strip() if fac is not None else "")
facilities_by_user[key]["access"] = (access or "").strip().upper() or None
facilities_by_user[key]["facilities"].add(fac_s)
if fac_s and fac_s != "*":
all_facilities.add(fac_s)
def sort_key(x):
try:
return (0, int(x))
except ValueError:
return (1, x)
sorted_facilities = sorted(all_facilities, key=sort_key)
print(f" {len(facilities_by_user)} users, {len(sorted_facilities)} facilities")
return facilities_by_user, sorted_facilities
def user_facility_flags(user, facilities_by_user, all_facilities):
udir = (user.get("userDirectory") or "").strip()
uid = (user.get("userId") or "").strip()
key = f"{udir}\\{uid}".lower()
entry = facilities_by_user.get(key)
flags = {"FACILITY_ALL": False}
for f in all_facilities:
flags[f"FACILITY_{f}"] = False
if not entry:
return flags
fac_set = entry["facilities"]
if "*" in fac_set:
flags["FACILITY_ALL"] = True
for f in all_facilities:
flags[f"FACILITY_{f}"] = True
else:
for f in fac_set:
col = f"FACILITY_{f}"
if col in flags:
flags[col] = True
return flags
# =========================
# 4. Admin / license flags
# =========================
def user_role_flags(user, license_by_user):
license_label = license_by_user.get(user.get("id"), "None")
roles = set(user.get("roles") or [])
is_admin = bool(roles & ADMIN_ROLES)
app_user = license_label != "Professional"
flag = is_admin and app_user
return license_label, is_admin, flag
# =========================
# MAIN
# =========================
def main():
users, streams, apps, license_by_user = fetch_qrs_data()
streams_by_cp = defaultdict(list)
for s in streams:
for cp in cp_set(s):
streams_by_cp[cp].append(s)
apps_by_stream = defaultdict(list)
for a in apps:
st = a.get("stream")
if st and st.get("id"):
apps_by_stream[st["id"]].append(a)
published_app_ids = [a["id"] for a in apps
if a.get("published") and a.get("stream")]
print(f"Engine: scanning {len(published_app_ids)} published apps for Section Access ...")
sa_by_app = detect_section_access(published_app_ids)
facilities_by_user, all_facilities = fetch_facility_access()
base_cols = [
"UserDirectory", "UserId", "UserName", "UserInactive",
"LicenseType", "IsAdmin", "AppUserButAdmin",
"StreamName", "AppName", "AppPublished", "AppUsesSectionAccess",
]
facility_cols = ["FACILITY_ALL"] + [f"FACILITY_{f}" for f in all_facilities]
fieldnames = base_cols + facility_cols
rows = []
for u in users:
user_cps = cp_set(u)
license_label, is_admin, app_user_but_admin = user_role_flags(u, license_by_user)
fac_flags = user_facility_flags(u, facilities_by_user, all_facilities)
seen_streams = {}
for cp in user_cps:
for s in streams_by_cp.get(cp, []):
seen_streams[s["id"]] = s
if not seen_streams:
row = {
"UserDirectory": u.get("userDirectory"),
"UserId": u.get("userId"),
"UserName": u.get("name"),
"UserInactive": u.get("inactive", False),
"LicenseType": license_label,
"IsAdmin": is_admin,
"AppUserButAdmin": app_user_but_admin,
"StreamName": "",
"AppName": "",
"AppPublished": "",
"AppUsesSectionAccess": "",
}
row.update(fac_flags)
rows.append(row)
continue
for s_id, stream in seen_streams.items():
stream_apps = apps_by_stream.get(s_id, [])
if not stream_apps:
row = {
"UserDirectory": u.get("userDirectory"),
"UserId": u.get("userId"),
"UserName": u.get("name"),
"UserInactive": u.get("inactive", False),
"LicenseType": license_label,
"IsAdmin": is_admin,
"AppUserButAdmin": app_user_but_admin,
"StreamName": stream.get("name"),
"AppName": "",
"AppPublished": "",
"AppUsesSectionAccess": "",
}
row.update(fac_flags)
rows.append(row)
continue
for app in stream_apps:
sa = sa_by_app.get(app["id"])
row = {
"UserDirectory": u.get("userDirectory"),
"UserId": u.get("userId"),
"UserName": u.get("name"),
"UserInactive": u.get("inactive", False),
"LicenseType": license_label,
"IsAdmin": is_admin,
"AppUserButAdmin": app_user_but_admin,
"StreamName": stream.get("name"),
"AppName": app.get("name"),
"AppPublished": app.get("published"),
"AppUsesSectionAccess": sa if sa is not None else "Unknown",
}
row.update(fac_flags)
rows.append(row)
with open(OUT_CSV, "w", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=fieldnames)
w.writeheader()
w.writerows(rows)
print(f"Wrote {len(rows)} rows -> {OUT_CSV}")
if __name__ == "__main__":
main()
Hope this helps provide some guidance/direction to anyone needing to get a detailed user audit.
If there are any other details needed I can try to help.
Hi,
I guess you can get all this information from Monitoring Apps like License Monitor or Operations Monitor.
Please check once
Thanks,
Ashutosh
Hello @AshutoshBhumkar,
Thank you for your reply.
The monitoring apps are insufficient because they do not show the actual permissions. For example:
If person JohnDoe accesses App A but does not access App B, then that is not logged and therefore App B does not appear next to the user.
Exporting from the audit tool in QMC shows "Yellow" coded "R"s or "Read"s as just another Read permission in the CSV. It does not differentiate between "Green" or "Yellow" permissions.
The only way to get the proper reading is with something like I am trying to do and read it directly from the DB.
Thanks.
I think you may be better off using the QRS API endpoints like /systemrule/security/audit
-Rob
Hello @rwunderlich,
Thank you for your suggestion but I don't know where to put or how to use APIs.
So far, (with some AI help), I managed the following query:
SQL
SELECT
u."Name" AS UserName,
u."UserId" AS UserDirectoryID,
u."UserDirectory",
CASE
WHEN prof."ID" IS NOT NULL THEN 'Professional'
WHEN anlz."ID" IS NOT NULL THEN 'Analyzer'
ELSE 'None/Login Access'
END AS LicenseType,
cpd."Name" AS PropertyName,
cpv."Value" AS PropertyValue,
CASE
WHEN last_usage_an."LastUsed" IS NOT NULL THEN MAX(last_usage_an."LastUsed")
WHEN last_usage_pr."LastUsed" IS NOT NULL THEN MAX(last_usage_pr."LastUsed")
ELSE NULL
END AS LastLoginDate
FROM "public"."Users" u
-- Join for Professional Licenses
LEFT JOIN "public"."LicenseProfessionalAccessTypes" prof ON u."ID" = prof."User_ID"
-- Join for Analyzer Licenses
LEFT JOIN "public"."LicenseAnalyzerAccessTypes" anlz ON u."ID" = anlz."User_ID"
-- Join for Custom Properties
LEFT JOIN "public"."CustomPropertyValues" cpv ON u."ID" = cpv."User_ID"
LEFT JOIN "public"."CustomPropertyDefinitions" cpd ON cpv."Definition_ID" = cpd."ID"
LEFT JOIN "public"."LicenseAnalyzerAccessTypes" last_usage_an ON u."ID" = last_usage_an."User_ID"
LEFT JOIN "public"."LicenseProfessionalAccessTypes" last_usage_pr ON u."ID" = last_usage_pr."User_ID"
WHERE u."RemovedExternally" = false
GROUP BY u."Name", u."UserId", u."UserDirectory", prof."ID", anlz."ID", cpd."Name", cpv."Value", last_usage_an."LastUsed", last_usage_pr."LastUsed";
The only thing about the result of this query, is that the relation between the streams and apps is vertical (in a column) instead of in a row.
I could extract it into an excel and manually finish the rest of the steps but I would like to have a complete solution.
Thank you.
Hello all,
I would like to update you on my progress:
I manage 2 environments.
The first one uses "Custom properties" and I posted the partial working solution and decided to edit the rest manually on excel after extracting the data.
The second one uses security rules. Using the following query directly on to the Qlik Sense PostgreSQL DB:
SELECT
u."ID" as UsersID,
u."UserId" as DirectoryUserID,
u."Name" as UserName,
s."ID" as StreamID,
s."Name" as StreamName,
a."ID" as AppID,
a."Name" as AppName,
l."LastUsed" as LastLogin,
CASE WHEN l."User_ID" IS NOT NULL THEN 'Analyzer' ELSE 'None' END AS LicenseType,
sr.*
FROM
(SELECT
sr."ID" AS systemrule_id,
sr."Name" AS systemrule_name,
resource_match[2] AS App_ID,
user_match[1] AS user_id
FROM public."SystemRules" sr
CROSS JOIN LATERAL
regexp_matches(sr."Rule", '\(\(.*?\)\)', 'g') AS block(rule_block)
CROSS JOIN LATERAL
regexp_match(block.rule_block[1], 'resource\.resourcetype\s*=\s*"App"') AS resource_type_match
CROSS JOIN LATERAL
regexp_match(block.rule_block[1], 'resource\.(id|name)\s*=\s*"([^"]+)"') AS resource_match
CROSS JOIN LATERAL
regexp_matches(block.rule_block[1], 'user\.userId\s*=\s*"([^"]+)"', 'g') AS user_match
WHERE sr."Type" = 'Custom' AND sr."Disabled" = 'No' AND CAST(sr."Actions" AS TEXT) LIKE '%2%') sr
LEFT JOIN public."Users" u ON u."UserId" = sr."user_id"
LEFT JOIN public."Apps" a ON (CAST(a."ID" AS TEXT) = CAST(sr.App_ID AS TEXT) OR a."Name" = sr.App_ID)
LEFT JOIN public."Streams" s ON s."ID" = a."Stream_ID"
LEFT JOIN public."LicenseAnalyzerAccessTypes" l ON l."User_ID" = u."ID"
WHERE u."RemovedExternally" = 'False' AND a."Published" = 'True'
This worked better than expected and there was no need to manually edit the excel file (other than formatting and coloring).
Note: This query assumes you use the directory "UserID" for the users and either the IDs or Names of your Apps.
Just posting the solution for the benefit of the community.
If I find a better solution for the "Custom properties" method, I will post that as well.
Thank you.
You’re on the right track focusing on CustomPropertyDefinitions and SystemRules, but the relationship can be tricky because rules reference custom properties indirectly through resource filters rather than simple foreign keys. Often, a bridge via CustomPropertyValues tied to users and streams/apps is required.
If joins become too complex in Qlik script, querying directly with SQL against the repository database may provide clearer relationship mapping. For tracking structured status data workflows, platforms like Buscador De Estatus show how organized status queries can simplify complex lookups.
Hello,
I (Claude AI actually), made a python script that retrieves the app and stream access privileges for users with a license.
What you need:
Note: This code contains a snippet to connect to an OracleDB as to retrieve "Section Access" data.
"""
Comprehensive Qlik Sense access audit report.
Combines four sources into one CSV (one row per User x Stream x App):
1. QRS REST -- users (with CPs + license type), streams (with CPs), apps
2. Qlik Engine (websocket) -- per-app load script, scanned for "section access"
3. Oracle DB -- YOUR_SECTION_ACCESS table, pivoted into per-facility flags
4. Admin role check -- flag users who are "App User" license but have any admin role
TLS NOTE: Qlik's self-generated root cert lacks the keyUsage extension and is
rejected by modern OpenSSL. We rely on the mutual client-certificate auth and
skip server-cert verification. Set VERIFY_TLS = True only if you've replaced
the Qlik root with a compliant one.
Output columns:
UserDirectory, UserId, UserName, UserInactive,
LicenseType, IsAdmin, AppUserButAdmin,
StreamName, AppName, AppPublished, AppUsesSectionAccess,
FACILITY_<n>... (one True/False column per facility seen in the Oracle table,
plus FACILITY_ALL for users with FACILITY_NO='*')
"""
import csv
import json
import ssl
import urllib3
from collections import defaultdict
import requests
import websocket
import oracledb
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# =========================
# CONFIG
# =========================
SERVER = "myqlikserver.company.com"
XRFKEY = "abcdefghij123nop"
USER_DIR = "company"
USER_ID = "qlikserviceuser"
CLIENT_CERT = ("client.pem", "client_key.pem")
# TLS: see module docstring. Set to a CA bundle path only if you have a
# RFC-5280-compliant Qlik root cert. Otherwise leave False.
VERIFY_TLS = False
# Engine WSS uses the same client cert pair in your environment
ENGINE_CERTFILE = "client.pem"
ENGINE_KEYFILE = "client_key.pem"
# Oracle connection (YOUR_SECTION_ACCESS table)
ORACLE_USER = "audit_reader"
ORACLE_PASSWORD = "REPLACE_ME"
ORACLE_DSN = "oracle-host:1521/SERVICE"
ORACLE_TABLE = "FACILITY_SECTION_ACCESS"
# Roles considered "admin" for the AppUserButAdmin check.
ADMIN_ROLES = {
"RootAdmin", "ContentAdmin", "DeploymentAdmin",
"AuditAdmin", "SecurityAdmin",
}
QRS_BASE = f"https://{SERVER}:4242/qrs"
OUT_CSV = "qlik_full_audit.csv"
# =========================
# HTTP HELPERS
# =========================
def qrs_headers():
return {
"X-Qlik-XrfKey": XRFKEY,
"X-Qlik-User": f"UserDirectory={USER_DIR};UserId={USER_ID}",
"Content-Type": "application/json",
"Accept": "application/json",
}
def qrs_get(path):
url = f"{QRS_BASE}{path}?xrfkey={XRFKEY}"
r = requests.get(url, headers=qrs_headers(),
cert=CLIENT_CERT, verify=VERIFY_TLS, timeout=180)
r.raise_for_status()
return r.json()
# =========================
# 1. QRS: users / streams / apps / licenses
# =========================
def cp_set(entity):
return {
(cp["definition"]["name"], cp["value"])
for cp in (entity.get("customProperties") or [])
}
def fetch_qrs_data():
print("QRS: users ...")
users = qrs_get("/user/full")
print(f" {len(users)}")
print("QRS: streams ...")
streams = qrs_get("/stream/full")
print(f" {len(streams)}")
print("QRS: apps ...")
apps = qrs_get("/app/full")
print(f" {len(apps)}")
print("QRS: license assignments ...")
license_by_user = {}
for endpoint, label in [
("/license/professionalaccesstype/full", "Professional"),
("/license/analyzeraccesstype/full", "Analyzer"),
("/license/useraccesstype/full", "User"),
("/license/loginaccessusage/full", "Login"),
]:
try:
for la in qrs_get(endpoint):
uid = (la.get("user") or {}).get("id")
if uid:
license_by_user[uid] = label
except requests.HTTPError as e:
print(f" {endpoint}: skipped ({e})")
print(f" {len(license_by_user)} users with assigned licenses")
return users, streams, apps, license_by_user
# =========================
# 2. Engine: section-access detection per app
# =========================
def detect_section_access(app_ids):
"""Return {app_id: True / False / None}. None = could not open / no script."""
out = {}
# Build SSL options once. CERT_NONE matches VERIFY_TLS=False above.
if VERIFY_TLS:
sslopt = {
"certfile": ENGINE_CERTFILE,
"keyfile": ENGINE_KEYFILE,
"ca_certs": VERIFY_TLS,
"cert_reqs": ssl.CERT_REQUIRED,
}
else:
sslopt = {
"certfile": ENGINE_CERTFILE,
"keyfile": ENGINE_KEYFILE,
"cert_reqs": ssl.CERT_NONE,
}
for i, app_id in enumerate(app_ids, 1):
try:
ws = websocket.create_connection(
f"wss://{SERVER}:4747/app/",
sslopt=sslopt,
header=[f"X-Qlik-User: UserDirectory={USER_DIR};UserId={USER_ID}"],
timeout=60,
)
ws.recv() # OnConnected
ws.send(json.dumps({
"jsonrpc": "2.0", "id": 1, "method": "OpenDoc",
"handle": -1, "params": [app_id],
}))
msg = None
while True:
msg = json.loads(ws.recv())
if msg.get("id") == 1:
break
if "result" not in msg:
out[app_id] = None
ws.close()
continue
handle = msg["result"]["qReturn"]["qHandle"]
ws.send(json.dumps({
"jsonrpc": "2.0", "id": 2, "method": "GetScript",
"handle": handle, "params": {},
}))
while True:
msg = json.loads(ws.recv())
if msg.get("id") == 2:
break
if "result" in msg:
script = msg["result"].get("qScript") or ""
out[app_id] = "section access" in script.lower()
else:
out[app_id] = None
ws.close()
except Exception as e:
print(f" engine: {app_id} ERROR {e}")
out[app_id] = None
if i % 25 == 0:
print(f" scanned {i}/{len(app_ids)} apps")
return out
# =========================
# 3. Oracle: facility section access
# =========================
def fetch_facility_access():
"""
Returns:
facilities_by_user: { 'userdir\\userid' (lowercased) :
{ 'access': 'ADMIN'/'USER', 'facilities': set([...]) } }
all_facilities: sorted list of distinct facility numbers (without '*')
"""
print(f"Oracle: reading {ORACLE_TABLE} ...")
with oracledb.connect(user=ORACLE_USER, password=ORACLE_PASSWORD,
dsn=ORACLE_DSN) as conn:
cur = conn.cursor()
cur.execute(f'SELECT "ACCESS", "USERID", "FACILITY_NO" FROM {ORACLE_TABLE}')
rows = cur.fetchall()
facilities_by_user = defaultdict(lambda: {"access": None, "facilities": set()})
all_facilities = set()
for access, userid, fac in rows:
if userid is None:
continue
key = userid.strip().lower()
fac_s = (str(fac).strip() if fac is not None else "")
facilities_by_user[key]["access"] = (access or "").strip().upper() or None
facilities_by_user[key]["facilities"].add(fac_s)
if fac_s and fac_s != "*":
all_facilities.add(fac_s)
def sort_key(x):
try:
return (0, int(x))
except ValueError:
return (1, x)
sorted_facilities = sorted(all_facilities, key=sort_key)
print(f" {len(facilities_by_user)} users, {len(sorted_facilities)} facilities")
return facilities_by_user, sorted_facilities
def user_facility_flags(user, facilities_by_user, all_facilities):
udir = (user.get("userDirectory") or "").strip()
uid = (user.get("userId") or "").strip()
key = f"{udir}\\{uid}".lower()
entry = facilities_by_user.get(key)
flags = {"FACILITY_ALL": False}
for f in all_facilities:
flags[f"FACILITY_{f}"] = False
if not entry:
return flags
fac_set = entry["facilities"]
if "*" in fac_set:
flags["FACILITY_ALL"] = True
for f in all_facilities:
flags[f"FACILITY_{f}"] = True
else:
for f in fac_set:
col = f"FACILITY_{f}"
if col in flags:
flags[col] = True
return flags
# =========================
# 4. Admin / license flags
# =========================
def user_role_flags(user, license_by_user):
license_label = license_by_user.get(user.get("id"), "None")
roles = set(user.get("roles") or [])
is_admin = bool(roles & ADMIN_ROLES)
app_user = license_label != "Professional"
flag = is_admin and app_user
return license_label, is_admin, flag
# =========================
# MAIN
# =========================
def main():
users, streams, apps, license_by_user = fetch_qrs_data()
streams_by_cp = defaultdict(list)
for s in streams:
for cp in cp_set(s):
streams_by_cp[cp].append(s)
apps_by_stream = defaultdict(list)
for a in apps:
st = a.get("stream")
if st and st.get("id"):
apps_by_stream[st["id"]].append(a)
published_app_ids = [a["id"] for a in apps
if a.get("published") and a.get("stream")]
print(f"Engine: scanning {len(published_app_ids)} published apps for Section Access ...")
sa_by_app = detect_section_access(published_app_ids)
facilities_by_user, all_facilities = fetch_facility_access()
base_cols = [
"UserDirectory", "UserId", "UserName", "UserInactive",
"LicenseType", "IsAdmin", "AppUserButAdmin",
"StreamName", "AppName", "AppPublished", "AppUsesSectionAccess",
]
facility_cols = ["FACILITY_ALL"] + [f"FACILITY_{f}" for f in all_facilities]
fieldnames = base_cols + facility_cols
rows = []
for u in users:
user_cps = cp_set(u)
license_label, is_admin, app_user_but_admin = user_role_flags(u, license_by_user)
fac_flags = user_facility_flags(u, facilities_by_user, all_facilities)
seen_streams = {}
for cp in user_cps:
for s in streams_by_cp.get(cp, []):
seen_streams[s["id"]] = s
if not seen_streams:
row = {
"UserDirectory": u.get("userDirectory"),
"UserId": u.get("userId"),
"UserName": u.get("name"),
"UserInactive": u.get("inactive", False),
"LicenseType": license_label,
"IsAdmin": is_admin,
"AppUserButAdmin": app_user_but_admin,
"StreamName": "",
"AppName": "",
"AppPublished": "",
"AppUsesSectionAccess": "",
}
row.update(fac_flags)
rows.append(row)
continue
for s_id, stream in seen_streams.items():
stream_apps = apps_by_stream.get(s_id, [])
if not stream_apps:
row = {
"UserDirectory": u.get("userDirectory"),
"UserId": u.get("userId"),
"UserName": u.get("name"),
"UserInactive": u.get("inactive", False),
"LicenseType": license_label,
"IsAdmin": is_admin,
"AppUserButAdmin": app_user_but_admin,
"StreamName": stream.get("name"),
"AppName": "",
"AppPublished": "",
"AppUsesSectionAccess": "",
}
row.update(fac_flags)
rows.append(row)
continue
for app in stream_apps:
sa = sa_by_app.get(app["id"])
row = {
"UserDirectory": u.get("userDirectory"),
"UserId": u.get("userId"),
"UserName": u.get("name"),
"UserInactive": u.get("inactive", False),
"LicenseType": license_label,
"IsAdmin": is_admin,
"AppUserButAdmin": app_user_but_admin,
"StreamName": stream.get("name"),
"AppName": app.get("name"),
"AppPublished": app.get("published"),
"AppUsesSectionAccess": sa if sa is not None else "Unknown",
}
row.update(fac_flags)
rows.append(row)
with open(OUT_CSV, "w", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=fieldnames)
w.writeheader()
w.writerows(rows)
print(f"Wrote {len(rows)} rows -> {OUT_CSV}")
if __name__ == "__main__":
main()
Hope this helps provide some guidance/direction to anyone needing to get a detailed user audit.
If there are any other details needed I can try to help.