def METHOD_NAME(self, data, parent_data):
    executor = parent_data.get_one_of_inputs("executor")
    client = get_client_by_user(executor)
    client.set_bk_api_ver("v2")
    if parent_data.get_one_of_inputs("language"):
        setattr(client, "language", parent_data.get_one_of_inputs("language"))
        translation.activate(parent_data.get_one_of_inputs("language"))
    biz_cc_id = data.get_one_of_inputs("biz_cc_id", parent_data.inputs.biz_cc_id)
    original_source_files = deepcopy(data.get_one_of_inputs("job_source_files", []))
    across_biz = data.get_one_of_inputs("job_across_biz", False)
    ip_is_exist = data.get_one_of_inputs("ip_is_exist")
    file_source = []
    for item in original_source_files:
        # filter 跨业务 IP
        clean_source_ip_result, server = self.get_target_server(
            executor, biz_cc_id, data, item["ip"], self.logger, False, is_across=across_biz
        )
        if not clean_source_ip_result:
            return False
        file_source.append(
            {
                "file_list": [_file.strip() for _file in item["files"].split("\n") if _file.strip()],
                "server": server,
                "account": {"alias": loose_strip(item["account"])},
            }
        )
    # 获取目标IP
    original_ip_list = data.get_one_of_inputs("job_ip_list")
    clean_result, target_server = self.get_target_server(
        executor,
        biz_cc_id,
        data,
        original_ip_list,
        logger_handle=self.logger,
        ip_is_exist=ip_is_exist,
        is_across=across_biz,
    )
    if not clean_result:
        return False
    job_timeout = data.get_one_of_inputs("job_timeout")
    job_kwargs = {
        "bk_scope_type": JobBizScopeType.BIZ.value,
        "bk_scope_id": str(biz_cc_id),
        "bk_biz_id": biz_cc_id,
        "file_source_list": file_source,
        "target_server": target_server,
        "account_alias": data.get_one_of_inputs("job_account"),
        "file_target_path": data.get_one_of_inputs("job_target_path"),
        "callback_url": get_node_callback_url(self.root_pipeline_id, self.id, getattr(self, "version", "")),
    }
    if job_timeout:
        job_kwargs["timeout"] = int(job_timeout)
    job_result = client.jobv3.fast_transfer_file(job_kwargs)
    self.logger.info("job_result: {result}, job_kwargs: {kwargs}".format(result=job_result, kwargs=job_kwargs))
    if job_result["result"]:
        job_instance_id = job_result["data"]["job_instance_id"]
        data.outputs.job_inst_id = job_instance_id
        data.outputs.job_inst_name = job_result["data"]["job_instance_name"]
        data.outputs.job_inst_url = get_job_instance_url(biz_cc_id, job_instance_id)
        data.outputs.client = client
        return True
    else:
        message = job_handle_api_error("jobv3.fast_transfer_file", job_kwargs, job_result)
        self.logger.error(message)
        data.outputs.ex_data = message
        return False