Refactor Muckrock Scraper

pdap_api_client/AccessManager.py

@@ -0,0 +1,123 @@
+from http import HTTPStatus
+from typing import Optional
+
+import requests
+
+from pdap_api_client.DTOs import RequestType, Namespaces, RequestInfo, ResponseInfo
+
+API_URL = "https://data-sources-v2.pdap.dev/api"
+request_methods = {
+    RequestType.POST: requests.post,
+    RequestType.PUT: requests.put,
+    RequestType.GET: requests.get,
+    RequestType.DELETE: requests.delete,
+}
+
+
+class CustomHTTPException(Exception):
+    pass
+
+
+def build_url(
+    namespace: Namespaces,
+    subdomains: Optional[list[str]] = None
+):
+    url = f"{API_URL}/{namespace.value}"
+    if subdomains is not None:
+        url = f"{url}/{'/'.join(subdomains)}"
+    return url
+
+
+class AccessManager:
+    """
+    Manages login, api key, access and refresh tokens
+    """
+    def __init__(self, email: str, password: str, api_key: Optional[str] = None):
+        self.access_token = None
+        self.refresh_token = None
+        self.api_key = api_key
+        self.login(email=email, password=password)
+
+    # TODO: Add means to refresh if token expired.
+
+    def load_api_key(self):
+        url = build_url(
+            namespace=Namespaces.AUTH,
+            subdomains=["api-key"]
+        )
+        request_info = RequestInfo(
+            type_ = RequestType.POST,
+            url=url,
+            headers=self.jwt_header()
+        )
+        response_info = self.make_request(request_info)
+        self.api_key = response_info.data["api_key"]
+
+    def refresh_access_token(self):
+        url = build_url(
+            namespace=Namespaces.AUTH,
+            subdomains=["refresh-session"],
+        )
+        raise NotImplementedError("Waiting on https://github.com/Police-Data-Accessibility-Project/data-sources-app/issues/566")
+
+    def make_request(self, ri: RequestInfo) -> ResponseInfo:
+        try:
+            response = request_methods[ri.type_](
+                ri.url,
+                json=ri.json,
+                headers=ri.headers,
+                params=ri.params,
+                timeout=ri.timeout
+            )
+            response.raise_for_status()
+        except requests.RequestException as e:
+            # TODO: Precise string matching here is brittle. Consider changing later.
+            if e.response.json().message == "Token is expired. Please request a new token.":
+                self.refresh_access_token()
+                return self.make_request(ri)
+            else:
+                raise CustomHTTPException(f"Error making {ri.type_} request to {ri.url}: {e}")
+        return ResponseInfo(
+            status_code=HTTPStatus(response.status_code),
+            data=response.json()
+        )
+
+    def login(self, email: str, password: str):
+        url = build_url(
+            namespace=Namespaces.AUTH,
+            subdomains=["login"]
+        )
+        request_info = RequestInfo(
+            type_=RequestType.POST,
+            url=url,
+            json={
+                "email": email,
+                "password": password
+            }
+        )
+        response_info = self.make_request(request_info)
+        data = response_info.data
+        self.access_token = data["access_token"]
+        self.refresh_token = data["refresh_token"]
+
+
+    def jwt_header(self) -> dict:
+        """
+        Retrieve JWT header
+        Returns: Dictionary of Bearer Authorization with JWT key
+        """
+        return {
+            "Authorization": f"Bearer {self.access_token}"
+        }
+
+    def api_key_header(self):
+        """
+        Retrieve API key header
+        Returns: Dictionary of Basic Authorization with API key
+
+        """
+        if self.api_key is None:
+            self.load_api_key()
+        return {
+            "Authorization": f"Basic {self.api_key}"
+        }

pdap_api_client/DTOs.py

