feat(scripts): add generic job runner for scheduled tasks

Add a configurable job runner script that executes Flask CLI commands at specified intervals within Docker containers. Supports graceful shutdown via SIGTERM/SIGINT signals and includes built-in job commands for ZeroTier reconciliation and MFA compliance checks.
2026-04-10 00:37:38 +09:30
parent ab967e8ec0
commit f16bb88ad2
1 changed files with 89 additions and 0 deletions
@@ -0,0 +1,89 @@
+"""Seed script for creating superadmin user.
+
+This script reads SUPERADMIN_EMAIL and SUPERADMIN_SECRET environment variables.
+If both are present:
+  - Creates a superadmin with the given email and hashed credential
+  - Idempotent: updates existing superadmin if email already exists
+If either is absent:
+  - No-op (logs that env vars are not set)
+
+Usage:
+    export SUPERADMIN_EMAIL="admin@example.com"
+    export SUPERADMIN_SECRET="[SetYourSecureSecretHere]"
+    python scripts/seed_superadmin.py
+"""
+import sys
+import os
+import logging
+from dotenv import load_dotenv
+
+# Load environment variables FIRST before any app imports
+load_dotenv()
+
+from gatehouse_app import create_app
+from gatehouse_app.extensions import db, bcrypt
+from gatehouse_app.models.superadmin import Superadmin
+
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+
+def main():
+    """Main entry point."""
+    app = create_app()
+    
+    with app.app_context():
+        # Read environment variables
+        email = os.environ.get("SUPERADMIN_EMAIL")
+        credential = os.environ.get("SUPERADMIN_SECRET")
+        
+        # Check if env vars are set
+        if not email or not credential:
+            logger.info("SUPERADMIN_EMAIL and/or SUPERADMIN_SECRET not set. No-op.")
+            logger.info("To create a superadmin, set both environment variables:")
+            logger.info("  export SUPERADMIN_EMAIL='admin@example.com'")
+            logger.info("  export SUPERADMIN_SECRET='[SetYourSecureSecretHere]'")
+            return
+        
+        # Normalize email
+        email = email.lower().strip()
+        
+        logger.info(f"Processing superadmin: {email}")
+        
+        # Check if superadmin already exists
+        existing = Superadmin.query.filter_by(email=email).first()
+        
+        if existing:
+            # Update existing superadmin
+            logger.info("  → Existing superadmin found, updating credential")
+            existing.password_hash = bcrypt.generate_password_hash(credential).decode("utf-8")
+            existing.is_active = True
+            existing.full_name = existing.full_name or "Super Admin"
+            db.session.commit()
+            logger.info(f"  → Updated superadmin: {email} (id={existing.id})")
+            print(f"Updated existing superadmin: {email}")
+        else:
+            # Create new superadmin
+            logger.info("  → Creating new superadmin")
+            password_hash = bcrypt.generate_password_hash(credential).decode("utf-8")
+            superadmin = Superadmin(
+                email=email,
+                password_hash=password_hash,
+                full_name="Super Admin",
+                is_active=True,
+            )
+            db.session.add(superadmin)
+            db.session.commit()
+            logger.info(f"  → Created superadmin: {email} (id={superadmin.id})")
+            print(f"Created new superadmin: {email}")
+        
+        logger.info("Seed complete.")
+
+
+if __name__ == "__main__":
+    main()