샘플 코드 (Python)

Python 학습 및 게시물 관리 예제

이 Python 스크립트는 다양한 문서(파일, 게시물 등)를 학습시키고 상태를 확인하는 함수들을 제공합니다. MongoDB와 연계하여 데이터를 관리하거나, 학습 실패한 게시물을 삭제하는 기능도 포함되어 있습니다.

필수 라이브러리

pip install requests pandas

주요 함수 설명
train_file_and_check_status
이 함수는 파일을 학습시키고 그 상태를 주기적으로 확인합니다.

def train_file_and_check_status(api_key, channel_id, file_path, doc_filter=None):
    # API URL 및 헤더 설정
    upload_url = "https://sejong-api.veluga.app/api/v1/document/file"
    check_url = "https://sejong-api.veluga.app/api/v1/document/file/{id}"

    headers = {
        "X-VELUGA-API-KEY": api_key,  # 발급받은 API 키
        # 파이썬에 requests 모듈을 사용하여 파일을 업로드할 때는 Content-Type 헤더를 따로 지정하지 않아야 합니다.
        # Post 메시지 함수에는 꼭 넣어야 합니다.
        #"Content-Type": "multipart/form-data"
    }

    data = {
        'channelId': channel_id,
        'forceUpdate': 'true',
        'docFilter': json.dumps(doc_filter)  # doc_filter를 JSON 문자열로 변환
    }

    files = {
        'file': open(file_path, 'rb')
    }

    response = requests.post(upload_url, headers=headers, data=data, files=files)

    if response.status_code == 200:
        response_data = response.json()
        file_id = response_data.get('data')
        print("파일 업로드 성공. 파일 ID:", file_id)
        success = response_data.get('success')

        if success == False:
            print("파일 업로드 실패. 응답 코드:", response.status_code)
            print("응답 내용:", response.text)
            return
        # 학습 상태 체크 (1초 간격)
        while True:
            check_response = requests.get(check_url.format(id=file_id), headers=headers)
            
            if check_response.status_code == 200:
                check_data = check_response.json().get('data')
                file_state = check_data.get('state')
                
                print(f"파일 상태: {file_state}")
                
                if file_state in ['success', 'fail']:
                    print(f"학습이 {file_state} 상태로 완료되었습니다.")
                    break
            else:
                print("학습 상태 체크 실패. 응답 코드:", check_response.status_code)
                break
            
            time.sleep(2)  # 2초 대기
    else:
        print("파일 업로드 실패. 응답 코드:", response.status_code)
        print("응답 내용:", response.text)

train_webUrl_and_check_status

웹 Url 학습 및 상태체크

def train_webUrl_and_check_status(api_key, channel_id, web_url, doc_filter=None):
    # API URL 및 헤더 설정
    upload_url = "https://sejong-api.veluga.app/api/v1/document/web-url"
    check_url = "https://sejong-api.veluga.app/api/v1/document/web-url/{id}"

    headers = {
        "X-VELUGA-API-KEY": api_key,  # 발급받은 API 키
        # 파이썬에 requests 모듈을 사용하여 파일을 업로드할 때는 Content-Type 헤더를 따로 지정하지 않아야 합니다.
        # Post 메시지 함수에는 꼭 넣어야 합니다.
        #"Content-Type": "application/json"
    }

    data = {
        'channelId': channel_id,
        'webUrl':  web_url,
        'forceUpdate': 'true',
        'docFilter': json.dumps(doc_filter)  # doc_filter를 JSON 문자열로 변환
    }

    response = requests.post(upload_url, headers=headers, data=data)

    if response.status_code == 200:
        response_data = response.json()
        web_id = response_data.get('data')
        print("웹 업로드 성공. 파일 ID:", web_id)
        success = response_data.get('success')
        if success == False:
            print("파일 업로드 실패. 응답 코드:", response.status_code)
            print("응답 내용:", response.text)
            return
        # 학습 상태 체크 (1초 간격)
        while True:
            check_response = requests.get(check_url.format(id=web_id), headers=headers)
            
            if check_response.status_code == 200:
                check_data = check_response.json().get('data')
                file_state = check_data.get('state')
                
                print(f"웹 url 상태: {file_state}")
                
                if file_state in ['success', 'fail']:
                    print(f"학습이 {file_state} 상태로 완료되었습니다.")
                    break
            else:
                print("학습 상태 체크 실패. 응답 코드:", check_response.status_code)
                break
            
            time.sleep(2)  # 2초 대기
    else:
        print("파일 업로드 실패. 응답 코드:", response.status_code)
        print("응답 내용:", response.text)

