diff --git a/README.md b/README.md index 6a439630..f7929f31 100644 --- a/README.md +++ b/README.md @@ -225,6 +225,24 @@ mapillary_tools process MY_CAPTURE_DIR \ --cutoff_time 120 \ ``` +To receive a machine-readable inventory of recognized inputs and unsupported +files without changing the image description format, specify an optional +process report: + +```sh +mapillary_tools process MY_CAPTURE_DIR \ + --process_report_path /tmp/mapillary_process_report.json +``` + +The version 1 report contains `discovered_file_count`, `processed_file_count`, +`skipped_file_count`, and `skipped_files`. Recognized files that produce a +description-level processing error are included in `processed_file_count`. +Unsupported entries include their filename, category, stable reason code, and +normalized extension. Hidden/system files and common GPS or metadata sidecars +are ignored during directory discovery, but an explicitly supplied unsupported +file is always reported. The report is written only to the requested local path +and contains absolute source filenames, so treat it as local capture metadata. + ## Upload After processing you should get the [image description file](#image-description). Pass it to the `upload` command to upload them: @@ -237,6 +255,27 @@ mapillary_tools upload MY_CAPTURE_DIR \ --organization_key "my_organization_id" ``` +## Check Upload History + +To check whether processed image sequences or videos were previously uploaded +from this device, use the same import paths and description file that would be +passed to `upload`: + +```sh +mapillary_tools check_upload_history MY_CAPTURE_DIR \ + --desc_path /tmp/mapillary_image_description.json +``` + +The command prints a JSON array with one entry per upload candidate. Each entry +contains the file type, sequence UUID when applicable, sequence checksum, +member filenames, the subset found in history as +`already_uploaded_filenames`, and an `already_uploaded` boolean. +`already_uploaded` is true only when every member filename is represented in +local history. For images, this includes files that were members of a larger +previously uploaded sequence. It reads only the local upload history: it does +not require authentication, make network requests, upload data, or modify +history. + # Advanced Usage ## Local Video Processing diff --git a/mapillary_tools/commands/__init__.py b/mapillary_tools/commands/__init__.py index fd0c99fb..96fadd72 100644 --- a/mapillary_tools/commands/__init__.py +++ b/mapillary_tools/commands/__init__.py @@ -6,6 +6,7 @@ # ruff: noqa: F401 from . import ( authenticate, + check_upload_history, process, process_and_upload, sample_video, diff --git a/mapillary_tools/commands/__main__.py b/mapillary_tools/commands/__main__.py index bba29dac..194643bd 100644 --- a/mapillary_tools/commands/__main__.py +++ b/mapillary_tools/commands/__main__.py @@ -16,6 +16,7 @@ from ..utils import configure_logger, get_app_name from . import ( authenticate, + check_upload_history, process, process_and_upload, sample_video, @@ -27,6 +28,7 @@ mapillary_tools_commands = [ process, + check_upload_history, upload, sample_video, video_process, @@ -62,13 +64,24 @@ def add_general_arguments(parser, command): default=False, required=False, ) - elif command in ["upload"]: + elif command in ["upload", "check_upload_history"]: parser.add_argument( "import_path", help="Paths to your images or videos.", nargs="+", type=Path, ) + if command == "check_upload_history": + parser.add_argument( + "--skip_subfolders", + help=( + "Skip all subfolders and import only files in the given " + "IMPORT_PATH." + ), + action="store_true", + default=False, + required=False, + ) elif command in ["process", "process_and_upload"]: parser.add_argument( "import_path", diff --git a/mapillary_tools/commands/check_upload_history.py b/mapillary_tools/commands/check_upload_history.py new file mode 100644 index 00000000..bad22fb3 --- /dev/null +++ b/mapillary_tools/commands/check_upload_history.py @@ -0,0 +1,41 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# +# This source code is licensed under the BSD license found in the +# LICENSE file in the root directory of this source tree. + +from __future__ import annotations + +import argparse +import inspect +import json +import sys + +from ..upload import check_upload_history +from .process import bold_text + + +class Command: + name = "check_upload_history" + help = "Check whether processed data exists in local upload history" + + def add_basic_arguments(self, parser: argparse.ArgumentParser): + group = parser.add_argument_group(bold_text("UPLOAD HISTORY OPTIONS")) + group.add_argument( + "--desc_path", + help=( + "Path to the description file with processed image and video metadata." + ), + default=None, + required=False, + ) + + def run(self, vars_args: dict): + results = check_upload_history( + **{ + key: value + for key, value in vars_args.items() + if key in inspect.getfullargspec(check_upload_history).args + } + ) + json.dump(results, sys.stdout, separators=(",", ":")) + sys.stdout.write("\n") diff --git a/mapillary_tools/commands/process.py b/mapillary_tools/commands/process.py index 4af4b95b..fc205c5d 100644 --- a/mapillary_tools/commands/process.py +++ b/mapillary_tools/commands/process.py @@ -9,7 +9,7 @@ import inspect from pathlib import Path -from .. import constants, types +from .. import constants, process_report, types from ..process_geotag_properties import ( DEFAULT_GEOTAG_SOURCE_OPTIONS, process_finalize, @@ -45,6 +45,16 @@ def add_basic_arguments(self, parser: argparse.ArgumentParser): default=False, required=False, ) + parser.add_argument( + "--process_report_path", + help=( + "Optional path to write a versioned JSON report describing " + "recognized process inputs and unsupported files." + ), + type=Path, + default=None, + required=False, + ) parser.add_argument( "--filetypes", "--file_types", @@ -217,6 +227,14 @@ def add_basic_arguments(self, parser: argparse.ArgumentParser): ) def run(self, vars_args: dict): + process_report_path = vars_args.get("process_report_path") + if process_report_path is not None: + report = process_report.build_process_report( + import_path=vars_args["import_path"], + skip_subfolders=vars_args.get("skip_subfolders", False), + ) + process_report.write_process_report(process_report_path, report) + metadatas = process_geotag_properties( **( { diff --git a/mapillary_tools/history.py b/mapillary_tools/history.py index 92c7121f..c43be869 100644 --- a/mapillary_tools/history.py +++ b/mapillary_tools/history.py @@ -63,6 +63,77 @@ def read_history_record(md5sum: str) -> None | T.Dict[str, T.Any]: return None +def _normalize_md5sum(value: T.Any) -> str | None: + if not isinstance(value, str) or len(value) != 32: + return None + + try: + _validate_hexdigits(value) + except ValueError: + return None + + return value.lower() + + +def find_uploaded_image_md5s(md5sums: T.Iterable[str]) -> set[str]: + """Find image checksums stored in upload-history descriptions. + + Upload history is sharded by the checksum of a whole sequence, so checking + whether images came from a previously uploaded larger sequence requires a + bounded scan of the records. Only requested checksums are retained, and the + scan stops as soon as all of them have been found. + """ + if not constants.MAPILLARY_UPLOAD_HISTORY_PATH: + return set() + + wanted = { + normalized + for md5sum in md5sums + if (normalized := _normalize_md5sum(md5sum)) is not None + } + if not wanted: + return set() + + root = Path(constants.MAPILLARY_UPLOAD_HISTORY_PATH) + if not root.is_dir(): + return set() + + found: set[str] = set() + try: + history_paths = root.glob("*/*.json") + for path in history_paths: + try: + with path.open("r", encoding="utf-8") as fp: + record = json.load(fp) + except (OSError, UnicodeError, json.JSONDecodeError) as ex: + LOG.warning("Failed to read upload history %s: %s", path, ex) + continue + + if not isinstance(record, dict): + LOG.warning("Invalid upload history record %s", path) + continue + + descs = record.get("descs") + if not isinstance(descs, list): + continue + + for desc in descs: + if not isinstance(desc, dict): + continue + if desc.get("filetype") != types.FileType.IMAGE.value: + continue + md5sum = _normalize_md5sum(desc.get("md5sum")) + if md5sum in wanted: + found.add(md5sum) + + if found == wanted: + break + except OSError as ex: + LOG.warning("Failed to scan upload history %s: %s", root, ex) + + return found + + def write_history( md5sum: str, params: JSONDict, diff --git a/mapillary_tools/process_report.py b/mapillary_tools/process_report.py new file mode 100644 index 00000000..f11f2e8c --- /dev/null +++ b/mapillary_tools/process_report.py @@ -0,0 +1,160 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# +# This source code is licensed under the BSD license found in the +# LICENSE file in the root directory of this source tree. + +from __future__ import annotations + +import json +import typing as T +from pathlib import Path + +from . import utils + + +PROCESS_REPORT_SCHEMA_VERSION = 1 + + +class ProcessReportDetails(T.TypedDict): + extension: str + + +class ProcessReportSkippedFile(T.TypedDict): + filename: str + category: T.Literal["unsupported"] + reason_code: T.Literal["unsupported_format"] + details: ProcessReportDetails + + +class ProcessReport(T.TypedDict): + schema_version: int + discovered_file_count: int + processed_file_count: int + skipped_file_count: int + skipped_files: list[ProcessReportSkippedFile] + + +# These files can accompany capture media but are not themselves process inputs. +# Explicitly supplied files are never ignored, regardless of their name or suffix. +_IGNORED_DISCOVERED_EXTENSIONS = { + ".exif", + ".fit", + ".gpx", + ".json", + ".kml", + ".kmz", + ".lrv", + ".nmea", + ".srt", + ".tcx", + ".thm", + ".vtt", + ".xml", + ".xmp", + ".zip", +} +_IGNORED_DISCOVERED_FILENAMES = { + ".ds_store", + "desktop.ini", + "ehthumbs.db", + "thumbs.db", +} +_IGNORED_DISCOVERED_DIRNAMES = {"__macosx"} + + +def _is_supported_process_file(path: Path) -> bool: + return utils.is_image_file(path) or utils.is_video_file(path) + + +def _is_ignored_discovered_file(path: Path, import_root: Path) -> bool: + if path.name.casefold() in _IGNORED_DISCOVERED_FILENAMES: + return True + if path.suffix.lower() in _IGNORED_DISCOVERED_EXTENSIONS: + return True + + try: + relative_parts = path.relative_to(import_root).parts[:-1] + except ValueError: + relative_parts = path.parts[:-1] + return any( + part.casefold() in _IGNORED_DISCOVERED_DIRNAMES for part in relative_parts + ) + + +def _unsupported_file(path: Path) -> ProcessReportSkippedFile: + return { + "filename": str(path.resolve()), + "category": "unsupported", + "reason_code": "unsupported_format", + "details": {"extension": path.suffix.lower()}, + } + + +def build_process_report( + import_path: Path | T.Sequence[Path], + skip_subfolders: bool = False, +) -> ProcessReport: + if isinstance(import_path, Path): + import_paths = [import_path] + else: + import_paths = list(import_path) + import_paths = list(utils.deduplicate_paths(import_paths)) + + processed_paths: dict[Path, Path] = {} + unsupported_paths: dict[Path, Path] = {} + explicit_paths: set[Path] = set() + + # Explicit regular files take precedence over directory filtering. In + # particular, explicitly selected hidden, system, and sidecar files must be + # reported as unsupported instead of silently ignored. + for path in import_paths: + if not path.is_file(): + continue + resolved = path.resolve() + explicit_paths.add(resolved) + if _is_supported_process_file(path): + processed_paths[resolved] = path + else: + unsupported_paths[resolved] = path + + for import_root in (path for path in import_paths if path.is_dir()): + discovered_paths = sorted( + utils.iterate_files(import_root, recursive=not skip_subfolders), + key=lambda path: str(path.resolve()), + ) + for path in discovered_paths: + if not path.is_file(): + continue + resolved = path.resolve() + if resolved in explicit_paths: + continue + if _is_supported_process_file(path): + processed_paths[resolved] = path + elif not _is_ignored_discovered_file(path, import_root): + unsupported_paths[resolved] = path + + skipped_files = sorted( + (_unsupported_file(path) for path in unsupported_paths.values()), + key=lambda skipped: skipped["filename"], + ) + processed_file_count = len(processed_paths) + skipped_file_count = len(skipped_files) + return { + "schema_version": PROCESS_REPORT_SCHEMA_VERSION, + "discovered_file_count": processed_file_count + skipped_file_count, + "processed_file_count": processed_file_count, + "skipped_file_count": skipped_file_count, + "skipped_files": skipped_files, + } + + +def write_process_report(path: Path, report: ProcessReport) -> None: + with path.open("w", encoding="utf-8") as fp: + json.dump( + report, + fp, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + fp.write("\n") diff --git a/mapillary_tools/upload.py b/mapillary_tools/upload.py index 5a732c2f..e78e84a2 100644 --- a/mapillary_tools/upload.py +++ b/mapillary_tools/upload.py @@ -36,6 +36,16 @@ JSONDict = T.Dict[str, T.Union[str, int, float, None]] + +class UploadHistoryCheckResult(T.TypedDict): + file_type: str + sequence_uuid: str | None + sequence_md5sum: str + filenames: list[str] + already_uploaded_filenames: list[str] + already_uploaded: bool + + LOG = logging.getLogger(__name__) @@ -136,6 +146,135 @@ def upload( _show_upload_summary(stats, upload_errors) +def _history_check_result( + file_type: types.FileType, + sequence_uuid: str | None, + sequence_md5sum: str, + filenames: T.Iterable[Path], + already_uploaded_filenames: T.Iterable[Path], +) -> UploadHistoryCheckResult: + filename_strings = [str(filename) for filename in filenames] + already_uploaded_filename_strings = [ + str(filename) for filename in already_uploaded_filenames + ] + return { + "file_type": file_type.value, + "sequence_uuid": sequence_uuid, + "sequence_md5sum": sequence_md5sum, + "filenames": filename_strings, + "already_uploaded_filenames": already_uploaded_filename_strings, + "already_uploaded": ( + bool(filename_strings) + and len(already_uploaded_filename_strings) == len(filename_strings) + ), + } + + +def _has_upload_history_record(md5sum: str) -> bool: + try: + record = history.read_history_record(md5sum) + except (OSError, UnicodeError) as ex: + LOG.warning("Failed to read upload history %s: %s", md5sum, ex) + return False + return isinstance(record, dict) + + +def check_upload_history( + import_path: Path | T.Sequence[Path], + desc_path: str | None = None, + skip_subfolders: bool = False, + _metadatas_from_process: T.Sequence[types.MetadataOrError] | None = None, +) -> list[UploadHistoryCheckResult]: + """Return upload-history matches without uploading or changing history.""" + import_paths = _normalize_import_paths(import_path) + metadatas = _load_descs(_metadatas_from_process, import_paths, desc_path) + candidates = _find_upload_candidates( + metadatas, import_paths, skip_subfolders=skip_subfolders + ) + + results: list[UploadHistoryCheckResult] = [] + image_results_to_check: list[ + tuple[UploadHistoryCheckResult, list[types.ImageMetadata]] + ] = [] + image_md5sums_to_check: set[str] = set() + + for sequence_uuid, sequence in types.group_and_sort_images( + candidates.image_metadatas + ).items(): + sequence_md5sum = types.update_sequence_md5sum(sequence) + filenames = [metadata.filename for metadata in sequence] + sequence_already_uploaded = _has_upload_history_record(sequence_md5sum) + if sequence_already_uploaded: + already_uploaded_filenames = filenames + else: + already_uploaded_filenames = [] + for metadata in sequence: + assert isinstance(metadata.md5sum, str), "md5sum should be calculated" + image_md5sums_to_check.add(metadata.md5sum) + + result = _history_check_result( + types.FileType.IMAGE, + sequence_uuid, + sequence_md5sum, + filenames, + already_uploaded_filenames, + ) + results.append(result) + + if not sequence_already_uploaded: + image_results_to_check.append((result, sequence)) + + uploaded_image_md5sums = history.find_uploaded_image_md5s(image_md5sums_to_check) + for result, sequence in image_results_to_check: + already_uploaded_filename_strings: list[str] = [] + for metadata in sequence: + assert isinstance(metadata.md5sum, str), "md5sum should be calculated" + if metadata.md5sum.lower() in uploaded_image_md5sums: + already_uploaded_filename_strings.append(str(metadata.filename)) + result["already_uploaded_filenames"] = already_uploaded_filename_strings + result["already_uploaded"] = len(already_uploaded_filename_strings) == len( + result["filenames"] + ) + + for video_metadata in sorted( + candidates.video_metadatas, key=lambda metadata: metadata.filename + ): + video_metadata.update_md5sum() + assert isinstance(video_metadata.md5sum, str), "md5sum should be calculated" + filenames = [video_metadata.filename] + already_uploaded_filenames = ( + filenames if _has_upload_history_record(video_metadata.md5sum) else [] + ) + results.append( + _history_check_result( + video_metadata.filetype, + None, + video_metadata.md5sum, + filenames, + already_uploaded_filenames, + ) + ) + + for zip_path in sorted(candidates.zip_paths): + with zip_path.open("rb") as zip_fp: + sequence_md5sum = uploader.ZipUploader._extract_sequence_md5sum(zip_fp) + filenames = [zip_path] + already_uploaded_filenames = ( + filenames if _has_upload_history_record(sequence_md5sum) else [] + ) + results.append( + _history_check_result( + types.FileType.ZIP, + None, + sequence_md5sum, + filenames, + already_uploaded_filenames, + ) + ) + + return results + + def zip_images(import_path: Path, zip_dir: Path, desc_path: str | None = None): if not import_path.is_dir(): raise exceptions.MapillaryFileNotFoundError( @@ -543,6 +682,12 @@ def _api_logging_failed(payload: dict, exc: Exception, dry_run: bool = False): _M = T.TypeVar("_M", bound=types.Metadata) +class _UploadCandidates(T.NamedTuple): + image_metadatas: list[types.ImageMetadata] + video_metadatas: list[types.VideoMetadata] + zip_paths: list[Path] + + def _find_metadata_with_filename_existed_in( metadatas: T.Iterable[_M], paths: T.Iterable[Path] ) -> list[_M]: @@ -550,32 +695,46 @@ def _find_metadata_with_filename_existed_in( return [d for d in metadatas if d.filename.resolve() in resolved_image_paths] -def _gen_upload_everything( - mly_uploader: uploader.Uploader, +def _find_upload_candidates( metadatas: T.Sequence[types.Metadata], import_paths: T.Sequence[Path], skip_subfolders: bool, -): - # Upload images +) -> _UploadCandidates: image_metadatas = _find_metadata_with_filename_existed_in( (m for m in metadatas if isinstance(m, types.ImageMetadata)), utils.find_images(import_paths, skip_subfolders=skip_subfolders), ) + video_metadatas = _find_metadata_with_filename_existed_in( + (m for m in metadatas if isinstance(m, types.VideoMetadata)), + utils.find_videos(import_paths, skip_subfolders=skip_subfolders), + ) + zip_paths = utils.find_zipfiles(import_paths, skip_subfolders=skip_subfolders) + return _UploadCandidates(image_metadatas, video_metadatas, zip_paths) + + +def _gen_upload_everything( + mly_uploader: uploader.Uploader, + metadatas: T.Sequence[types.Metadata], + import_paths: T.Sequence[Path], + skip_subfolders: bool, +): + candidates = _find_upload_candidates( + metadatas, import_paths, skip_subfolders=skip_subfolders + ) + + # Upload images image_uploader = uploader.ImageSequenceUploader( mly_uploader.upload_options, emitter=mly_uploader.emitter ) - yield from image_uploader.upload_images(image_metadatas) + yield from image_uploader.upload_images(candidates.image_metadatas) # Upload videos - video_metadatas = _find_metadata_with_filename_existed_in( - (m for m in metadatas if isinstance(m, types.VideoMetadata)), - utils.find_videos(import_paths, skip_subfolders=skip_subfolders), + yield from uploader.VideoUploader.upload_videos( + mly_uploader, candidates.video_metadatas ) - yield from uploader.VideoUploader.upload_videos(mly_uploader, video_metadatas) # Upload zip files - zip_paths = utils.find_zipfiles(import_paths, skip_subfolders=skip_subfolders) - yield from uploader.ZipUploader.upload_zipfiles(mly_uploader, zip_paths) + yield from uploader.ZipUploader.upload_zipfiles(mly_uploader, candidates.zip_paths) def _normalize_import_paths(import_path: Path | T.Sequence[Path]) -> list[Path]: diff --git a/tests/integration/fixtures.py b/tests/integration/fixtures.py index 80f1ec20..413639ab 100644 --- a/tests/integration/fixtures.py +++ b/tests/integration/fixtures.py @@ -392,7 +392,7 @@ def assert_descs_exact_equal(left: list[dict], right: list[dict]): def run_command(params: list[str], command: str, **kwargs): - subprocess.run( + return subprocess.run( [*shlex.split(EXECUTABLE), "--verbose", command, *params], check=True, **kwargs ) @@ -447,6 +447,16 @@ def run_upload(params: list[str], **kwargs): ) +def run_check_upload_history(params: list[str]): + result = run_command( + params, + command="check_upload_history", + capture_output=True, + text=True, + ) + return json.loads(result.stdout) + + def pytest_skip_if_not_ffmpeg_installed(): if not IS_FFMPEG_INSTALLED: pytest.skip("ffmpeg is not installed, skipping the test") diff --git a/tests/integration/test_history.py b/tests/integration/test_history.py index 1d6b9e92..f30b9e49 100644 --- a/tests/integration/test_history.py +++ b/tests/integration/test_history.py @@ -3,17 +3,111 @@ # This source code is licensed under the BSD license found in the # LICENSE file in the root directory of this source tree. +import json +import os +from pathlib import Path + import py.path import pytest from .fixtures import ( + run_check_upload_history, run_process_and_upload_for_descs, + run_process_for_descs, setup_config, setup_data, setup_upload, ) +@pytest.mark.usefixtures("setup_config") +def test_check_upload_history_is_read_only_and_matches_upload( + setup_data: py.path.local, setup_upload: py.path.local +): + process_args = [ + "--cutoff_time", + "1000", + "--cutoff_distance", + "1000", + str(setup_data), + ] + descs = run_process_for_descs(process_args) + desc_path = setup_data.join("check-upload-history-description.json") + desc_path.write(json.dumps(descs)) + command_args = [ + str(setup_data), + "--desc_path", + str(desc_path), + ] + + results = run_check_upload_history(command_args) + assert results + assert all(not result["already_uploaded"] for result in results) + assert not Path(os.environ["MAPILLARY_UPLOAD_HISTORY_PATH"]).exists() + + run_process_and_upload_for_descs(process_args) + + results = run_check_upload_history(command_args) + assert all(result["already_uploaded"] for result in results) + assert all( + result["already_uploaded_filenames"] == result["filenames"] + for result in results + ) + + image_descs_by_sequence = {} + for desc in descs: + if desc.get("filetype") == "image" and "error" not in desc: + image_descs_by_sequence.setdefault(desc["MAPSequenceUUID"], []).append(desc) + uploaded_sequence = next( + sequence for sequence in image_descs_by_sequence.values() if 1 < len(sequence) + ) + + # A smaller current sequence has a different sequence checksum, but each + # image is still identifiable in the stored descriptions of the old upload. + subset_descs = uploaded_sequence[:1] + subset_desc_path = setup_data.join("uploaded-subset-description.json") + subset_desc_path.write(json.dumps(subset_descs)) + subset_results = run_check_upload_history( + [ + *(desc["filename"] for desc in subset_descs), + "--desc_path", + str(subset_desc_path), + ] + ) + assert len(subset_results) == 1 + assert subset_results[0]["sequence_md5sum"] not in { + result["sequence_md5sum"] for result in results + } + assert subset_results[0]["already_uploaded_filenames"] == [ + desc["filename"] for desc in subset_descs + ] + assert subset_results[0]["already_uploaded"] is True + + new_image_path = Path(str(setup_data)).joinpath("not-uploaded.jpg") + new_image_path.write_bytes(b"not uploaded") + new_desc = { + **subset_descs[0], + "filename": str(new_image_path), + "md5sum": None, + "MAPCaptureTime": "2030_01_01_00_00_00_000", + } + mixed_descs = [subset_descs[0], new_desc] + mixed_desc_path = setup_data.join("mixed-subset-description.json") + mixed_desc_path.write(json.dumps(mixed_descs)) + mixed_results = run_check_upload_history( + [ + *(desc["filename"] for desc in mixed_descs), + "--desc_path", + str(mixed_desc_path), + ] + ) + assert len(mixed_results) == 1 + assert mixed_results[0]["already_uploaded_filenames"] == [ + subset_descs[0]["filename"] + ] + assert mixed_results[0]["already_uploaded"] is False + + @pytest.mark.usefixtures("setup_config") def test_upload_everything(setup_data: py.path.local, setup_upload: py.path.local): assert len(setup_upload.listdir()) == 0 diff --git a/tests/integration/test_process_report.py b/tests/integration/test_process_report.py new file mode 100644 index 00000000..437fefe9 --- /dev/null +++ b/tests/integration/test_process_report.py @@ -0,0 +1,114 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# +# This source code is licensed under the BSD license found in the +# LICENSE file in the root directory of this source tree. + +import json +from pathlib import Path + +import py.path + +from .fixtures import run_command + + +def _run_process_with_report(import_paths: list[Path], output_dir: Path): + desc_path = output_dir / "description.json" + report_path = output_dir / "process-report.json" + run_command( + [ + "--skip_process_errors", + "--desc_path", + str(desc_path), + "--process_report_path", + str(report_path), + *(str(path) for path in import_paths), + ], + command="process", + ) + return ( + json.loads(desc_path.read_text(encoding="utf-8")), + json.loads(report_path.read_text(encoding="utf-8")), + ) + + +def test_process_report_counts_recognized_errors_and_explicit_unsupported( + tmpdir: py.path.local, +): + root = Path(str(tmpdir)) + broken_image = root / "broken.JPG" + unsupported = root / "unsupported.WEBP" + broken_image.write_bytes(b"not an image") + unsupported.write_bytes(b"not supported") + + descs, report = _run_process_with_report([broken_image, unsupported], root) + + assert len(descs) == 1 + assert descs[0]["filename"] == str(broken_image.resolve()) + assert "error" in descs[0] + assert report == { + "schema_version": 1, + "discovered_file_count": 2, + "processed_file_count": 1, + "skipped_file_count": 1, + "skipped_files": [ + { + "filename": str(unsupported.resolve()), + "category": "unsupported", + "reason_code": "unsupported_format", + "details": {"extension": ".webp"}, + } + ], + } + + +def test_process_report_filters_folder_sidecars_but_reports_explicit_sidecars( + tmpdir: py.path.local, +): + root = Path(str(tmpdir)) + broken_image = root / "broken.jpg" + unsupported = root / "unsupported.webp" + sidecar = root / "track.GPX" + broken_image.write_bytes(b"not an image") + unsupported.write_bytes(b"not supported") + sidecar.write_text("metadata", encoding="utf-8") + (root / "Thumbs.db").write_bytes(b"system") + (root / ".hidden.webp").write_bytes(b"hidden") + macosx = root / "__MACOSX" + macosx.mkdir() + (macosx / "resource.webp").write_bytes(b"system") + + _, folder_report = _run_process_with_report([root], root) + assert folder_report["discovered_file_count"] == 2 + assert folder_report["processed_file_count"] == 1 + assert folder_report["skipped_file_count"] == 1 + assert folder_report["skipped_files"][0]["filename"] == str(unsupported.resolve()) + + explicit_output = root / "explicit" + explicit_output.mkdir() + descs, explicit_report = _run_process_with_report([sidecar], explicit_output) + assert descs == [] + assert explicit_report["discovered_file_count"] == 1 + assert explicit_report["processed_file_count"] == 0 + assert explicit_report["skipped_file_count"] == 1 + assert explicit_report["skipped_files"][0]["filename"] == str(sidecar.resolve()) + assert explicit_report["skipped_files"][0]["details"] == {"extension": ".gpx"} + + +def test_process_without_report_preserves_unsupported_omission(tmpdir: py.path.local): + root = Path(str(tmpdir)) + unsupported = root / "unsupported.webp" + desc_path = root / "description.json" + unsupported.write_bytes(b"not supported") + + run_command( + [ + "--skip_process_errors", + "--desc_path", + str(desc_path), + str(unsupported), + ], + command="process", + ) + + assert json.loads(desc_path.read_text(encoding="utf-8")) == [] + assert not (root / "process-report.json").exists() diff --git a/tests/unit/test_check_upload_history.py b/tests/unit/test_check_upload_history.py new file mode 100644 index 00000000..60613c7f --- /dev/null +++ b/tests/unit/test_check_upload_history.py @@ -0,0 +1,365 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# +# This source code is licensed under the BSD license found in the +# LICENSE file in the root directory of this source tree. + +import hashlib +import json +import sys +import zipfile +from pathlib import Path + +from mapillary_tools import constants, history, types, upload +from mapillary_tools.commands import __main__ as commands +from mapillary_tools.serializer.description import DescriptionJSONSerializer + + +def _image(filename: Path, sequence_uuid: str, capture_time: float): + return types.ImageMetadata( + filename=filename, + lat=1.0, + lon=2.0, + alt=None, + angle=None, + time=capture_time, + MAPSequenceUUID=sequence_uuid, + ) + + +def test_check_upload_history_for_images_and_video(tmp_path, monkeypatch): + history_path = tmp_path / "history" + monkeypatch.setattr(constants, "MAPILLARY_UPLOAD_HISTORY_PATH", str(history_path)) + + first_image_path = tmp_path / "first.jpg" + second_image_path = tmp_path / "second.jpg" + video_path = tmp_path / "video.mp4" + first_image_path.write_bytes(b"first image") + second_image_path.write_bytes(b"second image") + video_path.write_bytes(b"video") + + metadatas = [ + _image(second_image_path, "sequence", 2.0), + _image(first_image_path, "sequence", 1.0), + types.VideoMetadata( + filename=video_path, + filetype=types.FileType.CAMM, + points=[], + ), + ] + + results = upload.check_upload_history( + [first_image_path, second_image_path, video_path], + _metadatas_from_process=metadatas, + ) + + first_md5 = hashlib.md5(first_image_path.read_bytes()).hexdigest() + second_md5 = hashlib.md5(second_image_path.read_bytes()).hexdigest() + expected_sequence_md5 = hashlib.md5(f"{first_md5}{second_md5}".encode()).hexdigest() + expected_video_md5 = hashlib.md5(video_path.read_bytes()).hexdigest() + + assert results == [ + { + "file_type": "image", + "sequence_uuid": "sequence", + "sequence_md5sum": expected_sequence_md5, + "filenames": [str(first_image_path), str(second_image_path)], + "already_uploaded_filenames": [], + "already_uploaded": False, + }, + { + "file_type": "camm", + "sequence_uuid": None, + "sequence_md5sum": expected_video_md5, + "filenames": [str(video_path)], + "already_uploaded_filenames": [], + "already_uploaded": False, + }, + ] + + history.write_history( + expected_sequence_md5, + {"version": "test"}, + {"upload_end_time": 123.0}, + ) + history_record_path = history.history_desc_path(expected_sequence_md5) + history_record = history_record_path.read_bytes() + + results = upload.check_upload_history( + [first_image_path, second_image_path, video_path], + _metadatas_from_process=metadatas, + ) + + assert results[0]["already_uploaded"] is True + assert results[0]["already_uploaded_filenames"] == [ + str(first_image_path), + str(second_image_path), + ] + assert results[1]["already_uploaded"] is False + assert results[1]["already_uploaded_filenames"] == [] + assert history_record_path.read_bytes() == history_record + + history.write_history( + expected_video_md5, + {"version": "test"}, + {"upload_end_time": 456.0}, + ) + results = upload.check_upload_history( + [first_image_path, second_image_path, video_path], + _metadatas_from_process=metadatas, + ) + + assert results[0]["already_uploaded"] is True + assert results[1]["already_uploaded"] is True + assert results[1]["already_uploaded_filenames"] == [str(video_path)] + + +def test_check_upload_history_for_zip(tmp_path, monkeypatch): + history_path = tmp_path / "history" + monkeypatch.setattr(constants, "MAPILLARY_UPLOAD_HISTORY_PATH", str(history_path)) + + zip_path = tmp_path / "sequence.zip" + sequence_md5sum = "1" * 32 + with zipfile.ZipFile(zip_path, "w") as zip_file: + zip_file.comment = json.dumps({"sequence_md5sum": sequence_md5sum}).encode() + + results = upload.check_upload_history( + [zip_path], + _metadatas_from_process=[], + ) + assert results == [ + { + "file_type": "zip", + "sequence_uuid": None, + "sequence_md5sum": sequence_md5sum, + "filenames": [str(zip_path)], + "already_uploaded_filenames": [], + "already_uploaded": False, + } + ] + + history.write_history( + sequence_md5sum, + {"version": "test"}, + {"upload_end_time": 123.0}, + ) + history_record_path = history.history_desc_path(sequence_md5sum) + history_record = history_record_path.read_bytes() + + results = upload.check_upload_history( + [zip_path], + _metadatas_from_process=[], + ) + assert results[0]["already_uploaded_filenames"] == [str(zip_path)] + assert results[0]["already_uploaded"] is True + assert history_record_path.read_bytes() == history_record + + +def test_check_upload_history_matches_subset_of_uploaded_image_sequence( + tmp_path, monkeypatch +): + history_path = tmp_path / "history" + monkeypatch.setattr(constants, "MAPILLARY_UPLOAD_HISTORY_PATH", str(history_path)) + + first_image_path = tmp_path / "first.jpg" + second_image_path = tmp_path / "second.jpg" + first_image_path.write_bytes(b"first image") + second_image_path.write_bytes(b"second image") + current_sequence = [ + _image(first_image_path, "current-sequence", 1.0), + _image(second_image_path, "current-sequence", 2.0), + ] + types.update_sequence_md5sum(current_sequence) + + old_extra = _image(tmp_path / "old-extra.jpg", "old-sequence", 0.0) + old_extra.md5sum = hashlib.md5(b"old extra").hexdigest() + uploaded_sequence = [old_extra, *current_sequence] + history.write_history( + "1" * 32, + {"version": "test"}, + {"upload_end_time": 123.0}, + uploaded_sequence, + ) + + results = upload.check_upload_history( + [first_image_path, second_image_path], + _metadatas_from_process=current_sequence, + ) + + assert len(results) == 1 + assert results[0]["sequence_md5sum"] != "1" * 32 + assert results[0]["already_uploaded_filenames"] == [ + str(first_image_path), + str(second_image_path), + ] + assert results[0]["already_uploaded"] is True + + +def test_check_upload_history_reports_mixed_uploaded_image_subset( + tmp_path, monkeypatch +): + history_path = tmp_path / "history" + monkeypatch.setattr(constants, "MAPILLARY_UPLOAD_HISTORY_PATH", str(history_path)) + + first_image_path = tmp_path / "first.jpg" + second_image_path = tmp_path / "second.jpg" + first_image_path.write_bytes(b"first image") + second_image_path.write_bytes(b"second image") + current_sequence = [ + _image(first_image_path, "current-sequence", 1.0), + _image(second_image_path, "current-sequence", 2.0), + ] + types.update_sequence_md5sum(current_sequence) + + # The matching video checksum must not be mistaken for an uploaded image. + matching_video = types.VideoMetadata( + filename=tmp_path / "old-video.mp4", + filetype=types.FileType.CAMM, + points=[], + md5sum=current_sequence[1].md5sum, + ) + history.write_history( + "2" * 32, + {"version": "test"}, + {"upload_end_time": 123.0}, + [current_sequence[0], matching_video], + ) + + results = upload.check_upload_history( + [first_image_path, second_image_path], + _metadatas_from_process=current_sequence, + ) + + assert results[0]["already_uploaded_filenames"] == [str(first_image_path)] + assert results[0]["already_uploaded"] is False + + +def test_check_upload_history_skips_malformed_records_and_preserves_file_order( + tmp_path, monkeypatch +): + history_path = tmp_path / "history" + monkeypatch.setattr(constants, "MAPILLARY_UPLOAD_HISTORY_PATH", str(history_path)) + + image_paths = [tmp_path / f"image-{index}.jpg" for index in range(3)] + for index, image_path in enumerate(image_paths): + image_path.write_bytes(f"image {index}".encode()) + current_sequence = [ + _image(image_path, "current-sequence", float(index)) + for index, image_path in enumerate(image_paths) + ] + types.update_sequence_md5sum(current_sequence) + + malformed_records = { + history_path / "00" / "invalid-json.json": b"{", + history_path / "01" / "invalid-record.json": b"[]", + history_path / "02" / "invalid-descriptions.json": json.dumps( + { + "descs": [ + None, + {"filetype": "image", "md5sum": 123}, + {"filetype": "image", "md5sum": "not-an-md5"}, + ] + } + ).encode(), + } + for record_path, contents in malformed_records.items(): + record_path.parent.mkdir(parents=True, exist_ok=True) + record_path.write_bytes(contents) + + valid_record_path = history_path / "03" / "valid.json" + valid_record_path.parent.mkdir(parents=True) + valid_record_path.write_text( + json.dumps( + { + "descs": [ + { + "filetype": "image", + "md5sum": current_sequence[2].md5sum.upper(), + }, + { + "filetype": "video", + "md5sum": current_sequence[1].md5sum, + }, + { + "filetype": "image", + "md5sum": current_sequence[0].md5sum.upper(), + }, + ] + } + ), + encoding="utf-8", + ) + history_snapshot = { + path: path.read_bytes() for path in history_path.glob("*/*.json") + } + + results = upload.check_upload_history( + image_paths, + _metadatas_from_process=current_sequence, + ) + + assert len(results) == 1 + assert results[0]["already_uploaded_filenames"] == [ + str(image_paths[0]), + str(image_paths[2]), + ] + assert results[0]["already_uploaded"] is False + assert { + path: path.read_bytes() for path in history_path.glob("*/*.json") + } == history_snapshot + + +def test_check_upload_history_treats_corrupt_record_as_not_uploaded( + tmp_path, monkeypatch +): + history_path = tmp_path / "history" + monkeypatch.setattr(constants, "MAPILLARY_UPLOAD_HISTORY_PATH", str(history_path)) + + image_path = tmp_path / "image.jpg" + image_path.write_bytes(b"image") + metadata = _image(image_path, "sequence", 1.0) + sequence_md5sum = types.update_sequence_md5sum([metadata]) + history_record_path = history.history_desc_path(sequence_md5sum) + history_record_path.parent.mkdir(parents=True) + history_record_path.write_bytes(b"\xff") + + results = upload.check_upload_history( + [image_path], + _metadatas_from_process=[metadata], + ) + + assert results[0]["already_uploaded"] is False + assert results[0]["already_uploaded_filenames"] == [] + + +def test_check_upload_history_command_outputs_json_without_authentication( + tmp_path, monkeypatch, capsys +): + monkeypatch.setattr( + constants, "MAPILLARY_UPLOAD_HISTORY_PATH", str(tmp_path / "history") + ) + + image_path = tmp_path / "image.jpg" + image_path.write_bytes(b"image") + metadata = _image(image_path, "sequence", 1.0) + desc_path = tmp_path / "description.json" + desc_path.write_bytes(DescriptionJSONSerializer.serialize([metadata])) + + monkeypatch.setattr( + sys, + "argv", + [ + "mapillary_tools", + "check_upload_history", + str(image_path), + "--desc_path", + str(desc_path), + ], + ) + commands.main() + + results = json.loads(capsys.readouterr().out) + assert len(results) == 1 + assert results[0]["sequence_uuid"] == "sequence" + assert results[0]["filenames"] == [str(image_path)] + assert results[0]["already_uploaded_filenames"] == [] + assert results[0]["already_uploaded"] is False diff --git a/tests/unit/test_process_report.py b/tests/unit/test_process_report.py new file mode 100644 index 00000000..8897a4e8 --- /dev/null +++ b/tests/unit/test_process_report.py @@ -0,0 +1,123 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# +# This source code is licensed under the BSD license found in the +# LICENSE file in the root directory of this source tree. + +import json +from pathlib import Path + +from mapillary_tools import process_report + + +def _touch(path: Path, content: bytes = b"test") -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) + return path + + +def test_build_process_report_for_directory(tmp_path: Path): + _touch(tmp_path / "capture.JPG") + _touch(tmp_path / "clip.MP4") + unsupported_paths = [ + _touch(tmp_path / "document.PDF"), + _touch(tmp_path / "map.WEBP"), + _touch(tmp_path / "nested" / "地图.HEIC"), + ] + + # Folder discovery deliberately ignores hidden/system files and common + # capture metadata or GPS sidecars. + for path in [ + tmp_path / ".DS_Store", + tmp_path / ".hidden.webp", + tmp_path / "Thumbs.db", + tmp_path / "track.GPX", + tmp_path / "metadata.JSON", + tmp_path / "metadata.XML", + tmp_path / "video.LRV", + tmp_path / "archive.ZIP", + tmp_path / ".hidden" / "secret.webp", + tmp_path / "__MACOSX" / "resource.webp", + ]: + _touch(path) + + report = process_report.build_process_report(tmp_path) + + assert report == { + "schema_version": 1, + "discovered_file_count": 5, + "processed_file_count": 2, + "skipped_file_count": 3, + "skipped_files": [ + { + "filename": str(path.resolve()), + "category": "unsupported", + "reason_code": "unsupported_format", + "details": {"extension": path.suffix.lower()}, + } + for path in sorted(unsupported_paths, key=lambda path: str(path.resolve())) + ], + } + + +def test_explicit_files_are_reported_even_when_folder_discovery_ignores_them( + tmp_path: Path, +): + track_path = _touch(tmp_path / "track.GPX") + system_path = _touch(tmp_path / "Thumbs.db") + hidden_path = _touch(tmp_path / ".unsupported") + supported_path = _touch(tmp_path / "capture.JPEG") + + report = process_report.build_process_report( + [tmp_path, track_path, system_path, hidden_path, supported_path] + ) + + assert report["discovered_file_count"] == 4 + assert report["processed_file_count"] == 1 + assert report["skipped_file_count"] == 3 + assert [item["filename"] for item in report["skipped_files"]] == sorted( + [ + str(track_path.resolve()), + str(system_path.resolve()), + str(hidden_path.resolve()), + ] + ) + assert [item["details"]["extension"] for item in report["skipped_files"]] == [ + Path(filename).suffix.lower() + for filename in sorted( + [ + str(track_path.resolve()), + str(system_path.resolve()), + str(hidden_path.resolve()), + ] + ) + ] + + +def test_skip_subfolders_and_deterministic_utf8_output(tmp_path: Path): + unsupported_path = _touch(tmp_path / "é.Unsupported") + _touch(tmp_path / "nested" / "nested.webp") + + report = process_report.build_process_report(tmp_path, skip_subfolders=True) + first_path = tmp_path / "report-first.json" + second_path = tmp_path / "report-second.json" + process_report.write_process_report(first_path, report) + process_report.write_process_report(second_path, report) + + assert report["discovered_file_count"] == 1 + assert report["processed_file_count"] == 0 + assert report["skipped_file_count"] == 1 + assert report["skipped_files"][0]["filename"] == str(unsupported_path.resolve()) + assert report["skipped_files"][0]["details"] == {"extension": ".unsupported"} + assert first_path.read_bytes() == second_path.read_bytes() + assert "é" in first_path.read_text(encoding="utf-8") + assert json.loads(first_path.read_text(encoding="utf-8")) == report + + +def test_empty_process_report(tmp_path: Path): + assert process_report.build_process_report(tmp_path) == { + "schema_version": 1, + "discovered_file_count": 0, + "processed_file_count": 0, + "skipped_file_count": 0, + "skipped_files": [], + }