#!/usr/bin/env python3 """ API для управления P2P сканером """ import json import subprocess import psutil import os import signal from pathlib import Path from datetime import datetime from typing import Dict, Any, Optional # Supported (broad) Binance P2P assets and fiat lists SUPPORTED_ASSETS = [ 'USDT','USDC','BTC','ETH','BNB','BUSD','DAI','BCH','LTC','XRP','TRX','DOGE','SOL','DOT', 'MATIC','ADA','LINK','AVAX','XLM','XMR','ATOM','NEAR','APT','SUI','OP','ARB','TON','ETC', 'FIL','EOS','FTM','ICP','KAS','KAVA','TUSD','FDUSD' ] SUPPORTED_FIATS = [ 'USD','EUR','RUB','UAH','KZT','TRY','BRL','ARS','NGN','VND','IDR','INR','GBP','AED','SAR', 'PLN','CZK','RON','HUF','GEL','AMD','AZN','UZS','KGS','TJS','CNY','HKD','TWD','KRW','JPY', 'MYR','PHP','THB','SGD','AUD','CAD','MXN','CLP','PEN','COP','ZAR','EGP','ILS' ] class ScannerManager: """Менеджер для управления процессом сканера""" def __init__(self): self.project_dir = Path(__file__).parent self.pid_file = self.project_dir / "scanner.pid" self.config_file = self.project_dir / "scanner_config.json" # Ensure logs directory exists and write scanner stdout there self.logs_dir = self.project_dir / "logs" try: self.logs_dir.mkdir(parents=True, exist_ok=True) except Exception: pass self.log_file = self.logs_dir / "scanner.log" # Загрузить конфигурацию по умолчанию self.default_config = { "refresh_interval": 30, "fiat_currency": "USD", "assets_to_track": ["USDT", "USDC"], "auto_start": False } def get_config(self) -> Dict[str, Any]: """Получить текущую конфигурацию""" if self.config_file.exists(): try: with open(self.config_file, 'r', encoding='utf-8') as f: return json.load(f) except Exception: pass return self.default_config.copy() def save_config(self, config: Dict[str, Any]) -> bool: """Сохранить конфигурацию""" try: with open(self.config_file, 'w', encoding='utf-8') as f: json.dump(config, f, indent=2, ensure_ascii=False) return True except Exception as e: print(f"Ошибка сохранения конфигурации: {e}") return False def is_running(self) -> bool: """Проверить, запущен ли сканер""" if not self.pid_file.exists(): return False try: with open(self.pid_file, 'r') as f: pid = int(f.read().strip()) # Проверить, существует ли процесс return psutil.pid_exists(pid) except (ValueError, FileNotFoundError): return False def get_scanner_pid(self) -> Optional[int]: """Получить PID процесса сканера""" if not self.pid_file.exists(): return None try: with open(self.pid_file, 'r') as f: return int(f.read().strip()) except (ValueError, FileNotFoundError): return None def start_scanner(self) -> Dict[str, Any]: """Запустить сканер""" if self.is_running(): return { "success": False, "message": "Сканер уже запущен", "status": "running" } try: # Запустить main.py в фоновом режиме process = subprocess.Popen( ["python", "main.py"], cwd=self.project_dir, stdout=open(self.log_file, 'a'), stderr=subprocess.STDOUT, preexec_fn=os.setsid # Создать новую группу процессов ) # Сохранить PID with open(self.pid_file, 'w') as f: f.write(str(process.pid)) return { "success": True, "message": "Сканер успешно запущен", "status": "running", "pid": process.pid } except Exception as e: return { "success": False, "message": f"Ошибка запуска сканера: {str(e)}", "status": "error" } def stop_scanner(self) -> Dict[str, Any]: """Остановить сканер""" if not self.is_running(): # Удалить файл PID если он существует if self.pid_file.exists(): self.pid_file.unlink() return { "success": True, "message": "Сканер уже остановлен", "status": "stopped" } try: pid = self.get_scanner_pid() if pid: # Завершить группу процессов os.killpg(os.getpgid(pid), signal.SIGTERM) # Удалить файл PID if self.pid_file.exists(): self.pid_file.unlink() return { "success": True, "message": "Сканер успешно остановлен", "status": "stopped" } except Exception as e: return { "success": False, "message": f"Ошибка остановки сканера: {str(e)}", "status": "error" } def restart_scanner(self) -> Dict[str, Any]: """Перезапустить сканер""" stop_result = self.stop_scanner() if not stop_result["success"]: return stop_result # Небольшая пауза перед запуском import time time.sleep(2) return self.start_scanner() def get_status(self) -> Dict[str, Any]: """Получить статус сканера""" is_running = self.is_running() pid = self.get_scanner_pid() if is_running else None status = { "status": "running" if is_running else "stopped", "pid": pid, "timestamp": datetime.now().isoformat(), "config": self.get_config() } if is_running and pid: try: process = psutil.Process(pid) status["memory_usage"] = process.memory_info().rss / 1024 / 1024 # MB status["cpu_percent"] = process.cpu_percent() status["start_time"] = datetime.fromtimestamp(process.create_time()).isoformat() except psutil.NoSuchProcess: status["status"] = "stopped" status["pid"] = None return status # Глобальный экземпляр менеджера scanner_manager = ScannerManager() def handle_api_request(method: str, path: str, data: Dict[str, Any] = None) -> Dict[str, Any]: """Обработать API запрос""" if path == "/api/scanner/status" and method == "GET": return scanner_manager.get_status() elif path == "/api/scanner/start" and method == "POST": return scanner_manager.start_scanner() elif path == "/api/scanner/stop" and method == "POST": return scanner_manager.stop_scanner() elif path == "/api/scanner/restart" and method == "POST": return scanner_manager.restart_scanner() elif path == "/api/scanner/config" and method == "GET": return { "success": True, "config": scanner_manager.get_config() } elif path == "/api/scanner/config" and method == "POST": if data: success = scanner_manager.save_config(data) return { "success": success, "message": "Конфигурация сохранена" if success else "Ошибка сохранения конфигурации" } else: return { "success": False, "message": "Не переданы данные конфигурации" } elif path == "/api/scanner/options" and method == "GET": # Вернуть поддерживаемые списки фиатов и активов (можно расширять по мере надобности) return { "success": True, "assets": SUPPORTED_ASSETS, "fiats": SUPPORTED_FIATS } else: return { "success": False, "message": "Неизвестный API endpoint", "error": "Not Found" } if __name__ == "__main__": # Тестирование API print("Тестирование Scanner Manager API") print("Статус:", scanner_manager.get_status()) print("Конфигурация:", scanner_manager.get_config())