#!/usr/bin/env python3 """ Backend serwer dla Euro Barcode Scanner z zaawansowaną ochroną anty-botową oraz pełnym monitoringiem zużycia zasobów i ruchu sieciowego (Euro.com.pl & Klienci). """ import sys import os import json import time import re import sqlite3 import threading from urllib.parse import urlparse, parse_qs from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler try: from curl_cffi import requests except ImportError: print("BŁĄD: Brak biblioteki curl_cffi. Zainstaluj: pip install curl_cffi") sys.exit(1) try: import resource except ImportError: resource = None BASE_DIR = os.path.dirname(os.path.abspath(__file__)) DB_PATH = os.path.join(BASE_DIR, "euro_cache.db") STATS_FILE_PATH = os.path.join(BASE_DIR, "stats.json") DEFAULT_PORT = 8000 # Dozwolone wzorce źródeł (Origin / Referer) ALLOWED_ORIGIN_PATTERNS = [ r"^https?://euro-skaner\.krzysio-te1\.workers\.dev/?$", r"^https?://euroapi\.dsh\.yt/?$", r"^https?://localhost(:\d+)?/?$", r"^https?://127\.0\.0\.1(:\d+)?/?$" ] TRUSTED_CLIENT_HEADER = "euro-skaner" # --- Pomocnicze funkcje formatowania --- def format_bytes(num_bytes): if num_bytes is None: return "0 B" num = float(num_bytes) for unit in ['B', 'KB', 'MB', 'GB']: if abs(num) < 1024.0: return f"{num:3.1f} {unit}" num /= 1024.0 return f"{num:.1f} TB" def format_uptime(seconds): seconds = int(seconds) days, seconds = divmod(seconds, 86400) hours, seconds = divmod(seconds, 3600) minutes, seconds = divmod(seconds, 60) parts = [] if days > 0: parts.append(f"{days}d") if hours > 0: parts.append(f"{hours}h") if minutes > 0: parts.append(f"{minutes}m") parts.append(f"{seconds}s") return " ".join(parts) def get_process_memory_mb(): try: with open("/proc/self/status") as f: for line in f: if line.startswith("VmRSS:"): parts = line.split() kb = int(parts[1]) return round(kb / 1024.0, 2) except Exception: pass if resource: try: return round(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024.0, 2) except Exception: pass return 0.0 # --- Menedżer Statystyk (Lokalny & Trwały w SQLite / stats.json) --- class StatsManager: def __init__(self): self._lock = threading.Lock() self.start_time = time.time() # Statystyki sesji (od bieżącego startu) self.session_api_requests = 0 self.session_api_single = 0 self.session_api_batch = 0 self.session_api_stats = 0 self.session_blocked_requests = 0 self.session_cache_hits = 0 self.session_cache_misses = 0 self.session_client_bytes_in = 0 self.session_client_bytes_out = 0 self.session_euro_requests = 0 self.session_euro_requests_success = 0 self.session_euro_requests_failed = 0 self.session_euro_bytes_sent = 0 self.session_euro_bytes_received = 0 self.session_rotations = 0 # Statystyki łączne (lifetime z bazy danych) self.lifetime_api_requests = 0 self.lifetime_blocked_requests = 0 self.lifetime_cache_hits = 0 self.lifetime_cache_misses = 0 self.lifetime_client_bytes_in = 0 self.lifetime_client_bytes_out = 0 self.lifetime_euro_requests = 0 self.lifetime_euro_requests_success = 0 self.lifetime_euro_requests_failed = 0 self.lifetime_euro_bytes_sent = 0 self.lifetime_euro_bytes_received = 0 self.lifetime_rotations = 0 self.last_flush = time.time() self._load_lifetime_stats() def _load_lifetime_stats(self): try: conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute(""" CREATE TABLE IF NOT EXISTS server_stats ( key TEXT PRIMARY KEY, val_num INTEGER, updated_at INTEGER ) """) c.execute("SELECT key, val_num FROM server_stats") rows = dict(c.fetchall()) conn.close() self.lifetime_api_requests = rows.get("api_requests", 0) self.lifetime_blocked_requests = rows.get("blocked_requests", 0) self.lifetime_cache_hits = rows.get("cache_hits", 0) self.lifetime_cache_misses = rows.get("cache_misses", 0) self.lifetime_client_bytes_in = rows.get("client_bytes_in", 0) self.lifetime_client_bytes_out = rows.get("client_bytes_out", 0) self.lifetime_euro_requests = rows.get("euro_requests", 0) self.lifetime_euro_requests_success = rows.get("euro_requests_success", 0) self.lifetime_euro_requests_failed = rows.get("euro_requests_failed", 0) self.lifetime_euro_bytes_sent = rows.get("euro_bytes_sent", 0) self.lifetime_euro_bytes_received = rows.get("euro_bytes_received", 0) self.lifetime_rotations = rows.get("rotations", 0) except Exception as e: print(f"[STATS LOAD ERR] {e}") def record_client_request(self, path, bytes_in): with self._lock: self.session_api_requests += 1 self.lifetime_api_requests += 1 self.session_client_bytes_in += bytes_in self.lifetime_client_bytes_in += bytes_in if path in ("/api/product", "/api/lookup"): self.session_api_single += 1 elif path in ("/api/products/batch", "/api/lookup/batch"): self.session_api_batch += 1 elif path in ("/stats", "/api/stats"): self.session_api_stats += 1 def record_client_response(self, bytes_out): with self._lock: self.session_client_bytes_out += bytes_out self.lifetime_client_bytes_out += bytes_out def record_blocked(self): with self._lock: self.session_blocked_requests += 1 self.lifetime_blocked_requests += 1 def record_cache_hit(self): with self._lock: self.session_cache_hits += 1 self.lifetime_cache_hits += 1 def record_cache_miss(self): with self._lock: self.session_cache_misses += 1 self.lifetime_cache_misses += 1 def record_euro_call(self, bytes_sent, bytes_received, is_success): with self._lock: self.session_euro_requests += 1 self.lifetime_euro_requests += 1 self.session_euro_bytes_sent += bytes_sent self.lifetime_euro_bytes_sent += bytes_sent self.session_euro_bytes_received += bytes_received self.lifetime_euro_bytes_received += bytes_received if is_success: self.session_euro_requests_success += 1 self.lifetime_euro_requests_success += 1 else: self.session_euro_requests_failed += 1 self.lifetime_euro_requests_failed += 1 def record_rotation(self): with self._lock: self.session_rotations += 1 self.lifetime_rotations += 1 def flush(self): with self._lock: try: conn = sqlite3.connect(DB_PATH) c = conn.cursor() now = int(time.time()) stats_list = [ ("api_requests", self.lifetime_api_requests), ("blocked_requests", self.lifetime_blocked_requests), ("cache_hits", self.lifetime_cache_hits), ("cache_misses", self.lifetime_cache_misses), ("client_bytes_in", self.lifetime_client_bytes_in), ("client_bytes_out", self.lifetime_client_bytes_out), ("euro_requests", self.lifetime_euro_requests), ("euro_requests_success", self.lifetime_euro_requests_success), ("euro_requests_failed", self.lifetime_euro_requests_failed), ("euro_bytes_sent", self.lifetime_euro_bytes_sent), ("euro_bytes_received", self.lifetime_euro_bytes_received), ("rotations", self.lifetime_rotations) ] for k, v in stats_list: c.execute("INSERT OR REPLACE INTO server_stats (key, val_num, updated_at) VALUES (?, ?, ?)", (k, v, now)) conn.commit() conn.close() self.last_flush = time.time() # Zapisz czytelny plik stats.json obok serwera summary = self.get_summary() with open(STATS_FILE_PATH, "w", encoding="utf-8") as f: json.dump(summary, f, indent=2, ensure_ascii=False) except Exception as e: print(f"[STATS FLUSH ERR] {e}") def get_summary(self): now = time.time() uptime_sec = now - self.start_time # Baza danych db_size = 0 cached_total = 0 cached_found = 0 cached_not_found = 0 try: if os.path.exists(DB_PATH): db_size = os.path.getsize(DB_PATH) conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute("SELECT count(*), sum(case when found=1 then 1 else 0 end) FROM product_cache") row = c.fetchone() if row: cached_total = row[0] or 0 cached_found = row[1] or 0 cached_not_found = cached_total - cached_found conn.close() except Exception: pass total_cache_ops = self.lifetime_cache_hits + self.lifetime_cache_misses hit_ratio_lifetime = round((self.lifetime_cache_hits / total_cache_ops * 100), 1) if total_cache_ops > 0 else 0.0 session_cache_ops = self.session_cache_hits + self.session_cache_misses hit_ratio_session = round((self.session_cache_hits / session_cache_ops * 100), 1) if session_cache_ops > 0 else 0.0 return { "status": "online", "uptime": { "seconds": int(uptime_sec), "formatted": format_uptime(uptime_sec), "started_at": int(self.start_time) }, "system_resources": { "process_memory_mb": get_process_memory_mb(), "process_memory_formatted": f"{get_process_memory_mb()} MB", "active_threads": threading.active_count() }, "cache_efficiency": { "hit_ratio_lifetime_percent": hit_ratio_lifetime, "hit_ratio_session_percent": hit_ratio_session, "hits_lifetime": self.lifetime_cache_hits, "misses_lifetime": self.lifetime_cache_misses, "hits_session": self.session_cache_hits, "misses_session": self.session_cache_misses, "euro_queries_saved": self.lifetime_cache_hits, "cached_products_total": cached_total, "cached_products_found": cached_found, "cached_products_not_found": cached_not_found, "db_file_size_bytes": db_size, "db_file_size_formatted": format_bytes(db_size) }, "euro_network_impact": { "total_queries_sent": self.lifetime_euro_requests, "successful_queries": self.lifetime_euro_requests_success, "failed_or_blocked_queries": self.lifetime_euro_requests_failed, "bytes_downloaded_from_euro": self.lifetime_euro_bytes_received, "bytes_downloaded_formatted": format_bytes(self.lifetime_euro_bytes_received), "bytes_uploaded_to_euro": self.lifetime_euro_bytes_sent, "bytes_uploaded_formatted": format_bytes(self.lifetime_euro_bytes_sent), "session_rotations": self.lifetime_rotations, "session_rotations_current_run": self.session_rotations }, "client_traffic": { "total_requests": self.lifetime_api_requests, "session_requests": self.session_api_requests, "session_single_lookups": self.session_api_single, "session_batch_lookups": self.session_api_batch, "session_stats_views": self.session_api_stats, "bytes_received_from_clients": self.lifetime_client_bytes_in, "bytes_received_formatted": format_bytes(self.lifetime_client_bytes_in), "bytes_sent_to_clients": self.lifetime_client_bytes_out, "bytes_sent_formatted": format_bytes(self.lifetime_client_bytes_out) }, "security_and_protection": { "blocked_abuse_attempts": self.lifetime_blocked_requests, "trusted_origin": "https://euro-skaner.krzysio-te1.workers.dev", "trusted_client_rate_limit": "300 req/min", "bot_untrusted_rate_limit": "10 req/min" } } stats_manager = StatsManager() # --- Ochrona Anty-Botowa & Rate Limiter --- class RateLimiter: def __init__(self): self._lock = threading.Lock() self._records = {} # ip -> list of timestamps def check(self, ip, max_per_minute): now = time.time() cutoff = now - 60.0 with self._lock: timestamps = self._records.get(ip, []) timestamps = [t for t in timestamps if t > cutoff] if len(timestamps) >= max_per_minute: self._records[ip] = timestamps return False timestamps.append(now) self._records[ip] = timestamps if len(self._records) > 2000: self._records = {k: v for k, v in self._records.items() if v and v[-1] > cutoff} return True rate_limiter = RateLimiter() def is_allowed_origin(origin): if not origin: return False return any(re.match(pat, origin, re.IGNORECASE) for pat in ALLOWED_ORIGIN_PATTERNS) # --- SQLite Cache Produktów --- def init_db(): conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute(""" CREATE TABLE IF NOT EXISTS product_cache ( code TEXT PRIMARY KEY, found INTEGER, plu TEXT, name TEXT, price REAL, promo_price REAL, url TEXT, image_url TEXT, raw_json TEXT, updated_at INTEGER ) """) c.execute(""" CREATE TABLE IF NOT EXISTS server_stats ( key TEXT PRIMARY KEY, val_num INTEGER, updated_at INTEGER ) """) conn.commit() conn.close() def format_price(val): if val is None: return None try: val = float(val) if val.is_integer(): formatted = f"{int(val):,}".replace(",", " ") else: formatted = f"{val:,.2f}".replace(",", " ").replace(".", ",") return f"{formatted} zł" except Exception: return f"{val} zł" def get_cached_product(code): try: conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute(""" SELECT found, plu, name, price, promo_price, url, image_url, updated_at FROM product_cache WHERE code = ? """, (code,)) row = c.fetchone() conn.close() if row: found, plu, name, price, promo_price, url, image_url, updated_at = row ttl = 86400 if found else 300 if time.time() - updated_at < ttl: stats_manager.record_cache_hit() if not found: return {"code": code, "found": False, "cached": True} return { "code": code, "found": True, "plu": plu, "name": name, "price": price, "promoPrice": promo_price, "priceFormatted": format_price(price), "promoPriceFormatted": format_price(promo_price), "url": url, "imageUrl": image_url, "updatedAt": updated_at, "cached": True } except Exception as e: print(f"[CACHE GET ERR] {e}") stats_manager.record_cache_miss() return None def save_cached_product(code, data): try: conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute(""" INSERT OR REPLACE INTO product_cache (code, found, plu, name, price, promo_price, url, image_url, raw_json, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( code, 1 if data.get("found") else 0, data.get("plu"), data.get("name"), data.get("price"), data.get("promoPrice"), data.get("url"), data.get("imageUrl"), json.dumps(data), int(time.time()) )) conn.commit() conn.close() except Exception as e: print(f"[CACHE SAVE ERR] {e}") # --- Klient do euro.com.pl z omijaniem Akamai i pomiarem ruchu --- class EuroScraperClient: def __init__(self): self._session = None self._created_at = 0 self._lock = threading.Lock() self._profiles = ["edge101", "chrome124", "safari17_0"] self._profile_idx = 0 def get_session(self): with self._lock: now = time.time() if self._session is None or (now - self._created_at > 900): self._init_session() return self._session def _init_session(self): profile = self._profiles[self._profile_idx % len(self._profiles)] sess = requests.Session(impersonate=profile) sess.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", "Accept-Language": "pl-PL,pl;q=0.9,en-US;q=0.8,en;q=0.7", "Referer": "https://www.euro.com.pl/", "Origin": "https://www.euro.com.pl" }) try: r = sess.get("https://www.euro.com.pl/", timeout=10) stats_manager.record_euro_call(bytes_sent=280, bytes_received=len(r.content), is_success=(r.status_code == 200)) self._session = sess self._created_at = time.time() print(f"[EuroClient] Zainicjalizowano sesję ({profile}) [status={r.status_code}, cookies={len(sess.cookies)}]") except Exception as e: stats_manager.record_euro_call(bytes_sent=280, bytes_received=0, is_success=False) print(f"[EuroClient] Błąd sesji ({profile}): {e}") self._session = sess self._created_at = time.time() def rotate_session(self): with self._lock: stats_manager.record_rotation() self._profile_idx += 1 self._init_session() euro_client = EuroScraperClient() def fetch_product_from_euro(code): clean_code = str(code).strip() if not clean_code: return {"code": code, "found": False, "error": "Pusty kod"} cached = get_cached_product(clean_code) if cached: return cached # 1. Retail Search API search_url = "https://www.euro.com.pl/v2/projects/euro-search/locations/global/catalogs/default_catalog/servingConfigs/default_search:search" search_body = { "query": clean_code, "pageSize": 5, "offset": 0, "visitorId": "0", "branch": "projects/euro-search/locations/global/catalogs/default_catalog/branches/0" } headers_search = { "Referer": "https://www.euro.com.pl/", "Origin": "https://www.euro.com.pl", "Accept": "application/json", "Content-Type": "application/json", "x-goog-user-project": "euro-search" } plu = None search_succeeded = False for attempt in range(2): s = euro_client.get_session() payload_bytes = len(json.dumps(search_body)) + 300 try: r = s.post(search_url, json=search_body, headers=headers_search, timeout=10) stats_manager.record_euro_call(bytes_sent=payload_bytes, bytes_received=len(r.content), is_success=(r.status_code == 200)) if r.status_code == 200: search_succeeded = True res_json = r.json() results = res_json.get("results", []) if results and len(results) > 0: plu = str(results[0].get("id")) break elif r.status_code in (403, 429): print(f"[EuroClient] Status {r.status_code} dla {clean_code}, rotacja sesji...") euro_client.rotate_session() except Exception as e: stats_manager.record_euro_call(bytes_sent=payload_bytes, bytes_received=0, is_success=False) print(f"[SEARCH ERR] {clean_code} (próba {attempt+1}): {e}") time.sleep(0.3) # Fallback jeśli Retail Search nic nie zwrócił, a kod ma format PLU (5-8 cyfr) if not plu and re.fullmatch(r'\d{5,8}', clean_code): plu = clean_code if not plu: res = {"code": clean_code, "found": False, "error": "Produkt nie został znaleziony w euro.com.pl"} if search_succeeded: save_cached_product(clean_code, res) return res # 2. Pobranie szczegółów produktu po PLU plu_url = f"https://www.euro.com.pl/rest/api/products/simple-by-plu-list?pluList={plu}" headers_common = { "Referer": "https://www.euro.com.pl/", "Origin": "https://www.euro.com.pl", "Accept": "application/json" } for attempt in range(2): s = euro_client.get_session() req_bytes = len(plu_url) + 250 try: r = s.get(plu_url, headers=headers_common, timeout=10) stats_manager.record_euro_call(bytes_sent=req_bytes, bytes_received=len(r.content), is_success=(r.status_code == 200)) if r.status_code == 200: items = r.json() if items and len(items) > 0: item = items[0] name = item.get("name") prices = item.get("prices") or {} main_price = prices.get("mainPrice") promo_price = prices.get("promotionalPrice") image_url = None images = item.get("images") or [] if images and len(images) > 0: image_url = images[0].get("url") product_url = f"https://www.euro.com.pl/search.bhtml?keyword={clean_code}" links = item.get("links") or {} view_link = links.get("view") if view_link: if view_link.startswith("http"): product_url = view_link else: product_url = f"https://www.euro.com.pl{view_link}" res = { "code": clean_code, "found": True, "plu": str(item.get("id", plu)), "name": name, "price": main_price, "promoPrice": promo_price, "priceFormatted": format_price(main_price), "promoPriceFormatted": format_price(promo_price), "url": product_url, "imageUrl": image_url, "updatedAt": int(time.time()) } save_cached_product(clean_code, res) return res elif r.status_code in (403, 429): euro_client.rotate_session() except Exception as e: stats_manager.record_euro_call(bytes_sent=req_bytes, bytes_received=0, is_success=False) print(f"[PLU ERR] {plu} (próba {attempt+1}): {e}") res = {"code": clean_code, "found": False, "error": "Nie udało się pobrać szczegółów produktu"} return res # --- Tło: Okresowy zapis statystyk co 30 sekund --- def stats_flusher_loop(): while True: time.sleep(30) try: stats_manager.flush() except Exception: pass # --- Szablon Dashboardu /stats --- STATS_HTML_TEMPLATE = """ Euro Skaner API - Status & Statystyki

