datasources.xiaohongshu.search_rednote

  1"""
  2Import scraped RedNote data
  3
  4It's prohibitively difficult to scrape data from RedNote within 4CAT itself due
  5to its aggressive rate limiting. Instead, import data collected elsewhere.
  6"""
  7from datetime import datetime
  8
  9from backend.lib.search import Search
 10from common.lib.item_mapping import MappedItem, MissingMappedField
 11
 12
 13class SearchRedNote(Search):
 14    """
 15    Import scraped RedNote/Xiaohongshu/XSH data
 16    """
 17    type = "xiaohongshu-search"  # job ID
 18    category = "Search"  # category
 19    title = "Import scraped RedNote data"  # title displayed in UI
 20    description = "Import RedNote data collected with an external tool such as Zeeschuimer."  # description displayed in UI
 21    extension = "ndjson"  # extension of result file, used internally and in UI
 22    is_from_zeeschuimer = True
 23
 24    # not available as a processor for existing datasets
 25    accepts = [None]
 26    references = [
 27        "[Zeeschuimer browser extension](https://github.com/digitalmethodsinitiative/zeeschuimer)",
 28        "[Worksheet: Capturing TikTok data with Zeeschuimer and 4CAT](https://tinyurl.com/nmrw-zeeschuimer-tiktok)"
 29    ]
 30
 31    def get_items(self, query):
 32        """
 33        Run custom search
 34
 35        Not available for RedNote
 36        """
 37        raise NotImplementedError("RedNote/Xiaohongshu datasets can only be created by importing data from elsewhere")
 38
 39
 40    @staticmethod
 41    def map_item(post):
 42        """
 43        Map XSH object to 4CAT item
 44
 45        Depending on whether the object was captured from JSON or HTML, treat it
 46        differently. A lot of data is missing from HTML objects.
 47
 48        :param post:
 49        :return:
 50        """
 51        if post.get("_zs-origin") == "html":
 52            return SearchRedNote.map_item_from_html(post)
 53        else:
 54            if "note" in post:
 55                return SearchRedNote.map_item_from_json_embedded(post)
 56            else:
 57                return SearchRedNote.map_item_from_json_api_explore(post)
 58
 59    @staticmethod
 60    def map_item_from_json_api_explore(post):
 61        """
 62        Map API-sourced XSH object to 4CAT item
 63
 64        Most straightforward - JSON objects from the XSH web API, which do
 65        however not always contain the same fields.
 66
 67        :param dict post:
 68        :return MappedItem:
 69        """
 70        item = post["note_card"] if post.get("type") != "video" else post
 71        item_id = post.get("id", post.get("note_id"))
 72
 73
 74        # Images
 75        images = []
 76        if item.get("image_list"):
 77            for image in item["image_list"]:
 78                if "url_default" in image and image["url_default"]:
 79                    images.append(image["url_default"])
 80                elif "info_list" in image and image["info_list"]:
 81                    for img_info in image["info_list"]:
 82                        found = False
 83                        if img_info.get("image_scene") == "WB_DFT":
 84                            images.append(img_info["url"])
 85                            found = True
 86                            break
 87                    if not found:
 88                        images.append(image["info_list"][0]["url"])
 89        elif item.get("cover"):
 90            images.append(item["cover"]["url_default"])
 91        else:
 92            # no image found;
 93            images = MissingMappedField("")       
 94
 95        # permalinks need this token to work, else you get a 404 not found
 96        xsec_bit = f"?xsec_token={post['xsec_token']}" if post.get("xsec_token") else ""
 97        if item.get("video", {}).get("media"):
 98            video_url = item["video"]["media"]["stream"]["h264"][0]["master_url"]
 99        else:
100            video_url = MissingMappedField("")
101
102        timestamp = item.get("time", None)
103        return MappedItem({
104            "id": item_id,
105            "thread_id": item_id,
106            "url": f"https://www.xiaohongshu.com/explore/{post['id']}{xsec_bit}",
107            "title": item.get("display_title", ""),
108            "body": item.get("desc", "") if "desc" in item else MissingMappedField(""),
109            "timestamp": datetime.fromtimestamp(timestamp / 1000).strftime("%Y-%m-%d %H:%M:%S") if timestamp else MissingMappedField(""),
110            "author": item["user"]["nickname"],
111            "author_avatar_url": item["user"]["avatar"],
112            "image_urls": ",".join(images) if type(images) is list else images,
113            "video_url": video_url,
114            # only available when loading an individual post page, so skip
115            # "tags": ",".join(t["name"] for t in item["tag_list"]),
116            "likes": item["interact_info"]["liked_count"],
117            # "collects": item["interact_info"]["collected_count"],
118            # "comments": item["interact_info"]["comment_count"],
119            # "shares": item["interact_info"]["share_count"],
120            "unix_timestamp": int(timestamp / 1000) if timestamp else MissingMappedField(""),
121        })
122
123    @staticmethod
124    def map_item_from_json_embedded(item):
125        """
126        Map JSON object from an XHS HTML page
127
128        JSON objects from the HTML are formatted slightly differently, mostly
129        in that they use camelCase instead of underscores, but we can also
130        make a few more assumptions about the data
131
132        :param dict item:
133        :return MappedItem:
134        """
135        note = item["note"]
136        image = note["imageList"][0]["urlDefault"]
137        # permalinks need this token to work, else you get a 404 not found
138        xsec_bit = f"?xsec_token={note['xsecToken']}"
139        timestamp = item.get("time", None)
140
141        return MappedItem({
142            "id": item["id"],
143            "thread_id": item["id"],
144            "url": f"https://www.xiaohongshu.com/explore/{item['id']}{xsec_bit}",
145            "title": note.get("title", ""),
146            "body": note.get("desc", "") if "desc" in note else MissingMappedField(""),
147            "timestamp": datetime.fromtimestamp(timestamp / 1000).strftime("%Y-%m-%d %H:%M:%S") if timestamp else MissingMappedField(""),
148            "author": note["user"]["nickname"],
149            "author_avatar_url": note["user"]["avatar"],
150            "image_url": image,
151            "video_url": MissingMappedField(""),
152            # only available when loading an individual post page, so skip
153            # "tags": ",".join(t["name"] for t in item["tag_list"]),
154            "likes": item["interactInfo"]["likedCount"],
155            # "collects": item["interact_info"]["collected_count"],
156            # "comments": item["interact_info"]["comment_count"],
157            # "shares": item["interact_info"]["share_count"],
158            "unix_timestamp": int(timestamp / 1000) if timestamp else MissingMappedField(""),
159        })
160
161    def map_item_from_html(item):
162        """
163        Map pre-mapped item
164
165        These have been mapped by Zeeschuimer from the page HTML and contain
166        less data than JSON objects (but enough to be useful in some cases).
167
168        :param dict item:
169        :return MappedItem:
170        """
171        return MappedItem({
172            "id": item["id"],
173            "thread_id": item["id"],
174            "url": f"https://www.xiaohongshu.com{item['url']}",
175            "title": item["title"],
176            "body": MissingMappedField(""),
177            "timestamp": MissingMappedField(""),
178            "author": item["author_name"],
179            "author_avatar_url": item["author_avatar_url"],
180            "image_url": item["thumbnail_url"],
181            "video_url": MissingMappedField(""),
182            # "tags": MissingMappedField(""),
183            "likes": item["likes"],
184            # "collects": MissingMappedField(""),
185            # "comments": MissingMappedField(""),
186            # "shares": MissingMappedField(""),
187            "unix_timestamp": MissingMappedField(""),
188        })