datasources.tiktok_comments.search_tiktok_comments

 1"""
 2Import scraped TikTok comment data
 3
 4It's prohibitively difficult to scrape data from TikTok 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
11
12
13class SearchTikTokComments(Search):
14    """
15    Import scraped TikTok comment data
16    """
17    type = "tiktok-comments-search"  # job ID
18    category = "Search"  # category
19    title = "Import scraped Tiktok comment data"  # title displayed in UI
20    description = "Import Tiktok comment 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 TikTok comments
36        """
37        raise NotImplementedError("TikTok comment datasets can only be created by importing data from elsewhere")
38
39    @staticmethod
40    def map_item(item):
41        item_datetime = datetime.fromtimestamp(item["create_time"]).strftime("%Y-%m-%d %H:%M:%S")
42        thread_id = item["aweme_id"] if item["reply_id"] == "0" else item["reply_id"]
43        avatar_url = item["user"]["avatar_thumb"]["url_list"][0]
44
45        return MappedItem({
46            "id": item["cid"],
47            "thread_id": thread_id,
48            "author": item["user"]["unique_id"],
49            "author_full": item["user"]["nickname"],
50            "author_avatar_url": avatar_url,
51            "body": item["text"],
52            "timestamp": item_datetime,
53            "unix_timestamp": item["create_time"],
54            "likes": item["digg_count"],
55            "replies": item.get("reply_comment_total", 0),
56            "post_id": item["aweme_id"],
57            "post_url": item["share_info"]["url"].split(".html")[0],
58            "post_body": item["share_info"]["title"],
59            "comment_url": item["share_info"]["url"],
60            "is_liked_by_post_author": "yes" if bool(item.get("author_pin")) else "no",
61            "is_sticky": "yes" if bool(item["stick_position"]) else "no",
62            "is_comment_on_comment": "no" if bool(item["reply_id"] == "0") else "yes",
63            "language_guess": item["comment_language"]
64        })