| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960 |
- """Create tables and seed MySQL when the user table is empty."""
- import os
- import shutil
- import subprocess
- import sys
- from pathlib import Path
- from app import create_app
- from app.models import User, db
- def main() -> None:
- app = create_app()
- with app.app_context():
- db.create_all()
- if User.query.count() > 0:
- print("Database already initialized, skipping seed.")
- return
- sql_path = os.path.join(os.getcwd(), "sql", "seed_snapshot.sql")
- if not os.path.isfile(sql_path):
- sql_path = os.path.join(os.getcwd(), "sql", "init_db.sql")
- if not os.path.isfile(sql_path):
- print("Seed file not found: sql/seed_snapshot.sql or sql/init_db.sql", file=sys.stderr)
- sys.exit(1)
- host = os.environ["MYSQL_HOST"]
- port = os.environ.get("MYSQL_PORT", "3306")
- user = os.environ["MYSQL_USER"]
- password = os.environ["MYSQL_PASSWORD"]
- database = os.environ["MYSQL_DATABASE"]
- cmd = [
- "mysql",
- "--default-character-set=utf8mb4",
- f"-h{host}",
- f"-P{port}",
- f"-u{user}",
- f"-p{password}",
- database,
- ]
- print(f"Seeding database from {os.path.basename(sql_path)} ...")
- with open(sql_path, "rb") as f:
- subprocess.run(cmd, stdin=f, check=True)
- print("Seed completed.")
- snapshot_static = Path("seed_assets/snapshot/static")
- if snapshot_static.is_dir():
- static_root = Path("app/static")
- static_root.mkdir(parents=True, exist_ok=True)
- for child in snapshot_static.iterdir():
- dst = static_root / child.name
- if dst.exists():
- shutil.rmtree(dst)
- shutil.copytree(child, dst)
- print(f"Restored static assets from {snapshot_static}")
- if __name__ == "__main__":
- main()
|