Update patent_downloader.py

patent_downloader.py

@@ -1,170 +1,136 @@
 from typing import List, Union, Optional
 import os
 import requests
-import re
-import time
-import shutil
-import subprocess
-import pandas as pd
-from selenium import webdriver
-from selenium.webdriver.common.keys import Keys
-from selenium.webdriver.chrome.service import Service
-from selenium.webdriver.chrome.options import Options
+import tempfile
 from bs4 import BeautifulSoup
-from selenium.webdriver.common.by import By
-from selenium.webdriver.support.ui import WebDriverWait
-from selenium.webdriver.support import expected_conditions as EC
-import chromedriver_autoinstaller
 
 
 class PatentDownloader:
-    url = "https://patents.google.com"
+    """
+    A class to automate downloading patent PDFs from Google Patents.
+    """
+    base_url = "https://patents.google.com/patent"
 
     def __init__(self, verbose: bool = False):
         """
+        Initialize the downloader.
+
         Parameters
         ----------
         verbose : bool
-            Print additional debug information.
+            If True, print detailed debug information.
         """
         self.verbose = verbose
-        self.chrome_path = self.install_chrome()
 
-    def install_chrome(self) -> str:
+    def download(self, patents: Union[str, List[str]], output_path: Optional[str] = None) -> List[str]:
         """
-        Download and install Google Chrome dynamically.
+        Download single or multiple patent PDFs.
+
+        Parameters
+        ----------
+        patents : str or List[str]
+            Single patent number or a list of patent numbers.
+        output_path : Optional[str]
+            Directory to save the PDFs. Defaults to a temporary directory.
+
         Returns
         -------
-        str: Path to the Chrome binary.
-        """
-        chrome_path = "/usr/bin/google-chrome"
-
-        if not shutil.which("google-chrome"):
-            print("Downloading and installing Google Chrome...")
-            subprocess.run(
-                "wget https://dl.google.com/linux/direct/google-chrome-stable_current_amd64.deb -O chrome.deb",
-                shell=True,
-                check=True,
-            )
-            subprocess.run(
-                "apt-get update && apt-get install -y ./chrome.deb",
-                shell=True,
-                check=True,
-            )
-            os.remove("chrome.deb")
-
-        if not shutil.which("google-chrome"):
-            raise ValueError("Google Chrome installation failed!")
-        return chrome_path
-
-    def download(self, patent: Union[str, List[str]], output_path: str = "./",
-                 waiting_time: int = 10, remove_kind_codes: Optional[List[str]] = None) -> None:
+        List[str]
+            List of paths to the downloaded PDFs.
         """
-        Download patent document(s) as PDF.
-        """
-        if isinstance(patent, list) or os.path.isfile(patent):
-            self.get_pdfs(patent, output_path, waiting_time, remove_kind_codes)
-        else:
-            self.get_pdf(patent, output_path, waiting_time, remove_kind_codes)
+        if isinstance(patents, str):
+            patents = [patents]
 
-    def get_pdf(self, patent: str, output_path: str = "./", waiting_time: int = 10,
-                remove_kind_codes: Optional[List[str]] = None) -> None:
+        # Use a temporary directory if no output path is provided
+        output_dir = output_path or tempfile.gettempdir()
+        os.makedirs(output_dir, exist_ok=True)
+
+        downloaded_files = []
+
+        for i, patent in enumerate(patents):
+            try:
+                if self.verbose:
+                    print(f"🔍 Downloading {i+1}/{len(patents)}: {patent}")
+                file_path = self._download_single_pdf(patent, output_dir)
+                downloaded_files.append(file_path)
+                print(f"✅ Successfully downloaded: {file_path}")
+            except Exception as e:
+                print(f"❌ Failed to download {patent}: {e}")
+
+        return downloaded_files
+
+    def _download_single_pdf(self, patent_number: str, output_dir: str) -> str:
         """
         Download a single patent PDF.
+
+        Parameters
+        ----------
+        patent_number : str
+            The patent number (e.g., "US8676427B1").
+        output_dir : str
+            Directory to save the PDF.
+
+        Returns
+        -------
+        str
+            Path to the downloaded PDF file.
         """
