95 lines
3.2 KiB
Python
95 lines
3.2 KiB
Python
from mitmproxy import http, ctx
|
|
import json, requests
|
|
|
|
FEED_KEYWORD = "/api/sns/v1/feed"
|
|
LIKE_URL = "https://edith.xiaohongshu.com/api/sns/v1/like"
|
|
COMMENT_URL = "https://edith.xiaohongshu.com/api/sns/v1/comment"
|
|
COLLECT_URL = "https://edith.xiaohongshu.com/api/sns/v1/collect"
|
|
|
|
AUTH_HEADER_KEYS = ["Authorization", "Cookie", "X-Sign"]
|
|
COMMENT_TEXT = "好喜欢你的作品！"
|
|
FOLLOWER_THRESHOLD = 1000
|
|
|
|
session = requests.Session()
|
|
auth_headers = {}
|
|
|
|
def is_xhs_host(host: str):
|
|
return host and "xiaohongshu.com" in host
|
|
|
|
def request(flow: http.HTTPFlow):
|
|
if not is_xhs_host(flow.request.host):
|
|
# 不打印非小红书请求日志
|
|
return
|
|
# 只处理小红书请求，打印请求日志
|
|
ctx.log.info(f"[REQ] {flow.request.method} {flow.request.pretty_url}")
|
|
|
|
updated = False
|
|
for k in AUTH_HEADER_KEYS:
|
|
if k in flow.request.headers:
|
|
val = flow.request.headers[k]
|
|
if auth_headers.get(k) != val:
|
|
auth_headers[k] = val
|
|
updated = True
|
|
if updated:
|
|
session.headers.clear()
|
|
session.headers.update(auth_headers)
|
|
|
|
def response(flow: http.HTTPFlow):
|
|
if not is_xhs_host(flow.request.host):
|
|
# 不打印非小红书响应日志
|
|
return
|
|
ctx.log.info(f"[RES] {flow.request.method} {flow.request.pretty_url} - {flow.response.status_code}")
|
|
if FEED_KEYWORD not in flow.request.pretty_url:
|
|
return
|
|
try:
|
|
data = flow.response.json()
|
|
except Exception:
|
|
ctx.log.warn("响应不是 JSON")
|
|
return
|
|
|
|
posts = extract_posts(data)
|
|
for post in posts:
|
|
pid, fans, uid = post["id"], post["fans"], post["user_id"]
|
|
gender = post.get("gender", "").lower()
|
|
if fans is not None and fans < FOLLOWER_THRESHOLD and gender in ["f", "female", "女"]:
|
|
ctx.log.info(f"目标帖子: {pid}, 粉丝 {fans}, 性别 {gender}")
|
|
do_like(pid)
|
|
do_comment(pid)
|
|
do_collect(pid)
|
|
|
|
def extract_posts(data):
|
|
results = []
|
|
items = data.get("data", {}).get("items", [])
|
|
for item in items:
|
|
try:
|
|
pid = item["id"]
|
|
fans = item["user"]["fans"]
|
|
uid = item["user"]["id"]
|
|
gender = item["user"].get("gender", "")
|
|
results.append({"id": pid, "fans": fans, "user_id": uid, "gender": gender})
|
|
except Exception as e:
|
|
ctx.log.warn(f"解析帖子出错: {e}")
|
|
continue
|
|
return results
|
|
|
|
def do_like(pid):
|
|
try:
|
|
r = session.post(LIKE_URL, json={"note_id": pid, "action": "like"})
|
|
ctx.log.info(f"点赞 {pid}: {r.status_code} {r.text[:50]}")
|
|
except Exception as e:
|
|
ctx.log.warn(f"点赞失败 {pid}: {e}")
|
|
|
|
def do_comment(pid):
|
|
try:
|
|
r = session.post(COMMENT_URL, json={"note_id": pid, "content": COMMENT_TEXT})
|
|
ctx.log.info(f"评论 {pid}: {r.status_code} {r.text[:50]}")
|
|
except Exception as e:
|
|
ctx.log.warn(f"评论失败 {pid}: {e}")
|
|
|
|
def do_collect(pid):
|
|
try:
|
|
r = session.post(COLLECT_URL, json={"note_id": pid, "action": "collect"})
|
|
ctx.log.info(f"收藏 {pid}: {r.status_code} {r.text[:50]}")
|
|
except Exception as e:
|
|
ctx.log.warn(f"收藏失败 {pid}: {e}")
|