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 | None:
    """
    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 ou None si aucun résultat trouvé
    """
    merged_facebook_export = None

    with os.scandir(folder_path) as it:
        for entry in it:
            if not (entry.is_file() and entry.name.endswith('.json')): continue

            facebook_export = parse_facebook_export(entry.path)

            if merged_facebook_export is None:
                merged_facebook_export = facebook_export
            else:
                merged_facebook_export.messages.extend(facebook_export.messages)
    
    if merged_facebook_export is None: return None

    merged_facebook_export.messages = sorted(merged_facebook_export.messages, key=lambda m: m.date_time)
    merged_facebook_export.participants = sorted(merged_facebook_export.participants, key=lambda m: m.name)
    
    return merged_facebook_export

def parse_facebook_export(file_path: str) -> FacebookExport:
    """
    Parse un fichier JSON en instance de la classe FacebookExport

    :param file_path: Le chemin du fichier à ouvrir
    :return: Une instance de FacebookExport initialisée avec les données du fichier JSON
    """
    if not file_path.lower().endswith('.json') or not os.path.exists(file_path):
        raise ValueError(f"Le fichier {file_path} n'existe pas ou n'est pas un fichier JSON.")

    try:
        json_data = read_broken_fb_json(file_path)
        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_path: str) -> Any:
    """
    Lit un fichier JSON issu de Facebook et renvoie le contenu après correction de l'encodage

    :param file_path: 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_path, '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'))