Spaces:
Sleeping
Sleeping
File size: 5,989 Bytes
164d23a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 | """
Module for saving meeting data to Hugging Face Datasets.
Manages permanent persistence of analysis results.
"""
import json
import uuid
from datetime import datetime
from typing import Dict, Optional
try:
from datasets import Dataset
from huggingface_hub import HfApi, login
except ImportError:
Dataset = None
HfApi = None
login = None
# Configurazione logging
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Nome del dataset su Hugging Face
DATASET_NAME = "meeting-summarizer-data"
def save_meeting_to_dataset(meeting_data: Dict, hf_token: Optional[str] = None) -> bool:
"""
Save meeting data to Hugging Face Dataset.
Args:
meeting_data (Dict): Meeting data to save
hf_token (Optional[str]): Hugging Face token (optional)
Returns:
bool: True if saved successfully, False otherwise
"""
if not meeting_data:
logger.error("Meeting data not provided")
return False
if Dataset is None:
logger.error("datasets not installed. Install with: pip install datasets")
return False
try:
# Authentication if token provided
if hf_token:
try:
login(token=hf_token)
logger.info("Hugging Face authentication completed")
except Exception as e:
logger.warning(f"Error in HF authentication: {str(e)}")
logger.info("Continuing without authentication...")
# Prepare data for saving
meeting_record = _prepare_meeting_record(meeting_data)
# Create or load dataset
dataset = _get_or_create_dataset()
# Add new record
if dataset is None:
logger.error("Unable to create or load dataset")
return False
# Convert dataset to list to add record
records = list(dataset)
records.append(meeting_record)
# Create new dataset with added record
new_dataset = Dataset.from_list(records)
# Push to Hugging Face Hub (if authenticated)
if hf_token:
try:
new_dataset.push_to_hub(
DATASET_NAME,
private=True,
token=hf_token
)
logger.info(f"Dataset updated on Hugging Face Hub: {DATASET_NAME}")
except Exception as e:
logger.warning(f"Unable to push to HF Hub: {str(e)}")
logger.info("Data saved locally")
logger.info("Meeting saved successfully to dataset")
return True
except Exception as e:
logger.error(f"Error while saving meeting: {str(e)}")
return False
def _prepare_meeting_record(meeting_data: Dict) -> Dict:
"""
Prepare meeting record for saving.
Args:
meeting_data (Dict): Meeting data
Returns:
Dict: Record formatted for dataset
"""
current_time = datetime.now()
return {
"id": str(uuid.uuid4()),
"file_name": meeting_data.get("file_name", "unknown"),
"meeting_date": current_time.strftime("%Y-%m-%d"),
"transcription": meeting_data.get("transcription", ""),
"summary": meeting_data.get("summary", ""),
"topics": json.dumps(meeting_data.get("topics", [])),
"keywords": json.dumps(meeting_data.get("keywords", [])),
"created_at": current_time.isoformat()
}
def _get_or_create_dataset() -> Optional[Dataset]:
"""
Create or load Hugging Face dataset.
Returns:
Optional[Dataset]: Dataset or None if error
"""
try:
# Try to load existing dataset
try:
dataset = Dataset.from_hub(DATASET_NAME)
logger.info(f"Existing dataset loaded: {DATASET_NAME}")
return dataset
except Exception:
logger.info(f"Dataset {DATASET_NAME} not found, creating new dataset...")
# Create new empty dataset
empty_dataset = Dataset.from_dict({
"id": [],
"file_name": [],
"meeting_date": [],
"transcription": [],
"summary": [],
"topics": [],
"keywords": [],
"created_at": []
})
logger.info(f"New dataset created: {DATASET_NAME}")
return empty_dataset
except Exception as e:
logger.error(f"Error in creating/loading dataset: {str(e)}")
return None
def load_meetings_from_dataset(hf_token: Optional[str] = None) -> Optional[list]:
"""
Load all meetings from dataset.
Args:
hf_token (Optional[str]): Hugging Face token
Returns:
Optional[list]: List of meetings or None if error
"""
if Dataset is None:
logger.error("datasets not installed")
return None
try:
# Authentication if token provided
if hf_token:
try:
login(token=hf_token)
except Exception as e:
logger.warning(f"Error in HF authentication: {str(e)}")
# Load dataset
dataset = Dataset.from_hub(DATASET_NAME)
# Convert to list
meetings = list(dataset)
logger.info(f"Loaded {len(meetings)} meetings from dataset")
return meetings
except Exception as e:
logger.error(f"Error loading meetings: {str(e)}")
return None
def get_dataset_info() -> Dict:
"""
Return dataset information.
Returns:
Dict: Dataset information
"""
return {
"dataset_name": DATASET_NAME,
"description": "Dataset for persisting analyzed meetings",
"fields": [
"id", "file_name", "meeting_date", "transcription",
"summary", "topics", "keywords", "created_at"
]
}
|