#!/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()