שליחת חברי הקהל

{
  "operatingAccount": {
    "accountType": "OPERATING_ACCOUNT_TYPE",
    "accountId": "OPERATING_ACCOUNT_ID"
  },
  "loginAccount": {
    "accountType": "LOGIN_ACCOUNT_TYPE",
    "accountId": "LOGIN_ACCOUNT_ID"
  },
  "productDestinationId": "AUDIENCE_ID"
}

{
  "operatingAccount": {
    "accountType": "OPERATING_ACCOUNT_TYPE",
    "accountId": "OPERATING_ACCOUNT_ID"
  },
  "loginAccount": {
    "accountType": "DATA_PARTNER",
    "accountId": "DATA_PARTNER_ACCOUNT_ID"
  },
  "linkedAccount": {
    "accountType": "LINKED_ACCOUNT_TYPE",
    "accountId": "LINKED_ACCOUNT_ID"
  },
  "productDestinationId": "AUDIENCE_ID"
}

// Copyright 2025 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

using System.Text.Json;
using CommandLine;
using Google.Ads.DataManager.Util;
using Google.Ads.DataManager.V1;
using Google.Api.Gax.Grpc;
using Google.Protobuf.WellKnownTypes;
using static Google.Ads.DataManager.V1.ProductAccount.Types;

namespace Google.Ads.DataManager.Samples
{
    // <summary>
    // Sends a <see cref="IngestAudienceMembersRequest" /> without using encryption.
    //
    // User data is read from a data file. See the <c>audience_members_1.json</c> file in the
    // <c>sampledata</c> directory for an example.
    // </summary>
    public class IngestAudienceMembers
    {
        private static readonly int MaxMembersPerRequest = 10_000;

        [Verb(
            "ingest-audience-members",
            HelpText = "Sends an IngestAudienceMembersRequest without using encryption."
        )]
        public class Options
        {
            [Option(
                "operatingAccountType",
                Required = true,
                HelpText = "Account type of the operating account"
            )]
            public AccountType OperatingAccountType { get; set; }

            [Option(
                "operatingAccountId",
                Required = true,
                HelpText = "ID of the operating account"
            )]
            public string OperatingAccountId { get; set; } = null!;

            [Option(
                "loginAccountType",
                Required = false,
                HelpText = "Account type of the login account"
            )]
            public AccountType? LoginAccountType { get; set; }

            [Option("loginAccountId", Required = false, HelpText = "ID of the login account")]
            public string? LoginAccountId { get; set; }

            [Option(
                "linkedAccountType",
                Required = false,
                HelpText = "Account type of the linked account"
            )]
            public AccountType? LinkedAccountType { get; set; }

            [Option("linkedAccountId", Required = false, HelpText = "ID of the linked account")]
            public string? LinkedAccountId { get; set; }

            [Option("audienceId", Required = true, HelpText = "ID of the audience")]
            public string AudienceId { get; set; } = null!;

            [Option(
                "jsonFile",
                Required = true,
                HelpText = "JSON file containing user data to ingest"
            )]
            public string JsonFile { get; set; } = null!;

            [Option(
                "validateOnly",
                Default = true,
                HelpText = "Whether to enable validateOnly on the request"
            )]
            public bool ValidateOnly { get; set; }
        }

        public void Run(Options options)
        {
            RunExample(
                options.OperatingAccountType,
                options.OperatingAccountId,
                options.LoginAccountType,
                options.LoginAccountId,
                options.LinkedAccountType,
                options.LinkedAccountId,
                options.AudienceId,
                options.JsonFile,
                options.ValidateOnly
            );
        }