train_post_and_check_status
이 함수는 게시물(공지사항 등)을 학습시키고 그 상태를 주기적으로 확인합니다.

def train_post_and_check_status(api_key, channel_id, title, content, board=None, from_url=None, reg_datetime=None, metadata=None, doc_filter=None):
    # API URL 및 헤더 설정
    post_url = "https://sejong-api.veluga.app/api/v1/document/post"
    check_url = "https://sejong-api.veluga.app/api/v1/document/post/{id}"

    headers = {
        "X-VELUGA-API-KEY": api_key,
        "Content-Type": "application/json"
    }
    
    # 요청 바디 설정
    data = {
        "channelId": channel_id,
        "title": title,
        "content": content,
        "board": board,
        "fromUrl": from_url,
        "regDatetime": reg_datetime,
        "metadata": metadata,
        "docFilter": doc_filter
    }
    
    # 게시물 학습 요청
    response = requests.post(post_url, headers=headers, json=data)
    
    if response.status_code == 200:
        response_data = response.json()
        post_id = response_data.get('data')
        print("게시물 학습 요청 성공. 게시물 ID:", post_id)
    
        # 학습 상태 체크 (1초 간격)
        while True:
            check_response = requests.get(check_url.format(id=post_id), headers=headers)
            
            if check_response.status_code == 200:
                check_data = check_response.json().get('data')
                post_state = check_data.get('state')
                
                print(f"게시물 상태: {post_state}")
                
                if post_state in ['success', 'fail']:
                    print(f"학습이 {post_state} 상태로 완료되었습니다.")
                    break
            else:
                print("학습 상태 체크 실패. 응답 코드:", check_response.status_code)
                break
            
            time.sleep(2)  # 2초 대기
    else:
        print("게시물 학습 요청 실패. 응답 코드:", response.status_code)
        print("응답 내용:", response.text)

train_csv_and_check_status
이 함수는 CSV 파일을 학습시키고, 학습 완료 후 상태를 확인하여 결과를 반환합니다.

def train_csv_and_check_status(api_key, channel_id, file_path):
    # CSV 파일 학습 및 상태 확인
    upload_url = "[https://sejong-api.veluga.app/api/v1/document/file](https://sejong-api.veluga.app/api/v1/document/file)"
    check_url = "[https://sejong-api.veluga.app/api/v1/document/file/{id}](https://sejong-api.veluga.app/api/v1/document/file/%7Bid%7D)"

    headers = {
        "X-VELUGA-API-KEY": api_key
    }
    
    if not file_path.endswith('.csv'):
        print("Error: 이 함수는 CSV 파일만 업로드할 수 있습니다.")
        return
    
    files = {
        "file": open(file_path, "rb")
    }
    data = {
        "channelId": channel_id
    }
    
    response = requests.post(upload_url, headers=headers, files=files, data=data)
    
    if response.status_code == 200:
        response_data = response.json()
        file_id = response_data.get('data')
        print("CSV 파일 업로드 성공. 파일 ID:", file_id)
    
        # 학습 상태 체크 (1초 간격)
        while True:
            check_response = requests.get(check_url.format(id=file_id), headers=headers)
            
            if check_response.status_code == 200:
                check_data = check_response.json().get('data')
                file_state = check_data.get('state')
                
                print(f"CSV 파일 상태: {file_state}")
                
                if file_state in ['success', 'fail']:
                    print(f"학습이 {file_state} 상태로 완료되었습니다.")
                    return file_state, file_id
            else:
                print("학습 상태 체크 실패. 응답 코드:", check_response.status_code)
                break
            
            time.sleep(2)  # 2초 대기
    else:
        print("CSV 파일 업로드 실패. 응답 코드:", response.status_code)
        print("응답 내용:", response.text)
        return None, None

