Python: Add telemetry for Python analysis version

Here' `python_analysis_version` is the version of Python that we are
analysing the code as. In practice, all we care about is the major
version, but we might as well include the full thing (since it can be
overridden on the command line).

The `python_runtime_version` is the actual version of Python that ran
the extractor.

Author: tausbn

python/extractor/cli-integration-test/writing-diagnostics/diagnostics.expected

@@ -161,3 +161,23 @@
     "telemetry": true
   }
 }
+{
+  "attributes": {
+    "extractor_version": "7.1.10",
+    "python_analysis_version": "3.12",
+    "python_runtime_version": "3.12.3"
+  },
+  "markdownMessage": "Internal telemetry for the Python extractor.\n\nNo action needed.",
+  "severity": "note",
+  "source": {
+    "extractorName": "python",
+    "id": "py/extractor/summary",
+    "name": "Python extractor telemetry"
+  },
+  "timestamp": "2026-09-01T13:41:33.056818Z",
+  "visibility": {
+    "cliSummaryTable": false,
+    "statusPage": false,
+    "telemetry": true
+  }
+}

python/extractor/semmle/logging.py

@@ -8,6 +8,9 @@
 import multiprocessing
 import enum
 import datetime
+import platform
+
+from semmle.util import VERSION, get_analysis_version
 
 
 #Use standard Semmle logging levels
@@ -355,6 +358,15 @@ def with_timestamp(self, timestamp):
         self.timestamp = timestamp
         return self
 
+def extractor_telemetry_message():
+    return (DiagnosticMessage(Source("py/extractor/summary", "Python extractor telemetry"), Severity.NOTE)
+            .markdown("Internal telemetry for the Python extractor.\n\nNo action needed.")
+            .attribute("python_analysis_version", get_analysis_version())
+            .attribute("python_runtime_version", platform.python_version())
+            .attribute("extractor_version", VERSION)
+            .telemetry()
+    )
+
 def get_stack_trace_lines():
     """Creates a stack trace for inclusion into the `attributes` part of a diagnostic message.
     Limits the size of the stack trace to 5000 characters, so as to not make the SARIF file overly big.

python/extractor/semmle/util.py

@@ -10,7 +10,7 @@
 
 #Semantic version of extractor.
 #Update this if any changes are made
-VERSION = "7.1.9"
+VERSION = "7.1.10"
 
 PY_EXTENSIONS = ".py", ".pyw"
 

python/extractor/semmle/worker.py

@@ -10,7 +10,7 @@
 from semmle.extractors import SuperExtractor, ModulePrinter, SkippedBuiltin
 from semmle.profiling import get_profiler
 from semmle.path_rename import renamer_from_options_and_env
-from semmle.logging import WARN, recursion_error_message, internal_error_message, Logger
+from semmle.logging import WARN, recursion_error_message, internal_error_message, extractor_telemetry_message, Logger
 from semmle.util import FileExtractable, FolderExtractable
 
 class ExtractorFailure(Exception):
@@ -239,6 +239,12 @@ def _drain_queue(queue):
         #Emptied queue as best we can.
         pass
 
+def _write_extractor_telemetry(diagnostics_writer, logger: Logger):
+    try:
+        diagnostics_writer.write(extractor_telemetry_message())
+    except OSError as ex:
+        logger.warning("Failed to write extractor telemetry: %s", ex)
+
 class DiagnosticsWriter(object):
     def __init__(self, proc_id):
         self.proc_id = proc_id
@@ -276,6 +282,8 @@ def _extract_loop(proc_id, queue, trap_dir, archive, options, reply_queue, logge
         reply_queue.put(("INTERRUPT", None, None))
         sys.exit(2)
     logger.set_process_id(proc_id)
+    if write_global_data:
+        _write_extractor_telemetry(diagnostics_writer, logger)
     try:
         if options.trace_only:
             extractor = ModulePrinter(options, trap_dir, archive, renamer, logger)

python/extractor/tests/test_diagnostics.py

@@ -0,0 +1,54 @@
+import platform
+
+from semmle import logging
+from semmle import util
+from semmle import worker
+
+
+def test_extractor_telemetry_message(mocker):
+    mocker.patch("semmle.logging.get_analysis_version", return_value="3.13")
+
+    message = logging.extractor_telemetry_message().to_dict()
+    message.pop("timestamp")
+
+    assert message == {
+        "source": {
+            "id": "py/extractor/summary",
+            "name": "Python extractor telemetry",
+            "extractorName": "python",
+        },
+        "severity": "note",
+        "markdownMessage": "Internal telemetry for the Python extractor.\n\nNo action needed.",
+        "visibility": {
+            "statusPage": False,
+            "cliSummaryTable": False,
+            "telemetry": True,
+        },
+        "attributes": {
+            "python_analysis_version": "3.13",
+            "python_runtime_version": platform.python_version(),
+            "extractor_version": util.VERSION,
+        },
+    }
+
+
+def test_write_extractor_telemetry(mocker):
+    diagnostics_writer = mocker.Mock()
+    logger = mocker.Mock()
+
+    worker._write_extractor_telemetry(diagnostics_writer, logger)
+
+    diagnostics_writer.write.assert_called_once()
+    logger.warning.assert_not_called()
+
+
+def test_write_extractor_telemetry_handles_io_error(mocker):
+    diagnostics_writer = mocker.Mock()
+    diagnostics_writer.write.side_effect = OSError("write failed")
+    logger = mocker.Mock()
+
+    worker._write_extractor_telemetry(diagnostics_writer, logger)
+
+    logger.warning.assert_called_once_with(
+        "Failed to write extractor telemetry: %s", diagnostics_writer.write.side_effect
+    )