Merge pull request #14 from PitterPatterPython/flashpoint-connector

Flashpoint connector

.env.sample

@@ -10,4 +10,9 @@ VERIFY_SSL=
 ############
 SPYCLOUD_API_ATO_KEY=
 SPYCLOUD_API_INV_KEY=
-SPYCLOUD_API_SIP_KEY=
+SPYCLOUD_API_SIP_KEY=
+
+##############
+# FLASHPOINT #
+##############
+FLASHPOINT_API_KEY=

ppp_connectors/flashpoint.py

@@ -0,0 +1,140 @@
+from typing import Dict, Any, List
+from requests import Response
+from .broker import make_request
+from .helpers import check_required_env_vars, combine_env_configs
+
+
+env_config: Dict[str, Any] = combine_env_configs()
+
+def flashpoint_search_communities(query: str, **kwargs: Dict[str, Any]) -> Response:
+    """Communities Search allows search requests over article and conversation data.
+    Article data is made up of things like blogs and paste sites. Conversation data
+    is made up of chats, forums, and social media posts.
+
+    Args:
+        query (str): A word or phrase to search.
+
+    Returns:
+        Response: requests.Response object from the request
+    """
+
+    required_vars: List[str] = [
+        'FLASHPOINT_API_KEY'
+    ]
+
+    # Check and ensure that required variables are present, exits if not
+    check_required_env_vars(env_config, required_vars)
+
+    method: str = 'post'
+    url: str = 'https://api.flashpoint.io/sources/v2/communities'
+    headers: Dict = {
+        'accept': 'application/json',
+        'content-type': 'application/json',
+        'Authorization': f'Bearer {env_config["FLASHPOINT_API_KEY"]}'
+    }
+    payload: Dict = {
+        'query': query,
+        **kwargs
+    }
+
+    result: Response = make_request(method=method, url=url, headers=headers, json=payload)
+
+    return result
+
+def flashpoint_search_media(query: str, **kwargs: Dict[str, Any]) -> Response:
+    """Media search allows search requests over our media data, specifically
+    media that have been through our Optical Character Recogintion (OCR) process.
+    Once media have been through our OCR process, any text, classifications, or
+    logos found within the media are available for search.
+
+    Args:
+        query (str): A word or phrase to search.
+
+    Returns:
+        Response: requests.Response object from the request
+    """
+
+    required_vars: List[str] = [
+        'FLASHPOINT_API_KEY'
+    ]
+
+    # Check and ensure that required variables are present, exits if not
+    check_required_env_vars(env_config, required_vars)
+
+    method: str = 'post'
+    url: str = 'https://api.flashpoint.io/sources/v2/media'
+    headers: Dict = {
+        'accept': 'application/json',
+        'content-type': 'application/json',
+        'Authorization': f'Bearer {env_config["FLASHPOINT_API_KEY"]}'
+    }
+    payload: Dict = {
+        'query': query,
+        **kwargs
+    }
+
+    result: Response = make_request(method=method, url=url, headers=headers, json=payload)
+
+    return result
+
+def flashpoint_get_media_object(id: str) -> Response:
+    """Media ID request allows users to directly lookup the document based on the media ID provided.
+
+    Args:
+        id (str): the id of the media object to retrieve
+
+    Returns:
+        Response: requests.Response object from the request
+    """
+
+    required_vars: List[str] = [
+        'FLASHPOINT_API_KEY'
+    ]
+
+    # Check and ensure that required variables are present, exits if not
+    check_required_env_vars(env_config, required_vars)
+
+    method: str = 'get'
+    url: str = f'https://api.flashpoint.io/sources/v2/media/{id}'
+    headers: Dict = {
+        'accept': 'application/json',
+        'content-type': 'application/json',
+        'Authorization': f'Bearer {env_config["FLASHPOINT_API_KEY"]}'
+    }
+
+    result: Response = make_request(method=method, url=url, headers=headers)
+
+    return result
+
+def flashpoint_get_media_image(storage_uri: str) -> Response:
+    """Download the media from a media object by its storage_uri field
+
+    Args:
+        storage_uri (str): the storage_uri field from the media object
+
+    Returns:
+        Response: requests.Response object from the request
+    """
+
+    required_vars: List[str] = [
+        'FLASHPOINT_API_KEY'
+    ]
+
+    # Check and ensure that required variables are present, exits if not
+    check_required_env_vars(env_config, required_vars)
+
+    method: str = 'get'
+    url: str = f'https://api.flashpoint.io/sources/v1/media/'
+    headers: Dict = {
+        'accept': 'application/json',
+        'content-type': 'application/json',
+        'Authorization': f'Bearer {env_config["FLASHPOINT_API_KEY"]}'
+    }
+
+    params = {
+        "asset_id": storage_uri
+    }
+
+    result: Response = make_request(method=method, url=url, headers=headers, params=params)
+
+    return result

ppp_connectors/helpers.py

@@ -25,14 +25,12 @@ def check_required_env_vars(config: Dict[str, str], required_vars: List[str]) ->
         sys.exit(1)
 
 def combine_env_configs() -> Dict[str, Any]:
-    """_summary_
-
-    Args:
-        dotenv_config (Dict[str, str]): _description_
-        osenv_config (Dict[str, str]): _description_
+    """Find a .env file if it exists, and combine it with system environment
+        variables to form a "combined_config" dictionary of environment variables
 
     Returns:
-        Dict: _description_
+        Dict: a dictionary containing the output of a .env file (if found), and
+        system environment variables
     """
 
     env_config: Dict[str, Any] = dict(dotenv_values(find_dotenv()))

ppp_connectors/spycloud.py

@@ -1,4 +1,3 @@
-import sys
 from typing import Dict, Any, List
 from requests import Response
 from .broker import make_request

pyproject.toml

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "ppp-connectors"
 packages = [{ include = "ppp_connectors" }]
-version = "0.1.8"
+version = "0.2.0"
 description = "A simple, lightweight set of connectors and functions to various APIs, controlled by a central broker."
 authors = [
     "Rob D'Aveta <[email protected]>",