#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ vera_4_.py Фильтры 94/95/96 для сценария col_93 = 0. Изменение: выбор файлов для фильтров теперь зависит от usd_market_cap: - вместо white93.json / black93.json [0..10000] -> white010.json / black010.json [10001..15000] -> white1015.json / black1015.json [15001..20000] -> white1520.json / black1520.json [20001..25000] -> white2025.json / black2025.json [25001..30000] -> white2530.json / black2530.json [30001..80000] -> white3080.json / black3080.json - вместо whiteDiapazon93.json / blackDiapazon93.json [0..10000] -> whiteDiapazon010.json / blackDiapazon010.json [10001..15000] -> whiteDiapazon1015.json / blackDiapazon1015.json [15001..20000] -> whiteDiapazon1520.json / blackDiapazon1520.json [20001..25000] -> whiteDiapazon2025.json / blackDiapazon2025.json [25001..30000] -> whiteDiapazon2530.json / blackDiapazon2530.json [30001..80000] -> whiteDiapazon3080.json / blackDiapazon3080.json Примечание: - usd_market_cap появляется вначале и дальше не меняется. - Удаление токенов из этого файла не делаем (TTL чистит другой скрипт). """ import os import time import json import logging import re from typing import Set, List, Dict, Any, Optional, Tuple import pymysql import requests # ===== Настройки БД ===== DB_HOST = os.getenv("DB_HOST", "103.102.228.54") DB_USER = os.getenv("DB_USER", "phpmyadmin") DB_PASSWORD = os.getenv("DB_PASSWORD", "OE2hxy8iNUvB") DB_NAME = os.getenv("DB_NAME", "pumpfun") # ===== Пути для «старых» белых адресов (81) — как было ===== WHITELIST_PATH = os.getenv("WHITELIST_PATH", "/var/www/html/vera/white.json") # ===== Базовые пути (используем директории, а имена файлов подставляем динамически) ===== # Эти 4 переменные нужны только чтобы понять, в каких каталогах лежат файлы. BLACK93_PATH = os.getenv("BLACK93_PATH", "/var/www/html/vera/black93.json") BLACK93_DIAP_PATH = os.getenv("BLACK93_DIAP_PATH", "/var/www/html/vera/blackDiapazon93.json") WHITE93_PATH = os.getenv("WHITE93_PATH", "/var/www/html/vera/white93.json") WHITE93_DIAP_PATH = os.getenv("WHITE93_DIAP_PATH", "/var/www/html/vera/whiteDiapazon93.json") BASE_BLACK_DIR = os.path.dirname(BLACK93_PATH) or "/var/www/html/vera" BASE_WHITE_DIR = os.path.dirname(WHITE93_PATH) or "/var/www/html/vera" BASE_BD_DIR = os.path.dirname(BLACK93_DIAP_PATH) or "/var/www/html/vera" BASE_WD_DIR = os.path.dirname(WHITE93_DIAP_PATH) or "/var/www/html/vera" # ===== Telegram ===== TG_BOT_TOKEN = os.getenv("TG_BOT_TOKEN", "8442183787:AAF8o1PmZ72CQB9chVji91_roE18aVA9nlo") # Старая группа для col_81==1 (как было) TG_CHAT_ID_COL81 = int(os.getenv("TG_CHAT_ID", "-1002615752660")) # Новые группы (5 рассылок) TG_94_OK_CHAT_ID = int(os.getenv("TG_94_OK_CHAT_ID", "-1002615752660")) # 94 == 1 TG_94_FAIL_CHAT_ID = int(os.getenv("TG_94_FAIL_CHAT_ID", "-1002615752660")) # 94 == 0 TG_95_OK_CHAT_ID = int(os.getenv("TG_95_OK_CHAT_ID", "-1002615752660")) # 95 == 1 TG_96_OK_CHAT_ID = int(os.getenv("TG_96_OK_CHAT_ID", "-1002615752660")) # 96 == 1 TG_949596_ALL_ONES_CHAT = int(os.getenv("TG_949596_ALL_ONES_CHAT", "-1002615752660")) # 94=1 & 95=1 & 96=1 # Прочее POLL_SECONDS = int(os.getenv("POLL_SECONDS", "5")) LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper() logging.basicConfig(level=getattr(logging, LOG_LEVEL, logging.INFO), format="[%(asctime)s] %(levelname)s - %(message)s") # Колонки с адресами топ-10 WALLET_COLS = ["col_16","col_21","col_26","col_31","col_36","col_41","col_46","col_51","col_56","col_61"] # --- dip-файлы для col_81 (ветка 93=1): whiteDip*.json --- def _dip_suffix_for_81(usd_mc: Optional[float]) -> str: try: v = int(float(usd_mc)) except Exception: v = 0 if 0 <= v <= 30000: return "030" if 30001 <= v <= 60000:return "3060" if 60001 <= v <= 90000:return "6090" return "901000000" def _white_dip_path_for_81(usd_mc: Optional[float]) -> str: return os.path.join(BASE_WHITE_DIR, f"whiteDip{_dip_suffix_for_81(usd_mc)}.json") _WL_DIP_CACHE: Dict[str, 'Whitelist'] = {} def _get_wl_for_81(usd_mc: Optional[float]) -> 'Whitelist': p = _white_dip_path_for_81(usd_mc) inst = _WL_DIP_CACHE.get(p) if inst is None: inst = Whitelist(p); _WL_DIP_CACHE[p] = inst inst.load_if_changed() return inst # ===== Вспомогательные ===== COLNAME_RE = re.compile(r"^[A-Za-z0-9_]+$") def get_connection(): return pymysql.connect( host=DB_HOST, user=DB_USER, password=DB_PASSWORD, database=DB_NAME, charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, autocommit=True ) TG_SENT_CACHE = os.getenv("TG_SENT_CACHE", "tg_sent_cache.txt") _SENT_KEYS: Optional[Set[str]] = None def _load_sent_keys() -> Set[str]: global _SENT_KEYS if _SENT_KEYS is None: _SENT_KEYS = set() try: with open(TG_SENT_CACHE, "r", encoding="utf-8") as f: for line in f: _SENT_KEYS.add(line.strip()) except FileNotFoundError: pass return _SENT_KEYS def _mark_sent(key: str) -> None: s = _load_sent_keys() if key in s: return s.add(key) try: os.makedirs(os.path.dirname(TG_SENT_CACHE) or ".", exist_ok=True) with open(TG_SENT_CACHE, "a", encoding="utf-8") as f: f.write(key + "\n") except Exception: pass def send_tg_message(chat_id: int, text: str) -> None: """Антидубль: один и тот же text в один и тот же chat_id отправляется единожды.""" if not TG_BOT_TOKEN or not chat_id: return key = f"{chat_id}:{text}" if key in _load_sent_keys(): return try: url = f"https://api.telegram.org/bot{TG_BOT_TOKEN}/sendMessage" r = requests.post(url, timeout=15, json={ "chat_id": chat_id, "text": text, "disable_web_page_preview": True, }) if r.status_code != 200: logging.warning("Telegram non-200: %s %s", r.status_code, r.text[:200]) return _mark_sent(key) except Exception as e: logging.warning("Ошибка отправки в Telegram (%s): %s", text, e) # ---- whitelist loader (старый для col_81) ---- class Whitelist: def __init__(self, path: str): self.path = path self.mtime = 0.0 self.addrs: Set[str] = set() def load_if_changed(self): try: mtime = os.path.getmtime(self.path) if mtime != self.mtime: with open(self.path, "r", encoding="utf-8") as f: data = json.load(f) if isinstance(data, dict) and "wallets" in data: arr = data["wallets"] else: arr = data self.addrs = {str(x).strip() for x in arr if str(x).strip()} self.mtime = mtime logging.info("Whitelist перезагружен (%s): %d адресов.", self.path, len(self.addrs)) except FileNotFoundError: logging.warning("Не найден whitelist: %s", self.path) self.addrs = set() self.mtime = 0.0 except Exception as e: logging.error("Ошибка чтения whitelist %s: %s", self.path, e) # ---- Блеклист адресов ---- class Blacklist93: def __init__(self, path: str): self.path = path self.mtime = 0.0 self.addrs: Set[str] = set() def load_if_changed(self): try: mtime = os.path.getmtime(self.path) if mtime != self.mtime: with open(self.path, "r", encoding="utf-8") as f: data = json.load(f) if isinstance(data, dict) and "wallets" in data: arr = data["wallets"] else: arr = data self.addrs = {str(x).strip() for x in arr if str(x).strip()} self.mtime = mtime logging.info("Blacklist перезагружен (%s): %d адресов.", self.path, len(self.addrs)) except FileNotFoundError: self.addrs = set(); self.mtime = 0.0 except Exception as e: logging.error("Ошибка чтения blacklist %s: %s", self.path, e) # ---- Диапазоны (универсальный класс) ---- class Diapazon: def __init__(self, path: str): self.path = path self.mtime = 0.0 self.ranges: Dict[str, List[Tuple[float, float]]] = {} def load_if_changed(self): try: mtime = os.path.getmtime(self.path) if mtime != self.mtime: with open(self.path, "r", encoding="utf-8") as f: data = json.load(f) or {} rngs = (data or {}).get("ranges") or {} parsed: Dict[str, List[Tuple[float, float]]] = {} for col, arr in rngs.items(): if not isinstance(col, str) or not COLNAME_RE.match(col): # защита continue pairs: List[Tuple[float, float]] = [] for it in (arr or []): if not isinstance(it, (list, tuple)) or len(it) != 2: continue try: lo = float(it[0]); hi = float(it[1]) if lo > hi: lo, hi = hi, lo pairs.append((lo, hi)) except Exception: continue if pairs: parsed[col] = pairs self.ranges = parsed self.mtime = mtime logging.info("Diapazon перезагружен (%s): %d столбцов.", self.path, len(self.ranges)) except FileNotFoundError: self.ranges = {}; self.mtime = 0.0 except Exception as e: logging.error("Ошибка чтения %s: %s", self.path, e) def columns(self) -> List[str]: return list(self.ranges.keys()) def value_hits(self, col: str, value: Optional[float]) -> bool: if value is None: return False try: v = float(value) except Exception: return False for lo, hi in self.ranges.get(col, []): if lo <= v <= hi: return True return False # ============== Хелперы по БД ============== def fetch_candidates_col81(conn) -> List[Dict[str, Any]]: """ Кандидаты для col_81: только ветка 93=1, только где 78=1, 81 ещё NULL, и токены с достаточным MC (>=8000). """ cols = ", ".join(["mint", "col_81", "usd_market_cap"] + WALLET_COLS) sql = f""" SELECT {cols} FROM tokens WHERE col_93=1 AND col_78=1 AND col_81 IS NULL AND COALESCE(usd_market_cap,0) >= 8000 ORDER BY mint ASC LIMIT 250 """ with conn.cursor() as cur: cur.execute(sql) return cur.fetchall() def update_col81(conn, mint: str, value: int) -> int: """ Пишем col_81 только если было NULL, только в ветке 93=1, только при достаточном MC и при col_78=1. """ sql = """ UPDATE tokens SET col_81=%s WHERE LOWER(mint)=LOWER(%s) AND col_93=1 AND col_78=1 AND col_81 IS NULL AND COALESCE(usd_market_cap,0) >= 8000 """ with conn.cursor() as cur: cur.execute(sql, (value, mint)) return cur.rowcount def process_col81_batch(conn) -> int: """ Считает col_81 для кандидатов: - только ветка 93=1 - только где 78=1 - только где 81 IS NULL - только где usd_market_cap >= 8000 Логика: 81 = 1, если пересечение топ-10 адресов и whiteDip-списка не пусто; иначе 0. """ rows = fetch_candidates_col81(conn) if not rows: return 0 updated = 0 for r in rows: mint = r["mint"] umc = r.get("usd_market_cap") top10 = fetch_top10_from_row(r) if not top10: # ждём, пока подтянутся адреса – вернёмся к этому mint позже continue wl = _get_wl_for_81(umc).addrs # whiteDip*.json под диапазон MC val = 1 if (top10 and wl and (top10 & wl)) else 0 changed = update_col81(conn, mint, val) if changed: updated += changed if val == 1: send_tg_message(TG_CHAT_ID_COL81, mint) return updated def repair_invariants(conn) -> None: """ Страхующий ремонт: - если 78=0 в ветке 93=1, жёстко ставим 81=NULL и 82=NULL """ with conn.cursor() as cur: cur.execute(""" UPDATE tokens SET col_81=NULL, col_82=NULL WHERE col_93=1 AND col_78=0 AND (col_81 IS NOT NULL OR col_82 IS NOT NULL) """) def fetch_top10_from_row(row: Dict[str, Any]) -> Set[str]: return { (row.get(c) or "").strip() for c in WALLET_COLS if (row.get(c) or "").strip() } # --- Для 94 --- def fetch_rows_for_94(conn) -> List[Dict[str, Any]]: """Кандидаты для col_94: ветка 93=0, 94 ещё NULL, достаточный MC.""" cols = ", ".join(["mint","col_93","col_94","usd_market_cap"] + WALLET_COLS) sql = f""" SELECT {cols} FROM tokens WHERE col_93=0 AND col_94 IS NULL AND COALESCE(usd_market_cap,0) >= 8000 ORDER BY mint ASC LIMIT 250 """ with conn.cursor() as cur: cur.execute(sql); return cur.fetchall() def update_col94_atomic(conn, mint: str, val: int) -> bool: """94 считаем один раз в ветке 93=0, только при достаточном MC.""" sql = """ UPDATE tokens SET col_94=%s WHERE LOWER(mint)=LOWER(%s) AND col_93=0 AND col_94 IS NULL AND COALESCE(usd_market_cap,0) >= 8000 """ with conn.cursor() as cur: return cur.execute(sql, (val, mint)) > 0 def null_95_96_after_94_fail(conn, mint: str) -> None: """Фиксируем ветку 95/96 в NULL, если 94=0.""" with conn.cursor() as cur: cur.execute(""" UPDATE tokens SET col_95=NULL, col_96=NULL WHERE LOWER(mint)=LOWER(%s) AND col_93=0 """, (mint,)) # --- Для 95 --- def fetch_rows_for_95(conn) -> List[Dict[str, Any]]: cols = ", ".join(["mint","col_94","col_95"] + WALLET_COLS) sql = f""" SELECT {cols} FROM tokens WHERE col_93=0 AND col_94=1 AND col_95 IS NULL AND COALESCE(usd_market_cap,0) >= 8000 ORDER BY mint ASC LIMIT 250 """ with conn.cursor() as cur: cur.execute(sql); return cur.fetchall() def update_col95_atomic(conn, mint: str, val: int) -> bool: sql = """ UPDATE tokens SET col_95=%s WHERE LOWER(mint)=LOWER(%s) AND col_93=0 AND col_95 IS NULL AND COALESCE(usd_market_cap,0) >= 8000 """ with conn.cursor() as cur: return cur.execute(sql, (val, mint)) > 0 # --- Для 96 --- def fetch_rows_for_96_candidates(conn, limit: int=200) -> List[str]: sql = """ SELECT mint FROM tokens WHERE col_93=0 AND col_94=1 AND col_96 IS NULL AND COALESCE(usd_market_cap,0) >= 8000 ORDER BY mint ASC LIMIT %s """ with conn.cursor() as cur: cur.execute(sql, (limit,)) return [r["mint"] for r in cur.fetchall()] def update_col96_atomic(conn, mint: str, val: int) -> bool: sql = """ UPDATE tokens SET col_96=%s WHERE LOWER(mint)=LOWER(%s) AND col_93=0 AND col_96 IS NULL AND COALESCE(usd_market_cap,0) >= 8000 """ with conn.cursor() as cur: return cur.execute(sql, (val, mint)) > 0 def read_cols_94_95_96(conn, mint: str) -> Tuple[Optional[int], Optional[int], Optional[int]]: sql = "SELECT col_94, col_95, col_96 FROM tokens WHERE LOWER(mint)=LOWER(%s)" with conn.cursor() as cur: cur.execute(sql, (mint,)) r = cur.fetchone() or {} def to_int(x): try: return int(x) if x is not None else None except: return None return (to_int(r.get("col_94")), to_int(r.get("col_95")), to_int(r.get("col_96"))) def fetch_columns_for_mints(conn, mints: List[str], cols: List[str]) -> Dict[str, Dict[str, Any]]: if not mints or not cols: return {} safe_cols = [c for c in cols if COLNAME_RE.match(c)] if not safe_cols: return {} fields = ", ".join(["mint"] + safe_cols) placeholders = ",".join(["%s"] * len(mints)) sql = f"SELECT {fields} FROM tokens WHERE mint IN ({placeholders})" with conn.cursor() as cur: cur.execute(sql, mints) out: Dict[str, Dict[str, Any]] = {} for row in cur.fetchall(): m = row.pop("mint") out[m] = row return out def dogon_post_col94_ok(conn, sent_cache: Set[str]) -> None: """Досылка сообщений для всех токенов с col_93=0 и col_94=1.""" with conn.cursor() as cur: cur.execute(""" SELECT mint FROM tokens WHERE col_93=0 AND col_94=1 ORDER BY mint ASC LIMIT 200 """) for r in cur.fetchall(): m = r["mint"] if m not in sent_cache: send_tg_message(TG_94_OK_CHAT_ID, m) sent_cache.add(m) def dogon_post_col94_fail(conn, sent_cache: Set[str]) -> None: """Досылка для всех уже стоящих col_93=0 и col_94=0.""" with conn.cursor() as cur: cur.execute(""" SELECT mint FROM tokens WHERE col_93=0 AND col_94=0 ORDER BY mint ASC LIMIT 300 """) for r in cur.fetchall(): m = r["mint"] if m not in sent_cache: send_tg_message(TG_94_FAIL_CHAT_ID, m) sent_cache.add(m) def dogon_post_col81(conn, sent_cache: Set[str]) -> None: """Досылка для col_81==1 работает ТОЛЬКО в ветке col_93=1.""" with conn.cursor() as cur: cur.execute(""" SELECT mint FROM tokens WHERE col_93=1 AND col_78=1 AND col_81=1 ORDER BY mint ASC LIMIT 200 """) for r in cur.fetchall(): m = r["mint"] if m not in sent_cache: send_tg_message(TG_CHAT_ID_COL81, m) sent_cache.add(m) # ---- ЛОГИКА ---- def decide_col81(top10: Set[str], whitelist: Set[str]) -> bool: return bool(top10 and whitelist and (top10 & whitelist)) def decide_col94(top10: Set[str], bl_addrs: Set[str], diaps: Diapazon, row_vals: Dict[str, Any]) -> bool: """ Аналог col_78: если кошелёк в blacklist ИЛИ значение попадает в чёрный диапазон, то 94=0. Иначе 94=1. """ if top10 and (top10 & bl_addrs): return False for col in diaps.columns(): val = row_vals.get(col) if diaps.value_hits(col, val): return False return True # ---------- result.txt (оставлено без изменений логики удаления) ---------- RESULT_FILE = os.getenv("RESULT_FILE", "/var/www/html/vera/result.txt") MINT_LOG_FILE = os.getenv("MINT_LOG_FILE", "/var/www/html/vera/mint_log.txt") DEX_API_TOKEN = "https://api.dexscreener.com/latest/dex/tokens/{mint}" def append_result_lines(path: str, lines: List[str]) -> None: import hashlib, os os.makedirs(os.path.dirname(path), exist_ok=True) seen = set() try: with open(path, "r", encoding="utf-8") as f: for line in f: seen.add(hashlib.sha1(line.encode("utf-8")).hexdigest()) except FileNotFoundError: pass uniq = [] for ln in lines: h = hashlib.sha1(ln.encode("utf-8")).hexdigest() if h not in seen: uniq.append(ln); seen.add(h) if not uniq: return with open(path, "a", encoding="utf-8") as f: for ln in uniq: f.write(ln) def fetch_rows_for_result(conn, mints: List[str]) -> List[Dict[str, Any]]: if not mints: return [] placeholders = ",".join(["%s"] * len(mints)) cols = [ "mint","name","twitter","telegram","website","creator","bonding_curve", "created_timestamp","usd_market_cap", "market_cap","market_cap_bucket","volume_h24", "market_cap_bucket_max","axiom_total_count","axiom_migrated_count" ] + [f"col_{i}" for i in range(16, 108)] sql = f"SELECT {', '.join(cols)} FROM tokens WHERE mint IN ({placeholders})" with conn.cursor() as cur: cur.execute(sql, mints) return cur.fetchall() def serialize_row_for_result(row: Dict[str, Any]) -> str: fields = [ "mint","name","twitter","telegram","website","creator","bonding_curve", "created_timestamp","usd_market_cap", "market_cap","market_cap_bucket","volume_h24", "market_cap_bucket_max","axiom_total_count","axiom_migrated_count" ] + [f"col_{i}" for i in range(16, 108)] parts = [] for k in fields: parts.append(f"; {k} ; {row.get(k, '')}") return " ".join(parts) + "\n" # ===================== ДИНАМИЧЕСКИЙ ВЫБОР ФАЙЛОВ ПО usd_market_cap ===================== def _bucket_suffix(usd_mc: Optional[float]) -> str: """ Возвращает суффикс для имени файла по usd_market_cap. Если за пределами известных диапазонов (>80000), считаем как '3080'. """ try: v = int(float(usd_mc or 0.0)) except Exception: v = 0 if 0 <= v <= 10000: return "010" if 10001 <= v <= 15000: return "1015" if 15001 <= v <= 20000: return "1520" if 20001 <= v <= 25000: return "2025" if 25001 <= v <= 30000: return "2530" # 30001..80000 (и всё, что выше — считаем этим же набором) return "3080" def _white_path_for(usd_mc: Optional[float]) -> str: return os.path.join(BASE_WHITE_DIR, f"white{_bucket_suffix(usd_mc)}.json") def _black_path_for(usd_mc: Optional[float]) -> str: return os.path.join(BASE_BLACK_DIR, f"black{_bucket_suffix(usd_mc)}.json") def _white_diap_path_for(usd_mc: Optional[float]) -> str: return os.path.join(BASE_WD_DIR, f"whiteDiapazon{_bucket_suffix(usd_mc)}.json") def _black_diap_path_for(usd_mc: Optional[float]) -> str: return os.path.join(BASE_BD_DIR, f"blackDiapazon{_bucket_suffix(usd_mc)}.json") # Кэши инстансов по путям, чтобы не перечитывать файлы по сто раз. _WL_CACHE: Dict[str, Whitelist] = {} _BL_CACHE: Dict[str, Blacklist93] = {} _WD_CACHE: Dict[str, Diapazon] = {} _BD_CACHE: Dict[str, Diapazon] = {} def _get_wl(usd_mc: Optional[float]) -> Whitelist: p = _white_path_for(usd_mc) inst = _WL_CACHE.get(p) if inst is None: inst = Whitelist(p); _WL_CACHE[p] = inst inst.load_if_changed() return inst def _get_bl(usd_mc: Optional[float]) -> Blacklist93: p = _black_path_for(usd_mc) inst = _BL_CACHE.get(p) if inst is None: inst = Blacklist93(p); _BL_CACHE[p] = inst inst.load_if_changed() return inst def _get_wd(usd_mc: Optional[float]) -> Diapazon: p = _white_diap_path_for(usd_mc) inst = _WD_CACHE.get(p) if inst is None: inst = Diapazon(p); _WD_CACHE[p] = inst inst.load_if_changed() return inst def _get_bd(usd_mc: Optional[float]) -> Diapazon: p = _black_diap_path_for(usd_mc) inst = _BD_CACHE.get(p) if inst is None: inst = Diapazon(p); _BD_CACHE[p] = inst inst.load_if_changed() return inst # ===== Публикация 94=0 в index.html на 5 минут ===== FAIL_HTML_PATH = os.getenv("FAIL_HTML_PATH", "/var/www/html/vera/index.html") FAIL_HTML_STATE = os.getenv("FAIL_HTML_STATE", "/var/www/html/vera/index_state.json") FAIL_TTL_SECONDS = int(os.getenv("FAIL_TTL_SECONDS", "300")) # 5 минут по умолчанию def _now_ms() -> int: return int(time.time() * 1000) def _safe_mkdirs(path: str) -> None: d = os.path.dirname(path) or "." os.makedirs(d, exist_ok=True) def _load_state(path: str) -> Dict[str, int]: try: with open(path, "r", encoding="utf-8") as f: data = json.load(f) or {} # фильтруем мусор cleaned = {} for k, v in (data.items() if isinstance(data, dict) else []): try: ts = int(v) if k and ts > 0: cleaned[k] = ts except Exception: continue return cleaned except FileNotFoundError: return {} except Exception: return {} def _write_state_atomic(path: str, state: Dict[str, int]) -> None: _safe_mkdirs(path) tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(state, f, ensure_ascii=False, separators=(",", ":")) os.replace(tmp, path) def _render_html(path: str, mints: List[str]) -> None: _safe_mkdirs(path) # Как в примере index.html (заголовок, счётчик, список). Статические стили. html = [ "", '', "", ' ', ' ', " Отправленные токены (последние 5 минут)", " ", "", "", "

