MillionProjectMembersGraph/million/model/facebook_schema.py

from datetime import datetime
from typing import Any, List, Optional, Self
from uuid import uuid4

class AudioFile:
    uri: str
    creation_timestamp: int

    def __init__(self, uri: str, creation_timestamp: int):
        self.uri = uri
        self.creation_timestamp = creation_timestamp

    @classmethod
    def from_dict(cls, data: dict) -> Self:
        return cls(
            uri=data['uri'],
            creation_timestamp=data['creation_timestamp']
        )
    
class Gif:
    uri: str

    def __init__(self, uri: str):
        self.uri = uri

    @classmethod
    def from_dict(cls, data: dict) -> Self:
        return cls(
            uri=data['uri']
        )

class Photo:
    uri: str
    creation_timestamp: int

    def __init__(self, uri: str, creation_timestamp: int):
        self.uri = uri
        self.creation_timestamp = creation_timestamp

    @classmethod
    def from_dict(cls, data: dict) -> Self:
        return cls(
            uri=data['uri'],
            creation_timestamp=data['creation_timestamp']
        )
    
class Reaction:
    reaction: str
    actor: str

    def __init__(self, reaction: str, actor: str):
        self.reaction = reaction
        self.actor = actor

    @classmethod
    def from_dict(cls, data: dict) -> Self:
        return cls(
            reaction=data['reaction'],
            actor=data['actor']
        )
    
class Share:
    link: str
    share_text: str

    def __init__(self, link: str, share_text: str):
        self.link = link
        self.share_text = share_text

    @classmethod
    def from_dict(cls, data: dict) -> Self:
        return cls(
            link=data['link'],
            share_text=data['share_text']
        )

class Sticker:
    uri: str
    ai_stickers: Optional[List[Any]]

    def __init__(self, uri: str, ai_stickers: Optional[List[Any]] = None):
        self.uri = uri
        self.ai_stickers = ai_stickers if ai_stickers is not None else []

    @classmethod
    def from_dict(cls, data: dict) -> Self:
        return cls(
            uri=data['uri'],
            ai_stickers=data.get('ai_stickers', [])
        )
    
class Video:
    uri: str
    creation_timestamp: int

    def __init__(self, uri: str, creation_timestamp: int):
        self.uri = uri
        self.creation_timestamp = creation_timestamp

    @classmethod
    def from_dict(cls, data: dict) -> Self:
        return cls(
            uri=data['uri'],
            creation_timestamp=data['creation_timestamp']
        )
  
