Fix source_dir in FrameworkProcessor

[zhaoqizqwang]

fix(processing): Support S3 URIs for FrameworkProcessor source_dir

Restore v2 behavior where source_dir accepts S3 URIs pointing to
tar.gz archives or S3 prefixes. The v3 rewrite only handled local
directories, causing ValueError when customers passed S3 paths.

Add _is_s3_uri, _resolve_s3_source_dir, and _resolve_helper_scripts_prefix
helpers to cleanly route S3 vs local source_dir through _package_code and
_pack_and_upload_code. Update _generate_custom_framework_script to handle
the case where entry_point cannot be read locally.

[zhaoqizqwang]

Tested with real use case

def _upload_source_dir_to_s3(session, local_source_dir, s3_prefix):
    """Package local source_dir as sourcedir.tar.gz and upload to S3.

    This simulates what a customer would have pre-staged in S3.
    """
    bucket = session.default_bucket()
    s3_key = f"{s3_prefix}/sourcedir.tar.gz"

    with tempfile.NamedTemporaryFile(suffix=".tar.gz", delete=False) as tmp:
        with tarfile.open(tmp.name, "w:gz") as tar:
            for item in os.listdir(local_source_dir):
                item_path = os.path.join(local_source_dir, item)
                tar.add(item_path, arcname=item)

        s3_client = session.boto_session.client("s3")
        s3_client.upload_file(tmp.name, bucket, s3_key)
        os.unlink(tmp.name)

    return f"s3://{bucket}/{s3_key}"


def test_framework_processor_with_s3_source_dir(sagemaker_session, role):
    """FrameworkProcessor.run() with S3 source_dir should complete successfully.

    This test:
    1. Packages a local source dir as sourcedir.tar.gz
    2. Uploads it to S3 (simulating a pre-staged archive)
    3. Runs FrameworkProcessor with the S3 URI as source_dir
    4. Verifies the processing job completes successfully
    """
    region = sagemaker_session.boto_region_name
    bucket = sagemaker_session.default_bucket()
    s3_prefix = "integ-test-s3-source-dir"
    processing_job_name = "s3-srcdir-{}".format(strftime("%d-%H-%M-%S", gmtime()))
    output_destination = f"s3://{bucket}/{s3_prefix}/output"

    # Path to our test source code
    local_source_dir = os.path.join(
        os.path.dirname(__file__), "code", "s3_source_dir_processing"
    )

    try:
        # Upload source_dir to S3 as a tar.gz archive
        s3_source_dir_uri = _upload_source_dir_to_s3(
            sagemaker_session, local_source_dir, f"{s3_prefix}/code"
        )

        # Use a Python image (no framework-specific dependencies needed)
        image_uri = get_training_image_uri(
            region=region,
            framework="pytorch",
            framework_version="1.13",
            py_version="py39",
            instance_type="ml.m5.xlarge",
        )

        processor = FrameworkProcessor(
            image_uri=image_uri,
            role=role,
            instance_type="ml.m5.xlarge",
            instance_count=1,
            sagemaker_session=sagemaker_session,
        )

        # This is the critical call — source_dir is an S3 URI.
        # Before the fix, this would raise:
        #   ValueError: source_dir does not exist: /...mangled-path.../s3:/bucket/...
        processor.run(
            code="process.py",
            source_dir=s3_source_dir_uri,
            job_name=processing_job_name,
            outputs=[
                ProcessingOutput(
                    output_name="output",
                    s3_output=ProcessingS3Output(
                        s3_uri=output_destination,
                        local_path="/opt/ml/processing/output",
                        s3_upload_mode="EndOfJob",
                    ),
                ),
            ],
            wait=False,
        )

        # Wait for job to complete (10 min timeout)
        job = processor.latest_job
        timeout = 600
        start_time = time.time()

        while time.time() - start_time < timeout:
            job.refresh()
            status = job.processing_job_status

            if status == "Completed":
                break
            elif status in ["Failed", "Stopped"]:
                pytest.fail(
                    f"Processing job {status}. "
                    f"Job name: {processing_job_name}"
                )
            time.sleep(30)
        else:
            pytest.fail(f"Processing job timed out after {timeout} seconds")

        assert status == "Completed"

    finally:
        # Cleanup S3 resources
        s3 = boto3.resource("s3")
        bucket_obj = s3.Bucket(bucket)
        bucket_obj.objects.filter(Prefix=f"{s3_prefix}/").delete()