#!/usr/bin/env python3 # -*- coding: utf-8 -*- """GIAM_SAT.PY — theo dõi dịch vụ, báo Telegram khi hỏng / hồi phục. VÌ SAO (26/09/2026) ------------------- Chủ dự án: "làm script theo dõi báo tele" — sau khi bật Cloudflare, cần biết NGAY khi website, máy key hay Cloudflare trục trặc, trước khi khách phàn nàn. Hai máy THEO DÕI CHÉO: một máy không tự báo được lúc chính nó sập / mất mạng. - máy key : website qua Cloudflare (đúng như khách thấy) + dịch vụ của nó - máy web : api1/health + Redis máy key (đúng như EA thấy) + dịch vụ, đĩa, lượng truy cập tăng đột biến (báo sớm DDoS) CHỐNG SPAM: lỗi LIEN_TIEP lần liên tiếp mới báo; còn lỗi thì nhắc mỗi NHAC_PHUT phút; hồi phục thì báo kèm thời gian đã sập. Trạng thái giữ ở TEP_TRANG_THAI. Chỉ dùng thư viện chuẩn (chạy được trên cả hai máy, không cần venv). Cấu hình: /etc/tradingauto-giamsat.env (xem giam_sat.env.mau) — token Telegram KHÔNG nằm trong mã. Chạy mỗi phút bằng systemd timer (giam-sat.timer). Thử không gửi Telegram: python3 giam_sat.py --thu Tự kiểm: python3 giam_sat.py --tu-kiem """ from __future__ import annotations import json import os import re import shutil import socket import ssl import subprocess import sys import time import urllib.error import urllib.parse import urllib.request from typing import Any, Callable, Dict, List, Optional, Tuple TEP_TRANG_THAI = os.environ.get('GIAMSAT_TRANG_THAI', '/var/lib/giam-sat/trang_thai.json') LIEN_TIEP = int(os.environ.get('GIAMSAT_LIEN_TIEP', '2') or 2) NHAC_PHUT = int(os.environ.get('GIAMSAT_NHAC_PHUT', '30') or 30) UA = 'tradingauto-giam-sat/1.0' # --------------------------------------------------------------------- # CÁC PHÉP KIỂM — mỗi hàm trả (ok, chi_tiet) # --------------------------------------------------------------------- def kiem_http(url: str, ma_hop_le: Tuple[int, ...] = (200,), chu_can: str = '', cho: float = 15) -> Tuple[bool, str]: rq = urllib.request.Request(url, headers={'User-Agent': UA}) t = time.time() try: with urllib.request.urlopen(rq, timeout=cho, context=ssl.create_default_context()) as r: ma, than = r.status, r.read(200_000).decode('utf-8', 'replace') except urllib.error.HTTPError as e: ma, than = e.code, '' except Exception as e: # noqa: BLE001 return False, f'không kết nối được: {str(e)[:120]}' ms = int((time.time() - t) * 1000) if ma not in ma_hop_le: return False, f'HTTP {ma} ({ms} ms)' if chu_can and chu_can not in than: return False, f'HTTP {ma} nhưng thiếu nội dung «{chu_can}» ({ms} ms)' return True, f'HTTP {ma}, {ms} ms' def kiem_tcp(may: str, cong: int, cho: float = 8) -> Tuple[bool, str]: t = time.time() try: with socket.create_connection((may, cong), timeout=cho): return True, f'mở, {int((time.time() - t) * 1000)} ms' except Exception as e: # noqa: BLE001 return False, f'không kết nối được cổng {cong}: {str(e)[:100]}' def kiem_dich_vu(ten: str) -> Tuple[bool, str]: try: tt = subprocess.run(['systemctl', 'is-active', ten], capture_output=True, text=True, timeout=10).stdout.strip() except Exception as e: # noqa: BLE001 return False, str(e)[:100] return tt == 'active', tt or 'không rõ' def kiem_dia(duong: str = '/', nguong_phan_tram: int = 90) -> Tuple[bool, str]: tong, dung, _ = shutil.disk_usage(duong) pt = round(dung * 100 / tong) return pt < nguong_phan_tram, f'đã dùng {pt}% ({dung // 2**30}/{tong // 2**30} GB)' def kiem_tep_moi(mau: str, toi_da_phut: int) -> Tuple[bool, str]: """Tệp mới nhất khớp mẫu glob phải mới hơn toi_da_phut phút (vd bản sao lưu Redis).""" import glob ds = glob.glob(mau) if not ds: return False, f'không có tệp nào khớp {mau}' moi = max(ds, key=os.path.getmtime) tuoi = int((time.time() - os.path.getmtime(moi)) / 60) return tuoi <= toi_da_phut, f'{os.path.basename(moi)}, cách đây {tuoi} phút (tối đa {toi_da_phut})' def kiem_sql(csdl: str, cau: str, toi_da: int) -> Tuple[bool, str]: """Số (cột 1, dòng 1) của một câu SELECT phải <= toi_da. Cột 2 (nếu có) là chi tiết kèm theo. Chạy bằng client mariadb qua unix socket (dịch vụ chạy dưới root) — không cần mật khẩu trong cấu hình.""" r = subprocess.run(['mariadb', '-N', '-B', csdl, '-e', cau], capture_output=True, text=True, timeout=30) if r.returncode != 0: return False, f'truy vấn lỗi: {(r.stderr or r.stdout).strip()[:150]}' cot = (r.stdout.splitlines() or ['0'])[0].split('\t') n = int(float(cot[0] or 0)) them = f' — {cot[1][:200]}' if len(cot) > 1 and cot[1] not in ('', 'NULL') else '' return n <= toi_da, f'{n} (tối đa {toi_da}){them}' def kiem_lenh(lenh: str) -> Tuple[bool, str]: """Lệnh trả mã 0 là ổn; khác 0 là hỏng, kèm phần cuối đầu ra (vd php cron/ledger_check.php).""" r = subprocess.run(lenh, shell=True, capture_output=True, text=True, timeout=300) # Dòng ĐẦU: script tóm tắt ở đầu rồi mới liệt kê (ledger_check: "CANH BAO SO CAI: n/m..." + từng tài khoản). ra = ' / '.join(x.strip() for x in (r.stdout + r.stderr).strip().splitlines()[:6] if x.strip()) return r.returncode == 0, (ra[:400] or f'mã thoát {r.returncode}') _MAU_LOG = re.compile(r'^(\S+) \S+ \S+ \[(\d{2}/\w{3}/\d{4}):(\d{2}):(\d{2}):(\d{2})') def dem_truy_cap(tep_log: str, phut: int = 5, bo_ip: Tuple[str, ...] = ('127.0.0.1',), _bay_gio: Optional[float] = None, _doc: Optional[Callable[[str], List[str]]] = None) -> Tuple[int, List[Tuple[str, int]]]: """Số yêu cầu trong `phut` phút gần nhất (bỏ IP nội bộ) + 5 IP gọi nhiều nhất.""" bay_gio = _bay_gio or time.time() moc = time.strftime('%d/%b/%Y:%H:%M', time.localtime(bay_gio - phut * 60)) if _doc is None: def _doc(p: str) -> List[str]: with open(p, 'rb') as f: f.seek(0, 2) f.seek(max(0, f.tell() - 30 * 1024 * 1024)) # chỉ đọc ~30 MB cuối return f.read().decode('utf-8', 'replace').splitlines() dem, ip_dem = 0, {} for dong in _doc(tep_log): m = _MAU_LOG.match(dong) if not m or m.group(1) in bo_ip: continue khoa = time.strftime('%d/%b/%Y:%H:%M', time.strptime(f'{m.group(2)}:{m.group(3)}:{m.group(4)}', '%d/%b/%Y:%H:%M')) if time.mktime(time.strptime(khoa, '%d/%b/%Y:%H:%M')) < time.mktime(time.strptime(moc, '%d/%b/%Y:%H:%M')): continue dem += 1 ip_dem[m.group(1)] = ip_dem.get(m.group(1), 0) + 1 return dem, sorted(ip_dem.items(), key=lambda x: -x[1])[:5] def kiem_truy_cap(tep_log: str, nguong_5_phut: int, bo_ip: Tuple[str, ...]) -> Tuple[bool, str]: try: dem, top = dem_truy_cap(tep_log, 5, bo_ip) except OSError as e: return True, f'không đọc được log: {e}' # không đọc được log thì không báo động giả ip = ', '.join(f'{a} ({b})' for a, b in top) return dem < nguong_5_phut, f'{dem} yêu cầu / 5 phút (ngưỡng {nguong_5_phut}); nhiều nhất: {ip or "—"}' # --------------------------------------------------------------------- # CẤU HÌNH # --------------------------------------------------------------------- def doc_cau_hinh(tep: str = '/etc/tradingauto-giamsat.env') -> Dict[str, str]: ch: Dict[str, str] = {} try: with open(tep, encoding='utf-8') as f: for dong in f: dong = dong.strip() if dong and not dong.startswith('#') and '=' in dong: k, v = dong.split('=', 1) ch[k.strip()] = v.strip().strip('"').strip("'") except OSError: pass ch.update({k: v for k, v in os.environ.items() if k.startswith('GIAMSAT_')}) return ch def dung_danh_sach(ch: Dict[str, str]) -> List[Tuple[str, Callable[[], Tuple[bool, str]]]]: """GIAMSAT_HTTP=ten|url|ma1,ma2|chu_can ; GIAMSAT_TCP=ten|may|cong ; GIAMSAT_DICH_VU=a,b,c GIAMSAT_DIA=/:90 ; GIAMSAT_LOG=tep|nguong_5_phut|ip_bo_qua1,ip2 GIAMSAT_TEP_MOI=ten|mau_glob|toi_da_phut (nhiều mục: cách nhau bởi ';') GIAMSAT_SQL=ten|csdl|cau_select|toi_da (câu SQL không được chứa ';' hay '|') GIAMSAT_LENH=ten|lenh|chu_ky_phut (chạy khi phút hiện tại chia hết chu kỳ; hỏng là báo ngay)""" ds: List[Tuple[str, Callable[[], Tuple[bool, str]]]] = [] for muc in filter(None, ch.get('GIAMSAT_HTTP', '').split(';')): p = [x.strip() for x in muc.split('|')] + ['', '', ''] ma = tuple(int(x) for x in (p[2] or '200').split(',')) ds.append((p[0], (lambda u=p[1], m=ma, c=p[3]: kiem_http(u, m, c)))) for muc in filter(None, ch.get('GIAMSAT_TCP', '').split(';')): p = [x.strip() for x in muc.split('|')] ds.append((p[0], (lambda h=p[1], c=int(p[2]): kiem_tcp(h, c)))) for ten in filter(None, [x.strip() for x in ch.get('GIAMSAT_DICH_VU', '').split(',')]): ds.append((f'dịch vụ {ten}', (lambda t=ten: kiem_dich_vu(t)))) for muc in filter(None, ch.get('GIAMSAT_DIA', '').split(';')): d, _, n = muc.partition(':') ds.append((f'đĩa {d}', (lambda d=d, n=int(n or 90): kiem_dia(d, n)))) for muc in filter(None, ch.get('GIAMSAT_TEP_MOI', '').split(';')): p = [x.strip() for x in muc.split('|')] ds.append((p[0], (lambda m=p[1], n=int(p[2]): kiem_tep_moi(m, n)))) for muc in filter(None, ch.get('GIAMSAT_LOG', '').split(';')): p = [x.strip() for x in muc.split('|')] + ['', ''] bo = tuple(filter(None, p[2].split(','))) or ('127.0.0.1',) ds.append(('lượng truy cập web', (lambda t=p[0], n=int(p[1] or 3000), b=bo: kiem_truy_cap(t, n, b)))) for muc in filter(None, ch.get('GIAMSAT_SQL', '').split(';')): p = [x.strip() for x in muc.split('|')] ds.append((p[0], (lambda c=p[1], q=p[2], n=int(p[3] or 0): kiem_sql(c, q, n)))) return ds def dung_lenh(ch: Dict[str, str], phut: int) -> List[Tuple[str, Callable[[], Tuple[bool, str]]]]: """Việc định kỳ (GIAMSAT_LENH) ĐẾN LƯỢT ở phút này. Lượt không chạy thì không đụng trạng thái cũ.""" ds: List[Tuple[str, Callable[[], Tuple[bool, str]]]] = [] for muc in filter(None, ch.get('GIAMSAT_LENH', '').split(';')): p = [x.strip() for x in muc.split('|')] + [''] if phut % max(1, int(p[2] or 1)) == 0: ds.append((p[0], (lambda l=p[1]: kiem_lenh(l)))) return ds # --------------------------------------------------------------------- # TELEGRAM + TRẠNG THÁI # --------------------------------------------------------------------- def gui_telegram(token: str, chat: str, van_ban: str) -> bool: if not token or not chat: return False du = urllib.parse.urlencode({'chat_id': chat, 'text': van_ban[:3900], 'disable_web_page_preview': 'true'}).encode() try: with urllib.request.urlopen(f'https://api.telegram.org/bot{token}/sendMessage', du, timeout=15) as r: return r.status == 200 except Exception: # noqa: BLE001 - KHÔNG in lỗi kèm URL (URL chứa token) return False def _thoi_luong(giay: float) -> str: giay = int(giay) return f'{giay // 3600} giờ {giay % 3600 // 60} phút' if giay >= 3600 else f'{giay // 60} phút {giay % 60} giây' def xu_ly(ket_qua: Dict[str, Tuple[bool, str]], tt: Dict[str, Any], may: str, bay_gio: float, bao_ngay: frozenset = frozenset()) -> List[str]: """Cập nhật trạng thái; trả các tin cần gửi. Thuần tuý — dễ tự kiểm. `bao_ngay`: phép kiểm chạy thưa (GIAMSAT_LENH) — hỏng một lần là báo, không chờ LIEN_TIEP lượt.""" tin: List[str] = [] for ten, (ok, ct) in ket_qua.items(): can = 1 if ten in bao_ngay else LIEN_TIEP s = tt.setdefault(ten, {'hong': False, 'loi_lien': 0, 'tu': None, 'bao_luc': 0}) if ok: if s['hong']: tin.append(f'✅ [{may}] HỒI PHỤC: {ten}\nSập {_thoi_luong(bay_gio - (s["tu"] or bay_gio))}. Hiện: {ct}') s.update(hong=False, loi_lien=0, tu=None, bao_luc=0) continue s['loi_lien'] += 1 if s['tu'] is None: s['tu'] = bay_gio if not s['hong'] and s['loi_lien'] >= can: s['hong'] = True s['bao_luc'] = bay_gio tin.append(f'🔴 [{may}] HỎNG: {ten}\n{ct}') elif s['hong'] and bay_gio - s['bao_luc'] >= NHAC_PHUT * 60: s['bao_luc'] = bay_gio tin.append(f'⏰ [{may}] VẪN HỎNG ({_thoi_luong(bay_gio - s["tu"])}): {ten}\n{ct}') return tin def chay(thu: bool = False) -> int: ch = doc_cau_hinh() may = ch.get('GIAMSAT_TEN_MAY') or socket.gethostname() ket_qua: Dict[str, Tuple[bool, str]] = {} ds_lenh = dung_lenh(ch, 0 if thu else int(time.time() // 60)) # --thu: chạy mọi lệnh bao_ngay = frozenset(ten for ten, _ in ds_lenh) for ten, ham in dung_danh_sach(ch) + ds_lenh: try: ket_qua[ten] = ham() except Exception as e: # noqa: BLE001 - một phép kiểm hỏng không làm hỏng các phép khác ket_qua[ten] = (False, f'lỗi phép kiểm: {str(e)[:100]}') if thu: for ten, (ok, ct) in ket_qua.items(): print(('OK ' if ok else 'LOI ') + f'{ten}: {ct}') return 0 try: with open(TEP_TRANG_THAI, encoding='utf-8') as f: tt = json.load(f) except (OSError, ValueError): tt = {} for tin in xu_ly(ket_qua, tt, may, time.time(), bao_ngay): if not gui_telegram(ch.get('GIAMSAT_TELEGRAM_TOKEN', ''), ch.get('GIAMSAT_TELEGRAM_CHAT', ''), tin): print('Không gửi được Telegram:', tin.splitlines()[0], file=sys.stderr) os.makedirs(os.path.dirname(TEP_TRANG_THAI), exist_ok=True) tam = TEP_TRANG_THAI + '.tam' with open(tam, 'w', encoding='utf-8') as f: json.dump(tt, f, ensure_ascii=False) os.replace(tam, TEP_TRANG_THAI) return 0 def _tu_kiem() -> None: print('=== xu_ly: chống spam, nhắc lại, hồi phục ===') tt: Dict[str, Any] = {} assert xu_ly({'web': (False, 'HTTP 522')}, tt, 'm', 1000) == [], 'lỗi lần đầu chưa báo' t = xu_ly({'web': (False, 'HTTP 522')}, tt, 'm', 1060) assert len(t) == 1 and 'HỎNG' in t[0] and '522' in t[0] assert xu_ly({'web': (False, 'x')}, tt, 'm', 1120) == [], 'không spam mỗi phút' t = xu_ly({'web': (False, 'x')}, tt, 'm', 1060 + NHAC_PHUT * 60) assert len(t) == 1 and 'VẪN HỎNG' in t[0] t = xu_ly({'web': (True, 'HTTP 200')}, tt, 'm', 1000 + 3600) assert len(t) == 1 and 'HỒI PHỤC' in t[0] and '1 giờ 0 phút' in t[0], t assert xu_ly({'web': (False, 'x')}, tt, 'm', 5000) == [] and xu_ly({'web': (True, 'ok')}, tt, 'm', 5060) == [], 'lỗi 1 lần rồi tự hết: im lặng' print(' OK') print('=== dem_truy_cap: đếm 5 phút gần nhất, bỏ IP nội bộ ===') now = time.mktime(time.strptime('26/Sep/2026:10:10:30', '%d/%b/%Y:%H:%M:%S')) log = ['1.1.1.1 - - [26/Sep/2026:10:09:00 +0700] "GET / HTTP/1.1" 200 1', '1.1.1.1 - - [26/Sep/2026:10:08:00 +0700] "GET / HTTP/1.1" 200 1', '2.2.2.2 - - [26/Sep/2026:10:06:00 +0700] "GET / HTTP/1.1" 200 1', '192.168.8.203 - - [26/Sep/2026:10:09:00 +0700] "GET /api/keys HTTP/1.1" 200 1', '3.3.3.3 - - [26/Sep/2026:09:50:00 +0700] "GET / HTTP/1.1" 200 1', 'rác'] dem, top = dem_truy_cap('x', 5, ('192.168.8.203',), _bay_gio=now, _doc=lambda p: log) assert dem == 3 and top[0] == ('1.1.1.1', 2), (dem, top) print(' OK') print('=== dung_danh_sach ===') ds = dung_danh_sach({'GIAMSAT_HTTP': 'web|https://x|200,301|TradingAuto', 'GIAMSAT_TCP': 'redis|h|51200', 'GIAMSAT_DICH_VU': 'nginx, chatbot', 'GIAMSAT_DIA': '/:90', 'GIAMSAT_LOG': '/l|3000|192.168.8.203'}) assert [t for t, _ in ds] == ['web', 'redis', 'dịch vụ nginx', 'dịch vụ chatbot', 'đĩa /', 'lượng truy cập web'] print(' OK') print('=== GIAMSAT_LENH: chạy đúng chu kỳ, hỏng báo ngay ===') ch = {'GIAMSAT_LENH': 'so cai|exit 1|60;luon|true|1'} assert [t for t, _ in dung_lenh(ch, 120)] == ['so cai', 'luon'] and [t for t, _ in dung_lenh(ch, 121)] == ['luon'] ok, ct = kiem_lenh('echo lech 2 tai khoan; exit 1') assert not ok and 'lech 2' in ct and kiem_lenh('true')[0] tt = {} t = xu_ly({'so cai': (False, 'lech')}, tt, 'm', 1000, frozenset({'so cai'})) assert len(t) == 1 and 'HỎNG' in t[0], 'việc định kỳ hỏng một lần là báo' print(' OK') print('TỰ KIỂM giam_sat.py QUA.') if __name__ == '__main__': if '--tu-kiem' in sys.argv: _tu_kiem() sys.exit(0) sys.exit(chay(thu='--thu' in sys.argv))