| |
| import os |
| import json |
| import time |
| import threading |
| import requests |
| from http.server import HTTPServer, BaseHTTPRequestHandler |
| from tracker import load_tracker |
| import settings_manager |
|
|
| class DashboardHandler(BaseHTTPRequestHandler): |
| def do_GET(self): |
| |
| path = self.path.split('?')[0] |
| if path == '/': |
| self.serve_dashboard() |
| elif path == '/api/logs': |
| self.serve_logs() |
| elif path == '/api/stats': |
| self.serve_stats() |
| elif path == '/api/settings': |
| self.serve_settings() |
| else: |
| self.send_response(404) |
| self.end_headers() |
|
|
| def do_POST(self): |
| path = self.path.split('?')[0] |
| if path == '/api/save_settings': |
| self.save_settings() |
| else: |
| self.send_response(404) |
| self.end_headers() |
|
|
| def do_HEAD(self): |
| path = self.path.split('?')[0] |
| if path in ['/', '/api/logs', '/api/stats']: |
| self.send_response(200) |
| self.end_headers() |
| else: |
| self.send_response(404) |
| self.end_headers() |
|
|
| def serve_dashboard(self): |
| try: |
| with open("dashboard.html", "r", encoding="utf-8") as f: |
| html = f.read() |
| self.send_response(200) |
| self.send_header("Content-type", "text/html; charset=utf-8") |
| self.end_headers() |
| self.wfile.write(html.encode('utf-8')) |
| except Exception as e: |
| self.send_response(500) |
| self.end_headers() |
| self.wfile.write(f"Error loading dashboard: {e}".encode('utf-8')) |
|
|
| def serve_logs(self): |
| self.send_response(200) |
| self.send_header("Content-type", "application/json; charset=utf-8") |
| self.end_headers() |
| |
| log_content = "" |
| log_path = os.path.join("output", "bot.log") |
| if os.path.exists(log_path): |
| try: |
| with open(log_path, "r", encoding="utf-8") as f: |
| lines = f.readlines() |
| log_content = "".join(lines[-100:]) |
| except Exception: |
| pass |
| |
| if not log_content: |
| log_content = "Ожидание логов бота..." |
| |
| response = {"logs": log_content} |
| self.wfile.write(json.dumps(response).encode('utf-8')) |
|
|
| def serve_stats(self): |
| self.send_response(200) |
| self.send_header("Content-type", "application/json; charset=utf-8") |
| self.end_headers() |
| |
| try: |
| data = load_tracker() |
| total_posts = len(data) |
| total_likes = sum(post.get("likes", 0) for post in data) |
| total_replies = sum(post.get("replies", 0) for post in data) |
| |
| top_posts = sorted(data, key=lambda x: x.get("likes", 0) + x.get("replies", 0), reverse=True)[:3] |
| |
| response = { |
| "total_posts": total_posts, |
| "total_likes": total_likes, |
| "total_replies": total_replies, |
| "top_posts": top_posts |
| } |
| except Exception as e: |
| response = { |
| "total_posts": 0, |
| "total_likes": 0, |
| "total_replies": 0, |
| "top_posts": [] |
| } |
| |
| self.wfile.write(json.dumps(response, ensure_ascii=False).encode('utf-8')) |
|
|
| def serve_settings(self): |
| self.send_response(200) |
| self.send_header("Content-type", "application/json; charset=utf-8") |
| self.end_headers() |
| settings = settings_manager.load_settings() |
| self.wfile.write(json.dumps(settings, ensure_ascii=False).encode('utf-8')) |
|
|
| def save_settings(self): |
| content_length = int(self.headers.get('Content-Length', 0)) |
| post_data = self.rfile.read(content_length) |
| |
| try: |
| new_settings = json.loads(post_data.decode('utf-8')) |
| if settings_manager.save_settings(new_settings): |
| self.send_response(200) |
| self.send_header("Content-type", "application/json; charset=utf-8") |
| self.end_headers() |
| self.wfile.write(json.dumps({"status": "success"}).encode('utf-8')) |
| else: |
| self.send_response(500) |
| self.end_headers() |
| except Exception as e: |
| self.send_response(400) |
| self.end_headers() |
|
|
| def run_server(): |
| server = HTTPServer(("0.0.0.0", 7860), DashboardHandler) |
| print("[Keep-Alive] Веб-сервер дашборда запущен на порту 7860") |
| server.serve_forever() |
|
|
| def self_ping_loop(): |
| time.sleep(10) |
| hf_url = os.environ.get("HF_SPACE_URL") |
| if not hf_url: |
| print("[Keep-Alive] HF_SPACE_URL не задан. Self-ping отключен.") |
| return |
| |
| print(f"[Keep-Alive] Запуск механизма self-ping для URL: {hf_url}") |
| while True: |
| try: |
| r = requests.get(hf_url, timeout=10) |
| print(f"[Keep-Alive] Успешный пинг собственного Space ({r.status_code})") |
| except Exception as e: |
| print(f"[Keep-Alive] Ошибка при пинге: {e}") |
| time.sleep(600) |
|
|
| if __name__ == "__main__": |
| ping_thread = threading.Thread(target=self_ping_loop, daemon=True) |
| ping_thread.start() |
| run_server() |
|
|
