import os import re import sys import shutil import fileinput import subprocess from cement import App, Controller, ex from cement import shell from dhwp.core.git import Git from dhwp.core.wordpress import Wordpress class Stage(Controller): class Meta: label = 'stage' description = 'staging commands' stacked_on = 'base' stacked_type = 'nested' arguments = [ (['-p', '--path'], {'help': 'path to live wordpress install', 'action': 'store', 'dest': 'path', }), (['-d', '--destination'], {'help': 'path to staging', 'action': 'store', 'dest': 'destination', }), ] @ex( help='create a new staging environment', arguments=[ (['--url'], {'help': 'staging site url', 'action': 'store', 'dest': 'url'}), (['--dbname'], {'help': 'staging database name', 'action': 'store', 'dest': 'dbname'}), ] ) def create(self): if not self.app.pargs.destination: self.app.log.fatal("destination is required") sys.exit(1) destination = self.app.pargs.destination live_wp = Wordpress(self.app.pargs.path, app=self.app) live_url = live_wp.site_url() live_git = Git(self.app.pargs.path, app=self.app) self.delete() if not os.path.isdir(destination): os.makedirs(destination, 0o755) if os.listdir(destination): self.app.log.fatal("destination is not empty") sys.exit(1) self.app.log.info("Initalize live repository at %s" % self.app.pargs.path) live_git.cmd('init') self.app.log.info("Commiting all changes in %s" % self.app.pargs.path) live_git.commit('staging create', exclude_media=True, exclude_live_archives=True) self.app.log.info("Creating new git worktree at %s" % destination) live_git.cmd('worktree', 'add -b staging ' + destination + ' master -f') self._rsync_git_dirs(self.app.pargs.path, destination) self.app.log.info("Copying over wp-config from %s to %s" % (self.app.pargs.path, self.app.pargs.destination)) r = re.compile(r"https?://(www\.)?") domain = r.sub('', live_url).strip().strip('/') with open(os.path.join(self.app.pargs.path, 'wp-config.php'), 'r') as conf, \ open(os.path.join(self.app.pargs.destination, 'wp-config.php'), 'w') as stage_conf: for line in conf: wp_home = all(w in line for w in ('WP_HOME', domain)) wp_siteurl = all(w in line for w in ('WP_SITEURL', domain)) if wp_home or wp_siteurl: self.app.log.info( "Generating file %s: removed line %s" % ( os.path.join(self.app.pargs.destination, 'wp-config.php'), line)) else: stage_conf.write(line) stage_wp = Wordpress(destination, app=self.app) stage_r = re.compile(r"^www\.") stage_domain = stage_r.sub('', self.app.pargs.url) stage_url = 'https://' + stage_domain stage_db_file = destination + '/db.sql' self.app.log.info("Creating Live backup") live_wp.db_export(stage_db_file) self.app.log.info("Importing live DB backup to stage") stage_wp.cli('config set DB_NAME ' + self.app.pargs.dbname) stage_wp.db_import(stage_db_file) if os.path.exists(stage_db_file): os.unlink(stage_db_file) else: self.app.log.info(f"SQL backup file {stage_db_file} was already removed") self.app.log.info("Setting Jetpack to staging mode") stage_wp.cli('config set JETPACK_STAGING_MODE true --add --type=constant --raw') self.app.log.info("Setting WordPress to staging mode") stage_wp.cli('config set WP_ENVIRONMENT_TYPE staging --add --type=constant') self.app.log.info("Setting Redis plugin to use own separate DB") stage_wp.cli('config set WP_REDIS_DATABASE 1 --add --type=constant --raw') self.app.log.info("Setting WordPress to not auto-update") stage_wp.cli('config set AUTOMATIC_UPDATER_DISABLED true --add --type=constant') self.app.log.info("Renaming WP site from %s to %s" % (live_url, stage_url)) stage_wp.rename(live_url, stage_url) self.app.log.info("Regenerating the .htaccess file") stage_wp.cli('rewrite flush --hard') @ex( help='commit changes to environment', arguments=[ (['--all'], {'help': 'commit db and files', 'action': 'store_true', 'dest': 'all_changes', }), (['--files'], {'help': 'commit files only', 'action': 'store_true', 'dest': 'file_changes', }), (['--db'], {'help': 'commit database changes', 'action': 'store_true', 'dest': 'db_changes', }), ] ) def commit(self): stage_wp = Wordpress(self.app.pargs.destination, app=self.app) live_wp = Wordpress(self.app.pargs.path, app=self.app) live_git = Git(self.app.pargs.path, app=self.app) stage_git = Git(self.app.pargs.destination, app=self.app) live_url = live_wp.site_url() stage_url = stage_wp.site_url() if self.app.pargs.file_changes or self.app.pargs.all_changes: if stage_git.has_changes(): self.app.log.info("Commiting all changes to staging") stage_git.commit('staging changes') self.app.log.info("Merging staging changes") # TODO do we prefer changes to live or staging.. hrmm live_git.git_ignore(exclude_media=True, exclude_live_archives=True) shell.cmd("cd %s && for i in $(git ls-files . --exclude-standard --others); do rm -rf \"$i\"; done" % self.app.pargs.path) live_git.cmd('reset', '--hard') live_git.cmd('merge', '-X theirs staging') self._rsync_git_dirs(self.app.pargs.destination, self.app.pargs.path) else: self.app.log.info("No file changes were found to commit") if self.app.pargs.db_changes or self.app.pargs.all_changes: stage_sql_file = stage_wp.db_export() live_wp.db_import(stage_sql_file) if os.path.exists(stage_sql_file): os.unlink(stage_sql_file) else: self.app.log.info(f"SQL backup file {stage_sql_file} was already removed") live_wp.rename(stage_url, live_url) live_wp.cli('core update-db') live_wp.cli('cache flush') @ex( help='delete a staging environment', arguments=[ ] ) def delete(self): destination = self.app.pargs.destination live_git = Git(self.app.pargs.path, app=self.app) stage_git = Git(self.app.pargs.destination, app=self.app) live_git.wipe() stage_git.wipe() if os.path.isdir(destination): self.app.log.info("Clearing destination %s" % destination) for root, dirs, files in os.walk(destination): for f in files: os.unlink(os.path.join(root, f)) for d in dirs: if os.path.islink(os.path.join(root, d)): os.unlink(os.path.join(root, d)) else: shutil.rmtree(os.path.join(root, d)) def _rsync_git_dirs(self, source_path, destination_path): self.app.log.info("Searching for git submodules and rsyncing them to staging") for root, subdirs, files in os.walk(source_path): if '.git' in subdirs: if os.path.join(root, '.git') == os.path.join(source_path, '.git'): continue rel_path = os.path.relpath(root, source_path) source_dir = os.path.join(root, "") dest_dir = os.path.join(destination_path, rel_path, "") os.makedirs(os.path.dirname(dest_dir), exist_ok=True) self.app.log.info(f"Found git directory at {root}") self.app.log.info(f"Running rsync -avp --delete {source_dir} {dest_dir}") try: subprocess.run(["rsync", "-avp", source_dir, dest_dir], check=True) self.app.log.info(f"Successfully rsynced {source_dir} to {dest_dir}") except subprocess.CalledProcessError as e: self.app.log.error(f"Failed to rsync {source_dir} to {dest_dir}: {e}")