U ^j.@sddlmZddlmZmZmZddlmZmZddlm Z m Z m Z m Z m Z dZdZdZdd Zd d Zd d ZddZdddZddZddZdddZdS)) MongoClient) PyMongoErrorCollectionInvalidBulkWriteError)datetime timedelta)DictAnyListUnionOptionalFLUX__TSi,z --cCsP|d}||krHz|j|ddddWntk rFYnX||S)uF Helper para obtener/crear la colección TimeSeries correcta. _flux_ts timestampminutes) timeField granularity) timeseries)lowerlist_collection_namescreate_collectionr)db centro_idcollection_namer7/var/www/itgmanager/itgmanager/nanoox/nanoox_bolt_ts.pyget_flux_collection s  rcCs||d}|stdzt|d}Wn"tk rHtd|YnX|d}||d|d|dt|d S) zN Helper para validar y transformar un raw json en un documento mongo. rzCampo 'timestamp' faltante%Y-%m-%d %H:%M:%SuFormato fecha inválido: idz Flow ratezCumulant float point nummber connected)r flow_ratecumulantr received_at original_id)get ValueErrorrstrptimeutcnow)itemZ raw_timestampZ dt_objectZ external_idrrrparse_flux_document#s  r)c Csjz*|t}t||}t|}||WdStk rd}ztdt|WYdSd}~XYnXdS)u Procesa UN solo registro (Modo Síncrono) con manejo de errores. - Si tiene éxito: Retorna None. - Si falla: Retorna False. Nz[ERROR] save_flux_single: F)FLUX_DATABASE_NAMErr) insert_one Exceptionprintstr)clientdatarr collectiondocerrrsave_flux_single=s  r4c Cs|t}t||}g}g}|D]}zFt|}|||d} | dk rft| tr\|| n || Wqtk r} ztd| WYqW5d} ~ XYqXq|rz|j |ddWn2t k r} ztd| j W5d} ~ XYnX|S)u Procesa un ARRAY de registros (Modo Asíncrono). Siempre guarda. Retorna lista PLANA UNIDIMENSIONAL de IDs encontrados. r#Nz3Advertencia: Saltando item corrupto en batch Flux: F)orderedzAdvertencia Bulk: ) r*rr)append isinstancelistextendr%r- insert_manyrdetails) r/ data_listrrr1docs_to_insertZ ids_to_ackr(r2oidr3bwerrrsave_flux_batchWs,    "r@hc Cs|gggtgdi}z:||kr*|WS||}t}|jddddd}|tdd} d|| di} dddddd} || | dd |} t | } | s|WSt t | }|d d |}g}g}g}g}|D]b}| d}t |tr| nt|}|| d || d || d ||q||||d}||i}|WStk r}ztd||WYSd }~XYnXd S)u Obtiene datos históricos de Flux. Si no hay datos, retorna la estructura vacía estandarizada. Si el último dato es > 5 min, agrega estado desconectado al final. r r!Zconnection_statusTSrhourminutesecond microsecondrBdaysrz$gtez$ltr r!rr_idNr r!r"Error obteniendo historicos Flux: )DEFAULT_LAST_SYNC_TIMErrnowreplacerfindsortlimitr8reversedr$r7 isoformatr.r6r,r-)rr sensor_keyNstep empty_resultr1ahora inicio_hoy inicio_mananaquery projectioncursordocsdocs_asc docs_sampledflow_rate_vals cumulant_valsconnected_valsts_valsr2Zraw_tsts_strZvectors_package resultador3rrrget_flux_historical_datash        rlc Cst}z||kr|WS||}t}|jddddd}|tdd}|jd||didgdddd }|rd|kr|d}t|tr|d WSt |WS|WSt k r} zt d | |WYSd } ~ XYnXd S) u Busca el último registro recibido HOY para determinar el estado de conexión. Retorna la fecha formateada o ' --' si no hay datos. rrErBrJrrL)rrO)rrN)rUrarz#Advertencia en get_flux_heartbeat: N) rQrrrRrSrfind_oner7strftimer.r,r-) rrZ hora_defaultr1r]r^r_Zlast_heartbeat_docZrec_atr3rrrget_flux_heartbeats0     rocCs>t|tr|St|tr |Std|dt|ddS)ud @brief Verifica si un timestamp es un objeto datetime y lo convierte a ISO 8601. @param timestamp Puede ser un objeto datetime o una cadena. @return string con el timestamp en formato ISO 8601 si es datetime. Si ya es string, se devuelve sin cambios. @throws ValueError si el timestamp no es un datetime o string válido. u Formato de timestamp inválido: z ()N)r7rrXr.r%type)rrrrserialize_timestamp s  rrc Cs|dgdgdgtgdi}z>||kr0|WS||}t}|jddddd}|tdd} d|| d i} dddddd } || | dd |} t | } | s|WSt t | }|dd|}g}g}g}g}|D]j}t | dtr| d n t| d}|| d || d || d||q|||||di}|WStk r}ztd||WYSd}~XYnXdS)um Obtiene datos históricos de Flux. Si no hay datos, retorna la estructura vacía estandarizada. NFrCrrErBrJrrLrMrOr r!rrP)rQrrrRrSrrTrUrVr8rWr7r$rXr.r6r,r-)rrrYrZr[r\r1r]r^r_r`rarbrcrdrerfrgrhrir2rjrkr3rrrlegacy_get_flux_historical_datasd    ,  rsN)rArB)rArB)pymongorpymongo.errorsrrrrrtypingrr r r r r*ZUMBRAL_SEGUNDOSrQrr)r4r@rlrorrrsrrrrs ( _+