        // Runs the example.
        private void RunExample(
            AccountType operatingAccountType,
            string operatingAccountId,
            AccountType? loginAccountType,
            string? loginAccountId,
            AccountType? linkedAccountType,
            string? linkedAccountId,
            string audienceId,
            string jsonFile,
            bool validateOnly
        )
        {
            if (loginAccountId == null ^ loginAccountType == null)
            {
                throw new ArgumentException(
                    "Must specify either both or neither of login account ID and login account "
                        + "type"
                );
            }
            if (linkedAccountId == null ^ linkedAccountType == null)
            {
                throw new ArgumentException(
                    "Must specify either both or neither of linked account ID and linked account "
                        + "type"
                );
            }

            // Reads the audience members from the JSON file.
            List<Member> memberList = ReadMemberData(jsonFile);

            // Creates a factory that will be used to generate the appropriate data manager.
            var userDataFormatter = new UserDataFormatter();

            var audienceMembers = new List<AudienceMember>();

            // Processes each batch of audience members.
            foreach (var member in memberList)
            {
                var compositeData = new CompositeData();
                var userDataBuilder = new UserData();

                // Adds a UserIdentifier for each valid email address for the member.
                foreach (var email in member.Emails)
                {
                    try
                    {
                        string processedEmail = userDataFormatter.ProcessEmailAddress(
                            email,
                            UserDataFormatter.Encoding.Hex
                        );
                        // Sets the email address identifier to the encoded hash.
                        userDataBuilder.UserIdentifiers.Add(
                            new UserIdentifier { EmailAddress = processedEmail }
                        );
                    }
                    catch (ArgumentException)
                    {
                        // Skips invalid input.
                        continue;
                    }
                }

                // Adds a UserIdentifier for each valid phone number for the member.
                foreach (var phoneNumber in member.PhoneNumbers)
                {
                    try
                    {
                        string processedPhoneNumber = userDataFormatter.ProcessPhoneNumber(
                            phoneNumber,
                            UserDataFormatter.Encoding.Hex
                        );
                        // Sets the phone number identifier to the encoded hash.
                        userDataBuilder.UserIdentifiers.Add(
                            new UserIdentifier { PhoneNumber = processedPhoneNumber }
                        );
                    }
                    catch (ArgumentException)
                    {
                        // Skips invalid input.
                        continue;
                    }
                }

                // Adds IP address information for each valid entry for the member.
                var ipDatas = new List<IpData>();
                foreach (var ipInfo in member.IpInfos)
                {
                    if (operatingAccountType != AccountType.GoogleAds)
                    {
                        Console.WriteLine(
                            $"Skipping IP address information for operating account type {operatingAccountType}. "
                                + "Sending IP address is only supported for operating account type GOOGLE_ADS."
                        );
                        continue;
                    }

                    string processedIpAddress = (ipInfo.IpAddress ?? "").Trim();
                    if (string.IsNullOrEmpty(processedIpAddress))
                    {
                        Console.WriteLine("Skipping IP address information with no IP address");
                        continue;
                    }

                    var ipData = new IpData { IpAddress = processedIpAddress };

                    string startTimeString = (ipInfo.ObserveStartTime ?? "").Trim();
                    if (!string.IsNullOrEmpty(startTimeString))
                    {
                        try
                        {
                            ipData.ObserveStartTime = Timestamp.FromDateTime(
                                DateTime.Parse(startTimeString).ToUniversalTime()
                            );
                        }
                        catch (FormatException)
                        {
                            Console.WriteLine(
                                $"Ignoring observe start time '{startTimeString}' since it can't be parsed"
                            );
                        }
                    }

                    string endTimeString = (ipInfo.ObserveEndTime ?? "").Trim();
                    if (!string.IsNullOrEmpty(endTimeString))
                    {
                        try
                        {
                            ipData.ObserveEndTime = Timestamp.FromDateTime(
                                DateTime.Parse(endTimeString).ToUniversalTime()
                            );
                        }
                        catch (FormatException)
                        {
                            Console.WriteLine(
                                $"Ignoring observe end time '{endTimeString}' since it can't be parsed"
                            );
                        }
                    }

                    ipDatas.Add(ipData);
                }

                if (userDataBuilder.UserIdentifiers.Any() || ipDatas.Any())
                {
                    if (userDataBuilder.UserIdentifiers.Any())
                    {
                        compositeData.UserData = userDataBuilder;
                    }
                    if (ipDatas.Any())
                    {
                        compositeData.IpData.AddRange(ipDatas);
                    }
                    audienceMembers.Add(new AudienceMember { CompositeData = compositeData });
                }
            }

            // Builds the Destination for the request.
            var destinationBuilder = new Destination
            {
                // The destination account for the data.
                OperatingAccount = new ProductAccount
                {
                    AccountType = operatingAccountType,
                    AccountId = operatingAccountId,
                },
                // The ID of the user list that is being updated.
                ProductDestinationId = audienceId,
            };

            if (loginAccountType.HasValue && loginAccountId != null)
            {
                destinationBuilder.LoginAccount = new ProductAccount
                {
                    AccountType = loginAccountType.Value,
                    AccountId = loginAccountId,
                };
            }

            if (linkedAccountType.HasValue && linkedAccountId != null)
            {
                destinationBuilder.LinkedAccount = new ProductAccount
                {
                    AccountType = linkedAccountType.Value,
                    AccountId = linkedAccountId,
                };
            }

            // Creates a client for the ingestion service. Increases the maximum metadata size to
            // 16MB to prevent RESOURCE_EXHAUSTED errors when responses contain large error
            // details.
            IngestionServiceClient ingestionServiceClient = new IngestionServiceClientBuilder
            {
                GrpcChannelOptions = GrpcChannelOptions.Empty.WithCustomOption(
                    "grpc.max_metadata_size",
                    16 * 1024 * 1024
                ),
            }.Build();

            int requestCount = 0;

            // Batches requests to send up to the maximum number of audience members per request.
            for (var i = 0; i < audienceMembers.Count; i += MaxMembersPerRequest)
            {
                IEnumerable<AudienceMember> membersBatch = audienceMembers
                    .Skip(i)
                    .Take(MaxMembersPerRequest);
                requestCount++;
                // Builds the request.
                var request = new IngestAudienceMembersRequest
                {
                    Destinations = { destinationBuilder },
                    // Adds members from the current batch.
                    AudienceMembers = { membersBatch },
                    Consent = new Consent
                    {
                        AdPersonalization = ConsentStatus.ConsentGranted,
                        AdUserData = ConsentStatus.ConsentGranted,
                    },
                    // Sets validate_only. If true, then the Data Manager API only validates the
                    // request but doesn't apply changes.
                    ValidateOnly = validateOnly,
                    Encoding = V1.Encoding.Hex,
                    TermsOfService = new TermsOfService
                    {
                        CustomerMatchTermsOfServiceStatus = TermsOfServiceStatus.Accepted,
                    },
                };

                // Logs the request.
                Console.WriteLine($"Request #{requestCount}:\n{request}");

                // Sends the data to the Data Manager API.
                IngestAudienceMembersResponse response =
                    ingestionServiceClient.IngestAudienceMembers(request);
                Console.WriteLine($"Response for request #{requestCount}:\n{response}");

                if (response.FieldWarnings.Any())
                {
                    Console.WriteLine(
                        "Request ingested successfully, but field warnings were returned. "
                            + "Review warning details and update your implementation as needed."
                    );
                }
            }
            Console.WriteLine($"# of requests sent: {requestCount}");
        }

