From b31215057e312a9ffecad3ba60455614a780aa81 Mon Sep 17 00:00:00 2001 From: jackyu66git Date: Wed, 5 Aug 2026 18:09:43 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E5=B0=86=20bsp=5Fmonitor=20=E6=8B=86?= =?UTF-8?q?=E5=87=BA=E4=B8=BA=E7=8B=AC=E7=AB=8B=E4=BB=93=E5=BA=93=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 监控服务已迁移至 jack/bsp_monitor，不再随 chan 维护。 Co-authored-by: Cursor --- bsp_monitor/.env | 3 - bsp_monitor/__init__.py | 1 - bsp_monitor/engine.py | 174 ----------------- bsp_monitor/fetcher.py | 62 ------ bsp_monitor/main.py | 220 --------------------- bsp_monitor/notify.py | 61 ------ bsp_monitor/run.sh | 11 -- bsp_monitor/twitter_web.py | 380 ------------------------------------- 8 files changed, 912 deletions(-) delete mode 100644 bsp_monitor/.env delete mode 100644 bsp_monitor/__init__.py delete mode 100644 bsp_monitor/engine.py delete mode 100644 bsp_monitor/fetcher.py delete mode 100644 bsp_monitor/main.py delete mode 100644 bsp_monitor/notify.py delete mode 100755 bsp_monitor/run.sh delete mode 100644 bsp_monitor/twitter_web.py diff --git a/bsp_monitor/.env b/bsp_monitor/.env deleted file mode 100644 index df69084..0000000 --- a/bsp_monitor/.env +++ /dev/null @@ -1,3 +0,0 @@ -# bsp_monitor 复用 Hermes Agent 的 Telegram bot -# notify.py 从 ~/.hermes/.env 直接读取 TELEGRAM_BOT_TOKEN -# 此处无需重复配置 diff --git a/bsp_monitor/__init__.py b/bsp_monitor/__init__.py deleted file mode 100644 index e9d7b81..0000000 --- a/bsp_monitor/__init__.py +++ /dev/null @@ -1 +0,0 @@ -# bsp_monitor - 缠论买卖点监控 (BTC/USDT 1m) diff --git a/bsp_monitor/engine.py b/bsp_monitor/engine.py deleted file mode 100644 index e9983dd..0000000 --- a/bsp_monitor/engine.py +++ /dev/null @@ -1,174 +0,0 @@ -""" -engine.py - 缠论管线封装：DataFrame → KLU → KLC → BI → SEG → ZS → BSP。 - -复用 ~/Project/Chan/ 下的 TF_DF 模块，管线步骤对齐 TF_DF.get_bsp_state()。 -""" -import sys -import os -from typing import List, Optional - -_PARENT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) -if _PARENT not in sys.path: - sys.path.insert(0, _PARENT) - -import pandas as pd -from ChanEnum import ( - Chan_BSP_DIR, Chan_BSP_TYPE, Chan_KLC_FX, Chan_BI_DIR, - Chan_ZS_DIR, -) -from ChanBSP import ChanBSP -from ChanBI import ChanBI - -# 仅导入类，不触发 TF_DF.__init__ -from TF_DF import TF_DF as _TF_DF_Class - - -class ChanEngine: - """缠论管线，对齐 TF_DF.get_bsp_state() 的调用顺序。""" - - def __init__(self, df: pd.DataFrame): - if df.empty or len(df) < 50: - raise ValueError("DataFrame 至少需要 50 根 K 线") - - if "date" not in df.columns and "timestamp" in df.columns: - df["date"] = df["timestamp"] - - self.df = df - self._tf = _TF_DF_Class.__new__(_TF_DF_Class) # 不调用 __init__ - - # Step 0: 添加 TA 指标 (MACD/EMA/BB/RSI) - self._df_with_indicators = self._tf.add_indicators(df.copy()) - - # Step 1: KLU — get_klu_list → get_kl_data → cal_kl_data - self.klu_list = self._tf.get_klu_list(self._df_with_indicators) - - # Step 2: KLC — 内部已含 ChanMACD.cal_macd_state() + cal_trend() - self.klc_list = self._tf.get_klc_list(self.klu_list) - - # Step 3: BI (stroke) - self.bi_list = self._tf.cal_bi_list(self.klc_list) - - # Step 4: SEG (segment) - self.seg_list = self._tf.get_seg_list(self.bi_list) - - # Step 5: ZS — cal_bi_zs(seg_list) 对齐 get_bsp_state（从线段计算笔中枢） - self.bi_zs_list: List = self._tf.cal_bi_zs(self.seg_list) - - # Step 6: BSP (buy/sell points) - self.bsp_list: List[ChanBSP] = self._tf.find_all_bsp( - self.bi_list, self.bi_zs_list - ) - - def get_second_last_bi(self) -> Optional[ChanBI]: - """获取倒数第二笔（最新确认的笔）。""" - confirmed = [b for b in self.bi_list if b.is_sure] - if len(confirmed) >= 2: - return confirmed[-2] - elif len(confirmed) == 1: - return confirmed[-1] - return None - - def get_bsp_for_bi(self, bi: ChanBI) -> Optional[ChanBSP]: - """检查某个 Bi 的 end_klc 是否是买卖点。""" - if bi is None or not bi.is_sure: - return None - klc = bi.end_klc - if klc is None: - return None - if klc.bsp and klc.bsp_type != Chan_BSP_TYPE.NONE: - for bsp in self.bsp_list: - if bsp.klc is klc: - return bsp - return None - - # ── 格式化 ── - - @staticmethod - def _bsp_type_name(t: Chan_BSP_TYPE) -> str: - import ChanEnum - names = { - Chan_BSP_TYPE.B1: "一类买点(B1)", - Chan_BSP_TYPE.B2: "二类买点(B2)", - Chan_BSP_TYPE.B3: "三类买点(B3)", - Chan_BSP_TYPE.S1: "一类卖点(S1)", - Chan_BSP_TYPE.S2: "二类卖点(S2)", - Chan_BSP_TYPE.S3: "三类卖点(S3)", - } - return names.get(t, str(t)) - - @staticmethod - def _bi_dir_name(d) -> str: - return "⬆️ 向上" if d == Chan_BI_DIR.UP else "⬇️ 向下" - - @staticmethod - def _fx_strength_name(klc_fx_type) -> str: - import ChanEnum - names = { - Chan_KLC_FX.TOP0: "TOP0(弱)", Chan_KLC_FX.TOP1: "TOP1(标准)", - Chan_KLC_FX.TOP2: "TOP2(强)", Chan_KLC_FX.TOP3: "TOP3(二类)", - Chan_KLC_FX.TOP4: "TOP4(BB上轨)", Chan_KLC_FX.TOP5: "TOP5", - Chan_KLC_FX.TOP6: "TOP6(高位空)", Chan_KLC_FX.TOP7: "TOP7(背驰)", - Chan_KLC_FX.TOP8: "TOP8(信号线)", - Chan_KLC_FX.BOTTOM0: "BOTTOM0(弱)", Chan_KLC_FX.BOTTOM1: "BOTTOM1(标准)", - Chan_KLC_FX.BOTTOM2: "BOTTOM2(强)", Chan_KLC_FX.BOTTOM3: "BOTTOM3(二类)", - Chan_KLC_FX.BOTTOM4: "BOTTOM4(BB下轨)", Chan_KLC_FX.BOTTOM5: "BOTTOM5(零轴下)", - Chan_KLC_FX.BOTTOM6: "BOTTOM6(高位空)", Chan_KLC_FX.BOTTOM7: "BOTTOM7(背驰)", - Chan_KLC_FX.BOTTOM8: "BOTTOM8(信号线)", - } - return names.get(klc_fx_type, f"UNKNOWN({klc_fx_type})") - - @staticmethod - def _utc_to_cst(time_str: str) -> str: - """UTC 时间字符串 → 东八区 (UTC+8)。""" - from datetime import datetime, timedelta, timezone - dt = datetime.fromisoformat(str(time_str)) - if dt.tzinfo is None: - dt = dt.replace(tzinfo=timezone.utc) - cst = dt.astimezone(timezone(timedelta(hours=8))) - return cst.strftime("%Y-%m-%d %H:%M:%S CST") - - def format_bsp_detail(self, bsp: ChanBSP, symbol: str = "BTC/USDT:USDT", tf: str = "1m") -> str: - bi = bsp.bi - klc = bsp.klc - bsp_type = bsp.type - bsp_dir = bsp.dir - - emoji = "🟢" if bsp_dir == Chan_BSP_DIR.BUY else "🔴" - dir_label = "买点" if bsp_dir == Chan_BSP_DIR.BUY else "卖点" - - symbol_short = symbol.split(":")[0].replace("/", "") - lines = [ - f"{emoji} [{dir_label}] {self._bsp_type_name(bsp_type)} — {symbol_short} {tf}", - "", - f"⏰ 确认: {self._utc_to_cst(klc.end_time)}", - f"💰 价格: {klc.close:.2f}", - f"📐 笔方向: {self._bi_dir_name(bi.dir)}", - f"📏 笔高度: ${bi.height:.2f} 宽度: {bi.width}K 斜率: {bi.slop:.2f}", - f"🔩 分型强度: {self._fx_strength_name(klc.klc_fx_type)}", - ] - - if bsp.zs: - zs = bsp.zs - zs_dir = "UP" if hasattr(zs, 'dir') and hasattr(Chan_ZS_DIR, 'UP') and zs.dir == Chan_ZS_DIR.UP else "DOWN" - lines.append(f"🏠 中枢: {zs.zd:.2f} – {zs.zg:.2f} ({zs_dir}, #{getattr(zs, 'index', 0) + 1})") - - if bsp.type in (Chan_BSP_TYPE.B1, Chan_BSP_TYPE.S1): - lines.append("📊 MACD背驰: 有 (离开段能量 < 进入段)") - - if hasattr(klc, 'ema_status') and klc.ema_status: - ema52 = klc.ema_status.get('ema52', {}) - if ema52: - pos = str(ema52.get('pos', '?')) - lines.append(f"📈 EMA52: {pos} (值: {klc.ema52:.2f})") - - lines.append(f"📋 KLC状态: {klc.klc_state}") - - if bi.pre: - prev = bi.pre - lines.extend([ - "────", - f"⬅️ 前一笔: {self._bi_dir_name(prev.dir)} " - f"高度: ${prev.height:.2f} 宽度: {prev.width}K", - ]) - - return "\n".join(lines) diff --git a/bsp_monitor/fetcher.py b/bsp_monitor/fetcher.py deleted file mode 100644 index ac4dc90..0000000 --- a/bsp_monitor/fetcher.py +++ /dev/null @@ -1,62 +0,0 @@ -""" -fetcher.py - 从 data_provider HTTP API 拉取 K 线数据。 -""" - -from typing import List, Optional - -import requests -import pandas as pd -import logging - -logger = logging.getLogger(__name__) - -PROVIDER_URL = "http://103.179.242.166" -PROVIDER_URL = "http://127.0.0.1" -FETCH_LIMIT = 1000 - -_symbols_cache: Optional[List[str]] = None - - -# 只推送 BTC，其他币对暂不监控 -_SYMBOL_WHITELIST = {"BTC/USDT:USDT", "ETH/USDT:USDT", "SOL/USDT:USDT"} - - -def get_symbols() -> list[str]: - """获取要监控的币对列表（目前只监控 BTC）。""" - global _symbols_cache - if _symbols_cache is not None: - return _symbols_cache - try: - resp = requests.get(f"{PROVIDER_URL}/health", timeout=10) - resp.raise_for_status() - all_symbols = resp.json().get("symbols", []) - _symbols_cache = [s for s in all_symbols if s in _SYMBOL_WHITELIST] - logger.info(f"获取到 {len(all_symbols)} 个币对，过滤后监控 {len(_symbols_cache)} 个: {_symbols_cache}") - except Exception as e: - logger.error(f"获取币对列表失败: {e}") - _symbols_cache = ["BTC/USDT:USDT"] - return _symbols_cache - - -def fetch_ohlcv(symbol: str, tf: str = "1m") -> pd.DataFrame: - """从 data_provider API 拉取某个币对最近 FETCH_LIMIT 根 K 线。""" - url = f"{PROVIDER_URL}/api/candles" - params = { - "symbol": symbol, - "tf": tf, - "limit": FETCH_LIMIT, - } - resp = requests.get(url, params=params, timeout=30) - resp.raise_for_status() - data = resp.json() - - if not data: - logger.warning(f"{symbol}: API 返回空数据") - return pd.DataFrame() - - df = pd.DataFrame(data) - df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True) - df["date"] = df["timestamp"] - - df = df.drop_duplicates(subset="timestamp").sort_values("timestamp").reset_index(drop=True) - return df diff --git a/bsp_monitor/main.py b/bsp_monitor/main.py deleted file mode 100644 index aa3a304..0000000 --- a/bsp_monitor/main.py +++ /dev/null @@ -1,220 +0,0 @@ -#!/usr/bin/env python3 -""" -main.py - 缠论多周期买卖点监控。 - -每整分钟： - 1. 从 data_provider 拉取所有币对多周期 K 线 - 2. 每个币对 × 每个周期独立跑缠论管线 - 3. 检测新笔确认 → BSP 推送 -""" -import asyncio -import logging -import sys -import os -import time -from dataclasses import dataclass, field -from datetime import datetime, timezone, timedelta -from typing import Optional - -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) - -from fetcher import fetch_ohlcv, get_symbols -from engine import ChanEngine -from notify import send_bsp_alert, BOT_TOKEN, CHAT_ID - -_PARENT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) -if _PARENT not in sys.path: - sys.path.insert(0, _PARENT) -from ChanEnum import Chan_BI_DIR -# from ChanPivotMonitor import ChanPivotMonitor # 暂停中枢监控 - -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", -) -logger = logging.getLogger("bsp_monitor") - -TIMEFRAMES = ["1m", "5m", "15m", "1h"] - - -def _short(symbol: str) -> str: - """BTC/USDT:USDT → BTCUSDT""" - return symbol.split(":")[0].replace("/", "") - - -def _bi_id(bi) -> Optional[tuple]: - """笔的稳定标识，基于首K线时间戳。""" - if bi.start_klc is None: - return None - return (bi.start_klc.start_time,) - - -def _push_bsp(engine: ChanEngine, bsp, symbol: str, tf: str) -> bool: - """推送 BSP 到 Telegram（带去重）。""" - if bsp.klc is None: - return False - key = f"{symbol}_{bsp.type}_{bsp.klc.end_time}_{tf}" - msg = engine.format_bsp_detail(bsp, symbol, tf) - msg = _escape_html(msg) - if send_bsp_alert(msg, bsp_key=key): - logger.info(f"[{_short(symbol)} {tf}] ✅ BSP: {key}") - return True - return False - - -@dataclass -class TfState: - """单个周期的状态。""" - last_bi_id: Optional[tuple] = None - last_df_ts: object = None - first_run: bool = True - # pivot_monitor: ChanPivotMonitor = None # 暂停中枢监控 - - # def __post_init__(self): - # if self.pivot_monitor is None: - # self.pivot_monitor = ChanPivotMonitor() - - -@dataclass -class SymbolState: - symbol: str - tfs: dict = field(default_factory=dict) - - def __post_init__(self): - self.tfs = {tf: TfState() for tf in TIMEFRAMES} - - -class BSPMonitor: - def __init__(self): - symbols = get_symbols() - self._states: dict[str, SymbolState] = { - s: SymbolState(symbol=s) for s in symbols - } - logger.info(f"监控 {len(symbols)}×{len(TIMEFRAMES)} 币对×周期: " - f"{', '.join(_short(s) for s in symbols)}") - - async def tick(self): - tick_start = time.monotonic() - logger.info("── tick 开始 ──") - - for symbol, st in self._states.items(): - await self._tick_symbol(symbol, st) - - elapsed = (time.monotonic() - tick_start) * 1000 - logger.info(f"── tick 结束 ({elapsed:.0f}ms) ──") - - async def _tick_symbol(self, symbol: str, st: SymbolState): - name = _short(symbol) - - for tf in TIMEFRAMES: - await self._check_tf(symbol, tf, st.tfs[tf], name) - - async def _check_tf(self, symbol: str, tf: str, ts: TfState, name: str): - # 1. 拉取 K 线 - try: - df = fetch_ohlcv(symbol, tf) - except Exception as e: - logger.error(f"[{name} {tf}] 拉取失败: {e}") - return - - if df.empty: - return - - # 2. 检查是否有新 K 线 - latest_ts = df.iloc[-1]["timestamp"] - if ts.last_df_ts and latest_ts <= ts.last_df_ts: - return - ts.last_df_ts = latest_ts - - # 3. 运行缠论管线 - try: - engine = ChanEngine(df) - except Exception as e: - logger.error(f"[{name} {tf}] 缠论计算失败: {e}", exc_info=True) - return - - # 4. 中枢特征更新（暂停） - # try: - # ts.pivot_monitor.update(engine.bi_zs_list) - # except Exception as e: - # logger.debug(f"[{name} {tf}] 中枢特征更新失败: {e}") - - # 5. BSP 检测 - confirmed = [b for b in engine.bi_list if b.is_sure] - if len(confirmed) < 2: - return - - last_confirmed = confirmed[-1] - current_bi_id = _bi_id(last_confirmed) - if current_bi_id is None: - return - - if ts.first_run: - ts.first_run = False - ts.last_bi_id = current_bi_id - - bsp = engine.get_bsp_for_bi(last_confirmed) - if bsp: - _push_bsp(engine, bsp, symbol, tf) - - logger.info( - f"[{name} {tf}] 首次完成 — " - f"{len(confirmed)} 笔, {len(engine.bsp_list)} BSP" - ) - return - - if current_bi_id == ts.last_bi_id: - return - - ts.last_bi_id = current_bi_id - - bi_dir = "⬆️" if last_confirmed.dir == Chan_BI_DIR.UP else "⬇️" - logger.info(f"[{name} {tf}] 新笔确认 — #{len(confirmed)} " - f"{bi_dir} 高度: ${last_confirmed.height:.2f}") - - bsp = engine.get_bsp_for_bi(last_confirmed) - if bsp: - _push_bsp(engine, bsp, symbol, tf) - - async def run(self): - logger.info("=" * 50) - logger.info(f"bsp_monitor 启动 — {len(self._states)} 币对 " - f"× {len(TIMEFRAMES)} 周期 ({', '.join(TIMEFRAMES)})") - logger.info(f"Telegram: {'已配置' if BOT_TOKEN and CHAT_ID else '⚠️ 未配置'}") - logger.info("=" * 50) - - logger.info("首次运行（初始化）...") - await self.tick() - - while True: - now = datetime.now(timezone.utc) - next_minute = now.replace(second=0, microsecond=0) + timedelta(minutes=1) - wait_seconds = max(0.1, (next_minute - now).total_seconds()) - - logger.info(f"等待 {wait_seconds:.0f}s 到 {next_minute.strftime('%H:%M:%S')}UTC") - await asyncio.sleep(wait_seconds) - - try: - await self.tick() - except Exception as e: - logger.error(f"tick 异常: {e}", exc_info=True) - await asyncio.sleep(5) - - -def _escape_html(msg: str) -> str: - """HTML 转义，保留已有的 / 标签。""" - msg = msg.replace("&", "&") - msg = msg.replace("", "\x00B\x00").replace("", "\x00/B\x00") - msg = msg.replace("", "\x00C\x00").replace("", "\x00/C\x00") - msg = msg.replace("<", "<").replace(">", ">") - msg = msg.replace("\x00B\x00", "").replace("\x00/B\x00", "") - msg = msg.replace("\x00C\x00", "").replace("\x00/C\x00", "") - return msg - - -if __name__ == "__main__": - monitor = BSPMonitor() - try: - asyncio.run(monitor.run()) - except KeyboardInterrupt: - logger.info("收到中断信号，退出") diff --git a/bsp_monitor/notify.py b/bsp_monitor/notify.py deleted file mode 100644 index ee114b3..0000000 --- a/bsp_monitor/notify.py +++ /dev/null @@ -1,61 +0,0 @@ -""" -notify.py - Telegram 推送。 -""" -import logging -import requests - -logger = logging.getLogger(__name__) - -BOT_TOKEN = "8742822093:AAGzD1vS7ru7ROhgcOjA-UyHb4R8Cfcqv3Q" -CHAT_ID = "580807463" - - -def send_telegram_message(text: str) -> bool: - """发送 Telegram 消息（不去重，每次调用都发）。""" - if not BOT_TOKEN or not CHAT_ID: - logger.warning("Telegram 未配置，跳过推送") - return False - - url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage" - try: - resp = requests.post( - url, - json={ - "chat_id": CHAT_ID, - "text": text, - "parse_mode": "HTML", - "disable_web_page_preview": True, - }, - timeout=10, - ) - resp.raise_for_status() - return True - except Exception as e: - logger.error(f"Telegram 推送失败: {e}") - return False - - -def send_bsp_alert(text: str, bsp_key: str = "") -> bool: - """推送 BSP 消息。""" - if not BOT_TOKEN or not CHAT_ID: - logger.warning("Telegram 未配置，跳过推送") - return False - - url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage" - try: - resp = requests.post( - url, - json={ - "chat_id": CHAT_ID, - "text": text, - "parse_mode": "HTML", - "disable_web_page_preview": True, - }, - timeout=10, - ) - resp.raise_for_status() - logger.info(f"Telegram 推送成功: {bsp_key or 'no-key'}") - return True - except Exception as e: - logger.error(f"Telegram 推送失败: {e}") - return False diff --git a/bsp_monitor/run.sh b/bsp_monitor/run.sh deleted file mode 100755 index eb1588b..0000000 --- a/bsp_monitor/run.sh +++ /dev/null @@ -1,11 +0,0 @@ -#!/bin/bash -# bsp_monitor 启动脚本 -# 用法: bash run.sh - -cd "$(dirname "$0")" -echo "=== bsp_monitor ===" -echo "启动时间: $(date -u '+%Y-%m-%d %H:%M:%S UTC')" -echo "监控: BTC/USDT:USDT 1m 缠论买卖点" -echo "推送: Telegram (复用 Hermes bot)" -echo "===================" -exec /usr/bin/python3 -u main.py diff --git a/bsp_monitor/twitter_web.py b/bsp_monitor/twitter_web.py deleted file mode 100644 index cbdec57..0000000 --- a/bsp_monitor/twitter_web.py +++ /dev/null @@ -1,380 +0,0 @@ -#!/usr/bin/env python3 -""" -twitter_web.py — Twitter 监控账号管理 Web 界面。 -单文件，零依赖，只用到 Python 标准库。 -""" - -import json -import os -import sys -import re -from datetime import datetime, timezone -from http.server import HTTPServer, BaseHTTPRequestHandler -from urllib.parse import urlparse, parse_qs - -SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) -WATCHLIST_PATH = os.path.join(SCRIPT_DIR, "twitter_watchlist.json") -STATE_PATH = os.path.join(SCRIPT_DIR, "twitter_state.json") - -PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8010 - - -def extract_username(value: str) -> str: - value = value.strip().rstrip("/") - if value.startswith("@"): - return value[1:] - for pattern in [r"(?:twitter\.com|x\.com)/(\w+)(?:/|$)", r"/(\w+)$"]: - m = re.search(pattern, value) - if m: - return m.group(1) - if re.match(r"^\w+$", value): - return value - raise ValueError(f"无法提取用户名: {value}") - - -def load_json(path): - if os.path.exists(path): - with open(path) as f: - return json.load(f) - return {} - - -def save_json(path, data): - with open(path, "w") as f: - json.dump(data, f, indent=2, ensure_ascii=False) - - -def get_watchlist(): - return load_json(WATCHLIST_PATH).get("users", []) - - -def save_watchlist(users): - save_json(WATCHLIST_PATH, {"users": users}) - - -def get_state(): - return load_json(STATE_PATH) - - -HTML = """ - - - - - - - -Twitter 监控管理 - - - -

