implemented endpoints insights_tarb and insights_wiki

Author: mojomonger

docs/release_notes/release_notes.4.4.12.md

@@ -0,0 +1,3 @@
+### IARI version 4.4.6
+
+- Added exturls to insights endpoint, which is insights_webrx internally

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "Internet Archive Reference Inventory (IARI)"
-version = "4.4.10"
+version = "4.4.12"
 description = "API capable of fetching, extracting, transforming and storing reference information from Wikipedia articles, websites and PDFs as structured data."
 authors = [
   "Chris Lombardi <mojomonger@archive.org>",

src/__init__.py

@@ -25,8 +25,9 @@
 # # new stuff aug 2025
 # from src.views.v2.refs_lookup_v2 import GetArchiveV2
 
-# # new stuff oct 2025
-# from src.views.v2.insights_tarb_v2 import InsightsTarbV2
+# new stuff oct 2025
+from src.views.v2.insights_wiki_v2 import InsightsWikiV2
+from src.views.v2.insights_tarb_v2 import InsightsTarbV2
 # new stuff jun 2025
 from src.views.v2.refs_lookup_v2 import RefsLookupV2
 # new stuff mar 2025
@@ -145,7 +146,8 @@ def favicon():
 # bigger picture
 api.add_resource(RefsLookupV2, "/refs_lookup")      # James' Wiki Citations Database
 api.add_resource(InsightsWebRxV2, "/insights")      # Stephen's numbers
-# api.add_resource(InsightsTarbV2, "/tarb_insights")  # Sawood's numbers
+api.add_resource(InsightsTarbV2, "/tarb_insights")  # Sawood's numbers
+api.add_resource(InsightsWikiV2, "/wiki_insights")  # Wiki numbers (from stephen's wiki endpoint)
 
 # Other
 api.add_resource(ArticleCacheV2, "/article_cache")  # for offline article fetching

src/helpers/wiki_utils.py

@@ -0,0 +1,94 @@
+# probe_utils.py
+import requests
+from typing import Optional, Union
+
+from src.helpers.cache_utils import get_cache, set_cache, is_cached, CacheType
+
+from src.constants.constants import ProbeMethod
+from src.models.v2.probes.probe_test import ProbeTest
+from src.models.v2.probes.probe_trust_project import ProbeTrustProject
+from src.models.v2.probes.probe_verifyi import ProbeVerifyi
+
+
+class WikiUtils:
+
+    @staticmethod
+    def run_wiki_logic(data):
+        # complex logic here
+        return {"result": "ok"}
+
+
+
+    @staticmethod
+    def get_exturls():
+        """
+        returns wiki stats for external urls
+        returns {
+            stuff, stuff, stuff
+        }
+
+        """
+        from src import app
+        app.logger.debug(f"==> get_exturls_data")
+
+        results = {}
+        from_cache = False
+
+        # fetch external url data
+        results = WikiUtils.get_exturls_data()
+
+        # set cached data...but not for now
+
+        return {
+            "external_urls": results,
+        }
+
+
+    def get_exturls_data():
+
+        results = {}
+
+        user_agent = "IARI, see https://github.com/internetarchive/iari"
+        headers = {
+            # "Content-Type": "application/x-www-form-urlencoded",
+            "User-Agent": user_agent
+        }
+
+        # do "assess" endpoint
+        exturls_api_url = 'https://commons.wikimedia.org/w/index.php?title=Data%3AWikipedia%5Fstatistics%2Fexturls%2Etab&action=raw'
+
+        # TODO do we need to clean url param here?
+        response = requests.get(
+            exturls_api_url,
+            headers=headers,
+            # json={'url': url}
+        )
+
+        if response.status_code == 200:
+            data = response.json()
+            # results['raw'] = data
+            results = data
+
+        else:
+            # append error to errors array
+            msg = (
+                f"Error fetching Wiki external urls data"
+                f" Got {response.status_code} from {exturls_api_url}"
+                f" Text: {response.text}"
+            )
+
+            from src import app
+            app.logger.debug(msg)
+
+            results.setdefault('errors', []).append(msg)  # create errors entry if not there and append msg
+
+        return results
+
+
+    if __name__ == "__main__":
+        # Example command line test
+        wiki = WikiUtils()
+        test_url = "https://example.com"
+        result = wiki.get_exturls(test_url)
+        print(f"Test result for {test_url}:")
+        print(result)

src/models/v2/job/insights_tarb_job_v2.py

@@ -0,0 +1,22 @@
+from src import MissingInformationError
+from src.models.v2.job import JobV2
+
+
+class InsightsTarbJobV2(JobV2):
+    """job that supports InsightsWebRxV2 endpoint"""
+
+    # using marshmallow to describe parameters
+
+    date_start: str = ""
+    date_end: str = ""
+
+
+    def validate_fields(self):
+        """
+        parameter checking here, if any
+        """
+
+        pass
+
+
+

src/models/v2/job/insights_wiki_job_v2.py

@@ -0,0 +1,22 @@
+from src import MissingInformationError
+from src.models.v2.job import JobV2
+
+
+class InsightsWikiJobV2(JobV2):
+    """job that supports InsightsWebRxV2 endpoint"""
+
+    # using marshmallow to describe parameters
+
+    date_start: str = ""
+    date_end: str = ""
+
+
+    def validate_fields(self):
+        """
+        parameter checking here, if any
+        """
+
+        pass
+
+
+

src/models/v2/schema/insights_tarb_schema_v2.py

@@ -0,0 +1,28 @@
+from marshmallow import fields, pre_load, post_load
+
+from src.models.v2.job.insights_tarb_job_v2 import InsightsTarbJobV2
+from src.models.v2.schema import BaseSchemaV2
+
+
+class InsightsTarbSchemaV2(BaseSchemaV2):
+    # Defines expected parameters for endpoint "insights"
+    #   - default parameters are defined in BaseSchemaV2
+
+    date_start = fields.Str(default=None, required=False)
+    date_end = fields.Str(default=None, required=False)
+
+    # noinspection PyUnusedLocal
+    @post_load
+    # NB: post_load is a marshmallow directive;
+    #   this function is run after loading request args
+    #   it basically pulls the request object value into a Job object
+    #
+    #  **kwargs is needed here despite what the validator claims
+    def return_job_object(self, data, **kwargs) -> InsightsTarbJobV2:  # type: ignore # dead: disable
+        """Return Job object"""
+        job = InsightsTarbJobV2(**data)
+        job.validate_fields()
+
+        # NB here is where we can modify job field values before returning if we want
+
+        return job

src/models/v2/schema/insights_wiki_schema_v2.py

@@ -0,0 +1,32 @@
+from marshmallow import fields, pre_load, post_load
+
+from src.models.v2.job.insights_wiki_job_v2 import InsightsWikiJobV2
+from src.models.v2.schema import BaseSchemaV2
+
+
+class InsightsWikiSchemaV2(BaseSchemaV2):
+    # Defines expected parameters for endpoint "insights"
+    #   - default parameters are defined in BaseSchemaV2
+
+    date_start = fields.Str(default=None, required=False)
+    date_end = fields.Str(default=None, required=False)
+
+    # noinspection PyUnusedLocal
+    @post_load
+    # NB: post_load is a marshmallow directive;
+    #   this function is run after loading request args
+    #   it basically pulls the request object value into a Job object
+    #
+    #  **kwargs is needed here despite what the validator claims
+    def return_job_object(self, data, **kwargs) -> InsightsWikiJobV2:  # type: ignore # dead: disable
+        """Return Job object"""
+        from src import app
+        app.logger.debug("==> InsightsWikiJobV2::@post_load:return_job_object")
+        app.logger.debug(f"return_job_object data: {data}")
+
+        job = InsightsWikiJobV2(**data)
+        job.validate_fields()
+
+        # NB here is where we can modify job field values before returning if we want
+
+        return job

src/views/v2/insights_tarb_v2.py

@@ -0,0 +1,136 @@
+from typing import Any, Optional, Tuple, List, Dict
+import traceback
+import time
+
+import requests
+from bs4 import BeautifulSoup, NavigableString
+from flask import request
+
+from src.helpers.get_version import get_poetry_version
+
+from src.models.exceptions import MissingInformationError, WikipediaApiFetchError
+from src.models.wikimedia.enums import RequestMethods
+from src.views.v2.statistics import StatisticsViewV2
+
+from src.models.v2.job.insights_tarb_job_v2 import InsightsTarbJobV2
+from src.models.v2.schema.insights_tarb_schema_v2 import InsightsTarbSchemaV2
+
+
+class InsightsTarbV2(StatisticsViewV2):
+
+    """
+    returns IABot statistical data
+    """
+
+    schema = InsightsTarbSchemaV2()  # Defines expected parameters; Overrides StatisticsViewV2's "schema" property
+    job: InsightsTarbJobV2           # Holds usable variables, seeded from schema. Overrides StatisticsViewV2's "job"
+
+    return_data: Dict[str, Any] = {}  # holds parsed data from data processing
+    execution_errors: List[Dict[str, Any]] = None
+
+    def get(self):
+        """
+        flask entrypoint for GET
+        must return a tuple: (Any, response_code)
+        """
+        from src import app
+        app.logger.debug(f"==> InsightsTarbV2::get")
+
+        return self.__process_request__(method=RequestMethods.get)
+
+
+    def __process_request__(self, method=RequestMethods.post):  # default to POST method
+
+        from src import app
+        app.logger.debug(f"==> InsightsTarbV2::__process_request__, method = {method}")
+
+        # Start the timer
+        start_time = time.time()
+
+        # fetch the insight data
+        try:
+
+            # validate and setup params
+            self.__validate_and_get_job__(method)  # inherited/subclassed from StatisticsViewV2
+
+            # fetch the data, parse and return summary
+            insight_data = self.__get_insight_data__()
+
+            # Stop the timer and calculate execution time
+            end_time = time.time()
+            execution_time = end_time - start_time
+
+            self.return_data = {
+                "iari_version": get_poetry_version("pyproject.toml"),
+                "iari_command": "tarb_insights",
+                "endpoint": request.endpoint,
+                "execution_time": f"{execution_time:.4f} seconds",
+                "execution_errors": self.execution_errors,
+            }
+
+            self.return_data.update(insight_data)
+
+            return self.return_data, 200
+
+
+        except MissingInformationError as e:
+            traceback.print_exc()
+            return {"error": f"Missing Information Error: {str(e)}"}, 500
+
+        except Exception as e:
+            traceback.print_exc()
+            return {"error": f"General Error: {str(e)}"}, 500
+
+
+    def __get_insight_data__(self):
+        """
+        grabs appropriate data regarding media updates
+        """
+
+        # soup = self.__get_stats_soup__()
+        #
+        # table_names = self.__get_table_names__(soup)
+        # table_list = self.__get_all_tables__(soup, table_names)
+        # table_totals = self.__get_table_totals__(table_list)
+        #
+        # return {
+        #     "table_names": table_names,
+        #     "table_totals": table_totals,
+        #     "tables": table_list
+        # }
+
+
+        """
+        format of each returned record:
+        
+        {
+            "Wiki": "afwiki",
+            "Timestamp": "2021-08-14 00:00:00",
+            "TotalEdits": 2709,
+            "TotalLinks": 3543,
+            "ReactiveEdits": 1620,
+            "ProactiveEdits": 90,
+            "DeadEdits": 278,
+            "UnknownEdits": 721,
+            "LiveLinks": 163,
+            "DeadLinks": 2640,
+            "TagLinks": 740,
+            "UnknownLinks": 0
+        },
+        """
+
+        tarb_api_url = "https://iabot.wmcloud.org/api.php?action=statistics&format=flat"
+
+        r = requests.get(url_stats_yearly)
+        # TODO use a try/catch here
+
+        if r.status_code != 200:
+            return None
+
+        return {
+            "test_value_1": 1,
+            "test_value_2": 2,
+            "test_value_3": 3,
+        }
+
+