K . N . H :: BLOG

Python으로 Elasticsearch를 연결하려고 하는데 원인을 알 수 없는 이유로 자꾸 연결이 실패한다면...

지금 공유 드린 내용을 테스트 해보세요 그러면 해결 될 수도 있습니다.

Python으로 Elasticsearch를 연결할 때 디버깅 해보니 ID&PW는 맞게 넣었습니다.

하지만 이상하게 연결이 되지 않았는데요 저의 python 소스는 아래와 같았습니다.

  문제

def connect(self):
        # connect to elastic_search
        try:
            es = Elasticsearch(self.host, basic_auth=(self.id, self.pw), verify_certs=False, ssl_show_warn=False, request_timeout=60)

            if not es.ping():
                raise ValueError("ES Connection failed")
            print(' - ElasticSearch 연결 완료')
            return es

        except AuthenticationException as e:
            print("인증 실패: 아이디/비밀번호 또는 API Key 확인 필요")
            raise e
        except ConnectionError as e:
            print(f"연결 실패: {e}")
            raise e
        except Exception as e:
            print(f"알 수 없는 오류: {e}")
            raise e

  해결책

앞으로는 아래와 같이 try문 밑에 elastic 버전을 확인하는 문구를 넣어서 버전을 확인하고 그에 맞게 python 패키지도 설치해줘야 합니다. 

try:
    es = Elasticsearch(self.host, basic_auth=(self.id, self.pw), verify_certs=False, ssl_show_warn=False, request_timeout=60)

    # Elastic Search 버전 확인
    info = es.info()
    print(info['version']['number'])

만약 elastic 버전이 8.16 버전이라면 python 패키지 명령어는 아래와 같아야 합니다.

> pip install elasticsearch==8.16.0

이와 같이 하니 저 역시 문제를 해결할 수 있었습니다^^

이 글이 도움이 되셨기를 바랍니다.

내용이 유용하셨다면 좋아요&댓글 부탁드립니다.
이 블로그를 이끌어갈 수 있는 강력한 힘입니다!

caul334@gmail.com

개발 업무를 하다보면 회사 보안 정책 또는 인터넷 제한 때문에 Python 패키지를 다운받지 못하거나 설치하지 못하는 경우가 있습니다.

이러한 경우 아래와 같이 해결할 수 있는 방법 공유드립니다.

[ 가장 중요!!! ]

먼저, 어떤 환경에서 명령어를 입력할 지 구분하고 명령어를 입력해야 합니다.

패키지를 자유롭게 설치할 수 PC가 속해 있는 네트워크 망을 A망이라고 하고 인터넷이 제한되는 네트워크망을 B망이라고 하겠습니다.

네크워크 A망 : 제한 없고 패키지 자유롭게 설치 가능

네크워크 B망 : 보안 정책 또는 인터넷 제한으로 인해 패키지 설치 불가

  1. 현재 설치된 Python 패키지 목록을 txt로 저장

명령어 실행 환경 : 네크워크 A 망

명령어 : pip freeze > requirements.txt

  2. requirements.txt에 기입된 파이썬 패키지 목록 다운로드

명령어 실행 환경 : 네크워크 A 망

명령어 : pip download -r requirements.txt -d packages/

  3. 설치할 패키지를 다른 PC로 이동 (대상 프로젝트 위치로 이동합니다)

대상(2개)

    - (폴더) packages

    - (파일) requirements.txt

  4. 패키지 오프라인 설치

명령어 실행 환경 : 네크워크 B 망

명령어 : pip install --no-index --find-links=packages/ -r requirements.txt

도움이 되셨기를 바랍니다.

내용이 유용하셨다면 좋아요&댓글 부탁드립니다.
이 블로그를 이끌어갈 수 있는 강력한 힘입니다!

caul334@gmail.com

회사에서 파이썬으로 메일 업무를 자동화하고 싶은데 MFA 2차 인증이나 OTP 때문에 자동화를 못하신다구요?

