Skip to content

S
shadowban-eu-backend
Commit 90f622dd authored by nanahira's avatar nanahira
Browse files
Options

check every tweet instead of one

parent 0d755dee
Pipeline #18764 passed with stage
    in 39 seconds
    Hide whitespace changes
    Inline Side-by-side
    Showing with 97 additions and 93 deletions
    backend_requests.py
    View file @ 90f622dd
    ......@@ -248,10 +248,14 @@ def searchban(screen_name):
    replies = twitter_b.get("https://twitter.com/i/api/graphql/{}/{}".format(ENDPOINT["UserTweetsAndReplies"], "UserTweetsAndReplies"), params=get_reply_param)
    # print(replies.text)
    try:
    ghostban = True
    ghostTweetId = None
    ghostReplyId = None
    showmore = False
    showmoreTweetId = None
    showmoreReplyId = None
    repliesJson = replies.json()
    # print(repliesJson)
    maindata = repliesJson["data"]["user"]["result"]["timeline"]["timeline"]["instructions"]
    for d in maindata:
    if d["type"] == "TimelineAddEntries":
    ......@@ -260,95 +264,101 @@ def searchban(screen_name):
    tmp = ent["content"]["itemContent"]["tweet_results"]["result"]["legacy"]
    if "in_reply_to_status_id_str" in tmp:
    reply = tmp
    # print("Found a reply!", tmp["full_text"])
    break
    if reply is None:
    returnjson["tests"]["ghost"] = {}
    returnjson["tests"]["more_replies"] = {}
    else:
    tweet_detail_vars = {
    "focalTweetId": reply["in_reply_to_status_id_str"],
    "includePromotedContent":False,
    "withBirdwatchNotes":False,
    "withSuperFollowsUserFields":False,
    "withDownvotePerspective":False,
    "withReactionsMetadata":False,
    "withReactionsPerspective":False,
    "withSuperFollowsTweetFields":False,
    "withVoice":False,
    }
    tweetdetails = twitter_b.get("https://twitter.com/i/api/graphql/{}/{}".format(ENDPOINT["TweetDetail"], "TweetDetail"), params={"variables": json.dumps(tweet_detail_vars), "features": FeaturesJson})
    insts = tweetdetails.json()["data"]["threaded_conversation_with_injections"]["instructions"]
    ghostban = True
    inReplyToGhost = reply["id_str"]
    inReplyToShowMore = reply["id_str"]
    showmore = False
    for inst in insts:
    if inst["type"] == "TimelineAddEntries":
    for ent in inst["entries"]:
    print("Current entry of {} is: {}".format(screen_name, ent["entryId"]))
    if ent["entryId"].startswith("conversationthread") and ghostban:
    for item in ent["content"]["items"]:
    #print("Current sub entry of {} {} is: {}".format(screen_name, ent["entryId"], item["item"]["itemContent"]))
    if "tweet_results" in item["item"]["itemContent"] and item["item"]["itemContent"]["tweet_results"]["result"]["legacy"]["user_id_str"] == user_id:
    validId = item["item"]["itemContent"]["tweet_results"]["result"]["legacy"]["id_str"]
    returnjson["tests"]["ghost"] = {"ban": False, "in_reply_to": validId}
    ghostban = False
    print("Found valid reply {}, so {} is not ghost banned.".format(validId, screen_name))
    tweetId = reply["in_reply_to_status_id_str"]
    if ghostTweetId is None:
    ghostTweetId = tweetId
    ghostReplyId = reply["id_str"]
    if showmoreTweetId is None:
    showmoreTweetId = tweetId
    showmoreReplyId = reply["id_str"]
    tweet_detail_vars = {
    "focalTweetId": tweetId,
    "includePromotedContent":False,
    "withBirdwatchNotes":False,
    "withSuperFollowsUserFields":False,
    "withDownvotePerspective":False,
    "withReactionsMetadata":False,
    "withReactionsPerspective":False,
    "withSuperFollowsTweetFields":False,
    "withVoice":False,
    }
    tweetdetails = twitter_b.get("https://twitter.com/i/api/graphql/{}/{}".format(ENDPOINT["TweetDetail"], "TweetDetail"), params={"variables": json.dumps(tweet_detail_vars), "features": FeaturesJson})
    tweetData = tweetdetails.json()["data"]
    if "threaded_conversation_with_injections" not in tweetData:
    continue
    insts = tweetdetails.json()["data"]["threaded_conversation_with_injections"]["instructions"]
    for inst in insts:
    if not ghostban and showmore:
    # all checks done
    break
    if ent["entryId"].startswith("cursor-bottom"):
    print("{}'s test touched bottom.".format(screen_name))
    #returnjson["tests"]["ghost"] = {}
    #returnjson["tests"]["more_replies"] = {}
    break
    if ent["entryId"].startswith("cursor-showmorethreadsprompt") and not showmore:
    # showmore = True
    cursor_vars = tweet_detail_vars
    cursor_vars["cursor"] = ent["content"]["itemContent"]["value"]
    cursor = twitter_b.get("https://twitter.com/i/api/graphql/{}/{}".format(ENDPOINT["TweetDetail"], "TweetDetail"), params={"variables": json.dumps(cursor_vars), "features": FeaturesJson})
    cursor_insts = cursor.json()["data"]["threaded_conversation_with_injections"]["instructions"]
    for c_i in cursor_insts:
    if c_i["type"] == "TimelineAddEntries":
    if len(c_i["entries"]) == 0:
    returnjson["tests"]["more_replies"] = {"ban": True}
    showmore = True
    break
    for c_ent in c_i["entries"]:
    if c_ent["entryId"].startswith("conversationthread"):
    # more = True
    print("Checking more contents of {}".format(screen_name))
    for c_item in c_ent["content"]["items"]:
    if c_item["item"]["itemContent"]["tweet_results"]["result"]["legacy"]["user_id_str"] == user_id:
    inReplyToShowMore = c_item["item"]["itemContent"]["tweet_results"]["result"]["legacy"]["id_str"]
    # returnjson["tests"]["ghost"] = {"ban": False, "in_reply_to": inReplyToShowMore}
    returnjson["tests"]["more_replies"] = {"ban": True, "in_reply_to": inReplyToShowMore}
    print("{} is reply deboosted because of {}.".format(screen_name, inReplyToShowMore))
    showmore = True
    # more = False
    if inst["type"] == "TimelineAddEntries":
    for ent in inst["entries"]:
    if not ghostban and showmore:
    # all checks done
    break
    print("Current entry of {} is: {}".format(screen_name, ent["entryId"]))
    if ent["entryId"].startswith("conversationthread") and ghostban:
    for item in ent["content"]["items"]:
    if "tweet_results" in item["item"]["itemContent"] and item["item"]["itemContent"]["tweet_results"]["result"]["legacy"]["user_id_str"] == user_id:
    replyId = item["item"]["itemContent"]["tweet_results"]["result"]["legacy"]["id_str"]
    returnjson["tests"]["ghost"] = {"ban": False, "tweet": tweetId, "in_reply_to": replyId}
    ghostban = False
    print("Found valid reply {} => {}, so {} is not ghost banned.".format(tweetId, replyId, screen_name))
    break
    #if more:
    # print("{} is both ghost banned and reply deboosted.".format(screen_name))
    # returnjson["tests"]["ghost"] = {"ban": True}
    # returnjson["tests"]["more_replies"] = {"ban": True}
    if ghostban:
    print("{} is ghost banned.".format(screen_name))
    returnjson["tests"]["ghost"] = {"ban": True, "in_reply_to": inReplyToGhost}
    if not showmore:
    returnjson["tests"]["more_replies"] = {
    "ban": False,
    "in_reply_to": inReplyToShowMore,
    }
    replyTweet = reply["in_reply_to_status_id_str"]
    for testField in ["ghost", "more_replies"]:
    if testField in returnjson["tests"]:
    returnjson["tests"][testField]["tweet"] = replyTweet
    if ent["entryId"].startswith("cursor-showmorethreadsprompt") and not showmore:
    # showmore = True
    cursor_vars = tweet_detail_vars
    cursor_vars["cursor"] = ent["content"]["itemContent"]["value"]
    cursor = twitter_b.get("https://twitter.com/i/api/graphql/{}/{}".format(ENDPOINT["TweetDetail"], "TweetDetail"), params={"variables": json.dumps(cursor_vars), "features": FeaturesJson})
    cursor_insts = cursor.json()["data"]["threaded_conversation_with_injections"]["instructions"]
    for c_i in cursor_insts:
    if c_i["type"] == "TimelineAddEntries":
    if len(c_i["entries"]) == 0:
    returnjson["tests"]["more_replies"] = {"ban": True}
    showmore = True
    break
    for c_ent in c_i["entries"]:
    if c_ent["entryId"].startswith("conversationthread"):
    print("Checking more contents of {} by {}".format(tweetId, screen_name))
    for c_item in c_ent["content"]["items"]:
    if c_item["item"]["itemContent"]["tweet_results"]["result"]["legacy"]["user_id_str"] == user_id:
    replyId = c_item["item"]["itemContent"]["tweet_results"]["result"]["legacy"]["id_str"]
    returnjson["tests"]["more_replies"] = {"ban": True, "in_reply_to": replyId, "tweet": tweetId}
    print("{} is reply deboosted because of {} => {}.".format(screen_name, tweetId, replyId))
    showmore = True
    break
    if ghostban:
    print("{} is ghost banned.".format(screen_name))
    returnjson["tests"]["ghost"] = {
    "ban": True,
    "tweet": ghostTweetId,
    "in_reply_to": ghostReplyId
    }
    if not showmore:
    returnjson["tests"]["more_replies"] = {
    "ban": False,
    "in_reply_to": showmoreReplyId,
    "tweet": showmoreTweetId
    }
    # No search ban || more replies => No ghost ban
    if returnjson["tests"]["search"] and returnjson["tests"]["search"] != "_error" and "ban" in returnjson["tests"]["ghost"] and returnjson["tests"]["ghost"]["ban"] == True or "ban" in returnjson["tests"]["more_replies"] and returnjson["tests"]["more_replies"]["ban"] == True:
    returnjson["tests"]["ghost"] = {
    "ban": False,
    "tweet": ghostTweetId,
    "in_reply_to": ghostReplyId
    }
    # No ghost ban && unknown more replies => No more replies ban
    if "ban" not in returnjson["tests"]["more_replies"] and "ban" in returnjson["tests"]["ghost"] and returnjson["tests"]["ghost"]["ban"] == False:
    returnjson["tests"]["more_replies"] = {
    "ban": False,
    "in_reply_to": showmoreReplyId,
    "tweet": showmoreTweetId
    }
    except KeyError as e:
    print("Errored testing {}".format(screen_name))
    # print(Exception.with_traceback(e))
    print(e)
    returnjson["tests"]["ghost"] = {}
    returnjson["tests"]["more_replies"] = {}
    ......@@ -357,13 +367,7 @@ def searchban(screen_name):
    #print("ban" in returnjson["tests"]["ghost"])
    #print(returnjson["tests"]["ghost"])
    # No search ban => No ghost ban
    if returnjson["tests"]["search"] and returnjson["tests"]["search"] != "_error" and "ban" in returnjson["tests"]["ghost"] and returnjson["tests"]["ghost"]["ban"] == True:
    returnjson["tests"]["ghost"]["ban"] = False
    # No ghost ban && unknown more replies => No more replies ban
    if "ban" not in returnjson["tests"]["more_replies"] and "ban" in returnjson["tests"]["ghost"] and returnjson["tests"]["ghost"]["ban"] == False:
    returnjson["tests"]["more_replies"] = {"ban": False}
    print("Result of {}: {}".format(screen_name, json.dumps(returnjson)))
    return returnjson
    ......
    Markdown is supported
    0% or
    You are about to add 0 people to the discussion. Proceed with caution.
    Finish editing this message first!
    Please register or to comment