diff --git a/README.md b/README.md index 3835421..ff86b41 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,16 @@ Monitoring tool for Status App communities # Setup +## Environment Variables + +``` +POSTGRES_USERNAME +POSTGRES_PASSWORD +POSTGRES_DATABASE +POSTGRES_HOST +POSTGRES_PORT +``` + ## Docker 1. Login to `harbor.status.im`. Your password is your Harbor **CLI secret**. @@ -29,17 +39,12 @@ conda create -n status-monitoring python=3.12 ```bash pip install -r requirements.txt ``` +**Note**: If you are on Windows, you will have to install `psycopg2` instead of `psycopg2-binary`. ## Files - `create_account.py` - create a Status App account for the given `username` and `password`. Example runs: - -```bash -python create_account.py -u snt-maxxer -p StatusApp#123 -``` - -```bash -python create_account.py --username snt-maxxer --password StatusApp#123 -``` - -- `download.py` - download all messages and overall community info from the specified Status App channels in `config.yaml`. \ No newline at end of file + - `python create_account.py -u snt-maxxer -p StatusApp#123` + - `python create_account.py --username snt-maxxer --password StatusApp#123` +- `download.py` - download all messages and overall community info from the specified Status App channels in `config.yaml`. +- `upload.py` - upload data from `download.py` to Postgres \ No newline at end of file diff --git a/config.yaml b/config.yaml index 18afa35..e71fcaf 100644 --- a/config.yaml +++ b/config.yaml @@ -2,7 +2,8 @@ postgres: schema: "status_app_monitoring" tables: messages: "raw_messages" - community: "community_info" + community: "raw_community_info" + members: "raw_community_activity" status_app: bot_name: "snt-maxxer" diff --git a/postgres.py b/postgres.py new file mode 100644 index 0000000..1721fe6 --- /dev/null +++ b/postgres.py @@ -0,0 +1,61 @@ +""" +Minimum code to upload data taken from: +https://github.com/status-im/ift-data-py/blob/master/ift_data/clients/postgres.py +""" + +import psycopg2 +import pandas as pd +from typing import Optional +from sqlalchemy import create_engine +from sqlalchemy.dialects.postgresql import JSONB + +class Postgres: + + def __init__(self, username: str, password: str, port: int = 5432, database: str = "data-warehouse", host: str = "data-01.do-ams3.bi.test.status.im"): + + self.__params = { + "host": host, + "user": username, + "password": password, + "port": port, + "database": database + } + + self.__url = f"postgresql://{username}:{password}@{host}:{port}/{database}" + self.__conn: psycopg2.extensions.connection = psycopg2.connect(**self.__params) + self.__cursor: psycopg2.extensions.cursor = self.__conn.cursor() + + def insert(self, data: pd.DataFrame, table_name: str, schema: str, json_columns: Optional[list] = None): + """ + Insert the DataFrame in the specified schema > table. + If the schema / table name does not exist, it will be created. + + Parameters: + - `data` - the data to be inserted in Postgres + - `table_name` - the name of the table + - `schema` - the name of the schema + - `json_columns` - when creating the table, `dict` columns will be turned into JSON objects in Postgres + """ + engine = create_engine(self.__url) + + data.columns = [column.lower() for column in data.columns] + + params = { + "name": table_name, + "con": engine, + "schema": schema, + "if_exists": "append", + "index": False + } + if json_columns: + params["dtype"] = { + json_column: JSONB + for json_column in json_columns + } + + data.to_sql(**params) + + + def close(self): + self.__cursor.close() + self.__conn.close() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 55e7618..014501b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -14,4 +14,7 @@ eth-typing~=5.2.1 faker~=37.6.0 qrcode pandas -pyyaml \ No newline at end of file +pyyaml +python-dotenv +psycopg2-binary +sqlalchemy \ No newline at end of file diff --git a/upload.py b/upload.py new file mode 100644 index 0000000..0c737c9 --- /dev/null +++ b/upload.py @@ -0,0 +1,127 @@ +import constants +import json, datetime, os +import pandas as pd +from pathlib import Path, PosixPath +from postgres import Postgres +from dotenv import load_dotenv + +def get_community_members(file_path: PosixPath) -> pd.DataFrame: + """ + Get latest Status App login time for all members in the community. + + Parameters: + - `file_path` - the `.pkl` file that has overall member information from `data_utils.get_community_info` + + Output: + - processed data to be uploaded to the database + """ + info: dict = pd.read_pickle(file_path) + + data = pd.DataFrame(info["members"]["info"]).assign( + community_total = info["members"]["total"], + community_id = info["community_id"], + channels = len(info["channels"]) + ) + return data + +def get_community_info(file_path: PosixPath) -> pd.DataFrame: + """ + Get latest Status App community information + + Parameters: + - `file_path` - the `.pkl` file that has overall member information from `data_utils.get_community_info` + + Output: + - Single row for the given file. The data is returned in a DataFrame to make data uploading more robust + """ + info: dict = pd.read_pickle(file_path) + # members data can be found in function get_community_members + info.pop("members") + return pd.DataFrame([info]) + +def get_messages(file_paths: list[str]) -> pd.DataFrame: + """ + Convert all of the JSON messages into a DataFrame + + Parameters: + - `file_paths` - the `Path` of `*.json` files + + Output: + - DataFrame with all of the messages + """ + raw_data = [] + for file_path in file_paths: + with open(file_path, "r") as f: + data: dict = json.load(f) + + raw_data.append(pd.DataFrame(data["messages"])) + + data = pd.concat(raw_data, ignore_index=True) + for column in data.columns: + if not column.endswith("timestamp"): + continue + data[column] = pd.to_datetime(data[column], unit="s") + + return data.copy() + + +def upload(data: dict[str, pd.DataFrame], connector: Postgres): + """ + Upload the raw data to Postgres. + + Parameters: + - `data` - the raw concatenated data + - `connector` - initialized Postgres connection + """ + for data_key, table_name in constants.CONFIG["postgres"]["tables"].items(): + df = data[data_key] + if len(df) == 0: + continue + + df = df.assign( + upload_timestamp = datetime.datetime.now() + ) + + json_columns = [ + column + for column in df.columns + if isinstance(df[column].dropna().reset_index(drop=True).iloc[0], (dict, list)) + ] + connector.insert(df, table_name, constants.CONFIG["postgres"]["schema"], json_columns) + +if __name__ == "__main__": + + completed = [] + path = Path(constants.UPLOAD_PATH) + load_dotenv() + connector = connector = Postgres( + username = os.getenv("POSTGRES_USERNAME"), + password = os.getenv("POSTGRES_PASSWORD"), + host = os.getenv("POSTGRES_HOST"), + database = os.getenv("POSTGRES_DATABASE"), + port = int(os.getenv("POSTGRES_PORT")) + ) + # NOTE: the keys are the same as in config.yaml -> postgres.tables + data = { + "members": [], + "community": [], + } + + for file_path in path.rglob("*.pkl"): + data["community"].append(get_community_info(file_path)) + data["members"].append(get_community_members(file_path)) + completed.append(file_path) + + for key, value in data.items(): + data[key] = pd.concat(value, ignore_index=True) + + file_paths = list(path.rglob("*.json")) + data["messages"] = get_messages(file_paths) + completed += file_paths + + upload(data, connector) + connector.close() + + if completed: + for file_path in completed: + os.remove(file_path) \ No newline at end of file