web.py

from http.server import HTTPServer, BaseHTTPRequestHandler
import datetime
import json
import threading
import time
import requests
import logging
from urllib3.exceptions import InsecureRequestWarning

# SSL 경고 비활성화
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

# 로깅 설정
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

URLS = [
    "https://huggingface.co/spaces/ginipick/discord-openfree-LLM-chatgpt4",
    "https://huggingface.co/spaces/fantos/discord-openfree-LLM-qwen3-30b-a3b",
    "https://huggingface.co/spaces/fantos/discord-openfree-LLM-qwen3-235b-a22b",
    "https://huggingface.co/spaces/fantos/discord-openfree-LLM-llama4-maverick-instruct",
    "https://huggingface.co/spaces/fantos/discord-openfree-LLM-llama4-scout-instruct",
    "https://huggingface.co/spaces/fantos/discord-openfree-LLM-deepseek-v3-0324",
    "https://huggingface.co/spaces/fantos/discord-openfree-Image-sdxl-lightning",
    "https://huggingface.co/spaces/fantos/discord-openfree-Image-flux",
    "https://huggingface.co/spaces/fantos/discord-openfree-Image-3d",
    "https://huggingface.co/spaces/fantos/discord-openfree-video-luma",
    "https://huggingface.co/spaces/fantos/discord-openfree-video-luma2"
]

# 핑 간격 (초) - 허깅페이스 슬립 타임아웃보다 짧아야 함
PING_INTERVAL = 300  # 5분마다 핑

class HealthCheckHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        # 기본 헤더 설정
        self.send_response(200)
        self.send_header('Content-type', 'application/json')
        self.send_header('Access-Control-Allow-Origin', '*')
        self.end_headers()
        
        # 상태 정보 준비
        current_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        status_info = {
            "status": "ok",
            "timestamp": current_time,
            "message": "Service is running"
        }
        
        # JSON 응답 반환
        self.wfile.write(json.dumps(status_info).encode())

def check_url(url):
    """URL 상태를 확인하고 결과를 반환합니다."""
    try:
        # HEAD 요청이 GET보다 빠름 (본문 없이 헤더만 반환)
        response = requests.head(
            url, 
            verify=False,  # SSL 검증 비활성화
            timeout=10,    # 타임아웃 설정
            allow_redirects=True  # 리다이렉트 허용
        )
        status_code = response.status_code
        
        # 200 OK인 경우 성공
        if status_code == 200:
            result = f"상태 코드: {status_code}, 접속 상태: 접속 성공"
        else:
            result = f"상태 코드: {status_code}, 접속 상태: 접속 실패"
    except Exception as e:
        result = f"접속 실패: {str(e)}"
    
    logging.info(f"URL: {url} {result}")
    return result

def ping_thread():
    """
    주기적으로 URL들을 확인하여 슬립 모드로 전환되지 않도록 합니다.
    """
    logging.info("자동 핑 스레드 시작")
    
    while True:
        for url in URLS:
            check_url(url)
        
        # 다음 핑까지 대기
        time.sleep(PING_INTERVAL)

def run_server(port=7860):
    """
    헬스 체크 서버를 실행하고 동시에 백그라운드 핑 스레드를 시작합니다.
    """
    # 핑 스레드 시작
    ping_task = threading.Thread(target=ping_thread, daemon=True)
    ping_task.start()
    
    # 웹 서버 설정
    server_address = ('0.0.0.0', port)
    httpd = HTTPServer(server_address, HealthCheckHandler)
    
    logging.info(f"Starting health check server on port {port}")
    httpd.serve_forever()
    
if __name__ == "__main__":
    run_server()