""" AO3 反代后端 v5 — 粘性会话 + Service Worker + P2C 代理池 + 并发限流 v5 vs v4: - 粘性会话: ao3_sessid_proxy cookie (对标 go3) - Service Worker: /sw.js, /sw-YYYYMMDD.js for client-side caching - HTML 注入: 自动注入 SW 注册脚本 + 镜像域名 - P2C 代理选择: Power of Two Choices (对标 go3) - 并发限流: asyncio.Semaphore(400) (Webshare 上限) - 全量 5000 代理: 不再按端口过滤 - 状态机: healthy/unstable/blocked/probing """ import hashlib import html import json import logging import os import secrets import sys import threading import time from urllib.parse import urlparse, urlunparse from fastapi import FastAPI, Request, Response from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse, RedirectResponse import uvicorn sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from proxy_pool import get_proxy_pool, MAX_CONCURRENT_REQUESTS, OUTAGE_THRESHOLD from ao3_fetcher import fetch_url, is_cf_challenge, FAST_REQUEST_TIMEOUT, NORMAL_REQUEST_TIMEOUT from url_rewriter import rewrite_body, rewrite_response_headers, rewrite_response_headers_raw, needs_rewrite, MIRROR_DOMAIN from cache import get_cache, get_ttl_for_path from stats import get_stats_collector logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[logging.StreamHandler()], ) logger = logging.getLogger("ao3-backend") AO3_BASE = "https://archiveofourown.org" MIRROR_HOST = MIRROR_DOMAIN # Service Worker version date (change on SW updates) SERVICE_WORKER_DATE = "20260623" LOCAL_PATHS = {"/stats", "/health", "/metrics", "/favicon.ico", "/robots.txt"} # ─── Challenge Token Map (v4 retained) ────────────────────────────────────── _challenge_map: dict[str, tuple] = {} _challenge_lock = threading.Lock() CHALLENGE_TOKEN_TTL = 120 def _make_challenge_token() -> str: return secrets.token_urlsafe(16) def _store_challenge(token: str, method: str, ao3_url: str, headers: dict, body: bytes, cookies: dict, proxy_host: str): with _challenge_lock: now = time.time() expired = [k for k, v in _challenge_map.items() if v[6] < now] for k in expired: del _challenge_map[k] _challenge_map[token] = (method, ao3_url, headers, body, cookies, proxy_host, now + CHALLENGE_TOKEN_TTL) def _get_challenge(token: str) -> tuple | None: with _challenge_lock: entry = _challenge_map.get(token) if entry and entry[6] > time.time(): del _challenge_map[token] return entry if entry: del _challenge_map[token] return None # ─── Sticky Session ───────────────────────────────────────────────────────── STICKY_COOKIE_NAME = "ao3_sessid_proxy" def get_sticky_proxy_idx(request: Request) -> tuple[int, bool]: """Get sticky proxy index from cookie. Returns (idx, had_sticky).""" c = request.cookies.get(STICKY_COOKIE_NAME) if c is not None: try: idx = int(c) pool = get_proxy_pool() if 0 <= idx < len(pool.proxies): return idx, True except (ValueError, TypeError): pass # No sticky cookie — random import random pool = get_proxy_pool() return random.randint(0, max(len(pool.proxies) - 1, 0)), False # ─── App ─────────────────────────────────────────────────────────────────── app = FastAPI( title="AO3 Mirror", description="AO3 reverse proxy mirror for Chinese users", version="5.0.0", docs_url=None, redoc_url=None, ) def get_client_ip(request: Request) -> str: cf_ip = request.headers.get("CF-Connecting-IP") if cf_ip: return cf_ip forwarded = request.headers.get("X-Forwarded-For") if forwarded: return forwarded.split(",")[0].strip() return request.client.host if request.client else "unknown" def build_ao3_url(path: str, query: str = "") -> str: url = f"{AO3_BASE}{path}" if query: url = f"{url}?{query}" return url # ─── Challenge page URL rewriting ────────────────────────────────────────── def _rewrite_challenge_page(body: bytes, proxy_host: str, token: str) -> bytes: """Rewrite CF challenge page — inject retry UI + auto cookie.""" # Rewrite domain body = body.replace(b"archiveofourown.org", MIRROR_DOMAIN.encode()) # Rewrite CF challenge paths — our CF CDN would intercept /cdn-cgi/, # so we remap to /__cf/ which our proxy handler forwards to AO3 body = body.replace(b"/cdn-cgi/", b"/__cf/") body = body.replace(b"cdn-cgi/challenge", b"__cf/challenge") # Inject cookie + retry overlay retry_overlay = ( f''.encode() + b'' + b'
' + 'AO3 Mirror - CF challenge solving in background. '.encode() + 'Click to retry after a few seconds.'.encode() + b'
' ) body = body.replace(b" bytes: """Inject SW registration script into HTML, before .""" import re sw_script = ( b'' ) # Match with optional leading whitespace match = re.search(rb'\s*', html_body) if match: pos = match.start() return html_body[:pos] + sw_script + html_body[pos:] # Fallback: insert at beginning if no head tag return sw_script + html_body # ─── Response header helpers ─────────────────────────────────────────────── def filter_response_headers(headers: dict) -> dict: blocked = { "transfer-encoding", "content-encoding", "alt-svc", "cf-ray", "cf-cache-status", "cf-request-id", "server", "x-powered-by", "cross-origin-resource-policy", "cross-origin-embedder-policy", "cross-origin-opener-policy", "cross-origin-window-policy", "accept-ch", "critical-ch", } result = {} for key, value in headers.items(): if key.lower() in blocked: continue if key.lower().startswith("cf-"): continue result[key] = value result["X-Proxy"] = "AO3-Mirror/5.0" result["X-Cache"] = "MISS" result["Cache-Control"] = "public, max-age=60, s-maxage=60" return result def add_cors(response: Response): response.headers["Access-Control-Allow-Origin"] = "*" response.headers["Access-Control-Allow-Methods"] = "GET, POST, HEAD, OPTIONS, PUT, DELETE, PATCH" response.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization, Cookie, X-CSRF-Token" response.headers["Access-Control-Max-Age"] = "86400" response.headers["Access-Control-Allow-Credentials"] = "true" # ─── Local Routes ────────────────────────────────────────────────────────── @app.get("/health") async def health(): pool = get_proxy_pool() stats = pool.get_stats() return JSONResponse({ "status": "ok", "timestamp": time.time(), "version": "5.0.0", "proxy_pool": stats, }) @app.get("/robots.txt") async def robots(): return PlainTextResponse( "User-agent: *\nDisallow: /stats\nDisallow: /health\n" ) # ─── Service Worker Routes ───────────────────────────────────────────────── @app.get("/sw.js") async def service_worker_redirect(): """Redirect to versioned SW file.""" return RedirectResponse(url=f"/sw-{SERVICE_WORKER_DATE}.js", status_code=302) @app.get("/sw-{date}.js") async def service_worker(date: str): """Serve versioned Service Worker.""" sw_path = os.path.join(os.path.dirname(__file__), "static", "sw.js") try: with open(sw_path, "rb") as f: sw_content = f.read() except FileNotFoundError: return Response(status_code=404) return Response( content=sw_content, media_type="application/javascript", headers={ "Cache-Control": "public, max-age=86400, immutable", "Service-Worker-Allowed": "/", }, ) @app.get("/mirror-domains.json") async def mirror_domains(): """JSON endpoint for SW cross-domain awareness.""" return JSONResponse({ "domains": [MIRROR_DOMAIN], "primary": MIRROR_DOMAIN, }) # ─── Stats Dashboard ─────────────────────────────────────────────────────── @app.get("/stats", response_class=HTMLResponse) async def stats_page(): collector = get_stats_collector() pool = get_proxy_pool() cache = get_cache() stats = collector.get_overview() proxy_stats = pool.get_stats() cache_stats = cache.get_stats() hourly = stats.get("hourly", []) chart_labels = json.dumps([ time.strftime("%H:%M", time.localtime(h["hour"])) for h in hourly[-24:] ]) chart_requests = json.dumps([h["total"] for h in hourly[-24:]]) chart_success = json.dumps([h["successful"] for h in hourly[-24:]]) chart_elapsed = json.dumps([h["avg_elapsed"] * 1000 for h in hourly[-24:]]) return f""" AO3 Mirror - 统计面板 v5