@@ -0,0 +1,54 @@
+from enum import Enum
+from http import HTTPStatus
+from typing import Optional
+
+from pydantic import BaseModel
+
+
+class MatchAgencyInfo(BaseModel):
+    submitted_name: str
+    id: str
+
+class ApprovalStatus(Enum):
+    APPROVED = "approved"
+    REJECTED = "rejected"
+    PENDING = "pending"
+    NEEDS_IDENTIFICATION = "needs identification"
+
+
+
+class UniqueURLDuplicateInfo(BaseModel):
+    original_url: str
+    approval_status: ApprovalStatus
+    rejection_note: str
+
+class UniqueURLResponseInfo(BaseModel):
+    is_unique: bool
+    duplicates: list[UniqueURLDuplicateInfo]
+
+
+class Namespaces(Enum):
+    AUTH = "auth"
+    MATCH = "match"
+    CHECK = "check"
+
+
+class RequestType(Enum):
+    POST = "POST"
+    PUT = "PUT"
+    GET = "GET"
+    DELETE = "DELETE"
+
+
+class RequestInfo(BaseModel):
+    type_: RequestType
+    url: str
+    json: Optional[dict] = None
+    headers: Optional[dict] = None
+    params: Optional[dict] = None
+    timeout: Optional[int] = 10
+
+
+class ResponseInfo(BaseModel):
+    status_code: HTTPStatus
+    data: Optional[dict]

pdap_api_client/PDAPClient.py

@@ -0,0 +1,65 @@
+from typing import List
+
+from pdap_api_client.AccessManager import build_url, AccessManager
+from pdap_api_client.DTOs import MatchAgencyInfo, UniqueURLDuplicateInfo, UniqueURLResponseInfo, Namespaces, \
+    RequestType, RequestInfo
+
+
+class PDAPClient:
+
+    def __init__(self, access_manager: AccessManager):
+        self.access_manager = access_manager
+
+    def match_agency(
+            self,
+            name: str,
+            state: str,
+            county: str,
+            locality: str
+    ) -> List[MatchAgencyInfo]:
+        """
+        Returns agencies, if any, that match or partially match the search criteria
+        """
+        url = build_url(
+            namespace=Namespaces.MATCH,
+            subdomains=["agency"]
+        )
+        request_info = RequestInfo(
+            type_=RequestType.POST,
+            url=url,
+            json={
+                "name": name,
+                "state": state,
+                "county": county,
+                "locality": locality
+            }
+        )
+        response_info = self.access_manager.make_request(request_info)
+        return [MatchAgencyInfo(**agency) for agency in response_info.data["agencies"]]
+
+
+    def is_url_unique(
+        self,
+        url_to_check: str
+    ) -> UniqueURLResponseInfo:
+        """
+        Check if a URL is unique. Returns duplicate info otherwise
+        """
+        url = build_url(
+            namespace=Namespaces.CHECK,
+            subdomains=["unique-url"]
+        )
+        request_info = RequestInfo(
+            type_=RequestType.GET,
+            url=url,
+            params={
+                "url": url_to_check
+            }
+        )
+        response_info = self.access_manager.make_request(request_info)
+        duplicates = [UniqueURLDuplicateInfo(**entry) for entry in response_info.data["duplicates"]]
+        is_unique = (len(duplicates) == 0)
+        return UniqueURLResponseInfo(
+            is_unique=is_unique,
+            duplicates=duplicates
+        )

source_collectors/muckrock/README.md

@@ -56,8 +56,6 @@ pip install -r requirements.txt
 
 ### 2. Clone Muckrock database & search locally
 
-~~- `download_muckrock_foia.py` `search_local_foia_json.py`~~ (deprecated)
-
 - scripts to clone the MuckRock foia requests collection for fast local querying (total size <2GB at present)
 
 - `create_foia_data_db.py` creates and populates a SQLite database (`foia_data.db`) with all MuckRock foia requests. Various errors outside the scope of this script may occur; a counter (`last_page_fetched.txt`) is created to keep track of the most recent page fetched and inserted into the database. If the program exits prematurely, simply run `create_foia_data_db.py` again to continue where you left off. A log file is created to capture errors for later reference.