train_citation_file
이 함수는 참조 파일을 학습시키는 기능을 담당합니다.

def train_citation_file(api_key, channel_id, citation_file_path, training_file_id):
    # 참조 파일 학습 요청
    citation_url = "https://sejong-api.veluga.app/api/v1/document/file/citation"

    headers = {
        "X-VELUGA-API-KEY": api_key,
        #"Content-Type": "multipart/form-data"
    }

    files = {
        "file": open(citation_file_path, "rb")
    }
    data = {
        "channelId": channel_id,
        "trainingFileId": training_file_id
    }

    response = requests.post(citation_url, headers=headers, files=files, data=data)

    if response.status_code == 200:
        print("참조 파일 학습 요청 성공.")
    else:
        print("참조 파일 학습 요청 실패. 응답 코드:", response.status_code)
        print("응답 내용:", response.text)

delete_failed_post(api_key, post_id):
# 게시물 삭제 예시

def delete_failed_post(api_key, post_id):
    # 게시물 삭제 API URL 및 헤더 설정
    delete_url = f"https://sejong-api.veluga.app/api/v1/document/{post_id}"
    headers = {
        "X-VELUGA-API-KEY": api_key
    }

    # 게시물 삭제 요청
    response = requests.delete(delete_url, headers=headers)

    if response.status_code == 200:
        print(f"게시물 삭제 성공. 게시물 ID: {post_id}")
    else:
        print(f"게시물 삭제 실패. 게시물 ID: {post_id}, 응답 코드:", response.status_code)
        print("응답 내용:", response.text)

delete_all_failed_posts
이 함수는 학습이 실패한 모든 게시물을 삭제하는 기능을 제공합니다.

def delete_all_failed_posts(api_key, channel_id, board_name=None, keyword=None):
    # 학습이 실패한 게시물 리스트 가져오기
    failed_posts = get_failed_posts(api_key, channel_id, board_name, keyword)
      if failed_posts:
    print(f"총 {len(failed_posts)}개의 학습 실패 게시물이 발견되었습니다.")
    for post in failed_posts:
        delete_failed_post(api_key, post['_id'])
else:
    print("학습 실패한 게시물이 없습니다.")

if __name__ == "__main__":
    
    api_key = "YOUR_API_KEY"
    # 학습시킬 채널ID 미리 만들어져 있어야 합니다.
    channel_id = "YOUR_CHANNEL_ID"
    
    # 수업계획서 학습예제
    file_path = "12345.pdf"
    # 채팅창에 '일반'과 '수업계획서' 라는 선택 버튼이 생성됩니다. 
    # 수업계획서를 선택하고 질문을 던지면 수업계획서 docfilter로 학습한 문서들을 참고하여 답변합니다.
    doc_filter = {
        "docType": "수업계획서",
    }
    train_file_and_check_status(api_key, channel_id, file_path, doc_filter)
    
    # 일반파일 학습예제
    train_file_and_check_status(api_key, channel_id, file_path)
    
    # 공지사항 게시물 학습 예제
    title = "공지사항 제목"
    content = "<p>공지사항 내용입니다.</p>"
    board = "공지사항 게시판"
    from_url = "https://example.com/notice"
    reg_datetime = "2024-08-13T12:00:00Z"
    metadata = {
        "author": "관리자",
        "attachment": "파일명.pdf"
    }
    train_post_and_check_status(api_key, channel_id, title, content, board, from_url, reg_datetime, metadata)
    
    # CSV 파일 학습 및 참조 파일 학습 예제
    csv_file_path = "testcsv.csv"
    citation_file_path = "testpdf.pdf"
    
    state, file_id = train_csv_and_check_status(api_key, channel_id, csv_file_path)
    
    if state == "success":
        train_citation_file(api_key, channel_id, citation_file_path, file_id)
    
    # 전체 또는 실패한 게시물 삭제 예제
    delete_all_failed_posts(api_key, channel_id)

다운로드 샘플

sejong_sample.py15KB

Made with SlashPage