datasources.tiktok.search_tiktok

  1"""
  2Import scraped TikTok 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, timezone
  8from urllib.parse import urlparse, parse_qs
  9
 10from backend.lib.search import Search
 11from common.lib.item_mapping import MappedItem
 12from common.lib.helpers import normalize_url_encoding
 13
 14
 15class SearchTikTok(Search):
 16    """
 17    Import scraped TikTok data
 18    """
 19    type = "tiktok-search"  # job ID
 20    category = "Search"  # category
 21    title = "Import scraped Tiktok data"  # title displayed in UI
 22    description = "Import Tiktok data collected with an external tool such as Zeeschuimer."  # description displayed in UI
 23    extension = "ndjson"  # extension of result file, used internally and in UI
 24    is_from_zeeschuimer = True
 25
 26    # not available as a processor for existing datasets
 27    accepts = [None]
 28    references = [
 29        "[Zeeschuimer browser extension](https://github.com/digitalmethodsinitiative/zeeschuimer)",
 30        "[Worksheet: Capturing TikTok data with Zeeschuimer and 4CAT](https://tinyurl.com/nmrw-zeeschuimer-tiktok)"
 31    ]
 32
 33    def get_items(self, query):
 34        """
 35        Run custom search
 36
 37        Not available for TikTok
 38        """
 39        raise NotImplementedError("TikTok datasets can only be created by importing data from elsewhere")
 40
 41    @staticmethod
 42    def map_item(post):
 43        # Zeeschuimer metadata
 44        metadata = post.get("__import_meta", {})
 45
 46        challenges = [challenge["title"] for challenge in post.get("challenges", [])]
 47
 48        hashtags = [extra["hashtagName"] for extra in post.get("textExtra", []) if
 49                    "hashtagName" in extra and extra["hashtagName"]]
 50
 51        labels = ",".join(post["diversificationLabels"]) if type(post.get("diversificationLabels")) is list else ""
 52
 53        if type(post.get("author")) is dict:
 54            # from intercepted API response
 55            user_nickname = post["author"]["uniqueId"]
 56            user_fullname = post["author"]["nickname"]
 57            user_thumbnail = post["author"].get("avatarThumb", "")
 58        elif post.get("author"):
 59            # from embedded JSON object
 60            user_nickname = post["author"]
 61            user_fullname = post["nickname"]
 62            user_thumbnail = ""
 63        else:
 64            user_nickname = ""
 65            user_fullname = ""
 66            user_thumbnail = ""
 67
 68        # there are various thumbnail URLs, some of them expire later than
 69        # others. Try to get the highest-resolution one that hasn't expired
 70        # yet
 71        thumbnail_options = []
 72
 73        if post["video"].get("shareCover"):
 74            thumbnail_options.append(post["video"]["shareCover"].pop())
 75
 76        if post["video"].get("cover"):
 77            thumbnail_options.append(post["video"]["cover"])
 78
 79        now = int(datetime.now(tz=timezone.utc).timestamp())
 80        thumbnail_url = [url for url in thumbnail_options if int(parse_qs(urlparse(url).query).get("x-expires", [now])[0]) >= now]
 81        thumbnail_url = thumbnail_url.pop() if thumbnail_url else ""
 82
 83        return MappedItem({
 84            "collected_from_url": normalize_url_encoding(metadata.get("source_platform_url")) if metadata.get("source_platform_url") else "",
 85            "id": post["id"],
 86            "thread_id": post["id"],
 87            "author": user_nickname,
 88            "author_full": user_fullname,
 89            "author_followers": post.get("authorStats", {}).get("followerCount", ""),
 90            "author_likes": post.get("authorStats", {}).get("diggCount", ""),
 91            "author_videos": post.get("authorStats", {}).get("videoCount", ""),
 92            "author_avatar": user_thumbnail,
 93            "body": post["desc"],
 94            "stickers": "\n".join(" ".join(s["stickerText"]) for s in post.get("stickersOnItem", [])),
 95            "timestamp": datetime.utcfromtimestamp(int(post["createTime"])).strftime('%Y-%m-%d %H:%M:%S'),
 96            "unix_timestamp": int(post["createTime"]),
 97            "is_duet": "yes" if (post.get("duetInfo", {}).get("duetFromId") != "0" if post.get("duetInfo", {}) else False) else "no",
 98            "is_ad": "yes" if post.get("isAd", False) else "no",
 99            "is_paid_partnership": "yes" if post.get("adAuthorization") else "no",
100            "is_sensitive": "yes" if post.get("maskType") == 3 else "no",
101            "is_photosensitive": "yes" if post.get("maskType") == 4 else "no",
102            "music_name": post["music"]["title"],
103            "music_id": post["music"]["id"],
104            "music_url": post["music"].get("playUrl", ""),
105            "music_thumbnail": post["music"].get("coverLarge", ""),
106            "music_author": post["music"].get("authorName", ""),
107            "video_url": post["video"].get("downloadAddr", ""),
108            "tiktok_url": "https://www.tiktok.com/@%s/video/%s" % (user_nickname, post["id"]),
109            "thumbnail_url": thumbnail_url,
110            "likes": post["stats"]["diggCount"],
111            "comments": post["stats"]["commentCount"],
112            "shares": post["stats"]["shareCount"],
113            "plays": post["stats"]["playCount"],
114            "hashtags": ",".join(hashtags),
115            "challenges": ",".join(challenges),
116            "diversification_labels": labels,
117            "location_created": post.get("locationCreated", ""),
118            "effects": ",".join([e["name"] for e in post.get("effectStickers", [])]),
119            "warning": ",".join([w["text"] for w in post.get("warnInfo", [])])
120        })