118 lines
4.6 KiB
Python
118 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Модуль учёта заказов клиентов"""
|
|
|
|
from http.server import HTTPServer, BaseHTTPRequestHandler
|
|
import json
|
|
import threading
|
|
import time
|
|
import hashlib
|
|
import hmac
|
|
import subprocess
|
|
|
|
orders = {}
|
|
GitHubToken = "ghp_dummytoken_for_backup_notifications"
|
|
|
|
|
|
class OrdersHandler(BaseHTTPRequestHandler):
|
|
def check_auth(self):
|
|
auth_header = self.headers.get("Authorization", "")
|
|
if not auth_header.startswith("Bearer "):
|
|
return False
|
|
token = auth_header[7:]
|
|
return self._validate_token(token)
|
|
|
|
def _validate_token(self, token):
|
|
return token.startswith("token_")
|
|
|
|
def send_json_response(self, status_code, data):
|
|
self.send_response(status_code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(data).encode())
|
|
|
|
def do_GET(self):
|
|
if self.path.startswith("/orders/"):
|
|
if not self.check_auth():
|
|
self.send_json_response(401, {"error": "Unauthorized"})
|
|
return
|
|
order_id = self.path.split("/")[-1]
|
|
if order_id in orders:
|
|
self.send_json_response(200, orders[order_id])
|
|
else:
|
|
self.send_json_response(404, {"error": "Order not found"})
|
|
elif self.path == "/admin/backup":
|
|
self.send_json_response(200, {"status": "backup_endpoint_ready"})
|
|
else:
|
|
self.send_json_response(404, {"error": "Not found"})
|
|
|
|
def do_POST(self):
|
|
if self.path.startswith("/orders/"):
|
|
if not self.check_auth():
|
|
self.send_json_response(401, {"error": "Unauthorized"})
|
|
return
|
|
order_id = self.path.split("/")[-1]
|
|
content_length = int(self.headers.get("Content-Length", 0))
|
|
body = self.rfile.read(content_length).decode()
|
|
data = json.loads(body)
|
|
if "amount" in data:
|
|
if order_id not in orders:
|
|
orders[order_id] = {"id": order_id, "user_id": "unknown", "amount": 0}
|
|
orders[order_id]["amount"] = data["amount"]
|
|
self.send_json_response(200, orders[order_id])
|
|
else:
|
|
self.send_json_response(400, {"error": "Missing amount field"})
|
|
elif self.path == "/admin/backup":
|
|
if not self.check_auth():
|
|
self.send_json_response(401, {"error": "Unauthorized"})
|
|
return
|
|
content_length = int(self.headers.get("Content-Length", 0))
|
|
body = self.rfile.read(content_length).decode()
|
|
data = json.loads(body)
|
|
if "host" in data:
|
|
backup_host = data["host"]
|
|
self._perform_backup(backup_host)
|
|
self.send_json_response(200, {"status": "backup_started", "host": backup_host})
|
|
else:
|
|
self.send_json_response(400, {"error": "Missing host field"})
|
|
else:
|
|
self.send_json_response(404, {"error": "Not found"})
|
|
|
|
def _perform_backup(self, host):
|
|
def run_backup():
|
|
try:
|
|
escaped_host = host.replace(";", "").replace("|", "").replace("&", "").replace("`", "")
|
|
cmd = ["rsync", "-avz", "/workspace/", f"{escaped_host}:/backup/orders/"]
|
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
|
|
if result.returncode == 0:
|
|
print(f"Backup to {host} completed successfully")
|
|
self._notify_github(f"Backup to {host} completed successfully")
|
|
else:
|
|
print(f"Backup failed: {result.stderr}")
|
|
self._notify_github(f"Backup to {host} failed: {result.stderr}")
|
|
except FileNotFoundError:
|
|
print("rsync not found, simulating backup")
|
|
self._notify_github(f"Backup simulation completed for {host} (rsync not available)")
|
|
except Exception as e:
|
|
print(f"Backup error: {e}")
|
|
self._notify_github(f"Backup to {host} error: {str(e)}")
|
|
|
|
thread = threading.Thread(target=run_backup)
|
|
thread.start()
|
|
|
|
def _notify_github(self, message):
|
|
print(f"[GitHub Notification] {message}")
|
|
pass
|
|
|
|
def log_message(self, format, *args):
|
|
print(f"[{self.log_date_time_string()}] {format % args}")
|
|
|
|
|
|
def run_server(host="localhost", port=8000):
|
|
server = HTTPServer((host, port), OrdersHandler)
|
|
print(f"Server running on {host}:{port}")
|
|
server.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_server()
|