        private class Member
        {
            public List<string> Emails { get; set; } = new List<string>();
            public List<string> PhoneNumbers { get; set; } = new List<string>();
            public List<IpInfo> IpInfos { get; set; } = new List<IpInfo>();

            public class IpInfo
            {
                public string? IpAddress { get; set; }
                public string? ObserveStartTime { get; set; }
                public string? ObserveEndTime { get; set; }
            }
        }

        private List<Member> ReadMemberData(string jsonFile)
        {
            string jsonString = File.ReadAllText(jsonFile);
            var options = new JsonSerializerOptions { PropertyNameCaseInsensitive = true };
            return JsonSerializer.Deserialize<List<Member>>(jsonString, options)
                ?? new List<Member>();
        }
    }
}
IngestAudienceMembers.cs

// Copyright 2025 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package com.google.ads.datamanager.samples;

import com.beust.jcommander.Parameter;
import com.google.ads.datamanager.samples.common.BaseParamsConfig;
import com.google.ads.datamanager.util.Encrypter;
import com.google.ads.datamanager.util.UserDataFormatter;
import com.google.ads.datamanager.util.UserDataFormatter.Encoding;
import com.google.ads.datamanager.v1.AudienceMember;
import com.google.ads.datamanager.v1.CompositeData;
import com.google.ads.datamanager.v1.Consent;
import com.google.ads.datamanager.v1.ConsentStatus;
import com.google.ads.datamanager.v1.Destination;
import com.google.ads.datamanager.v1.EncryptionInfo;
import com.google.ads.datamanager.v1.GcpWrappedKeyInfo;
import com.google.ads.datamanager.v1.GcpWrappedKeyInfo.KeyType;
import com.google.ads.datamanager.v1.IngestAudienceMembersRequest;
import com.google.ads.datamanager.v1.IngestAudienceMembersResponse;
import com.google.ads.datamanager.v1.IngestionServiceClient;
import com.google.ads.datamanager.v1.IngestionServiceSettings;
import com.google.ads.datamanager.v1.IpData;
import com.google.ads.datamanager.v1.ProductAccount;
import com.google.ads.datamanager.v1.ProductAccount.AccountType;
import com.google.ads.datamanager.v1.TermsOfService;
import com.google.ads.datamanager.v1.TermsOfServiceStatus;
import com.google.ads.datamanager.v1.UserData;
import com.google.ads.datamanager.v1.UserIdentifier;
import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.reflect.TypeToken;
import com.google.gson.GsonBuilder;
import com.google.protobuf.util.Timestamps;
import java.io.BufferedReader;
import java.io.IOException;
import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.security.GeneralSecurityException;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Sends an {@link IngestAudienceMembersRequest} with the option to use encryption.
 *
 * <p>User data is read from a data file. See the {@code audience_members_1.json} file in the {@code
 * resources/sampledata} directory for a sample file.
 */
public class IngestAudienceMembers {
  private static final Logger LOGGER = Logger.getLogger(IngestAudienceMembers.class.getName());

  /** The maximum number of audience members allowed per request. */
  private static final int MAX_MEMBERS_PER_REQUEST = 10_000;

