cap/bot.py at main · deeptrust-ai/cap · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
from datetime import datetime
import logging
import time
import os

from modal import Function
from tweepy.errors import TooManyRequests
import pytz

from cap import CapClient, CapType, ValidMention

FORMAT = "%(asctime)s: %(message)s"
logging.basicConfig(format=FORMAT, level=logging.INFO)

SLEEP_TIME = int(os.environ.get("SLEEP_TIME", "30"))

cap = CapClient()
logging.info("Capper Checker starting...")


def launch_deepfake_job(mention: ValidMention) -> None:
    # will launch a modal job to fact check tweet (API request)
    # TODO: Use api to launch tweet job
    logging.info(
        f"Launching twitter predict job for mention(id={mention.mention.id})..."
    )
    twitter_predict = Function.lookup("rawnet-predict-jobs", "twitter_predict")
    mention_tweet = mention.mention
    parent_tweet = mention.parent_tweet.data
    predict_job = twitter_predict.spawn(parent_tweet.id, "ss")
    job_id = predict_job.object_id

    # launch a poller job to update with tweet (modal function launch)
    logging.info(f"Launching poller(job_id={job_id})...")
    poller = Function.lookup("cap-jobs", "poller")
    poller.spawn(job_id, mention_tweet.id)

def launch_fact_check_job(mention: ValidMention) -> None:
    logging.info(f"Launching a fact check job for mention(id{mention.mention.id})...")

    # the tweet to analyze
    parent_tweet_id = mention.parent_tweet.data.id

    # the tweet to respond to
    mention_tweet_id = mention.mention.id

    verity_job = Function.lookup("cap-jobs", "verity_job")
    verity_job.spawn(parent_tweet_id, mention_tweet_id)

start_time = datetime.now(tz=pytz.utc)
while True:
    # Get the mentions for your bot's user ID.
    logging.info(f"Getting mentions starting from time ({start_time})")

    try:
        mentions = cap.get_mentions(start_time=start_time)
    except TooManyRequests:
        sleep_time = SLEEP_TIME * 5
        logging.error(f"TooManyRequests thrown. Sleeping for {sleep_time}s.")

        time.sleep(sleep_time)
        continue

    for mention in mentions:
        if mention.cap_type == CapType.DEEPFAKE:
            print(
                f"Starting deepfake job for mention(tweet_id={mention.mention.id})..."
            )
            launch_deepfake_job(mention)
        elif mention.cap_type == CapType.FACTCHECK:
            print(
                f"Starting factcheck job for mention(tweet_id={mention.mention.id})..."
            )
            launch_fact_check_job(mention)

    # update start time
    start_time = datetime.now(tz=pytz.utc)

    # sleep
    logging.info(f"Sleeping for {SLEEP_TIME}s...")
    time.sleep(SLEEP_TIME)