From 56cc2f317ae4dabbd337ce37c64d02911e8e9439 Mon Sep 17 00:00:00 2001 From: ReaJason Date: Sun, 16 Jun 2024 14:24:53 +0800 Subject: [PATCH] feat: add qrcode login of creator (#111) --- example/login_qrcode_from_creator.py | 45 +++++++ tests/test_xhs.py | 195 +++++++++++++++++++-------- xhs/core.py | 74 ++++++++-- 3 files changed, 246 insertions(+), 68 deletions(-) create mode 100644 example/login_qrcode_from_creator.py diff --git a/example/login_qrcode_from_creator.py b/example/login_qrcode_from_creator.py new file mode 100644 index 0000000..1993083 --- /dev/null +++ b/example/login_qrcode_from_creator.py @@ -0,0 +1,45 @@ +import datetime +import json +from time import sleep + +import qrcode +import requests + +from xhs import XhsClient + + +def sign(uri, data=None, a1="", web_session=""): + # 填写自己的 flask 签名服务端口地址 + res = requests.post("http://localhost:5555/sign", + json={"uri": uri, "data": data, "a1": a1, "web_session": web_session}) + signs = res.json() + return { + "x-s": signs["x-s"], + "x-t": signs["x-t"] + } + + +# pip install qrcode +if __name__ == '__main__': + xhs_client = XhsClient(sign=sign) + print(datetime.datetime.now()) + qr_res = xhs_client.get_qrcode_from_creator() + qr_id = qr_res["id"] + qr = qrcode.QRCode(version=1, error_correction=qrcode.ERROR_CORRECT_L, + box_size=50, + border=1) + qr.add_data(qr_res["url"]) + qr.make() + qr.print_ascii() + + while True: + check_qrcode = xhs_client.check_qrcode_from_creator(qr_id) + print(check_qrcode) + sleep(1) + if check_qrcode["status"] == 1: + ticket = check_qrcode["ticket"] + xhs_client.customer_login(ticket) + xhs_client.login_from_creator() + break + + print(json.dumps(xhs_client.get_self_info_from_creator(), ensure_ascii=False, indent=4)) diff --git a/tests/test_xhs.py b/tests/test_xhs.py index d615982..e113847 100644 --- a/tests/test_xhs.py +++ b/tests/test_xhs.py @@ -1,8 +1,12 @@ +import datetime + import pytest import requests +import xhs.help from xhs import FeedType, IPBlockError, XhsClient -from xhs.exception import SignError, DataFetchError +from xhs.exception import DataFetchError + from . import test_cookie from .utils import beauty_print @@ -10,7 +14,7 @@ @pytest.fixture def xhs_client(): def sign(uri, data, a1="", web_session=""): - res = requests.post("http://localhost:5005/sign", + res = requests.post("http://localhost:5555/sign", json={"uri": uri, "data": data, "a1": a1, "web_session": web_session}) signs = res.json() return { @@ -21,39 +25,44 @@ def sign(uri, data, a1="", web_session=""): return XhsClient(cookie=test_cookie, sign=sign) -# def test_xhs_client_init(): -# xhs_client = XhsClient() -# assert xhs_client +def test_xhs_client_init(): + xhs_client = XhsClient() + assert xhs_client + note = xhs_client.get_note_by_id_from_html("646837b9000000001300a4c3") + beauty_print(note) + assert note -# def test_cookie_setter_getter(): -# xhs_client = XhsClient() -# cd = xhs_client.cookie_dict -# beauty_print(cd) -# assert "web_session" in cd +def test_cookie_setter_getter(): + xhs_client = XhsClient() + cd = xhs_client.cookie_dict + beauty_print(cd) + assert "a1" in cd def test_external_sign_func(): - def sign(url, data=None, a1=""): + def sign(url, data=None, a1="", web_session=""): """signature url and data in here""" return {} - with pytest.raises(SignError): + with pytest.raises(DataFetchError): xhs_client = XhsClient(sign=sign) xhs_client.get_qrcode() def test_get_note_by_id(xhs_client: XhsClient): - note_id = "6413cf6b00000000270115b5" + # 13 我是DM发布了一篇小红书笔记，快来看吧！ 😆 F02MULzeoQVJ7YY 😆 http://xhslink.com/GQ0MHB，复制本条信息，打开【小红书】App查看精彩内容！ + note_id = "65682d4500000000380339a5" data = xhs_client.get_note_by_id(note_id) beauty_print(data) assert data["note_id"] == note_id def test_get_note_by_id_from_html(xhs_client: XhsClient): - note_id = "6413cf6b00000000270115b5" + note_id = "65a025ea000000001d03799b" data = xhs_client.get_note_by_id_from_html(note_id) beauty_print(data) + print(xhs.help.get_imgs_url_from_note(data)) assert data["note_id"] == note_id @@ -79,10 +88,19 @@ def test_get_self_info2(xhs_client: XhsClient): assert isinstance(data, dict) +def test_get_user_by_keyword(xhs_client: XhsClient): + keyword = "Python" + data = xhs_client.get_user_by_keyword(keyword, page=15) + # beauty_print(data) + print(len(data["users"])) + assert len(data["users"]) > 0 + + def test_get_user_info(xhs_client: XhsClient): user_id = "5ff0e6410000000001008400" data = xhs_client.get_user_info(user_id) basic_info = data["basic_info"] + print(datetime.datetime.now()) beauty_print(data) assert (basic_info["red_id"] == "hh06ovo" or basic_info["nickname"] == "小王不爱睡") @@ -101,6 +119,16 @@ def test_get_home_feed(xhs_client: XhsClient): assert len(data["items"]) > 0 +def test_comment_note(xhs_client: XhsClient): + data = xhs_client.comment_note("65ddbbda000000000700708a", "测试笔记评论功能") + beauty_print(data) + comment_id = data["comment"]["id"] + assert comment_id + delete_status = xhs_client.delete_note_comment("65ddbbda000000000700708a", comment_id) + beauty_print(delete_status) + assert bool(delete_status) + + def test_get_search_suggestion(xhs_client: XhsClient): res = xhs_client.get_search_suggestion("jvm") beauty_print(res) @@ -121,11 +149,10 @@ def test_get_user_notes(xhs_client: XhsClient): assert len(data["notes"]) > 0 -# @pytest.mark.skip(reason="it take much request and time") def test_get_user_all_notes(xhs_client: XhsClient): - user_id = "576e7b1d50c4b4045222de08" + user_id = "63273a77000000002303cc9b" notes = xhs_client.get_user_all_notes(user_id, 0) - beauty_print(notes) + assert len(notes) def test_get_note_comments(xhs_client: XhsClient): @@ -143,12 +170,10 @@ def test_get_note_sub_comments(xhs_client: XhsClient): assert len(comments["comments"]) > 0 -@pytest.mark.skip(reason="it take much request and time") def test_get_note_all_comments(xhs_client: XhsClient): - note_id = "63db8819000000001a01ead1" + note_id = "658120c0000000000602325f" note = xhs_client.get_note_by_id(note_id) comments_count = int(note["interact_info"]["comment_count"]) - print(comments_count) comments = xhs_client.get_note_all_comments(note_id) beauty_print(comments) assert len(comments) == comments_count @@ -167,27 +192,65 @@ def test_check_qrcode(xhs_client: XhsClient): assert "code_status" in data -@pytest.mark.skip() -def test_comment_note(xhs_client: XhsClient): - data = xhs_client.comment_note("642b96640000000014027cd2", "你最好说你在说你自己") +def test_get_qrcode_from_creator(xhs_client: XhsClient): + qrcode = xhs_client.get_qrcode() + beauty_print(qrcode) + + +def test_check_qrcode_from_creator(xhs_client: XhsClient): + creator = xhs_client.check_qrcode_from_creator("682141718111245888") + beauty_print(creator) + + +def test_login_from_creator(xhs_client: XhsClient): + xhs_client.login_from_creator() + beauty_print(xhs_client.cookie_dict) + + +def test_get_self_info_from_creator(xhs_client: XhsClient): + data = xhs_client.get_self_info_from_creator() beauty_print(data) - assert data["comment"]["id"] -@pytest.mark.skip() def test_comment_user(xhs_client: XhsClient): - data = xhs_client.comment_user("642b96640000000014027cd2", - "642f801000000000150037f8", - "我评论你了") + data = xhs_client.comment_user("65ddbbda000000000700708a", + "65f548eb000000001202e725", # 回复的评论 ID + "测试回复评论功能") beauty_print(data) - assert data["comment"]["id"] + comment_id = data["comment"]["id"] + assert comment_id + delete_status = xhs_client.delete_note_comment("65ddbbda000000000700708a", comment_id) + beauty_print(delete_status) + assert bool(delete_status) -@pytest.mark.skip() -def test_delete_comment(xhs_client: XhsClient): - data = xhs_client.delete_note_comment("642b96640000000014027cd2", - "642f801000000000150037f8") - beauty_print(data) +def test_follow_user(xhs_client: XhsClient): + follow_status = xhs_client.follow_user("5c3ef37200000000060137dc") + assert follow_status["fstatus"] == "follows" + + unfollow_status = xhs_client.unfollow_user("5c3ef37200000000060137dc") + assert unfollow_status["fstatus"] == "none" + + +def test_collect_note(xhs_client: XhsClient): + collect_status = xhs_client.collect_note("65ddbbda000000000700708a") + assert bool(collect_status) + uncollect_status = xhs_client.uncollect_note("65ddbbda000000000700708a") + assert bool(uncollect_status) + + +def test_like_note(xhs_client: XhsClient): + like_status = xhs_client.like_note("65ddbbda000000000700708a") + assert bool(like_status) + unlike_status = xhs_client.dislike_note("65ddbbda000000000700708a") + assert bool(unlike_status) + + +def test_like_comment(xhs_client: XhsClient): + like_status = xhs_client.like_comment("65ddbbda000000000700708a", "65f548eb000000001202e725") + assert bool(like_status) + unlike_status = xhs_client.dislike_comment("65ddbbda000000000700708a", "65f548eb000000001202e725") + assert bool(unlike_status) @pytest.mark.parametrize("note_id", [ @@ -205,23 +268,21 @@ def test_save_files_from_note_id(xhs_client: XhsClient, note_id: str): ]) @pytest.mark.skip() def test_save_files_from_note_id_invalid_title(xhs_client: XhsClient, note_id): - xhs_client.save_files_from_note_id(note_id, r"C:\Users\ReaJason\Desktop") + xhs_client.save_files_from_note_id(note_id, r"./tests/test_save_files") -@pytest.mark.skip() def test_get_user_collect_notes(xhs_client: XhsClient): notes = xhs_client.get_user_collect_notes( user_id="63273a77000000002303cc9b")["notes"] beauty_print(notes) - assert len(notes) == 1 + assert len(notes) -@pytest.mark.skip() def test_get_user_like_notes(xhs_client: XhsClient): notes = xhs_client.get_user_like_notes( user_id="63273a77000000002303cc9b")["notes"] beauty_print(notes) - assert len(notes) == 2 + assert len(notes) @pytest.mark.skip(reason="i don't want to block by ip") @@ -232,6 +293,7 @@ def test_ip_block_error(xhs_client: XhsClient): xhs_client.get_note_by_id(note_id) +@pytest.mark.skip(reason="current this func is useless") def test_activate(xhs_client: XhsClient): info = xhs_client.activate() beauty_print(info) @@ -262,25 +324,41 @@ def test_get_follow_notifications(xhs_client: XhsClient): assert len(mentions["message_list"]) +def test_get_notes_summary(xhs_client: XhsClient): + notes = xhs_client.get_notes_summary() + assert notes + + +def test_get_notes_statistics(xhs_client: XhsClient): + notes = xhs_client.get_notes_statistics() + beauty_print(notes) + + def test_get_upload_image_ids(xhs_client: XhsClient): - count = 5 - ids = xhs_client.get_upload_image_ids(count) - beauty_print(ids) - assert len(ids[0]["fileIds"]) == count + image_id, token = xhs_client.get_upload_files_permit("image") + upload_id = xhs_client.get_upload_id(image_id, token) + print(upload_id) + beauty_print(image_id) + assert image_id def test_upload_image(xhs_client: XhsClient): - ids = xhs_client.get_upload_image_ids(1) - file_info = ids[0] - file_id = file_info["fileIds"][0] - file_token = file_info["token"] - file_path = "/Users/reajason/Downloads/4538306CF3BDC215721FCC0532AF4D3D.jpg" - res = xhs_client.upload_image(file_id, file_token, file_path) + file_id, file_token = xhs_client.get_upload_files_permit("image") + file_path = "/Users/reajason/Downloads/wall/wallhaven-x6k21l.png" + res = xhs_client.upload_file(file_id, file_token, file_path) assert res.status_code == 200 - print(res.headers["X-Ros-Preview-Url"]) + assert res.headers["X-Ros-Preview-Url"] with pytest.raises(DataFetchError, match="file already exists"): - xhs_client.upload_image(file_id, file_token, file_path) + xhs_client.upload_file(file_id, file_token, file_path) + + +def test_upload_video(xhs_client: XhsClient): + file_id, file_token = xhs_client.get_upload_files_permit("video") + file_path = "/Users/reajason/Downloads/1.mp4" + res = xhs_client.upload_file(file_id, file_token, file_path, content_type="video/mp4") + assert res.status_code == 200 + assert res.headers["X-Ros-Preview-Url"] def test_get_suggest_topic(xhs_client: XhsClient): @@ -297,21 +375,23 @@ def test_get_suggest_ats(xhs_client: XhsClient): assert ats_keyword.upper() in ats[0]["user_base_dto"]["user_nickname"].upper() +@pytest.mark.skip() def test_create_simple_note(xhs_client: XhsClient): title = "我是标题" desc = "下面我说两点 \n 1. 第一点 \n 2. 第二点" images = [ - "/Users/reajason/Downloads/221686462282_.pic.png", + "/Users/reajason/Downloads/wall/wallhaven-x6k21l.png", ] note = xhs_client.create_image_note(title, desc, images, is_private=True, post_time="2023-07-25 23:59:59") beauty_print(note) +# @pytest.mark.skip() def test_create_note_with_ats_topics(xhs_client: XhsClient): title = "我是通过自动发布脚本发送的笔记" desc = "deployed by GitHub xhs， #Python[话题]# @ReaJason" files = [ - "/Users/reajason/Downloads/221686462282_.pic.png", + "/Users/reajason/ReaJason/wall/1687363223539@0.5x.jpg", ] # 可以通过 xhs_client.get_suggest_ats(ats_keyword) 接口获取用户数据 @@ -331,14 +411,17 @@ def test_create_note_with_ats_topics(xhs_client: XhsClient): beauty_print(note) +@pytest.mark.skip() def test_create_video_note(xhs_client: XhsClient): - note = xhs_client.create_video_note(title="123123", video_path="/Users/reajason/Downloads/2.mp4", desc="", + note = xhs_client.create_video_note(title="123123", desc="测试使用 github.com/reajason/xhs 发布视频笔记", + video_path="/Users/reajason/Downloads/1.mp4", is_private=True) beauty_print(note) +@pytest.mark.skip() def test_create_video_note_with_cover(xhs_client: XhsClient): - note = xhs_client.create_video_note(title="123123", video_path="/Users/reajason/Downloads/2.mp4", desc="", - cover_path="/Users/reajason/Downloads/221686462282_.pic.jpg", + note = xhs_client.create_video_note(title="123123", video_path="/Users/reajason/Downloads/1.mp4", desc="", + cover_path="/Users/reajason/Downloads/wall/wallhaven-x6k21l.png", is_private=True) beauty_print(note) diff --git a/xhs/core.py b/xhs/core.py index a5a2cf0..2f4d001 100644 --- a/xhs/core.py +++ b/xhs/core.py @@ -101,6 +101,7 @@ def __init__( self.external_sign = sign self._host = "https://edith.xiaohongshu.com" self._creator_host = "https://creator.xiaohongshu.com" + self._customer_host = "https://customer.xiaohongshu.com" self.home = "https://www.xiaohongshu.com" self.user_agent = user_agent or ( "Mozilla/5.0 " @@ -131,8 +132,8 @@ def cookie_dict(self): def session(self): return self.__session - def _pre_headers(self, url: str, data=None, is_creator: bool = False): - if is_creator: + def _pre_headers(self, url: str, data=None, quick_sign: bool = False): + if quick_sign: signs = sign(url, data, a1=self.cookie_dict.get("a1")) self.__session.headers.update({"x-s": signs["x-s"]}) self.__session.headers.update({"x-t": signs["x-t"]}) @@ -157,7 +158,6 @@ def request(self, method, url, **kwargs): data = response.json() except json.decoder.JSONDecodeError: return response - print(data) if response.status_code == 471 or response.status_code == 461: # someday someone maybe will bypass captcha verify_type = response.headers['Verifytype'] @@ -174,21 +174,34 @@ def request(self, method, url, **kwargs): else: raise DataFetchError(data, response=response) - def get(self, uri: str, params=None, is_creator: bool = False, **kwargs): + def get(self, uri: str, params=None, is_creator: bool = False, is_customer: bool = False, **kwargs): final_uri = uri if isinstance(params, dict): final_uri = f"{uri}?" f"{'&'.join([f'{k}={v}' for k, v in params.items()])}" - self._pre_headers(final_uri, is_creator=is_creator) - return self.request(method="GET", url=f"{self._creator_host if is_creator else self._host}{final_uri}", + self._pre_headers(final_uri, quick_sign=is_creator or is_customer) + endpoint = self._host + if is_customer: + endpoint = self._customer_host + elif is_creator: + endpoint = self._creator_host + return self.request(method="GET", url=f"{endpoint}{final_uri}", **kwargs) - def post(self, uri: str, data: dict, is_creator: bool = False, **kwargs): - self._pre_headers(uri, data, is_creator=is_creator) + def post(self, uri: str, data: dict | None, is_creator: bool = False, is_customer: bool = False, **kwargs): json_str = json.dumps(data, separators=(",", ":"), ensure_ascii=False) - return self.request( - method="POST", url=f"{self._creator_host if is_creator else self._host}{uri}", data=json_str.encode(), - **kwargs - ) + self._pre_headers(uri, data, quick_sign=is_creator or is_customer) + endpoint = self._host + if is_customer: + endpoint = self._customer_host + elif is_creator: + endpoint = self._creator_host + if data: + return self.request( + method="POST", url=f"{endpoint}{uri}", data=json_str.encode(), + **kwargs + ) + else: + return self.request(method="POST", url=f"{endpoint}{uri}", **kwargs) def get_note_by_id(self, note_id: str): """ @@ -323,6 +336,13 @@ def get_self_info2(self): uri = "/api/sns/web/v2/user/me" return self.get(uri) + def get_self_info_from_creator(self): + uri = "/api/galaxy/creator/home/personal_info" + headers = { + "referer": "https://creator.xiaohongshu.com/creator/home" + } + return self.get(uri, is_creator=True, headers=headers) + def get_user_by_keyword(self, keyword: str, page: int = 1, page_size: int = 20, ): @@ -653,6 +673,36 @@ def login_code(self, phone: str, mobile_token: str, zone: str = 86): data = {"mobile_token": mobile_token, "zone": zone, "phone": phone} return self.post(uri, data) + def get_qrcode_from_creator(self): + uri = "/api/cas/customer/web/qr-code" + data = {"service": "https://creator.xiaohongshu.com"} + return self.post(uri, data, is_customer=True) + + def check_qrcode_from_creator(self, qr_code_id: str): + uri = "/api/cas/customer/web/qr-code" + params = { + "service": "https://creator.xiaohongshu.com", + "qr_code_id": qr_code_id, + } + return self.get(uri, params, is_customer=True) + + def customer_login(self, ticket: str): + uri = "/sso/customer_login" + data = { + "ticket": ticket, + "login_service": "https://creator.xiaohongshu.com", + "subsystem_alias": "creator", + "set_global_domain": True + } + return self.post(uri, data, is_creator=True) + + def login_from_creator(self): + uri = "/api/galaxy/user/cas/login" + headers = { + "referer": "https://creator.xiaohongshu.com/login" + } + return self.post(uri, None, is_creator=True, headers=headers) + def get_user_collect_notes(self, user_id: str, num: int = 30, cursor: str = ""): uri = "/api/sns/web/v2/note/collect/page" params = {"user_id": user_id, "num": num, "cursor": cursor}