backend.workers.datasource_metrics

  1"""
  2Calculate various 4CAT metrics
  3
  4Two types of metrics are currently calculated:
  5
  6- General metrics. Currently this is mostly the total dataset size, which is
  7  useful to know for admins
  8- Datasource metrics. This is used for both processors (e.g. to calculate
  9  relative) and to show how many posts a local datasource contains.
 10"""
 11import os
 12
 13from datetime import datetime, time, timezone
 14
 15from backend.lib.worker import BasicWorker
 16
 17
 18class DatasourceMetrics(BasicWorker):
 19    """
 20    Calculate metrics
 21
 22    This will be stored in a separate PostgreSQL table.
 23    """
 24    type = "datasource-metrics"
 25    max_workers = 1
 26
 27    @classmethod
 28    def ensure_job(cls, config=None):
 29        """
 30        Ensure that the datasource metrics worker is always running
 31
 32        This is used to ensure that the datasource metrics worker is always
 33        running, and if it is not, it will be started by the WorkerManager.
 34
 35        :return:  Job parameters for the worker
 36        """
 37        return {"remote_id": "localhost", "interval": 40000}
 38
 39    def work(self):
 40        self.general_stats()
 41        self.data_stats()
 42
 43    @staticmethod
 44    def folder_size(path='.'):
 45        """
 46        Get the size of a folder using os.scandir for efficiency
 47        """
 48        total = 0
 49        for entry in os.scandir(path):
 50            if entry.is_file():
 51                try:
 52                    total += entry.stat().st_size
 53                except FileNotFoundError:
 54                    # If the file was removed while scanning, skip it
 55                    continue
 56            elif entry.is_dir():
 57                total += DatasourceMetrics.folder_size(entry.path)
 58        return total
 59
 60    def general_stats(self):
 61        """
 62        Calculate general 4CAT stats
 63
 64        These sizes can be very slow to calculate, which is why we do it in
 65        this worker instead of on demand.
 66        """
 67        metrics = {
 68            "size_data": DatasourceMetrics.folder_size(self.config.get("PATH_DATA")),
 69            "size_logs": DatasourceMetrics.folder_size(self.config.get("PATH_LOGS")),
 70            "size_db": self.db.fetchone("SELECT pg_database_size(%s) AS num", (self.config.get("DB_NAME"),))["num"]
 71        }
 72
 73        for metric, value in metrics.items():
 74            self.db.upsert("metrics", {
 75                "metric": metric,
 76                "count": value,
 77                "datasource": "4cat",
 78                "board": "",
 79                "date": "now"
 80            }, constraints=["metric", "datasource", "board", "date"])
 81
 82    def data_stats(self):
 83        """
 84        Go through all local datasources, and update the posts per day
 85        if they haven't been calculated yet. These data can then be used
 86        to calculate e.g. posts per month.
 87        :return:
 88        """
 89
 90        # Get a list of all database tables
 91        all_tables = [row["tablename"] for row in self.db.fetchall(
 92            "SELECT tablename FROM pg_catalog.pg_tables WHERE schemaname != 'pg_catalog' AND schemaname != 'information_schema';")]
 93
 94        # Check if the metrics table is already present
 95        metrics_exists = True if "metrics" in all_tables else False
 96
 97        # If not, make it.
 98        if not metrics_exists:
 99            self.db.execute("""
100				CREATE TABLE IF NOT EXISTS metrics (
101				  metric             text,
102				  datasource         text,
103				  board              text,
104				  date               text,
105				  count              integer
106				);
107
108			""")
109
110        added_datasources = [row["datasource"] for row in self.db.fetchall("SELECT DISTINCT(datasource) FROM metrics")]
111        enabled_datasources = self.config.get("datasources.enabled", {})
112
113        for datasource_id in self.modules.datasources:
114            if datasource_id not in enabled_datasources:
115                continue
116
117            datasource = self.modules.workers.get(datasource_id + "-search")
118            if not datasource:
119                continue
120
121            # Database IDs may be different from the Datasource ID (e.g. the datasource "4chan" became "fourchan" but the database ID remained "4chan")
122            database_db_id = datasource.prefix if hasattr(datasource, "prefix") else datasource_id
123
124            is_local = True if hasattr(datasource, "is_local") and datasource.is_local else False
125            is_static = True if hasattr(datasource, "is_static") and datasource.is_static else False
126
127            # Only update local datasources
128            if is_local:
129
130                # Some translating..
131                settings_id = datasource_id
132                if datasource_id == "4chan":
133                    settings_id = "fourchan"
134                elif datasource_id == "8chan":
135                    settings_id = "eightchan"
136
137                boards = [b for b in self.config.get(settings_id + "-search.boards", [])]
138
139                # If a datasource is static (so not updated) and it
140                # is already present in the metrics table, we don't
141                # need to update its metrics anymore.
142                if is_static and datasource_id in added_datasources:
143                    continue
144                else:
145
146                    # -------------------------
147                    #   Posts per day metric
148                    # -------------------------
149
150                    # Get the name of the posts table for this datasource
151                    posts_table = datasource_id if "posts_" + database_db_id not in all_tables else "posts_" + database_db_id
152
153                    # Count and update for every board individually
154                    for board in boards:
155
156                        if not board:
157                            board_sql = " board = '' OR board = NULL"
158                        else:
159                            board_sql = " board='" + board + "'"
160
161                        # Midnight of this day in UTC epoch timestamp
162                        midnight = int(
163                            datetime.combine(datetime.today(), time.min).replace(tzinfo=timezone.utc).timestamp())
164
165                        # We only count passed days
166                        time_sql = "timestamp < " + str(midnight)
167
168                        # If the datasource is dynamic, we also only update days
169                        # that haven't been added yet - these are heavy queries.
170                        if not is_static:
171                            days_added = self.db.fetchall(
172                                "SELECT date FROM metrics WHERE datasource = '%s' AND board = '%s' AND metric = 'posts_per_day';" % (
173                                database_db_id, board))
174
175                            if days_added:
176
177                                last_day_added = max([row["date"] for row in days_added])
178                                last_day_added = datetime.strptime(last_day_added, '%Y-%m-%d').replace(
179                                    tzinfo=timezone.utc)
180
181                                # If the last day added is today, there's no need to update yet
182                                if last_day_added.date() == datetime.today().replace(tzinfo=timezone.utc).date():
183                                    self.log.info(
184                                        "No new posts per day to count for %s%s" % (datasource_id, "/" + board))
185                                    continue
186
187                                # Change to UTC epoch timestamp for postgres query
188                                after_timestamp = int(last_day_added.timestamp())
189
190                                time_sql += " AND timestamp > " + str(after_timestamp) + " "
191
192                        self.log.info(
193                            "Calculating metric posts_per_day for datasource %s%s" % (datasource_id, "/" + board))
194
195                        # Get those counts
196                        query = """
197							SELECT 'posts_per_day' AS metric, '%s' AS datasource, board, to_char(to_timestamp(timestamp), 'YYYY-MM-DD') AS date, count(*)COUNT
198							FROM %s
199							WHERE %s AND %s
200							GROUP BY metric, datasource, board, date;
201							""" % (database_db_id, posts_table, board_sql, time_sql)
202                        # Add to metrics table
203                        rows = [dict(row) for row in self.db.fetchall(query)]
204
205                        if rows:
206                            for row in rows:
207                                self.db.upsert("metrics", row, constraints=["metric", "datasource", "board", "date"])
208
209                # -------------------------------
210                #   no other metrics added yet
211                # -------------------------------
212
213        self.job.finish()