from sqlalchemy.orm import Session from typing import Dict, Any, Optional, List from uuid import UUID from datetime import datetime, timezone import requests import os from ..models.stream import Stream from ..models.chat_message import ChatMessage as ChatMessageModel from ..models.chapter import Chapter from ..services.chat_service import ChatService class ChatJsonService: def __init__(self, db: Session): self.db = db self.chat_service = ChatService(db) def parse_chat_json(self, json_data: Dict[str, Any]) -> Dict[str, Any]: """Parse chat.json and extract relevant data""" video_data = json_data.get("video", {}) streamer_data = json_data.get("streamer", {}) comments = json_data.get("comments", []) chapters = video_data.get("chapters", []) embedded_data = json_data.get("embeddedData", {}) return { "title": video_data.get("title"), "description": video_data.get("description"), "game": video_data.get("game"), "streamer": streamer_data.get("name"), "streamer_login": streamer_data.get("login"), "streamer_id": streamer_data.get("id"), "duration": video_data.get("length"), "created_at": video_data.get("created_at"), "view_count": video_data.get("viewCount"), "video_id": video_data.get("id"), "chapters": chapters, "comments": comments, "embedded_data": embedded_data } def create_stream_from_chat_json( self, user_id: UUID, hls_path: Optional[str], parsed_data: Dict[str, Any], thumbnail_timestamp: Optional[float] = None, is_public: bool = False, is_twitch_import: bool = False ) -> Stream: """Create a stream from parsed chat.json data""" # Normalize path - remove /app prefix but keep leading / normalized_path = hls_path if hls_path and hls_path.startswith('/app/'): normalized_path = hls_path[4:] # Remove /app but keep / # Set status to pending if no HLS path provided status = "pending" if not normalized_path else "ready" # Generate Twitch thumbnail URL from video ID thumbnail_url = None video_id = parsed_data.get("video_id") if video_id: # Set thumbnail URL to Twitch URL initially (will be downloaded later if HLS path is set) thumbnail_url = f"https://static-cdn.jtvnw.net/cf_vods/d2nvs3xx597ldy/{video_id}/thumb/thumb-640-360.jpg" stream = Stream( title=parsed_data.get("title", "Untitled Stream"), description=parsed_data.get("description"), hls_path=normalized_path or None, status=status, created_by=user_id, duration_seconds=parsed_data.get("duration", 0) or 0, view_count=0, # Always start with 0 views, regardless of chat.json is_public=is_public, thumbnail_url=thumbnail_url, twitch_vod_id=str(parsed_data.get("video_id", "")) or None if is_twitch_import else None, streamer=parsed_data.get("streamer"), stream_date=self._parse_datetime(parsed_data.get("created_at")), stream_metadata={ "game": parsed_data.get("game"), "streamer_login": parsed_data.get("streamer_login"), "streamer_id": parsed_data.get("streamer_id") } ) self.db.add(stream) self.db.commit() self.db.refresh(stream) # Regenerate master ffconcat if stream is ready if status == "ready": try: from ..services.ffconcat_service import regenerate_master_ffconcat regenerate_master_ffconcat() except Exception as e: import sys sys.stderr.write(f"[FFCONCAT ERROR] Failed to regenerate after chat.json import: {e}\n") # Renumber all streams after creation try: from ..services.stream_numbering_service import renumber_all_streams renumber_all_streams() except Exception as e: import sys sys.stderr.write(f"[NUMBERING ERROR] Failed to renumber after chat.json import: {e}\n") return stream def import_chapters_from_chat_json( self, stream_id: UUID, chapters_data: List[Dict[str, Any]] ): """Import chapters from chat.json""" for chapter_data in chapters_data: chapter = Chapter( stream_id=stream_id, game_display_name=chapter_data.get("gameDisplayName", chapter_data.get("description", "")), game_id=str(chapter_data.get("gameId", "")) or None, description=chapter_data.get("subDescription", ""), start_milliseconds=chapter_data.get("startMilliseconds", 0) or 0, length_milliseconds=chapter_data.get("lengthMilliseconds", 0) or 0, thumbnail_url=chapter_data.get("thumbnailUrl", "") ) self.db.add(chapter) self.db.commit() def import_chat_messages_from_chat_json( self, stream_id: UUID, comments_data: List[Dict[str, Any]], embedded_data: Dict[str, Any] ): """Import chat messages from chat.json""" # Build emote mapping from embedded data emote_mapping = {} # Add first party emotes first_party_emotes = embedded_data.get("firstParty", []) for emote in first_party_emotes: emote_id = emote.get("id") if emote_id: emote_mapping[emote_id] = { "id": emote_id, "imageScale": emote.get("imageScale"), "data": emote.get("data"), "name": emote.get("name"), "width": emote.get("width"), "height": emote.get("height"), "source": "firstParty" } # Add third party emotes third_party_emotes = embedded_data.get("thirdParty", []) for emote in third_party_emotes: emote_id = emote.get("id") if emote_id: emote_mapping[emote_id] = { "id": emote_id, "imageScale": emote.get("imageScale"), "data": emote.get("data"), "name": emote.get("name"), "width": emote.get("width"), "height": emote.get("height"), "source": "thirdParty" } # Build badge mapping from embedded data badge_mapping = {} twitch_badges = embedded_data.get("twitchBadges", []) if isinstance(twitch_badges, list): for badge in twitch_badges: badge_name = badge.get("name") if badge_name and isinstance(badge, dict) and "versions" in badge: versions = badge.get("versions", {}) for version_num, version_data in versions.items(): if isinstance(version_data, dict): badge_key = f"{badge_name}:{version_num}" badge_mapping[badge_key] = { "name": badge_name, "version": version_num, "title": version_data.get("title", badge_name), "description": version_data.get("description", ""), "bytes": version_data.get("bytes", ""), "source": "twitchBadges" } for comment in comments_data: commenter = comment.get("commenter", {}) message = comment.get("message", {}) # Map emoticons to emotes with embedded data emoticons = message.get("emoticons", []) mapped_emotes = [] for emoticon in emoticons: emoticon_id = emoticon.get("_id") if emoticon_id and emoticon_id in emote_mapping: mapped_emotes.append({ "id": emoticon_id, "begin": emoticon.get("begin"), "end": emoticon.get("end"), "image_data": emote_mapping[emoticon_id] }) # Map user badges to embedded badge data user_badges = message.get("user_badges", []) mapped_badges = [] for badge in user_badges: badge_name = badge.get("_id") badge_version = badge.get("version", "1") badge_key = f"{badge_name}:{badge_version}" if badge_key in badge_mapping: mapped_badges.append({ **badge, "embedded_data": badge_mapping[badge_key] }) else: # Keep original badge if no embedded data found mapped_badges.append(badge) self.chat_service.create_chat_message( stream_id=stream_id, username=commenter.get("display_name", "Unknown"), message=message.get("body", ""), timestamp_offset=comment.get("content_offset_seconds", 0) or 0, color=message.get("user_color", "#9146FF"), emotes=mapped_emotes, badges=mapped_badges, user_avatar_url=commenter.get("logo") ) def _parse_datetime(self, datetime_str: Optional[str]) -> Optional[datetime]: """Parse ISO datetime string to datetime object""" if not datetime_str: return None try: return datetime.fromisoformat(datetime_str.replace("Z", "+00:00")) except (ValueError, AttributeError): return None