🛡 AO3 Mirror 统计面板 v5

总请求数
{stats['total_requests']:,}
运行 {stats['uptime_human']}
最近 1 分钟
{stats['recent_1m']:,} req
最近 5 分钟: {stats['recent_5m']:,}
成功率
{stats['success_rate']}%
失败: {stats['failed']:,}
缓存命中率
{stats['cache_rate']}%
已缓存: {stats['cached']:,}
平均延迟
{stats['avg_elapsed_ms']} ms
P99: {stats['p99_elapsed_ms']} ms
代理池 ({proxy_stats['total']})
{proxy_stats['healthy']}/{proxy_stats['total']}
不稳定: {proxy_stats['unstable']} | 封禁: {proxy_stats['blocked']} | 探测: {proxy_stats['probing']}
Cookie 代理
{proxy_stats['with_cookies']}/{proxy_stats['total']}
inflight: {proxy_stats['total_inflight']}
本地缓存
{cache_stats['size']}/{cache_stats['capacity']}
已用: {cache_stats['usage_pct']}%

请求趋势 (最近 24 小时)

响应延迟 (最近 24 小时)

热门路径

{''.join(f'' for p in stats['top_paths'][:15])}
路径请求数成功平均延迟
{html.escape(p["path"])}{p["requests"]:,}0.8 else "badge-red"}">{round(p["successful"]/max(p["requests"],1)*100)}%{p["avg_elapsed"]*1000:.0f}ms
""" # ─── Metrics ─────────────────────────────────────────────────────────────── @app.get("/metrics") async def metrics(): collector = get_stats_collector() pool = get_proxy_pool() cache = get_cache() stats = collector.get_overview() proxy_stats = pool.get_stats() cache_stats = cache.get_stats() lines = [ "# HELP ao3_mirror_requests_total Total proxy requests", "# TYPE ao3_mirror_requests_total counter", f'ao3_mirror_requests_total {stats["total_requests"]}', "# HELP ao3_mirror_successful_requests Successful requests", "# TYPE ao3_mirror_successful_requests counter", f'ao3_mirror_successful_requests {stats["successful"]}', "# HELP ao3_mirror_failed_requests Failed requests", "# TYPE ao3_mirror_failed_requests counter", f'ao3_mirror_failed_requests {stats["failed"]}', "# HELP ao3_mirror_cache_hits Cache hit count", "# TYPE ao3_mirror_cache_hits counter", f'ao3_mirror_cache_hits {stats["cached"]}', "# HELP ao3_mirror_avg_elapsed_ms Average response time", "# TYPE ao3_mirror_avg_elapsed_ms gauge", f'ao3_mirror_avg_elapsed_ms {stats["avg_elapsed_ms"]}', "# HELP ao3_mirror_proxy_pool Proxy pool status", "# TYPE ao3_mirror_proxy_pool gauge", f'ao3_mirror_proxy_total {proxy_stats["total"]}', f'ao3_mirror_proxy_healthy {proxy_stats["healthy"]}', f'ao3_mirror_proxy_unstable {proxy_stats["unstable"]}', f'ao3_mirror_proxy_blocked {proxy_stats["blocked"]}', f'ao3_mirror_proxy_probing {proxy_stats["probing"]}', f'ao3_mirror_proxy_with_cookies {proxy_stats["with_cookies"]}', f'ao3_mirror_proxy_inflight {proxy_stats["total_inflight"]}', "# HELP ao3_mirror_cache_size Current cache size", "# TYPE ao3_mirror_cache_size gauge", f'ao3_mirror_cache_size {cache_stats["size"]}', ] return PlainTextResponse("\n".join(lines)) # ─── Proxy handler (v5) ────────────────────────────────────────────────── @app.api_route("/{path:path}", methods=["GET", "POST", "HEAD", "OPTIONS", "PUT", "DELETE", "PATCH"]) async def proxy_handler(request: Request, path: str): """Main proxy handler — sticky sessions + P2C selection + semaphore-limited.""" start_time = time.time() client_ip = get_client_ip(request) if path == "" or path == "/": path = "" full_path = f"/{path}" if path else "/" if full_path in LOCAL_PATHS or full_path.startswith("/sw"): return JSONResponse({"error": "Not found"}, status_code=404) # Remap /__cf/ → /cdn-cgi/ (CF challenge paths bypass our CDN) if full_path.startswith("/__cf/"): path = "cdn-cgi" + path[4:] full_path = "/" + path if request.method == "OPTIONS": resp = Response() add_cors(resp) return resp pool = get_proxy_pool() query_string = request.url.query ao3_url = build_ao3_url(f"/{path}" if path else "/", query_string) # ── v4: Challenge token check ──────────────────────────────────────── cf_token = request.cookies.get("_cf_token") or request.query_params.get("_cf_token") if cf_token: challenge_entry = _get_challenge(cf_token) if challenge_entry: orig_method, orig_url, orig_headers, orig_body, orig_cookies, proxy_host, _ = challenge_entry logger.info(f"Challenge resolution: using proxy {proxy_host}") challenge_proxy_obj = pool.get_by_addr(proxy_host) result = await fetch_url( url=orig_url, method=orig_method, proxy=challenge_proxy_obj, headers=dict(request.headers), body=await request.body() if orig_method in ("POST", "PUT", "PATCH") else None, cookies=orig_cookies, is_api="/api/" in orig_url, ) if result["success"]: elapsed = time.time() - start_time rewritten_body = rewrite_body(result["body"], result["headers"].get("Content-Type", result["headers"].get("content-type", ""))) content_type = result["headers"].get("Content-Type", result["headers"].get("content-type", "")) if "text/html" in content_type: rewritten_body = inject_service_worker(rewritten_body) rewritten_headers = rewrite_response_headers(result["headers"]) final_headers = filter_response_headers(rewritten_headers) final_headers.pop("Set-Cookie", None) final_headers.pop("set-cookie", None) final_headers["X-Cache"] = "MISS" final_headers["X-CF-Status"] = "solved" resp = Response(content=rewritten_body, status_code=result["status"], headers=final_headers) # Append ALL Set-Cookie from raw (preserves multi-value) # Filter: strip AO3 CF cookies SKIP_CF = {"__cf_bm", "_cfuvid"} ao3_headers_raw = result.get("headers_raw", []) if ao3_headers_raw: rewritten_raw = rewrite_response_headers_raw(ao3_headers_raw) for key, value in rewritten_raw: if key.lower() == "set-cookie": cookie_name = value.split("=")[0].split(";")[0].strip() if cookie_name in SKIP_CF: continue resp.headers.append("Set-Cookie", value) resp.delete_cookie("_cf_token", path="/") collector = get_stats_collector() collector.log_request(method=orig_method, path=full_path, status=result["status"], elapsed=elapsed, cached=False, proxy_host=proxy_host, client_ip=client_ip) add_cors(resp) return resp # ── Static cache check (static assets only) ──────────────────── from cache import is_static_path cache = get_cache() if request.method == "GET" and is_static_path(full_path): cached = cache.get(ao3_url) if cached: body, resp_headers, status = cached elapsed = time.time() - start_time collector = get_stats_collector() collector.log_request(method=request.method, path=full_path, status=status, elapsed=elapsed, cached=True, client_ip=client_ip) return Response(content=body, status_code=status, headers=resp_headers) # ── Prepare request ────────────────────────────────────────────────── method = request.method client_headers = dict(request.headers) req_body = None if method in ("POST", "PUT", "PATCH"): req_body = await request.body() cookies = {} cookie_header = request.headers.get("cookie", "") if cookie_header: for pair in cookie_header.split(";"): if "=" in pair: k, v = pair.split("=", 1) cookies[k.strip()] = v.strip() # ── v5: Sticky proxy + P2C selection with semaphore ────────────────── sticky_idx, had_sticky = get_sticky_proxy_idx(request) final_proxy_idx = sticky_idx tried_idxs: set[int] = set() import random is_fast = method in ("POST", "PUT", "PATCH") or "/users/" in full_path request_timeout = FAST_REQUEST_TIMEOUT if is_fast else NORMAL_REQUEST_TIMEOUT # Try with sticky first, then P2C fallback has_session = bool(cookies) # Stateful request: has user cookies session_cookie_val = cookies.get("_otwarchive_session", "")[:80] logger.info(f"Request {method} {full_path} | cookies={list(cookies.keys())} | session={session_cookie_val}... | has_session={has_session}") async with pool.concurrency_limiter: # First attempt: sticky proxy result = None sticky_proxy = pool.sticky_select(sticky_idx, tried_idxs) if sticky_proxy: sticky_proxy.inflight += 1 tried_idxs.add(sticky_proxy.idx) try: result = await fetch_url( url=ao3_url, method=method, proxy=sticky_proxy, headers=client_headers, body=req_body, cookies=cookies, is_api="/api/" in full_path, ) if result["success"]: final_proxy_idx = sticky_proxy.idx sticky_proxy.mark_success(result.get("elapsed", 0)) sticky_proxy.inflight -= 1 elif result.get("is_challenge"): sticky_proxy.mark_challenged() sticky_proxy.inflight -= 1 else: status = result.get("status", 0) if status in (403, 525): sticky_proxy.mark_blocked() else: sticky_proxy.mark_failure(is_network=True) sticky_proxy.inflight -= 1 result = None # Force retry except Exception: if sticky_proxy: sticky_proxy.mark_failure(is_network=True) sticky_proxy.inflight -= 1 result = None # Retry loop with P2C — but NEVER for stateful requests # If the sticky proxy failed on a request with cookies, switching # proxies breaks CSRF/session affinity. Forward CF challenge instead. if (not result or not result["success"]) and not has_session: max_retries = 5 for attempt in range(max_retries): if attempt > 0: await _sleep(0.1 * (2 ** min(attempt, 3))) proxy = pool.p2c_select(exclude_idxs=tried_idxs) if not proxy: break proxy.inflight += 1 tried_idxs.add(proxy.idx) try: result = await fetch_url( url=ao3_url, method=method, proxy=proxy, headers=client_headers, body=req_body, cookies=cookies, is_api="/api/" in full_path, ) if result["success"]: final_proxy_idx = proxy.idx proxy.mark_success(result.get("elapsed", 0)) proxy.inflight -= 1 break elif result.get("is_challenge"): proxy.mark_challenged() proxy.inflight -= 1 else: status = result.get("status", 0) if status in (403, 525): proxy.mark_blocked() elif status > 0: proxy.mark_failure(is_network=False) else: proxy.mark_failure(is_network=True) proxy.inflight -= 1 result = None except Exception: proxy.mark_failure(is_network=True) proxy.inflight -= 1 result = None elapsed = time.time() - start_time # ── All proxies exhausted — outage ─────────────────────────────────── if not result or not result["success"]: if result and result.get("is_challenge") and result.get("challenge_body"): challenge_proxy = result.get("challenge_proxy", "unknown") token = _make_challenge_token() _store_challenge(token, method, ao3_url, client_headers, req_body or b"", cookies, challenge_proxy) rewritten_challenge = _rewrite_challenge_page( result["challenge_body"], challenge_proxy, token ) logger.warning(f"CF Challenge via {challenge_proxy}, sending to browser") collector = get_stats_collector() collector.log_request(method=method, path=full_path, status=503, elapsed=elapsed, cached=False, proxy_host=challenge_proxy, client_ip=client_ip) resp = HTMLResponse( content=rewritten_challenge, status_code=200, # 200 so browser renders and executes CF JS headers={ "X-CF-Challenge": "true", "X-CF-Challenge-Proxy": challenge_proxy, }, ) resp.set_cookie( key="_cf_token", value=token, path="/", max_age=CHALLENGE_TOKEN_TTL, httponly=False, samesite="lax", ) resp.delete_cookie("cf_clearance", path="/") add_cors(resp) return resp # Check if we hit outage threshold if pool.pool_unavailable_ratio() >= OUTAGE_THRESHOLD: logger.error(f"OUTAGE: {pool.pool_unavailable_ratio():.1%} proxies unavailable") return HTMLResponse( content=_outage_page(), status_code=503 ) logger.error(f"Failed: {ao3_url[:80]}: all retries exhausted") collector = get_stats_collector() collector.log_request(method=method, path=full_path, status=502, elapsed=elapsed, cached=False, client_ip=client_ip) return HTMLResponse(content=_transient_error_page(), status_code=502) # ── Success ────────────────────────────────────────────────────────── ao3_status = result["status"] ao3_headers = result.get("headers", {}) ao3_headers_raw = result.get("headers_raw", []) raw_body = result.get("body", b"") content_type = ao3_headers.get("Content-Type", ao3_headers.get("content-type", "")) rewritten_body = rewrite_body(raw_body, content_type) # Inject Service Worker into HTML responses if "text/html" in content_type: rewritten_body = inject_service_worker(rewritten_body) rewritten_headers = rewrite_response_headers(ao3_headers) final_headers = filter_response_headers(rewritten_headers) # Remove Set-Cookie from dict — multi-value Set-Cookie from raw below final_headers.pop("Set-Cookie", None) final_headers.pop("set-cookie", None) final_headers["X-Cache"] = "MISS" # Cache successful static GET responses (disk MD5 cache, like go3) if method == "GET" and ao3_status == 200 and is_static_path(full_path): cache.set(ao3_url, rewritten_body, rewritten_headers, ao3_status) # Handle redirects if 300 <= ao3_status < 400 and "location" in rewritten_headers: redirect_url = rewritten_headers["location"] if AO3_BASE in redirect_url: redirect_url = redirect_url.replace(AO3_BASE, "").replace("http://", "https://") return RedirectResponse(url=redirect_url, status_code=ao3_status) collector = get_stats_collector() collector.log_request(method=method, path=full_path, status=ao3_status, elapsed=elapsed, cached=False, proxy_host=result.get("proxy_host"), client_ip=client_ip) resp = Response(content=rewritten_body, status_code=ao3_status, headers=final_headers) # Append ALL Set-Cookie headers from raw (preserves multi-value — pitfall #18a) # Filter: strip AO3 CF cookies — our own CF edge handles these SKIP_COOKIES = {"__cf_bm", "_cfuvid"} if ao3_headers_raw: rewritten_raw = rewrite_response_headers_raw(ao3_headers_raw) for key, value in rewritten_raw: if key.lower() == "set-cookie": cookie_name = value.split("=")[0].split(";")[0].strip() if cookie_name in SKIP_COOKIES: continue logger.info(f"[SET_COOKIE_FWD] {value[:120]}...") resp.headers.append("Set-Cookie", value) # Set sticky session cookie resp.set_cookie( key=STICKY_COOKIE_NAME, value=str(final_proxy_idx), path="/", httponly=True, samesite="lax", ) add_cors(resp) return resp # ─── Error pages ─────────────────────────────────────────────────────────── def _outage_page() -> str: return """ AO3 Mirror - 暂时不可用

