f0c30d8005
방문자 사전신청·승인, 입·출입 체크인/아웃, QR 배지, 재실현황, 블랙리스트, 대시보드 통계, 방문 리포트(엑셀)까지 7단계 전 기능 구현. - backend: Spring Boot 3.4.5 / Java 21 (JDK 26 빌드), 세션 인증, JPA, H2/PostgreSQL, POI, ZXing, Flyway - frontend: React 19 / Vite 6 / TypeScript - infra: Docker Compose (db·app·web nginx), Flyway V1__init, Python 사용자 시드 - docs: 워크플로우 / 시퀀스 다이어그램(Mermaid) / 이슈·유의사항 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
seed-load.py — Load users into the ACS (Access Control System) database.
|
|
|
|
CSV/XLSX columns:
|
|
username, full_name, email, department, password, must_change_password, roles
|
|
- roles: semicolon-separated, e.g. "ADMIN;SECURITY" (values: ADMIN, SECURITY, HOST)
|
|
|
|
Usage:
|
|
python scripts/seed-load.py --file scripts/seeds/users.csv
|
|
|
|
Connection via env (PGHOST/PGPORT/PGUSER/PGPASSWORD/PGDATABASE) or --dsn.
|
|
Idempotent: existing usernames are updated (password re-hashed), roles replaced.
|
|
"""
|
|
import argparse
|
|
import os
|
|
|
|
import pandas as pd
|
|
import psycopg2
|
|
from passlib.hash import bcrypt
|
|
|
|
VALID_ROLES = {"ADMIN", "SECURITY", "HOST"}
|
|
|
|
|
|
def read_sheet(path):
|
|
if path.lower().endswith(".csv"):
|
|
return pd.read_csv(path, dtype=str).fillna("")
|
|
return pd.read_excel(path, dtype=str).fillna("")
|
|
|
|
|
|
def get_conn(dsn=None):
|
|
if dsn:
|
|
return psycopg2.connect(dsn)
|
|
return psycopg2.connect(
|
|
host=os.getenv("PGHOST", "localhost"),
|
|
port=os.getenv("PGPORT", "5432"),
|
|
user=os.getenv("PGUSER", "acs"),
|
|
password=os.getenv("PGPASSWORD", ""),
|
|
dbname=os.getenv("PGDATABASE", "acs"),
|
|
)
|
|
|
|
|
|
def parse_bool(v):
|
|
return str(v).strip().lower() in ("1", "true", "yes", "y")
|
|
|
|
|
|
def upsert_user(cur, row):
|
|
username = row["username"].strip()
|
|
if not username:
|
|
return False
|
|
password = row.get("password", "").strip() or "ChangeMe123!"
|
|
pw_hash = bcrypt.hash(password)
|
|
roles = [r.strip().upper() for r in str(row.get("roles", "")).split(";") if r.strip()]
|
|
roles = [r for r in roles if r in VALID_ROLES] or ["HOST"]
|
|
|
|
cur.execute("SELECT id FROM users WHERE username = %s", (username,))
|
|
existing = cur.fetchone()
|
|
|
|
if existing:
|
|
user_id = existing[0]
|
|
cur.execute(
|
|
"""UPDATE users SET updated_at = now(), password_hash = %s, full_name = %s,
|
|
email = %s, department = %s, must_change_password = %s WHERE id = %s""",
|
|
(pw_hash, row.get("full_name", "").strip() or username,
|
|
row.get("email", "").strip() or None, row.get("department", "").strip() or None,
|
|
parse_bool(row.get("must_change_password", "true")), user_id),
|
|
)
|
|
else:
|
|
cur.execute(
|
|
"""INSERT INTO users (created_at, updated_at, username, password_hash, full_name,
|
|
email, department, must_change_password, enabled, locked)
|
|
VALUES (now(), now(), %s, %s, %s, %s, %s, %s, TRUE, FALSE) RETURNING id""",
|
|
(username, pw_hash, row.get("full_name", "").strip() or username,
|
|
row.get("email", "").strip() or None, row.get("department", "").strip() or None,
|
|
parse_bool(row.get("must_change_password", "true"))),
|
|
)
|
|
user_id = cur.fetchone()[0]
|
|
|
|
cur.execute("DELETE FROM user_roles WHERE user_id = %s", (user_id,))
|
|
for role in set(roles):
|
|
cur.execute("INSERT INTO user_roles (user_id, role) VALUES (%s, %s)", (user_id, role))
|
|
return True
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--file", required=True)
|
|
parser.add_argument("--dsn", default=None)
|
|
args = parser.parse_args()
|
|
|
|
df = read_sheet(args.file)
|
|
conn = get_conn(args.dsn)
|
|
count = 0
|
|
try:
|
|
with conn, conn.cursor() as cur:
|
|
for _, row in df.iterrows():
|
|
if upsert_user(cur, row):
|
|
count += 1
|
|
print(f"[seed-load] upserted {count} user(s).")
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|