alterware/BetterTwitFix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #103 from dylanpdx/main

Browse Source 
Logging changes & Cache-Control header addition
This commit is contained in:
Dylan 2023-06-25 23:36:01 +01:00 committed by GitHub
commit dbf699aba0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 72 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
cache.py
View File

@ -4,6 +4,7 @@ from datetime import date,datetime
import json import json
import os import os
import boto3 import boto3
import vxlogging as log
link_cache_system = config['config']['link_cache'] link_cache_system = config['config']['link_cache']
link_cache = {} link_cache = {}
@ -21,16 +22,16 @@ if link_cache_system == "json":
f = open('links.json',) f = open('links.json',)
link_cache = json.load(f) link_cache = json.load(f)
except json.decoder.JSONDecodeError: except json.decoder.JSONDecodeError:
print(" ➤ [ X ] Failed to load cache JSON file. Creating new file.") log.warn("Failed to load cache JSON file. Creating new file.")
link_cache = {} link_cache = {}
except FileNotFoundError: except FileNotFoundError:
print(" ➤ [ X ] Failed to load cache JSON file. Creating new file.") log.warn("Failed to load cache JSON file. Creating new file.")
link_cache = {} link_cache = {}
finally: finally:
f.close() f.close()
elif link_cache_system == "ram": elif link_cache_system == "ram":
link_cache = {} link_cache = {}
print("Your link_cache_system is set to 'ram' which is not recommended; this is only intended to be used for tests") log.warn("Your link_cache_system is set to 'ram' which is not recommended; this is only intended to be used for tests")
elif link_cache_system == "db": elif link_cache_system == "db":
client = pymongo.MongoClient(config['config']['database'], connect=False) client = pymongo.MongoClient(config['config']['database'], connect=False)
table = config['config']['table'] table = config['config']['table']
@ -49,17 +50,17 @@ def addVnfToLinkCache(video_link, vnf):
try: try:
if link_cache_system == "db": if link_cache_system == "db":
out = db.linkCache.update_one(vnf) out = db.linkCache.update_one(vnf)
print(" ➤ [ + ] Link added to DB cache ") log.debug("Link added to DB cache ")
return True return True
elif link_cache_system == "json": elif link_cache_system == "json":
link_cache[video_link] = vnf link_cache[video_link] = vnf
with open("links.json", "w") as outfile: with open("links.json", "w") as outfile:
json.dump(link_cache, outfile, indent=4, sort_keys=True, default=serializeUnknown) json.dump(link_cache, outfile, indent=4, sort_keys=True, default=serializeUnknown)
print(" ➤ [ + ] Link added to JSON cache ") log.debug("Link added to JSON cache ")
return True return True
elif link_cache_system == "ram": # FOR TESTS ONLY elif link_cache_system == "ram": # FOR TESTS ONLY
link_cache[video_link] = vnf link_cache[video_link] = vnf
print(" ➤ [ + ] Link added to RAM cache ") log.debug("Link added to RAM cache ")
elif link_cache_system == "dynamodb": # pragma: no cover elif link_cache_system == "dynamodb": # pragma: no cover
vnf["ttl"] = int(vnf["ttl"].strftime('%s')) vnf["ttl"] = int(vnf["ttl"].strftime('%s'))
table = client.Table(DYNAMO_CACHE_TBL) table = client.Table(DYNAMO_CACHE_TBL)
@ -70,11 +71,10 @@ def addVnfToLinkCache(video_link, vnf):
'ttl':vnf["ttl"] 'ttl':vnf["ttl"]
} }
) )
print(" ➤ [ + ] Link added to dynamodb cache ") log.debug("Link added to dynamodb cache ")
return True return True
except Exception as e: except Exception as e:
print(" ➤ [ X ] Failed to add link to DB cache") log.error("Failed to add link to DB cache: "+str(e)+" "+video_link)
print(e)
return False return False
def getVnfFromLinkCache(video_link): def getVnfFromLinkCache(video_link):
@ -83,24 +83,23 @@ def getVnfFromLinkCache(video_link):
if link_cache_system == "db": if link_cache_system == "db":
collection = db.linkCache collection = db.linkCache
vnf = collection.find_one({'tweet': video_link}) vnf = collection.find_one({'tweet': video_link})
# print(vnf)
if vnf != None: if vnf != None:
hits = ( vnf['hits'] + 1 ) hits = ( vnf['hits'] + 1 )
print(" ➤ [ ✔ ] Link located in DB cache. " + "hits on this link so far: [" + str(hits) + "]") log.debug("Link located in DB cache.")
query = { 'tweet': video_link } query = { 'tweet': video_link }
change = { "$set" : { "hits" : hits } } change = { "$set" : { "hits" : hits } }
out = db.linkCache.update_one(query, change) out = db.linkCache.update_one(query, change)
return vnf return vnf
else: else:
print(" ➤ [ X ] Link not in DB cache") log.debug("Link not in DB cache")
return None return None
elif link_cache_system == "json": elif link_cache_system == "json":
if video_link in link_cache: if video_link in link_cache:
print("Link located in json cache") log.debug("Link located in json cache")
vnf = link_cache[video_link] vnf = link_cache[video_link]
return vnf return vnf
else: else:
print(" ➤ [ X ] Link not in json cache") log.debug("Link not in json cache")
return None return None
elif link_cache_system == "dynamodb": # pragma: no cover elif link_cache_system == "dynamodb": # pragma: no cover
table = client.Table(DYNAMO_CACHE_TBL) table = client.Table(DYNAMO_CACHE_TBL)
@ -110,19 +109,19 @@ def getVnfFromLinkCache(video_link):
} }
) )
if 'Item' in response: if 'Item' in response:
print("Link located in dynamodb cache") log.debug("Link located in dynamodb cache")
vnf = response['Item']['vnf'] vnf = response['Item']['vnf']
return vnf return vnf
else: else:
print(" ➤ [ X ] Link not in dynamodb cache") log.debug("Link not in dynamodb cache")
return None return None
elif link_cache_system == "ram": # FOR TESTS ONLY elif link_cache_system == "ram": # FOR TESTS ONLY
if video_link in link_cache: if video_link in link_cache:
print("Link located in json cache") log.debug("Link located in json cache")
vnf = link_cache[video_link] vnf = link_cache[video_link]
return vnf return vnf
else: else:
print(" ➤ [ X ] Link not in cache") log.debug("Link not in cache")
return None return None
elif link_cache_system == "none": elif link_cache_system == "none":
return None return None

