fantos's picture
Update web.py
6fb051b verified
raw
history blame
3.05 kB
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'
)
# λͺ¨λ‹ˆν„°λ§ν•  URL λͺ©λ‘
URLS = [
"https://huggingface.co/spaces/ginipick/discord-openfree-LLM-chatgpt4",
"https://huggingface.co/spaces/fantos/discord-openfree-LLM-qwen3-30b-a3b"
]
# ν•‘ 간격 (초) - ν—ˆκΉ…νŽ˜μ΄μŠ€ 슬립 νƒ€μž„μ•„μ›ƒλ³΄λ‹€ μ§§μ•„μ•Ό 함
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()