그렇다면 이 글이 도움 될 수 있습니다.

  해결 방법

해결 방법은 Python을 통해 메일 모듈(smtplib)을 활용하는게 아니라 win32com 모듈을 활용하면 됩니다

win32com 모듈은 Windows에 있는 애플리케이션을 조작할 수 있는 기능이 있어서 이 모듈을 통해 Outlook 앱을 직접 조작할 수 있습니다.

1. win32 모듈 설치

pip 명령어를 통해 win32 모듈을 설치합니다.

# ---------- 1. 필수 모듈 설치 ---------- #
pip install pywin32

2. Outlook 앱 연결 테스트 코드

[★중요★]

전제조건 : 반.드.시. 코드 편집기 프로그램과 Outlook, 이렇게 2개 프로그램은 모두 관리자 권한으로 실행해야 합니다!

저의 경우는 코드 편집기로 Visual Studio Code를 사용하였습니다.

# ---------- 2. Outlook 연결 테스트 ---------- #
import win32com.client

try:
    outlook = win32com.client.Dispatch('outlook.application')
    print("Outlook 연결 성공")
except Exception as e:
    print("Outlook 연결 실패:", e)

Outlook과 Visual Studio Code를 모두 관리자 권한으로 실행 후 연결 테스트 코드를 실행하보면 정상적으로 연결 됩니다.

참고로 Outlook은 구버전 신버전 모두 테스트 해봤는데 두 버전 모두 정상적으로 연결 되었습니다.

3. 메일 발송 테스트

"Outlook 연결 성공" 메시지를 확인했다면 이제는 아래 코드를 통해 메일 발송을 테스트 해봐야 합니다.

아래 코드에서 수신자 메일 주소를 넣어 테스트 합니다.

# ---------- 3. Outlook 간단한 메시지 전송 테스트 ---------- #
import win32com.client as win32

# Outlook 앱 실행 연결
outlook = win32.Dispatch('outlook.application')
mail = outlook.CreateItem(0)  # 0: MailItem

# 메일 작성
mail.To = '<수신자 메일 주소>'
mail.Subject = 'Python을 통해 Outlook 앱으로 보낸 메일'
mail.Body = '이 메일은 Python 코드가 로컬 Outlook 앱을 제어해 전송한 것입니다.'

# 첨부파일 추가 (선택)
# mail.Attachments.Add('C:\\path\\to\\file.txt')

# 메일 발송
mail.Send()

print("메일 발송 완료")

이 글이 도움이 되셨기를 바랍니다.

내용이 유용하셨다면 좋아요&댓글 부탁드립니다.
이 블로그를 이끌어갈 수 있는 강력한 힘입니다!

caul334@gmail.com

Python 프로그램을 개발을 할 때 개발 환경을 격리하기 위한 가장 간편한 방법은 어플리케이션 레벨 격리 입니다.

격리 방법은 파이썬 내장 모듈인 venv을 활용하면 됩니다.

  격리(가상) 환경  생성 명령어

# 1. 격리 환경(디렉토리) 생성
python3.<버전> -m venv <격리 환경 이름>

# 2. 격리 환경 활성화
source <격리 환경 이름>/bin/activate

# 3. python3 경로 출력
(<격리 환경 이름>) $ which python3

# 4. 격리 환경 비활성화
(<격리 환경 이름>) $ deactivate

# 5. python3 경로 출력
which python3

위 명령어를 실행해보면 which python3 명령어를 입력했을 때

격리 환경과 격리되지 않은 환경에서의 python3 경로가 다른 것을 확인할 수 있습니다.

위 명령어를 참고하셔서 Python 프로그램을 개발할 때 안전 하게 격리환경 안에서 개발을 진행하시면 됩니다.

이 글 이 도움이 되셨기를 바랍니다.

caul334@gmail.com