48
twitfix.py
View File

@ -15,6 +15,7 @@ from configHandler import config
from cache import addVnfToLinkCache,getVnfFromLinkCache from cache import addVnfToLinkCache,getVnfFromLinkCache
from yt_dlp.utils import ExtractorError from yt_dlp.utils import ExtractorError
from twitter.api import TwitterHTTPError from twitter.api import TwitterHTTPError
import vxlogging as log
app = Flask(__name__) app = Flask(__name__)
CORS(app) CORS(app)
@ -69,11 +70,11 @@ def twitfix(sub_path):
if e is not None: if e is not None:
return message(msgs.failedToScan+msgs.failedToScanExtra+e) return message(msgs.failedToScan+msgs.failedToScanExtra+e)
return message(msgs.failedToScan) return message(msgs.failedToScan)
return getTemplate("rawvideo.html",vnf,"","",clean,"","","","") return make_cached_vnf_response(vnf,getTemplate("rawvideo.html",vnf,"","",clean,"","","",""))
elif request.url.startswith("https://d.vx"): # Matches d.fx? Try to give the user a direct link elif request.url.startswith("https://d.vx"): # Matches d.fx? Try to give the user a direct link
if user_agent in generate_embed_user_agents: if user_agent in generate_embed_user_agents:
twitter_url = config['config']['url'] + "/"+sub_path twitter_url = config['config']['url'] + "/"+sub_path
print( " ➤ [ D ] d.vx link shown to discord user-agent!") log.debug( "d.vx link shown to discord user-agent!")
if request.url.endswith(".mp4") and "?" not in request.url: if request.url.endswith(".mp4") and "?" not in request.url:
if "?" not in request.url: if "?" not in request.url:
clean = twitter_url[:-4] clean = twitter_url[:-4]
@ -83,7 +84,7 @@ def twitfix(sub_path):
clean = twitter_url clean = twitter_url
return redirect(clean+".mp4", 301) return redirect(clean+".mp4", 301)
else: else:
print(" ➤ [ R ] Redirect to MP4 using d.fxtwitter.com") log.debug("Redirect to MP4 using d.fxtwitter.com")
return dir(sub_path) return dir(sub_path)
elif request.url.endswith("/1") or request.url.endswith("/2") or request.url.endswith("/3") or request.url.endswith("/4") or request.url.endswith("%2F1") or request.url.endswith("%2F2") or request.url.endswith("%2F3") or request.url.endswith("%2F4"): elif request.url.endswith("/1") or request.url.endswith("/2") or request.url.endswith("/3") or request.url.endswith("/4") or request.url.endswith("%2F1") or request.url.endswith("%2F2") or request.url.endswith("%2F3") or request.url.endswith("%2F4"):
twitter_url = "https://twitter.com/" + sub_path twitter_url = "https://twitter.com/" + sub_path
@ -107,7 +108,7 @@ def twitfix(sub_path):
return res return res
else: else:
print(" ➤ [ R ] Redirect to " + twitter_url) log.debug("Redirect to " + twitter_url)
return redirect(twitter_url, 301) return redirect(twitter_url, 301)
else: else:
return message("This doesn't appear to be a twitter URL") return message("This doesn't appear to be a twitter URL")
@ -129,7 +130,7 @@ def dir(sub_path):
return res return res
else: else:
print(" ➤ [ R ] Redirect to direct MP4 URL") log.debug("Redirect to direct MP4 URL")
return direct_video(twitter_url) return direct_video(twitter_url)
else: else:
return redirect(url, 301) return redirect(url, 301)
@ -160,7 +161,7 @@ def rendercombined():
finalImg = finalImg.convert("RGB") finalImg = finalImg.convert("RGB")
finalImg.save(imgIo, 'JPEG',quality=70) finalImg.save(imgIo, 'JPEG',quality=70)
imgIo.seek(0) imgIo.seek(0)
return send_file(imgIo, mimetype='image/jpeg') return send_file(imgIo, mimetype='image/jpeg',max_age=86400)
def upgradeVNF(vnf): def upgradeVNF(vnf):
# Makes sure any VNF object passed through this has proper fields if they're added in later versions # Makes sure any VNF object passed through this has proper fields if they're added in later versions
@ -183,6 +184,23 @@ def upgradeVNF(vnf):
def getDefaultTTL(): # TTL for deleting items from the database def getDefaultTTL(): # TTL for deleting items from the database
return datetime.today().replace(microsecond=0) + timedelta(days=1) return datetime.today().replace(microsecond=0) + timedelta(days=1)
def secondsUntilTTL(ttl):
if ttl < datetime.today().replace(microsecond=0):
return 0
return (ttl - datetime.today().replace(microsecond=0)).total_seconds()
def make_cached_vnf_response(vnf,response):
try:
if vnf['ttl'] == None or vnf['ttl'] < datetime.today().replace(microsecond=0) or 'ttl' not in vnf:
return response
resp = make_response(response)
resp.cache_control.max_age = secondsUntilTTL(vnf['ttl'])
resp.cache_control.public = True
return resp
except Exception as e:
log.error("Error making cached response: " + str(e))
return response
def vnfFromCacheOrDL(video_link): def vnfFromCacheOrDL(video_link):
cached_vnf = getVnfFromLinkCache(video_link) cached_vnf = getVnfFromLinkCache(video_link)
if cached_vnf == None: if cached_vnf == None:
@ -204,7 +222,7 @@ def vnfFromCacheOrDL(video_link):
else: else:
return None,None return None,None
except Exception as e: except Exception as e:
print(e) log.error(e)
return None,None return None,None
else: else:
return upgradeVNF(cached_vnf),None return upgradeVNF(cached_vnf),None
@ -268,7 +286,7 @@ def tweetInfo(url, tweet="", desc="", thumb="", uploader="", screen_name="", pfp
def link_to_vnf_from_tweet_data(tweet,video_link): def link_to_vnf_from_tweet_data(tweet,video_link):
imgs = ["","","","", ""] imgs = ["","","","", ""]
print(" ➤ [ + ] Tweet Type: " + tweetType(tweet)) log.debug("Tweet Type: " + tweetType(tweet))
isGif=False isGif=False
# Check to see if tweet has a video, if not, make the url passed to the VNF the first t.co link in the tweet # Check to see if tweet has a video, if not, make the url passed to the VNF the first t.co link in the tweet
if tweetType(tweet) == "Video": if tweetType(tweet) == "Video":
@ -301,7 +319,6 @@ def link_to_vnf_from_tweet_data(tweet,video_link):
imgs[i] = media['media_url_https'] imgs[i] = media['media_url_https']
i = i + 1 i = i + 1
#print(imgs)
imgs[4] = str(i) imgs[4] = str(i)
url = "" url = ""
images= imgs images= imgs
@ -363,15 +380,16 @@ def link_to_vnf_from_tweet_data(tweet,video_link):
def link_to_vnf_from_unofficial_api(video_link): def link_to_vnf_from_unofficial_api(video_link):
tweet=None tweet=None
print(" ➤ [ + ] Attempting to download tweet info from UNOFFICIAL Twitter API") log.info("Attempting to download tweet info: "+video_link)
tweet = twExtract.extractStatus(video_link) tweet = twExtract.extractStatus(video_link)
print (" ➤ [ ✔ ] Unofficial API Success") log.success("Unofficial API Success")
if "extended_entities" not in tweet: if "extended_entities" not in tweet:
# check if any entities with urls ending in /video/XXX or /photo/XXX exist # check if any entities with urls ending in /video/XXX or /photo/XXX exist
if "entities" in tweet and "urls" in tweet["entities"]: if "entities" in tweet and "urls" in tweet["entities"]:
for url in tweet["entities"]["urls"]: for url in tweet["entities"]["urls"]:
if "/video/" in url["expanded_url"] or "/photo/" in url["expanded_url"]: if "/video/" in url["expanded_url"] or "/photo/" in url["expanded_url"]:
log.info("Extra tweet info found in entities: "+video_link+" -> "+url["expanded_url"])
subTweet = twExtract.extractStatus(url["expanded_url"]) subTweet = twExtract.extractStatus(url["expanded_url"])
if "extended_entities" in subTweet: if "extended_entities" in subTweet:
tweet["extended_entities"] = subTweet["extended_entities"] tweet["extended_entities"] = subTweet["extended_entities"]
@ -423,7 +441,7 @@ def getTemplate(template,vnf,desc,image,video_link,color,urlDesc,urlUser,urlLink
videoSize = embedVNF['size'] ) videoSize = embedVNF['size'] )
def embed(video_link, vnf, image): def embed(video_link, vnf, image):
print(" ➤ [ E ] Embedding " + vnf['type'] + ": " + video_link) log.info("Embedding " + vnf['type'] + ": " + video_link)
desc = re.sub(r' http.*t\.co\S+', '', vnf['description']) desc = re.sub(r' http.*t\.co\S+', '', vnf['description'])
urlUser = urllib.parse.quote(vnf['uploader']) urlUser = urllib.parse.quote(vnf['uploader'])
@ -480,14 +498,14 @@ def embed(video_link, vnf, image):
if vnf['nsfw'] == True: if vnf['nsfw'] == True:
color = "#800020" # Red color = "#800020" # Red
return getTemplate(template,vnf,desc,image,video_link,color,urlDesc,urlUser,urlLink,appNamePost,embedVNF) return make_cached_vnf_response(vnf,getTemplate(template,vnf,desc,image,video_link,color,urlDesc,urlUser,urlLink,appNamePost,embedVNF))
def embedCombined(video_link): def embedCombined(video_link):
vnf,e = vnfFromCacheOrDL(video_link) vnf,e = vnfFromCacheOrDL(video_link)
if vnf != None: if vnf != None:
return embedCombinedVnf(video_link, vnf) return make_cached_vnf_response(vnf,embedCombinedVnf(video_link, vnf))
else: else:
if e is not None: if e is not None:
return message(msgs.failedToScan+msgs.failedToScanExtra+e) return message(msgs.failedToScan+msgs.failedToScanExtra+e)
@ -523,7 +541,7 @@ def embedCombinedVnf(video_link,vnf):
if vnf['nsfw'] == True: if vnf['nsfw'] == True:
color = "#800020" # Red color = "#800020" # Red
return getTemplate('image.html',vnf,desc,image,video_link,color,urlDesc,urlUser,urlLink,appNameSuffix=" - View original tweet for full quality") return make_cached_vnf_response(vnf,getTemplate('image.html',vnf,desc,image,video_link,color,urlDesc,urlUser,urlLink,appNameSuffix=" - View original tweet for full quality"))
def getPollObject(card): def getPollObject(card):

22
vxlogging.py Normal file
View File

@ -0,0 +1,22 @@
IS_DEBUG = False
def info(message):
message = str(message)
print(f" ➤ [ I ] {message}")
def success(message):
message = str(message)
print(f" ➤ [ ✔ ] {message}")
def error(message):
message = str(message)
print(f" ➤ [ ✘ ] {message}")
def warn(message):
message = str(message)
print(f" ➤ [ ! ] {message}")
def debug(message):
if IS_DEBUG:
message = str(message)
print(f" ➤ [ D ] {message}")