175 lines
6.9 KiB
Python
175 lines
6.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
リアルタイム チェックイン監視ツール
|
|
スマホアプリからのチェックイン試行をリアルタイムで監視
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import django
|
|
import subprocess
|
|
import time
|
|
import json
|
|
import requests
|
|
from datetime import datetime, timedelta
|
|
import threading
|
|
from collections import defaultdict
|
|
|
|
# Django設定
|
|
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
|
|
django.setup()
|
|
|
|
from rog.models import GpsLog, Entry, Team, NewEvent2
|
|
|
|
class CheckinMonitor:
|
|
def __init__(self):
|
|
self.last_check = datetime.now()
|
|
self.request_counts = defaultdict(int)
|
|
|
|
def check_recent_gpslog(self):
|
|
"""最近のGpsLog エントリーをチェック"""
|
|
try:
|
|
recent_logs = GpsLog.objects.filter(
|
|
create_at__gte=self.last_check
|
|
).order_by('-create_at')
|
|
|
|
if recent_logs.exists():
|
|
print(f"\n🆕 新しいGpsLogエントリー ({recent_logs.count()}件):")
|
|
for log in recent_logs:
|
|
print(f" ✅ ID:{log.id} イベント:{log.event_code} ゼッケン:{log.zekken_number} CP:{log.cp_number} 時刻:{log.create_at}")
|
|
return True
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ GpsLog確認エラー: {e}")
|
|
return False
|
|
|
|
def check_recent_entries(self):
|
|
"""最近のEntry更新をチェック"""
|
|
try:
|
|
recent_entries = Entry.objects.filter(
|
|
start_time__gte=self.last_check
|
|
).order_by('-start_time')
|
|
|
|
if recent_entries.exists():
|
|
print(f"\n🏁 新しいスタート ({recent_entries.count()}件):")
|
|
for entry in recent_entries:
|
|
team_name = entry.team.team_name if entry.team else "N/A"
|
|
event_name = entry.event.event_name if entry.event else "N/A"
|
|
print(f" 🚀 エントリーID:{entry.id} チーム:{team_name} イベント:{event_name} スタート時刻:{entry.start_time}")
|
|
return True
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ Entry確認エラー: {e}")
|
|
return False
|
|
|
|
def check_nginx_logs(self):
|
|
"""nginxログから最近のAPIアクセスを確認"""
|
|
try:
|
|
# Dockerログからnginxのアクセスログを取得
|
|
cmd = ["docker", "compose", "logs", "--tail=20", "nginx"]
|
|
result = subprocess.run(cmd, capture_output=True, text=True, cwd="/Volumes/PortableSSD1TB/main/GifuTabi/rogaining_srv_exdb-2/rogaining_srv")
|
|
|
|
if result.returncode == 0:
|
|
lines = result.stdout.split('\n')
|
|
api_requests = []
|
|
|
|
for line in lines:
|
|
if 'checkin_from_rogapp' in line or 'start_from_rogapp' in line:
|
|
api_requests.append(line)
|
|
elif any(endpoint in line for endpoint in ['/api/user/', '/api/teams/', '/api/entry/']):
|
|
api_requests.append(line)
|
|
|
|
if api_requests:
|
|
print(f"\n📡 最近のAPI アクセス ({len(api_requests)}件):")
|
|
for req in api_requests[-5:]: # 最新5件のみ表示
|
|
print(f" 📥 {req.strip()}")
|
|
return True
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ nginxログ確認エラー: {e}")
|
|
return False
|
|
|
|
def check_app_logs(self):
|
|
"""アプリケーションログから最近のエラーを確認"""
|
|
try:
|
|
cmd = ["docker", "compose", "logs", "--tail=10", "app"]
|
|
result = subprocess.run(cmd, capture_output=True, text=True, cwd="/Volumes/PortableSSD1TB/main/GifuTabi/rogaining_srv_exdb-2/rogaining_srv")
|
|
|
|
if result.returncode == 0:
|
|
lines = result.stdout.split('\n')
|
|
error_logs = []
|
|
|
|
for line in lines:
|
|
if any(keyword in line.lower() for keyword in ['error', 'warning', 'exception', 'failed', 'api_play']):
|
|
error_logs.append(line)
|
|
|
|
if error_logs:
|
|
print(f"\n⚠️ 最近のアプリケーションログ ({len(error_logs)}件):")
|
|
for log in error_logs[-3:]: # 最新3件のみ表示
|
|
print(f" 🔍 {log.strip()}")
|
|
return True
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ アプリケーションログ確認エラー: {e}")
|
|
return False
|
|
|
|
def test_checkin_endpoints(self):
|
|
"""チェックインエンドポイントの動作テスト"""
|
|
endpoints = [
|
|
"http://localhost:8100/api/checkin_from_rogapp",
|
|
"http://localhost:8100/gifuroge/checkin_from_rogapp"
|
|
]
|
|
|
|
print(f"\n🔧 チェックインエンドポイント動作確認:")
|
|
for endpoint in endpoints:
|
|
try:
|
|
response = requests.get(endpoint, timeout=5)
|
|
status_color = "✅" if response.status_code == 405 else "❌"
|
|
print(f" {status_color} {endpoint} → HTTP {response.status_code}")
|
|
except Exception as e:
|
|
print(f" ❌ {endpoint} → エラー: {e}")
|
|
|
|
def run_monitor(self):
|
|
"""メイン監視ループ"""
|
|
print("🚀 リアルタイム チェックイン監視開始")
|
|
print("=" * 60)
|
|
|
|
while True:
|
|
try:
|
|
current_time = datetime.now()
|
|
print(f"\n⏰ 監視時刻: {current_time.strftime('%Y-%m-%d %H:%M:%S')}")
|
|
|
|
# 各種チェック実行
|
|
has_new_data = False
|
|
has_new_data |= self.check_recent_gpslog()
|
|
has_new_data |= self.check_recent_entries()
|
|
has_new_data |= self.check_nginx_logs()
|
|
has_new_data |= self.check_app_logs()
|
|
|
|
# 10分毎にエンドポイントテスト
|
|
if current_time.minute % 10 == 0:
|
|
self.test_checkin_endpoints()
|
|
|
|
if not has_new_data:
|
|
print(" 💤 新しいアクティビティなし")
|
|
|
|
# 次回チェック時刻を更新
|
|
self.last_check = current_time
|
|
|
|
print("-" * 40)
|
|
time.sleep(30) # 30秒間隔で監視
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n🛑 監視を停止します")
|
|
break
|
|
except Exception as e:
|
|
print(f"❌ 監視エラー: {e}")
|
|
time.sleep(5)
|
|
|
|
def main():
|
|
monitor = CheckinMonitor()
|
|
monitor.run_monitor()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|