Introduce GoogleDrive Fetcher for tika-pipes

tika-grpc/pom.xml

@@ -222,6 +222,11 @@
       <artifactId>tika-fetcher-http</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.tika</groupId>
+      <artifactId>tika-fetcher-google</artifactId>
+      <version>${project.version}</version>
+    </dependency>
     <dependency>
       <groupId>com.fasterxml.jackson.module</groupId>
       <artifactId>jackson-module-jsonSchema</artifactId>

tika-parent/pom.xml

@@ -643,6 +643,16 @@
         <groupId>com.google.guava</groupId>
         <artifactId>guava</artifactId>
         <version>${guava.version}</version>
+        <exclusions>
+            <exclusion>
+                <groupId>com.google.errorprone</groupId>
+                <artifactId>error_prone_annotations</artifactId>
+            </exclusion>
+            <exclusion>
+                <groupId>com.google.j2objc</groupId>
+                <artifactId>j2objc-annotations</artifactId>
+            </exclusion>
+        </exclusions>
       </dependency>
       <dependency>
         <groupId>com.googlecode.json-simple</groupId>

tika-pipes/tika-fetchers/pom.xml

@@ -37,6 +37,7 @@
     <module>tika-fetcher-gcs</module>
     <module>tika-fetcher-az-blob</module>
     <module>tika-fetcher-microsoft-graph</module>
+    <module>tika-fetcher-google</module>
   </modules>
 
   <dependencies>
@@ -45,4 +46,4 @@
   <scm>
     <tag>3.0.0-rc1</tag>
   </scm>
-</project>
+</project>

tika-pipes/tika-fetchers/tika-fetcher-google/pom.xml

