diff --git a/bsp_monitor/twitter_web.py b/bsp_monitor/twitter_web.py new file mode 100644 index 0000000..cbdec57 --- /dev/null +++ b/bsp_monitor/twitter_web.py @@ -0,0 +1,380 @@ +#!/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() diff --git a/web/templates/chan_tv.html b/web/templates/chan_tv.html index 50a5171..a8a2993 100644 --- a/web/templates/chan_tv.html +++ b/web/templates/chan_tv.html @@ -1,6 +1,15 @@ + + + + 缠论 Chart — TradingView diff --git a/web/templates/index.html b/web/templates/index.html index fe8cca4..4f4ddf0 100644 --- a/web/templates/index.html +++ b/web/templates/index.html @@ -1,6 +1,15 @@ + + + + 缠论分析系统