diff --git a/medcat-trainer/webapp/api/api/data_utils.py b/medcat-trainer/webapp/api/api/data_utils.py
index e22e5b89b..20ffeab4e 100644
--- a/medcat-trainer/webapp/api/api/data_utils.py
+++ b/medcat-trainer/webapp/api/api/data_utils.py
@@ -73,6 +73,221 @@ def delete_orphan_docs(dataset: Dataset):
Document.objects.filter(dataset__id=dataset.id).delete()
+def _prepare_state(proj: dict) -> tuple[set, dict, set, set, dict]:
+ # ensure current deployment has the neccessary - Entity, MetaTask, Relation, and warn on not present User objects.
+ ent_labels, meta_tasks, rels, unavailable_users, available_users = set(), defaultdict(set), set(), set(), dict()
+ for doc in proj['documents']:
+ for anno in doc['annotations']:
+ ent_labels.add(anno['cui'])
+ for meta_anno in anno['meta_anns'].values():
+ meta_tasks[meta_anno['name']].add(meta_anno['value'])
+ user_obj = User.objects.filter(username=anno['user']).first()
+ if user_obj is None:
+ unavailable_users.add(anno['user'])
+ elif anno['user'] not in available_users:
+ available_users[anno['user']] = user_obj
+ for rel in doc.get('relations', []):
+ rels.add(rel['relation'])
+ return ent_labels, meta_tasks, rels, unavailable_users, available_users
+
+
+def _init_proj_ann_ents(
+ proj: dict,
+ modelpack_id: str,
+ cdb_id: str,
+ project_name_suffix: str,
+ vocab_id: str
+) -> ProjectAnnotateEntities:
+ p = ProjectAnnotateEntities()
+ p.name = f"{proj['name']}{project_name_suffix}"
+ if len(proj['cuis']) > 1000:
+ # store large CUI lists in a json file.
+ cuis_file_name = MEDIA_ROOT + '/' + re.sub('/|\.', '_', p.name + '_cuis_file') + '.json'
+ json.dump(proj["cuis"].split(','), open(cuis_file_name, 'w'))
+ p.cuis = ""
+ p.cuis_file.name = cuis_file_name
+ else:
+ p.cuis = proj['cuis']
+
+ if cdb_id is not None and vocab_id is not None:
+ p.concept_db = ConceptDB.objects.get(id=cdb_id)
+ p.vocab = Vocabulary.objects.get(id=vocab_id)
+ elif modelpack_id is not None:
+ p.model_pack = ModelPack.objects.get(id=modelpack_id)
+ else:
+ raise InvalidParameterError("No cdb, vocab, or modelpack provided")
+ return p
+
+
+def _create_dataset(proj: dict, p: ProjectAnnotateEntities) -> tuple[Dataset, dict[str, str]]:
+ # escape - filename
+ ds_file_name = MEDIA_ROOT + '/' + re.sub('/|\.', '_', p.name + '_dataset') + '.csv'
+ names = [doc['name'] for doc in proj['documents']]
+ if len(set(names)) != len(names): # ensure names are unique for docs
+ names = [f'{i} - {names[i]}' for i in range(len(names))]
+ pd.DataFrame({'name': names,
+ 'text': [doc['text'] for doc in proj['documents']]}).to_csv(ds_file_name)
+ ds_mod = Dataset()
+ ds_mod.name = p.name + '_dataset'
+ ds_mod.original_file.name = ds_file_name
+ ds_mod.save()
+ p.dataset = ds_mod
+ # creating text 2 name mapping so we can find the doucments based on the name
+ # even if the text has been processed through pandas conversion and/or sanitisation
+ text2name: dict[str, str] = {
+ doc["text"]: name
+ for doc, name in zip(proj['documents'], names)
+ }
+ return ds_mod, text2name
+
+
+def _upload_project(
+ proj: dict,
+ cdb_id: str,
+ vocab_id: str,
+ modelpack_id: str,
+ project_name_suffix: str = ' IMPORTED',
+ cdb_search_filter_id: str = None,
+ members: List[str] = None,
+ set_validated_docs: bool = False,
+):
+ if len(proj['documents']) == 0:
+ # don't add projects with no documents
+ return
+ p = _init_proj_ann_ents(proj, modelpack_id, cdb_id, project_name_suffix, vocab_id)
+
+ ent_labels, meta_tasks, rels, unavailable_users, available_users = _prepare_state(proj)
+
+ ds_mod, text2name = _create_dataset(proj, p)
+
+ p.save()
+
+ if cdb_search_filter_id is not None:
+ p.cdb_search_filter.set([ConceptDB.objects.get(id=cdb_search_filter_id)])
+
+ if members is not None:
+ p.members.set(members)
+
+ # create django ORM model instances that are referenced in the upload if they don't exist.
+ for u in unavailable_users:
+ logger.warning(f'Username: {u} - not present in this trainer deployment.')
+ for ent_lab in ent_labels:
+ ent = Entity.objects.filter(label=ent_lab).first()
+ if ent is None:
+ ent = Entity()
+ ent.label = ent_lab
+ ent.save()
+ for task in meta_tasks:
+ if MetaTask.objects.filter(name=task).first() is None:
+ m_task = MetaTask()
+ m_task.name = task
+ m_task.save()
+ # create the MetaTask Values.
+ for task_val in meta_tasks[task]:
+ if MetaTaskValue.objects.filter(name=task_val).first() is None:
+ mt_value = MetaTaskValue()
+ mt_value.name = task_val
+ mt_value.save()
+ m_task = MetaTask.objects.filter(name=task).first()
+ curr_vals = m_task.values.all()
+ task_vals = [MetaTaskValue.objects.filter(name=m_t).first() for m_t in meta_tasks[task]]
+ m_task.values.set(set(list(curr_vals) + task_vals))
+
+ for rel in rels:
+ if Relation.objects.filter(label=rel).first() is None:
+ r = Relation()
+ r.label = rel
+ r.save()
+
+ if set_validated_docs:
+ p.validated_documents.set(list(Document.objects.filter(dataset=ds_mod)))
+ else:
+ p.validated_documents.clear()
+
+
+ for doc in proj['documents']:
+ _process_doc(doc, p, ds_mod, available_users, text2name)
+
+
+def _process_doc(
+ doc: dict,
+ p: ProjectAnnotateEntities,
+ ds_mod: Dataset,
+ available_users: dict,
+ text2name: dict[str, str],
+):
+ # NOTE: using text2name mapping here since the text in the database
+ # can be sanitised and changed during pandas serialisation (and deserialisation),
+ # however we've kpet track of per-text names (which are unique) so will use these
+ # instead here
+ doc_mod = Document.objects.filter(Q(dataset=ds_mod) & Q(name=text2name[doc['text']])).first()
+ annos = []
+ for anno in doc['annotations']:
+ a = AnnotatedEntity()
+ a.user = available_users[anno['user']]
+ a.project = p
+ a.document = doc_mod
+ e = Entity.objects.get(label=anno['cui'])
+ a.entity = e
+ a.value = anno['value']
+ a.start_ind = anno['start']
+ a.end_ind = anno['end']
+ a.validated = anno['validated']
+ a.correct = anno['correct']
+ a.deleted = anno['deleted']
+ a.alternative = anno['alternative']
+ a.killed = anno['killed']
+ a.irrelevant = anno.get('irrelevant', False) # Added later - so False by default for compatibility
+ if anno.get('last_modified') is not None:
+ try:
+ a.last_modified = datetime.strptime(anno['last_modified'], _dt_fmt)
+ except ValueError:
+ a.last_modified = datetime.now()
+ if anno.get('create_time') is not None:
+ try:
+ a.create_time = datetime.strptime(anno['create_time'], _dt_fmt)
+ except ValueError:
+ a.create_time = datetime.now()
+ a.comment = anno.get('comment')
+ a.manually_created = anno['manually_created']
+
+ a.acc = anno['acc']
+ a.save()
+ annos.append(a)
+ for task_name, meta_anno in anno['meta_anns'].items():
+ m_a = MetaAnnotation()
+ m_a.annotated_entity = a
+ # there will be at least one or more of these available.
+ m_a.meta_task = MetaTask.objects.filter(name=task_name).first()
+ m_a.validated = meta_anno['validated']
+ m_a.acc = meta_anno['acc']
+ m_a.meta_task_value = MetaTaskValue.objects.filter(name=meta_anno['value']).first()
+ m_a.save()
+ # missing acc on the model
+ anno_to_doc_ind = {a.start_ind: a for a in annos}
+
+ for relation in doc.get('relations', []):
+ er = EntityRelation()
+ er.user = available_users[relation['user']]
+ er.project = p
+ er.document = doc_mod
+ # there will be at least one or more of these available.
+ er.relation = Relation.objects.filter(label=relation['relation']).first()
+ er.validated = er.validated
+ # link relations with start and end anno ents
+ er.start_entity = anno_to_doc_ind[relation['start_entity_start_idx']]
+ er.end_entity = anno_to_doc_ind[relation['end_entity_start_idx']]
+ if relation.get('create_time') is not None:
+ er.create_time = datetime.strptime(relation['create_time'], _dt_fmt)
+ else:
+ er.create_time = datetime.now()
+ if relation.get('last_modified_time') is not None:
+ er.last_modified = datetime.strptime(relation['last_modified_time'], _dt_fmt)
+ else:
+ er.last_modified = datetime.now()
+ er.save()
+
+
def upload_projects_export(
medcat_export: Dict,
cdb_id: str,
@@ -82,168 +297,11 @@ def upload_projects_export(
cdb_search_filter_id: str = None,
members: List[str] = None,
import_project_name_suffix: str = ' IMPORTED',
- set_validated_docs: bool = False
+ set_validated_docs: bool = False,
):
for proj in medcat_export['projects']:
- if len(proj['documents']) == 0:
- # don't add projects with no documents
- continue
- p = ProjectAnnotateEntities()
- p.name = f"{proj['name']}{project_name_suffix}"
- if len(proj['cuis']) > 1000:
- # store large CUI lists in a json file.
- cuis_file_name = MEDIA_ROOT + '/' + re.sub('/|\.', '_', p.name + '_cuis_file') + '.json'
- json.dump(proj["cuis"].split(','), open(cuis_file_name, 'w'))
- p.cuis = ""
- p.cuis_file.name = cuis_file_name
- else:
- p.cuis = proj['cuis']
-
- if cdb_id is not None and vocab_id is not None:
- p.concept_db = ConceptDB.objects.get(id=cdb_id)
- p.vocab = Vocabulary.objects.get(id=vocab_id)
- elif modelpack_id is not None:
- p.model_pack = ModelPack.objects.get(id=modelpack_id)
- else:
- raise InvalidParameterError("No cdb, vocab, or modelpack provided")
-
- # ensure current deployment has the neccessary - Entity, MetaTak, Relation, and warn on not present User objects.
- ent_labels, meta_tasks, rels, unavailable_users, available_users = set(), defaultdict(set), set(), set(), dict()
- for doc in proj['documents']:
- for anno in doc['annotations']:
- ent_labels.add(anno['cui'])
- for meta_anno in anno['meta_anns'].values():
- meta_tasks[meta_anno['name']].add(meta_anno['value'])
- user_obj = User.objects.filter(username=anno['user']).first()
- if user_obj is None:
- unavailable_users.add(anno['user'])
- elif anno['user'] not in available_users:
- available_users[anno['user']] = user_obj
- for rel in doc.get('relations', []):
- rels.add(rel['relation'])
- # escape - filename
- ds_file_name = MEDIA_ROOT + '/' + re.sub('/|\.', '_', p.name + '_dataset') + '.csv'
- names = [doc['name'] for doc in proj['documents']]
- if len(set(names)) != len(names): # ensure names are unique for docs
- names = [f'{i} - {names[i]}' for i in range(len(names))]
- pd.DataFrame({'name': names,
- 'text': [doc['text'] for doc in proj['documents']]}).to_csv(ds_file_name)
- ds_mod = Dataset()
- ds_mod.name = p.name + '_dataset'
- ds_mod.original_file.name = ds_file_name
- ds_mod.save()
- p.dataset = ds_mod
- p.save()
-
- if cdb_search_filter_id is not None:
- p.cdb_search_filter.set([ConceptDB.objects.get(id=cdb_search_filter_id)])
-
- if members is not None:
- p.members.set(members)
-
- # create django ORM model instances that are referenced in the upload if they don't exist.
- for u in unavailable_users:
- logger.warning(f'Username: {u} - not present in this trainer deployment.')
- for ent_lab in ent_labels:
- ent = Entity.objects.filter(label=ent_lab).first()
- if ent is None:
- ent = Entity()
- ent.label = ent_lab
- ent.save()
- for task in meta_tasks:
- if MetaTask.objects.filter(name=task).first() is None:
- m_task = MetaTask()
- m_task.name = task
- m_task.save()
- # create the MetaTask Values.
- for task_val in meta_tasks[task]:
- if MetaTaskValue.objects.filter(name=task_val).first() is None:
- mt_value = MetaTaskValue()
- mt_value.name = task_val
- mt_value.save()
- m_task = MetaTask.objects.filter(name=task).first()
- curr_vals = m_task.values.all()
- task_vals = [MetaTaskValue.objects.filter(name=m_t).first() for m_t in meta_tasks[task]]
- m_task.values.set(set(list(curr_vals) + task_vals))
-
- for rel in rels:
- if Relation.objects.filter(label=rel).first() is None:
- r = Relation()
- r.label = rel
- r.save()
-
- if set_validated_docs:
- p.validated_documents.set(list(Document.objects.filter(dataset=ds_mod)))
- else:
- p.validated_documents.clear()
-
-
- for doc in proj['documents']:
- doc_mod = Document.objects.filter(Q(dataset=ds_mod) & Q(text=doc['text'])).first()
- annos = []
- for anno in doc['annotations']:
- a = AnnotatedEntity()
- a.user = available_users[anno['user']]
- a.project = p
- a.document = doc_mod
- e = Entity.objects.get(label=anno['cui'])
- a.entity = e
- a.value = anno['value']
- a.start_ind = anno['start']
- a.end_ind = anno['end']
- a.validated = anno['validated']
- a.correct = anno['correct']
- a.deleted = anno['deleted']
- a.alternative = anno['alternative']
- a.killed = anno['killed']
- a.irrelevant = anno.get('irrelevant', False) # Added later - so False by default for compatibility
- if anno.get('last_modified') is not None:
- try:
- a.last_modified = datetime.strptime(anno['last_modified'], _dt_fmt)
- except ValueError:
- a.last_modified = datetime.now()
- if anno.get('create_time') is not None:
- try:
- a.create_time = datetime.strptime(anno['create_time'], _dt_fmt)
- except ValueError:
- a.create_time = datetime.now()
- a.comment = anno.get('comment')
- a.manually_created = anno['manually_created']
-
- a.acc = anno['acc']
- a.save()
- annos.append(a)
- for task_name, meta_anno in anno['meta_anns'].items():
- m_a = MetaAnnotation()
- m_a.annotated_entity = a
- # there will be at least one or more of these available.
- m_a.meta_task = MetaTask.objects.filter(name=task_name).first()
- m_a.validated = meta_anno['validated']
- m_a.acc = meta_anno['acc']
- m_a.meta_task_value = MetaTaskValue.objects.filter(name=meta_anno['value']).first()
- m_a.save()
- # missing acc on the model
- anno_to_doc_ind = {a.start_ind: a for a in annos}
-
- for relation in doc.get('relations', []):
- er = EntityRelation()
- er.user = available_users[relation['user']]
- er.project = p
- er.document = doc_mod
- # there will be at least one or more of these available.
- er.relation = Relation.objects.filter(label=relation['relation']).first()
- er.validated = er.validated
- # link relations with start and end anno ents
- er.start_entity = anno_to_doc_ind[relation['start_entity_start_idx']]
- er.end_entity = anno_to_doc_ind[relation['end_entity_start_idx']]
- if relation.get('create_time') is not None:
- er.create_time = datetime.strptime(relation['create_time'], _dt_fmt)
- else:
- er.create_time = datetime.now()
- if relation.get('last_modified_time') is not None:
- er.last_modified = datetime.strptime(relation['last_modified_time'], _dt_fmt)
- else:
- er.last_modified = datetime.now()
- er.save()
+ _upload_project(
+ proj, cdb_id, vocab_id, modelpack_id, project_name_suffix,
+ cdb_search_filter_id, members, set_validated_docs)
logger.info(f"Finished annotation import for project {proj['name']}")
logger.info('Finished importing all projects')
diff --git a/medcat-trainer/webapp/api/api/tests/test_data_integrity.py b/medcat-trainer/webapp/api/api/tests/test_data_integrity.py
new file mode 100644
index 000000000..e5a22e905
--- /dev/null
+++ b/medcat-trainer/webapp/api/api/tests/test_data_integrity.py
@@ -0,0 +1,113 @@
+import json
+import os
+from unittest.mock import patch
+
+from django.test import TestCase
+from django.core.files.base import ContentFile
+from rest_framework.test import APIClient
+from api.models import Dataset, Document, ModelPack
+
+from ._helpers import create_user
+
+
+PROBLEM_PAYLOAD = {
+ "projects": [
+ {
+ "id": 1,
+ "name": "HTML-Tag-Challenge-Project",
+ "cuis": "",
+ "tuis": "",
+ "documents": [
+ {
+ "id": 50,
+ "name": "Document_With_HTML",
+ # The saboteur string: contains
tags
+ "text": "Patient presented with
acute chest pain.",
+ "annotations": [
+ {
+ "cui": "12345",
+ "value": "chest pain",
+ "start": 29, # This may shift if tags are stripped early
+ "end": 39,
+ "correct": True,
+ "validated": True,
+ "deleted": False,
+ "alternative": False,
+ "killed": False,
+ "manually_created": False,
+ "acc": 1.0,
+ "user": "admin",
+ "meta_anns": {}
+ }
+ ]
+ }
+ ]
+ }
+ ]
+}
+
+class MedCatTrainerImportTests(TestCase):
+
+ def _prepare_model_pack(self, name="data-integrity-test", mp_id: int = 10):
+ """Create a ModelPack with a fake unpacked dir (cdb dir + vocab file)."""
+ model_pack = ModelPack(name=name, id=mp_id)
+ model_pack.model_pack.save(f"{name}.zip", ContentFile(b"fake"), save=False)
+ unpacked = model_pack.model_pack.path[: -len(".zip")]
+ os.makedirs(os.path.join(unpacked, "cdb"), exist_ok=True)
+ with open(os.path.join(unpacked, "vocab"), "w", encoding="utf-8") as fh:
+ fh.write("")
+ self._register_model_pack(model_pack)
+ return model_pack, unpacked
+
+ def _register_model_pack(self, model_pack):
+ with patch("api.models.CAT.attempt_unpack"), \
+ patch("api.models.CDB.load"), \
+ patch("api.models.Vocab.load"), \
+ patch("api.utils._load_global_cnf_addon_cnfs", return_value=[]):
+ model_pack.save()
+
+ def setUp(self):
+ # 1. Create required database dependencies for the import
+ self.user = create_user(username='admin', password='password', is_staff=True)
+ self.client = APIClient()
+ self.client.force_login(self.user)
+
+ self.model_pack, self.unpacked = self._prepare_model_pack()
+
+ def test_upload_project_with_html_tags_in_text(self):
+ """
+ Ensures that importing a project containing document text with HTML tags
+ succeeds without throwing a NotNullViolation/IntegrityError.
+ """
+ # Define a project layout mirroring your 'bad' dataset payload structure
+
+ # Build the exact wrapper format expected by the API view / Admin form data
+ post_data = {
+ "exported_projects": PROBLEM_PAYLOAD,
+ "modelpack_id": self.model_pack.id,
+ "project_name_suffix": " TEST_IMPORT"
+ }
+ print("MP ID", self.model_pack.id)
+
+ # Target the API view directly (or change URL to test your Admin /add/ view form submission)
+ url = "/api/upload-deployment/"
+
+ print("\nš Executing import payload containing HTML formatting breaks...")
+
+ # Fire the request
+ response = self.client.post(
+ url,
+ data=json.dumps(post_data),
+ content_type="application/json",
+ )
+
+ # 2. Assertions
+ # If the bug is active, this will catch a 500 status code with an IntegrityError traceback
+ self.assertEqual(
+ response.status_code, 200,
+ msg=f"Upload failed! Server responded with status {response.status_code}: {response.content}"
+ )
+
+ # Verify the database successfully managed to store the documents and annotations
+ self.assertTrue(Dataset.objects.filter(name__contains="HTML-Tag-Challenge").exists())
+ self.assertTrue(Document.objects.filter(text__contains="chest pain").exists())