안녕하세요. 휴먼스케이프 june입니다.
Django를 통해 크롤링 데이터 적재하는 방법에 대해 알아보도록 하겠습니다.
프로젝트를 진행한 코드는 깃허브에 올라가 있으니 참고하시기 바랍니다.
파이썬 설치
자연어 처리를 하기 위해선 파이썬을 설치해야 합니다.
아래 사이트에 접속해 파이썬을 설치해줍니다.
Django 프로젝트 생성
django project를 생성하기 위해 python에 django를 설치해줍니다.
$ python -m pip install Djangodjango project를 생성해줍니다.
$ django-admin startproject django_practice패키지 버전 관리(Poetry)
파이썬 패키지 버전 관리는 poetry를 사용해서 진행합니다.
아래 문서를 보고 poetry install을 진행합니다.
poetry install이 완료되었으면 poetry init 명령어를 실행해 pyproject.toml 파일을 생해줍니다.
새로 생성된 pyproject.toml에는 django pkg가 포함되어있지 않으니 django를 추가합니다.
$ poetry add djangopoetry의 python virtualenv에 들어가려면 다음 명령어를 입력하면 됩니다. poetry shell
데이터를 가져올 사이트 탐색
우리는 https://clinicaltrials.gov/ 라는 곳에서 임상연구 데이터를 가져와 적재할 예정입니다.
아래 가이드를 보면 api 사용 방법이 상세하게 나와있습니다.
(사실 압축 파일을 통쨰로 다운받아 적재하는 방법이 있지만, api호출을 해야하는 경우를 설명하기 위해 건별로 api를 호출하는 방식을 사용하겠습니다.)
clinicaltrials.gov에서 서빙하는 임상연구의 데이터 구조는 다음과 같습니다.
모든 데이터를 저장하게 되면 고통스러울 수 있으니 몇개만 추출해 저장해보도록 하겠습니다.
데이터 모델링
먼저 임상연구(study)를 저장할 앱을 생성합니다.
python manage.py startapp studiesmain app(django_practice)의 settings.py에 들어가 방금 생성한 앱(studies)를 INSTALLED_APPS에 추가합니다.
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
"studies",
]models.py에 들어가 우리가 사용할 모델 몇개만 추려 입력합니다.
from django.db import models
class Study(models.Model):
nct_id = models.CharField(verbose_name="임상연구 번호", max_length=50)
results_first_submitted_date = models.DateField(verbose_name="최초 제출 날짜")
last_update_submitted_date = models.DateField(verbose_name="최근 수정 날짜")
start_date = models.DateField(verbose_name="임상연구 시작 날짜")
completion_date = models.DateField(verbose_name="임상연구 종료 날짜")
title = models.CharField(max_length=300, verbose_name="제목")
overall_status = models.CharField(max_length=50, verbose_name="진행 상태")
phase = models.CharField(max_length=50, verbose_name="임상 단계")
enrollment = models.IntegerField(verbose_name="대상자 수")
class Intervention(models.Model):
study = models.ForeignKey(Study, on_delete=models.CASCADE, related_name="interventions")
intervention_type = models.CharField(max_length=50, verbose_name="치료 타입")
name = models.CharField(max_length=100)
description = models.CharField(max_length=300)
class Condition(models.Model):
studies = models.ManyToManyField(Study, related_name="conditions")
name = models.CharField(max_length=100)
class Eligibility(models.Model):
study = models.ForeignKey(Study, on_delete=models.CASCADE, related_name="eligibilities")
gender = models.CharField(max_length=10, verbose_name="성별")
minimum_age = models.CharField(max_length=30, verbose_name="최소 나이")
maximum_age = models.CharField(max_length=30, verbose_name="최대 나이")
healthy_volunteers = models.CharField(max_length=100)
criteria = models.TextField()$ python3 manage.py makemigrations 명령어를 실행하면 우리가 작성한 models에 맞춰 migration file이 자동으로 생성됩니다.
migration이란
마이그레이션은 당신의 모델에 생긴 변화(필드를 추가했다던가 모델을 삭제했다던가 등등)를 반영하는 Django의 방식입니다. 대체로 자동실행되도록 설계되었지만 언제 마이그레이션을 만들고 언제 실행하고 보통 어떤 문제에 맞닥트리는지는 알아둘 필요가 있습니다.
데이터 적재 코드 작성
studies app 내 batch_tasks.py에서 데이터 적재 코드를 관리합니다.
먼저 clinicaltrials.gov에서 어떤 방식으로 임상연구를 가져올지 구상합니다.
등록된 임상연구의 수가 매우 많기(약 40만개 이상)때문에 간단하게 전체 임상연구의 수를 구한 뒤, min_rnk, max_rnk 파라미터를 활용해 조금씩 나눠서 가져오기로 합니다.
위의 방식을 구현하기 위해 전체 임상연구를 가져오는 함수 get_all_studies 를 이루는 get_studies_count, get_studies 함수를 구현합니다.
이때 request를 사용하기 위한 패키지 requests를 설치합니다.
$ poetry add requests
import requests
def get_studies_num():
"""
clinicaltrials.gov 에 존재하는 임상 연구 개수를 가져오는 메소드
"""
CLINICALTRIALS_STATISTICS_URL = 'https://clinicaltrials.gov/api/info/study_statistics'
response = requests.get(CLINICALTRIALS_STATISTICS_URL, params={'fmt': 'json'})
studies_num = int(response.json()['StudyStatistics']['ElmtDefs']['Study']['nInstances'])
return studies_num
def get_studies(start, end):
"""
clinicaltrials.gov 에서 제공하는 API 에서 임상 연구 목록을 가져오는 메소드
"""
CLINICALTRIALS_OPEN_API_BASE_URL = 'https://clinicaltrials.gov/api/query/full_studies'
response = requests.get(CLINICALTRIALS_OPEN_API_BASE_URL, params={'fmt': 'json', 'min_rnk': start, 'max_rnk': end})
response_json = response.json()
if response_json['FullStudiesResponse']['NStudiesFound'] == 0:
return []
return response_json['FullStudiesResponse']['FullStudies']
def get_all_studies():
"""
clinicaltrials.gov 에서 제공하는 API 에서 전체 임상 연구 목록을 가져오는 메소드
"""
studies_num = get_studies_num()
studies = []
for start in range(1, studies_num, 100):
end = start + 99
studies += get_studies(start, end)
return studies임상연구를 가져오는 함수를 간단하게 구현했습니다.
이제 get_all_studies를 수정해 모든 임상연구를 db에 저장하도록 할 것입니다.
지금 가져오는 studies는 clinicaltrials.gov의 response를 그대로 가져오고 있어 우리가 저장할 수 있는 형식이 아닙니다.
위에서 우리가 생성한 model 구조에 맞게 response를 변경해주는 로직을 작성합니다.
def convert_study(study):
return {
'nct_id': study['Study']['ProtocolSection']['IdentificationModule']['NCTId'],
'title': study['Study']['ProtocolSection']['DescriptionModule']['OfficialTitle'],
'results_first_submitted_date': study['Study']['ProtocolSection']['DescriptionModule']['ResultsFirstSubmittedDate'],
'last_update_submitted_date': study['Study']['ProtocolSection']['DescriptionModule']['LastUpdateSubmittedDate'],
'start_date': study['Study']['ProtocolSection']['DescriptionModule']['StartDate'],
'completion_date': study['Study']['ProtocolSection']['DescriptionModule']['CompletionDate'],
'overall_status': study['Study']['ProtocolSection']['DescriptionModule']['OverallStatus'],
'phase': study['Study']['ProtocolSection']['DescriptionModule']['Phase'],
'enrollment': study['Study']['ProtocolSection']['DescriptionModule']['Enrollment'],
'intervensions': [
{
'intervention_type': intervention['InterventionType'],
'name': intervention['InterventionName'],
'description': intervention['InterventionDescription'],
}
for intervention in study['Study']['ProtocolSection']['InterventionModule']['InterventionList']['Intervention']
],
'conditions': [
condition['Condition']
for condition in study['Study']['ProtocolSection']['ConditionModule']['ConditionList']['Condition']
],
'eligibilities': [
{
'gender': study['Study']['ProtocolSection']['EligibilityModule']['Gender'],
'minimum_age': study['Study']['ProtocolSection']['EligibilityModule']['MinimumAge'],
'maximum_age': study['Study']['ProtocolSection']['EligibilityModule']['MaximumAge'],
'healthy_volunteers': study['Study']['ProtocolSection']['EligibilityModule']['HealthyVolunteers'],
'criteria': study['Study']['ProtocolSection']['EligibilityModule']['EligibilityCriteria'],
}
]
}일단 model 구조에 맞게 변환까지는 했는데.. 저장하려니 조금 막막합니다.
drf serializer의 힘을 빌려 간편하게 저장하도록 합시다.
(drf serializer는 주로 rest api에서 model에 저장된 데이터를 json으로 변환하고, 반대로 json으로 들어온 데이터를 model 데이터로 변환해 저장하는 기능을 합니다.)
먼저 drf를 설치하고, settings의 INSTALLED_APPS에 추가해줍니다.
poetry add djangorestframework
기본 serializer는 연결된 모델 데이터까지 저장하지 못하니 drf-writable-nested 를 설치합니다.
serializers.py 파일을 생성해 Study를 포함한 model들의 serializer를 정의합니다.
from rest_framework.serializers import ModelSerializer
from drf_writable_nested.serializers import WritableNestedModelSerializer
from .models import Study, Intervention, Condition, Eligibility
class InterventionSerializer(ModelSerializer):
class Meta:
model = Intervention
fields = ['intervention_type', 'name', 'description']
class ConditionSerializer(ModelSerializer):
class Meta:
model = Condition
fields = ['name']
class EligibilitySerializer(ModelSerializer):
class Meta:
model = Eligibility
fields = ['gender', 'minimum_age', 'maximum_age', 'healthy_volunteers', 'criteria']
class StudySerializer(WritableNestedModelSerializer):
interventions = InterventionSerializer(many=True)
conditions = ConditionSerializer(many=True)
eligibilities = EligibilitySerializer(many=True)
class Meta:
model = Study
fields = '__all__'
다시 batch_tasks.py로 돌아와 get_all_studies를 save_all_studies로 변경하고, 코드를 수정합니다.
def save_all_studies():
"""
clinicaltrials.gov 에서 제공하는 API 에서 전체 임상 연구 목록을 저장하는 메소드
"""
studies_num = get_studies_num()
for start in range(1, studies_num, 100):
end = start + 99
studies = get_studies(start, end)
for study in studies:
study = convert_study(study)
study_serializer = StudySerializer(data=study)
try:
study_serializer.is_valid(raise_exception=True)
except ValidationError as e:
if e.detail.get('nct_id', None) is not None and e.detail['nct_id'][0].code == 'unique':
continue
raise e
study_serializer.save()이제 python3 manage.py shell로 접속해 해당 함수를 실행하면 임상연구가 저장되는것을 확인할 수 있습니다.
실제로 실행해보니 데이터가 비어있는 경우가 많아 KeyError가 많이 발생합니다.
적당히 없는 데이터는 None으로 들어가도록 model과 convert 함수를 수정해줍니다.
class Study(models.Model):
nct_id = models.CharField(verbose_name="임상연구 번호", max_length=50, unique=True)
results_first_submitted_date = models.DateField(verbose_name="최초 제출 날짜", null=True, blank=True)
last_update_submitted_date = models.DateField(verbose_name="최근 수정 날짜", null=True, blank=True)
start_date = models.DateField(verbose_name="임상연구 시작 날짜", null=True, blank=True)
completion_date = models.DateField(verbose_name="임상연구 종료 날짜", null=True, blank=True)
title = models.TextField(verbose_name="제목", null=True, blank=True)
overall_status = models.CharField(max_length=50, verbose_name="진행 상태", null=True, blank=True)
phase = models.CharField(max_length=50, verbose_name="임상 단계", null=True, blank=True)
enrollment = models.IntegerField(verbose_name="대상자 수", null=True, blank=True)
class Intervention(models.Model):
study = models.ForeignKey(Study, on_delete=models.CASCADE, related_name="interventions")
intervention_type = models.CharField(max_length=50, verbose_name="치료 타입", null=True, blank=True)
name = models.CharField(max_length=100, null=True, blank=True)
description = models.TextField(null=True, blank=True)
class Condition(models.Model):
studies = models.ManyToManyField(Study, related_name="conditions")
name = models.CharField(max_length=100, null=True, blank=True)
class Eligibility(models.Model):
study = models.ForeignKey(Study, on_delete=models.CASCADE, related_name="eligibilities")
gender = models.CharField(max_length=10, verbose_name="성별", null=True, blank=True)
minimum_age = models.CharField(max_length=30, verbose_name="최소 나이", null=True, blank=True)
maximum_age = models.CharField(max_length=30, verbose_name="최대 나이", null=True, blank=True)
healthy_volunteers = models.CharField(max_length=100, null=True, blank=True)
criteria = models.TextField(null=True, blank=True)
def convert_study(study):
description_module = study['Study']['ProtocolSection'].get('DescriptionModule', {})
if 'OfficialTitle' in description_module:
title = description_module['OfficialTitle']
elif 'BriefTitle' in description_module:
title = description_module['BriefTitle']
elif 'BriefSummary' in description_module:
title = description_module['BriefSummary']
else:
title = None
interventions = study['Study']['ProtocolSection'].get('ArmsInterventionsModule', {}).get('InterventionList', {}).get('Intervention', [])
conditions = study['Study']['ProtocolSection'].get('ConditionsModule', {}).get('ConditionList', {}).get('Condition', [])
eligibility_module = study['Study']['ProtocolSection'].get('EligibilityModule', None)
if eligibility_module is None:
eligibility = []
else:
eligibility = [
{
'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),
}
]
return {
'nct_id': study['Study']['ProtocolSection']['IdentificationModule']['NCTId'],
'title': title,
'results_first_submitted_date': description_module.get('ResultsFirstSubmittedDate', None),
'last_update_submitted_date': description_module.get('LastUpdateSubmittedDate', None),
'start_date': description_module.get('StartDate', None),
'completion_date': description_module.get('CompletionDate', None),
'overall_status': description_module.get('OverallStatus', None),
'phase': description_module.get('Phase', None),
'enrollment': description_module.get('Enrollment', None),
'interventions': [
{
'intervention_type': intervention.get('InterventionType', None),
'name': intervention.get('InterventionName', None),
'description': intervention.get('InterventionDescription', None),
}
for intervention in interventions
],
'conditions': [
{
'name': condition,
}
for condition in conditions
],
'eligibilities': eligibility,
}이렇게 1장에서 간단한 임상연구 데이터 적재를 실습해 보았습니다.
2장부터는 django crontab을 이용해 주기적인 적재, 적재 도중 api서버가 오류를 뱉을때의 핸들링 방법에 대해 알아보도록 하겠습니다.
