Ajout des données JSON facebook messenger

+ les dataclasses et le module de parsing

python/model/facebook_export.py

@@ -0,0 +1,291 @@
+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
+        )

python/parsing/facebook_export.py

@@ -0,0 +1,72 @@
+import json
+import os
+import re
+from typing import Any
+from model.facebook_export import FacebookExport
+
+def parse_facebook_exports_from_folder(folder_path: str) -> FacebookExport:
+    """
+    Parcourt un dossier, lit chaque fichier JSON et les fusionne dans un seul objet FacebookExport.
+
+    :param folder_path: Chemin vers le dossier contenant les fichiers JSON
+    :return: Un objet FacebookExport
+    """
+    if not os.path.isdir(folder_path):
+        raise FileNotFoundError(f"Le dossier {folder_path} n'existe pas.")
+    
+    json_files = [f for f in os.listdir(folder_path) if f.endswith('.json')]
+    
+    if not json_files:
+        raise FileNotFoundError(f"Aucun fichier JSON trouvé dans le dossier {folder_path}.")
+
+    merged_facebook_export = None
+
+    for file_name in json_files:
+        #file_path = os.path.join(folder_path, file_name)
+
+        facebook_export = parse_facebook_export(file_name)
+
+        if merged_facebook_export is None:
+            merged_facebook_export = facebook_export
+        else:
+            merged_facebook_export.messages.extend(facebook_export.messages)
+            merged_facebook_export.participants.update(facebook_export.participants)
+
+    if merged_facebook_export is None:
+        raise ValueError(f"Aucun fichier valide n'a été trouvé ou parsé dans le dossier {folder_path}.")
+    
+    return merged_facebook_export
+
+def parse_facebook_export(file_name: str) -> FacebookExport:
+    """
+    Parse un fichier JSON en instance de la classe model.FacebookExport
+
+    :param file_name: Le chemin du fichier à ouvrir
+    :return: Une instance de FacebookExport initialisée avec les données du fichier JSON
+    """
+    if not file_name.lower().endswith('.json') or not os.path.exists(file_name):
+        raise ValueError(f"Le fichier {file_name} n'existe pas ou n'est pas un fichier JSON.")
+
+    try:
+        json_data = read_broken_fb_json(file_name)
+        return FacebookExport.from_dict(json_data)
+    except Exception as e:
+        raise Exception(f"Erreur lors de la conversion du JSON en FacebookExport: {str(e)}")
+
+def read_broken_fb_json(file_name: str) -> Any:
+    """
+    Lit un fichier JSON issu de Facebook et renvoie le contenu après correction de l'encodage
+
+    :param file_name: Le chemin du fichier à ouvrir
+    :return: Le contenu du fichier JSON sous forme d'objet
+    """
+    # https://stackoverflow.com/questions/50008296/facebook-json-badly-encoded
+    with open(file_name, 'rb') as f:
+        binary_data = f.read()
+        repaired = re.sub(
+            rb'\\u00([\da-f]{2})',
+            lambda m: bytes.fromhex(m.group(1).decode()),
+            binary_data
+        )
+
+        return json.loads(repaired.decode('utf8'))