#!/usr/bin/env python3 """ 小红书签名服务 — 纯 Python，无需 Playwright 部署在用户本地电脑，通过 cloudflared 暴露给服务器远程调用。 启动: pip install requests && python sign_server.py 测试: curl http://localhost:5005/health 暴露: cloudflared tunnel --url http://localhost:5005 服务器端调用示例： GET /search?keyword=高职高考&count=10 GET /me GET /api/sns/web/v1/... (通用 API 代理) POST /api/sns/web/v1/... (通用 API 代理) POST /sign (纯签名) Cookie 通过 Authorization: Bearer 传递。 """ import hashlib, json, time, base64, os from http.server import HTTPServer, BaseHTTPRequestHandler import requests as req # ============================================================ # 签名算法 # ============================================================ def _h(n): d = "A4NjFqYu5wPHsO0XTdDgMa2r1ZQocVte9UJBvk6/7=yRnhISGKblCWi+LpfE8xzm3" m = "" for i in range(0, 32, 3): o, g, h2 = ord(n[i]), ord(n[i+1]) if i+1<32 else 0, ord(n[i+2]) if i+2<32 else 0 x = ((o & 3) << 4) | (g >> 4) p = ((15 & g) << 2) | (h2 >> 6) v = o >> 2 b = h2 & 63 if h2 else 64 if not g: p = b = 64 m += d[v] + d[x] + d[p] + d[b] return m def _encode_utf8(s): result = "" for c in s: code = ord(c) if code < 128: result += c elif code < 2048: result += chr(192 | (code >> 6)) + chr(128 | (code & 63)) else: result += chr(224 | (code >> 12)) + chr(128 | ((code >> 6) & 63)) + chr(128 | (code & 63)) return result CRC_TABLE = [0, 1996959894, 3993919788, 2567524794, 124634137, 1886057615, 3915621685, 2657392035, 249268274, 2044508324, 3772115230, 2547177864, 162941995, 2125561021, 3887607047, 2428444049, 498536548, 1789927666, 4089016648, 2227061214, 450548861, 1843258603, 4107580753, 2211677639, 325883990, 1684777152, 4251122042, 2321926636, 335633487, 1661365465, 4195302755, 2366115317, 997073096, 1281953886, 3579855332, 2724688242, 1006888145, 1258607687, 3524101629, 2768942443, 901097722, 1119000684, 3686517206, 2898065728, 853044451, 1172266101, 3705015759, 2882616665, 651767980, 1373503546, 3369554304, 3218104598, 565507253, 1454621731, 3485111705, 3099436303, 671266974, 1594198024, 3322730930, 2970347812, 795835527, 1483230225, 3244367275, 3060149565, 1994146192, 31158534, 2563907772, 4023717930, 1907459465, 112637215, 2680153253, 3904427059, 2013776290, 251722036, 2517215374, 3775830040, 2137656763, 141376813, 2439277719, 3865271297, 1802195444, 476864866, 2238001368, 4066508878, 1812370925, 453092731, 2181625025, 4111451223, 1706088902, 314042704, 2344532202, 4240017532, 1658658271, 366619977, 2362670323, 4224994405, 1303535960, 984961486, 2747007092, 3569037538, 1256170817, 1037604311, 2765210733, 3554079995, 1131014506, 879679996, 2909243462, 3663771856, 1141124467, 855842277, 2852801631, 3708648649, 1342533948, 654459306, 3188396048, 3373015174, 1466479909, 544179635, 3110523913, 3462522015, 1591671054, 702138776, 2966460450, 3352799412, 1504918807, 783551873, 3082640443, 3233442989, 3988292384, 2596254646, 62317068, 1957810842, 3939845945, 2647816111, 81470997, 1943803523, 3814918930, 2489596804, 225274430, 2053790376, 3826175755, 2466906013, 167816743, 2097651377, 4027552580, 2265490386, 503444072, 1762050814, 4150417245, 2154129355, 426522225, 1852507879, 4275313526, 2312317920, 282753626, 1742555852, 4189708143, 2394877945, 397917763, 1622183637, 3604390888, 2714866558, 953729732, 1340076626, 3518719985, 2797360999, 1068828381, 1219638859, 3624741850, 2936675148, 906185462, 1090812512, 3747672003, 2825379669, 829329135, 1181335161, 3412177804, 3160834842, 628085408, 1382605366, 3423369109, 3138078467, 570562233, 1426400815, 3317316542, 2998733608, 733239954, 1555261956, 3268935591, 3050360625, 752459403, 1541320221, 2607071920, 3965973030, 1969922972, 40735498, 2617837225, 3943577151, 1913087877, 83908371, 2512341634, 3803740692, 2075208622, 213261112, 2463272603, 3855990285, 2094854071, 198958881, 2262029012, 4057260610, 1759359992, 534414190, 2176718541, 4139329115, 1873836001, 414664567, 2282248934, 4279200368, 1711684554, 285281116, 2405801727, 4167216745, 1634467795, 376229701, 2685067896, 3608007406, 1308918612, 956543938, 2808555105, 3495958263, 1231636301, 1047427035, 2932959818, 3654703836, 1088359270, 936918000, 2847714899, 3736837829, 1202900863, 817233897, 3183342108, 3401237130, 1404277552, 615818150, 3134207493, 3453421203, 1423857449, 601450431, 3009837614, 3294710456, 1567103746, 711928724, 3020668471, 3272380065, 1510334235, 755167117] def _mrc(e): o = -1 for n in range(len(e)): o = CRC_TABLE[(o ^ ord(e[n])) & 255] ^ (o >> 8) return (o ^ -1) & 2147483647 def generate_sign(uri, data=None, a1="", web_session=""): v = int(round(time.time() * 1000)) data_str = json.dumps(data, separators=(",", ":"), ensure_ascii=False) if isinstance(data, dict) else "" raw_str = f"{v}test{uri}{data_str}" md5_str = hashlib.md5(raw_str.encode("utf-8")).hexdigest() x_s = _h(md5_str) x_t = str(v) common = { "s0": 5, "s1": "", "x0": "1", "x1": "3.2.0", "x2": "Windows", "x3": "xhs-pc-web", "x4": "2.3.1", "x5": a1, "x6": x_t, "x7": x_s, "x8": web_session, "x9": str(_mrc(x_t + x_s)), "x10": 1, } es = _encode_utf8(json.dumps(common, separators=(",", ":"))) x_s_common = base64.b64encode(es.encode("latin-1")).decode() return {"x-s": x_s, "x-t": x_t, "x-s-common": x_s_common} # ============================================================ # API 代理 # ============================================================ API_HOST = "https://edith.xiaohongshu.com" def api_request(method, uri, data=None, cookies=None): a1 = cookies.get("a1", "") ws = cookies.get("web_session", "") if method == "GET" and data: params_str = "&".join([f"{k}={v}" for k, v in data.items()]) final_uri = f"{uri}?{params_str}" sign_data = None else: final_uri = uri sign_data = data sign = generate_sign(final_uri, sign_data, a1, ws) headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Content-Type": "application/json;charset=UTF-8", "Cookie": "; ".join([f"{k}={v}" for k, v in cookies.items()]), "x-s": sign["x-s"], "x-t": sign["x-t"], "x-s-common": sign["x-s-common"], "Origin": "https://www.xiaohongshu.com", "Referer": "https://www.xiaohongshu.com/", } url = f"{API_HOST}{final_uri}" if method == "GET": resp = req.get(url, headers=headers, timeout=15) else: resp = req.post(url, data=json.dumps(data or {}, separators=(",", ":")).encode(), headers=headers, timeout=15) try: return resp.json() except: return {"_raw": resp.text, "_status": resp.status_code} # ============================================================ # HTTP 服务 # ============================================================ class SignHandler(BaseHTTPRequestHandler): def _send_json(self, data, status=200): body = json.dumps(data, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(body) def do_GET(self): from urllib.parse import urlparse, parse_qs parsed = urlparse(self.path) if self.path == "/health": self._send_json({"status": "ok", "service": "xhs-sign-server"}) elif self.path == "/me": cookies = self._get_cookies() result = api_request("GET", "/api/sns/web/v2/user/me", cookies=cookies) self._send_json(result) elif self.path == "/search": qs = parse_qs(parsed.query) keyword = qs.get("keyword", ["高职高考"])[0] count = int(qs.get("count", [5])[0]) cookies = self._get_cookies() data = { "keyword": keyword, "page": 1, "page_size": count, "search_id": hashlib.md5(str(time.time()).encode()).hexdigest()[:20].upper(), "sort": "general", "note_type": 0, } result = api_request("POST", "/api/sns/web/v1/search/notes", data, cookies) self._send_json(result) elif self.path.startswith("/api/"): uri = parsed.path.replace("/api", "", 1) params = {k: v[0] for k, v in parse_qs(parsed.query).items()} cookies = self._get_cookies() result = api_request("GET", uri, data=params or None, cookies=cookies) self._send_json(result) else: self._send_json({"error": "not found"}, 404) def do_POST(self): content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) if content_length > 0 else b"{}" if self.path == "/sign": payload = json.loads(body) result = generate_sign( payload.get("uri", ""), payload.get("data"), payload.get("a1", ""), payload.get("web_session", ""), ) self._send_json(result) elif self.path.startswith("/api/"): uri = self.path.replace("/api", "", 1) payload = json.loads(body) cookies = self._get_cookies() result = api_request("POST", uri, payload, cookies) self._send_json(result) else: self._send_json({"error": "not found"}, 404) def do_OPTIONS(self): self.send_response(200) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Authorization, Content-Type") self.end_headers() def _get_cookies(self): auth = self.headers.get("Authorization", "") if auth.startswith("Bearer "): try: return json.loads(base64.b64decode(auth[7:]).decode()) except: pass return {} def log_message(self, format, *args): print(f"[{time.strftime('%H:%M:%S')}] {args[0]}") def main(): port = int(os.environ.get("PORT", "5005")) server = HTTPServer(("0.0.0.0", port), SignHandler) print(f"\n小红书 API 签名代理 :{port}\n测试: curl http://localhost:{port}/health\n") try: server.serve_forever() except KeyboardInterrupt: print("\n已停止") server.server_close() if __name__ == "__main__": main()