secstate/vrobbler
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[videos] Add preliminary Twitch video support

Browse Source
This commit is contained in:
Colin Powell
2026-04-02 11:10:39 -04:00
parent 1bf558938d
commit b6af201ba3
10 changed files with 430 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
vrobbler/apps/scrobbles/constants.py
View File

@ -46,6 +46,7 @@ SCROBBLE_CONTENT_URLS = {
"-p": ["https://www.ipdb.plus/IPDb/puzzle.php?id="],
"-l": ["https://brickset.com/sets/"],
"-c": ["https://readcomicsonline.ru"],
"-h": ["https://www.twitch.tv/"],
}
EXCLUDE_FROM_NOW_PLAYING = ("GeoLocation",)
@ -63,6 +64,7 @@ MANUAL_SCROBBLE_FNS = {
"-l": "manual_scrobble_brickset",
"-c": "manual_scrobble_book",
"-f": "manual_scrobble_food",
"-h": "manual_scrobble_twitch_channel",
}

11
vrobbler/apps/scrobbles/management/commands/update_twitch_scrobbles.py Normal file
View File

@ -0,0 +1,11 @@
from django.core.management.base import BaseCommand
class Command(BaseCommand):
help = "Check recent Twitch channel scrobbles for matching VODs"
def handle(self, *args, **options):
from scrobbles.tasks import check_twitch_channels_for_vods
check_twitch_channels_for_vods()
self.stdout.write(self.style.SUCCESS("Done"))

54
vrobbler/apps/scrobbles/migrations/0075_add_channel_scrobble.py Normal file
View File

@ -0,0 +1,54 @@
# Generated by Django 4.2.29 on 2026-04-01 16:18
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
dependencies = [
("videos", "0028_channel_twitch_id_video_twitch_id_and_more"),
("scrobbles", "0074_add_tags_to_scrobble"),
]
operations = [
migrations.AddField(
model_name="scrobble",
name="channel",
field=models.ForeignKey(
blank=True,
null=True,
on_delete=django.db.models.deletion.DO_NOTHING,
to="videos.channel",
),
),
migrations.AlterField(
model_name="scrobble",
name="media_type",
field=models.CharField(
choices=[
("Video", "Video"),
("Track", "Track"),
("PodcastEpisode", "Podcast episode"),
("SportEvent", "Sport event"),
("Book", "Book"),
("Paper", "Paper"),
("VideoGame", "Video game"),
("BoardGame", "Board game"),
("GeoLocation", "GeoLocation"),
("Trail", "Trail"),
("Beer", "Beer"),
("Puzzle", "Puzzle"),
("Food", "Food"),
("Task", "Task"),
("WebPage", "Web Page"),
("LifeEvent", "Life event"),
("Mood", "Mood"),
("BrickSet", "Brick set"),
("Channel", "Channel"),
],
default="Video",
max_length=14,
),
),
]

4
vrobbler/apps/scrobbles/models.py
View File

