diff --git a/docs/dsh/dsh.md b/docs/dsh/dsh.md index e394881..7c5c88e 100644 --- a/docs/dsh/dsh.md +++ b/docs/dsh/dsh.md @@ -33,9 +33,40 @@ This allows for incremental uploads; that is, the addition of extra data for dates that have already had an upload in the past. *However*, the extracted files will clash in name, as the names of the parquets within are anchored to the original event date. -It would be the job of a future DSH extractor script to do the right thing here. -Eg. to have a rule that extracts from later uploads always take precedence. -See [issue #84](https://github.com/SAFEHR-data/waveform-controller/issues/84) . The uploaded file name is stored in JSON on the GAE in the daily uploaded sentinel file: eg. `waveform-export/ftps-logs/2024-09-12/2024-09-12.uploaded.json` + +### Extraction on the DSH side +See +[UCL DSH docs on File Transfer Portal]https://www.ucl.ac.uk/isd/services/file-storage-sharing/data-safe-haven-dsh/data-safe-haven-user-guide-faqs#File%20Transfer%20Portal +for background information. + +Uploaded files go into the Q: drive on the DSH and are auto deleted after 30 days, unless moved to the S: drive. + +There is [a script to make this transfer on the DSH side](../../scripts/dsh/extract.py). + +It must be manually run by a member of the `WAVEFORM` DSH share more often than every 30 days. + +#### To run it + +First deploy the script onto the DSH. This is manual but very simple. We do everything under `S:\WAVEFORM` +so we can all see what extraction has taken place. + +1. Inside this repository, check out the version you wish to deploy. +2. Log into the DSH File Transfer Portal as your own user +3. Upload `scripts/dsh/extract.py` to the portal +4. On the DSH Desktop, in a Git Bash window, run `mv /q/${YOUR_USER_ID}/extract.py /s/WAVEFORM/extract_script/extract.py` + +Now you can run the script in a Git Bash Window: +``` +cd /s/WAVEFORM/extract_script +python extract.py --help +# eg. +python extract.py --instance-name gae-testing --dry-run +``` +When we go into production, instance name can be omitted. + +Later TAR uploads can overwrite earlier files when they are extracted. + +All output is logged to a directory adjacent to the script. diff --git a/scripts/dsh/extract.py b/scripts/dsh/extract.py new file mode 100644 index 0000000..5f8e5fa --- /dev/null +++ b/scripts/dsh/extract.py @@ -0,0 +1,89 @@ +# Script to run on DSH (Windows) to move data from incoming to shared drive. +# To avoid having to deal with Python envs on the DSH, target Python 3.8 +# and don't introduce non-core dependencies. +import argparse +import logging +import tarfile +from datetime import datetime, timezone +from pathlib import Path + +# base paths +WAVEFORM_INCOMING = Path("Q:/emap-waveform-uploader-prod/waveform-export") +# It is very important that we don't write to any share except this one +WAVEFORM_SHARED = Path("S:/WAVEFORM") +PSEUDONYMISED = "pseudonymised" + + +def setup_logger(): + """Log to both screen and a log file.""" + logger = logging.getLogger(__name__) + logger.setLevel(logging.DEBUG) + datetime_str = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H%M%SZ") + log_dir = Path(__file__).parent / "extract_logs" + log_dir.mkdir(exist_ok=True) + file_handler = logging.FileHandler(log_dir / f"extract{datetime_str}.log") + file_handler.setLevel(logging.DEBUG) + console_handler = logging.StreamHandler() + console_handler.setLevel(logging.DEBUG) + logger.addHandler(file_handler) + logger.addHandler(console_handler) + return logger + + +logger = setup_logger() + + +def main(args): + incoming_full_path = WAVEFORM_INCOMING / args.instance_name / PSEUDONYMISED + extraction_output_full_path = WAVEFORM_SHARED / args.instance_name / PSEUDONYMISED + extraction_output_full_path.parent.mkdir(parents=True, exist_ok=True) + # Tar file names contain ISO datetime strings so will sort in the order they were uploaded. + # By design, extracted files that appear in later uploads can overwrite those from earlier uploads if the paths match. + ordered_tar_files = sorted(incoming_full_path.glob("*.tar")) + logger.info(f"Found {len(ordered_tar_files)} tar files in {incoming_full_path}") + dry_run_marker = "[DRY RUN] " if args.dry_run else "" + for tar_to_extract in ordered_tar_files: + logger.info( + f"{dry_run_marker}Extracting tar file ({tar_to_extract.stat().st_size:,} bytes) {tar_to_extract}" + ) + if not args.interactive or "y" == input("Extract? "): + with tarfile.TarFile(tar_to_extract, mode="r") as tar_obj: + all_members = sorted( + tar_obj.getmembers(), key=lambda member: member.name + ) + for tar_member in all_members: + logger.info( + " file [%s bytes]: %s", + "{: 12,}".format(tar_member.size), + tar_member.name, + ) + if not args.dry_run: + tar_obj.extractall(path=extraction_output_full_path) + if not args.no_delete: + if not args.dry_run: + tar_to_extract.unlink() + logger.info(f"{dry_run_marker}Deleted {tar_to_extract}") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument( + "--dry-run", action="store_true", help="Only print what would be done." + ) + parser.add_argument( + "--instance-name", + default="production", + help="From which named uploader instance to take data from.", + ) + parser.add_argument( + "--interactive", + action="store_true", + help="Interactively prompt before extracting/deleting each tar file.", + ) + parser.add_argument( + "--no-delete", + action="store_true", + help="Do not delete the source TAR file after a successful extraction.", + ) + args = parser.parse_args() + main(args)