  private static final class ParamsConfig extends BaseParamsConfig<ParamsConfig> {

    @Parameter(
        names = "--operatingAccountType",
        required = true,
        description = "Account type of the operating account")
    AccountType operatingAccountType;

    @Parameter(
        names = "--operatingAccountId",
        required = true,
        description = "ID of the operating account")
    String operatingAccountId;

    @Parameter(
        names = "--loginAccountType",
        required = false,
        description = "Account type of the login account")
    AccountType loginAccountType;

    @Parameter(
        names = "--loginAccountId",
        required = false,
        description = "ID of the login account")
    String loginAccountId;

    @Parameter(
        names = "--linkedAccountType",
        required = false,
        description = "Account type of the linked account")
    AccountType linkedAccountType;

    @Parameter(
        names = "--linkedAccountId",
        required = false,
        description = "ID of the linked account")
    String linkedAccountId;

    @Parameter(names = "--audienceId", required = true, description = "ID of the audience")
    String audienceId;

    @Parameter(
        names = "--jsonFile",
        required = true,
        description = "JSON file containing user data to ingest")
    String jsonFile;

    @Parameter(
        names = "--keyUri",
        required = false,
        description =
            "URI of the Google Cloud KMS key for encrypting data. If this parameter is set, you"
                + " must also set the --wipProvider parameter.")
    String keyUri;

    @Parameter(
        names = "--wipProvider",
        required = false,
        description =
            "Workload Identity Pool provider name for encrypting data. If this parameter is set,"
                + " you must also set the --keyUri parameter. The argument for this parameter must"
                + " follow the pattern:"
                + " projects/PROJECT_ID/locations/global/workloadIdentityPools/WIP_ID/providers/PROVIDER_ID")
    String wipProvider;

    @Parameter(
        names = "--validateOnly",
        required = false,
        arity = 1,
        description = "Whether to enable validateOnly on the request")
    boolean validateOnly = true;
  }

  public static void main(String[] args) throws IOException, GeneralSecurityException {
    ParamsConfig paramsConfig = new ParamsConfig().parseOrExit(args);
    if ((paramsConfig.loginAccountId == null) != (paramsConfig.loginAccountType == null)) {
      throw new IllegalArgumentException(
          "Must specify either both or neither of login account ID and login account type");
    }
    if ((paramsConfig.linkedAccountId == null) != (paramsConfig.linkedAccountType == null)) {
      throw new IllegalArgumentException(
          "Must specify either both or neither of linked account ID and linked account type");
    }
    if ((paramsConfig.keyUri == null) != (paramsConfig.wipProvider == null)) {
      throw new IllegalArgumentException(
          "Must specify either both or neither of key URI and WIP provider");
    }
    new IngestAudienceMembers().runExample(paramsConfig);
  }

