안녕하세요. 휴먼스케이프 june입니다.
이번시간엔 임상연구 업데이트 시 변경 부분을 감지하고, 해당 부분만을 업데이트하는 방법에 대해 알아보겠습니다.
프로젝트를 진행한 코드는 깃허브에 올라가 있으니 참고하시기 바랍니다.
실제 서비스하는 임상연구는 유료 번역api를 사용해 번역의 퀄리티를 높이기 때문에 관련 비용이 굉장히 많이 나오게 되고, 번역 태스크에서의 병목현상이 많이 일어나기 때문에 이 부분을 최적화하는게 굉장히 중요합니다.
진행 과정은 다음과 같습니다.
- convert태스크에서 변경된 부분을 확인합니다.
study model의 필드는 <updated>와 같은 값으로 채워 변경됨을 알리고, 연결된 모델은 인스턴스를 특정할 수 없는 고유 키가 없기 때문에(clinicaltrials에서 id값을 제공하지 않음) 모든 값이 같을 경우에만 유지, 하나라도 다를 경우 생성/제거합니다.
2. translate태스크에서 변경된 부분(marked study model field, created related model instance)에 대한 번역을 새로 생성합니다.
이 때 제거된 related model instance의 번역본의 경우, 이미 on_delete 규칙에 따라 convert 태스크에서 같이 제거되었기 때문에 신경쓰지 않아도 됩니다.
먼저, 간단하게 구현할 수 있는 study model의 필드 변경 감지 코드를 작성해보겠습니다.
번역필드만 확인하면 되기 때문에 TRANSLATE_FIELDS로 지정해두고 사용하도록 하겠습니다.
def mark_updated_sutdy_field(study, convert_data):
TRANSLATE_FIELDS = ['title', 'overall_status', 'phase']
translated_studies = Study.objects.filter(translate_from_study=study)
for field in TRANSLATE_FIELDS:
if getattr(study, field) != convert_data[field]:
for translated_study in translated_studies:
setattr(translated_study, field, "<updated>")
for translated_study in translated_studies:
translated_study.save()이제 related model instance에서 기존 instance를 재활용하도록 수정해줍니다.
def convert_conditions(condition_module):
conditions = []
condition_instances = list(Condition.objects.filter(name__in=condition_module))
for condition in condition_module:
condition_id = None
for condition_instance in condition_instances:
if condition_instance.name == condition:
condition_id = condition_instance.id
break
conditions.append({
'id': condition_id,
'name': condition,
'locale': 'en',
})
return conditions
def convert_interventions(study, intervention_module):
interventions = []
intervention_instances = list(Intervention.objects.filter(study=study))
for intervention in intervention_module:
intervention_id = None
for intervention_instance in intervention_instances:
if (
intervention_instance.name == intervention.get('InterventionName', None) and \
intervention_instance.intervention_type == intervention.get('InterventionType', None) and \
intervention_instance.description == intervention.get('InterventionDescription', None)
):
intervention_id = intervention_instance.pk
intervention_instances.remove(intervention_instance)
break
interventions.append({
'id': intervention_id,
'intervention_type': intervention.get('InterventionType', None),
'name': intervention.get('InterventionName', None),
'description': intervention.get('InterventionDescription', None),
'locale': 'en',
})
return interventions
def convert_eligibilities(study, eligibility_module):
if eligibility_module is None:
return []
eligibility_instances = list(Eligibility.objects.filter(study=study))
eligibility_id = None
for eligibility_instance in eligibility_instances:
if (
eligibility_instance.gender == eligibility_module.get('Gender', None) and \
eligibility_instance.minimum_age == eligibility_module.get('MinimumAge', None) and \
eligibility_instance.maximum_age == eligibility_module.get('MaximumAge', None) and \
eligibility_instance.healthy_volunteers == eligibility_module.get('HealthyVolunteers', None) and \
eligibility_instance.criteria == eligibility_module.get('EligibilityCriteria', None)
):
eligibility_id = eligibility_instance.pk
eligibility_instances.remove(eligibility_instance)
break
return [{
'id': eligibility_id,
'gender': eligibility_module.get('Gender', None),
'minimum_age': eligibility_module.get('MinimumAge', None),
'maximum_age': eligibility_module.get('MaximumAge', None),
'healthy_volunteers': eligibility_module.get('HealthyVolunteers', None),
'criteria': eligibility_module.get('EligibilityCriteria', None),
'locale': 'en',
}]convert 태스크를 실행하면 변경사항이 없는 부분은 재활용하고, 변경된 부분만 인식함을 확인할 수 있습니다.
이제 translate 태스크를 수정합니다.
- trnaslated_studies를 제거하는 코드 삭제
- related model instance중 translated instance가 없는 부분만 번역하도록 수정
- <updated>마크 되어있는 study model컬럼만 다시 번역하도록 수정
1번은 다들 금방 할 수 있기 때문에 2, 3번에 대해서만 설명하도록 하겠습니다.
- translated_study 인스턴스를 업데이트 하는 방식으로 진행하기 때문에, translate 태스크 진행 시 translated_study 인스턴스를 찾아 translate_study함수에 인자로 넣어줍니다.
- translate_study함수는 인자로 받은 translated_study(default: None)에서 <update>마크가 되어있는 부분만 새로 번역하고, 나머지는 기존 데이터를 그대로 사용합니다.
- intervention, condition 등의 연결된 모델들은 translated object가 있는지 확인하고, 없으면 생성해 사용합니다.
코드를 보면 다음과 같습니다.
def translate_study(study, translated_study=None):
"""
임상 연구 데이터를 번역하는 메소드
"""
translated_text_dict = {translate_field:None for translate_field in TRANSLATE_FIELDS}
if translated_study is not None:
for translate_field in TRANSLATE_FIELDS:
if getattr(translated_study, translate_field) != "<updated>":
translated_text_dict[translate_field] = getattr(translated_study, translate_field)
for translate_field in TRANSLATE_FIELDS:
if translated_text_dict[translate_field] is None and getattr(study, translate_field) is not None:
translated_text_dict[translate_field] = translate(getattr(study, translate_field))
conditions = []
for condition in study.conditions.all().prefetch_related(Prefetch("translated_conditions", queryset=Condition.objects.filter(locale='ko'))):
translated_condition = condition.translated_conditions.first()
if translated_condition is not None:
conditions.append({
'id': translated_condition.pk,
'name': translated_condition.name,
'locale': 'ko',
'translate_from_condition': condition.pk
})
else:
conditions.append({
'name': translate(condition.name),
'locale': 'ko',
'translate_from_condition': condition.pk
})
interventions = []
for intervention in Intervention.objects.filter(study=study).prefetch_related(Prefetch("translated_interventions", queryset=Intervention.objects.filter(locale='ko'))):
translated_intervention = intervention.translated_interventions.first()
if translated_intervention is not None:
interventions.append({
'id': translated_intervention.pk,
'intervention_type': translated_intervention.intervention_type,
'name': translated_intervention.name,
'description': translated_intervention.description,
'locale': 'ko',
'translate_from_intervention': intervention.pk
})
else:
interventions.append({
'intervention_type': translate(intervention.intervention_type),
'name': translate(intervention.name),
'description': translate(intervention.description),
'locale': 'ko',
'translate_from_intervention': intervention.pk
})
eligibilities = []
for eligibility in Eligibility.objects.filter(study=study).prefetch_related(Prefetch("translated_eligibilities", queryset=Eligibility.objects.filter(locale='ko'))):
translated_eligibility = eligibility.translated_eligibilities.first()
if translated_eligibility is not None:
eligibilities.append({
'id': translated_eligibility.pk,
'gender': translated_eligibility.gender,
'minimum_age': translated_eligibility.minimum_age,
'maximum_age': translated_eligibility.maximum_age,
'healthy_volunteers': translated_eligibility.healthy_volunteers,
'criteria': translated_eligibility.criteria,
'locale': 'ko',
'translate_from_eligibility': eligibility.pk
})
else:
eligibilities.append({
'gender': translate(eligibility.gender),
'minimum_age': eligibility.minimum_age,
'maximum_age': eligibility.maximum_age,
'healthy_volunteers': translate(eligibility.healthy_volunteers),
'criteria': translate(eligibility.criteria),
'locale': 'ko',
'translate_from_eligibility': eligibility.pk
})
return {
'nct_id': study.nct_id,
'title': translated_text_dict['title'],
'results_first_submitted_date': study.results_first_submitted_date,
'last_update_submitted_date': study.last_update_submitted_date,
'start_date': study.start_date,
'completion_date': study.completion_date,
'overall_status': translated_text_dict['overall_status'],
'phase': translated_text_dict['phase'],
'enrollment': study.enrollment,
'interventions': interventions,
'conditions': conditions,
'eligibilities': eligibilities,
'locale': 'ko',
'translate_from_study': study.pk,
'control_status_type': ControlStatusType.COMPLETED,
}이렇게 임상연구 적재, 번역, 간단한 최적화까지 빠르게 진행해보았습니다.
여기서 조금만 더 응용한다면 다국어 번역, 여러 데이터 소스에서 임상연구 데이터 적재, 더 좋은 최적화 등 개선할 부분이 많으니 시도해보셔도 좋을듯 합니다.
Django 데이터 적재 시리즈는 여기서 마무리 짓고, 이후 하루만에 완성하는 Django+DRF 서비스시리즈로 찾아오도록 하겠습니다.
