Spaces:
Running
Running
| """Bounded document attachments for the existing Sofiel conversation pipeline.""" | |
| from __future__ import annotations | |
| import copy | |
| import hashlib | |
| from pathlib import Path | |
| import re | |
| from uuid import uuid4 | |
| from sofiel_documents import DocumentError, Limits, packed | |
| def attachment_name(path): | |
| """A short, single-line name; never show a server path in the conversation.""" | |
| name = Path(path).name | |
| return re.sub(r"[\x00-\x1f\x7f<>`\[\]*_#]", "", name)[:160] or "documento" | |
| def freeze_document(path, directory, *, encoding=None, delimiter=None, limits=None): | |
| """Snapshot the exact submitted bytes before clearing/locking the composer.""" | |
| limits = limits or Limits.from_env() | |
| source = Path(path) | |
| if source.suffix.lower() not in {".txt", ".md", ".csv", ".docx", ".pdf"}: | |
| raise DocumentError("unsupported_format") | |
| if not source.is_file() or source.stat().st_size > limits.file_bytes: | |
| raise DocumentError("file_size_or_missing") | |
| attachment_id = "doc_" + uuid4().hex | |
| target_dir = Path(directory) / attachment_id | |
| target_dir.mkdir() | |
| target = target_dir / attachment_name(source) | |
| # Keep the extension even when a pathological original name was truncated. | |
| if target.suffix.lower() != source.suffix.lower(): | |
| target = target_dir / (target.stem[:140] + source.suffix.lower()) | |
| sha = hashlib.sha256() | |
| total = 0 | |
| try: | |
| with source.open("rb") as incoming, target.open("xb") as outgoing: | |
| while chunk := incoming.read(65536): | |
| total += len(chunk) | |
| if total > limits.file_bytes: | |
| raise DocumentError("file_size_limit") | |
| sha.update(chunk) | |
| outgoing.write(chunk) | |
| except BaseException: | |
| target.unlink(missing_ok=True) | |
| target_dir.rmdir() | |
| raise | |
| return {"attachment_id": attachment_id, "path": str(target), | |
| "name": attachment_name(source), "sha256": sha.hexdigest(), | |
| "encoding": encoding or None, "delimiter": delimiter or None} | |
| def document_evidence(owner, attachment, result, human_text, *, max_chars=11000): | |
| """Focus on the submitted source even for 'read this' or a file-only turn. | |
| Retrieval remains unchanged for later conversation. This bounded envelope | |
| carries external passages and their attribution, never a human utterance. | |
| Caller holds the owner's state lock. | |
| """ | |
| store = owner.memory.data.get("documents", {}) | |
| source_id = "src_" + attachment["sha256"] | |
| source = store.get("sources", {}).get(source_id) | |
| if source is None: | |
| raise DocumentError("submitted_source_missing") | |
| if store.get('schema_version') == 2: | |
| docs = owner.document_service.evidence(human_text, focus=source_id, max_chars=max_chars-1200) | |
| if not docs: | |
| docs = {'revision':store['revision'],'instance_id':store['instance_id'],'passages':[],'knowledge':[]} | |
| docs['upload_event'] = {'kind':'document_upload','source_id':source_id,'file_name':attachment['name'], | |
| 'content_is_user_message':False,'participant_message_provided':bool(human_text.strip()), | |
| 'extraction_status':result.extraction_status,'interpretation_status':result.interpretation_status, | |
| 'incorporated':result.incorporated,'pending_segments':result.pending_segments,'error_code':result.error_code, | |
| 'coverage':copy.deepcopy(source['coverage'])} | |
| return docs | |
| event = {"kind": "document_upload", "source_id": source_id, | |
| "file_name": attachment["name"], "content_is_user_message": False, | |
| "participant_message_provided": bool(human_text.strip()), | |
| "extraction_status": result.extraction_status, | |
| "interpretation_status": result.interpretation_status, | |
| "incorporated": result.incorporated, | |
| "pending_segments": result.pending_segments, | |
| "error_code": result.error_code, | |
| "coverage": copy.deepcopy(source["coverage"])} | |
| docs = {"revision": store["revision"], "instance_id": store["instance_id"], | |
| "retrieval": "submitted_document", "upload_event": event, | |
| "passages": [], "knowledge": []} | |
| terms = set(re.findall(r"\w{3,}", human_text.casefold())) | |
| segments = [s for s in store["segments"].values() if s["source_id"] == source_id] | |
| segments.sort(key=lambda s: (-len(terms & set(re.findall(r"\w{3,}", s["text"].casefold()))), | |
| s["locator"]["unit"], s["char_start"])) | |
| for segment in segments: | |
| candidate = copy.deepcopy(docs) | |
| candidate["passages"].append({**copy.deepcopy(segment), "source_name": source["name"], | |
| "declared_authors": copy.deepcopy(source["declared_authors"])}) | |
| if len(packed(candidate)) <= max_chars: | |
| docs = candidate | |
| if len(docs["passages"]) >= 4: | |
| break | |
| retained = {p["segment_id"] for p in docs["passages"]} | |
| for knowledge in store["knowledge"].values(): | |
| if (knowledge["source_id"] == source_id and knowledge["state"] == "incorporated" | |
| and all(r["segment_id"] in retained for r in knowledge["support_refs"])): | |
| candidate = copy.deepcopy(docs) | |
| candidate["knowledge"].append(copy.deepcopy(knowledge)) | |
| if len(packed(candidate)) <= max_chars: | |
| docs = candidate | |
| if len(packed(docs)) > max_chars: | |
| raise DocumentError("evidence_budget_exceeded") | |
| return docs | |
| def document_status(result): | |
| if result.status == "failed": | |
| return "No se pudo leer el archivo. Se conservaron el adjunto y el mensaje. " + (result.error_code or "") | |
| details = f"Archivo recibido 路 {result.incorporated} conocimientos incorporados." | |
| if result.extraction_status == "partial" or result.interpretation_status != "complete": | |
| details += " Lectura parcial; pod茅s continuarla desde Documentos y procedencia." | |
| if result.error_code: | |
| details += " No se complet贸 la interpretaci贸n: " + result.error_code + "." | |
| return details | |