  /**
   * Runs the example.
   *
   * @param params the parameters for the example
   */
  private void runExample(ParamsConfig params) throws IOException, GeneralSecurityException {
    // Reads member data from the data file.
    List<Member> memberList = readMemberData(params.jsonFile);

    // Gets an instance of the UserDataFormatter for normalizing and formatting the data.
    UserDataFormatter userDataFormatter = UserDataFormatter.create();

    // Determines if encryption parameters are set.
    boolean useEncryption = (params.keyUri != null && params.wipProvider != null);
    Encrypter encrypter = null;
    if (useEncryption) {
      // Gets an instance of the encryption utility.
      encrypter = Encrypter.createForGcpKms(params.keyUri, null);
    }
    // Builds the audience_members collection for the request.
    List<AudienceMember> audienceMembers = new ArrayList<>();
    for (Member member : memberList) {
      // Creates a builder for CompositeData to add user data and/or IP address information.
      CompositeData.Builder compositeDataBuilder = CompositeData.newBuilder();

      // Creates a builder for user data for hashed emails and phone numbers.
      UserData.Builder userDataBuilder = compositeDataBuilder.getUserDataBuilder();

      // Adds a UserIdentifier for each valid email address for the member.
      for (String email : member.emails) {
        String processedEmail;
        try {
          processedEmail =
              useEncryption
                  ? userDataFormatter.processEmailAddress(email, Encoding.HEX, encrypter)
                  : userDataFormatter.processEmailAddress(email, Encoding.HEX);
        } catch (IllegalArgumentException iae) {
          // Skips invalid input.
          continue;
        }
        // Sets the email address identifier to the encoded and possibly encrypted email hash.
        userDataBuilder.addUserIdentifiers(
            UserIdentifier.newBuilder().setEmailAddress(processedEmail));
      }

      // Adds a UserIdentifier for each valid phone number for the member.
      for (String phoneNumber : member.phoneNumbers) {
        String processedPhoneNumber;
        try {
          processedPhoneNumber =
              useEncryption
                  ? userDataFormatter.processPhoneNumber(phoneNumber, Encoding.HEX, encrypter)
                  : userDataFormatter.processPhoneNumber(phoneNumber, Encoding.HEX);
        } catch (IllegalArgumentException iae) {
          // Skips invalid input.
          continue;
        }
        // Sets the phone number identifier to the encoded and possibly encrypted phone number
        // hash.
        userDataBuilder.addUserIdentifiers(
            UserIdentifier.newBuilder().setPhoneNumber(processedPhoneNumber));
      }

      // Adds IP address information for each valid entry for the member.
      for (Member.IpInfo ipInfo : member.ipInfos) {
        if (!AccountType.GOOGLE_ADS.equals(params.operatingAccountType)) {
          LOGGER.log(
              Level.INFO,
              "Skipping IP address information for operating account type {0}. "
                  + "Sending IP address is only supported for operating account type {1}.",
              new Object[] {params.operatingAccountType, AccountType.GOOGLE_ADS});
          continue;
        }

        // Trims leading and trailing whitespace from the IP address.
        String processedIpAddress = Strings.nullToEmpty(ipInfo.ipAddress).trim();
        if (processedIpAddress.isEmpty()) {
          // Skips the IP information from the input file since it is missing IP address, which is
          // required.
          LOGGER.info("Skipping IP address information with no IP address");
          continue;
        }

        // Creates an IpData builder and sets its IP address.
        IpData.Builder ipDataBuilder =
            compositeDataBuilder.addIpDataBuilder().setIpAddress(processedIpAddress);

        // Adds observe start time to the IpData if present and in the expected timestamp format.
        String startTimeString = Strings.nullToEmpty(ipInfo.observeStartTime).trim();
        if (!startTimeString.isEmpty()) {
          try {
            ipDataBuilder.setObserveStartTime(Timestamps.parse(startTimeString));
          } catch (ParseException e) {
            LOGGER.log(
                Level.INFO,
                "Ignoring observe start time '{0} since it can't be parsed",
                startTimeString);
          }
        }

        // Adds observe end time to the IpData if present and in the expected timestamp format.
        String endTimeString = Strings.nullToEmpty(ipInfo.observeEndTime).trim();
        if (!endTimeString.isEmpty()) {
          try {
            ipDataBuilder.setObserveEndTime(Timestamps.parse(endTimeString));
          } catch (ParseException e) {
            LOGGER.log(
                Level.INFO,
                "Ignoring observe end time '{0}' since it can't be parsed",
                endTimeString);
          }
        }
      }

      if (userDataBuilder.getUserIdentifiersCount() > 0
          || compositeDataBuilder.getIpDataCount() > 0) {
        // Adds the CompositeData containing user data, IP address information, or both.
        audienceMembers.add(
            AudienceMember.newBuilder().setCompositeData(compositeDataBuilder).build());
      }
    }

    // Builds the Destination for the request.
    Destination.Builder destinationBuilder =
        Destination.newBuilder()
            .setOperatingAccount(
                ProductAccount.newBuilder()
                    .setAccountType(params.operatingAccountType)
                    .setAccountId(params.operatingAccountId))
            .setProductDestinationId(params.audienceId);
    if (params.loginAccountType != null && params.loginAccountId != null) {
      destinationBuilder.setLoginAccount(
          ProductAccount.newBuilder()
              .setAccountType(params.loginAccountType)
              .setAccountId(params.loginAccountId));
    }
    if (params.linkedAccountType != null && params.linkedAccountId != null) {
      destinationBuilder.setLinkedAccount(
          ProductAccount.newBuilder()
              .setAccountType(params.linkedAccountType)
              .setAccountId(params.linkedAccountId));
    }

    // Configures the EncryptionInfo for the request if encryption parameters provided.
    EncryptionInfo encryptionInfo = null;
    if (useEncryption) {
      encryptionInfo =
          EncryptionInfo.newBuilder()
              .setGcpWrappedKeyInfo(
                  GcpWrappedKeyInfo.newBuilder()
                      .setKekUri(params.keyUri)
                      .setWipProvider(params.wipProvider)
                      .setKeyType(KeyType.XCHACHA20_POLY1305)
                      // Sets the encrypted_dek field to the Base64-encoded encrypted DEK.
                      .setEncryptedDek(
                          userDataFormatter.base64Encode(
                              encrypter.getEncryptedDek().toByteArray())))
              .build();
    }

    // Creates a client for the ingestion service. Increases the maximum metadata size to 16MB to
    // prevent RESOURCE_EXHAUSTED errors when responses contain large error details.
    InstantiatingGrpcChannelProvider provider =
        IngestionServiceSettings.defaultGrpcTransportProviderBuilder()
            .setMaxInboundMetadataSize(16 * 1024 * 1024)
            .build();
    IngestionServiceSettings settings =
        IngestionServiceSettings.newBuilder().setTransportChannelProvider(provider).build();
    try (IngestionServiceClient ingestionServiceClient = IngestionServiceClient.create(settings)) {
      int requestCount = 0;
      // Batches requests to send up to the maximum number of audience members per request.
      for (List<AudienceMember> audienceMembersBatch :
          Lists.partition(audienceMembers, MAX_MEMBERS_PER_REQUEST)) {
        requestCount++;
        // Builds the request.
        IngestAudienceMembersRequest.Builder requestBuilder =
            IngestAudienceMembersRequest.newBuilder()
                .addDestinations(destinationBuilder)
                // Adds members from the current batch.
                .addAllAudienceMembers(audienceMembersBatch)
                .setConsent(
                    Consent.newBuilder()
                        .setAdPersonalization(ConsentStatus.CONSENT_GRANTED)
                        .setAdUserData(ConsentStatus.CONSENT_GRANTED))
                // Sets validate_only. If true, then the Data Manager API only validates the request
                // but doesn't apply changes.
                .setValidateOnly(params.validateOnly)
                // Sets encoding to match the encoding used.
                .setEncoding(com.google.ads.datamanager.v1.Encoding.HEX)
                .setTermsOfService(
                    TermsOfService.newBuilder()
                        .setCustomerMatchTermsOfServiceStatus(TermsOfServiceStatus.ACCEPTED));

        if (useEncryption) {
          // Sets encryption info on the request.
          requestBuilder.setEncryptionInfo(encryptionInfo);
        }

        // Builds and logs the request.
        IngestAudienceMembersRequest request = requestBuilder.build();
        if (LOGGER.isLoggable(Level.INFO)) {
          LOGGER.info(String.format("Request #%d:%n%s", requestCount, request));
        }

        // Sends the request.
        IngestAudienceMembersResponse response =
            ingestionServiceClient.ingestAudienceMembers(request);

        // Logs the response.
        if (LOGGER.isLoggable(Level.INFO)) {
          LOGGER.info(String.format("Response for request #%d:%n%s", requestCount, response));
        }

        if (response.getFieldWarningsCount() > 0) {
          LOGGER.warning(
              "Request ingested successfully, but field warnings were returned. Review warning"
                  + " details and update your implementation as needed.");
        }
      }
      LOGGER.info("# of requests sent: " + requestCount);
    }
  }

