hr-portal/backend/check_architecture.py

"""
檢查多租戶架構完整性
"""
import psycopg2

conn = psycopg2.connect('postgresql://admin:DC1qaz2wsx@10.1.0.20:5433/hr_portal')
cur = conn.cursor()

print('=' * 80)
print('Multi-Tenant Architecture Analysis')
print('=' * 80)

# 1. 租戶隔離檢查
print('\n[1] Tenant Isolation (tenant_id FK)')
cur.execute("""
    SELECT
        tc.table_name,
        kcu.column_name,
        ccu.table_name AS foreign_table_name
    FROM information_schema.table_constraints AS tc
    JOIN information_schema.key_column_usage AS kcu
        ON tc.constraint_name = kcu.constraint_name
    JOIN information_schema.constraint_column_usage AS ccu
        ON ccu.constraint_name = tc.constraint_name
    WHERE tc.constraint_type = 'FOREIGN KEY'
        AND ccu.table_name = 'tenants'
    ORDER BY tc.table_name
""")

tenant_fk_tables = []
for row in cur.fetchall():
    tenant_fk_tables.append(row[0])
    print(f'  [OK] {row[0]:35s} -> tenants.id')

print(f'\n  Total: {len(tenant_fk_tables)} tables with tenant isolation')

# 2. 通用欄位檢查
print(f'\n[2] Common Fields Check')
required_cols = ['is_active', 'edit_by', 'created_at', 'updated_at']

all_tables = []
cur.execute("""
    SELECT table_name
    FROM information_schema.tables
    WHERE table_schema = 'public'
        AND table_type = 'BASE TABLE'
        AND table_name LIKE 'tenant_%'
    ORDER BY table_name
""")
all_tables = [row[0] for row in cur.fetchall()]

complete_count = 0
for table in all_tables:
    cur.execute(f"""
        SELECT column_name
        FROM information_schema.columns
        WHERE table_name = '{table}'
            AND column_name IN ('is_active', 'edit_by', 'created_at', 'updated_at')
    """)
    existing = [row[0] for row in cur.fetchall()]
    missing = [col for col in required_cols if col not in existing]

    if not missing:
        print(f'  [OK] {table}')
        complete_count += 1
    else:
        print(f'  [!!] {table:35s} missing: {", ".join(missing)}')

print(f'\n  Complete: {complete_count}/{len(all_tables)} tables')

# 3. 複合主鍵檢查
print(f'\n[3] Composite Primary Key')
cur.execute("""
    SELECT
        tc.table_name,
        string_agg(kcu.column_name, ', ' ORDER BY kcu.ordinal_position) as pk_columns
    FROM information_schema.table_constraints tc
    JOIN information_schema.key_column_usage kcu
        ON tc.constraint_name = kcu.constraint_name
    WHERE tc.constraint_type = 'PRIMARY KEY'
        AND tc.table_schema = 'public'
    GROUP BY tc.table_name
    HAVING COUNT(kcu.column_name) > 1
    ORDER BY tc.table_name
""")

composite_pk = cur.fetchall()
if composite_pk:
    for row in composite_pk:
        print(f'  [OK] {row[0]:35s} PRIMARY KEY ({row[1]})')
else:
    print('  (No composite primary key tables)')

# 4. 租戶內序號檢查
print(f'\n[4] Tenant-Scoped Sequence (seq_no)')
cur.execute("""
    SELECT table_name
    FROM information_schema.columns
    WHERE column_name = 'seq_no'
        AND table_schema = 'public'
    ORDER BY table_name
""")
seq_tables = cur.fetchall()
for row in seq_tables:
    print(f'  [OK] {row[0]}')

print(f'\n  Total: {len(seq_tables)} tables with seq_no')

# 5. 觸發器檢查
print(f'\n[5] Automation Triggers')
cur.execute("""
    SELECT
        trigger_name,
        event_object_table
    FROM information_schema.triggers
    WHERE trigger_schema = 'public'
        AND (trigger_name LIKE '%tenant%'
             OR trigger_name LIKE '%seq%'
             OR trigger_name LIKE '%emp_code%')
    ORDER BY event_object_table, trigger_name
""")
triggers = cur.fetchall()
for row in triggers:
    print(f'  [OK] {row[1]:35s} <- {row[0]}')

print(f'\n  Total: {len(triggers)} triggers')

# 6. 個人化服務
print(f'\n[6] Personal Services')
cur.execute('SELECT service_code, service_name FROM personal_services WHERE is_active = true ORDER BY id')
for row in cur.fetchall():
    print(f'  [OK] {row[0]:10s} - {row[1]}')

# 7. 核心業務表統計
print(f'\n[7] Table Statistics')
cur.execute("""
    SELECT
        CASE
            WHEN table_name LIKE 'tenant_%' THEN 'tenant_* (Core Business)'
            WHEN table_name IN ('personal_services', 'system_functions_cache', 'tenants') THEN 'System Support'
            WHEN table_name IN ('business_units', 'employee_identities') THEN 'Deprecated'
            ELSE 'Other'
        END as category,
        COUNT(*) as count
    FROM information_schema.tables
    WHERE table_schema = 'public'
        AND table_type = 'BASE TABLE'
        AND table_name != 'alembic_version'
    GROUP BY category
    ORDER BY category
""")
for row in cur.fetchall():
    print(f'  {row[0]:30s} {row[1]:3d} tables')

# 8. 關鍵設計驗證
print(f'\n[8] Key Design Validation')

# 8.1 員工工號自動生成
cur.execute("""
    SELECT proname, prosrc
    FROM pg_proc
    WHERE proname = 'generate_tenant_emp_code'
""")
if cur.fetchone():
    print('  [OK] Employee Code Auto-Generation Function Exists')
else:
    print('  [!!] Employee Code Auto-Generation Function Missing')

# 8.2 租戶資料
cur.execute('SELECT id, code, name, prefix, is_sysmana FROM tenants WHERE id = 1')
tenant = cur.fetchone()
if tenant:
    print(f'  [OK] Tenant 1: {tenant[1]} ({tenant[2]}), prefix={tenant[3]}, sysmana={tenant[4]}')

# 8.3 複合主鍵使用情況
cur.execute("""
    SELECT COUNT(*) FROM tenant_emp_setting
""")
emp_setting_count = cur.fetchone()[0]
print(f'  [OK] tenant_emp_setting (composite PK): {emp_setting_count} records')

print('\n' + '=' * 80)
print('Architecture Check Complete')
print('=' * 80)

conn.close()