class Message:
    sender_name: str
    date_time: datetime
    content: Optional[str]
    sticker: Optional[Sticker]
    share: Optional[Share]
    photos: Optional[List[Photo]]
    videos: Optional[List[Video]]
    gifs: Optional[List[Gif]]
    audio_files: Optional[List[AudioFile]]
    call_duration: Optional[int]
    reactions: Optional[List[Reaction]]
    is_unsent: Optional[bool]
    is_geoblocked_for_viewer: bool
    _id: str

    def __init__(self,
                 sender_name: str,
                 date_time: datetime,
                 is_geoblocked_for_viewer: bool,
                 content: Optional[str] = None,
                 sticker: Optional[Sticker] = None,
                 share: Optional[Share] = None,
                 photos: Optional[List[Photo]] = None,
                 videos: Optional[List[Video]] = None,
                 gifs: Optional[List[Gif]] = None,
                 audio_files: Optional[List[AudioFile]] = None,
                 call_duration: Optional[int] = None,
                 reactions: Optional[List[Reaction]] = None,
                 is_unsent: Optional[bool] = None
                 ):
        self.sender_name = sender_name
        self.date_time = date_time
        self.content = content
        self.sticker = sticker
        self.share = share
        self.photos = photos
        self.videos = videos
        self.gifs = gifs
        self.audio_files = audio_files
        self.call_duration = call_duration
        self.reactions = reactions
        self.is_unsent = is_unsent
        self.is_geoblocked_for_viewer = is_geoblocked_for_viewer
        self._id = str(uuid4())

    @property
    def message_id(self) -> str:
        return self._id

    @classmethod
    def from_dict(cls, data: dict) -> Self:
        _timestamp_ms = data.get('timestamp_ms')
        _sender_name = data.get('sender_name')
        _is_geoblocked_for_viewer = data.get('is_geoblocked_for_viewer')
        _content = data.get('content')
        _sticker_dict = data.get('sticker')
        _share_dict = data.get('share')
        _photos_dict = data.get('photos')
        _videos_dict = data.get('videos')
        _gifs_dict = data.get('gifs')
        _audio_files_dict = data.get('audio_files')
        _call_duration = data.get('call_duration')
        _reactions_dict = data.get('reactions')
        _is_unsent = data.get('is_unsent')

        _date_time = datetime.fromtimestamp(_timestamp_ms / 1000)
        _sticker = Sticker.from_dict(_sticker_dict) if _sticker_dict else None
        _share = Share.from_dict(_share_dict) if _share_dict else None
        _photos = [Photo.from_dict(photo) for photo in _photos_dict] if _photos_dict else None
        _videos = [Video.from_dict(video) for video in _videos_dict] if _videos_dict else None
        _gifs = [Gif.from_dict(gif) for gif in _gifs_dict] if _gifs_dict else None
        _audio_files = [AudioFile.from_dict(audio_file) for audio_file in _audio_files_dict] if _audio_files_dict else None
        _reactions = [Reaction.from_dict(reaction) for reaction in _reactions_dict] if _reactions_dict else None

        return cls(
            sender_name=_sender_name,
            date_time=_date_time,
            is_geoblocked_for_viewer=_is_geoblocked_for_viewer,
            content=_content,
            sticker=_sticker,
            share=_share,
            photos=_photos,
            videos=_videos,
            gifs=_gifs,
            audio_files=_audio_files,
            call_duration=_call_duration,
            reactions=_reactions,
            is_unsent=_is_unsent
        )
 
class Image:
    creation_timestamp: int
    uri: str

    def __init__(self, creation_timestamp: int, uri: str):
        self.creation_timestamp = creation_timestamp
        self.uri = uri

    @classmethod
    def from_dict(cls, data: dict) -> Self:
        return cls(
            creation_timestamp=data['creation_timestamp'],
            uri=data['uri']
        )

class JoinableMode:
    mode: int
    link: str

    def __init__(self, mode: int, link: str):
        self.mode = mode
        self.link = link

    @classmethod
    def from_dict(cls, data: dict) -> Self:
        return cls(
            mode=data['mode'],
            link=data['link']
        )

class Participant:
    name: str

    def __init__(self, name: str) -> None:
        self.name = name

    @classmethod
    def from_dict(cls, data: dict) -> Self:
        return cls(
            name = data.get('name')
        )

class FacebookExport:
    messages: List[Message]
    participants: List[Participant]
    title: str
    is_still_participant: bool
    thread_path: str
    magic_words: List[Any]
    image: Image
    joinable_mode: JoinableMode

    def __init__(self,
                 messages: List[Message],
                 participants: List[Participant],
                 title: str,
                 is_still_participant: bool,
                 thread_path: str,
                 magic_words: List[Any],
                 image: Image,
                 joinable_mode: JoinableMode):
        self.messages = messages
        self.participants = participants
        self.title = title
        self.is_still_participant = is_still_participant
        self.thread_path = thread_path
        self.magic_words = magic_words
        self.image = image
        self.joinable_mode = joinable_mode

    @classmethod
    def from_dict(cls, data: dict) -> Self:
        messages_data = data.get('messages', [])
        participants_data = data.get('participants', [])
        title = data['title']
        is_still_participant = data['is_still_participant']
        thread_path = data['thread_path']
        magic_words = data.get('magic_words', [])
        image = Image.from_dict(data['image'])
        joinable_mode = JoinableMode.from_dict(data['joinable_mode'])
        
        messages = [Message.from_dict(m) for m in messages_data]
        participants = {Participant.from_dict(p) for p in participants_data}

        return cls(
            messages=messages,
            participants=participants,
            title=title,
            is_still_participant=is_still_participant,
            thread_path=thread_path,
            magic_words=magic_words,
            image=image,
            joinable_mode=joinable_mode
        )