MillionProjectMembersGraph/million/parsing/facebook_parser.py

import json
import os
import re
from typing import Any
from million.model.facebook_schema 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'))