@ -397,6 +397,7 @@ class Scrobble(TimeStampedModel):
LIFE_EVENT = "LifeEvent", "Life event"
MOOD = "Mood", "Mood"
BRICKSET = "BrickSet", "Brick set"
CHANNEL = "Channel", "Channel"
@classmethod
def list(cls):
@ -404,6 +405,7 @@ class Scrobble(TimeStampedModel):
uuid = models.UUIDField(editable=False, **BNULL)
video = models.ForeignKey(Video, on_delete=models.DO_NOTHING, **BNULL)
channel = models.ForeignKey("videos.Channel", on_delete=models.DO_NOTHING, **BNULL)
track = models.ForeignKey(Track, on_delete=models.DO_NOTHING, **BNULL)
podcast_episode = models.ForeignKey(
PodcastEpisode, on_delete=models.DO_NOTHING, **BNULL
@ -961,6 +963,8 @@ class Scrobble(TimeStampedModel):
media_obj = self.task
if self.food:
media_obj = self.food
if self.channel:
media_obj = self.channel
return media_obj
def __str__(self):

51
vrobbler/apps/scrobbles/scrobblers.py
View File

@ -35,8 +35,8 @@ from scrobbles.notifications import ScrobbleNtfyNotification
from scrobbles.utils import (
convert_to_seconds,
extract_domain,
remove_last_part,
next_url_if_exists,
remove_last_part,
)
from sports.models import SportEvent
from sports.thesportsdb import lookup_event_from_thesportsdb
@ -188,6 +188,15 @@ def web_scrobbler_scrobble_media(
return video.scrobble_for_user(user_id, status, source="Web Scrobbler")
def twitch_scrobble_channel(
twitch_handle: str, user_id: int, status: str = "started"
) -> Optional[Scrobble]:
from videos.models import Channel
channel = Channel.find_or_create(twitch_handle)
return channel.scrobble_for_user(user_id, status, source="Twitch")
def manual_scrobble_video(
video_id: str,
user_id: int,
@ -629,6 +638,8 @@ def manual_scrobble_from_url(
item_id = url
elif content_key == "-i" and "title/tt" in url:
item_id = "tt" + str(item_id)
elif content_key == "-h" and "twitch.tv" in url:
item_id = url
scrobble_fn = MANUAL_SCROBBLE_FNS[content_key]
return eval(scrobble_fn)(item_id, user_id, source=source, action=action)
@ -984,6 +995,44 @@ def manual_scrobble_webpage(
return scrobble
def manual_scrobble_twitch_channel(
url: str,
user_id: int,
source: str = "Manual",
action: Optional[str] = None,
):
from videos.models import Channel
handle = (
url.replace("https://www.twitch.tv/", "")
.replace("https://twitch.tv/", "")
.rstrip("/")
)
channel = Channel.find_or_create(handle)
scrobble_dict = {
"user_id": user_id,
"timestamp": timezone.now(),
"playback_position_seconds": 0,
"source": source,
}
logger.info(
"[vrobbler-scrobble] twitch channel scrobble request received",
extra={
"channel_id": channel.id,
"handle": handle,
"user_id": user_id,
"scrobble_dict": scrobble_dict,
"media_type": Scrobble.MediaType.CHANNEL,
},
)
scrobble = Scrobble.create_or_update(channel, user_id, scrobble_dict)
return scrobble
def gpslogger_scrobble_location(data_dict: dict, user_id: int) -> Scrobble:
location = GeoLocation.find_or_create(data_dict)

65
vrobbler/apps/scrobbles/tasks.py
View File

@ -1,12 +1,13 @@
import logging
from datetime import timedelta
from celery import shared_task
from charts.utils import (
build_charts_since,
build_daily_charts,
build_yearly_charts,
build_monthly_charts,
build_weekly_charts,
build_yearly_charts,
)
from django.apps import apps
from django.contrib.auth import get_user_model
@ -29,6 +30,68 @@ MEDIA_TYPES = [
]
@shared_task
def check_twitch_channels_for_vods():
"""Check recent Twitch channel scrobbles for matching VODs."""
from scrobbles.models import Scrobble
cutoff = timezone.now() - timedelta(hours=48)
channel_scrobbles = Scrobble.objects.filter(
media_type=Scrobble.MediaType.CHANNEL,
timestamp__gte=cutoff,
)
logger.info(f"[twitch_vods] Checking {channel_scrobbles.count()} channel scrobbles")
matched_count = 0
for scrobble in channel_scrobbles:
if not scrobble.channel or not scrobble.channel.twitch_id:
continue
try:
from videos.sources.twitch import get_channel_vods
vods = get_channel_vods(scrobble.channel.twitch_id)
scrobble_time = scrobble.timestamp
for vod in vods:
if not vod.get("published_at"):
continue
from videos.sources import twitch as twitch_source
vod_time = twitch_source.parse_twitch_datetime(vod["published_at"])
if not vod_time:
continue
time_diff = (vod_time - scrobble_time).total_seconds()
if 0 < time_diff <= 86400:
from videos.models import Video
video = Video.get_from_twitch_id(vod["id"], overwrite=True)
if video:
video.scrobble_for_user(
scrobble.user_id,
status="stopped",
source="Twitch VOD",
)
matched_count += 1
logger.info(
f"[twitch_vods] Matched VOD {vod['id']} for channel scrobble {scrobble.id}"
)
break
except Exception as e:
logger.warning(
f"[twitch_vods] Error processing scrobble {scrobble.id}: {e}"
)
logger.info(f"[twitch_vods] Matched {matched_count} VODs")
@shared_task
def process_retroarch_import(import_id):
RetroarchImport = apps.get_model("scrobbles", "RetroarchImport")

3
vrobbler/apps/videos/metadata.py
View File

@ -59,6 +59,9 @@ class VideoMetadata:
# YouTube specific
channel_id: Optional[int]
# Twitch specific
upload_date: Optional[str]
# General
cover_url: Optional[str]
genres: list[str]

35
vrobbler/apps/videos/migrations/0029_alter_channel_options_channel_base_run_time_seconds_and_more.py Normal file
View File

@ -0,0 +1,35 @@
# Generated by Django 4.2.29 on 2026-04-01 16:24
from django.db import migrations, models
import taggit.managers
class Migration(migrations.Migration):
dependencies = [
("taggit", "0004_alter_taggeditem_content_type_alter_taggeditem_tag"),
("videos", "0028_channel_twitch_id_video_twitch_id_and_more"),
]
operations = [
migrations.AlterModelOptions(
name="channel",
options={"verbose_name_plural": "channels"},
),
migrations.AddField(
model_name="channel",
name="base_run_time_seconds",
field=models.IntegerField(blank=True, null=True),
),
migrations.AddField(
model_name="channel",
name="tags",
field=taggit.managers.TaggableManager(
blank=True,
help_text="A comma-separated list of tags.",
through="taggit.TaggedItem",
to="taggit.Tag",
verbose_name="Tags",
),
),
]

117
vrobbler/apps/videos/models.py
View File

@ -40,7 +40,7 @@ class VideoLogData(BaseLogData, WithPeopleLogData):
rating: Optional[int] = None
class Channel(TimeStampedModel):
class Channel(ScrobblableMixin):
uuid = models.UUIDField(default=uuid4, editable=False, **BNULL)
name = models.CharField(max_length=255)
cover_image = models.ImageField(upload_to="videos/channels/", **BNULL)
@ -60,32 +60,29 @@ class Channel(TimeStampedModel):
twitch_id = models.CharField(max_length=255, **BNULL)
genre = TaggableManager(through=ObjectWithGenres)
class Meta:
verbose_name_plural = "channels"
SECONDS_TO_STALE = 7200 # 2 hours
COMPLETION_PERCENT = 100
def __str__(self) -> str:
return self.name
def get_absolute_url(self):
return reverse("videos:channel_detail", kwargs={"slug": self.uuid})
@property
def run_time_seconds(self) -> int:
return self.base_run_time_seconds or 7200
@property
def youtube_url(self):
return YOUTUBE_CHANNEL_URL + self.youtube_id
def title(self):
return self.name
@property
def primary_image_url(self) -> str:
url = ""
if self.cover_image:
url = self.cover_image_medium.url
return url
@property
def safe_cover_image_url(self) -> str:
if self.cover_image:
try:
if self.cover_image.storage.exists(self.cover_image.name):
return self.cover_medium.url
except Exception:
pass
return "/static/images/not-found.jpg"
def save_image_from_url(self, url: str, force_update: bool = False):
if not self.cover_image or (force_update and url):
r = requests.get(url)
if r.status_code == 200:
fname = f"{self.name}_{self.uuid}.jpg"
self.cover_image.save(fname, ContentFile(r.content), save=True)
def scrobbles_for_user(self, user_id: int, include_playing=False):
from scrobbles.models import Scrobble
@ -95,10 +92,51 @@ class Channel(TimeStampedModel):
played_query = models.Q()
return Scrobble.objects.filter(
played_query,
video__channel=self,
channel=self,
user=user_id,
).order_by("-timestamp")
def get_absolute_url(self):
return reverse("videos:channel_detail", kwargs={"slug": self.uuid})
@classmethod
def find_or_create(cls, twitch_handle: str, overwrite: bool = False):
channel, created = cls.objects.get_or_create(twitch_id=twitch_handle)
if not created and not overwrite:
return channel
try:
from videos.sources.twitch import lookup_channel_from_twitch
metadata = lookup_channel_from_twitch(twitch_handle)
if metadata.name:
channel.name = metadata.name
if metadata.profile_image_url:
channel.save_image_from_url(metadata.profile_image_url)
channel.save()
except Exception as e:
logger.warning(f"Failed to get channel metadata: {e}")
return channel
def scrobble_for_user(
self,
user_id: int,
status: str = "started",
source: str = "Twitch",
**kwargs,
):
from scrobbles.models import Scrobble
return Scrobble.create_or_update(
media_object=self,
user_id=user_id,
status=status,
source=source,
**kwargs,
)
def fix_metadata(self, force: bool = False):
# TODO Scrape channel info from Youtube
logger.warning("Not implemented yet")
@ -415,6 +453,39 @@ class Video(ScrobblableMixin):
return video
@classmethod
def get_from_twitch_id(cls, twitch_id: str, overwrite: bool = False) -> "Video":
video, created = cls.objects.get_or_create(twitch_id=twitch_id)
if not created and not overwrite:
return video
from videos.sources.twitch import lookup_video_from_twitch
metadata = lookup_video_from_twitch(twitch_id)
if created or overwrite:
video.title = metadata.title or f"Twitch VOD {twitch_id}"
video.overview = metadata.overview
video.cover_url = metadata.cover_url
video.base_run_time_seconds = metadata.base_run_time_seconds
if metadata.upload_date:
video.upload_date = metadata.upload_date
if metadata.year:
video.year = metadata.year
video.video_type = Video.VideoType.TWITCH
if metadata.channel_id:
from videos.models import Channel
video.channel = Channel.objects.filter(id=metadata.channel_id).first()
video.save()
if metadata.cover_url:
video.save_image_from_url(metadata.cover_url)
return video
@classmethod
def find_or_create(
cls, source_id: Optional[str], overwrite: bool = False
@ -426,6 +497,8 @@ class Video(ScrobblableMixin):
return cls.get_from_imdb_id(source_id, overwrite)
if bool(YOUTUBE_ID_PATTERN.match(source_id)):
return cls.get_from_youtube_id(source_id, overwrite)
if source_id.isdigit():
return cls.get_from_twitch_id(source_id, overwrite)
logger.warning("Video ID not recognized, not scrobbling")

112
vrobbler/apps/videos/sources/twitch.py
View File

@ -1,4 +1,5 @@
import logging
from dataclasses import dataclass
import pendulum
import requests
@ -8,6 +9,117 @@ from videos.metadata import VideoMetadata, VideoType
logger = logging.getLogger(__name__)
TWITCH_VIDEO_URL = "https://www.twitch.tv/videos/{video_id}"
TWITCH_CHANNEL_URL = "https://www.twitch.tv/{handle}"
@dataclass
class ChannelMetadata:
name: str = ""
display_name: str = ""
profile_image_url: str = ""
bio: str = ""
twitch_id: str = ""
def lookup_channel_from_twitch(handle: str) -> ChannelMetadata:
channel_metadata = ChannelMetadata(twitch_id=handle)
url = TWITCH_CHANNEL_URL.format(handle=handle)
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
}
try:
response = requests.get(url, headers=headers, timeout=10)
except requests.RequestException as e:
logger.warning(f"Failed to fetch Twitch channel page: {e}")
return channel_metadata
if response.status_code != 200:
logger.warning(f"Twitch channel page returned status {response.status_code}")
return channel_metadata
soup = BeautifulSoup(response.text, "html.parser")
title_tag = soup.find("meta", property="og:title")
if title_tag:
content = title_tag.get("content", "")
if " on Twitch" in content:
channel_metadata.display_name = content.replace(" on Twitch", "").strip()
else:
channel_metadata.display_name = content
channel_metadata.name = channel_metadata.display_name.replace(
" - Twitch", ""
).strip()
description_tag = soup.find("meta", property="og:description")
if description_tag:
channel_metadata.bio = description_tag.get("content", "")
image_tag = soup.find("meta", property="og:image")
if image_tag:
channel_metadata.profile_image_url = image_tag.get("content", "")
return channel_metadata
def get_channel_vods(handle: str) -> list:
"""Get recent VODs for a channel by scraping the channel's videos page."""
import re
url = f"https://www.twitch.tv/{handle}/videos"
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
}
try:
response = requests.get(url, headers=headers, timeout=10)
except requests.RequestException as e:
logger.warning(f"Failed to fetch Twitch channel videos page: {e}")
return []
if response.status_code != 200:
logger.warning(f"Twitch videos page returned status {response.status_code}")
return []
page_text = response.text
vods = []
video_pattern = r'"url":\s*"https://www\.twitch\.tv/videos/(\d+)"'
url_matches = re.findall(video_pattern, page_text)
upload_pattern = r'"uploadDate":\s*"([^"]+)"'
date_matches = re.findall(upload_pattern, page_text)
name_pattern = r'"name":\s*"([^"]+)"'
name_matches = re.findall(name_pattern, page_text)
for i, video_id in enumerate(url_matches):
title = name_matches[i] if i < len(name_matches) else ""
published_at = date_matches[i] if i < len(date_matches) else ""
vods.append(
{
"id": video_id,
"title": title,
"published_at": published_at,
"thumbnail": None,
}
)
logger.info(f"[get_channel_vods] Found {len(vods)} VODs for {handle}")
print(vods)
return vods
def parse_twitch_datetime(date_str: str):
"""Parse Twitch datetime string to timezone-aware datetime."""
if not date_str:
return None
try:
return pendulum.parse(date_str)
except Exception:
return None
def lookup_video_from_twitch(twitch_video_id: str) -> VideoMetadata: