국회의원 회의록 정규식으로 구분하기, docker를 이용해 elastic search로 설치, 저장하기, 오류 해결하기

 

https://github.com/woongCat/goverment_check_project

GitHub - woongCat/goverment_check_project: 국회의원의 국회의원 회의 참석율과 공약이행률 확인하는 airflow

이제는 api를 통해서 pdf링크를 전부 가져왔다.

그리고 pdfplumber로 pdf에 텍스트를 추출하고

pdf 그 자체는 저장하지 않고 io.BytesIO를 이용해 메모리로만 읽어오기로 했다.

 

굳이 pdf 그 자체를 저장하지 않는 이유는 텍스트를 전부 추출하고 링크를 갖고 있기 때문에 pdf가 차지하는 용량은 안 가져오기로 했다.

 

그래서 아래와 같은 pdf 파일을 가져왔다. 이제 이 추출된 pdf를 어떻게 저장할지 고민해야할 차례다.

회의록의 기본적인 모양

[STEP 1 : 발언자와 발언으로 구분 짓기]

    speaker_pattern = re.compile(r"◯([\w]+ [\w]+)\s*\n*([\s\S]+?)(?=\n◯|\Z)", re.MULTILINE)

1. 발언자 이름 앞에는 ◯으로 시작한다. = ◯

2. 발언자는 직책과 이름 2단어로 구성되어 있고 그 사이에 공백이 있다. = ([\w]+ [\w]+)

3. 발언자와 발언 사이에는 공백이 여러개가 있다. = \s*\n*

4. 그 다음에 발언이 온다. +?를 사용한 이유는 ◯을 만나면 멈추기 위함이다. = ([\s\S]+?)

5. 다음 발언자 또는 문서 끝까지 매칭한다. = (?=\n◯|\Z)

 

[STEP 2 : 발언자, 발언 내용을 이용해 엘라스틱 서치에 들어갈 내용을 추가한다.]

def get_speaker_data(id,title,date,text):
    """ 텍스트에서 발언자와 발언 내용을 추출하는 함수 """
    speaker_pattern = re.compile(r"◯([\w]+ [\w]+)\s*\n*([\s\S]+?)(?=\n◯|\Z)", re.MULTILINE)
    speech_list = []
    
    for match in speaker_pattern.finditer(text):
        speaker = match.group(1).strip()
        speech = match.group(2).strip()

        speech_list.append({
            "document_id": id,
            "title": title,
            "date": date,
            "speaker": speaker, # 발언자
            "text": speech, # 발언
            "summary": None,  # 요약은 나중에 추가
            'timestamp': datetime.now(), # 편집날짜
        })

    return speech_list

 

근데, elastic search는 어떻게 사용하나요?

아하, 저도 잘 몰라서 요것저것 알아봤습니다!

생각보다 딱 정리된 블로그가 없어서 애 좀 먹었습니다.

 

먼저 local로 설치하는 방법과 docker로 사용하는 방법이 있는데 손쉽게 하기 위해서 docker를 사용하기로 했습니다!

[STEP 3 : Docker를 이용해 elasticSearch를 가져온다.]

 

docker run -d --name elasticsearch -p 9200:9200 -e "discovery.type=single-node" docker.elastic.co/elasticsearch/elasticsearch:8.6.0

이렇게 설정해주면 된다.

그리고 .env에 넣어주면 된다.

근데, password라고 환경변수를 설정하지 말자. 자꾸 오류가 난다. 

나는 결국, ELASTIC_SEARCH_PW라고 변수명을 바꾸고 나서야 됐다.

만약, 이걸 통해 비번을 바꿔주지 않는다면 로그를 확인하면 http traffic on an https channel을 볼 수 있고

elastic_transport.ConnectionError: 가 나오는 것을 확인할 수 있다. 

docker logs elasticsearch

elastic_transport.ConnectionError: Connection error caused by: ConnectionError(Connection error caused by: ProtocolError(('Connection aborted.', RemoteDisconnected('Remote end closed connection without response'))))

[STEP 4 : elasticSearch에 pdf data를 추가한다.]

def connect_to_elasticsearch():
    """Elasticsearch에 연결하는 함수"""
    load_dotenv()

    Elasticsearch_key = os.getenv("ELASTIC_SEARCH_PW")
    # Elasticsearch 연결
    es = Elasticsearch(
        "https://localhost:9200",
        basic_auth=("elastic", Elasticsearch_key),  # 인증 추가
        verify_certs=False  # 자체 서명된 인증서 문제 방지
        )

    # 인덱스 설정 (인덱스가 없으면 생성)
    index_name = "congress_meetings"

    if not es.indices.exists(index=index_name):
        es.indices.create(index=index_name)
        
    return es,index_name

def save_to_elasticsearch(data):
    """ 추출된 국회의원 발언 데이터를 Elasticsearch에 저장하는 함수 """
    es, index_name = connect_to_elasticsearch()
    for entry in data:
        es.index(index=index_name, body=entry)
    print("Elasticsearch 저장 완료!")

 

[STEP 5: elasticSearch에 들어간 내용 확인하기.]

from elasticsearch import Elasticsearch
import os

# .env에서 환경변수 불러오기
from dotenv import load_dotenv
load_dotenv()

# 환경변수에서 비밀번호 가져오기
ELASTIC_PASSWORD = os.getenv("ELASTIC_SEARCH_PW")

# Elasticsearch 클라이언트 설정
es = Elasticsearch(
    "https://localhost:9200",
    basic_auth=("elastic", ELASTIC_PASSWORD),  # 인증 추가
    verify_certs=False  # 자체 서명된 인증서 문제 방지
)

# 인덱스명 설정 (저장된 데이터가 들어 있는 인덱스)
index_name = "congress_meetings"

# 모든 데이터 조회
def get_all_documents():
    query = {
        "query": {
            "match_all": {}
        }
    }
    
    res = es.search(index=index_name, body=query, size=10)  # 최대 10개 문서 조회
    return res['hits']['hits']

# 데이터 가져오기
documents = get_all_documents()
for doc in documents:
    print(f"문서 ID: {doc['_id']}, 내용: {doc['_source']}")

원하는 모습으로 나온 것을 확인할 수 있다~