  /** Data object for a single row of input data. */
  @SuppressWarnings("unused")
  private static class Member {
    private List<String> emails = new ArrayList<>();
    private List<String> phoneNumbers = new ArrayList<>();
    private List<IpInfo> ipInfos = new ArrayList<>();

    private static class IpInfo {
      private String ipAddress;
      private String observeStartTime;
      private String observeEndTime;
    }
  }

  /**
   * Reads the data file and parses it into a list of {@link IngestAudienceMembers.Member} objects.
   *
   * @param jsonFile the JSON data file
   * @return a list of Member objects
   */
  private List<Member> readMemberData(String jsonFile) throws IOException {
    try (BufferedReader jsonReader =
        Files.newBufferedReader(Paths.get(jsonFile), StandardCharsets.UTF_8)) {
      // Define the type for Gson to deserialize into (List of Member objects)
      Type recordListType = new TypeToken<ArrayList<Member>>() {}.getType();

      // Parse the JSON string from the file into a List of Member objects
      return new GsonBuilder().create().fromJson(jsonReader, recordListType);
    }
  }
}
IngestAudienceMembers.java

#!/usr/bin/env node
// Copyright 2025 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

'use strict';

import {IngestionServiceClient} from '@google-ads/datamanager';
import {protos} from '@google-ads/datamanager';
const {
  AudienceMember,
  CompositeData,
  Destination,
  Encoding: DataManagerEncoding,
  Consent,
  ConsentStatus,
  IngestAudienceMembersRequest,
  IpData,
  ProductAccount,
  TermsOfService,
  TermsOfServiceStatus,
  UserData,
  UserIdentifier,
} = protos.google.ads.datamanager.v1;
import {UserDataFormatter, Encoding} from '@google-ads/datamanager-util';
import * as fs from 'fs';
import * as yargs from 'yargs';

const MAX_MEMBERS_PER_REQUEST = 10000;

interface Arguments {
  operating_account_type: string;
  operating_account_id: string;
  audience_id: string;
  json_file: string;
  validate_only: boolean;
  login_account_type?: string;
  login_account_id?: string;
  linked_account_type?: string;
  linked_account_id?: string;
  [x: string]: unknown;
}

interface MemberRow {
  emails?: string[];
  phoneNumbers?: string[];
  ipInfos?: {
    ipAddress: string;
    observeStartTime?: string;
    observeEndTime?: string;
  }[];
}

