chore: reinitialize project with vite architecture
This commit is contained in:
194
tools/update_oracle_authorized_objects.py
Normal file
194
tools/update_oracle_authorized_objects.py
Normal file
@@ -0,0 +1,194 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Generate a list of accessible TABLE/VIEW objects under a specific owner (default: DWH)
|
||||
and update docs/Oracle_Authorized_Objects.md.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
from collections import Counter
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import oracledb
|
||||
|
||||
|
||||
def load_env() -> None:
|
||||
"""Load .env if available (best-effort)."""
|
||||
try:
|
||||
from dotenv import load_dotenv # type: ignore
|
||||
|
||||
env_path = Path(__file__).resolve().parent.parent / ".env"
|
||||
load_dotenv(env_path)
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
env_path = Path(__file__).resolve().parent.parent / ".env"
|
||||
if not env_path.exists():
|
||||
return
|
||||
for line in env_path.read_text().splitlines():
|
||||
if not line or line.strip().startswith("#") or "=" not in line:
|
||||
continue
|
||||
key, value = line.split("=", 1)
|
||||
os.environ.setdefault(key.strip(), value.strip())
|
||||
|
||||
|
||||
def get_connection():
|
||||
host = os.getenv("DB_HOST", "")
|
||||
port = os.getenv("DB_PORT", "1521")
|
||||
service = os.getenv("DB_SERVICE", "")
|
||||
user = os.getenv("DB_USER", "")
|
||||
password = os.getenv("DB_PASSWORD", "")
|
||||
dsn = (
|
||||
"(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)"
|
||||
f"(HOST={host})(PORT={port})))(CONNECT_DATA=(SERVICE_NAME={service})))"
|
||||
)
|
||||
return oracledb.connect(user=user, password=password, dsn=dsn)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
owner = "DWH"
|
||||
output_path = Path("docs/Oracle_Authorized_Objects.md")
|
||||
if len(sys.argv) > 1:
|
||||
owner = sys.argv[1].strip().upper()
|
||||
|
||||
load_env()
|
||||
conn = get_connection()
|
||||
cur = conn.cursor()
|
||||
|
||||
cur.execute("SELECT USER FROM DUAL")
|
||||
user = cur.fetchone()[0]
|
||||
|
||||
# Roles
|
||||
cur.execute("SELECT GRANTED_ROLE FROM USER_ROLE_PRIVS")
|
||||
roles = [r[0] for r in cur.fetchall()]
|
||||
|
||||
# Accessible objects under owner
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT OBJECT_NAME, OBJECT_TYPE
|
||||
FROM ALL_OBJECTS
|
||||
WHERE OWNER = :p_owner
|
||||
AND OBJECT_TYPE IN ('TABLE', 'VIEW')
|
||||
ORDER BY OBJECT_NAME
|
||||
""",
|
||||
p_owner=owner,
|
||||
)
|
||||
objects = cur.fetchall()
|
||||
|
||||
# Direct + PUBLIC grants
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT o.object_name, o.object_type, p.privilege,
|
||||
CASE WHEN p.grantee = 'PUBLIC' THEN 'PUBLIC' ELSE 'DIRECT' END AS source
|
||||
FROM all_tab_privs p
|
||||
JOIN all_objects o
|
||||
ON o.owner = p.table_schema
|
||||
AND o.object_name = p.table_name
|
||||
WHERE p.grantee IN (:p_user, 'PUBLIC')
|
||||
AND o.owner = :p_owner
|
||||
AND o.object_type IN ('TABLE', 'VIEW')
|
||||
""",
|
||||
p_user=user,
|
||||
p_owner=owner,
|
||||
)
|
||||
direct_rows = cur.fetchall()
|
||||
|
||||
# Role grants
|
||||
role_rows = []
|
||||
for role in roles:
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT o.object_name, o.object_type, p.privilege, p.role AS source
|
||||
FROM role_tab_privs p
|
||||
JOIN all_objects o
|
||||
ON o.owner = p.owner
|
||||
AND o.object_name = p.table_name
|
||||
WHERE p.role = :p_role
|
||||
AND o.owner = :p_owner
|
||||
AND o.object_type IN ('TABLE', 'VIEW')
|
||||
""",
|
||||
p_role=role,
|
||||
p_owner=owner,
|
||||
)
|
||||
role_rows.extend(cur.fetchall())
|
||||
|
||||
# Aggregate privileges by object
|
||||
info = {}
|
||||
for name, otype in objects:
|
||||
info[(name, otype)] = {"privs": set(), "sources": set()}
|
||||
|
||||
for name, otype, priv, source in direct_rows + role_rows:
|
||||
key = (name, otype)
|
||||
if key not in info:
|
||||
info[key] = {"privs": set(), "sources": set()}
|
||||
info[key]["privs"].add(priv)
|
||||
info[key]["sources"].add(source)
|
||||
|
||||
# Fill in missing privilege/source if object is visible but not in grants
|
||||
for key, data in info.items():
|
||||
if not data["privs"]:
|
||||
data["privs"].add("UNKNOWN")
|
||||
if not data["sources"]:
|
||||
data["sources"].add("SYSTEM")
|
||||
|
||||
type_counts = Counter(k[1] for k in info.keys())
|
||||
source_counts = Counter()
|
||||
for data in info.values():
|
||||
for s in data["sources"]:
|
||||
if s in ("DIRECT", "PUBLIC"):
|
||||
source_counts[s] += 1
|
||||
elif s == "SYSTEM":
|
||||
source_counts["SYSTEM"] += 1
|
||||
else:
|
||||
source_counts["ROLE"] += 1
|
||||
|
||||
# Render markdown
|
||||
lines = []
|
||||
lines.append("# Oracle 可使用 TABLE/VIEW 清單(DWH)")
|
||||
lines.append("")
|
||||
lines.append(f"**產生時間**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
lines.append(f"**使用者**: {user}")
|
||||
lines.append(f"**Schema**: {owner}")
|
||||
lines.append("")
|
||||
lines.append("## 摘要")
|
||||
lines.append("")
|
||||
lines.append(f"- 可使用物件總數: {len(info):,}")
|
||||
lines.append(f"- TABLE: {type_counts.get('TABLE', 0):,}")
|
||||
lines.append(f"- VIEW: {type_counts.get('VIEW', 0):,}")
|
||||
lines.append(
|
||||
"- 來源 (去重後物件數): "
|
||||
f"DIRECT {source_counts.get('DIRECT', 0):,}, "
|
||||
f"PUBLIC {source_counts.get('PUBLIC', 0):,}, "
|
||||
f"ROLE {source_counts.get('ROLE', 0):,}, "
|
||||
f"SYSTEM {source_counts.get('SYSTEM', 0):,}"
|
||||
)
|
||||
lines.append("")
|
||||
lines.append("## 物件清單")
|
||||
lines.append("")
|
||||
lines.append("| 物件 | 類型 | 權限 | 授權來源 |")
|
||||
lines.append("|------|------|------|----------|")
|
||||
|
||||
for name, otype in sorted(info.keys()):
|
||||
data = info[(name, otype)]
|
||||
obj = f"{owner}.{name}"
|
||||
privs = ", ".join(sorted(data["privs"]))
|
||||
sources = ", ".join(
|
||||
sorted(
|
||||
"ROLE" if s not in ("DIRECT", "PUBLIC", "SYSTEM") else s
|
||||
for s in data["sources"]
|
||||
)
|
||||
)
|
||||
lines.append(f"| `{obj}` | {otype} | {privs} | {sources} |")
|
||||
|
||||
output_path.write_text("\n".join(lines), encoding="utf-8")
|
||||
|
||||
cur.close()
|
||||
conn.close()
|
||||
print(f"Wrote {output_path} ({len(info)} objects)")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user