HoloTech/twitch-eventtracker-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add channel notification event tracking.

Browse Source 
Bumps schema version to 2.
Adds Profiles module version dep on schema init.
Add notice_events, notice_sub_events, and notice_resub_events tables.
Add corresponding event manager.
Add ChatNotificationWithPayload event payload extension.
Add ChannelNotification subscription.
This commit is contained in:
Interitio
2026-07-22 12:21:16 +03:00
parent 7744aaab0e
commit dd6fc013ee
4 changed files with 304 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tracker/component.py
+203 -85
View File
@@ -1,8 +1,10 @@
from typing import Optional
import random
import twitchio
from twitchio import PartialUser, Scopes, eventsub
from twitchio.ext import commands as cmds
from psycopg.types.json import Jsonb
from botdata import BotChannel
from meta import Bot
@@ -10,6 +12,7 @@ from utils.lib import utc_now
from . import logger
from .data import EventData, TrackingChannel
from .payloads import ChatNotificationWithPayload
class TrackerComponent(cmds.Component):
@@ -32,43 +35,50 @@ class TrackerComponent(cmds.Component):
# ----- Methods -----
async def start_tracking(self, channel: TrackingChannel):
# TODO: Make sure that we aren't trying to make duplicate subscriptions here
logger.debug(
"Initialising event tracking for %s",
channel.userid
)
logger.debug("Initialising event tracking for %s", channel.userid)
# Get associated auth scopes
rows = await self.bot.data.user_auth_scopes.select_where(userid=channel.userid)
scopes = Scopes([row['scope'] for row in rows])
scopes = Scopes([row["scope"] for row in rows])
# Build subscription payloads based on available scopes
subs = []
usersubs = []
subcls = []
if Scopes.channel_read_redemptions in scopes or Scopes.channel_manage_redemptions in scopes:
if (
Scopes.channel_read_redemptions in scopes
or Scopes.channel_manage_redemptions in scopes
):
subcls.append(eventsub.ChannelPointsRedeemAddSubscription)
subcls.append(eventsub.ChannelPointsRedeemUpdateSubscription)
if Scopes.bits_read in scopes:
subcls.append(eventsub.ChannelBitsUseSubscription)
subcls.append(eventsub.ChannelCheerSubscription)
if Scopes.channel_read_subscriptions in scopes:
subcls.extend((
eventsub.ChannelSubscribeSubscription,
eventsub.ChannelSubscribeMessageSubscription,
eventsub.ChannelSubscriptionGiftSubscription,
))
subcls.extend(
(
eventsub.ChannelSubscribeSubscription,
eventsub.ChannelSubscribeMessageSubscription,
eventsub.ChannelSubscriptionGiftSubscription,
)
)
if Scopes.channel_read_polls in scopes or Scopes.channel_manage_polls in scopes:
subcls.append(eventsub.ChannelPollEndSubscription)
if Scopes.channel_read_vips in scopes or Scopes.channel_manage_vips in scopes:
subcls.extend((
eventsub.ChannelVIPAddSubscription,
eventsub.ChannelVIPRemoveSubscription,
))
subcls.extend(
(
eventsub.ChannelVIPAddSubscription,
eventsub.ChannelVIPRemoveSubscription,
)
)
subcls.extend((
eventsub.StreamOnlineSubscription,
eventsub.StreamOfflineSubscription,
eventsub.ChannelUpdateSubscription,
))
subcls.extend(
(
eventsub.StreamOnlineSubscription,
eventsub.StreamOfflineSubscription,
eventsub.ChannelUpdateSubscription,
eventsub.ChannelNotificationSubscription,
)
)
for subbr in subcls:
subs.append(subbr(broadcaster_user_id=channel.userid))
@@ -90,10 +100,14 @@ class TrackerComponent(cmds.Component):
)
)
subs.extend((
eventsub.ChannelRaidSubscription(to_broadcaster_user_id=channel.userid),
eventsub.ChannelRaidSubscription(from_broadcaster_user_id=channel.userid),
))
subs.extend(
(
eventsub.ChannelRaidSubscription(to_broadcaster_user_id=channel.userid),
eventsub.ChannelRaidSubscription(
from_broadcaster_user_id=channel.userid
),
)
)
responses = []
for sub in subs:
@@ -110,25 +124,33 @@ class TrackerComponent(cmds.Component):
if self.bot.using_webhooks:
resp = await self.bot.subscribe_webhook(sub)
else:
resp = await self.bot.subscribe_websocket(sub, token_for=channel.userid, as_bot=False)
resp = await self.bot.subscribe_websocket(
sub, token_for=channel.userid, as_bot=False
)
responses.append(resp)
except Exception:
logger.exception("Failed to subscribe to %s", str(sub))
logger.info("Finished tracker subscription to %s: %s", channel.userid, ', '.join(map(str, responses)))
logger.info(
"Finished tracker subscription to %s: %s",
channel.userid,
", ".join(map(str, responses)),
)
# ----- Events -----
@cmds.Component.listener()
async def event_safe_channel_joined(self, payload: BotChannel):
# Check if the channel is tracked
# Check if the channel is tracked
# If it is, call start_tracking
tracked = await TrackingChannel.fetch(payload.userid)
if tracked and tracked.joined:
await self.start_tracking(tracked)
@cmds.Component.listener()
async def event_custom_redemption_add(self, payload: twitchio.ChannelPointsRedemptionAdd):
async def event_custom_redemption_add(
self, payload: twitchio.ChannelPointsRedemptionAdd
):
tracked = await TrackingChannel.fetch(payload.broadcaster.id)
if tracked and tracked.joined:
community = await self.bot.profiles.fetch_community(payload.broadcaster)
@@ -137,7 +159,7 @@ class TrackerComponent(cmds.Component):
pid = profile.profileid
event_row = await self.data.events.insert(
event_type='redemption_add',
event_type="redemption_add",
communityid=cid,
channel_id=payload.broadcaster.id,
profileid=pid,
@@ -145,7 +167,7 @@ class TrackerComponent(cmds.Component):
occurred_at=payload.redeemed_at,
)
detail_row = await self.data.redemption_add_events.insert(
event_id=event_row['event_id'],
event_id=event_row["event_id"],
redeem_id=payload.reward.id,
redeem_title=payload.reward.title,
redeem_cost=payload.reward.cost,
@@ -155,7 +177,9 @@ class TrackerComponent(cmds.Component):
)
@cmds.Component.listener()
async def event_custom_redemption_update(self, payload: twitchio.ChannelPointsRedemptionUpdate):
async def event_custom_redemption_update(
self, payload: twitchio.ChannelPointsRedemptionUpdate
):
tracked = await TrackingChannel.fetch(payload.broadcaster.id)
if tracked and tracked.joined:
community = await self.bot.profiles.fetch_community(payload.broadcaster)
@@ -164,20 +188,20 @@ class TrackerComponent(cmds.Component):
pid = profile.profileid
event_row = await self.data.events.insert(
event_type='redemption_update',
event_type="redemption_update",
communityid=cid,
channel_id=payload.broadcaster.id,
profileid=pid,
user_id=payload.user.id,
)
detail_row = await self.data.redemption_update_events.insert(
event_id=event_row['event_id'],
event_id=event_row["event_id"],
redeem_id=payload.reward.id,
redeem_title=payload.reward.title,
redeem_cost=payload.reward.cost,
redemption_id=payload.id,
redemption_status=payload.status,
redeemed_at=utc_now()
redeemed_at=utc_now(),
)
@cmds.Component.listener()
@@ -189,19 +213,18 @@ class TrackerComponent(cmds.Component):
profile = await self.bot.profiles.fetch_profile(payload.user)
pid = profile.profileid
# Computer follower count
# Computer follower count
followers = await payload.broadcaster.fetch_followers()
event_row = await self.data.events.insert(
event_type='follow',
event_type="follow",
communityid=cid,
channel_id=payload.broadcaster.id,
profileid=pid,
user_id=payload.user.id,
)
detail_row = await self.data.follow_events.insert(
event_id=event_row['event_id'],
follower_count=followers.total
event_id=event_row["event_id"], follower_count=followers.total
)
@cmds.Component.listener()
@@ -214,20 +237,20 @@ class TrackerComponent(cmds.Component):
pid = profile.profileid
event_row = await self.data.events.insert(
event_type='bits',
event_type="bits",
communityid=cid,
channel_id=payload.broadcaster.id,
profileid=pid,
user_id=payload.user.id,
)
detail_row = await self.data.bits_events.insert(
event_id=event_row['event_id'],
event_id=event_row["event_id"],
bits=payload.bits,
bits_type=payload.type,
message=payload.text,
powerup_type=payload.power_up.type if payload.power_up else None
powerup_type=payload.power_up.type if payload.power_up else None,
)
self.bot.safe_dispatch('bits_use', payload=(event_row, detail_row, payload))
self.bot.safe_dispatch("bits_use", payload=(event_row, detail_row, payload))
@cmds.Component.listener()
async def event_subscription(self, payload: twitchio.ChannelSubscribe):
@@ -239,18 +262,20 @@ class TrackerComponent(cmds.Component):
pid = profile.profileid
event_row = await self.data.events.insert(
event_type='subscribe',
event_type="subscribe",
communityid=cid,
channel_id=payload.broadcaster.id,
profileid=pid,
user_id=payload.user.id,
)
detail_row = await self.data.subscribe_events.insert(
event_id=event_row['event_id'],
event_id=event_row["event_id"],
tier=int(payload.tier),
gifted=payload.gift,
)
self.bot.safe_dispatch('subscription', payload=(event_row, detail_row, payload))
self.bot.safe_dispatch(
"subscription", payload=(event_row, detail_row, payload)
)
@cmds.Component.listener()
async def event_subscription_gift(self, payload: twitchio.ChannelSubscriptionGift):
@@ -265,21 +290,25 @@ class TrackerComponent(cmds.Component):
pid = None
event_row = await self.data.events.insert(
event_type='gift',
event_type="gift",
communityid=cid,
channel_id=payload.broadcaster.id,
profileid=pid,
user_id=payload.user.id if payload.user else None,
)
detail_row = await self.data.gift_events.insert(
event_id=event_row['event_id'],
event_id=event_row["event_id"],
tier=int(payload.tier),
gifted_count=payload.total,
)
self.bot.safe_dispatch('subscription_gift', payload=(event_row, detail_row, payload))
self.bot.safe_dispatch(
"subscription_gift", payload=(event_row, detail_row, payload)
)
@cmds.Component.listener()
async def event_subscription_message(self, payload: twitchio.ChannelSubscriptionMessage):
async def event_subscription_message(
self, payload: twitchio.ChannelSubscriptionMessage
):
tracked = await TrackingChannel.fetch(payload.broadcaster.id)
if tracked and tracked.joined:
community = await self.bot.profiles.fetch_community(payload.broadcaster)
@@ -288,21 +317,23 @@ class TrackerComponent(cmds.Component):
pid = profile.profileid
event_row = await self.data.events.insert(
event_type='subscribe_message',
event_type="subscribe_message",
communityid=cid,
channel_id=payload.broadcaster.id,
profileid=pid,
user_id=payload.user.id,
)
detail_row = await self.data.subscribe_message_events.insert(
event_id=event_row['event_id'],
event_id=event_row["event_id"],
tier=int(payload.tier),
duration_months=payload.months,
cumulative_months=payload.cumulative_months,
streak_months=payload.streak_months,
message=payload.text,
)
self.bot.safe_dispatch('subscription_message', payload=(event_row, detail_row, payload))
self.bot.safe_dispatch(
"subscription_message", payload=(event_row, detail_row, payload)
)
@cmds.Component.listener()
async def event_stream_online(self, payload: twitchio.StreamOnline):
@@ -312,13 +343,13 @@ class TrackerComponent(cmds.Component):
cid = community.communityid
event_row = await self.data.events.insert(
event_type='stream_online',
event_type="stream_online",
communityid=cid,
channel_id=payload.broadcaster.id,
occurred_at=payload.started_at,
)
detail_row = await self.data.stream_online_events.insert(
event_id=event_row['event_id'],
event_id=event_row["event_id"],
stream_id=payload.id,
stream_type=payload.type,
)
@@ -331,12 +362,12 @@ class TrackerComponent(cmds.Component):
cid = community.communityid
event_row = await self.data.events.insert(
event_type='stream_offline',
event_type="stream_offline",
communityid=cid,
channel_id=payload.broadcaster.id,
)
detail_row = await self.data.stream_offline_events.insert(
event_id=event_row['event_id'],
event_id=event_row["event_id"],
)
@cmds.Component.listener()
@@ -352,42 +383,121 @@ class TrackerComponent(cmds.Component):
payload.viewer_count,
)
async def _event_raid_out(self, broadcaster: PartialUser, to_broadcaster: PartialUser, viewer_count: int):
async def _event_raid_out(
self, broadcaster: PartialUser, to_broadcaster: PartialUser, viewer_count: int
):
tracked = await TrackingChannel.fetch(broadcaster.id)
if tracked and tracked.joined:
community = await self.bot.profiles.fetch_community(broadcaster)
cid = community.communityid
event_row = await self.data.events.insert(
event_type='raidout',
event_type="raidout",
communityid=cid,
channel_id=broadcaster.id,
)
detail_row = await self.data.raid_out_events.insert(
event_id=event_row['event_id'],
event_id=event_row["event_id"],
target_id=to_broadcaster.id,
target_name=to_broadcaster.name,
viewer_count=viewer_count
viewer_count=viewer_count,
)
async def _event_raid_in(self, broadcaster: PartialUser, from_broadcaster: PartialUser, viewer_count: int):
async def _event_raid_in(
self, broadcaster: PartialUser, from_broadcaster: PartialUser, viewer_count: int
):
tracked = await TrackingChannel.fetch(broadcaster.id)
if tracked and tracked.joined:
community = await self.bot.profiles.fetch_community(broadcaster)
cid = community.communityid
event_row = await self.data.events.insert(
event_type='raidin',
event_type="raidin",
communityid=cid,
channel_id=broadcaster.id,
)
detail_row = await self.data.raid_in_events.insert(
event_id=event_row['event_id'],
event_id=event_row["event_id"],
source_id=from_broadcaster.id,
source_name=from_broadcaster.name,
viewer_count=viewer_count
viewer_count=viewer_count,
)
@cmds.Component.listener()
async def event_chat_notification(self, payload: ChatNotificationWithPayload):
tracked = await TrackingChannel.fetch(payload.broadcaster.id)
if tracked and tracked.joined:
community = await self.bot.profiles.fetch_community(payload.broadcaster)
cid = community.communityid
profile = await self.bot.profiles.fetch_profile(payload.chatter)
pid = profile.profileid
event_payload = Jsonb(payload._raw)
event_row = await self.data.events.insert(
event_type="notice",
communityid=cid,
channel_id=payload.broadcaster.id,
profileid=pid,
user_id=payload.chatter.id,
event_payload=event_payload,
)
parent_event_id = event_row["event_id"]
detail_row = await self.data.notice_events.insert(
event_id=parent_event_id,
message_id=payload.id,
system_message=payload.system_message,
)
self.bot.safe_dispatch(
"chat_notice", payload=(event_row, detail_row, payload)
)
# Dispatch sub-events
if payload.notice_type == "sub":
event_row = await self.data.events.insert(
event_type="notice_sub",
communityid=cid,
channel_id=payload.broadcaster.id,
profileid=pid,
user_id=payload.chatter.id,
parent_event_id=parent_event_id,
)
detail_row = await self.data.notice_sub_events.insert(
event_id=event_row["event_id"],
tier=int(payload.tier),
is_prime=payload.sub.prime,
duration_months=payload.sub.months,
)
self.bot.safe_dispatch(
"chat_notice_sub", payload=(event_row, detail_row, payload)
)
elif payload.notice_type == "resub":
event_row = await self.data.events.insert(
event_type="notice_resub",
communityid=cid,
channel_id=payload.broadcaster.id,
profileid=pid,
user_id=payload.chatter.id,
parent_event_id=parent_event_id,
)
detail_row = await self.data.notice_resub_events.insert(
event_id=event_row["event_id"],
message_id=payload.id,
tier=payload.resub.tier,
is_prime=payload.resub.prime,
is_gift=payload.resub.gift,
cumulative_months=payload.resub.cumulative_months,
duration_months=payload.resub.months,
streak_months=payload.resub.streak_months,
gifter_is_anonymous=payload.resub.anonymous,
gifter_user_id=(
payload.resub.gifter.id if payload.resub.gifter else None
),
)
self.bot.safe_dispatch(
"chat_notice_resub", payload=(event_row, detail_row, payload)
)
@cmds.Component.listener()
async def event_message(self, payload: twitchio.ChatMessage):
tracked = await TrackingChannel.fetch(payload.broadcaster.id)
@@ -398,64 +508,72 @@ class TrackerComponent(cmds.Component):
pid = profile.profileid
event_row = await self.data.events.insert(
event_type='message',
event_type="message",
communityid=cid,
channel_id=payload.broadcaster.id,
profileid=pid,
user_id=payload.chatter.id,
)
detail_row = await self.data.message_events.insert(
event_id=event_row['event_id'],
event_id=event_row["event_id"],
message_id=payload.id,
message_type=payload.type,
content=payload.text,
source_channel_id=payload.source_id
source_channel_id=payload.source_id,
)
# ----- Commands -----
@cmds.command(name='starttracking')
@cmds.command(name="starttracking")
async def cmd_starttracking(self, ctx: cmds.Context):
if ctx.broadcaster:
tracking = await TrackingChannel.fetch_or_create(ctx.channel.id, joined=True)
tracking = await TrackingChannel.fetch_or_create(
ctx.channel.id, joined=True
)
if not tracking.joined:
await tracking.update(joined=True)
rows = await self.bot.data.user_auth_scopes.select_where(userid=ctx.channel.id)
scopes = Scopes([row['scope'] for row in rows])
rows = await self.bot.data.user_auth_scopes.select_where(
userid=ctx.channel.id
)
scopes = Scopes([row["scope"] for row in rows])
url = self.bot.get_auth_url(
Scopes({
Scopes.channel_read_subscriptions,
Scopes.channel_read_redemptions,
Scopes.bits_read,
Scopes.channel_read_polls,
Scopes.channel_read_vips,
Scopes.moderator_read_followers,
*scopes
})
Scopes(
{
Scopes.channel_read_subscriptions,
Scopes.channel_read_redemptions,
Scopes.bits_read,
Scopes.channel_read_polls,
Scopes.channel_read_vips,
Scopes.moderator_read_followers,
*scopes,
}
)
)
await ctx.reply(
f"Tracking enabled! Please authorise me to track events in this channel: {url}"
)
await ctx.reply(f"Tracking enabled! Please authorise me to track events in this channel: {url}")
else:
await ctx.reply("Only the broadcaster can enable tracking.")
@cmds.command(name='stoptracking')
@cmds.command(name="stoptracking")
async def cmd_stoptracking(self, ctx: cmds.Context):
if ctx.broadcaster:
tracking = await TrackingChannel.fetch(ctx.channel.id)
if tracking and tracking.joined:
await tracking.update(joined=False)
# TODO: Actually disable the subscriptions instead of just on the next restart
# TODO: Actually disable the subscriptions instead of just on the next restart
# This is tricky because some of the subscriptions may have been requested by other modules
# Requires keeping track of the source of subscriptions, and having a central manager disable them when no-one is listening anymore.
pass
await ctx.reply("Event tracking has been disabled.")
else:
await ctx.reply("Event tracking is not enabled!")
else:
await ctx.reply("Only the broadcaster can enable tracking.")
@cmds.command(name='join')
@cmds.command(name="join")
async def cmd_join(self, ctx: cmds.Context):
url = self.bot.get_auth_url()
await ctx.reply(f"Invite me to your channel with: {url}")
tracker/data.py
+25 -21
View File
@@ -3,39 +3,43 @@ from data.columns import String, Timestamp, Integer, Bool
class TrackingChannel(RowModel):
_tablename_ = 'tracking_channels'
_tablename_ = "tracking_channels"
_cache_ = {}
userid = String(primary=True)
joined = Bool
joined = Bool
joined_at = Timestamp()
_timestamp = Timestamp()
class EventData(Registry):
VERSION = ('EVENT_TRACKER', 1)
VERSION = ("EVENT_TRACKER", 2)
tracking_channels = TrackingChannel.table
events = Table('events')
follow_events = Table('follow_events')
bits_events = Table('bits_events')
events = Table("events")
follow_events = Table("follow_events")
bits_events = Table("bits_events")
subscribe_events = Table('subscribe_events')
gift_events = Table('gift_events')
subscribe_message_events = Table('subscribe_message_events')
subscribe_events = Table("subscribe_events")
gift_events = Table("gift_events")
subscribe_message_events = Table("subscribe_message_events")
cheer_events = Table('cheer_events')
redemption_add_events = Table('redemption_add_events')
redemption_update_events = Table('redemption_update_events')
poll_end_events = Table('poll_end_events')
stream_online_events = Table('stream_online_events')
stream_offline_events = Table('stream_offline_events')
channel_update_events = Table('channel_update_events')
cheer_events = Table("cheer_events")
redemption_add_events = Table("redemption_add_events")
redemption_update_events = Table("redemption_update_events")
poll_end_events = Table("poll_end_events")
stream_online_events = Table("stream_online_events")
stream_offline_events = Table("stream_offline_events")
channel_update_events = Table("channel_update_events")
vip_add_events = Table('vip_add_events')
vip_remove_events = Table('vip_remove_events')
vip_add_events = Table("vip_add_events")
vip_remove_events = Table("vip_remove_events")
raid_out_events = Table('raid_out_events')
raid_in_events = Table('raid_in_events')
message_events = Table('message_events')
raid_out_events = Table("raid_out_events")
raid_in_events = Table("raid_in_events")
message_events = Table("message_events")
notice_events = Table("notice_events")
notice_sub_events = Table("notice_sub_events")
notice_resub_events = Table("notice_resub_events")
tracker/payloads.py
+15
View File
@@ -0,0 +1,15 @@
import twitchio
class ChatNotificationWithPayload(twitchio.ChatNotification):
"""
Extends twitchio.ChatNotification to include a _raw field with the original data.
The subclass registry in twitchio.BaseEvent should automatically register this as the correct event handler.
"""
__slots__ = ("_raw",)
def __init__(self, payload: twitchio.ChannelChatNotificationEvent, *args, **kwargs):
super().__init__(payload, *args, **kwargs)
self._raw = payload