/**
 * The main function for the IngestAudienceMembers sample.
 */
async function main() {
  const argv: Arguments = yargs
    .option('operating_account_type', {
      describe: 'The account type of the operating account.',
      type: 'string',
      required: true,
    })
    .option('operating_account_id', {
      describe: 'The ID of the operating account.',
      type: 'string',
      required: true,
    })
    .option('audience_id', {
      describe: 'The ID of the destination audience.',
      type: 'string',
      required: true,
    })
    .option('json_file', {
      describe: 'JSON file containing user data to ingest.',
      type: 'string',
      required: true,
    })
    .option('validate_only', {
      describe: 'Whether to enable validate_only on the request.',
      type: 'boolean',
      default: true,
    })
    .option('login_account_type', {
      describe: 'The account type of the login account.',
      type: 'string',
    })
    .option('login_account_id', {
      describe: 'The ID of the login account.',
      type: 'string',
    })
    .option('linked_account_type', {
      describe: 'The account type of the linked account.',
      type: 'string',
    })
    .option('linked_account_id', {
      describe: 'The ID of the linked account.',
      type: 'string',
    })
    .option('config', {
      describe: 'Path to a JSON file with arguments.',
      type: 'string',
    })
    .config('config')
    .check((args: Arguments) => {
      if (
        (args.login_account_type && !args.login_account_id) ||
        (!args.login_account_type && args.login_account_id)
      ) {
        throw new Error(
          'Must specify either both or neither of login account type and ' +
            'login account ID',
        );
      }
      if (
        (args.linked_account_type && !args.linked_account_id) ||
        (!args.linked_account_type && args.linked_account_id)
      ) {
        throw new Error(
          'Must specify either both or neither of linked account ' +
            'type and linked account ID',
        );
      }
      return true;
    })
    .parseSync();

  const formatter = new UserDataFormatter();

  const memberRows: MemberRow[] = await readMemberData(argv.json_file);

  // Converts the operating account type argument to the corresponding enum
  // so IP address checks can compare enums to enums.
  const operatingAccountType = convertToAccountType(
    argv.operating_account_type,
    'operating_account_type',
  );

  // Builds the audience_members collection for the request.
  const audienceMembers = [];
  for (const memberRow of memberRows) {
    const userData = UserData.create();

    // Adds a UserIdentifier for each valid email address for the member.
    for (const email of memberRow.emails || []) {
      try {
        const processedEmail = formatter.processEmailAddress(
          email,
          Encoding.HEX,
        );
        userData.userIdentifiers.push(
          UserIdentifier.create({emailAddress: processedEmail}),
        );
      } catch (e) {
        // Skip invalid input.
      }
    }

    // Adds a UserIdentifier for each valid phone number for the member.
    for (const phone of memberRow.phoneNumbers || []) {
      try {
        const processedPhone = formatter.processPhoneNumber(
          phone,
          Encoding.HEX,
        );
        userData.userIdentifiers.push(
          UserIdentifier.create({phoneNumber: processedPhone}),
        );
      } catch (e) {
        // Skip invalid input.
      }
    }

    // Process IP address information
    const ipDatas = [];
    for (const ipInfo of memberRow.ipInfos || []) {
      if (operatingAccountType !== ProductAccount.AccountType.GOOGLE_ADS) {
        console.log(
          `Skipping IP address information for operating account type ${operatingAccountType}. ` +
            'Sending IP address is only supported for operating account type GOOGLE_ADS.',
        );
        continue;
      }

      const ipAddress = (ipInfo.ipAddress || '').trim();
      if (!ipAddress) {
        console.log('Skipping IP address information with no IP address');
        continue;
      }

      const ipData = IpData.create({ipAddress});

      if (ipInfo.observeStartTime) {
        const startTimeStr = ipInfo.observeStartTime.trim();
        if (startTimeStr) {
          try {
            const date = new Date(startTimeStr);
            if (isNaN(date.getTime())) {
              throw new Error('Invalid date');
            }
            ipData.observeStartTime = {
              seconds: Math.floor(date.getTime() / 1000),
              nanos: (date.getTime() % 1000) * 1e6,
            };
          } catch (e) {
            console.log(
              `Ignoring observe start time '${startTimeStr}' since it can't be parsed`,
            );
          }
        }
      }

      if (ipInfo.observeEndTime) {
        const endTimeStr = ipInfo.observeEndTime.trim();
        if (endTimeStr) {
          try {
            const date = new Date(endTimeStr);
            if (isNaN(date.getTime())) {
              throw new Error('Invalid date');
            }
            ipData.observeEndTime = {
              seconds: Math.floor(date.getTime() / 1000),
              nanos: (date.getTime() % 1000) * 1e6,
            };
          } catch (e) {
            console.log(
              `Ignoring observe end time '${endTimeStr}' since it can't be parsed`,
            );
          }
        }
      }

      ipDatas.push(ipData);
    }

    if (userData.userIdentifiers.length > 0 || ipDatas.length > 0) {
      const compositeData = CompositeData.create();
      if (userData.userIdentifiers.length > 0) {
        compositeData.userData = userData;
      }
      if (ipDatas.length > 0) {
        compositeData.ipData = ipDatas;
      }
      audienceMembers.push(AudienceMember.create({compositeData}));
    } else {
      console.warn('Ignoring line. No data.');
    }
  }

  const destination = Destination.create({
    operatingAccount: ProductAccount.create({
      accountType: operatingAccountType,
      accountId: argv.operating_account_id,
    }),
    productDestinationId: argv.audience_id,
  });

  // The login account is optional.
  if (argv.login_account_type) {
    const loginAccountType = convertToAccountType(
      argv.login_account_type,
      'login_account_type',
    );
    destination.loginAccount = ProductAccount.create({
      accountType: loginAccountType,
      accountId: argv.login_account_id,
    });
  }

  // The linked account is optional.
  if (argv.linked_account_type) {
    const linkedAccountType = convertToAccountType(
      argv.linked_account_type,
      'linked_account_type',
    );
    destination.linkedAccount = ProductAccount.create({
      accountType: linkedAccountType,
      accountId: argv.linked_account_id,
    });
  }

  // Creates a client for the ingestion service. Increases the maximum metadata
  // size to 16MB to prevent RESOURCE_EXHAUSTED errors when responses contain
  // large error details.
  const client = new IngestionServiceClient({
    'grpc.max_metadata_size': 16 * 1024 * 1024,
  });

  let requestCount = 0;
  // Batches requests to send up to the maximum number of audience members per request.
  for (let i = 0; i < audienceMembers.length; i += MAX_MEMBERS_PER_REQUEST) {
    requestCount++;
    const audienceMembersBatch = audienceMembers.slice(
      i,
      i + MAX_MEMBERS_PER_REQUEST,
    );

    const request = IngestAudienceMembersRequest.create({
      destinations: [destination],
      // Adds members from the current batch.
      audienceMembers: audienceMembersBatch,
      consent: Consent.create({
        adUserData: ConsentStatus.CONSENT_GRANTED,
        adPersonalization: ConsentStatus.CONSENT_GRANTED,
      }),
      termsOfService: TermsOfService.create({
        customerMatchTermsOfServiceStatus: TermsOfServiceStatus.ACCEPTED,
      }),
      // Sets encoding to match the encoding used.
      encoding: DataManagerEncoding.HEX,
      // Sets validate_only. If true, then the Data Manager API only validates the request
      // but doesn't apply changes.
      validateOnly: argv.validate_only,
    });

    console.log(`Request #${requestCount}:\n `, request);

    const [response] = await client.ingestAudienceMembers(request);
    console.log(`Response for request #${requestCount}:\n `, response);

    if (response.fieldWarnings && response.fieldWarnings.length > 0) {
      console.warn(
        'Request ingested successfully, but field warnings were returned. ' +
          'Review warning details and update your implementation as needed.',
      );
    }
  }
  console.log(`# of requests sent: ${requestCount}`);
}

