diff --git a/.env.sample b/.env.sample index 5fdcbd9..8dbc976 100644 --- a/.env.sample +++ b/.env.sample @@ -10,4 +10,9 @@ VERIFY_SSL= ############ SPYCLOUD_API_ATO_KEY= SPYCLOUD_API_INV_KEY= -SPYCLOUD_API_SIP_KEY= \ No newline at end of file +SPYCLOUD_API_SIP_KEY= + +############## +# FLASHPOINT # +############## +FLASHPOINT_API_KEY= \ No newline at end of file diff --git a/ppp_connectors/flashpoint.py b/ppp_connectors/flashpoint.py new file mode 100644 index 0000000..5306dae --- /dev/null +++ b/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 \ No newline at end of file diff --git a/ppp_connectors/helpers.py b/ppp_connectors/helpers.py index b566fff..fc0b2c8 100644 --- a/ppp_connectors/helpers.py +++ b/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())) diff --git a/ppp_connectors/spycloud.py b/ppp_connectors/spycloud.py index 29b2f6b..6147c30 100644 --- a/ppp_connectors/spycloud.py +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index a223ce6..ebe1d17 100644 --- a/pyproject.toml +++ b/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 ",