"""Corpus domain models: tenants, workspaces, documents, chunks, extractions.
"""
from __future__ import annotations
import re
import uuid
from pathlib import Path
from django.db import models
from django.utils import timezone
from agency.utils import sha256_file
[docs]
def slugify_for_collection(value: str) -> str:
"""
Slug simple y estable para nombres de colección.
Chroma suele tolerar bastante, pero conviene restringir a [a-z0-9_:-].
"""
value = (value or "").strip().lower()
value = re.sub(r"\s+", "_", value)
value = re.sub(r"[^a-z0-9_:\-]", "", value)
return value[:80] or "default"
[docs]
class Tenant(models.Model):
"""
Organización / cliente. Límite duro de aislamiento de datos.
"""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
name = models.CharField(max_length=200)
slug = models.SlugField(max_length=80, unique=True)
is_active = models.BooleanField(default=True)
created_at = models.DateTimeField(auto_now_add=True)
def __str__(self) -> str:
return self.name
[docs]
class TenantMembership(models.Model):
"""Links an operator to a tenant for corpus / workspace access."""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
tenant = models.ForeignKey(
Tenant,
on_delete=models.CASCADE,
related_name="memberships",
)
user = models.ForeignKey(
"auth.User",
on_delete=models.CASCADE,
related_name="tenant_memberships",
)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
constraints = [
models.UniqueConstraint(fields=["tenant", "user"], name="uniq_tenant_user"),
]
indexes = [
models.Index(fields=["user", "tenant"]),
]
def __str__(self) -> str:
return f"{self.user_id}@{self.tenant.slug}"
[docs]
class Workspace(models.Model):
"""
Espacio dentro de un tenant (proyecto/departamento).
Normalmente: 1 colección por workspace (recomendado) o por tenant+workspace.
"""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE, related_name="workspaces")
name = models.CharField(max_length=200)
slug = models.SlugField(max_length=80)
description = models.TextField(blank=True, default="")
# Opcional: control de configuración por workspace
default_embedding_model = models.CharField(max_length=120, default="mxbai-embed-large")
default_chunk_size = models.PositiveIntegerField(default=256)
default_chunk_overlap = models.PositiveIntegerField(default=64)
created_at = models.DateTimeField(auto_now_add=True)
def __str__(self) -> str:
return f"{self.tenant.slug}/{self.slug}"
[docs]
def collection_name(self, env: str = "prod", schema_version: int = 1) -> str:
"""
Nombre determinista de colección.
- env: dev/prod/staging
- schema_version: súbelo si cambias metadatos/estrategia de ids
"""
emb = slugify_for_collection(self.default_embedding_model)
return f"rag:{env}:t_{self.tenant.slug}:w_{self.slug}:emb_{emb}:v{schema_version}"
[docs]
class Document(models.Model):
""" A document is a file or a URL that is indexed by the system.
"""
[docs]
class Status(models.TextChoices):
UPLOADED = "UPLOADED", "Uploaded"
PARSING = "PARSING", "Parsing"
INDEXING = "INDEXING", "Indexing"
INDEXED = "INDEXED", "Indexed"
FAILED = "FAILED", "Failed"
DELETED = "DELETED", "Deleted"
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
workspace = models.ForeignKey(Workspace, on_delete=models.CASCADE, related_name="documents")
title = models.CharField(max_length=255)
# Si guardas archivos en Django:
file = models.FileField(upload_to="corpus/", blank=True, null=True)
# Si a veces es URL en vez de file:
source_url = models.URLField(blank=True, default="")
summary = models.TextField(blank=True, default="")
# Para detectar cambios y reindexar:
checksum_sha256 = models.CharField(max_length=64, blank=True, default="")
# Config de indexado efectiva (congelada por doc, útil si luego cambias defaults del workspace)
#embedding_model = models.CharField(max_length=120, default="mxbai-embed-large")
#chunk_size = models.PositiveIntegerField(default=256)
#chunk_overlap = models.PositiveIntegerField(default=64)
# Colección objetivo en Chroma (muy útil para auditoría/borrado/reindex)
#collection_name = models.CharField(max_length=255, blank=True, default="")
status = models.CharField(max_length=20, choices=Status.choices, default=Status.UPLOADED)
error_message = models.TextField(blank=True, default="")
# Métricas (opcionales, pero ayudan)
page_count = models.PositiveIntegerField(blank=True, null=True)
chunk_count = models.PositiveIntegerField(blank=True, null=True)
indexed_at = models.DateTimeField(blank=True, null=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
indexes = [
models.Index(fields=["workspace", "status"]),
models.Index(fields=["workspace", "created_at"]),
models.Index(fields=["checksum_sha256"]),
]
def __str__(self) -> str:
return self.title
[docs]
def checksum_changes(self) -> bool:
file_path = Path(self.file.path)
checksum = sha256_file(file_path)
if checksum == self.checksum_sha256:
return False
else:
self.checksum_sha256 = checksum
self.save(update_fields=["checksum_sha256", "updated_at"])
return True
[docs]
def mark_indexed(self, chunk_count: int | None = None) -> None:
self.status = self.Status.INDEXED
self.indexed_at = timezone.now()
if chunk_count is not None:
self.chunk_count = chunk_count
self.error_message = ""
[docs]
class Section(models.Model):
"""One section can be se
"""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
extraction = models.ForeignKey(Extraction, on_delete=models.CASCADE, related_name="sections")
title = models.CharField(max_length=255)
[docs]
class Chunk(models.Model):
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
extract = models.ForeignKey(Extract, on_delete=models.CASCADE, related_name="chunks")
section = models.ForeignKey(Section, on_delete=models.CASCADE, related_name="chunks", null=True, blank=True)
metadata = models.JSONField()
text = models.TextField()
clear = models.BooleanField(default=True)