-        if remove_kind_codes:
-            for kind_code in remove_kind_codes:
-                patent = re.sub(kind_code + "$", "", patent)
-
-        # Automatically install ChromeDriver
-        chromedriver_autoinstaller.install()
-
-        # Set up Chrome options
-        chrome_options = Options()
-        chrome_options.binary_location = self.chrome_path
-        chrome_options.add_argument("--headless")
-        chrome_options.add_argument("--no-sandbox")
-        chrome_options.add_argument("--disable-dev-shm-usage")
-
-        # Initialize Selenium WebDriver
-        service = Service()
-        driver = webdriver.Chrome(service=service, options=chrome_options)
-        pdf_link = None  # Ensure pdf_link is defined
-
-        try:
-            driver.get(self.url)
-
-            # Wait for the search input field and interact with it
-            print("Waiting for the search input field...")
-            search_input_xpath = "//input[@aria-label='Search patents']"
-            WebDriverWait(driver, 20).until(
-                EC.presence_of_element_located((By.XPATH, search_input_xpath))
-            )
-            element = driver.find_element(By.XPATH, search_input_xpath)
-            print("Search input field located.")
-
-            element.send_keys(patent)
-            element.send_keys(Keys.RETURN)
-
-            # Wait for search results to load
-            print("Waiting for search results to load...")
-            WebDriverWait(driver, 20).until(
-                EC.presence_of_element_located((By.TAG_NAME, "body"))
-            )
-            time.sleep(waiting_time)
-
-            # Parse HTML and get the PDF link
-            soup = BeautifulSoup(driver.page_source, "html.parser")
-            pdf_link = self.get_pdf_link(soup, patent)
-        except Exception as e:
-            print(f"Error occurred: {e}")
-        finally:
-            driver.quit()
-
-        # Download the PDF
-        if pdf_link:
-            validate_directory(output_path)
-            pdf_content = requests.get(pdf_link).content
-            with open(os.path.join(output_path, f"{patent}.pdf"), "wb") as file:
-                file.write(pdf_content)
-            print(f">>> Patent {patent} successfully downloaded <<<")
-        else:
-            print(f"Error: PDF link for patent {patent} not found!")
-
-    def get_pdfs(self, patents: Union[List[str], str], output_path: str = "./",
-                 waiting_time: int = 10, remove_kind_codes: Optional[List[str]] = None) -> None:
-        """
-        Download multiple patent PDFs from a list or file.
-        """
-        if isinstance(patents, str):
-            if patents.lower().endswith('csv'):
-                df_patents = pd.read_csv(patents)
-                patents = df_patents['patent_number'].to_list()
-            elif patents.lower().endswith('txt'):
-                with open(patents, 'r') as txt_file:
-                    patents = txt_file.read().splitlines()
-            else:
-                raise NotImplementedError(f'Unsupported file type: {patents}')
+        # Construct the Google Patents URL
+        patent_url = f"{self.base_url}/{patent_number}/en"
 
-        for i, patent in enumerate(patents):
-            print(len(patents) - i, "patent(s) remaining.")
-            self.get_pdf(patent, output_path, waiting_time, remove_kind_codes)
+        if self.verbose:
+            print(f"Fetching patent page: {patent_url}")
+
+        # Fetch the HTML content of the patent page
+        response = requests.get(patent_url)
+        if response.status_code != 200:
+            raise Exception(f"Failed to fetch patent page for {patent_number}. HTTP Status: {response.status_code}")
+
+        # Parse the HTML content and extract the PDF link
+        soup = BeautifulSoup(response.content, "html.parser")
+        pdf_url = self._extract_pdf_link(soup)
+
+        if not pdf_url:
+            raise Exception(f"No PDF link found for patent {patent_number}.")
+
+        if self.verbose:
+            print(f"Found PDF link: {pdf_url}")
+
+        # Download the PDF file
+        pdf_response = requests.get(pdf_url)
+        if pdf_response.status_code != 200:
+            raise Exception(f"Failed to download PDF for {patent_number}. HTTP Status: {pdf_response.status_code}")
+
+        # Save the PDF to the specified output directory
+        file_path = os.path.join(output_dir, f"{patent_number}.pdf")
+        with open(file_path, "wb") as pdf_file:
+            pdf_file.write(pdf_response.content)
+
+        return file_path
 
     @staticmethod
-    def get_pdf_link(soup: BeautifulSoup, patent: str) -> Optional[str]:
+    def _extract_pdf_link(soup: BeautifulSoup) -> Optional[str]:
         """
-        Extract the PDF link from parsed HTML.
+        Extract the PDF link from the page's metadata.
+
+        Parameters
+        ----------
+        soup : BeautifulSoup
+            Parsed HTML content of the patent page.
+
+        Returns
+        -------
+        Optional[str]
+            The direct PDF link if found.
         """
-        pdf_links = [link['href'] for link in soup.find_all('a', href=True) if link['href'].lower().endswith("pdf")]
-        for link in pdf_links:
-            if patent.lower() in link.lower():
-                return link
-        return None
+        # Look for the 'citation_pdf_url' meta tag
+        pdf_meta = soup.find("meta", {"name": "citation_pdf_url"})
+        if pdf_meta and pdf_meta.get("content"):
+            return pdf_meta["content"]
 
+        # Fallback: search for any <a> tag containing '.pdf' in its href
+        pdf_links = [a['href'] for a in soup.find_all("a", href=True) if a['href'].endswith(".pdf")]
+        if pdf_links:
+            return pdf_links[0]  # Return the first matching PDF link
 
-def validate_directory(directory: str) -> None:
-    """
-    Ensure the output directory exists.
-    """
-    if not os.path.exists(directory):
-        os.makedirs(directory)
+        return None