Отправленные в Telegram токены (живут 5 минут)

", f'
Сейчас в списке: {len(mints)}
', '
' ] if mints: for m in mints: html.append(f'
{m}
') else: html.append('
Пока пусто
') html.extend([ "
", "", "", "" ]) tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: f.write("\n".join(html)) os.replace(tmp, path) def _purge_expired_and_render(state: Dict[str, int]) -> Dict[str, int]: """Убираем протухшие записи и перерисовываем HTML при изменениях.""" now = _now_ms() ttl_ms = FAIL_TTL_SECONDS * 1000 alive = {m: ts for m, ts in state.items() if now - ts <= ttl_ms} if len(alive) != len(state): # были изменения — перерендер _render_html(FAIL_HTML_PATH, sorted(alive.keys())) _write_state_atomic(FAIL_HTML_STATE, alive) return alive return state def publish_94_fail_to_html(mint: str) -> None: """ Добавляет mint в HTML-страницу (с TTL), если его ещё нет. Вызывать всякий раз, когда выставили 94=0 (или при «догоне» 94=0). """ try: state = _load_state(FAIL_HTML_STATE) state = _purge_expired_and_render(state) if mint not in state: state[mint] = _now_ms() # сохраняем и перерисовываем _write_state_atomic(FAIL_HTML_STATE, state) _render_html(FAIL_HTML_PATH, sorted(state.keys())) except Exception as e: logging.warning("Не удалось опубликовать 94=0 в index.html для %s: %s", mint, e) # ================= ГЛАВНЫЙ ЦИКЛ ================= def loop(): conn = get_connection() logging.info("Подключено к MySQL: %s/%s", DB_HOST, DB_NAME) # Кэши для антидублей рассылок (как у вас принято) sent_94_ok: Set[str] = set() sent_94_fail: Set[str] = set() sent_95_ok: Set[str] = set() sent_96_ok: Set[str] = set() sent_all_949596: Set[str] = set() sent_81_ok: Set[str] = set() # Кэш «живых» публикаций в HTML поддерживается файлами state; здесь — только регулярная очистка _ = _purge_expired_and_render(_load_state(FAIL_HTML_STATE)) # Локальные кэш-объекты для файлов (чтобы не дёргать диск каждый раз) _BL_CACHE: Dict[str, Blacklist93] = {} _BD_CACHE: Dict[str, Diapazon] = {} _WL_CACHE: Dict[str, Whitelist] = {} _WD_CACHE: Dict[str, Diapazon] = {} def _suffix_umc(usd_mc: Optional[float]) -> str: try: v = int(float(usd_mc or 0)) except Exception: v = 0 if 0 <= v <= 10000: return "010" if 10001 <= v <= 15000: return "1015" if 15001 <= v <= 20000: return "1520" if 20001 <= v <= 25000: return "2025" if 25001 <= v <= 30000: return "2530" return "3080" def _get_bl(usd_mc: Optional[float]) -> Blacklist93: suf = _suffix_umc(usd_mc) path = os.path.join(BASE_BLACK_DIR, f"black{suf}.json") inst = _BL_CACHE.get(path) if inst is None: inst = Blacklist93(path); _BL_CACHE[path] = inst inst.load_if_changed() return inst def _get_bd(usd_mc: Optional[float]) -> Diapazon: suf = _suffix_umc(usd_mc) path = os.path.join(BASE_BD_DIR, f"blackDiapazon{suf}.json") inst = _BD_CACHE.get(path) if inst is None: inst = Diapazon(path); _BD_CACHE[path] = inst inst.load_if_changed() return inst def _get_wl(usd_mc: Optional[float]) -> Whitelist: # для 95: white010.json/… из BASE_WHITE_DIR suf = _suffix_umc(usd_mc) path = os.path.join(BASE_WHITE_DIR, f"white{suf}.json") inst = _WL_CACHE.get(path) if inst is None: inst = Whitelist(path); _WL_CACHE[path] = inst inst.load_if_changed() return inst def _get_wd(usd_mc: Optional[float]) -> Diapazon: # для 96: whiteDiapazon010.json/… из BASE_WD_DIR suf = _suffix_umc(usd_mc) path = os.path.join(BASE_WD_DIR, f"whiteDiapazon{suf}.json") inst = _WD_CACHE.get(path) if inst is None: inst = Diapazon(path); _WD_CACHE[path] = inst inst.load_if_changed() return inst while True: try: try: conn.ping(reconnect=True) except Exception: conn = get_connection() # регулярная чистка HTML по TTL (чтобы минты уходили через 5 минут) try: st = _load_state(FAIL_HTML_STATE) _ = _purge_expired_and_render(st) except Exception: pass # ===== col_81 при col_93=1 и col_78=1 (MC >= 8000) ===== try: repair_invariants(conn) # страховка, быстро upd81 = process_col81_batch(conn) # основной расчёт 81 if upd81: logging.info("Обновлено col_81: %d", upd81) except Exception as e: logging.error("Ошибка обработки col_81: %s", e) # ===== col_94 при col_93=0 ===== try: rows94 = fetch_rows_for_94(conn) # mint, col_93, col_94, usd_market_cap, +col_16..col_61 if rows94: # собрать, какие столбцы нужны для чёрных диапазонов (по mint — свой файл) need_mints = [r["mint"] for r in rows94] diaps_by_mint: Dict[str, Diapazon] = {} needed_cols: Set[str] = set() for r in rows94: umc = r.get("usd_market_cap") d = _get_bd(umc) diaps_by_mint[r["mint"]] = d needed_cols.update(d.columns()) vals_by_mint = fetch_columns_for_mints(conn, need_mints, sorted(needed_cols)) if needed_cols else {} for r in rows94: mint = r["mint"] if int(r.get("col_93") or 0) != 0: continue top10 = fetch_top10_from_row(r) if not top10: continue # ждём заполнения топ-10 umc = r.get("usd_market_cap") bl_addrs = _get_bl(umc).addrs # Всегда берём актуальный Diapazon под MC (blackDiapazon*.json) bd = diaps_by_mint.get(mint) or _get_bd(umc) cols_needed = bd.columns() # Текущее знание по значениям строки (что уже батчем подтянули выше) row_vals = vals_by_mint.get(mint, {}) if needed_cols else {} # 1) Догрузим недостающие колонки точечно для этого mint missing_cols = [c for c in cols_needed if c not in row_vals] if missing_cols: extra = fetch_columns_for_mints(conn, [mint], missing_cols).get(mint, {}) if row_vals: row_vals.update(extra) else: row_vals = extra vals_by_mint[mint] = row_vals # чтобы следующие проходы видели актуальные данные # 2) Если какие-то нужные колонки всё ещё отсутствуют или равны None — пропускаем этот mint still_missing = [c for c in cols_needed if (c not in row_vals or row_vals.get(c) is None)] if still_missing: continue # подождём следующего тика, когда колонки будут заполнены ok = decide_col94(top10, bl_addrs, bd, row_vals) val = 1 if ok else 0 changed = update_col94_atomic(conn, mint, val) if not changed: continue already_ok = mint in sent_94_ok already_fail = mint in sent_94_fail if val == 1: if not already_ok and not already_fail: send_tg_message(TG_94_OK_CHAT_ID, mint) sent_94_ok.add(mint) else: # 94 = 0 — публикуем в HTML c TTL + Telegram + фиксируем 95/96=NULL if not already_fail and not already_ok: send_tg_message(TG_94_FAIL_CHAT_ID, mint) sent_94_fail.add(mint) publish_94_fail_to_html(mint) # <<< ВСТАВКА null_95_96_after_94_fail(conn, mint) except Exception as e: logging.error("Ошибка обработки col_94: %s", e) # «догон» для всех уже стоящих 94=0 — и в TG, и в HTML try: with conn.cursor() as cur: cur.execute(""" SELECT mint FROM tokens WHERE col_93=0 AND col_94=0 ORDER BY mint ASC LIMIT 300 """) for rr in cur.fetchall(): m = rr["mint"] if m not in sent_94_fail: send_tg_message(TG_94_FAIL_CHAT_ID, m) sent_94_fail.add(m) publish_94_fail_to_html(m) # <<< ВСТАВКА except Exception as e: logging.error("Ошибка догона 94=0: %s", e) # ===== col_95 при col_93=0 и col_94=1 ===== try: rows95 = fetch_rows_for_95(conn) upd95 = 0 if rows95: mints95 = [r["mint"] for r in rows95] mc_by_mint = fetch_columns_for_mints(conn, mints95, ["usd_market_cap"]) for r in rows95: mint = r["mint"] top10 = fetch_top10_from_row(r) if not top10: continue umc = (mc_by_mint.get(mint) or {}).get("usd_market_cap") wl_set = _get_wl(umc).addrs ok = bool(top10 & wl_set) val = 1 if ok else 0 if update_col95_atomic(conn, mint, val): upd95 += 1 if val == 1 and mint not in sent_95_ok: send_tg_message(TG_95_OK_CHAT_ID, mint) sent_95_ok.add(mint) if upd95: logging.info("Обновлено col_95: %d", upd95) except Exception as e: logging.error("Ошибка обработки col_95: %s", e) # ===== col_96 при col_93=0 и col_94=1 ===== try: mints96 = fetch_rows_for_96_candidates(conn, 150) if mints96: mc_by_mint = fetch_columns_for_mints(conn, mints96, ["usd_market_cap"]) wd_by_mint: Dict[str, Diapazon] = {} needed_cols: Set[str] = set() for mint in mints96: umc = (mc_by_mint.get(mint) or {}).get("usd_market_cap") wd = _get_wd(umc) wd_by_mint[mint] = wd needed_cols.update(wd.columns()) rows_vals = fetch_columns_for_mints(conn, mints96, sorted(needed_cols)) if needed_cols else {} for mint in mints96: wd = wd_by_mint.get(mint) if not wd: continue vals = rows_vals.get(mint, {}) hits = 0 for c in wd.columns(): if wd.value_hits(c, vals.get(c)): hits += 1 val = 1 if hits >= 3 else 0 changed96 = update_col96_atomic(conn, mint, val) c94, c95, c96 = read_cols_94_95_96(conn, mint) if changed96 and val == 1 and mint not in sent_96_ok: send_tg_message(TG_96_OK_CHAT_ID, mint) sent_96_ok.add(mint) if c94 == 1 and c95 == 1 and c96 == 1 and mint not in sent_all_949596: send_tg_message(TG_949596_ALL_ONES_CHAT, mint) sent_all_949596.add(mint) # догоны 95/96/общей группы with conn.cursor() as cur: # 95 OK cur.execute(""" SELECT mint FROM tokens WHERE col_93=0 AND col_94=1 AND col_95=1 ORDER BY mint ASC LIMIT 200 """) for rr in cur.fetchall(): m = rr["mint"] if m not in sent_95_ok: send_tg_message(TG_95_OK_CHAT_ID, m) sent_95_ok.add(m) # 96 OK cur.execute(""" SELECT mint FROM tokens WHERE col_93=0 AND col_94=1 AND col_96=1 ORDER BY mint ASC LIMIT 200 """) for rr in cur.fetchall(): m = rr["mint"] if m not in sent_96_ok: send_tg_message(TG_96_OK_CHAT_ID, m) sent_96_ok.add(m) # 94=1 & 95=1 & 96=1 cur.execute(""" SELECT mint FROM tokens WHERE col_93=0 AND col_94=1 AND col_95=1 AND col_96=1 ORDER BY mint ASC LIMIT 200 """) for rr in cur.fetchall(): m = rr["mint"] if m not in sent_all_949596: send_tg_message(TG_949596_ALL_ONES_CHAT, m) sent_all_949596.add(m) except Exception as e: logging.error("Ошибка обработки col_96/догона: %s", e) # (опционально) догоним старую рассылку col_81==1 в ветке 93=1 (если используете) try: dogon_post_col81(conn, sent_81_ok) except Exception: pass time.sleep(POLL_SECONDS) except KeyboardInterrupt: logging.info("Остановка по Ctrl+C") break except Exception as e: logging.error("Критическая ошибка цикла: %s", e) time.sleep(5) if __name__ == "__main__": loop()