import os
import shlex
import subprocess
import time
import zipfile
from typing import List, Tuple, Optional, Dict, Any, Iterable
import redis
import re
import tempfile
import boto3
from .config.config import ccat_data_transfer_settings
from .exceptions import ServiceExit, ArchiveCorruptionError
from .logging_utils import get_structured_logger
from ccat_ops_db import models
# from .bbcp_settings import BBCPSettings
logger = get_structured_logger(__name__)
# SSH options applied to all outgoing SSH commands for defense-in-depth.
# The ccat_transfer user's ~/.ssh/config (managed by Ansible) also enforces these,
# but explicit options here guard against config drift.
SSH_OPTS = ["-o", "StrictHostKeyChecking=yes"]
[docs]
def deduplicate_by_id(objects: Iterable[Any]) -> List[Any]:
"""Return ORM objects with duplicate primary-key ``id`` removed, order preserved.
Drop-in replacement for ``Query.distinct()`` on entity queries that join a
one-to-many table (e.g. RawDataPackage joined to its physical copies). Such a
join can emit the same parent row more than once, which ``.distinct()`` used
to collapse in the database.
That broke once the failure-tracking ``error_context`` columns (ops-db #85)
added a PostgreSQL ``json`` column to the operation models: the ``json`` type
has no equality operator, so ``SELECT DISTINCT`` over the full row raises
``UndefinedFunction: could not identify an equality operator for type json``.
Deduplicating by ``id`` in Python avoids comparing whole rows and behaves
identically on Postgres and the SQLite test database.
"""
seen = set()
unique: List[Any] = []
for obj in objects:
if obj.id not in seen:
seen.add(obj.id)
unique.append(obj)
return unique
[docs]
def safe_join(base: str, *paths: str) -> str:
"""Join base path with untrusted relative path(s), preventing directory traversal.
Raises ValueError if the resolved path escapes the base directory.
"""
base = os.path.realpath(base)
joined = os.path.realpath(os.path.join(base, *paths))
if not joined.startswith(base + os.sep) and joined != base:
raise ValueError(
f"Path traversal detected: resolved path {joined!r} is outside base {base!r}"
)
return joined
# Singleton Redis client
_redis_client = None
# Singleton S3 client
_s3_client = None
[docs]
def get_redis_connection() -> redis.StrictRedis:
"""
Establish a connection to the Redis server.
This function implements a singleton pattern to reuse the same Redis connection.
Returns
-------
redis.StrictRedis
"""
global _redis_client
if _redis_client is None:
# Create the Redis client only once
_redis_client = redis.Redis(
host=ccat_data_transfer_settings.REDIS_HOST,
port=ccat_data_transfer_settings.REDIS_PORT,
db=0,
decode_responses=True,
ssl=True,
ssl_cert_reqs="required",
ssl_ca_certs=ccat_data_transfer_settings.REDIS_CA_CERT,
password=ccat_data_transfer_settings.REDIS_PASSWORD,
ssl_certfile=ccat_data_transfer_settings.REDIS_CERTFILE,
ssl_keyfile=ccat_data_transfer_settings.REDIS_KEYFILE,
# Add connection pooling settings
max_connections=10, # Limit the number of connections
socket_timeout=5, # Set a reasonable timeout
socket_connect_timeout=5,
retry_on_timeout=True,
)
logger.info("redis connection created")
return _redis_client
[docs]
def get_s3_client(
location: Optional[models.S3DataLocation] = None, site_name: Optional[str] = None
) -> boto3.client:
"""
Establish a connection to the S3 server.
This function implements a singleton pattern to reuse the same S3 connection.
Parameters
----------
location : Optional[models.S3DataLocation]
S3DataLocation object to get specific configuration for.
If None, uses default configuration.
site_name : Optional[str]
Name of the site for credential loading.
Required if location is provided.
Returns
-------
boto3.client
"""
global _s3_client
# Get configuration for the specific location or use default
if location and site_name:
# Get location-specific credentials
access_key_id, secret_access_key = location.get_s3_credentials(site_name)
# Resolve endpoint URL: prefer the value stored on the location (admin
# self-serve), then fall back to dynaconf S3_ENDPOINTS keyed by name,
# then the global default. See ADR 0001 (ops-db).
endpoints = ccat_data_transfer_settings.S3_ENDPOINTS
endpoint_url = getattr(location, "endpoint_url", None) or endpoints.get(
location.name, ccat_data_transfer_settings.S3_ENDPOINT_URL
)
# Create a new client for this specific location
client = boto3.client(
"s3",
endpoint_url=endpoint_url,
aws_access_key_id=access_key_id,
aws_secret_access_key=secret_access_key,
region_name=location.region,
config=_s3_boto_config(location),
)
# Log the exact connection details so credential/endpoint/addressing
# mismatches are diagnosable from the worker logs. Endpoint, bucket and a
# masked access-key fingerprint at INFO; the secret only ever as a
# first/last-char fingerprint at DEBUG. The env var name the credential
# lookup expects is logged so a name mismatch (silent global fallback) is
# immediately visible.
expected_var = (
f"CCAT_DATA_TRANSFER_{site_name}_{location.name}_S3_ACCESS_KEY_ID"
)
logger.info(
"s3 client configured",
location=location.name,
site=site_name,
endpoint=endpoint_url,
region=location.region,
bucket=getattr(location, "bucket_name", None),
access_method=getattr(location, "access_method", None),
addressing_style=_addressing_style(location) or "auto",
access_key=_fingerprint(access_key_id, keep=4),
)
logger.debug(
"s3 credential fingerprint",
location=location.name,
expected_env_var=expected_var,
access_key=_fingerprint(access_key_id, keep=4),
secret_key=_fingerprint(secret_access_key, keep=1),
)
if not access_key_id or access_key_id == "MUST_SET_VIA_ENV":
# Per-location lookup missed and fell back to the global placeholder;
# the upload will fail with InvalidAccessKeyId. Surface the exact var.
logger.warning(
"S3 access key unresolved — using global fallback/placeholder; "
"set the per-location credential env var",
location=location.name,
site=site_name,
expected_env_var=expected_var,
)
return client
# Use default configuration (singleton pattern)
if _s3_client is None:
# Create the S3 client only once
_s3_client = boto3.client(
"s3",
endpoint_url=ccat_data_transfer_settings.s3_endpoint_url,
aws_access_key_id=ccat_data_transfer_settings.s3_access_key_id,
aws_secret_access_key=ccat_data_transfer_settings.s3_secret_access_key,
region_name=ccat_data_transfer_settings.s3_region_name,
config=_s3_boto_config(None),
)
logger.info("s3 default connection created")
return _s3_client
def _fingerprint(secret: Optional[str], keep: int = 1) -> str:
"""Masked fingerprint of a credential for debug logs.
Shows the first/last ``keep`` chars and the length, never the middle, so a
log reader can tell *which* key is in play without exposing the secret. The
unset placeholder is flagged explicitly so a silent fallback to the global
default is obvious in the worker logs.
"""
if not secret:
return "<empty>"
if secret == "MUST_SET_VIA_ENV":
return "<unset: MUST_SET_VIA_ENV>"
if len(secret) <= keep * 2:
return f"…(len={len(secret)})"
return f"{secret[:keep]}…{secret[-keep:]} (len={len(secret)})"
def _resolve_path_style(location=None):
"""Effective path-style for an S3 location: per-location first, then global.
Per-location ``path_style`` wins; otherwise the deployment-wide
``S3_PATH_STYLE`` is used. Returns True (path), False (virtual), or None.
"""
path_style = getattr(location, "path_style", None)
if path_style is None:
path_style = ccat_data_transfer_settings.get("S3_PATH_STYLE", None)
return path_style
def _addressing_style(location=None) -> Optional[str]:
"""Human-readable boto3 addressing style: ``"path"``, ``"virtual"`` or None."""
ps = _resolve_path_style(location)
return None if ps is None else ("path" if ps else "virtual")
def _s3_boto_config(location=None):
"""boto3 ``Config`` for an S3 client: s3v4 signature + addressing style.
The addressing (path vs virtual-host) is endpoint-specific — RDS needs
path-style, the NRW Datastorage needs virtual-host — so it is resolved from
the location's ``path_style`` first, then a deployment-wide ``S3_PATH_STYLE``:
True -> "path" (https://endpoint/bucket/key) e.g. RDS, MinIO
False -> "virtual" (https://bucket.endpoint/key) e.g. NRW Datastorage
None -> boto3 default ("auto")
"""
s3_opts = {}
style = _addressing_style(location)
if style is not None:
s3_opts["addressing_style"] = style
return boto3.session.Config(signature_version="s3v4", s3=s3_opts)
[docs]
def resolve_s3_method(location=None) -> str:
"""Resolve how to reach an S3 location: ``"boto3"`` or ``"coscine"``.
Prefers the per-location ``access_method`` (``"s3"`` ⇒ boto3, ``"coscine"``),
falling back to the deployment-wide ``S3_METHOD`` when the location does not
specify one. See ADR 0001 (ops-db).
"""
method = getattr(location, "access_method", None)
if not method:
return ccat_data_transfer_settings.s3_method
return "boto3" if method == "s3" else method
[docs]
def resolve_coscine_config(location, site_name):
"""Resolve (api_token, project, resource) for a Coscine S3 location.
Non-secret project/resource come from the location with a global fallback;
the secret token is resolved from the environment per-location via
``S3DataLocation.get_coscine_api_token`` (global ``COSCINE_API_TOKEN`` fallback).
"""
s = ccat_data_transfer_settings
project = getattr(location, "coscine_project", None) or s.coscine_project
resource = getattr(location, "coscine_resource", None) or s.coscine_resource
if location is not None and hasattr(location, "get_coscine_api_token") and site_name:
token = location.get_coscine_api_token(site_name)
else:
token = s.coscine_api_token
return token, project, resource
[docs]
def service_shutdown(signum: int, frame) -> None:
"""
Handle service shutdown signal.
Parameters
----------
signum : int
The signal number.
frame : frame
Current stack frame.
Raises
------
ServiceExit
Raised to initiate the service exit process.
"""
logger.info("caught shutdown signal", signal=signum)
raise ServiceExit
[docs]
def unique_id() -> str:
"""
Generate a unique ID using UUID4.
Returns
-------
str
A 16-character hexadecimal string (64 bits of randomness).
"""
import uuid
return uuid.uuid4().hex[:16]
[docs]
def create_archive(
files: List, archive_name: str, base_path: str
) -> Tuple[str, List[str]]:
"""
Create a tar archive optimized for high-speed transfer using system tar command.
Parameters
----------
files : List
A list of RawDataPackage objects, each with a 'relative_path' attribute.
archive_name : str
The name (including path) of the tar archive to be created.
base_path : str
The base path to prepend to the relative paths.
Returns
-------
Tuple[str, List[str]]
A tuple containing the archive name and a list of file names included in the archive.
"""
# Add timing instrumentation
start_time = time.time()
# Ensure the archive ends with .tar
if not archive_name.endswith(".tar"):
archive_name = archive_name.rsplit(".", 1)[0] + ".tar"
logger.info("creating archive", archive_name=os.path.basename(archive_name))
file_names = []
existing_files = []
# First, collect all valid files
for file in files:
full_path = safe_join(base_path, file.relative_path)
if os.path.exists(full_path):
existing_files.append(
(file.relative_path, full_path)
) # Store relative path first
file_names.append(file.relative_path)
else:
logger.error(
"file missing — excluded from archive",
path=full_path,
)
if not existing_files:
logger.error("no valid files to add to archive")
return archive_name, []
# Create a temporary file listing relative paths for tar
with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file:
for rel_path, _ in existing_files:
temp_file.write(f"{rel_path}\n")
temp_file_path = temp_file.name
try:
# Use system tar command for much better performance
tar_cmd = [
"tar",
"-cf", # Create uncompressed tar file
archive_name, # Output file
"-C",
base_path, # Change to base directory
"--files-from",
temp_file_path, # Read file list from temp file
]
logger.debug("running tar command", cmd=" ".join(tar_cmd))
# Add detailed timing for subprocess
cmd_start = time.time()
result = subprocess.run(tar_cmd, capture_output=True, text=True, check=False)
cmd_end = time.time()
logger.debug("tar command finished", duration_s=round(cmd_end - cmd_start, 2))
if result.returncode != 0:
logger.error("tar archive creation failed", stderr=result.stderr)
return archive_name, []
logger.info("archive created", file_count=len(file_names))
except Exception as e:
logger.error("tar command error", error=str(e))
return archive_name, []
finally:
# Clean up temp file
try:
os.unlink(temp_file_path)
except Exception as e:
logger.error("temp file cleanup failed", error=str(e))
end_time = time.time()
logger.debug("archive creation total time", duration_s=round(end_time - start_time, 2))
return archive_name, file_names
[docs]
def unpack_local(archive_path: str, destination: str) -> Tuple[bool, List[str]]:
"""
Unpack a file locally.
Parameters
----------
archive_path : str
The path to the archive file (tar or zip)
destination : str
The path where the archive should be extracted.
Returns
-------
Tuple[bool, List[str]]
A tuple containing a boolean indicating success (True) or failure (False), and a
list of unpacked files (empty if failed).
Raises
------
ArchiveCorruptionError
If the archive is corrupted or incomplete.
"""
try:
extracted_files = []
if archive_path.endswith(".tar"):
# First verify the archive integrity
verify_cmd = ["tar", "-tf", archive_path]
verify_result = subprocess.run(
verify_cmd, capture_output=True, text=True, check=False
)
if verify_result.returncode != 0:
error_msg = verify_result.stderr.strip()
if "Unexpected EOF" in error_msg:
raise ArchiveCorruptionError(
f"Archive is corrupted or incomplete: {error_msg}",
archive_path=archive_path,
)
logger.error("tar archive verification failed", error=error_msg)
return False, []
# Check for path traversal in tar members
dest_real = os.path.realpath(destination)
for member_name in verify_result.stdout.strip().split("\n"):
if member_name.startswith("/") or ".." in member_name.split("/"):
raise ValueError(
f"Tar member {member_name!r} contains path traversal, "
f"refusing to extract archive {archive_path}"
)
# Use system tar command for better performance and control
tar_cmd = [
"tar",
"-xf", # Extract files
archive_path, # Input file
"-C", # Change to directory
destination, # Destination directory
]
logger.debug("running tar extraction", cmd=" ".join(tar_cmd))
result = subprocess.run(
tar_cmd, capture_output=True, text=True, check=False
)
if result.returncode != 0:
error_msg = result.stderr.strip()
if "Unexpected EOF" in error_msg:
raise ArchiveCorruptionError(
f"Archive is corrupted or incomplete: {error_msg}",
archive_path=archive_path,
)
logger.error("tar extraction failed", error=error_msg)
return False, []
# Get the list of files from the tar archive
list_cmd = ["tar", "-tf", archive_path]
list_result = subprocess.run(
list_cmd, capture_output=True, text=True, check=False
)
if list_result.returncode == 0:
extracted_files = list_result.stdout.strip().split("\n")
elif archive_path.endswith(".zip"):
try:
with zipfile.ZipFile(archive_path, "r") as zip_ref:
# Test the zip file integrity
if zip_ref.testzip() is not None:
raise ArchiveCorruptionError(
"ZIP file is corrupted", archive_path=archive_path
)
# Check for path traversal in zip members
dest_real = os.path.realpath(destination)
for name in zip_ref.namelist():
member_path = os.path.realpath(
os.path.join(destination, name)
)
if not member_path.startswith(dest_real + os.sep) and member_path != dest_real:
raise ValueError(
f"Zip member {name!r} contains path traversal, "
f"refusing to extract archive {archive_path}"
)
zip_ref.extractall(destination)
extracted_files = zip_ref.namelist()
except zipfile.BadZipFile as e:
raise ArchiveCorruptionError(
f"ZIP file is corrupted: {str(e)}", archive_path=archive_path
)
else:
logger.error(
"unsupported archive format",
archive_path=archive_path,
supported_formats=".tar, .zip",
)
return False, []
logger.info(
"archive unpacked locally",
archive_path=archive_path,
destination=destination,
)
return True, extracted_files
except ArchiveCorruptionError:
raise
except Exception as e:
logger.error(
"error unpacking archive locally",
archive_path=archive_path,
error=str(e),
)
return False, []
[docs]
def calculate_checksum(filepath: str) -> Optional[str]:
"""Calculate xxHash64 checksum of a file for fast integrity verification."""
start_time = time.time()
try:
result = subprocess.run(
["xxh64sum", filepath], capture_output=True, text=True, check=True
)
checksum = result.stdout.strip().split()[0]
elapsed = time.time() - start_time
logger.info(
"checksum calculated",
path=filepath,
checksum=checksum,
duration_s=round(elapsed, 2),
)
return checksum
except Exception as e:
logger.error("checksum calculation failed", path=filepath, error=str(e))
return None
[docs]
def make_bbcp_command(source_url: str, destination_url: str) -> List[str]:
"""
Construct the bbcp command.
Parameters
----------
source_url : str
The source URL for the bbcp transfer.
destination_url : str
The destination URL for the bbcp transfer.
Returns
-------
List[str]
A list of strings representing the bbcp command and its arguments.
"""
command = ["/usr/bin/bbcp"]
# preserve source mode, ownership, and dates.
command += ["-p"]
command += ["-P", "2"]
# Add verbose options
if ccat_data_transfer_settings.get("BBCP_VERBOSE") == 1:
command += ["-v"]
elif ccat_data_transfer_settings.get("BBCP_VERBOSE") == 2:
command += ["-V"]
# Add window size if specified
if window_size := ccat_data_transfer_settings.get("BBCP_WINDOW_SIZE"):
command.extend(["-w", str(window_size)])
# Add parallel streams if specified
if streams := ccat_data_transfer_settings.get("BBCP_PARALLEL_STREAMS"):
command.extend(["-s", str(streams)])
# Add port range for data stream connections (-Z first:last)
if port_range := ccat_data_transfer_settings.get("BBCP_PORT_RANGE"):
command.extend(["-Z", str(port_range)])
# Add source path if specified
if source_path := ccat_data_transfer_settings.get("BBCP_SOURCE_PATH"):
command.extend(["-S", str(source_path)])
# Add target path if specified
if target_path := ccat_data_transfer_settings.get("BBCP_TARGET_PATH"):
command.extend(["-T", str(target_path)])
# Add source and destination URLs
command.extend([source_url, destination_url])
# Log the command after ensuring all elements are strings
command_str = " ".join(str(arg) for arg in command)
logger.debug("BBCP command constructed", cmd=command_str)
return command
[docs]
def create_local_folder(folder: str) -> bool:
"""
Create a local folder.
Parameters
----------
folder : str
The path of the folder to be created.
Returns
-------
bool
True if the folder was created successfully or already exists, False otherwise.
"""
try:
os.makedirs(folder, exist_ok=True)
logger.info("local folder created or already exists", path=folder)
return True
except Exception as e:
logger.error("local folder creation failed", path=folder, error=str(e))
return False
[docs]
def create_remote_folder(user: str, host: str, folder: str) -> bool:
"""
Create a remote folder.
Parameters
----------
user : str
The username for SSH connection.
host : str
The hostname or IP address of the remote machine.
folder : str
The path of the folder to be created.
Returns
-------
bool
True if the folder was created successfully or already exists, False otherwise.
"""
try:
ssh_command = ["ssh", *SSH_OPTS, f"{user}@{host}", f"mkdir -p {shlex.quote(folder)}"]
subprocess.run(ssh_command, check=True)
logger.info(
"remote folder created or already exists",
path=folder,
host=host,
user=user,
)
return True
except Exception as e:
logger.error(
"remote folder creation failed",
path=folder,
host=host,
user=user,
error=str(e),
)
return False
[docs]
def make_long_term_archive_copy_command(
source_url: str, destination_url: str
) -> List[str]:
"""
Construct the bbcp command for copying to long term archive.
Parameters
----------
source_url : str
The source URL for the bbcp transfer.
destination_url : str
The destination URL for the bbcp transfer.
Returns
-------
List[str]
A list of strings representing the bbcp command and its arguments.
"""
command = make_bbcp_command(source_url, destination_url)
logger.debug("LTA copy command constructed", cmd=" ".join(command))
return command
[docs]
def run_ssh_command(user, host, command):
"""Run an SSH command on a remote host and return the output."""
ssh_command = ["ssh", *SSH_OPTS, f"{user}@{host}", command]
result = subprocess.run(
ssh_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
)
if result.returncode == 0:
return result.stdout.strip()
else:
logger.error(
"SSH command failed",
host=host,
user=user,
command=command,
stderr=result.stderr.strip(),
)
return None
[docs]
def check_remote_folder_size_gb(user, host, parent_path):
"""Check the size of a remote folder and return it in gigabytes."""
size_command = f"du -sb {shlex.quote(parent_path)} | cut -f1"
size_output = run_ssh_command(user, host, size_command)
if size_output:
try:
size_gb = int(size_output) / (1024**3) # Convert bytes to GB
return round(size_gb, 2) # Round to 2 decimal places
except ValueError:
logger.error("failed to parse remote folder size", raw_output=size_output)
return None
[docs]
def parse_bbcp_output(stdout: bytes, stderr: bytes, duration: float) -> Dict[str, Any]:
"""
Parse BBCP command output to extract transfer metrics.
Parameters
----------
stdout : bytes
Standard output from BBCP command
stderr : bytes
Standard error from BBCP command
duration : float
Total duration of the transfer
Returns
-------
Dict[str, Any]
Dictionary containing parsed metrics
"""
metrics = {
"duration": duration,
"peak_transfer_rate_mbps": 0,
"average_transfer_rate_mbps": 0,
"bytes_transferred": 0,
"number_of_streams": 0,
"network_errors": 0,
}
def convert_to_mbps(value: float, unit: str) -> float:
"""Convert various transfer rates to MB/s"""
unit = unit.upper()
if unit == "GB/S":
return value * 1024
elif unit == "KB/S":
return value / 1024
elif unit == "MB/S":
return value
return 0
def extract_rate(rate_str: str) -> float:
"""Extract numeric rate and unit, convert to MB/s"""
try:
# Match number and unit (e.g., "1.6 GB/s" or "15.8 MB/s")
match = re.search(r"([\d.]+)\s*([KMG]B/s)", rate_str, re.IGNORECASE)
if match:
value = float(match.group(1))
unit = match.group(2)
return convert_to_mbps(value, unit)
except (ValueError, AttributeError):
pass
return 0
stdout_text = stdout
stderr_text = stderr
combined_output = stdout_text + stderr_text
# Parse BBCP output
for line in combined_output.split("\n"):
# Look for bytes transferred and transfer rate
if "created;" in line and "bytes at" in line:
try:
# Extract bytes value
bytes_str = line.split("created;")[1].split("bytes")[0].strip()
metrics["bytes_transferred"] = int(bytes_str)
logger.debug(
"extracted bytes transferred",
bytes_transferred=metrics["bytes_transferred"],
)
# Extract peak rate - look for pattern "bytes at X MB/s"
try:
rate_part = line.split("bytes at")[-1].strip()
metrics["peak_transfer_rate_mbps"] = extract_rate(rate_part)
logger.debug(
"extracted peak transfer rate",
peak_transfer_rate_mbps=metrics["peak_transfer_rate_mbps"],
)
except (ValueError, IndexError) as e:
logger.error(
"failed to parse peak transfer rate",
error=e,
)
except (ValueError, IndexError) as e:
logger.error(
"failed to parse bytes transferred",
error=e,
)
# Look for effective transfer rate
if "effectively" in line:
try:
rate_str = line.split("effectively")[1].strip()
metrics["average_transfer_rate_mbps"] = extract_rate(rate_str)
logger.debug(
"extracted average transfer rate",
average_transfer_rate_mbps=metrics["average_transfer_rate_mbps"],
)
except (ValueError, IndexError) as e:
logger.error(
"failed to parse average transfer rate",
error=e,
)
# Look for network errors
if "error" in line.lower() or "failed" in line.lower():
metrics["network_errors"] += 1
logger.warning(
"detected network error in BBCP output",
network_errors_count=metrics["network_errors"],
)
logger.debug(
"BBCP output parsing complete",
stdout_length=len(stdout_text),
stderr_length=len(stderr_text),
bytes_transferred=metrics["bytes_transferred"],
peak_transfer_rate_mbps=metrics["peak_transfer_rate_mbps"],
average_transfer_rate_mbps=metrics["average_transfer_rate_mbps"],
network_errors=metrics["network_errors"],
)
return metrics
[docs]
def calculate_transfer_rate(file_size: int, duration: int) -> float:
"""
Calculate transfer rate in Mbps with float precision.
Parameters
----------
file_size : int
File size in bytes
duration : int
Transfer duration in seconds
Returns
-------
float
Transfer rate in Mbps
"""
if duration <= 0:
return 0.0
# Convert bytes to bits (multiply by 8)
# Convert to Mbps (divide by 1,000,000)
# Maintain float precision
return (file_size * 8.0) / (duration * 1_000_000.0)
[docs]
def generate_readable_filename(
raw_data_package, hash_value, file_type="raw", extension="tar"
):
"""
Generate a human-readable filename that includes metadata and a hash suffix.
Parameters
----------
raw_data_package : models.RawDataPackage
The raw data package containing metadata
hash_value : str
The original hash or UUID used for uniqueness
file_type : str
Type of file (e.g., "raw" or "transfer")
extension : str
File extension (without the dot)
Returns
-------
str
A human-readable filename with hash suffix
"""
# Extract date from package metadata or creation date
date_str = (
raw_data_package.created_at.strftime("%Y%m%d")
if hasattr(raw_data_package, "created_at") and raw_data_package.created_at
else time.strftime("%Y%m%d")
)
# Use first 8 chars of hash (still provides good uniqueness)
short_hash = hash_value[:8]
# Build filename with consistent extension
filename = f"{date_str}_{file_type}_{short_hash}.{extension}"
# Replace invalid characters
return re.sub(r"[^\w\.-]", "_", filename)
[docs]
def get_s3_key_for_package(
data_location: models.S3DataLocation, raw_data_package: models.RawDataPackage
) -> str:
"""
Construct S3 object key for a raw data package using consistent logic.
This function implements the same S3 key construction logic used in archive_manager.py
to ensure consistency between upload and download operations.
Parameters
----------
data_location : models.S3DataLocation
The S3 data location where the package is stored
raw_data_package : models.RawDataPackage
The raw data package to construct the key for
Returns
-------
str
The S3 object key for the package
Notes
-----
The S3 key is constructed as:
1. Replace underscores with slashes in the location name
2. Join with the package's relative path
3. Replace all slashes with underscores
4. Remove leading slash
"""
# Use the same logic as in archive_manager.py
location_path = data_location.name.replace("_", "/")
# Construct the destination path similar to the old implementation
destination = os.path.normpath(
os.path.join(
location_path,
raw_data_package.relative_path,
)
)
# Apply the same reformatting: replace / with _ and remove leading /
s3_key = destination.lstrip("/").replace("/", "_")
logger.debug(
"s3 key constructed for package",
location_name=data_location.name,
location_path=location_path,
package_relative_path=raw_data_package.relative_path,
s3_key=s3_key,
)
return s3_key
[docs]
def get_s3_key_for_file(
data_location: models.S3DataLocation, raw_data_file: models.RawDataFile
) -> str:
"""
Construct S3 object key for a raw data file using consistent logic.
This function implements the same S3 key construction logic for individual files.
Parameters
----------
data_location : models.S3DataLocation
The S3 data location where the file is stored
raw_data_file : models.RawDataFile
The raw data file to construct the key for
Returns
-------
str
The S3 object key for the file
"""
# Use the same logic as for packages
location_path = data_location.name.replace("_", "/")
# Construct the destination path
destination = os.path.normpath(
os.path.join(
location_path,
raw_data_file.relative_path,
)
)
# Apply the same reformatting: replace / with _ and remove leading /
s3_key = destination.lstrip("/").replace("/", "_")
logger.debug(
"s3 key constructed for file",
location_name=data_location.name,
file_relative_path=raw_data_file.relative_path,
s3_key=s3_key,
)
return s3_key
