datasources.threads.search_threads

 1"""
 2Import scraped Threads data
 3
 4It's prohibitively difficult to scrape data from Threads within 4CAT itself due
 5to its aggressive rate limiting. Instead, import data collected elsewhere.
 6"""
 7from datetime import datetime
 8from urllib.parse import urlparse, parse_qs
 9import re
10
11from backend.lib.search import Search
12from common.lib.item_mapping import MappedItem
13from common.lib.helpers import normalize_url_encoding
14
15
16class SearchThreads(Search):
17    """
18    Import scraped Threads data
19    """
20    type = "threads-search"  # job ID
21    category = "Search"  # category
22    title = "Import scraped Threads data"  # title displayed in UI
23    description = "Import Threads data collected with an external tool such as Zeeschuimer."  # description displayed in UI
24    extension = "ndjson"  # extension of result file, used internally and in UI
25    is_from_zeeschuimer = True
26
27    # not available as a processor for existing datasets
28    accepts = [None]
29    references = [
30        "[Zeeschuimer browser extension](https://github.com/digitalmethodsinitiative/zeeschuimer)",
31        "[Worksheet: Capturing TikTok data with Zeeschuimer and 4CAT](https://tinyurl.com/nmrw-zeeschuimer-tiktok)"
32    ]
33
34    def get_items(self, query):
35        """
36        Run custom search
37
38        Not available for 9gag
39        """
40        raise NotImplementedError("Threads datasets can only be created by importing data from elsewhere")
41
42    @staticmethod
43    def map_item(post):
44        post_timestamp = datetime.fromtimestamp(post["taken_at"])
45
46        if post["carousel_media"]:
47            image_urls = [c["image_versions2"]["candidates"].pop(0)["url"] for c in post["carousel_media"] if c["image_versions2"] and c["image_versions2"].get("candidates")]
48            video_urls = [c["video_versions"].pop(0)["url"] for c in post["carousel_media"] if c["video_versions"]]
49
50        else:
51            image_urls = [post["image_versions2"]["candidates"].pop(0)["url"]] if post["image_versions2"] and post["image_versions2"].get("candidates") else []
52            video_urls = [post["video_versions"].pop(0)["url"]] if post["video_versions"] else []
53        audio_url = post["audio"]["audio_src"] if post["audio"] else ""
54        
55        linked_url = ""
56        link_thumbnail = ""
57        if post["text_post_app_info"].get("link_preview_attachment"):
58            linked_url = post["text_post_app_info"]["link_preview_attachment"]["url"]
59            parsed_url = parse_qs(urlparse(linked_url).query).get("u")
60            if parsed_url:
61                linked_url = parsed_url.pop()
62                link_thumbnail = post["text_post_app_info"]["link_preview_attachment"].get("image_url")
63            else:
64                link_thumbnail = linked_url
65
66        return MappedItem({
67            "collected_from_url": normalize_url_encoding(post.get("__import_meta", {}).get("source_platform_url", "")),  # Zeeschuimer metadata
68            "id": post["code"],
69            "thread_id": post["code"],
70            "url": f"https://www.threads.com/@{post['user']['username']}/post/{post['code']}",
71            "body": post["caption"]["text"] if post["caption"] else "",
72            "timestamp": post_timestamp.strftime("%Y-%m-%d %H:%M:%S"),
73            "author": post["user"]["username"],
74            "author_is_verified": "yes" if post["user"].get("is_verified") else "no",
75            "author_avatar": post["user"].get("profile_pic_url"),
76            "image_url": ",".join(image_urls),
77            "video_url": ",".join(video_urls),
78            "audio_url": audio_url,
79            "link_url": linked_url,
80            "link_thumbnail_url": link_thumbnail if link_thumbnail else "",
81            "is_paid_partnership": "yes" if post["is_paid_partnership"] else "no",
82            "likes": post["like_count"],
83            "reposts": post["text_post_app_info"]["repost_count"],
84            "replies": post["text_post_app_info"]["direct_reply_count"],
85            "quotes": post["text_post_app_info"]["quote_count"],
86            "hashtags": ",".join(re.findall(r"#([^\s!@#$%ˆ&*()_+{}:\"|<>?\[\];'\,./`~']+)", post["caption"]["text"])) if post["caption"] else "",
87            "unix_timestamp": int(post_timestamp.timestamp()),
88        })