⚠️ AO3 暂时不可用

所有代理目前都无法连接 AO3。可能 AO3 正在维护或大规模封锁。

请稍后重试。如果持续出现，请切换到数据流量。

刷新页面
""" def _transient_error_page() -> str: return """ AO3 Mirror - 连接失败

🔄 正在尝试连接 AO3

镜像站正在通过代理重新连接 AO3 服务器。

请稍后刷新页面重试。

刷新页面
""" # ─── Helper ──────────────────────────────────────────────────────────────── async def _sleep(seconds: float): import asyncio await asyncio.sleep(seconds) # ─── Startup / Shutdown ──────────────────────────────────────────────────── @app.on_event("startup") async def startup(): logger.info("AO3 Mirror backend v5 starting up...") pool = get_proxy_pool() get_cache() get_stats_collector() logger.info(f"AO3 Mirror backend v5 started ({len(pool.proxies)} proxies, " f"P2C selection, sticky sessions, SW deploy)") @app.on_event("shutdown") async def shutdown(): logger.info("AO3 Mirror backend shutting down...") pool = get_proxy_pool() await pool.close_all() logger.info("AO3 Mirror backend stopped") # ─── Entry point ─────────────────────────────────────────────────────────── if __name__ == "__main__": port = int(os.environ.get("PORT", "8080")) workers = int(os.environ.get("WORKERS", "4")) logger.info(f"Starting AO3 Mirror v5 on port {port} with {workers} workers") uvicorn.run( "app:app", host="127.0.0.1", port=port, workers=workers, log_level="info", timeout_keep_alive=30, )