ABOUT ME

-

Today
-
Yesterday
-
Total
-
  • 장고에서 스케줄러로 서버 아이피 변동 확인
    Python/초간단 장고 Django 2024. 1. 19. 08:33
    반응형

    이전 글 참고:

     

    파이썬으로 사용 중인 컴퓨터의 ip 찾기

    파이썬으로 본인 컴의 아이피를 찾아보자. import json import socket import urllib.request with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: sock.connect(("8.8.8.8", 80)) print('내부 :', sock.getsockname()[0]) with urllib.reque

    comdoc.tistory.com

    pip install apscheduler

    django-apscheduler라는 라이브러리도 있지만,
    하나의 일만을 실행하는 단순한 코드라...
    apscheduler만 사용했다. 

    from django.apps import AppConfig
    
    
    class CheckIpConfig(AppConfig):
        default_auto_field = 'django.db.models.BigAutoField'
        name = 'check_ip'
    
        def ready(self):
            from .runapscheduler import start
            start()

    장고의  settings.py 의 인스톨드_앱스에서 앱을 등록하지 않아도 urls.py는 작동한다...
    그럼 인스톨드_앱스는 뭐에 쓰는 물건일까?

    DB 관련해서 인스톨드_앱스를 활용하는 것은
    migrate를 해보면서 경험해 본 적 있을 것이다.

    ready 메소드로 앱의 초기화 코드를 설정할 수 있는데,
    인스톨드 앱에 등록되어 있지 않으면 작동하지 않는다. 

    즉 '장고 프레임워크에 설치된 앱을 알려준다'는 것은
    장고에서 그 앱의 DB를 관리하게 해주고,
    초기화에 필요한 코드들을 실행해준다는 의미가 있다.  

    INSTALLED_APPS = [
        'check_ip',
        'django.contrib.admin',
        'django.contrib.auth',
        'django.contrib.contenttypes',
        'django.contrib.sessions',
        'django.contrib.messages',
        'django.contrib.staticfiles',
    ]

    텔레그램 메시지 송신에는 별다른 라이브러리를 쓰지 않았고,
    텔레그램 서버와의 웹 통신에도 흔히 쓰는 외부 라이브러리인 requests 말고
    기본 라이브러리인 urllib를 이용했다. 

    이 과정에서 'ssl._create_default_https_context = ssl._create_unverified_context' 같은 편법이 사용된 것이 함정...
    파이썬을 배웠다면 _으로 시작하는 것은 외부에서 건드리지 않는 것을 원칙으로 한다는 것을 알고 있을 것이다.

    그렇다 그냥 requests를 쓰자 ~! 

    #runapscheduler.py 2024.1.20 Comdoc
    
    import json
    import logging
    import os
    import ssl
    from urllib import request, parse
    
    from apscheduler.schedulers.background import BackgroundScheduler
    
    TELEGRAM_BOT_TOKEN = ''  # 
    TELEGRAM_CHAT_ID = ''  #
    IP_FILE_PATH = "check_ip.json"
    ssl._create_default_https_context = ssl._create_unverified_context
    
    
    def get_current_ip():
        try:
            url = 'https://api64.ipify.org?format=json'
            req = request.Request(url)
            res = request.urlopen(req)
            decoded = res.read().decode("utf8")
            json_dict = json.loads(decoded)
            return json_dict['ip']
        except Exception as e:
            logging.error(f'Error getting IP address: {e}')
            return None
    
    
    def get_previous_ip():
        if os.path.exists(IP_FILE_PATH):
            with open(IP_FILE_PATH, 'r') as file:
                return json.load(file).get('ip')
        return None
    
    
    def save_current_ip(ip):
        with open(IP_FILE_PATH, 'w') as file:
            json.dump({'ip': ip}, file)
    
    
    def send_telegram_message(message):
        try:
            url = f'https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage'
            req = request.Request(url)
            data = {'chat_id': TELEGRAM_CHAT_ID, 'text': message}
            post_data = parse.urlencode(data).encode('UTF-8')
            res = request.urlopen(req, post_data)
            decoded = res.read().decode('UTF-8')
            json_dict = json.loads(decoded)
            if not json_dict.get('ok'):
                logging.error(json_dict.get('result'))
            else:
                result = json_dict['result']
                logging.info(f"[telegram] date:{result['date']}, text:{result['text']}")
        except Exception as e:
            logging.error(f'Error sending message: {e}')
            return None
    
    
    def check_ip_changes():
        current_ip = get_current_ip()
        if current_ip:
            previous_ip = get_previous_ip()
            print(current_ip, previous_ip)
            logging.debug(f'IP CHECK:{current_ip}, {previous_ip}')
            if current_ip != previous_ip:
                message = f'IP address has changed!\n\nPrevious IP: {previous_ip}\nCurrent IP: {current_ip}'
                send_telegram_message(message)
                save_current_ip(current_ip)
    
    
    def start():
        logging.basicConfig(filename="check_ip.log", level=logging.INFO)
    
        scheduler = BackgroundScheduler()
        scheduler.add_job(check_ip_changes, 'interval', hours=1)
    
        check_ip_changes()
        scheduler.start()

    로컬에서 runserver사용시 2번씩 실행되는 것은 이렇게 임시 해결책이 있고,
    굳이 이런 방법을 안 쓰더라도, 서버 환경에 올라가면 정상적으로 작동하니 큰 문제는 없다. 

    python manage.py runserver --noreload
    반응형
Designed by Tistory.