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'))