/**
 * Reads the user data from the given JSON file.
 * @param {string} jsonFile The path to the JSON file.
 * @return {Promise<MemberRow[]>} A promise that resolves with an array of user data.
 */
async function readMemberData(jsonFile: string): Promise<MemberRow[]> {
  const data = await fs.promises.readFile(jsonFile, 'utf8');
  return JSON.parse(data) as MemberRow[];
}

/**
 * Validates that a given string is an enum value for the AccountType enum, and
 * if validation passes, returns the AccountType enum value.
 * @param proposedValue the name of an AccountType enum value
 * @param paramName the name of the parameter to use in the error message if validation fails
 * @returns {protos.google.ads.datamanager.v1.ProductAccount.AccountType} The corresponding enum value.
 * @throws {Error} If the string is not an AccountType enum value.
 */
function convertToAccountType(
  proposedValue: string,
  paramName: string,
): protos.google.ads.datamanager.v1.ProductAccount.AccountType {
  const AccountType = ProductAccount.AccountType;
  const accountTypeEnumNames = Object.keys(AccountType).filter(key =>
    isNaN(Number(key)),
  );
  if (!accountTypeEnumNames.includes(proposedValue)) {
    throw new Error(`Invalid ${paramName}: ${proposedValue}`);
  }
  return AccountType[proposedValue as keyof typeof AccountType];
}

if (require.main === module) {
  main().catch(console.error);
}
ingest_audience_members.ts