@@ -0,0 +1,121 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF licenses this file
+  to you 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <artifactId>tika-fetchers</artifactId>
+        <groupId>org.apache.tika</groupId>
+        <version>4.0.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>tika-fetcher-google</artifactId>
+    <name>Google Tika Pipes Fetcher</name>
+
+    <properties>
+        <google.api.client.version>2.7.0</google.api.client.version>
+    </properties>
+
+    <dependencies>
+        <!-- Apache Tika Core -->
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>tika-core</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <!-- Google Drive API Client -->
+        <dependency>
+            <groupId>com.google.api-client</groupId>
+            <artifactId>google-api-client</artifactId>
+            <version>${google.api.client.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.auth</groupId>
+                    <artifactId>google-auth-library-oauth2-http</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.auth</groupId>
+            <artifactId>google-auth-library-oauth2-http</artifactId>
+            <version>1.30.0</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.errorprone</groupId>
+                    <artifactId>error_prone_annotations</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.google.j2objc</groupId>
+                    <artifactId>j2objc-annotations</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <!-- Google Drive API -->
+        <dependency>
+            <groupId>com.google.apis</groupId>
+            <artifactId>google-api-services-drive</artifactId>
+            <version>v3-rev20241027-2.0.0</version>
+        </dependency>
+
+        <!-- Logging -->
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+
+        <!-- Apache Commons IO -->
+        <dependency>
+            <groupId>commons-io</groupId>
+            <artifactId>commons-io</artifactId>
+        </dependency>
+
+        <!-- Test Dependencies -->
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                <archive>
+                    <manifestEntries>
+                    <Automatic-Module-Name>org.apache.tika.pipes.fetcher.s3</Automatic-Module-Name>
+                    </manifestEntries>
+                </archive>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+  <scm>
+    <tag>3.0.0-BETA-rc1</tag>
+  </scm>
+</project>

tika-pipes/tika-fetchers/tika-fetcher-google/src/main/java/org/apache/tika/pipes/fetchers/google/GoogleDriveFetcher.java

@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.tika.pipes.fetchers.google;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.List;
+import java.util.Map;
+
+import com.google.api.client.googleapis.javanet.GoogleNetHttpTransport;
+import com.google.api.client.http.HttpRequestInitializer;
+import com.google.api.client.json.JsonFactory;
+import com.google.api.client.json.gson.GsonFactory;
+import com.google.api.services.drive.Drive;
+import com.google.api.services.drive.DriveScopes;
+import com.google.auth.http.HttpCredentialsAdapter;
+import com.google.auth.oauth2.GoogleCredentials;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.tika.config.Field;
+import org.apache.tika.config.Initializable;
+import org.apache.tika.config.InitializableProblemHandler;
+import org.apache.tika.config.Param;
+import org.apache.tika.exception.TikaConfigException;
+import org.apache.tika.exception.TikaException;
+import org.apache.tika.io.TemporaryResources;
+import org.apache.tika.io.TikaInputStream;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.parser.ParseContext;
+import org.apache.tika.pipes.fetcher.AbstractFetcher;
+import org.apache.tika.pipes.fetchers.google.config.GoogleDriveFetcherConfig;
+
+
+/**
+ * GoogleDrive Fetcher allows the fetching of files from a Google Drive, using a
+ * service account key.
+ *
+ * Fetch Keys are ${fileId},${subjectUser}, where the subject user is the
+ * organizer of the file. This user is necessary as part of the key as the
+ * service account must act on behalf of the user when querying for the file.
+ */
+public class GoogleDriveFetcher extends AbstractFetcher implements Initializable {
+    private static final Logger LOGGER = LoggerFactory.getLogger(GoogleDriveFetcher.class);
+    private static final JsonFactory JSON_FACTORY = GsonFactory.getDefaultInstance();
+
+    private GoogleCredentials baseCredentials;
+
+    private Drive driveService;
+    private boolean spoolToTemp;
+    private List<String> scopes;
+
+    private GoogleDriveFetcherConfig config = new GoogleDriveFetcherConfig();
+
+    public GoogleDriveFetcher() {
+        scopes = new ArrayList<>();
+        scopes.add(DriveScopes.DRIVE_READONLY);
+    }
+
+    public GoogleDriveFetcher(GoogleDriveFetcherConfig config) {
+        this.config = config;
+    }
+
+    @Field
+    public void setThrottleSeconds(String commaDelimitedLongs) throws TikaConfigException {
+        String[] longStrings = (commaDelimitedLongs == null ? "" : commaDelimitedLongs).split(",");
+        long[] seconds = new long[longStrings.length];
+        for (int i = 0; i < longStrings.length; i++) {
+            try {
+                seconds[i] = Long.parseLong(longStrings[i]);
+            } catch (NumberFormatException e) {
+                throw new TikaConfigException(e.getMessage());
+            }
+        }
+        setThrottleSeconds(seconds);
+    }
+
+    public void setThrottleSeconds(long[] throttleSeconds) {
+        config.setThrottleSeconds(throttleSeconds);
+    }
+
+    @Field
+    public void setSpoolToTemp(boolean spoolToTemp) {
+        config.setSpoolToTemp(spoolToTemp);
+    }
+
+    @Field
+    public void setServiceAccountKeyBase64(String serviceAccountKeyBase64) {
+        config.setServiceAccountKeyBase64(serviceAccountKeyBase64);
+    }
+
+    @Field
+    public void setSubjectUser(String subjectUser) {
+        config.setSubjectUser(subjectUser);
+    }
+
+    @Field
+    public void setScopes(List<String> scopes) {
+        config.setScopes(new ArrayList<>(scopes));
+        if (config.getScopes().isEmpty()) {
+            config.getScopes().add(DriveScopes.DRIVE_READONLY);
+        }
+    }
+
+    @Override
+    public void initialize(Map<String, Param> map) throws TikaConfigException {
+        try {
+            baseCredentials = GoogleCredentials
+                    .fromStream(new ByteArrayInputStream(Base64.getDecoder().decode(config.getServiceAccountKeyBase64())))
+                    .createScoped(scopes);
+        } catch (IOException e) {
+            throw new TikaConfigException("Failed to initialize Google Drive service", e);
+        }
+    }
+
+    @Override
+    public void checkInitialization(InitializableProblemHandler initializableProblemHandler) throws TikaConfigException {
+    }
+
+    @Override
+    public InputStream fetch(String fetchKey, Metadata metadata, ParseContext parseContext) throws TikaException, IOException {
+        int tries = 0;
+        Exception ex = null;
+        TemporaryResources tmp = null;
+
+        do {
+            long start = System.currentTimeMillis();
+            try {
+                String[] fetchKeySplit = fetchKey.split(",");
+                if (fetchKeySplit.length != 2) {
+                    throw new TikaException("Invalid fetch key, expected format ${fileId},${subjectUser}: " + fetchKey);
+                }
+
+                String fileId = fetchKeySplit[0];
+                String subjectUser = fetchKeySplit[1];
+
+                GoogleCredentials delegatedCredentials = baseCredentials.createDelegated(subjectUser);
+                final HttpRequestInitializer requestInitializer = new HttpCredentialsAdapter(delegatedCredentials);
+
+                driveService = new Drive.Builder(
+                    GoogleNetHttpTransport.newTrustedTransport(), 
+                    JSON_FACTORY, 
+                    requestInitializer).setApplicationName("tika-fetcher-google").build();
+
+                InputStream is = driveService.files()
+                        .get(fileId)
+                        .executeMediaAsInputStream();
+
+                if (is == null) {
+                    throw new IOException("Empty input stream when we tried to parse " + fetchKey);
+                }
+
+                if (spoolToTemp) {
+                    tmp = new TemporaryResources();
+                    Path tmpPath = tmp.createTempFile(fileId + ".dat");
+                    Files.copy(is, tmpPath);
+                    return TikaInputStream.get(tmpPath);
+                }
+                return TikaInputStream.get(is);
+
+            } catch (Exception e) {
+                LOGGER.warn("Exception fetching on retry=" + tries, e);
+                ex = e;
+            } finally {
+                long elapsed = System.currentTimeMillis() - start;
+                LOGGER.debug("Total to fetch {}", elapsed);
+            }
+
+            long[] throttleSeconds = config.getThrottleSeconds();
+
+            LOGGER.warn("Sleeping for {} seconds before retry", throttleSeconds[tries]);
+            try {
+                Thread.sleep(throttleSeconds[tries] * 1000);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        } while (++tries < config.getThrottleSeconds().length);
+
+        throw new TikaException("Could not fetch " + fetchKey, ex);
+    }
+}