source_collectors/muckrock/classes/FOIADBSearcher.py

@@ -0,0 +1,65 @@
+import os
+import sqlite3
+
+import pandas as pd
+
+from source_collectors.muckrock.constants import FOIA_DATA_DB
+
+check_results_table_query = """
+                SELECT name FROM sqlite_master
+                WHERE (type = 'table')
+                AND (name = 'results')
+                """
+
+search_foia_query = """
+        SELECT * FROM results
+        WHERE (title LIKE ? OR tags LIKE ?)
+        AND (status = 'done')
+        """
+
+
+class FOIADBSearcher:
+
+    def __init__(self, db_path = FOIA_DATA_DB):
+        self.db_path = db_path
+        if not os.path.exists(self.db_path):
+            raise FileNotFoundError("foia_data.db does not exist.\nRun create_foia_data_db.py first to create and populate it.")
+
+
+    def search(self, search_string: str) -> pd.DataFrame | None:
+        """
+        Searches the foia_data.db database for FOIA request entries matching the provided search string.
+
+        Args:
+            search_string (str): The string to search for in the `title` and `tags` of the `results` table.
+
+        Returns:
+            Union[pandas.DataFrame, None]:
+                - pandas.DataFrame: A DataFrame containing the matching entries from the database.
+                - None: If an error occurs during the database operation.
+
+        Raises:
+            sqlite3.Error: If any database operation fails, prints error and returns None.
+            Exception: If any unexpected error occurs, prints error and returns None.
+        """
+        try:
+            with sqlite3.connect(self.db_path) as conn:
+                results_table = pd.read_sql_query(check_results_table_query, conn)
+                if results_table.empty:
+                    print("The `results` table does not exist in the database.")
+                    return None
+
+                df = pd.read_sql_query(
+                    sql=search_foia_query,
+                    con=conn,
+                    params=[f"%{search_string}%", f"%{search_string}%"]
+                )
+
+        except sqlite3.Error as e:
+            print(f"Sqlite error: {e}")
+            return None
+        except Exception as e:
+            print(f"An unexpected error occurred: {e}")
+            return None
+
+        return df

source_collectors/muckrock/classes/FOIASearcher.py

@@ -0,0 +1,58 @@
+from typing import Optional
+
+from source_collectors.muckrock.classes.muckrock_fetchers import FOIAFetcher
+from tqdm import tqdm
+
+class FOIASearcher:
+    """
+    Used for searching FOIA data from MuckRock
+    """
+
+    def __init__(self, fetcher: FOIAFetcher, search_term: Optional[str] = None):
+        self.fetcher = fetcher
+        self.search_term = search_term
+
+    def fetch_page(self) -> dict | None:
+        """
+        Fetches the next page of results using the fetcher.
+        """
+        data = self.fetcher.fetch_next_page()
+        if data is None or data.get("results") is None:
+            return None
+        return data
+
+    def filter_results(self, results: list[dict]) -> list[dict]:
+        """
+        Filters the results based on the search term.
+        Override or modify as needed for custom filtering logic.
+        """
+        if self.search_term:
+            return [result for result in results if self.search_term.lower() in result["title"].lower()]
+        return results
+
+    def update_progress(self, pbar: tqdm, results: list[dict]) -> int:
+        """
+        Updates the progress bar and returns the count of results processed.
+        """
+        num_results = len(results)
+        pbar.update(num_results)
+        return num_results
+
+    def search_to_count(self, max_count: int) -> list[dict]:
+        """
+        Fetches and processes results up to a maximum count.
+        """
+        count = max_count
+        all_results = []
+        with tqdm(total=max_count, desc="Fetching results", unit="result") as pbar:
+            while count > 0:
+                data = self.fetch_page()
+                if not data:
+                    break
+
+                results = self.filter_results(data["results"])
+                all_results.extend(results)
+                count -= self.update_progress(pbar, results)
+
+        return all_results
+