Spaces:
Running
Running
File size: 6,086 Bytes
2e1f79e | 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 | """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
|