📦 EURO SKANER API & MONITOR

Serwer proxy z cachem SQLite & ochroną anty-botową

ONLINE RAW JSON
💻 SYSTEM & ZASOBYPID: __PID__
Czas działania (Uptime):__UPTIME__
Zużycie pamięci RAM (RSS):__RAM__
Wątki procesora:__THREADS__
Rozmiar bazy euro_cache.db:__DB_SIZE__
⚡ EFEKTYWNOŚĆ CACHE__HIT_RATIO__% HIT
Trafienia z cache (Hits):__CACHE_HITS__
Odpytania Euro (Misses):__CACHE_MISSES__
Zaoszczędzone zapytania do Euro:__SAVED_QUERIES__
Zapisane produkty w bazie:__CACHED_PRODS__
🌐 OBCIĄŻENIE EURO.COM.PLOUTBOUND
Łącznie wysłanych zapytań:__EURO_REQ_TOTAL__
Prawidłowe odpowiedzi (200):__EURO_REQ_OK__
Błędy / Odrzucenia Akamai:__EURO_REQ_FAIL__
Pobrane dane z Euro.com.pl:__EURO_BYTES_RECV__
Wysłane dane do Euro:__EURO_BYTES_SENT__
📡 RUCH TWOJEJ SIECIINBOUND
Żądania od klientów (Lifetime):__CLIENT_REQ_TOTAL__
Żądania w bieżącej sesji:__CLIENT_REQ_SESSION__
Odebrane od klientów:__CLIENT_BYTES_IN__
Wysłane do klientów:__CLIENT_BYTES_OUT__
🛡️ BEZPIECZEŃSTWO & ANTY-ABUSEAKTYWNA
Zablokowane boty / nadużycia:__BLOCKED_REQ__
Limit autoryzowanej strony:300 req/min
Limit obcych botów / curl:10 req/min
Zaufany Origin:euro-skaner.krzysio-te1.workers.dev
""" # --- HTTP Handler --- class EuroServerHandler(SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=BASE_DIR, **kwargs) def _get_client_ip(self): cf_ip = self.headers.get("CF-Connecting-IP") if cf_ip: return cf_ip.strip() fwd = self.headers.get("X-Forwarded-For") if fwd: return fwd.split(",")[0].strip() real_ip = self.headers.get("X-Real-IP") if real_ip: return real_ip.strip() return self.client_address[0] def _is_trusted_client(self): origin = self.headers.get("Origin", "") referer = self.headers.get("Referer", "") client_tag = self.headers.get("X-Euro-Client", "") # 1. Custom nagłówek autoryzowanego klienta z naszej aplikacji if client_tag == TRUSTED_CLIENT_HEADER: return True # 2. Sprawdzenie Origin / Referer if is_allowed_origin(origin): return True if referer and any(re.match(pat, referer, re.IGNORECASE) for pat in ALLOWED_ORIGIN_PATTERNS): return True # 3. Połączenia lokalne ip = self._get_client_ip() if ip in ("127.0.0.1", "::1", "localhost"): return True return False def _check_access_and_rate_limit(self): ip = self._get_client_ip() origin = self.headers.get("Origin") # Jeśli request przyszedł z przeglądarki z NIEAUTORYZOWANEGO origin (inna strona próbuje ukraść API) if origin and not is_allowed_origin(origin): stats_manager.record_blocked() self._send_json({"error": "Forbidden: nieautoryzowane źródło (Origin)"}, 403) return False # Jeśli to zaufany klient (nasza strona na Cloudflare lub z poprawnym nagłówkiem): wysoki limit (300/min) if self._is_trusted_client(): if not rate_limiter.check(ip, max_per_minute=300): stats_manager.record_blocked() self._send_json({"error": "Zbyt wiele zapytań (Rate Limit). Zwolnij tempo.", "retry_after": 60}, 429) return False return True # Jeśli to obcy bot / bezpośredni skaner: ścisły limit (10/min) if not rate_limiter.check(ip, max_per_minute=10): stats_manager.record_blocked() self._send_json({"error": "Too Many Requests (Rate limit dla niezaufanych klientów: 10/min)", "retry_after": 60}, 429) return False return True def _send_cors_headers(self): origin = self.headers.get("Origin", "") # Odsyłaj origin jeśli dozwolony, w przeciwnym razie ogólny * if is_allowed_origin(origin): self.send_header("Access-Control-Allow-Origin", origin) else: self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization, X-Euro-Client") self.send_header("Access-Control-Max-Age", "86400") def do_OPTIONS(self): self.send_response(204) self._send_cors_headers() self.end_headers() def do_GET(self): parsed = urlparse(self.path) path = parsed.path bytes_in = int(self.headers.get("Content-Length", 0)) + len(self.path) + 150 stats_manager.record_client_request(path, bytes_in) # 1. Endpointy API z ochroną anty-botową if path in ("/api/product", "/api/lookup"): if not self._check_access_and_rate_limit(): return query = parse_qs(parsed.query) code = query.get("code", [None])[0] if not code: self._send_json({"error": "Brak parametru code"}, 400) return result = fetch_product_from_euro(code) self._send_json(result) return # 2. Statystyki w formacie JSON elif path == "/api/stats": summary = stats_manager.get_summary() self._send_json(summary) return # 3. Wizualny Dashboard Statystyk HTML (/stats) elif path in ("/stats", "/stats/"): summary = stats_manager.get_summary() html = STATS_HTML_TEMPLATE replacements = { "__PID__": str(os.getpid()), "__UPTIME__": summary["uptime"]["formatted"], "__RAM__": summary["system_resources"]["process_memory_formatted"], "__THREADS__": str(summary["system_resources"]["active_threads"]), "__DB_SIZE__": summary["cache_efficiency"]["db_file_size_formatted"], "__HIT_RATIO__": str(summary["cache_efficiency"]["hit_ratio_lifetime_percent"]), "__CACHE_HITS__": str(summary["cache_efficiency"]["hits_lifetime"]), "__CACHE_MISSES__": str(summary["cache_efficiency"]["misses_lifetime"]), "__SAVED_QUERIES__": str(summary["cache_efficiency"]["euro_queries_saved"]), "__CACHED_PRODS__": str(summary["cache_efficiency"]["cached_products_total"]), "__EURO_REQ_TOTAL__": str(summary["euro_network_impact"]["total_queries_sent"]), "__EURO_REQ_OK__": str(summary["euro_network_impact"]["successful_queries"]), "__EURO_REQ_FAIL__": str(summary["euro_network_impact"]["failed_or_blocked_queries"]), "__EURO_BYTES_RECV__": summary["euro_network_impact"]["bytes_downloaded_formatted"], "__EURO_BYTES_SENT__": summary["euro_network_impact"]["bytes_uploaded_formatted"], "__CLIENT_REQ_TOTAL__": str(summary["client_traffic"]["total_requests"]), "__CLIENT_REQ_SESSION__": str(summary["client_traffic"]["session_requests"]), "__CLIENT_BYTES_IN__": summary["client_traffic"]["bytes_received_formatted"], "__CLIENT_BYTES_OUT__": summary["client_traffic"]["bytes_sent_formatted"], "__BLOCKED_REQ__": str(summary["security_and_protection"]["blocked_abuse_attempts"]) } for k, v in replacements.items(): html = html.replace(k, v) body = html.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self._send_cors_headers() self.end_headers() self.wfile.write(body) stats_manager.record_client_response(len(body)) return elif path == "/api/health": summary = stats_manager.get_summary() self._send_json({"status": "ok", "uptime": summary["uptime"]["formatted"], "cached_products": summary["cache_efficiency"]["cached_products_total"]}) return # Domyślnie serwuj pliki statyczne (HTML, JS, obrazy) super().do_GET() def do_POST(self): parsed = urlparse(self.path) path = parsed.path bytes_in = int(self.headers.get("Content-Length", 0)) + len(self.path) + 150 stats_manager.record_client_request(path, bytes_in) if path in ("/api/products/batch", "/api/lookup/batch"): if not self._check_access_and_rate_limit(): return try: content_len = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_len) data = json.loads(body.decode("utf-8")) codes = data.get("codes", []) results = {} for code in codes: results[str(code)] = fetch_product_from_euro(str(code)) self._send_json({"results": results}) except Exception as e: self._send_json({"error": str(e)}, 500) return self._send_json({"error": "Nieznana ścieżka POST"}, 404) def _send_json(self, data, status=200): body = json.dumps(data, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self._send_cors_headers() self.end_headers() self.wfile.write(body) stats_manager.record_client_response(len(body)) def log_message(self, format, *args): sys.stderr.write(f"[{self.log_date_time_string()}] {format % args}\n") def run(port=DEFAULT_PORT): init_db() # Uruchom w tle flusher statystyk (zapisuje co 30s do SQLite i stats.json) flusher_thread = threading.Thread(target=stats_flusher_loop, daemon=True) flusher_thread.start() server_address = ("0.0.0.0", port) httpd = ThreadingHTTPServer(server_address, EuroServerHandler) print(f"🚀 Euro Skaner Backend uruchomiony na porcie {port}") print(f"🔗 Aplikacja: http://localhost:{port}") print(f"📊 Dashboard statystyk: http://localhost:{port}/stats") print(f"📄 Statystyki JSON: http://localhost:{port}/api/stats") print(f"🛡️ Ochrona anty-botowa: WŁĄCZONA (Zaufany Origin: https://euro-skaner.krzysio-te1.workers.dev)") try: httpd.serve_forever() except KeyboardInterrupt: print("\nZatrzymywanie serwera i zapisywanie statystyk...") stats_manager.flush() httpd.server_close() if __name__ == "__main__": port = int(os.environ.get("PORT", sys.argv[1] if len(sys.argv) > 1 else DEFAULT_PORT)) run(port)