🐦 Twitter 账号监控

-

管理 twitterapi.io 监控账号 · 增删改查

- -
- - -
- -
- -
- - - -""" - - -class Handler(BaseHTTPRequestHandler): - def log_message(self, format, *args): - pass # silent - - def _send(self, code, body, content_type="application/json"): - body = body.encode() if isinstance(body, str) else json.dumps(body, ensure_ascii=False).encode() - self.send_response(code) - self.send_header("Content-Type", content_type + "; charset=utf-8") - self.send_header("Content-Length", str(len(body))) - self.send_header("Access-Control-Allow-Origin", "*") - self.end_headers() - self.wfile.write(body) - - def _json(self, code, data): - self._send(code, data) - - def _error(self, code, msg): - self._json(code, {"error": msg}) - - def do_OPTIONS(self): - self.send_response(204) - self.send_header("Access-Control-Allow-Origin", "*") - self.send_header("Access-Control-Allow-Methods", "GET,POST,PUT,DELETE,OPTIONS") - self.send_header("Access-Control-Allow-Headers", "Content-Type") - self.end_headers() - - def do_GET(self): - path = urlparse(self.path).path - if path == "/" or path == "/index.html": - self._send(200, HTML, "text/html") - return - if path.startswith("/api/accounts"): - username = path[len("/api/accounts"):].strip("/") - if username: - # GET /api/accounts/ — single account - users = get_watchlist() - state = get_state() - for u in users: - if u["username"].lower() == username.lower(): - entry = dict(u) - entry["last_check"] = state.get(u["username"], {}).get("last_check") - self._json(200, entry) - return - self._error(404, "账号不存在") - return - # GET /api/accounts — list all - users = get_watchlist() - state = get_state() - accounts = [] - for u in users: - entry = dict(u) - sc = state.get(u["username"], {}) - ts = sc.get("last_check") - if ts: - try: - ts = datetime.fromisoformat(ts).strftime("%m-%d %H:%M") - except Exception: - pass - else: - ts = "从未" - entry["last_check"] = ts - accounts.append(entry) - self._json(200, {"accounts": accounts}) - else: - self._error(404, "Not Found") - - def do_POST(self): - path = urlparse(self.path).path - if path != "/api/accounts": - self._error(404, "Not Found") - return - length = int(self.headers.get("Content-Length", 0)) - body = json.loads(self.rfile.read(length)) if length else {} - url = body.get("url", "").strip() - if not url: - self._error(400, "缺少 url 参数") - return - try: - username = extract_username(url) - except ValueError: - self._error(400, "无法从输入中提取用户名，请输入 Twitter/X 链接或 @用户名") - return - - users = get_watchlist() - if any(u["username"].lower() == username.lower() for u in users): - self._error(409, f"@{username} 已在监控列表中") - return - - display_name = body.get("display_name", "").strip() or username - users.append({ - "username": username, - "display_name": display_name, - "added_at": datetime.now(timezone.utc).isoformat(), - }) - save_watchlist(users) - self._json(201, {"message": f"✅ 已添加 @{username}", "username": username}) - - def do_PUT(self): - path = urlparse(self.path).path - username = path[len("/api/accounts"):].strip("/") - if not username: - self._error(400, "缺少用户名") - return - length = int(self.headers.get("Content-Length", 0)) - body = json.loads(self.rfile.read(length)) if length else {} - display_name = body.get("display_name", "").strip() - - users = get_watchlist() - for u in users: - if u["username"].lower() == username.lower(): - if display_name: - u["display_name"] = display_name - save_watchlist(users) - self._json(200, {"message": f"✅ @{username} 已更新"}) - return - self._error(404, "账号不存在") - - def do_DELETE(self): - path = urlparse(self.path).path - username = path[len("/api/accounts"):].strip("/") - if not username: - self._error(400, "缺少用户名") - return - users = get_watchlist() - before = len(users) - users = [u for u in users if u["username"].lower() != username.lower()] - if len(users) < before: - save_watchlist(users) - self._json(200, {"message": f"🗑 已移除 @{username}"}) - else: - self._error(404, "账号不存在") - - -def main(): - print(f"🐦 Twitter 监控管理: http://0.0.0.0:{PORT}") - server = HTTPServer(("0.0.0.0", PORT), Handler) - try: - server.serve_forever() - except KeyboardInterrupt: - print("\n已停止") - server.server_close() - - -if __name__ == "__main__": - main()