import json import re from datetime import datetime, timezone from pymongo import ASCENDING, IndexModel from pymongo.errors import CollectionInvalid NOTES_COLLECTION_NAME = "scapp_notas" MAX_BATCH_SIZE = 50 MAX_QUERY_NOTES = 5000 MAX_NOTE_TEXT_LENGTH = 10_000 MAX_NOTE_METADATA_BYTES = 64 * 1024 MAX_NOTE_TAGS = 50 MAX_NOTE_TAG_LENGTH = 100 MAX_NOTES_REQUEST_BYTES = 8 * 1024 * 1024 def _coerce_datetime(value, *, local_naive=False): if isinstance(value, datetime): parsed = value elif isinstance(value, str): raw_value = value.strip() if raw_value.endswith("Z"): raw_value = raw_value[:-1] + "+00:00" try: parsed = datetime.fromisoformat(raw_value) except ValueError as exc: raise ValueError("datetime invalido") from exc else: raise ValueError("datetime invalido") if parsed.tzinfo is None: return parsed if local_naive: return parsed.replace(tzinfo=None) return parsed.astimezone(timezone.utc).replace(tzinfo=None) def ensure_notes_collection(scapp_db, collection_name=NOTES_COLLECTION_NAME): if collection_name not in scapp_db.list_collection_names(): try: scapp_db.create_collection(collection_name) except CollectionInvalid: pass collection = scapp_db[collection_name] # Los tombstones no usan TTL: conservar la versión evita que un replay antiguo # pueda resucitar una nota después de eliminar su registro de borrado. collection.create_indexes([ IndexModel( [("centro_id", ASCENDING), ("note_id", ASCENDING)], unique=True, name="idx_scapp_notas_centro_note", background=True, ), IndexModel( [("centro_id", ASCENDING), ("timestamp", ASCENDING)], name="idx_scapp_notas_centro_timestamp", background=True, ), IndexModel( [("centro_id", ASCENDING), ("updated_at", ASCENDING)], name="idx_scapp_notas_centro_updated", background=True, ), ]) return collection def validate_and_normalize_notes_payload(payload): docs = [payload] if isinstance(payload, dict) else payload if not isinstance(docs, list) or not docs: raise ValueError("Payload debe ser un dict o una lista no vacia") if len(docs) > MAX_BATCH_SIZE: raise ValueError(f"Payload excede el maximo de {MAX_BATCH_SIZE} notas") normalized = [] for doc in docs: if not isinstance(doc, dict): raise ValueError(f"Documento invalido: {doc!r}") note_id = str(doc.get("note_id") or "").strip() if not note_id: raise ValueError("Falta 'note_id' en alguno de los documentos") raw_cam_ids = doc.get("cam_ids") if not isinstance(raw_cam_ids, list): raise ValueError("'cam_ids' debe ser una lista") cam_ids = list(dict.fromkeys( str(camera_id).strip() for camera_id in raw_cam_ids if str(camera_id).strip() )) if not cam_ids: raise ValueError("Falta 'cam_ids' en alguno de los documentos") text = doc.get("text") if not isinstance(text, str) or not text.strip(): raise ValueError("Falta 'text' en alguno de los documentos") if len(text) > MAX_NOTE_TEXT_LENGTH: raise ValueError(f"'text' excede el maximo de {MAX_NOTE_TEXT_LENGTH} caracteres") tags = doc.get("tags", []) metadata = doc.get("metadata", {}) deleted = doc.get("deleted", False) if not isinstance(tags, list): raise ValueError("'tags' debe ser una lista") if len(tags) > MAX_NOTE_TAGS: raise ValueError(f"'tags' excede el maximo de {MAX_NOTE_TAGS} elementos") if any( not isinstance(tag, str) or not tag.strip() or len(tag) > MAX_NOTE_TAG_LENGTH for tag in tags ): raise ValueError( f"cada tag debe ser texto no vacio de hasta {MAX_NOTE_TAG_LENGTH} caracteres" ) if not isinstance(metadata, dict): raise ValueError("'metadata' debe ser un objeto") metadata_size = len( json.dumps(metadata, ensure_ascii=False, separators=(",", ":")).encode("utf-8") ) if metadata_size > MAX_NOTE_METADATA_BYTES: raise ValueError(f"'metadata' excede el maximo de {MAX_NOTE_METADATA_BYTES} bytes") if not isinstance(deleted, bool): raise ValueError("'deleted' debe ser booleano") created_at = doc.get("created_at") normalized.append({ "note_id": note_id, "cam_ids": cam_ids, # Contrato SCAPP: timestamp conserva la hora local escrita, sin reinterpretar zona. "timestamp": _coerce_datetime(doc.get("timestamp"), local_naive=True), "text": text.strip(), "tags": tags, "metadata": metadata, "deleted": deleted, "created_at": _coerce_datetime(created_at) if created_at is not None else None, "updated_at": _coerce_datetime(doc.get("updated_at")), }) return normalized def save_notes_payload(scapp_db, docs, centro_id): collection = ensure_notes_collection(scapp_db) received_at = datetime.now(timezone.utc).replace(tzinfo=None) for doc in docs: replacement = { **doc, "centro_id": centro_id, "received_at": received_at, } # Cada nota se reemplaza atómicamente solo si la versión entrante no es más antigua. # El sender limita lotes a 50; si ese techo crece, migrar este loop a bulk_write(UpdateOne). collection.update_one( {"centro_id": centro_id, "note_id": doc["note_id"]}, [{ "$replaceWith": { "$cond": [ { "$or": [ {"$eq": [{"$type": "$updated_at"}, "missing"]}, {"$lte": ["$updated_at", doc["updated_at"]]}, ] }, {"$literal": replacement}, "$$ROOT", ] } }], upsert=True, ) return len(docs) def serialize_note(note): serialized = { key: value.isoformat() if isinstance(value, datetime) else value for key, value in note.items() if key != "_id" } # El frontend local identifica notas con ``_id``; se conserva ``note_id`` como clave cloud. serialized["_id"] = serialized.get("note_id") return serialized def fetch_notes_for_range(collection, start_day, end_day, *, centro_id=None, scope_prefix=None): if not centro_id and not scope_prefix: raise ValueError("Se requiere centro_id o scope_prefix") query = { "timestamp": {"$gte": start_day, "$lt": end_day}, "deleted": {"$ne": True}, } if centro_id: query["centro_id"] = centro_id else: query["centro_id"] = {"$regex": f"^{re.escape(scope_prefix)}-"} cursor = ( collection.find(query, {"_id": 0}) .sort([ ("timestamp", ASCENDING), ("centro_id", ASCENDING), ("note_id", ASCENDING), ]) .limit(MAX_QUERY_NOTES) ) return [serialize_note(note) for note in cursor]