Initial translation implementation #273

This commit is contained in:
Dylan
2026-04-18 20:11:03 +01:00
parent 4794857c0a
commit 1a6f424e69
8 changed files with 130 additions and 47 deletions
+11 -2
View File
@@ -8,11 +8,20 @@ def tweetDataToActivity(tweetData,embedIndex = -1):
if tweetData['replyingTo'] is not None: if tweetData['replyingTo'] is not None:
content += f"<blockquote>↪️ <i>Replying to @{tweetData['replyingTo']}</i></blockquote>" content += f"<blockquote>↪️ <i>Replying to @{tweetData['replyingTo']}</i></blockquote>"
content+=f"<p>{tweetData['text']}</p>"
mainText = tweetData['text']
if tweetData["translation"] is not None:
content += f"<blockquote>🌐 <i>{tweetData['translation']['source_language'].upper()}{tweetData['translation']['destination_language'].upper()}</i></blockquote>"
mainText=tweetData['translation']["text"]
content+=f"<p>{mainText}</p>"
attachments=[] attachments=[]
if tweetData['qrt'] is not None: if tweetData['qrt'] is not None:
content += f"<blockquote><b>QRT: <a href=\"{tweetData['qrtURL']}\">{tweetData['qrt']['user_screen_name']}</a></b><br>{tweetData['qrt']['text']}</blockquote>" qrtText = tweetData['qrt']['text']
if "translation" in tweetData['qrt'] and tweetData['qrt']["translation"] is not None:
qrtText = tweetData['qrt']["translation"]["text"]
content += f"<blockquote><b>QRT: <a href=\"{tweetData['qrtURL']}\">{tweetData['qrt']['user_screen_name']}</a></b><br>{qrtText}</blockquote>"
if tweetData['pollData'] is not None: if tweetData['pollData'] is not None:
content += f"<p>{msgs.genPollDisplay(tweetData['pollData'])}</p>" content += f"<p>{msgs.genPollDisplay(tweetData['pollData'])}</p>"
content += "</p>" content += "</p>"
+12 -3
View File
@@ -17,8 +17,11 @@ def genLikesDisplay(vnf):
return ("💖 " + numerize.numerize(vnf['likes'])) return ("💖 " + numerize.numerize(vnf['likes']))
def genQrtDisplay(qrt): def genQrtDisplay(qrt):
text = qrt['text']
if "translation" in qrt:
text = qrt["translation"]["text"]
verifiedCheck = "☑️" if ('verified' in qrt and qrt['verified']) else "" verifiedCheck = "☑️" if ('verified' in qrt and qrt['verified']) else ""
return ("\n\n【QRT of " + qrt['user_name'] + " (@" + qrt['user_screen_name'] + ")"+ verifiedCheck+":】\n\n'" + qrt['text'] + "'") return ("\n\n【QRT of " + qrt['user_name'] + " (@" + qrt['user_screen_name'] + ")"+ verifiedCheck+":】\n\n'" + text + "'")
def genPollDisplay(poll): def genPollDisplay(poll):
pctSplit=10 pctSplit=10
@@ -27,6 +30,9 @@ def genPollDisplay(poll):
output+=choice["name"]+"\n"+(""*int(choice["percent"]/pctSplit)) +" "+str(choice["percent"])+"%\n" output+=choice["name"]+"\n"+(""*int(choice["percent"]/pctSplit)) +" "+str(choice["percent"])+"%\n"
return output return output
def genTranslationDisplay(translation):
return f"【TL "+translation["source_language"]+""+translation["destination_language"]+"\n\n"+translation["text"]
# formats the top text of the embed # formats the top text of the embed
def formatProvider(base,vnf): def formatProvider(base,vnf):
finalText = base finalText = base
@@ -38,13 +44,16 @@ def formatProvider(base,vnf):
finalText = base finalText = base
return finalText return finalText
def formatEmbedDesc(type,body,qrt,pollData): def formatEmbedDesc(type,body,qrt,pollData,translation):
# Trim the embed description to 248 characters, prioritizing poll and likes # Trim the embed description to 248 characters, prioritizing poll and likes
qrtType=None qrtType=None
if qrt!=None: if qrt!=None:
qrtType="Text" qrtType="Text"
if translation is not None:
body = genTranslationDisplay(translation)
limit = videoDescLimit if type=="Video" or (qrt!=None and (qrtType=="Video")) else tweetDescLimit limit = videoDescLimit if type=="Video" or (qrt!=None and (qrtType=="Video")) else tweetDescLimit
output = "" output = ""
@@ -73,6 +82,6 @@ def formatEmbedDesc(type,body,qrt,pollData):
diff = len(output)-limit diff = len(output)-limit
# remove the characters from body, add ellipsis # remove the characters from body, add ellipsis
body = body[:-(diff+1)]+"" body = body[:-(diff+1)]+""
return formatEmbedDesc(type,body,qrt,pollData) return formatEmbedDesc(type,body,qrt,pollData,None)
else: else:
return output return output
+1 -1
View File
@@ -8,4 +8,4 @@ Werkzeug==2.3.7
numerize==0.12 numerize==0.12
oauthlib==3.2.2 oauthlib==3.2.2
PyRTF3==0.47.5 PyRTF3==0.47.5
XClientTransaction==0.0.2 XClientTransaction==1.0.2
+7
View File
@@ -0,0 +1,7 @@
validLanguages = ['AF', 'AN', 'AR', 'AS', 'AY', 'AZ', 'BA', 'BE', 'BG', 'BN', 'BR', 'BS', 'CA', 'CS', 'CY', 'DA', 'DE', 'DE-DE', 'EL', 'EN', 'EO', 'ES', 'ES-419', 'ET', 'EU', 'FA', 'FI', 'FR', 'FR-FR', 'GA', 'GL', 'GN', 'GU', 'HA', 'HE', 'HI', 'HR', 'HT', 'HU', 'HY', 'ID', 'IG', 'IS', 'IT', 'JA', 'JV', 'KA', 'KK', 'KO', 'KY', 'LA', 'LB', 'LN', 'LT', 'LV', 'MG', 'MI', 'MK', 'ML', 'MN', 'MR', 'MS', 'MT', 'MY', 'NB', 'NE', 'NL', 'OC', 'OM', 'PA', 'PL', 'PS', 'PT-BR', 'PT-PT', 'QU', 'RO', 'RU', 'SA', 'SK', 'SL', 'SQ', 'SR', 'ST', 'SU', 'SV', 'SW', 'TA', 'TE', 'TG', 'TH', 'TK', 'TL', 'TN', 'TR', 'TS', 'TT', 'UK', 'UR', 'UZ', 'VI', 'WO', 'XH', 'YI', 'ZH', 'ZH-HANS', 'ZH-HANT', 'ZU']
languageAliases = {
#"EN":"EN-US",
"JP":"JA",
}
mergedLangs=(validLanguages+list(languageAliases.keys()))
+22 -18
View File
@@ -113,7 +113,7 @@ def cycleBearerTokenGet(url,headers):
return tweet return tweet
raise TwExtractError(400, "Extract error") raise TwExtractError(400, "Extract error")
def twitterApiGet(url,btoken=None,authToken=None,guestToken=None,userAgent=None): def twitterApiGet(url,btoken=None,authToken=None,guestToken=None,userAgent=None,language=None):
if authToken != None and authToken.startswith("oa|"): if authToken != None and authToken.startswith("oa|"):
url = url.replace("https://x.com/i/api/graphql/","https://api.twitter.com/graphql/") url = url.replace("https://x.com/i/api/graphql/","https://api.twitter.com/graphql/")
@@ -122,7 +122,7 @@ def twitterApiGet(url,btoken=None,authToken=None,guestToken=None,userAgent=None)
secret = authToken.split("|")[1] secret = authToken.split("|")[1]
twt = oauth1.Client(client_key='3nVuSoBZnx6U4vzUxf5w',client_secret='Bcs59EFbbsdF6Sl9Ng71smgStWEGwXXKSjYvPVt7qys',resource_owner_key=key,resource_owner_secret=secret) twt = oauth1.Client(client_key='3nVuSoBZnx6U4vzUxf5w',client_secret='Bcs59EFbbsdF6Sl9Ng71smgStWEGwXXKSjYvPVt7qys',resource_owner_key=key,resource_owner_secret=secret)
hdr = getAuthHeaders(androidBearer) hdr = getAuthHeaders(androidBearer,language=language)
del hdr["Authorization"] del hdr["Authorization"]
hdr["X-Twitter-Client"] = "TwitterAndroid" hdr["X-Twitter-Client"] = "TwitterAndroid"
if userAgent is not None: if userAgent is not None:
@@ -135,12 +135,12 @@ def twitterApiGet(url,btoken=None,authToken=None,guestToken=None,userAgent=None)
if btoken is None: if btoken is None:
btoken = v2bearer btoken = v2bearer
#return cycleBearerTokenGet(url,getAuthHeaders(bearer,authToken=authToken,guestToken=guestToken)) #return cycleBearerTokenGet(url,getAuthHeaders(bearer,authToken=authToken,guestToken=guestToken))
headers = getAuthHeaders(btoken,authToken=authToken,guestToken=guestToken) headers = getAuthHeaders(btoken,authToken=authToken,guestToken=guestToken,language=language)
response = requests.get(url, headers=headers) response = requests.get(url, headers=headers)
return response return response
def getAuthHeaders(btoken,authToken=None,guestToken=None): def getAuthHeaders(btoken,authToken=None,guestToken=None,language=None):
csrfToken=str(uuid.uuid4()).replace('-', '') csrfToken=str(uuid.uuid4()).replace('-', '')
headers = {"x-twitter-active-user":"yes","x-twitter-client-language":"en","x-csrf-token":csrfToken,"User-Agent":requestUserAgent} headers = {"x-twitter-active-user":"yes","x-twitter-client-language":"en","x-csrf-token":csrfToken,"User-Agent":requestUserAgent}
@@ -152,6 +152,9 @@ def getAuthHeaders(btoken,authToken=None,guestToken=None):
if guestToken is not None: if guestToken is not None:
headers["x-guest-token"] = guestToken headers["x-guest-token"] = guestToken
if language is not None:
headers["x-twitter-client-language"] = language
return headers return headers
def getGuestToken(): def getGuestToken():
@@ -268,7 +271,7 @@ def extractStatus_twExtractProxy(url,workaroundTokens=None):
continue continue
return output return output
def extractStatusV2(url,workaroundTokens): def extractStatusV2(url,workaroundTokens,tlLanguage=None):
# get tweet ID # get tweet ID
m = re.search(pathregex, url) m = re.search(pathregex, url)
if m is None: if m is None:
@@ -282,7 +285,7 @@ def extractStatusV2(url,workaroundTokens):
def request_with_token(twid, authToken): def request_with_token(twid, authToken):
vars = json.loads('{"includeTweetImpression":true,"includeHasBirdwatchNotes":false,"includeEditPerspective":false,"rest_ids":["x"],"includeEditControl":true,"includeCommunityTweetRelationship":true,"includeTweetVisibilityNudge":true}') vars = json.loads('{"includeTweetImpression":true,"includeHasBirdwatchNotes":false,"includeEditPerspective":false,"rest_ids":["x"],"includeEditControl":true,"includeCommunityTweetRelationship":true,"includeTweetVisibilityNudge":true}')
vars['rest_ids'][0] = str(twid) vars['rest_ids'][0] = str(twid)
tweet = twitterApiGet(f"https://x.com/i/api/graphql/{v2graphql_api}/TweetResultsByIdsQuery?variables={urllib.parse.quote(json.dumps(vars))}&features={urllib.parse.quote(v2Features)}",authToken=authToken) tweet = twitterApiGet(f"https://x.com/i/api/graphql/{v2graphql_api}/TweetResultsByIdsQuery?variables={urllib.parse.quote(json.dumps(vars))}&features={urllib.parse.quote(v2Features)}",authToken=authToken,language=tlLanguage)
try: try:
rateLimitRemaining = tweet.headers.get("x-rate-limit-remaining") rateLimitRemaining = tweet.headers.get("x-rate-limit-remaining")
print(f"Twitter Token Rate limit remaining: {rateLimitRemaining}") print(f"Twitter Token Rate limit remaining: {rateLimitRemaining}")
@@ -321,7 +324,7 @@ def extractStatusV2(url,workaroundTokens):
return tweet return tweet
return parallel_token_request(twid, tokens, request_with_token) return parallel_token_request(twid, tokens, request_with_token)
def extractStatusV2Android(url,workaroundTokens): def extractStatusV2Android(url,workaroundTokens,tlLanguage=None):
# get tweet ID # get tweet ID
m = re.search(pathregex, url) m = re.search(pathregex, url)
if m is None: if m is None:
@@ -335,7 +338,7 @@ def extractStatusV2Android(url,workaroundTokens):
try: try:
vars = json.loads('{"referrer":"home","includeTweetImpression":true,"includeHasBirdwatchNotes":false,"isReaderMode":false,"includeEditPerspective":false,"includeEditControl":true,"focalTweetId":0,"includeCommunityTweetRelationship":true,"includeTweetVisibilityNudge":true}') vars = json.loads('{"referrer":"home","includeTweetImpression":true,"includeHasBirdwatchNotes":false,"isReaderMode":false,"includeEditPerspective":false,"includeEditControl":true,"focalTweetId":0,"includeCommunityTweetRelationship":true,"includeTweetVisibilityNudge":true}')
vars['focalTweetId'] = int(twid) vars['focalTweetId'] = int(twid)
tweet = twitterApiGet(f"https://x.com/i/api/graphql/{androidGraphql_api}/ConversationTimelineV2?variables={urllib.parse.quote(json.dumps(vars))}&features={urllib.parse.quote(androidGraphqlFeatures)}", authToken=authToken,btoken=androidBearer,userAgent="TwitterAndroid/11.61.0-release.0 (311610000-r-0) G011A/9 (google;G011A;google;G011A;0;;1;2016)") tweet = twitterApiGet(f"https://x.com/i/api/graphql/{androidGraphql_api}/ConversationTimelineV2?variables={urllib.parse.quote(json.dumps(vars))}&features={urllib.parse.quote(androidGraphqlFeatures)}", authToken=authToken,btoken=androidBearer,userAgent="TwitterAndroid/11.61.0-release.0 (311610000-r-0) G011A/9 (google;G011A;google;G011A;0;;1;2016)",language=tlLanguage)
try: try:
rateLimitRemaining = tweet.headers.get("x-rate-limit-remaining") rateLimitRemaining = tweet.headers.get("x-rate-limit-remaining")
print(f"Twitter Android Token Rate limit remaining: {rateLimitRemaining}") print(f"Twitter Android Token Rate limit remaining: {rateLimitRemaining}")
@@ -380,7 +383,7 @@ def extractStatusV2Android(url,workaroundTokens):
return tweet return tweet
return parallel_token_request(twid, tokens, request_with_token) return parallel_token_request(twid, tokens, request_with_token)
def extractStatusV2TweetDetail(url,workaroundTokens): def extractStatusV2TweetDetail(url,workaroundTokens,tlLanguage=None):
# get tweet ID # get tweet ID
m = re.search(pathregex, url) m = re.search(pathregex, url)
if m is None: if m is None:
@@ -395,7 +398,7 @@ def extractStatusV2TweetDetail(url,workaroundTokens):
try: try:
vars = json.loads('{"focalTweetId":"0","with_rux_injections":false,"includePromotedContent":true,"withCommunity":true,"withQuickPromoteEligibilityTweetFields":true,"withBirdwatchNotes":true,"withVoice":true,"withV2Timeline":true}') vars = json.loads('{"focalTweetId":"0","with_rux_injections":false,"includePromotedContent":true,"withCommunity":true,"withQuickPromoteEligibilityTweetFields":true,"withBirdwatchNotes":true,"withVoice":true,"withV2Timeline":true}')
vars['focalTweetId'] = str(twid) vars['focalTweetId'] = str(twid)
tweet = twitterApiGet(f"https://x.com/i/api/graphql/{tweetDetailGraphql_api}/TweetDetail?variables={urllib.parse.quote(json.dumps(vars))}&features={urllib.parse.quote(tweetDetailGraphqlFeatures)}", authToken=authToken,btoken=v2bearer) tweet = twitterApiGet(f"https://x.com/i/api/graphql/{tweetDetailGraphql_api}/TweetDetail?variables={urllib.parse.quote(json.dumps(vars))}&features={urllib.parse.quote(tweetDetailGraphqlFeatures)}", authToken=authToken,btoken=v2bearer,language=tlLanguage)
try: try:
rateLimitRemaining = tweet.headers.get("x-rate-limit-remaining") rateLimitRemaining = tweet.headers.get("x-rate-limit-remaining")
print(f"Twitter Token Rate limit remaining: {rateLimitRemaining}") print(f"Twitter Token Rate limit remaining: {rateLimitRemaining}")
@@ -440,10 +443,10 @@ def extractStatusV2TweetDetail(url,workaroundTokens):
return tweet return tweet
return parallel_token_request(twid, tokens, request_with_token) return parallel_token_request(twid, tokens, request_with_token)
def extractStatusV2Rest_Anon(url,workaroundTokens): def extractStatusV2Rest_Anon(url,workaroundTokens,tlLanguage=None):
return extractStatusV2Rest(url,None) return extractStatusV2Rest(url,None,tlLanguage=tlLanguage)
def extractStatusV2Rest(url,workaroundTokens): def extractStatusV2Rest(url,workaroundTokens,tlLanguage=None):
# get tweet ID # get tweet ID
m = re.search(pathregex, url) m = re.search(pathregex, url)
if m is None: if m is None:
@@ -461,11 +464,11 @@ def extractStatusV2Rest(url,workaroundTokens):
random.shuffle(tokens) random.shuffle(tokens)
for authToken in tokens: for authToken in tokens:
try: try:
tweet = twitterApiGet(f"https://x.com/i/api/graphql/{v2AnonGraphql_api}/TweetResultByRestId?variables={urllib.parse.quote(json.dumps(vars))}&features={urllib.parse.quote(v2AnonFeatures)}", btoken=v2bearer,authToken=authToken,guestToken=guestToken) tweet = twitterApiGet(f"https://x.com/i/api/graphql/{v2AnonGraphql_api}/TweetResultByRestId?variables={urllib.parse.quote(json.dumps(vars))}&features={urllib.parse.quote(v2AnonFeatures)}", btoken=v2bearer,authToken=authToken,guestToken=guestToken,language=tlLanguage)
except Exception as e: except Exception as e:
continue continue
else: else:
tweet = twitterApiGet(f"https://x.com/i/api/graphql/{v2AnonGraphql_api}/TweetResultByRestId?variables={urllib.parse.quote(json.dumps(vars))}&features={urllib.parse.quote(v2AnonFeatures)}", btoken=v2bearer,guestToken=guestToken) tweet = twitterApiGet(f"https://x.com/i/api/graphql/{v2AnonGraphql_api}/TweetResultByRestId?variables={urllib.parse.quote(json.dumps(vars))}&features={urllib.parse.quote(v2AnonFeatures)}", btoken=v2bearer,guestToken=guestToken,language=tlLanguage)
try: try:
rateLimitRemaining = tweet.headers.get("x-rate-limit-remaining") rateLimitRemaining = tweet.headers.get("x-rate-limit-remaining")
@@ -512,12 +515,13 @@ def fixTweetData(tweet):
pass pass
return tweet return tweet
def extractStatus(url,workaroundTokens=None): def extractStatus(url,workaroundTokens=None,tlLanguage=None):
# TODO: commented out methods are too slow/unreliable at the moment
methods=[extractStatusV2Rest_Anon,extractStatusV2,extractStatusV2Rest,extractStatusV2Android]#,extractStatusV2TweetDetail] methods=[extractStatusV2Rest_Anon,extractStatusV2,extractStatusV2Rest,extractStatusV2Android]#,extractStatusV2TweetDetail]
if tlLanguage is not None: # prioritize endpoints that return translations
methods = [extractStatusV2Rest,extractStatusV2TweetDetail,extractStatusV2Android,extractStatusV2,extractStatusV2Rest_Anon]
for method in methods: for method in methods:
try: try:
result = method(url,workaroundTokens) result = method(url,workaroundTokens,tlLanguage=tlLanguage)
if 'legacy' not in result: if 'legacy' not in result:
print(f"{method.__name__} method failed: Legacy not found for {url}") print(f"{method.__name__} method failed: Legacy not found for {url}")
# try another method # try another method
+55 -22
View File
@@ -14,7 +14,7 @@ import msgs
import twExtract as twExtract import twExtract as twExtract
from cache import addVnfToLinkCache,getVnfFromLinkCache from cache import addVnfToLinkCache,getVnfFromLinkCache
import vxlogging as log import vxlogging as log
from utils import getTweetIdFromUrl, pathregex, determineMediaToEmbed, determineEmbedTweet, BytesIOWrapper, fixMedia from utils import getTweetIdFromUrl, pathregex, determineMediaToEmbed, determineEmbedTweet, BytesIOWrapper, fixMedia, indexOfAny
from vxApi import getApiResponse, getApiUserResponse from vxApi import getApiResponse, getApiUserResponse
from urllib.parse import urlparse from urllib.parse import urlparse
from PyRTF.Elements import Document from PyRTF.Elements import Document
@@ -22,12 +22,14 @@ from PyRTF.document.section import Section
from PyRTF.document.paragraph import Paragraph from PyRTF.document.paragraph import Paragraph
from copy import deepcopy from copy import deepcopy
import json import json
import datetime import translation
import activity as activitymg import activity as activitymg
app = Flask(__name__) app = Flask(__name__)
CORS(app) CORS(app)
user_agent="" user_agent=""
activityseparator="zqj"
staticFiles = { # TODO: Use flask static files instead of this staticFiles = { # TODO: Use flask static files instead of this
"favicon.ico": {"mime": "image/vnd.microsoft.icon","path": "favicon.ico"}, "favicon.ico": {"mime": "image/vnd.microsoft.icon","path": "favicon.ico"},
"apple-touch-icon.png": {"mime": "image/png","path": "apple-touch-icon.png"}, "apple-touch-icon.png": {"mime": "image/png","path": "apple-touch-icon.png"},
@@ -82,7 +84,11 @@ def generateActivityLink(tweetData,media=None,mediatype=None,embedIndex=-1):
return None return None
try: try:
embedIndex = embedIndex+1 embedIndex = embedIndex+1
return f"{config['config']['url']}/users/{tweetData['user_screen_name']}/statuses/{str(embedIndex)}{tweetData['tweetID']}"
extras=""
if tweetData["translation"] is not None:
extras+=activityseparator+tweetData["translation"]["destination_language"]
return f"{config['config']['url']}/users/{tweetData['user_screen_name']}/statuses/{str(embedIndex)}{activityseparator}{tweetData['tweetID']}{extras}"
except Exception as e: except Exception as e:
log.error("Error generating activity link: "+str(e)) log.error("Error generating activity link: "+str(e))
return None return None
@@ -95,7 +101,7 @@ def getAppName(tweetData,appnameSuffix=""):
def renderImageTweetEmbed(tweetData,image,appnameSuffix="",embedIndex=-1): def renderImageTweetEmbed(tweetData,image,appnameSuffix="",embedIndex=-1):
qrt = tweetData['qrt'] qrt = tweetData['qrt']
embedDesc = msgs.formatEmbedDesc("Image",tweetData['text'],qrt,tweetData['pollData']) embedDesc = msgs.formatEmbedDesc("Image",tweetData['text'],qrt,tweetData['pollData'],tweetData["translation"])
if image.startswith("https://pbs.twimg.com") and "?" not in image: if image.startswith("https://pbs.twimg.com") and "?" not in image:
image = f"{image}?name=orig" image = f"{image}?name=orig"
@@ -115,7 +121,7 @@ def renderImageTweetEmbed(tweetData,image,appnameSuffix="",embedIndex=-1):
def renderVideoTweetEmbed(tweetData,mediaInfo,appnameSuffix="",embedIndex=-1): def renderVideoTweetEmbed(tweetData,mediaInfo,appnameSuffix="",embedIndex=-1):
qrt = tweetData['qrt'] qrt = tweetData['qrt']
embedDesc = msgs.formatEmbedDesc("Video",tweetData['text'],qrt,tweetData['pollData']) embedDesc = msgs.formatEmbedDesc("Video",tweetData['text'],qrt,tweetData['pollData'],tweetData["translation"])
mediaInfo=fixMedia(mediaInfo) mediaInfo=fixMedia(mediaInfo)
@@ -138,7 +144,7 @@ def renderVideoTweetEmbed(tweetData,mediaInfo,appnameSuffix="",embedIndex=-1):
def renderTextTweetEmbed(tweetData,appnameSuffix=""): def renderTextTweetEmbed(tweetData,appnameSuffix=""):
qrt = tweetData['qrt'] qrt = tweetData['qrt']
embedDesc = msgs.formatEmbedDesc("Text",tweetData['text'],qrt,tweetData['pollData']) embedDesc = msgs.formatEmbedDesc("Text",tweetData['text'],qrt,tweetData['pollData'],tweetData["translation"])
return render_template("text.html", return render_template("text.html",
tweet=tweetData, tweet=tweetData,
@@ -154,7 +160,7 @@ def renderTextTweetEmbed(tweetData,appnameSuffix=""):
def renderArticleTweetEmbed(tweetData,appnameSuffix=""): def renderArticleTweetEmbed(tweetData,appnameSuffix=""):
articlePreview=tweetData['article']["title"]+"\n\n"+tweetData['article']["preview_text"]+"" articlePreview=tweetData['article']["title"]+"\n\n"+tweetData['article']["preview_text"]+""
embedDesc = msgs.formatEmbedDesc("Image",articlePreview,None,None) embedDesc = msgs.formatEmbedDesc("Image",articlePreview,None,None,None)
return render_template("image.html", return render_template("image.html",
tweet=tweetData, tweet=tweetData,
@@ -260,9 +266,9 @@ def userJson():
}, },
} }
def getTweetData(twitter_url,include_txt="false",include_rtf="false"): def getTweetData(twitter_url,include_txt="false",include_rtf="false",tlLanguage=None):
cachedVNF = getVnfFromLinkCache(twitter_url) cachedVNF = getVnfFromLinkCache(twitter_url)
if cachedVNF is not None and include_txt == "false" and include_rtf == "false": if cachedVNF is not None and include_txt == "false" and include_rtf == "false" and tlLanguage is None:
return cachedVNF return cachedVNF
try: try:
@@ -270,7 +276,7 @@ def getTweetData(twitter_url,include_txt="false",include_rtf="false"):
workaroundTokens = config['config']['workaroundTokens'].split(",") workaroundTokens = config['config']['workaroundTokens'].split(",")
else: else:
workaroundTokens = None workaroundTokens = None
rawTweetData = twExtract.extractStatus(twitter_url,workaroundTokens=workaroundTokens) rawTweetData = twExtract.extractStatus(twitter_url,workaroundTokens=workaroundTokens,tlLanguage=tlLanguage)
except: except:
rawTweetData = None rawTweetData = None
if rawTweetData == None or 'error' in rawTweetData: if rawTweetData == None or 'error' in rawTweetData:
@@ -281,7 +287,15 @@ def getTweetData(twitter_url,include_txt="false",include_rtf="false"):
tweetData = getApiResponse(rawTweetData,include_txt,include_rtf) tweetData = getApiResponse(rawTweetData,include_txt,include_rtf)
if tweetData is None: if tweetData is None:
return None return None
if include_txt == "false" and include_rtf == "false":
if tweetData["translation"] is None and tlLanguage is not None and tlLanguage != tweetData["lang"]:
# translation probably failed on twitter's end
pass # TODO: implement DeepL api
if tweetData["translation"] is not None and tlLanguage is None:
tweetData["translation"] = None # didn't ask for TL
if include_txt == "false" and include_rtf == "false" and tlLanguage is None:
addVnfToLinkCache(twitter_url,tweetData) addVnfToLinkCache(twitter_url,tweetData)
return tweetData return tweetData
@@ -307,6 +321,7 @@ def getUserData(twitter_url,includeFeed=False):
@app.route('/<path:sub_path>') # Default endpoint used by everything @app.route('/<path:sub_path>') # Default endpoint used by everything
def twitfix(sub_path): def twitfix(sub_path):
splitPath=sub_path.split("/")
global user_agent global user_agent
user_agent = request.headers.get('User-Agent', None) user_agent = request.headers.get('User-Agent', None)
if user_agent is None: if user_agent is None:
@@ -355,24 +370,33 @@ def twitfix(sub_path):
include_txt="false" include_txt="false"
include_rtf="false" include_rtf="false"
translationIndex=indexOfAny(splitPath,translation.mergedLangs,caseInsensitive=True)
translationLang = None
if translationIndex != -1:
if splitPath[translationIndex].upper() in translation.languageAliases:
translationLang = translation.languageAliases[splitPath[translationIndex].upper()]
else:
translationLang = splitPath[translationIndex].lower()
if isApiRequest: if isApiRequest:
if "include_txt" in request.args: if "include_txt" in request.args:
include_txt = request.args.get("include_txt") include_txt = request.args.get("include_txt")
if "include_rtf" in request.args: if "include_rtf" in request.args:
include_rtf = request.args.get("include_rtf") include_rtf = request.args.get("include_rtf")
tweetData = getTweetData(twitter_url,include_txt,include_rtf) tweetData = getTweetData(twitter_url,include_txt,include_rtf,tlLanguage=translationLang)
if tweetData is None: if tweetData is None:
log.error("Tweet Data Get failed for "+twitter_url) log.error("Tweet Data Get failed for "+twitter_url)
return message(msgs.failedToScan) return message(msgs.failedToScan)
qrt = None qrt = None
if 'qrtURL' in tweetData and tweetData['qrtURL'] is not None: if 'qrtURL' in tweetData and tweetData['qrtURL'] is not None:
qrt = getTweetData(tweetData['qrtURL']) qrt = getTweetData(tweetData['qrtURL'],tlLanguage=translationLang)
tweetData['qrt'] = qrt tweetData['qrt'] = qrt
retweet = None retweet = None
if 'retweetURL' in tweetData and tweetData['retweetURL'] is not None: if 'retweetURL' in tweetData and tweetData['retweetURL'] is not None:
retweet = getTweetData(tweetData['retweetURL']) retweet = getTweetData(tweetData['retweetURL'],tlLanguage=translationLang)
tweetData['retweet'] = retweet tweetData['retweet'] = retweet
tweetData = deepcopy(tweetData) tweetData = deepcopy(tweetData)
@@ -403,9 +427,10 @@ def twitfix(sub_path):
return send_file(BytesIOWrapper(rtf), mimetype='application/rtf', as_attachment=True, download_name=f'{tweetData["user_screen_name"]}_{tweetData["tweetID"]}.rtf') return send_file(BytesIOWrapper(rtf), mimetype='application/rtf', as_attachment=True, download_name=f'{tweetData["user_screen_name"]}_{tweetData["tweetID"]}.rtf')
embedIndex = -1 embedIndex = -1
# if url ends with /1, /2, /3, or /4, we'll use that as the index # if url has /1, /2, /3, or /4, we'll use that as the index
if sub_path[-2:] in ["/1","/2","/3","/4"]: numIdx=indexOfAny(splitPath,["1","2","3","4"]);
embedIndex = int(sub_path[-1])-1 if numIdx != -1:
embedIndex = int(splitPath[numIdx])-1
sub_path = sub_path[:-2] sub_path = sub_path[:-2]
if isApiRequest: # Directly return the API response if the request is from the API if isApiRequest: # Directly return the API response if the request is from the API
@@ -484,21 +509,29 @@ def rendercombined():
@app.route("/api/v1/statuses/<string:tweet_id>") @app.route("/api/v1/statuses/<string:tweet_id>")
def api_v1_status(tweet_id): def api_v1_status(tweet_id):
embedIndex = int(tweet_id[0])-1 lang = None
tweet_id = int(tweet_id[1:]) if activityseparator not in tweet_id: #old embeds
embedIndex = int(tweet_id[0])-1
tweet_id = int(tweet_id[1:])
else:
args = tweet_id.split(activityseparator)
embedIndex = int(args[0])
tweet_id = int(args[1])
if len(args)>2:
lang=args[2]
twitter_url=f"https://twitter.com/i/status/{tweet_id}" twitter_url=f"https://twitter.com/i/status/{tweet_id}"
tweetData = getTweetData(twitter_url) tweetData = getTweetData(twitter_url,tlLanguage=lang)
if tweetData is None: if tweetData is None:
log.error("Tweet Data Get failed for "+twitter_url) log.error("Tweet Data Get failed for "+twitter_url)
return message(msgs.failedToScan) return message(msgs.failedToScan)
qrt = None qrt = None
if 'qrtURL' in tweetData and tweetData['qrtURL'] is not None: if 'qrtURL' in tweetData and tweetData['qrtURL'] is not None:
qrt = getTweetData(tweetData['qrtURL']) qrt = getTweetData(tweetData['qrtURL'],tlLanguage=lang)
tweetData['qrt'] = qrt tweetData['qrt'] = qrt
retweet = None retweet = None
if 'retweetURL' in tweetData and tweetData['retweetURL'] is not None: if 'retweetURL' in tweetData and tweetData['retweetURL'] is not None:
retweet = getTweetData(tweetData['retweetURL']) retweet = getTweetData(tweetData['retweetURL'],tlLanguage=lang)
tweetData['retweet'] = retweet tweetData['retweet'] = retweet
if tweetData is None: if tweetData is None:
+12 -1
View File
@@ -91,4 +91,15 @@ def determineMediaToEmbed(tweetData,embedIndex = -1,convertGif = True):
media['url'] = gcApi + "/convert.avif?url=" + vurl media['url'] = gcApi + "/convert.avif?url=" + vurl
suffix += " • GIF" suffix += " • GIF"
media["suffix"] = suffix media["suffix"] = suffix
return media return media
def indexOfAny(list,items,caseInsensitive=False):
if caseInsensitive:
list = [item.upper() for item in list]
items = [item.upper() for item in items]
for item in items:
try:
return list.index(item)
except:
continue
return -1
+10
View File
@@ -77,6 +77,7 @@ def getApiResponse(tweet,include_txt=False,include_rtf=False):
oldTweetVersion = False oldTweetVersion = False
tweetArticle=None tweetArticle=None
lang=None lang=None
translation=None
if "screen_name" not in userL: if "screen_name" not in userL:
userL["screen_name"] = user["core"]["screen_name"] userL["screen_name"] = user["core"]["screen_name"]
@@ -85,6 +86,14 @@ def getApiResponse(tweet,include_txt=False,include_rtf=False):
if "profile_image_url_https" not in userL: if "profile_image_url_https" not in userL:
userL["profile_image_url_https"] = user["avatar"]["image_url"] userL["profile_image_url_https"] = user["avatar"]["image_url"]
if "grok_translated_post_with_availability" in tweet and tweet["grok_translated_post_with_availability"]["is_available"] and "data" in tweet["grok_translated_post_with_availability"]:
tlData = tweet["grok_translated_post_with_availability"]["data"]
translation = {
"source_language":tlData["source_language"],
"destination_language":tlData["destination_language"],
"text":tlData["translation"]
}
#editedTweet=False #editedTweet=False
try: try:
if "birdwatch_pivot" in tweet: if "birdwatch_pivot" in tweet:
@@ -305,6 +314,7 @@ def getApiResponse(tweet,include_txt=False,include_rtf=False):
"replyingToID": replyingToID, "replyingToID": replyingToID,
"fetched_on": int(datetime.now().timestamp()), "fetched_on": int(datetime.now().timestamp()),
"retweetURL":retweetURL, "retweetURL":retweetURL,
"translation":translation
} }
try: try:
apiObject["date_epoch"] = int(datetime.strptime(tweetL["created_at"], "%a %b %d %H:%M:%S %z %Y").timestamp()) apiObject["date_epoch"] = int(datetime.strptime(tweetL["created_at"], "%a %b %d %H:%M:%S %z %Y").timestamp())