내용이 유용하셨다면 좋아요&댓글 부탁드립니다.
이 블로그를 이끌어갈 수 있는 강력한 힘입니다!

  Poetry는?

패키지 버전을 관리하지 않아도 손쉽게 Dependency 관리를 가능하게 해주는 도구

즉 poetry를 이용한 프로젝트는 패키지 버전에 종속되지 않고 어느 환경에서든 잘 돌아가게 만들어주는 도구이다.

pip를 이용하여 패키지를 설치하면 패키지를 설치할 때마다 requirements.txt를 작성해야 Dependency 문제가 해결됩니다.

하지만 poetry add <package_name> 명령어를 이용하면 따로 패키지 버전을 관리하지 않아도

편하게 Dependency 관리가 가능해집니다.

  Poetry 설치 방법

1. Ubuntu (linux)
sudo apt install python3-poetry

2. Powershell (Windows)
(Invoke-WebRequest -Uri https://install.python-poetry.org -UseBasicParsing).Content | py -

3. curl 이용
curl -sSL https://install.python-poetry.org | python3

  Poetry 기본 명령어

# 설치 및 버전 확인
poetry --version

# 새 프로젝트 생성
$ poetry new <project_name>

# 가상환경 실행
$ cd <project_name>
$ poetry shell

# 패키지 설치
$ poetry add <package_name>

해당 글이 도움이 되셨기를 바랍니다.

caul334@gmail.com

내용이 유용하셨다면 좋아요&댓글 부탁드립니다.
이 블로그를 이끌어갈 수 있는 강력한 힘입니다!

Ubuntu(Linux) 에서 파이썬 디펜던시와 가상환경 관리를 위해 poetry(포어트리)를 설치하려 했습니다.

하지만 아래와 같은 에러를 만났습니다.

  에러 메시지 내용

nick@ubuntu:~/python$ pip install --user poetry

error: externally-managed-environment

  에러 발생 원인

에러 발생 원인은 python 3.11 이상 버전부터 패키지 관리 정책이 변경되어
pip 명령어를 이용한 직접 설치 보다는 Ubuntu 시스템 패키지 관리자인 apt를 통해 패키지를 설치하도록 권장하기 때문입니다.

  문제 해결 방법

아래 3가지 해결 방법 중 원하는 것을 선택해 해결할 수 있습니다.

1. 시스템 패키지 관리자를 이용하여 설치

# 무조건 가상 환경 밖에 설치해야 하는 경우
sudo apt install python3-<package_name>

2. 프로젝트가 가상환경 안에 있다면 가상환경 안에 설치

# 격리 환경(디렉토리) 생성
python3.<버전> -m venv <가상 환경 이름>

# 격리 환경 활성화
source <가상 환경 이름>/bin/activate

# 패키지 설치
pip install <package name>

3. curl 유틸리티 이용한 설치

# 유틸리티를 이용한 설치
curl -sSL https://install.python-poetry.org | python3

위 방법을 통해 문제가 해결 되셨기를 바랍니다.

caul334@gmail.com

내용이 유용하셨다면 좋아요&댓글 부탁드립니다.
이 블로그를 이끌어갈 수 있는 강력한 힘입니다!

회사 업무를 수행하다보면 종종 통계를 위해 메일 제목 또는 메일 내에 있는 특정 문구가 얼마나 있는지 필요한 경우가 있습니다.

아래 Python 코드로 그러한 어려움을 해결할 수 있습니다.

필요하신 분은 자유롭게 사용하시면 되겠습니다^^

Python 코드

# mail_extractor.py
import re
import os
import extract_msg
# pip install extract-msg

def extract_subjects_from_msg(folder_path):
    # 메일에서 제목과 내용 return

    # 결과를 저장할 리스트
    objects = []

    for file_name in os.listdir(folder_path):
        tmp = []
        if file_name.endswith(".msg"):
            file_path = os.path.join(folder_path, file_name)
            try:
                # .msg 파일 열기
                with extract_msg.Message(file_path) as message:

                    # 메일 제목
                    subject = message.subject
                    # 메일 본문
                    body = message.body

                    tmp.append(subject)
                    tmp.append(body)
                    objects.append(tmp)

            except Exception as e:
                print(f"Error reading {file_name}: {e}")

    return objects



def find_content(pattern, body):
    # 일치하는 내용 메일 본문에서 찾기
    content = []

    # 일치 여부
    matches = re.findall(pattern, body)

    # 결과 저장
    for i, match in enumerate(matches, start=1):
        content.append(i)
        content.append(match)

    return content



if __name__ == '__main__':

    # 메일 들이 저장되어 있는 특정 폴더 경로
    folder_path = r"C:\Users\<경로>"

    # 제목, 내용 추출
    results = extract_subjects_from_msg(folder_path)

    # 결과 출력
    for subject, body in results:
        # 제목과 메일 본분만 출력
        print(f"Subject: {subject}, Body: {body}")

        # 찾을 특정패턴이 있는 경우 (아래 코드 활용)
        # 패턴 정의
        pattern = r"\[Rule\]\s*\n(.+)"
        content = find_content(pattern, body)
        print("Subject: {}, Content: {}".format(subject, content))

부족한 부분 또는 개선이 필요한 사항은 댓글 남겨주시기 바라면서

이 글이 도움이 되셨기를 바랍니다.

caul334@gmail.com

내용이 유용하셨다면 좋아요&댓글 부탁드립니다.
이 블로그를 이끌어갈 수 있는 강력한 힘입니다!

해당 글은 Python과  Elasticsearch를 연동함에 있어서 나중에 참고를 목적으로 쓴 글입니다.

  아래 class Elastic() 구현된 기능

1. Elasticsearch 접속 : connect()

2. Index 크기 검색 : count_index_size()

3. 전체 문서 검색 : search_all_docs()

4. 특정 id에 대한 문서 검색 : search_doc_by_risk_id()

# Elasticsearch 관련 모듈 import
from elasticsearch import Elasticsearch


class Elastic:
    # Elastic Search 클래스 선언
    def __init__(self, host, id, pw):
        self.host = host
        self.id = id
        self.pw = pw


    def connect(self):
        # connect to elasticsearch
        es = Elasticsearch(self.host, basic_auth=(self.id, self.pw), verify_certs=False, ssl_show_warn=False)
        if not es.ping():
            raise ValueError("ES Connection failed")
        print('ElasticSearch 연결 완료')
        return es


    def count_index_size(self, es, index_name):
        # count index size
        count = es.count(index=index_name)
        index_size = count.body['count']
        return index_size


    def search_all_docs(self, es, index_name):
        # index에 저장된 모든 docs 검색
        index_size = self.count_index_size(es, index_name)
        search_query = {
            # 모든 docs 검색 및 riskid 기준 내림차순 정렬
            "sort": [
                {
                    "riskid": {
                        "order": "desc"
                    }
                }
            ],
            "size": index_size,
            "query": {
                "match_all": {}
            }
        }

        search_result = es.search(index=index_name, body=search_query)

        return search_result["hits"]["hits"]


    def search_doc_by_risk_id(self, es, index_name, risk_id):
        # index에 저장된 특정id doc 검색
        search_query = {
            "query": {
                "match": {
                    "riskid": risk_id
                }
            }
        }
        search_result = es.search(index=index_name, body=search_query)

        return search_result["hits"]["hits"]


if __name__ == "__main__":

    # 1. Connect ElasticSearch
    elastic = Elastic("https://localhost:9200", "elastic", "elastic_pw")
    es = elastic.connect()

    index_name = "test_index"

    # test_index 내에 있는 모든 문서 검색
    all_docs_list = elastic.search_all_docs(es, index_name)

    # test_index 내에 risk_id 가 12345인 문서 검색
    risk_id = 12345
    doc = elastic.search_doc_by_risk_id(es, index_name, risk_id)

caul334@gmail.com

내용이 유용하셨다면 좋아요&댓글 부탁드립니다.
이 블로그를 이끌어갈 수 있는 강력한 힘입니다!

회사 인터넷망 위에서 Python 프로그램을 개발 중인데

Selenium 또는 Selenium-Wire 모듈을 사용하면 반드시 googlechromelabs.github.io  사이트에 접속되어야만 합니다.

하지만 제가 다니는 회사는 프록시 서버를 지나지 않는 이상 Github 사이트에 접속이 불가능하다는 사실...

방화벽 신청까지 해봤지만 결과는 반려 처리ㅜㅜ 어쩔 수 없이 프록시로...

제 코드는 3가지 모듈 (selenium-wire, selenium, request)을 사용합니다.

이 경우 Proxy 설정 방법에 관해 공유드리겠습니다.

import certifi
import json
import requests
from seleniumwire import webdriver
from selenium.webdriver.chrome.service import Service


# Selenium 또는 Selnium-Wire를 사용하는 경우
def driver_load():
    # webdriver 로드 #
    chromedriver = ".\\chromedriver.exe"
    service = Service(executable_path=chromedriver)

    seleniumwire_options = {
        "proxy": {
            "http": "<Proxy Server>:<Proxy Port>",
            "https": "<Proxy Server>:<Proxy Port>"
        }
    }

    # Set Options
    options = webdriver.ChromeOptions()
    options.add_argument('--proxy-server=<Proxy Server>:<Proxy Port>')

    # Set up the Chrome WebDriver
    driver = webdriver.Chrome(service=service, seleniumwire_options=seleniumwire_options, options=options)

    return driver


# Requests를 사용하는 경우
def request_post(post_url, params, headers):
    proxies = {
        "http": "<Proxy Server>:<Proxy Port>",
        "https": "<Proxy Server>:<Proxy Port>",
        "no_proxy": "localhost,127.0.0.1 "
    }

    params_json = json.dumps(params)
    response = requests.post(post_url, data=params_json, headers=headers, proxies=proxies, verify=certifi.where())

    return response

내용이 유용하셨다면 좋아요&댓글 부탁드립니다.
이 블로그를 이끌어갈 수 있는 강력한 힘입니다!

caul334@gmail.com

Selenium wire 모듈을 사용하다보면 가끔 POST 요청으로 백그라운드에서 XHR 요청이 발생하는 경우 있습니다.

하지만 문제는 POST 요청에 대한 파라미터를 변경한 값을 결과 값으로 가져오고 싶은데 그게 어렵다는 문제입니다ㅠㅠ

이런 경우 아래와 같은 코드를 사용하면 Selenium wire 모듈을 통해서도 얼마든지 세션을 유지하면서 Post 요청을 사용할 수 있습니다.

  해결책 - Python 코드

from seleniumwire import webdriver
import requests

# WebDriver 설정
options = webdriver.ChromeOptions()
driver = webdriver.Chrome(seleniumwire_options={}, options=options)

# Selenium 모듈로 홈페이지 요청
url = 'https://example.com'
driver.get(url)

# POST 요청을 하는 특정 페이지 요청 값을 수정하여 requests 모듈로 재요청
for idx, request in enumerate(driver.requests):
	if '</api와 같은 홈페이지 uri 값>' in request.url and request.method == 'POST':
		post_url = request.url
		headers = request.headers
		params = {
			'key1': 'val1',
			'key2': 'val2'
		params_json = json.dumps(params)
		response = requests.post(post_url, data=params_json, headers=headers)
		print(response.text)
		break

caul334@gmail.com

내용이 유용하셨다면 좋아요&댓글 부탁드립니다.
이 블로그를 이끌어갈 수 있는 강력한 힘입니다!