feat(runemetrics_scraper): emit players.banned on ban transition

bases/bot_detector/runemetrics_scraper/core.py

@@ -13,7 +13,11 @@
 )
 from bot_detector.event_queue.core import Queue, QueueProducer
 from bot_detector.event_queue.factory import QueueFactory
-from bot_detector.event_queue.structs import NotFoundStruct, ScrapedStruct
+from bot_detector.event_queue.structs import (
+    NotFoundStruct,
+    PlayerBannedStruct,
+    ScrapedStruct,
+)
 from bot_detector.proxy_manager import ProxyManager
 from bot_detector.proxy_manager import Settings as ProxySettings
 from bot_detector.retry_tracker import RetryTracker
@@ -191,6 +195,7 @@ async def work(
     rate_limiter: RateLimiter,
     player_nf_queue: Queue[NotFoundStruct],
     player_sc_producer: QueueProducer[ScrapedStruct],
+    player_banned_producer: QueueProducer[PlayerBannedStruct],
 ):
     retry_tracker = RetryTracker(base_delay=10.0, max_delay=300.0, decay_window=300.0)
     async with ClientSession() as session:
@@ -257,10 +262,25 @@ async def work(
                 continue
 
             # update player data
+            old_label = player_data.label_jagex
             player_data = await update_player(
                 player_data=player_data,
                 runemetrics_response=runemetrics_response,
             )
+
+            # emit a ban event only on the transition into label_jagex = 2
+            if old_label != 2 and player_data.label_jagex == 2:
+                banned_event = PlayerBannedStruct(
+                    metadata=MetaData(version=1, source="runemetrics_scraper"),
+                    player_id=player_data.id,
+                    name=player_data.name,
+                )
+                produce_error = await player_banned_producer.put([banned_event])
+                if produce_error:
+                    logger.error(
+                        f"[{worker_id}]: Failed to produce banned player: {produce_error}"
+                    )
+
             # create hiscore data
             try:
                 scraped_data = ScrapedStruct(
@@ -338,15 +358,30 @@ async def main():
             ),
         ),
     )
-    for queue in (player_nf_queue, player_sc_producer):
+    player_banned_producer = QueueFactory.create_queue(
+        model=PlayerBannedStruct,
+        queue_type="producer",
+        backend_type="kafka",
+        config=KafkaConfig(
+            topic="players.banned",
+            bootstrap_servers=b_server,
+            producer=True,
+            producer_config=KafkaProducerConfig(
+                partition_key_fn=lambda message: str(message.player_id % 10)
+            ),
+        ),
+    )
+    for queue in (player_nf_queue, player_sc_producer, player_banned_producer):
         if isinstance(queue, Exception):
             raise queue
 
     assert isinstance(player_nf_queue, Queue)
     assert isinstance(player_sc_producer, QueueProducer)
+    assert isinstance(player_banned_producer, QueueProducer)
 
     await player_nf_queue.start()
     await player_sc_producer.start()
+    await player_banned_producer.start()
 
     # start workers
     workers = [
@@ -360,6 +395,7 @@ async def main():
                 ),
                 player_nf_queue=player_nf_queue,
                 player_sc_producer=player_sc_producer,
+                player_banned_producer=player_banned_producer,
             )
         )
         for worker_id in range(len(proxies))

test/bases/bot_detector/runemetrics_scraper/test_core.py

@@ -112,6 +112,7 @@ async def test_work_commits_offset_after_successful_handle(
             rate_limiter=AsyncMock(),
             player_nf_queue=player_nf_queue,
             player_sc_producer=player_sc_producer,
+            player_banned_producer=AsyncMock(),
         )
 
     player_nf_queue.commit.assert_awaited_once()
@@ -161,6 +162,7 @@ async def test_work_commits_only_after_successful_requeue(
             rate_limiter=AsyncMock(),
             player_nf_queue=player_nf_queue,
             player_sc_producer=player_sc_producer,
+            player_banned_producer=AsyncMock(),
         )
 
     assert player_nf_queue.commit.await_count == expected_commit_calls
@@ -252,6 +254,7 @@ async def test_work_backoff_increases_across_consecutive_errors(
             rate_limiter=AsyncMock(),
             player_nf_queue=player_nf_queue,
             player_sc_producer=player_sc_producer,
+            player_banned_producer=AsyncMock(),
         )
 
     delays = [c[0][0] for c in sleep_mock.call_args_list]
@@ -319,6 +322,7 @@ async def test_work_backoff_resets_after_success(
             rate_limiter=AsyncMock(),
             player_nf_queue=player_nf_queue,
             player_sc_producer=player_sc_producer,
+            player_banned_producer=AsyncMock(),
         )
 
     delays = [c[0][0] for c in sleep_mock.call_args_list]
@@ -375,7 +379,79 @@ async def test_work_validation_error_requeues_without_backoff(
             rate_limiter=AsyncMock(),
             player_nf_queue=player_nf_queue,
             player_sc_producer=player_sc_producer,
+            player_banned_producer=AsyncMock(),
         )
 
     player_nf_queue.put.assert_awaited_once()
     sleep_mock.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "old_label,scrape_error,should_emit",
+    [
+        (1, "NOT_A_MEMBER", True),
+        (0, "NOT_A_MEMBER", True),
+        (2, "NOT_A_MEMBER", False),
+        (0, None, False),
+        (0, "NO_PROFILE", False),
+        (1, None, False),
+    ],
+)
+async def test_work_emits_banned_event_only_on_transition_into_banned(
+    monkeypatch: pytest.MonkeyPatch,
+    player_struct: PlayerStruct,
+    old_label: int,
+    scrape_error: str | None,
+    should_emit: bool,
+):
+    player = player_struct.model_copy(update={"label_jagex": old_label})
+    player_message = NotFoundStruct(
+        metadata=MetaData(version=1, source="test"),
+        player_data=player,
+    )
+    player_nf_queue = AsyncMock()
+    player_nf_queue.get_one = AsyncMock(return_value=player_message)
+    player_nf_queue.commit = AsyncMock(return_value=None)
+    player_sc_producer = AsyncMock()
+    player_sc_producer.put = AsyncMock(return_value=None)
+    player_banned_producer = AsyncMock()
+    player_banned_producer.put = AsyncMock(return_value=None)
+
+    runemetrics_response = RuneMetricsResponse()
+    if scrape_error is not None:
+        runemetrics_response.error = RuneMetricsError(
+            error=scrape_error,
+            loggedIn=False,
+        )
+
+    monkeypatch.setattr(core, "ClientSession", _DummySession)
+    monkeypatch.setattr(
+        core,
+        "get_proxy",
+        AsyncMock(side_effect=["http://user@proxy", asyncio.CancelledError()]),
+    )
+    monkeypatch.setattr(
+        core,
+        "scrape_player",
+        AsyncMock(return_value=(runemetrics_response, 0.1, None)),
+    )
+    monkeypatch.setattr(core.asyncio, "sleep", AsyncMock(return_value=None))
+
+    with pytest.raises(asyncio.CancelledError):
+        await core.work(
+            worker_id=1,
+            proxy_manager=AsyncMock(),
+            rate_limiter=AsyncMock(),
+            player_nf_queue=player_nf_queue,
+            player_sc_producer=player_sc_producer,
+            player_banned_producer=player_banned_producer,
+        )
+
+    if should_emit:
+        player_banned_producer.put.assert_awaited_once()
+        event = player_banned_producer.put.call_args.args[0][0]
+        assert event.player_id == player.id
+        assert event.name == player.name
+    else:
+        player_banned_producer.put.assert_not_awaited()