-
Notifications
You must be signed in to change notification settings - Fork 0
/
jobs.py
123 lines (92 loc) · 3.93 KB
/
jobs.py
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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
from datetime import datetime
import logging
from modal import functions, Image, Stub, Secret
from api import twitter_transcribe, verity
from cap import CapClient
from heatmap import create_heatmap, delete_heatmap
DEFAULT_TIMEOUT=700
FORMAT = '%(asctime)s: %(message)s'
logging.basicConfig(format=FORMAT, level=logging.INFO)
class ScorelessException(Exception):
pass
image = (
Image.debian_slim(python_version="3.10")
.pip_install_from_requirements("requirements.txt")
)
stub = Stub(
"cap-jobs",
image=image,
secrets=(
Secret.from_dotenv(".env"),
),
)
@stub.function(timeout=DEFAULT_TIMEOUT)
async def verity_job(tweet_id: str, mention_tweet_id: str):
print("Transcribing...")
transcription = twitter_transcribe(tweet_id)
print(f"Transcription Result: {transcription}\n----------")
print("Fact checking...")
fact_check = verity(tweet_id, transcription)
print(f'Fact Check Output: {fact_check}')
print('Tweeting...')
cap = CapClient()
response_tweet = cap.tweet(text=_parse_fact_check_tweet(tweet_id, fact_check), in_reply_to_tweet_id=mention_tweet_id)
response_tweet_id = response_tweet.data['id']
print(f"Tweeted! See https://twitter.com/twitter/status/{response_tweet_id}")
return fact_check, response_tweet_id
@stub.function(timeout=DEFAULT_TIMEOUT)
async def poller(id: str, mention_tweet_id: str):
logging.info("Capper poller starting...")
then = datetime.now()
cap = CapClient()
logging.info(f"Getting results for job(id={id}, mention_tweet_id={mention_tweet_id})...")
function_call = functions.FunctionCall.from_id(id)
try:
result = function_call.get(timeout=DEFAULT_TIMEOUT)
except TimeoutError as e:
logging.error(f"Polling job(id={id}, mention_tweet_id={mention_tweet_id}) has timed out.")
raise e
logging.info(f"Job(id={id}, mention_tweet_id={mention_tweet_id}, completion_length=({datetime.now() - then}s)) completed with this result:\n{result}")
scores = result.get("scores")
segmented_predictions = result.get("segmented_predictions")
if not scores:
raise ScorelessException(f"Job(id={id}, mention_tweet_id={mention_tweet_id}) returned empty scores.")
logging.info(f"Job(id={id}, mention_tweet_id={mention_tweet_id}) posting to twitter...")
media_ids = []
for sg in segmented_predictions:
create_heatmap(sg)
media = cap.api.media_upload("heatmap.png")
media_ids.append(media.media_id)
delete_heatmap()
tweet1, tweet2 = _parse_voice_clone_tweet(scores[0])
created_tweet = cap.tweet(text=tweet1, in_reply_to_tweet_id=mention_tweet_id,)
created_tweet_id = created_tweet.data['id']
cap.tweet(text=tweet2, in_reply_to_tweet_id=created_tweet_id, media_ids=media_ids)
return
def _parse_voice_clone_tweet(score: int):
icon = "🟢"
message = "DeepTrust Alpha did not detect generated speech. Yay! 🎉"
percent = round(score * 100, 2)
if 50 <= percent <= 85:
icon = "🟡"
message = "DeepTrust Alpha detects some traces that resemble generated speech. Tread carefully. 👀"
elif percent > 85:
icon = "🔴"
message = "🚨 CAPPER ALERT 🚨\nDeepTrust Alpha is certain there is generated speech. We detect a capper. 🧢"
tweet1 = f"""Speech Analysis Complete!
{icon} {message}
============
Check out the tweet below for more details via heatmap!
"""
# tweet2 = f"""Deepfake Speech Heatmap Generated By ䷼ @deeptrustAI
# ============
# Disclaimer: DeepTrust Speech is in early alpha. Research is still undergoing. Results may vary.
# """
tweet2 = "Deepfake Speech Heatmap generated by ䷼ @deeptrustAI"
return tweet1, tweet2
def _parse_fact_check_tweet(id: str, fact_check: str):
return f"""AI Fact Checking Complete!
Here's what Verity had to say:
"{fact_check[:150]}{"..." if len(fact_check) > 150 else ""}"
See more details at https://app.deeptrustai.com/verity/twitter/{id}
"""