From 8fbc1372273774aa337b415edf5c1d4c9360239b Mon Sep 17 00:00:00 2001 From: eveleighoj <35256612+eveleighoj@users.noreply.github.com> Date: Mon, 30 Mar 2026 23:01:26 +0100 Subject: [PATCH 1/2] update transform process to error if the resource doesnt exist and to properly download the relevant files --- .env.example | 6 + .gitignore | 3 +- bin/download_resources.py | 61 +++- bin/transform.sh | 5 +- bin/transform_resources.py | 264 +----------------- src/collection_task.egg-info/PKG-INFO | 5 - src/collection_task.egg-info/SOURCES.txt | 10 - .../dependency_links.txt | 1 - src/collection_task.egg-info/top_level.txt | 1 - src/collection_task/filtering.py | 59 ++++ src/collection_task/transform.py | 242 ++++++++++++++++ tests/integration/__init__.py | 0 tests/integration/test_transform.py | 47 ++++ 13 files changed, 406 insertions(+), 298 deletions(-) create mode 100644 .env.example delete mode 100644 src/collection_task.egg-info/PKG-INFO delete mode 100644 src/collection_task.egg-info/SOURCES.txt delete mode 100644 src/collection_task.egg-info/dependency_links.txt delete mode 100644 src/collection_task.egg-info/top_level.txt create mode 100644 src/collection_task/transform.py create mode 100644 tests/integration/__init__.py create mode 100644 tests/integration/test_transform.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..e84fb31 --- /dev/null +++ b/.env.example @@ -0,0 +1,6 @@ +# TRANSFORM VARIABLES +export COLLECTION=conservation-area +export DATASET=conservation-area +export DATASET_NAME=conservation-area +export TRANSFORMATION_OFFSET=1600 +export TRANSFORMATION_LIMIT=200 diff --git a/.gitignore b/.gitignore index 2e26e64..a960cc6 100644 --- a/.gitignore +++ b/.gitignore @@ -24,4 +24,5 @@ /log # python .venv -__pycache__/ \ No newline at end of file +__pycache__/ +*.egg-info/ \ No newline at end of file diff --git a/bin/download_resources.py b/bin/download_resources.py index beae993..0f974ba 100644 --- a/bin/download_resources.py +++ b/bin/download_resources.py @@ -7,18 +7,20 @@ from collection_task.downloading import download_files from collection_task.filtering import ( build_redirect_map, - build_dataset_resource_pairs, - apply_offset_and_limit, + select_resources_to_process, ) logger = logging.getLogger(__name__) -def download_resources(collection, collection_dir: str, bucket=None, base_url=None, collection_name=None, dataset=None, transformaiton_offset=None, transformation_limit=None, max_threads=4) -> None: - """Download resources for a collection., can limit and offset based on how many. transformation - rresource numberr may differ to +def download_resources(collection, collection_dir: str, bucket=None, base_url=None, collection_name=None, dataset=None, transformaiton_offset=None, transformation_limit=None, dataset_resource_dir="var/dataset-resource/", pipeline_dir="pipeline/", specification_dir="specification/", reprocess=False, max_threads=4) -> None: + """Download resources for a collection. + + Uses the same selection logic as transform_resources (via select_resources_to_process) + so only resources that will actually be transformed are downloaded. + Args: - collection: The collection object or None to load from collection_dir + collection: Unused, kept for backwards compatibility. collection_dir (str): The directory of the collection. bucket (str, optional): S3 bucket name to download from. If provided, will construct s3:// URLs. base_url (str, optional): Base URL for HTTP downloads (e.g., https://files.planning.data.gov.uk/). @@ -26,6 +28,10 @@ def download_resources(collection, collection_dir: str, bucket=None, base_url=No dataset (str, optional): Filter resources to only this dataset. transformaiton_offset (int, optional): Offset for filtering resources. transformation_limit (int, optional): Limit for filtering resources. + dataset_resource_dir (str, optional): Path to dataset resource logs (used for skip check). + pipeline_dir (str, optional): Path to pipeline config (used for config hash). + specification_dir (str, optional): Path to specification (used for specification hash). + reprocess (bool, optional): If True, skip the dataset-resource log check and download all resources. max_threads (int, optional): Maximum number of concurrent download threads. Defaults to 4. """ # Validate that either bucket or base_url is provided @@ -44,19 +50,21 @@ def download_resources(collection, collection_dir: str, bucket=None, base_url=No dataset_resource_map = collection.dataset_resource_map() redirect = build_redirect_map(collection.old_resource.entries) - dataset_resource_pairs = build_dataset_resource_pairs(dataset_resource_map, dataset=dataset) - total_pairs = len(dataset_resource_pairs) - dataset_resource_pairs = apply_offset_and_limit( - dataset_resource_pairs, + dataset_resource_pairs = select_resources_to_process( + dataset_resource_map=dataset_resource_map, + dataset_resource_dir=dataset_resource_dir, + pipeline_dir=pipeline_dir, + specification_dir=specification_dir, + dataset=dataset, offset=transformaiton_offset, limit=transformation_limit, - dataset=dataset, + reprocess=reprocess, ) # Extract unique resources to download (a resource only needs to be downloaded once) resources_to_download = list(set([res for _, res in dataset_resource_pairs])) - logger.info(f"Downloading resources for {len(dataset_resource_pairs)} transformation tasks (out of {total_pairs} total)") + logger.info(f"Downloading resources for {len(dataset_resource_pairs)} transformation tasks") # Build download map with URLs and output paths download_map = {} @@ -131,6 +139,27 @@ def download_resources(collection, collection_dir: str, bucket=None, base_url=No type=int, help="Limit for filtering resources" ) +@click.option( + "--dataset-resource-dir", + default="var/dataset-resource/", + help="Path to dataset resource logs (used to skip already up-to-date resources)" +) +@click.option( + "--pipeline-dir", + default="pipeline/", + help="Path to the pipeline configuration directory (used for config hash)" +) +@click.option( + "--specification-dir", + default="specification/", + help="Path to the specification directory (used for specification hash)" +) +@click.option( + "--reprocess", + is_flag=True, + default=False, + help="Download all resources, ignoring the dataset-resource skip check" +) @click.option( "--max-threads", default=4, @@ -147,7 +176,7 @@ def download_resources(collection, collection_dir: str, bucket=None, base_url=No is_flag=True, help="Enable debug logging" ) -def run_command(collection_dir, bucket, base_url, collection_name, dataset, offset, limit, max_threads, quiet, debug): +def run_command(collection_dir, bucket, base_url, collection_name, dataset, offset, limit, dataset_resource_dir, pipeline_dir, specification_dir, reprocess, max_threads, quiet, debug): """Download resources for a collection from S3 or HTTP(S) URLs. Either --bucket or --base-url must be provided. @@ -184,7 +213,11 @@ def run_command(collection_dir, bucket, base_url, collection_name, dataset, offs dataset=dataset, transformaiton_offset=offset, transformation_limit=limit, - max_threads=max_threads + dataset_resource_dir=dataset_resource_dir, + pipeline_dir=pipeline_dir, + specification_dir=specification_dir, + reprocess=reprocess, + max_threads=max_threads, ) click.echo("Download complete!") except ValueError as e: diff --git a/bin/transform.sh b/bin/transform.sh index aa333dd..edd766b 100755 --- a/bin/transform.sh +++ b/bin/transform.sh @@ -98,9 +98,8 @@ if [ -n "$DOWNLOAD_THREADS" ]; then DOWNLOAD_CMD="$DOWNLOAD_CMD --max-threads $DOWNLOAD_THREADS" fi -# Add verbose flag if needed -if [ -n "$VERBOSE" ]; then - DOWNLOAD_CMD="$DOWNLOAD_CMD --verbose" +if [ -n "$REPROCESS" ]; then + DOWNLOAD_CMD="$DOWNLOAD_CMD --reprocess" fi echo "Downloading resources..." diff --git a/bin/transform_resources.py b/bin/transform_resources.py index 5358421..de3f93f 100755 --- a/bin/transform_resources.py +++ b/bin/transform_resources.py @@ -1,271 +1,12 @@ import logging import sys import click -from pathlib import Path -from multiprocessing import Pool, cpu_count -from tqdm import tqdm -from digital_land import __version__ as dl_version -from digital_land.collection import Collection -from digital_land.pipeline import Pipeline -from digital_land.specification import Specification -from digital_land.commands import pipeline_run -from digital_land.utils.hash_utils import hash_directory -from digital_land.utils.dataset_resource_utils import resource_needs_processing - -from collection_task.filtering import ( - build_redirect_map, - build_dataset_resource_pairs, - apply_offset_and_limit, -) +from collection_task.transform import process_resources logger = logging.getLogger(__name__) -def process_single_resource(args): - """Process a single resource through the digital-land pipeline using Python library directly. - - Args: - args: Tuple of (old_resource, dataset, resource_path, endpoints, organisations, entry_date, config) - where old_resource is the original resource identifier (used for output filename) - and resource_path points to the actual file (may be redirected) - - Returns: - Tuple of (old_resource, success, error_message) - """ - old_resource, dataset, resource_path, endpoints, organisations, entry_date, config = args - - try: - # Build output directories - transformed_dir = Path(config['transformed_dir']) / dataset - issue_dir = Path(config['issue_dir']) / dataset - operational_issue_dir = Path(config['operational_issue_dir']) - output_log_dir = Path(config['output_log_dir']) - column_field_dir = Path(config['column_field_dir']) / dataset - dataset_resource_dir = Path(config['dataset_resource_dir']) / dataset - converted_resource_dir = Path(config['converted_resource_dir']) / dataset - - # Create directories - for directory in [transformed_dir, issue_dir, operational_issue_dir, - output_log_dir, column_field_dir, dataset_resource_dir, - converted_resource_dir]: - directory.mkdir(parents=True, exist_ok=True) - - # Build output path using old_resource (original resource identifier) - output_path = transformed_dir / f"{old_resource}.csv" - - # Initialize pipeline and specification objects - pipeline = Pipeline(path=config['pipeline_dir'], dataset=dataset) - specification = Specification(config.get('specification_dir', 'specification/')) - - # Parse endpoints and organisations (convert space-separated strings to lists) - endpoints_list = endpoints.split() if endpoints else [] - organisations_list = organisations.split() if organisations else [] - - # Call the pipeline_run function directly instead of subprocess - pipeline_run( - dataset=dataset, - pipeline=pipeline, - specification=specification, - input_path=str(resource_path), - output_path=output_path, - collection_dir=config.get('collection_dir', 'collection/'), - issue_dir=str(issue_dir), - operational_issue_dir=str(operational_issue_dir), - column_field_dir=str(column_field_dir), - dataset_resource_dir=str(dataset_resource_dir), - converted_resource_dir=str(converted_resource_dir), - organisation_path=config['organisation_path'], - config_path=config['config_path'], - endpoints=endpoints_list, - organisations=organisations_list, - entry_date=entry_date, - cache_dir=config.get('cache_dir', 'var/cache'), - resource=config.get('resource'), # For redirected resources - output_log_dir=str(output_log_dir), - ) - - return (old_resource, True, None) - - except Exception as e: - error_msg = str(e) - logger.error(f"Error processing {old_resource} for dataset {dataset}: {error_msg}") - return (old_resource, False, error_msg) - - -def process_resources( - collection_dir, - pipeline_dir="pipeline/", - specification_dir="specification/", - cache_dir="var/cache/", - transformed_dir="transformed/", - issue_dir="issue/", - operational_issue_dir="performance/operational_issue/", - output_log_dir="log/", - column_field_dir="var/column-field/", - dataset_resource_dir="var/dataset-resource/", - converted_resource_dir="var/converted-resource/", - dataset=None, - offset=None, - limit=None, - max_workers=None, - reprocess=False, -): - """Process resources using multiprocessing. - - Args: - collection_dir: Path to the collection directory - pipeline_dir: Path to the pipeline configuration directory - cache_dir: Path to the cache directory - transformed_dir: Path to the transformed output directory - issue_dir: Path to the issue directory - operational_issue_dir: Path to the operational issue directory - output_log_dir: Path to the output log directory - column_field_dir: Path to the column field directory - dataset_resource_dir: Path to the dataset resource directory - converted_resource_dir: Path to the converted resource directory - dataset: Optional dataset filter - only process resources from this dataset - offset: Optional offset for filtering resources - limit: Optional limit for filtering resources - max_workers: Number of worker processes (defaults to CPU count) - """ - # Load collection - collection = Collection(name=None, directory=collection_dir) - collection.load() - dataset_resource_map = collection.dataset_resource_map() - - redirect = build_redirect_map(collection.old_resource.entries) - dataset_resource_pairs = build_dataset_resource_pairs(dataset_resource_map, dataset=dataset) - total_pairs = len(dataset_resource_pairs) - if not reprocess: - config_hash = hash_directory(pipeline_dir) - specification_hash = hash_directory(specification_dir) - before_skip = len(dataset_resource_pairs) - dataset_resource_pairs = [ - (ds, resource) for ds, resource in dataset_resource_pairs - if resource_needs_processing( - dataset_resource_dir, ds, resource, - dl_version, config_hash, specification_hash, - ) - ] - skipped = before_skip - len(dataset_resource_pairs) - logger.info( - f"Skipping {skipped} already up-to-date resources, " - f"{len(dataset_resource_pairs)} to process" - ) - - dataset_resource_pairs = apply_offset_and_limit( - dataset_resource_pairs, - offset=offset, - limit=limit, - dataset=dataset, - ) - - # Now build tasks from the filtered list, considering redirects - tasks = [] - for ds, old_resource in dataset_resource_pairs: - # Get the actual resource to process (may be redirected) - resource = redirect.get(old_resource, old_resource) - - # Skip resources that have been removed (redirect to empty) - if not resource: - logger.info(f"Skipping removed resource: {old_resource}") - continue - - # Use the redirected resource path, but old_resource for metadata - resource_path = collection.resource_path(resource) - endpoints = " ".join(collection.resource_endpoints(old_resource)) - organisations = " ".join(collection.resource_organisations(old_resource)) - entry_date = collection.resource_start_date(old_resource) - - config = { - 'pipeline_dir': pipeline_dir, - 'cache_dir': cache_dir, - 'collection_dir': collection_dir, - 'transformed_dir': transformed_dir, - 'issue_dir': issue_dir, - 'operational_issue_dir': operational_issue_dir, - 'output_log_dir': output_log_dir, - 'column_field_dir': column_field_dir, - 'dataset_resource_dir': dataset_resource_dir, - 'converted_resource_dir': converted_resource_dir, - 'config_path': f"{cache_dir}config.sqlite3", - 'organisation_path': f"{cache_dir}organisation.csv", - } - - # If resource was redirected, include the old_resource in config - if resource != old_resource: - config['resource'] = old_resource - - tasks.append((old_resource, ds, resource_path, endpoints, organisations, entry_date, config)) - - logger.info(f"Processing {len(tasks)} transformation tasks (out of {total_pairs} total)") - - if not tasks: - logger.warning("No transformation tasks to process after applying filters") - return - - # Determine number of workers - if max_workers is None: - max_workers = cpu_count() - - logger.info(f"Using {max_workers} worker processes") - - # Process resources in parallel - use_progress_bar = sys.stdout.isatty() - - successful = 0 - failed = 0 - errors = [] - - with Pool(processes=max_workers) as pool: - if use_progress_bar: - # Interactive mode with progress bar - results = list(tqdm( - pool.imap(process_single_resource, tasks), - total=len(tasks), - desc="Processing resources" - )) - else: - # Non-interactive mode with progress logging at 10% intervals - results = [] - total_tasks = len(tasks) - logger.info(f"Starting processing of {total_tasks} transformation tasks...") - - # Calculate 10% interval (at least 1, at most total_tasks) - progress_interval = max(1, total_tasks // 10) - last_logged_percent = 0 - - for i, result in enumerate(pool.imap(process_single_resource, tasks), 1): - results.append(result) - - # Log at 10% intervals - current_percent = (i * 100) // total_tasks - if current_percent >= last_logged_percent + 10 or i == total_tasks: - logger.info(f"Progress: {i}/{total_tasks} tasks ({current_percent}%)") - last_logged_percent = current_percent - - logger.info(f"Completed processing of {total_tasks} transformation tasks") - - # Collect results - for resource, success, error_msg in results: - if success: - successful += 1 - else: - failed += 1 - errors.append((resource, error_msg)) - - # Log summary - logger.info(f"Processing complete: {successful} successful, {failed} failed") - - if errors: - logger.error(f"Failed resources:") - for resource, error_msg in errors: - logger.error(f" - {resource}: {error_msg}") - - return successful, failed, errors - - @click.command() @click.option( "--collection-dir", @@ -383,7 +124,6 @@ def run_command( debug ): """Process resources using multiprocessing instead of make.""" - # Configure logging if debug: logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s: %(message)s') elif quiet: @@ -392,7 +132,6 @@ def run_command( logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s: %(message)s') try: - # Process resources result = process_resources( collection_dir=collection_dir, pipeline_dir=pipeline_dir, @@ -412,7 +151,6 @@ def run_command( reprocess=reprocess, ) - # Handle case where no tasks were processed if result is None: click.echo("\nNo transformation tasks to process") sys.exit(0) diff --git a/src/collection_task.egg-info/PKG-INFO b/src/collection_task.egg-info/PKG-INFO deleted file mode 100644 index 78ffada..0000000 --- a/src/collection_task.egg-info/PKG-INFO +++ /dev/null @@ -1,5 +0,0 @@ -Metadata-Version: 2.4 -Name: collection-task -Version: 0.1.0 -License-File: LICENSE -Dynamic: license-file diff --git a/src/collection_task.egg-info/SOURCES.txt b/src/collection_task.egg-info/SOURCES.txt deleted file mode 100644 index a34a657..0000000 --- a/src/collection_task.egg-info/SOURCES.txt +++ /dev/null @@ -1,10 +0,0 @@ -LICENSE -README.md -pyproject.toml -src/collection_task/__init__.py -src/collection_task/downloading.py -src/collection_task/filtering.py -src/collection_task.egg-info/PKG-INFO -src/collection_task.egg-info/SOURCES.txt -src/collection_task.egg-info/dependency_links.txt -src/collection_task.egg-info/top_level.txt \ No newline at end of file diff --git a/src/collection_task.egg-info/dependency_links.txt b/src/collection_task.egg-info/dependency_links.txt deleted file mode 100644 index 8b13789..0000000 --- a/src/collection_task.egg-info/dependency_links.txt +++ /dev/null @@ -1 +0,0 @@ - diff --git a/src/collection_task.egg-info/top_level.txt b/src/collection_task.egg-info/top_level.txt deleted file mode 100644 index a0687cc..0000000 --- a/src/collection_task.egg-info/top_level.txt +++ /dev/null @@ -1 +0,0 @@ -collection_task diff --git a/src/collection_task/filtering.py b/src/collection_task/filtering.py index ff7e646..8b78057 100644 --- a/src/collection_task/filtering.py +++ b/src/collection_task/filtering.py @@ -3,6 +3,10 @@ import logging from typing import Dict, List, Optional, Tuple +from digital_land import __version__ as dl_version +from digital_land.utils.hash_utils import hash_directory +from digital_land.utils.dataset_resource_utils import resource_needs_processing + logger = logging.getLogger(__name__) @@ -85,6 +89,61 @@ def apply_offset_and_limit( return dataset_resource_pairs +def select_resources_to_process( + dataset_resource_map: Dict[str, List[str]], + dataset_resource_dir: str, + pipeline_dir: str, + specification_dir: str, + dataset: Optional[str] = None, + offset: Optional[int] = None, + limit: Optional[int] = None, + reprocess: bool = False, +) -> List[Tuple[str, str]]: + """Select the (dataset, resource) pairs to process or download. + + Applies the same logic in both download_resources and transform_resources + so the two steps always operate on the same set of resources: + + 1. Build the full list of (dataset, resource) pairs + 2. If not reprocess: filter to only those whose dataset-resource log is + out-of-date (different code version, config hash, or spec hash) + 3. Apply offset/limit to the resulting list + + Args: + dataset_resource_map: Dictionary mapping datasets to lists of resources + dataset_resource_dir: Path to dataset resource logs + pipeline_dir: Path to pipeline config directory (used for config hash) + specification_dir: Path to specification directory (used for spec hash) + dataset: Optional dataset name to filter to + offset: Optional offset into the final list + limit: Optional maximum number of pairs to return + reprocess: If True, skip the dataset-resource log check + + Returns: + List of (dataset, resource) tuples to process + """ + pairs = build_dataset_resource_pairs(dataset_resource_map, dataset=dataset) + + if not reprocess: + config_hash = hash_directory(pipeline_dir) + specification_hash = hash_directory(specification_dir) + before_skip = len(pairs) + pairs = [ + (ds, resource) for ds, resource in pairs + if resource_needs_processing( + dataset_resource_dir, ds, resource, + dl_version, config_hash, specification_hash, + ) + ] + skipped = before_skip - len(pairs) + logger.info( + f"Skipping {skipped} already up-to-date resources, " + f"{len(pairs)} to process" + ) + + return apply_offset_and_limit(pairs, offset=offset, limit=limit, dataset=dataset) + + def build_retired_resources_set(old_resource_entries: List[Dict]) -> set: """Build a set of retired resources (status 410). diff --git a/src/collection_task/transform.py b/src/collection_task/transform.py new file mode 100644 index 0000000..e17da43 --- /dev/null +++ b/src/collection_task/transform.py @@ -0,0 +1,242 @@ +"""Transform functions for processing collection resources through the pipeline.""" + +import logging +import sys +from pathlib import Path +from multiprocessing import Pool, cpu_count +from tqdm import tqdm + +from digital_land.collection import Collection +from digital_land.pipeline import Pipeline +from digital_land.specification import Specification +from digital_land.commands import pipeline_run + +from collection_task.filtering import ( + build_redirect_map, + select_resources_to_process, +) + +logger = logging.getLogger(__name__) + + +def process_single_resource(args): + """Process a single resource through the digital-land pipeline. + + Args: + args: Tuple of (old_resource, dataset, resource_path, endpoints, organisations, entry_date, config) + where old_resource is the original resource identifier (used for output filename) + and resource_path points to the actual file (may be redirected) + + Returns: + Tuple of (old_resource, success, error_message) + + Raises: + FileNotFoundError: If the resource file does not exist at resource_path + """ + old_resource, dataset, resource_path, endpoints, organisations, entry_date, config = args + + if not Path(resource_path).exists(): + raise FileNotFoundError( + f"Resource file not found for {old_resource} (dataset={dataset}): {resource_path}" + ) + + try: + # Build output directories + transformed_dir = Path(config['transformed_dir']) / dataset + issue_dir = Path(config['issue_dir']) / dataset + operational_issue_dir = Path(config['operational_issue_dir']) + output_log_dir = Path(config['output_log_dir']) + column_field_dir = Path(config['column_field_dir']) / dataset + dataset_resource_dir = Path(config['dataset_resource_dir']) / dataset + converted_resource_dir = Path(config['converted_resource_dir']) / dataset + + # Create directories + for directory in [transformed_dir, issue_dir, operational_issue_dir, + output_log_dir, column_field_dir, dataset_resource_dir, + converted_resource_dir]: + directory.mkdir(parents=True, exist_ok=True) + + # Build output path using old_resource (original resource identifier) + output_path = transformed_dir / f"{old_resource}.csv" + + # Initialize pipeline and specification objects + pipeline = Pipeline(path=config['pipeline_dir'], dataset=dataset) + specification = Specification(config.get('specification_dir', 'specification/')) + + # Parse endpoints and organisations (convert space-separated strings to lists) + endpoints_list = endpoints.split() if endpoints else [] + organisations_list = organisations.split() if organisations else [] + + pipeline_run( + dataset=dataset, + pipeline=pipeline, + specification=specification, + input_path=str(resource_path), + output_path=output_path, + collection_dir=config.get('collection_dir', 'collection/'), + issue_dir=str(issue_dir), + operational_issue_dir=str(operational_issue_dir), + column_field_dir=str(column_field_dir), + dataset_resource_dir=str(dataset_resource_dir), + converted_resource_dir=str(converted_resource_dir), + organisation_path=config['organisation_path'], + config_path=config['config_path'], + endpoints=endpoints_list, + organisations=organisations_list, + entry_date=entry_date, + cache_dir=config.get('cache_dir', 'var/cache'), + resource=config.get('resource'), # For redirected resources + output_log_dir=str(output_log_dir), + ) + + return (old_resource, True, None) + + except Exception as e: + error_msg = str(e) + logger.error(f"Error processing {old_resource} for dataset {dataset}: {error_msg}") + return (old_resource, False, error_msg) + + +def process_resources( + collection_dir, + pipeline_dir="pipeline/", + specification_dir="specification/", + cache_dir="var/cache/", + transformed_dir="transformed/", + issue_dir="issue/", + operational_issue_dir="performance/operational_issue/", + output_log_dir="log/", + column_field_dir="var/column-field/", + dataset_resource_dir="var/dataset-resource/", + converted_resource_dir="var/converted-resource/", + dataset=None, + offset=None, + limit=None, + max_workers=None, + reprocess=False, +): + """Process resources using multiprocessing. + + Args: + collection_dir: Path to the collection directory + pipeline_dir: Path to the pipeline configuration directory + cache_dir: Path to the cache directory + transformed_dir: Path to the transformed output directory + issue_dir: Path to the issue directory + operational_issue_dir: Path to the operational issue directory + output_log_dir: Path to the output log directory + column_field_dir: Path to the column field directory + dataset_resource_dir: Path to the dataset resource directory + converted_resource_dir: Path to the converted resource directory + dataset: Optional dataset filter - only process resources from this dataset + offset: Optional offset for filtering resources + limit: Optional limit for filtering resources + max_workers: Number of worker processes (defaults to CPU count) + reprocess: If True, skip the dataset-resource log check and reprocess all resources + """ + collection = Collection(name=None, directory=collection_dir) + collection.load() + dataset_resource_map = collection.dataset_resource_map() + total_pairs = len(dataset_resource_map) + + redirect = build_redirect_map(collection.old_resource.entries) + dataset_resource_pairs = select_resources_to_process( + dataset_resource_map=dataset_resource_map, + dataset_resource_dir=dataset_resource_dir, + pipeline_dir=pipeline_dir, + specification_dir=specification_dir, + dataset=dataset, + offset=offset, + limit=limit, + reprocess=reprocess, + ) + + tasks = [] + for ds, old_resource in dataset_resource_pairs: + resource = redirect.get(old_resource, old_resource) + + if not resource: + logger.info(f"Skipping removed resource: {old_resource}") + continue + + resource_path = collection.resource_path(resource) + endpoints = " ".join(collection.resource_endpoints(old_resource)) + organisations = " ".join(collection.resource_organisations(old_resource)) + entry_date = collection.resource_start_date(old_resource) + + config = { + 'pipeline_dir': pipeline_dir, + 'cache_dir': cache_dir, + 'collection_dir': collection_dir, + 'transformed_dir': transformed_dir, + 'issue_dir': issue_dir, + 'operational_issue_dir': operational_issue_dir, + 'output_log_dir': output_log_dir, + 'column_field_dir': column_field_dir, + 'dataset_resource_dir': dataset_resource_dir, + 'converted_resource_dir': converted_resource_dir, + 'config_path': f"{cache_dir}config.sqlite3", + 'organisation_path': f"{cache_dir}organisation.csv", + } + + if resource != old_resource: + config['resource'] = old_resource + + tasks.append((old_resource, ds, resource_path, endpoints, organisations, entry_date, config)) + + logger.info(f"Processing {len(tasks)} transformation tasks (out of {total_pairs} total)") + + if not tasks: + logger.warning("No transformation tasks to process after applying filters") + return + + if max_workers is None: + max_workers = cpu_count() + + logger.info(f"Using {max_workers} worker processes") + + use_progress_bar = sys.stdout.isatty() + successful = 0 + failed = 0 + errors = [] + + with Pool(processes=max_workers) as pool: + if use_progress_bar: + results = list(tqdm( + pool.imap(process_single_resource, tasks), + total=len(tasks), + desc="Processing resources" + )) + else: + results = [] + total_tasks = len(tasks) + logger.info(f"Starting processing of {total_tasks} transformation tasks...") + + progress_interval = max(1, total_tasks // 10) + last_logged_percent = 0 + + for i, result in enumerate(pool.imap(process_single_resource, tasks), 1): + results.append(result) + + current_percent = (i * 100) // total_tasks + if current_percent >= last_logged_percent + 10 or i == total_tasks: + logger.info(f"Progress: {i}/{total_tasks} tasks ({current_percent}%)") + last_logged_percent = current_percent + + logger.info(f"Completed processing of {total_tasks} transformation tasks") + + for resource, success, error_msg in results: + if success: + successful += 1 + else: + failed += 1 + errors.append((resource, error_msg)) + + logger.info(f"Processing complete: {successful} successful, {failed} failed") + + if errors: + logger.error("Failed resources:") + for resource, error_msg in errors: + logger.error(f" - {resource}: {error_msg}") + + return successful, failed, errors diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integration/test_transform.py b/tests/integration/test_transform.py new file mode 100644 index 0000000..95be835 --- /dev/null +++ b/tests/integration/test_transform.py @@ -0,0 +1,47 @@ +"""Integration tests for collection_task.transform""" + +import pytest +from collection_task.transform import process_single_resource + + +def _make_args(resource_path, old_resource="resource-hash-abc", dataset="some-dataset", **config_overrides): + config = { + "pipeline_dir": "pipeline/", + "cache_dir": "var/cache/", + "collection_dir": "collection/", + "transformed_dir": "transformed/", + "issue_dir": "issue/", + "operational_issue_dir": "performance/operational_issue/", + "output_log_dir": "log/", + "column_field_dir": "var/column-field/", + "dataset_resource_dir": "var/dataset-resource/", + "converted_resource_dir": "var/converted-resource/", + "config_path": "var/cache/config.sqlite3", + "organisation_path": "var/cache/organisation.csv", + **config_overrides, + } + return (old_resource, dataset, resource_path, "", "", None, config) + + +def test_process_single_resource_raises_if_resource_file_missing(tmp_path): + """Should raise FileNotFoundError when the resource file does not exist.""" + missing_path = tmp_path / "does-not-exist" + + with pytest.raises(FileNotFoundError, match="resource-hash-abc"): + process_single_resource(_make_args(missing_path)) + + +def test_process_single_resource_error_includes_dataset_name(tmp_path): + """FileNotFoundError message should include the dataset name for easier debugging.""" + missing_path = tmp_path / "does-not-exist" + + with pytest.raises(FileNotFoundError, match="some-dataset"): + process_single_resource(_make_args(missing_path)) + + +def test_process_single_resource_error_includes_path(tmp_path): + """FileNotFoundError message should include the expected path.""" + missing_path = tmp_path / "does-not-exist" + + with pytest.raises(FileNotFoundError, match=str(missing_path)): + process_single_resource(_make_args(missing_path)) From b983be14b38a40e487d16359bb47dbe1a68026f0 Mon Sep 17 00:00:00 2001 From: eveleighoj <35256612+eveleighoj@users.noreply.github.com> Date: Tue, 31 Mar 2026 18:07:40 +0100 Subject: [PATCH 2/2] make relevant changes to process from state --- .env.example | 3 +- .gitignore | 2 ++ bin/download_resources.py | 12 +++++-- bin/transform.sh | 34 +++++++++++++++++--- bin/transform_resources.py | 8 +++++ src/collection_task/filtering.py | 55 ++++++++++++++++++++++++++------ src/collection_task/transform.py | 2 ++ 7 files changed, 99 insertions(+), 17 deletions(-) diff --git a/.env.example b/.env.example index e84fb31..65cb8c8 100644 --- a/.env.example +++ b/.env.example @@ -2,5 +2,6 @@ export COLLECTION=conservation-area export DATASET=conservation-area export DATASET_NAME=conservation-area -export TRANSFORMATION_OFFSET=1600 +export TRANSFORMATION_OFFSET=1200 export TRANSFORMATION_LIMIT=200 +# export REPROCESS='' diff --git a/.gitignore b/.gitignore index a960cc6..6a7b085 100644 --- a/.gitignore +++ b/.gitignore @@ -16,6 +16,8 @@ /expectations /task/metadata.json /metadata.json +/task/state.json +/state.json /task/specification /specification /task/issue diff --git a/bin/download_resources.py b/bin/download_resources.py index 0f974ba..1ca1322 100644 --- a/bin/download_resources.py +++ b/bin/download_resources.py @@ -13,7 +13,7 @@ logger = logging.getLogger(__name__) -def download_resources(collection, collection_dir: str, bucket=None, base_url=None, collection_name=None, dataset=None, transformaiton_offset=None, transformation_limit=None, dataset_resource_dir="var/dataset-resource/", pipeline_dir="pipeline/", specification_dir="specification/", reprocess=False, max_threads=4) -> None: +def download_resources(collection, collection_dir: str, bucket=None, base_url=None, collection_name=None, dataset=None, transformaiton_offset=None, transformation_limit=None, dataset_resource_dir="var/dataset-resource/", pipeline_dir="pipeline/", specification_dir="specification/", reprocess=False, max_threads=4, state_path=None) -> None: """Download resources for a collection. Uses the same selection logic as transform_resources (via select_resources_to_process) @@ -59,6 +59,7 @@ def download_resources(collection, collection_dir: str, bucket=None, base_url=No offset=transformaiton_offset, limit=transformation_limit, reprocess=reprocess, + state_path=state_path, ) # Extract unique resources to download (a resource only needs to be downloaded once) @@ -154,6 +155,12 @@ def download_resources(collection, collection_dir: str, bucket=None, base_url=No default="specification/", help="Path to the specification directory (used for specification hash)" ) +@click.option( + "--state-path", + default=None, + type=click.Path(exists=True), + help="Path to state.json for stable ordered resource list" +) @click.option( "--reprocess", is_flag=True, @@ -176,7 +183,7 @@ def download_resources(collection, collection_dir: str, bucket=None, base_url=No is_flag=True, help="Enable debug logging" ) -def run_command(collection_dir, bucket, base_url, collection_name, dataset, offset, limit, dataset_resource_dir, pipeline_dir, specification_dir, reprocess, max_threads, quiet, debug): +def run_command(collection_dir, bucket, base_url, collection_name, dataset, offset, limit, dataset_resource_dir, pipeline_dir, specification_dir, state_path, reprocess, max_threads, quiet, debug): """Download resources for a collection from S3 or HTTP(S) URLs. Either --bucket or --base-url must be provided. @@ -218,6 +225,7 @@ def run_command(collection_dir, bucket, base_url, collection_name, dataset, offs specification_dir=specification_dir, reprocess=reprocess, max_threads=max_threads, + state_path=state_path, ) click.echo("Download complete!") except ValueError as e: diff --git a/bin/transform.sh b/bin/transform.sh index edd766b..6d2fc3c 100755 --- a/bin/transform.sh +++ b/bin/transform.sh @@ -38,10 +38,34 @@ make init echo "Step 3: Building collection database..." make collection -# Step 4: Download dataset resource logs (used to skip already up-to-date resources) +# Step 4: Download state.json (used for stable batch ordering) and dataset resource logs +# Use local state.json if it exists, allowing manual edits +STATE_PATH=${STATE_PATH:-"state.json"} + +if [ -f "$STATE_PATH" ]; then + echo "Step 4a: Using existing state.json at $STATE_PATH" +else + echo "Step 4a: Downloading state.json..." + if [ -n "$COLLECTION_DATASET_BUCKET_NAME" ]; then + aws s3 cp "s3://${COLLECTION_DATASET_BUCKET_NAME}/${COLLECTION_NAME}-collection/state.json" "$STATE_PATH" + elif [ -n "$DATASTORE_URL" ]; then + base=$(echo "$DATASTORE_URL" | sed 's:/*$::') + curl --fail -o "$STATE_PATH" "${base}/${COLLECTION_NAME}-collection/state.json" + else + echo "Error: Either COLLECTION_DATASET_BUCKET_NAME or DATASTORE_URL must be set" + exit 1 + fi + + if [ ! -f "$STATE_PATH" ]; then + echo "Error: Failed to download state.json" + exit 1 + fi +fi + +# Download dataset resource logs (used to skip already up-to-date resources within a batch) # Skipped when REPROCESS is set - logs will be freshly written by the transform step if [ -z "$REPROCESS" ]; then - echo "Step 4: Downloading dataset resource logs..." + echo "Step 4b: Downloading dataset resource logs..." DATASET_RESOURCE_CMD="python bin/download_dataset_resource.py --collection-dir $COLLECTION_DIR --collection-name $COLLECTION_NAME --dataset-resource-dir $DATASET_RESOURCE_DIR" @@ -61,14 +85,14 @@ if [ -z "$REPROCESS" ]; then echo "Command: $DATASET_RESOURCE_CMD" eval $DATASET_RESOURCE_CMD else - echo "Step 4: Skipping dataset resource log download - REPROCESS is set, logs will be written fresh" + echo "Step 4b: Skipping dataset resource log download - REPROCESS is set, logs will be written fresh" fi # Step 5: Download resources echo "Step 5: Downloading resources..." # Build the download command -DOWNLOAD_CMD="python bin/download_resources.py --collection-dir $COLLECTION_DIR" +DOWNLOAD_CMD="python bin/download_resources.py --collection-dir $COLLECTION_DIR --state-path $STATE_PATH" # Add bucket or base URL (bucket takes precedence, matching makefile convention) if [ -n "$COLLECTION_DATASET_BUCKET_NAME" ]; then @@ -111,7 +135,7 @@ echo "Resources downloaded successfully" # Step 6: Transform resources using Python multiprocessing echo "Step 6: Transforming resources..." -TRANSFORM_CMD="python bin/transform_resources.py --collection-dir $COLLECTION_DIR" +TRANSFORM_CMD="python bin/transform_resources.py --collection-dir $COLLECTION_DIR --state-path $STATE_PATH" # Add directory parameters TRANSFORM_CMD="$TRANSFORM_CMD --pipeline-dir $PIPELINE_DIR" diff --git a/bin/transform_resources.py b/bin/transform_resources.py index de3f93f..ccccbac 100755 --- a/bin/transform_resources.py +++ b/bin/transform_resources.py @@ -87,6 +87,12 @@ default="specification/", help="Path to the specification directory" ) +@click.option( + "--state-path", + default=None, + type=click.Path(exists=True), + help="Path to state.json for stable ordered resource list" +) @click.option( "--reprocess", is_flag=True, @@ -119,6 +125,7 @@ def run_command( offset, limit, max_workers, + state_path, reprocess, quiet, debug @@ -148,6 +155,7 @@ def run_command( offset=offset, limit=limit, max_workers=max_workers, + state_path=state_path, reprocess=reprocess, ) diff --git a/src/collection_task/filtering.py b/src/collection_task/filtering.py index 8b78057..faf9bc7 100644 --- a/src/collection_task/filtering.py +++ b/src/collection_task/filtering.py @@ -1,5 +1,6 @@ """Filtering and resource management functions for collection tasks.""" +import json import logging from typing import Dict, List, Optional, Tuple @@ -89,6 +90,31 @@ def apply_offset_and_limit( return dataset_resource_pairs +def load_state_resources(state_path: str, dataset: str) -> List[Tuple[str, str]]: + """Load the ordered resource list for a dataset from a state.json file. + + Args: + state_path: Path to state.json + dataset: Dataset name to load resources for + + Returns: + List of (dataset, resource) tuples in the order defined by state.json + + Raises: + FileNotFoundError: If state_path does not exist + KeyError: If state.json does not contain 'transform_resources' or dataset is not found + """ + with open(state_path) as f: + state = json.load(f) + + transform_resources = state["transform_resources"] + + if dataset not in transform_resources: + raise KeyError(f"Dataset '{dataset}' not found in state.json transform_resources") + + return [(dataset, resource) for resource in transform_resources[dataset]] + + def select_resources_to_process( dataset_resource_map: Dict[str, List[str]], dataset_resource_dir: str, @@ -98,16 +124,18 @@ def select_resources_to_process( offset: Optional[int] = None, limit: Optional[int] = None, reprocess: bool = False, + state_path: Optional[str] = None, ) -> List[Tuple[str, str]]: """Select the (dataset, resource) pairs to process or download. - Applies the same logic in both download_resources and transform_resources - so the two steps always operate on the same set of resources: + If state_path is provided, the stable ordered list from state.json is used + as the base for offset/limit. Otherwise falls back to building from + dataset_resource_map. The skip check (resource_needs_processing) is always + applied after offset/limit so batch boundaries remain stable. - 1. Build the full list of (dataset, resource) pairs - 2. If not reprocess: filter to only those whose dataset-resource log is - out-of-date (different code version, config hash, or spec hash) - 3. Apply offset/limit to the resulting list + 1. Build the ordered list (from state.json if available, else dataset_resource_map) + 2. Apply offset/limit to get a stable batch slice + 3. If not reprocess: filter the slice to only those needing processing Args: dataset_resource_map: Dictionary mapping datasets to lists of resources @@ -115,14 +143,23 @@ def select_resources_to_process( pipeline_dir: Path to pipeline config directory (used for config hash) specification_dir: Path to specification directory (used for spec hash) dataset: Optional dataset name to filter to - offset: Optional offset into the final list + offset: Optional offset into the stable ordered list limit: Optional maximum number of pairs to return reprocess: If True, skip the dataset-resource log check + state_path: Optional path to state.json for stable ordered resource list Returns: List of (dataset, resource) tuples to process """ - pairs = build_dataset_resource_pairs(dataset_resource_map, dataset=dataset) + if state_path: + if not dataset: + raise ValueError("dataset is required when state_path is provided") + pairs = load_state_resources(state_path, dataset=dataset) + else: + pairs = build_dataset_resource_pairs(dataset_resource_map, dataset=dataset) + + # Apply offset/limit to stable list first so batch boundaries never shift + pairs = apply_offset_and_limit(pairs, offset=offset, limit=limit, dataset=dataset) if not reprocess: config_hash = hash_directory(pipeline_dir) @@ -141,7 +178,7 @@ def select_resources_to_process( f"{len(pairs)} to process" ) - return apply_offset_and_limit(pairs, offset=offset, limit=limit, dataset=dataset) + return pairs def build_retired_resources_set(old_resource_entries: List[Dict]) -> set: diff --git a/src/collection_task/transform.py b/src/collection_task/transform.py index e17da43..c7f1bd5 100644 --- a/src/collection_task/transform.py +++ b/src/collection_task/transform.py @@ -114,6 +114,7 @@ def process_resources( limit=None, max_workers=None, reprocess=False, + state_path=None, ): """Process resources using multiprocessing. @@ -149,6 +150,7 @@ def process_resources( offset=offset, limit=limit, reprocess=reprocess, + state_path=state_path, ) tasks = []