dark-firepit
/
Twxtter-main
Archived
5
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Updates and changes 

Signed-off-by: Chloe Carver-Brown <admin@twxtter.com>
This commit is contained in:
Chloe Carver-Brown 2022-05-18 15:24:27 +01:00
parent 5b58c0d9bc
commit ef42c26a81
No known key found for this signature in database
GPG Key ID: 71938715093BE620
9 changed files with 605 additions and 336 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
.github/ISSUE_TEMPLATE/bug_report.md vendored Normal file
View File

@ -0,0 +1,38 @@
---
name: Bug report
about: Create a report to help us improve
title: "[BUG]"
labels: bug
assignees: ''
---
**Describe the bug**
A clear and concise description of what the bug is.
**To Reproduce**
Steps to reproduce the behavior:
1. Go to '...'
2. Click on '....'
3. Scroll down to '....'
4. See error
**Expected behavior**
A clear and concise description of what you expected to happen.
**Screenshots**
If applicable, add screenshots to help explain your problem.
**Desktop (please complete the following information):**
- OS: [e.g. iOS]
- Browser [e.g. chrome, safari]
- Version [e.g. 22]
**Smartphone (please complete the following information):**
- Device: [e.g. iPhone6]
- OS: [e.g. iOS8.1]
- Browser [e.g. stock browser, safari]
- Version [e.g. 22]
**Additional context**
Add any other context about the problem here.

20
.github/ISSUE_TEMPLATE/feature_request.md vendored Normal file
View File

@ -0,0 +1,20 @@
---
name: Feature request
about: Suggest an idea for this project
title: "[FEATURE REQUEST]"
labels: enhancement
assignees: ''
---
**Is your feature request related to a problem? Please describe.**
A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]
**Describe the solution you'd like**
A clear and concise description of what you want to happen.
**Describe alternatives you've considered**
A clear and concise description of any alternative solutions or features you've considered.
**Additional context**
Add any other context or screenshots about the feature request here.

7
.vscode/settings.json vendored Normal file
View File

@ -0,0 +1,7 @@
{
"python.linting.flake8Enabled": true,
"python.linting.enabled": true,
"python.linting.flake8Args": [
"--ignore=E501, E261, E302, E231, E226, F401, E221, W503, E722, E711, E712, F841",
],
}

67
readme.md
View File

@ -1,54 +1,18 @@
# TwitFix
# Twxtter/sixFix
(A fork of TwitFix)
Basic flask server that serves fixed twitter video embeds to desktop discord by using either the Twitter API or Youtube-DL to grab tweet video information. This also automatically embeds the first link in the text of non video tweets (API Only)
Regarding what happened to TwitFix: http://blog.hirob.in/2022-05-16-Goodbye-TwitFix/
## How to use (discord side)
just put the url to the server, and directly after, the full URL to the tweet you want to embed
**I now have a copy of this running on a Linode server, you can use it via the following url**
```
https://fxtwitter.com/[twitter video url] or [last half of twitter url] (everything past twitter.com/)
https://twxtter.com/[twitter video url] or [last half of twitter url])
```
You can also simply type out 'fx' directly before 'twitter.com' in any valid twitter video url, and that will convert it into a working TwitFix url, for example:
You can also simply type out `s/i/x` on PC and iOS after sending a twitter link to discord, and it will edit the last message (The link) to replace the first i with x.
![example](example.gif)
**Note**: If you enjoy this service, please considering donating via [Ko-Fi](https://ko-fi.com/robin_universe) to help cover server costs
## Child Projects:
[TwitFix-Bot](https://github.com/robinuniverse/TwitFix-Bot) - A discord bot for automatically converting normal twitter links posted by users into twitfix links
[TwitFix-Extension](https://github.com/robinuniverse/TwitFix-Extension) - A browser extention that lets you right click twitter videos to copy a twitfix link to your clipboard
# Monthly Contributors
TwitFix is run for free, period, I have no plans to monetize it directly in any way ( no ads, no premium accounts with more features ) so I rely on donations to keep TwitFix running, and I have created the option to [donate on a monthly basis using my KoFi](https://ko-fi.com/robin_universe#tier16328580186740)
Here's a a list of the people who help to keep this project alive! ( current total monthly - $49!!! )
1. [$3] First Contributor and Twitter Funnyman **Chris Burwell** ( [@countchrisdo](https://twitter.com/countchrisdo) on Twitter )
2. [$9] Previously highest Contributor, Suspciously wealthy furry, and a very loving friend **Vectrobe** ( [@Vectrobe](https://twitter.com/Vectrobe) on Twitter )
3. [$10] New highest monthly contributor, **helloitscrash**!
4. [$6] A Mysterious and **Anonymous** contributor...
5. [$10] One of the highest contributors, **Ryan Vilbrandt**!
6. [$3] **Starcat13**, the one with the coolest sounding name
7. [$5] THE LIGHT THROUGH WHICH GOD SPEAKS TO THIS EARTH: **Statek**
8. [$3] **Impulse**, probably the source cheat
9. [$3] a STRONG contendor for coolest name, "**Lost in Art & Magic**"
**Note**: If you enjoy this service, please considering donating via [Ko-Fi](https://ko-fi.com/twxtter) to help cover server costs
## How to run (server side)
@ -58,7 +22,7 @@ I have included some files to give you a head start on setting this server up wi
### Config
TwitFix generates a config.json in its root directory the first time you run it, the options are:
Twxtter generates a config.json in its root directory the first time you run it, the options are:
**API** - This will be where you put the credentials for your twitter API if you use this method
@ -90,13 +54,19 @@ This project is licensed under the **Do What The Fuck You Want Public License**
## Other stuff
Going to `https://fxtwitter.com/latest/` will present a page that shows the all the latest tweets that were added to the database, use with caution as results may be nsfw! Current page created by @DorukSaga
We check for t.co links in non video tweets, and if one is found, we direct the discord useragent to embed that link directly, this means that twitter links containing youtube / vimeo links will automatically embed those as if you had just directly linked to that content
## Other stuff
Going to `https://twxtter.com/latest/` will present a page that shows the all the latest tweets that were added to the database, use with caution as results may be nsfw! Current page created by @DorukSaga
Using the `/dir/<video-url>` endpoint will return a redirect to the direct MP4 link, this can be useful for downloading a video
Using the `/dl/<video-url>` or appending a `.mp4` will make the server download the video and return a static, locally hosted copy
Using the subdomain `d.fxtwitter.com/<video-url>` will redirect to a direct MP4 url hosted on Twitter
Using the subdomain `d.twxtter.com/<video-url>` will redirect to a direct MP4 url hosted on Twitter
Using the `/info/<video-url>` endpoint will return a json that contains all video info that youtube-dl can grab about any given video
@ -108,6 +78,7 @@ Using `/api/top/` will return a json with the most hit tweet in the database. Ta
Using `/api/stats/` will return a json with some stats about TwitFix's activity (embeds, new cached links, API hits, downloads). Takes param `?=date"YYYY-MM-DD"` to return a specific day, otherwise will return today's stats to far
Advanced embeds are provided via a `/oembed.json?` endpoint - This is manually pointing at my server in `/templates/index.html` and should be changed from `https://fxtwitter.com/` to whatever your domain is
Advanced embeds are provided via a `/oembed.json?` endpoint - This is manually pointing at my server in `/templates/index.html` and should be changed from `https://twxtter.com/` to whatever your domain is
We check for t.co links in non video tweets, and if one is found, we direct the discord useragent to embed that link directly, this means that twitter links containing youtube / vimeo links will automatically embed those as if you had just directly linked to that content
# NOTICE
## This is _**NOT**_ actively monitored by anyone working on Twxtter. All tweets are public ally accessible.

BIN
static/coconut.jpg Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 13 KiB

BIN
static/favicon.ico
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 15 KiB

After

Width:  |  Height:  |  Size: 4.2 KiB

2
twitfix.ini
View File

@ -4,7 +4,7 @@ module = wsgi:app
master = true
processes = 5
socket = twitfix.sock
socket = sixFix.sock
chmod-socket = 660
vacuum = true

597
twitfix.py
View File

@ -1,4 +1,15 @@
from flask import Flask, render_template, request, redirect, Response, send_from_directory, url_for, send_file, make_response, jsonify
from flask import (
Flask,
render_template,
request,
redirect,
Response,
send_from_directory,
url_for,
send_file,
make_response,
jsonify,
)
from flask_cors import CORS
import youtube_dl
import textwrap
@ -29,7 +40,8 @@ generate_embed_user_agents = [
"Mozilla/5.0 (compatible; Discordbot/2.0; +https://discordapp.com)",
"TelegramBot (like TwitterBot)",
"Mozilla/5.0 (compatible; January/1.0; +https://gitlab.insrt.uk/revolt/january)",
"test"]
"test",
]
# Read config from config.json. If it does not exist, create new.
if not os.path.exists("config.json"):
@ -43,13 +55,14 @@ if not os.path.exists("config.json"):
"color": "#43B581",
"appname": "TwitFix",
"repo": "https://github.com/robinuniverse/twitfix",
"url": "https://fxtwitter.com"
"url": "https://fxtwitter.com",
},
"api":{"api_key":"[api_key goes here]",
"api": {
"api_key": "[api_key goes here]",
"api_secret": "[api_secret goes here]",
"access_token": "[access_token goes here]",
"access_secret":"[access_secret goes here]"
}
"access_secret": "[access_secret goes here]",
},
}
json.dump(default_config, outfile, indent=4, sort_keys=True)
@ -61,11 +74,16 @@ else:
f.close()
# If method is set to API or Hybrid, attempt to auth with the Twitter API
if config['config']['method'] in ('api', 'hybrid'):
auth = twitter.oauth.OAuth(config['api']['access_token'], config['api']['access_secret'], config['api']['api_key'], config['api']['api_secret'])
if config["config"]["method"] in ("api", "hybrid"):
auth = twitter.oauth.OAuth(
config["api"]["access_token"],
config["api"]["access_secret"],
config["api"]["api_key"],
config["api"]["api_secret"],
)
twitter_api = twitter.Twitter(auth=auth)
link_cache_system = config['config']['link_cache']
link_cache_system = config["config"]["link_cache"]
if link_cache_system == "json":
link_cache = {}
@ -74,53 +92,97 @@ if link_cache_system == "json":
default_link_cache = {"test": "test"}
json.dump(default_link_cache, outfile, indent=4, sort_keys=True)
f = open('links.json',)
f = open(
"links.json",
)
link_cache = json.load(f)
f.close()
elif link_cache_system == "db":
client = pymongo.MongoClient(config['config']['database'], connect=False)
table = config['config']['table']
client = pymongo.MongoClient(config["config"]["database"], connect=False)
table = config["config"]["table"]
db = client[table]
@app.route('/bidoof/')
@app.route("/bidoof/")
def bidoof():
return redirect("https://cdn.discordapp.com/attachments/291764448757284885/937343686927319111/IMG_20211226_202956_163.webp", 301)
return redirect(
"https://cdn.discordapp.com/attachments/291764448757284885/937343686927319111/IMG_20211226_202956_163.webp",
301,
)
@app.route('/discord/')
@app.route("/thisiswhatrunstwxtter/")
def discord():
return redirect("https://discord.gg/ztz2hHwZXv", 301)
return redirect(
"https://cdn.discordapp.com/attachments/932899721767448607/976472084974829568/unknown.png",
301,
)
@app.route('/stats/')
@app.route("/stats/")
def statsPage():
today = str(date.today())
stats = getStats(today)
return render_template('stats.html', embeds=stats['embeds'], downloadss=stats['downloads'], api=stats['api'], linksCached=stats['linksCached'], date=today)
return render_template(
"stats.html",
embeds=stats["embeds"],
downloadss=stats["downloads"],
api=stats["api"],
linksCached=stats["linksCached"],
date=today,
)
@app.route('/latest/')
@app.route("/latest/")
def latest():
return render_template('latest.html')
return render_template("latest.html")
@app.route('/copy.svg') # Return a SVG needed for Latest
@app.route("/copy.svg") # Return a SVG needed for Latest
def icon():
return send_from_directory(os.path.join(app.root_path, 'static'),
'copy.svg',mimetype='image/svg+xml')
return send_from_directory(
os.path.join(app.root_path, "static"), "copy.svg", mimetype="image/svg+xml"
)
@app.route('/font.ttf') # Return a font needed for Latest
@app.route("/font.ttf") # Return a font needed for Latest
def font():
return send_from_directory(os.path.join(app.root_path, 'static'),
'NotoColorEmoji.ttf',mimetype='application/octet-stream')
return send_from_directory(
os.path.join(app.root_path, "static"),
"NotoColorEmoji.ttf",
mimetype="application/octet-stream",
)
@app.route('/top/') # Try to return the most hit video
@app.route("/top/") # Try to return the most hit video
def top():
vnf = db.linkCache.find_one(sort = [('hits', pymongo.DESCENDING)])
desc = re.sub(r' http.*t\.co\S+', '', vnf['description'])
urlUser = urllib.parse.quote(vnf['uploader'])
vnf = db.linkCache.find_one(sort=[("hits", pymongo.DESCENDING)])
desc = re.sub(r" http.*t\.co\S+", "", vnf["description"])
urlUser = urllib.parse.quote(vnf["uploader"])
urlDesc = urllib.parse.quote(desc)
urlLink = urllib.parse.quote(vnf['url'])
print(" ➤ [ ✔ ] Top video page loaded: " + vnf['tweet'] )
return render_template('inline.html', page="Top", vidlink=vnf['url'], vidurl=vnf['url'], desc=desc, pic=vnf['thumbnail'], user=vnf['uploader'], video_link=vnf['url'], color=config['config']['color'], appname=config['config']['appname'], repo=config['config']['repo'], url=config['config']['url'], urlDesc=urlDesc, urlUser=urlUser, urlLink=urlLink, tweet=vnf['tweet'])
urlLink = urllib.parse.quote(vnf["url"])
print(" ➤ [ ✔ ] Top video page loaded: " + vnf["tweet"])
return render_template(
"inline.html",
page="Top",
vidlink=vnf["url"],
vidurl=vnf["url"],
desc=desc,
pic=vnf["thumbnail"],
user=vnf["uploader"],
video_link=vnf["url"],
color=config["config"]["color"],
appname=config["config"]["appname"],
repo=config["config"]["repo"],
url=config["config"]["url"],
urlDesc=urlDesc,
urlUser=urlUser,
urlLink=urlLink,
tweet=vnf["tweet"],
)
@app.route('/api/latest/') # Return some raw VNF data sorted by top tweets
@app.route("/api/latest/") # Return some raw VNF data sorted by top tweets
def apiLatest():
bigvnf = []
@ -130,16 +192,25 @@ def apiLatest():
if tweets > 15:
tweets = 1
vnf = db.linkCache.find(sort = [('_id', pymongo.DESCENDING)]).skip(tweets * page).limit(tweets)
vnf = (
db.linkCache.find(sort=[("_id", pymongo.DESCENDING)])
.skip(tweets * page)
.limit(tweets)
)
for r in vnf:
bigvnf.append(r)
print(" ➤ [ ✔ ] Latest video API called")
addToStat('api')
return Response(response=json.dumps(bigvnf, default=str), status=200, mimetype="application/json")
addToStat("api")
return Response(
response=json.dumps(bigvnf, default=str),
status=200,
mimetype="application/json",
)
@app.route('/api/top/') # Return some raw VNF data sorted by top tweets
@app.route("/api/top/") # Return some raw VNF data sorted by top tweets
def apiTop():
bigvnf = []
@ -149,36 +220,57 @@ def apiTop():
if tweets > 15:
tweets = 1
vnf = db.linkCache.find(sort = [('hits', pymongo.DESCENDING )]).skip(tweets * page).limit(tweets)
vnf = (
db.linkCache.find(sort=[("hits", pymongo.DESCENDING)])
.skip(tweets * page)
.limit(tweets)
)
for r in vnf:
bigvnf.append(r)
print(" ➤ [ ✔ ] Top video API called")
addToStat('api')
return Response(response=json.dumps(bigvnf, default=str), status=200, mimetype="application/json")
addToStat("api")
return Response(
response=json.dumps(bigvnf, default=str),
status=200,
mimetype="application/json",
)
@app.route('/api/stats/') # Return a json of a usage stats for a given date (defaults to today)
@app.route(
"/api/stats/"
) # Return a json of a usage stats for a given date (defaults to today)
def apiStats():
try:
addToStat('api')
addToStat("api")
today = str(date.today())
desiredDate = request.args.get("date", default=today, type=str)
stat = getStats(desiredDate)
print(" ➤ [ ✔ ] Stats API called")
return Response(response=json.dumps(stat, default=str), status=200, mimetype="application/json")
return Response(
response=json.dumps(stat, default=str),
status=200,
mimetype="application/json",
)
except:
print(" ➤ [ ✔ ] Stats API failed")
@app.route('/') # If the useragent is discord, return the embed, if not, redirect to configured repo directly
def default():
user_agent = request.headers.get('user-agent')
if user_agent in generate_embed_user_agents:
return message("TwitFix is an attempt to fix twitter video embeds in discord! created by Robin Universe :)\n\n💖\n\nClick me to be redirected to the repo!")
else:
return redirect(config['config']['repo'], 301)
@app.route('/oembed.json') #oEmbed endpoint
@app.route(
"/"
) # If the useragent is discord, return the embed, if not, redirect to configured repo directly
def default():
user_agent = request.headers.get("user-agent")
if user_agent in generate_embed_user_agents:
return message(
"Twxtter is an attempt to fix twitter video embeds in discord! :)\n\n💖\n\nClick me to be redirected to the repo!"
)
else:
return redirect(config["config"]["repo"], 301)
@app.route("/oembed.json") # oEmbed endpoint
def oembedend():
desc = request.args.get("desc", None)
user = request.args.get("user", None)
@ -186,19 +278,24 @@ def oembedend():
ttype = request.args.get("ttype", None)
return oEmbedGen(desc, user, link, ttype)
@app.route('/<path:sub_path>') # Default endpoint used by everything
@app.route("/<path:sub_path>") # Default endpoint used by everything
def twitfix(sub_path):
user_agent = request.headers.get('user-agent')
user_agent = request.headers.get("user-agent")
match = pathregex.search(sub_path)
print(request.url)
if request.url.startswith("https://d.fx"): # Matches d.fx? Try to give the user a direct link
if request.url.startswith(
"https://d.fx"
): # Matches d.fx? Try to give the user a direct link
if user_agent in generate_embed_user_agents:
print(" ➤ [ D ] d.fx link shown to discord user-agent!")
if request.url.endswith(".mp4") and "?" not in request.url:
return dl(sub_path)
else:
return message("To use a direct MP4 link in discord, remove anything past '?' and put '.mp4' at the end")
return message(
"To use a direct MP4 link in discord, remove anything past '?' and put '.mp4' at the end"
)
else:
print(" ➤ [ R ] Redirect to MP4 using d.fxtwitter.com")
return dir(sub_path)
@ -226,11 +323,27 @@ def twitfix(sub_path):
vnf = link_to_vnf_from_api(clean.replace(".json", ""))
if user_agent in generate_embed_user_agents:
return message("VNF Data: ( discord useragent preview )\n\n"+ json.dumps(vnf, default=str))
return message(
"VNF Data: ( discord useragent preview )\n\n"
+ json.dumps(vnf, default=str)
)
else:
return Response(response=json.dumps(vnf, default=str), status=200, mimetype="application/json")
return Response(
response=json.dumps(vnf, default=str),
status=200,
mimetype="application/json",
)
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
if "?" not in request.url:
@ -238,9 +351,29 @@ def twitfix(sub_path):
else:
clean = twitter_url
image = ( int(request.url[-1]) - 1 )
image = int(request.url[-1]) - 1
return embed_video(clean, image)
elif (
request.url.endswith("/1p")
or request.url.endswith("/2p")
or request.url.endswith("/3p")
or request.url.endswith("/4p")
or request.url.endswith("%2F1p")
or request.url.endswith("%2F2p")
or request.url.endswith("%2F3p")
or request.url.endswith("%2F4p")
):
twitter_url = "https://twitter.com/" + sub_path
if "?" not in request.url:
clean = twitter_url[:-3]
else:
clean = twitter_url
image = int(request.url[-2]) - 1
return embed_video(clean, image, raw=True)
if match is not None:
twitter_url = sub_path
@ -257,25 +390,32 @@ def twitfix(sub_path):
else:
return message("This doesn't appear to be a twitter URL")
@app.route('/other/<path:sub_path>') # Show all info that Youtube-DL can get about a video as a json
@app.route(
"/other/<path:sub_path>"
) # Show all info that Youtube-DL can get about a video as a json
def other(sub_path):
otherurl = request.url.split("/other/", 1)[1].replace(":/", "://")
print(" ➤ [ OTHER ] Other URL embed attempted: " + otherurl)
res = embed_video(otherurl)
return res
@app.route('/info/<path:sub_path>') # Show all info that Youtube-DL can get about a video as a json
@app.route(
"/info/<path:sub_path>"
) # Show all info that Youtube-DL can get about a video as a json
def info(sub_path):
infourl = request.url.split("/info/", 1)[1].replace(":/", "://")
print(" ➤ [ INFO ] Info data requested: " + infourl)
with youtube_dl.YoutubeDL({'outtmpl': '%(id)s.%(ext)s'}) as ydl:
with youtube_dl.YoutubeDL({"outtmpl": "%(id)s.%(ext)s"}) as ydl:
result = ydl.extract_info(infourl, download=False)
return result
@app.route('/dl/<path:sub_path>') # Download the tweets video, and rehost it
@app.route("/dl/<path:sub_path>") # Download the tweets video, and rehost it
def dl(sub_path):
print(' ➤ [[ !!! TRYING TO DOWNLOAD FILE !!! ]] Downloading file from ' + sub_path)
print(" ➤ [[ !!! TRYING TO DOWNLOAD FILE !!! ]] Downloading file from " + sub_path)
url = sub_path
match = pathregex.search(url)
if match is not None:
@ -284,28 +424,39 @@ def dl(sub_path):
twitter_url = "https://twitter.com/" + url
mp4link = direct_video_link(twitter_url)
filename = (sub_path.split('/')[-1].split('.mp4')[0] + '.mp4')
filename = sub_path.split("/")[-1].split(".mp4")[0] + ".mp4"
PATH = ( './static/' + filename )
PATH = "./static/" + filename
if os.path.isfile(PATH) and os.access(PATH, os.R_OK):
print(" ➤ [[ FILE EXISTS ]]")
else:
print(" ➤ [[ FILE DOES NOT EXIST, DOWNLOADING... ]]")
addToStat('downloads')
addToStat("downloads")
mp4file = urllib.request.urlopen(mp4link)
with open(('/home/robin/twitfix/static/' + filename), 'wb') as output:
with open(("/home/robin/twitfix/static/" + filename), "wb") as output:
output.write(mp4file.read())
print(' ➤ [[ PRESENTING FILE: '+ filename +', URL: https://fxtwitter.com/static/'+ filename +' ]]')
r = make_response(send_file(('static/' + filename), mimetype='video/mp4', max_age=100))
r.headers['Content-Type'] = 'video/mp4'
r.headers['Sec-Fetch-Site'] = 'none'
r.headers['Sec-Fetch-User'] = '?1'
print(
" ➤ [[ PRESENTING FILE: "
+ filename
+ ", URL: https://fxtwitter.com/static/"
+ filename
+ " ]]"
)
r = make_response(
send_file(("static/" + filename), mimetype="video/mp4", max_age=100)
)
r.headers["Content-Type"] = "video/mp4"
r.headers["Sec-Fetch-Site"] = "none"
r.headers["Sec-Fetch-User"] = "?1"
return r
@app.route('/dir/<path:sub_path>') # Try to return a direct link to the MP4 on twitters servers
@app.route(
"/dir/<path:sub_path>"
) # Try to return a direct link to the MP4 on twitters servers
def dir(sub_path):
user_agent = request.headers.get('user-agent')
user_agent = request.headers.get("user-agent")
url = sub_path
match = pathregex.search(url)
if match is not None:
@ -324,10 +475,15 @@ def dir(sub_path):
else:
return redirect(url, 301)
@app.route('/favicon.ico') # This shit don't work
@app.route("/favicon.ico") # This shit don't work
def favicon():
return send_from_directory(os.path.join(app.root_path, 'static'),
'favicon.ico',mimetype='image/vnd.microsoft.icon')
return send_from_directory(
os.path.join(app.root_path, "static"),
"favicon.ico",
mimetype="image/vnd.microsoft.icon",
)
def direct_video(video_link): # Just get a redirect to a MP4 link from any tweet link
cached_vnf = getVnfFromLinkCache(video_link)
@ -335,63 +491,87 @@ def direct_video(video_link): # Just get a redirect to a MP4 link from any tweet
try:
vnf = link_to_vnf(video_link)
addVnfToLinkCache(video_link, vnf)
return redirect(vnf['url'], 301)
print(" ➤ [ D ] Redirecting to direct URL: " + vnf['url'])
return redirect(vnf["url"], 301)
print(" ➤ [ D ] Redirecting to direct URL: " + vnf["url"])
except Exception as e:
print(e)
return message("Failed to scan your link!")
else:
return redirect(cached_vnf['url'], 301)
print(" ➤ [ D ] Redirecting to direct URL: " + vnf['url'])
return redirect(cached_vnf["url"], 301)
print(" ➤ [ D ] Redirecting to direct URL: " + vnf["url"])
def direct_video_link(video_link): # Just get a redirect to a MP4 link from any tweet link
def direct_video_link(
video_link,
): # Just get a redirect to a MP4 link from any tweet link
cached_vnf = getVnfFromLinkCache(video_link)
if cached_vnf == None:
try:
vnf = link_to_vnf(video_link)
addVnfToLinkCache(video_link, vnf)
return vnf['url']
print(" ➤ [ D ] Redirecting to direct URL: " + vnf['url'])
return vnf["url"]
print(" ➤ [ D ] Redirecting to direct URL: " + vnf["url"])
except Exception as e:
print(e)
return message("Failed to scan your link!")
else:
return cached_vnf['url']
print(" ➤ [ D ] Redirecting to direct URL: " + vnf['url'])
return cached_vnf["url"]
print(" ➤ [ D ] Redirecting to direct URL: " + vnf["url"])
def addToStat(stat):
# print(stat)
today = str(date.today())
try:
collection = db.stats.find_one({'date': today})
delta = ( collection[stat] + 1 )
collection = db.stats.find_one({"date": today})
delta = collection[stat] + 1
query = {"date": today}
change = {"$set": {stat: delta}}
out = db.stats.update_one(query, change)
except:
collection = db.stats.insert_one({'date': today, "embeds" : 1, "linksCached" : 1, "api" : 1, "downloads" : 1 })
collection = db.stats.insert_one(
{"date": today, "embeds": 1, "linksCached": 1, "api": 1, "downloads": 1}
)
def getStats(day):
collection = db.stats.find_one({'date': day})
collection = db.stats.find_one({"date": day})
return collection
def embed_video(video_link, image=0): # Return Embed from any tweet link
def embed_video(video_link, image=0, raw=False): # Return Embed from any tweet link
cached_vnf = getVnfFromLinkCache(video_link)
if cached_vnf == None:
try:
vnf = link_to_vnf(video_link)
addVnfToLinkCache(video_link, vnf)
return embed(video_link, vnf, image)
return embed(video_link, vnf, image, raw)
except Exception as e:
print(e)
return message("Failed to scan your link!")
else:
return embed(video_link, cached_vnf, image)
return embed(video_link, cached_vnf, image, raw)
def tweetInfo(url, tweet="", desc="", thumb="", uploader="", screen_name="", pfp="", tweetType="", images="", hits=0, likes=0, rts=0, time="", qrt={}, nsfw=False): # Return a dict of video info with default values
def tweetInfo(
url,
tweet="",
desc="",
thumb="",
uploader="",
screen_name="",
pfp="",
tweetType="",
images="",
hits=0,
likes=0,
rts=0,
time="",
qrt={},
nsfw=False,
): # Return a dict of video info with default values
vnf = {
"tweet": tweet,
"url": url,
@ -407,13 +587,16 @@ def tweetInfo(url, tweet="", desc="", thumb="", uploader="", screen_name="", pfp
"rts": rts,
"time": time,
"qrt": qrt,
"nsfw" : nsfw
"nsfw": nsfw,
}
return vnf
def link_to_vnf_from_api(video_link):
print(" ➤ [ + ] Attempting to download tweet info from Twitter API")
twid = int(re.sub(r'\?.*$','',video_link.rsplit("/", 1)[-1])) # gets the tweet ID as a int from the passed url
twid = int(
re.sub(r"\?.*$", "", video_link.rsplit("/", 1)[-1])
) # gets the tweet ID as a int from the passed url
tweet = twitter_api.statuses.show(_id=twid, tweet_mode="extended")
# For when I need to poke around and see what a tweet looks like
# print(tweet)
@ -421,83 +604,97 @@ def link_to_vnf_from_api(video_link):
print(" ➤ [ + ] Tweet Type: " + tweetType(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 tweet['extended_entities']['media'][0]['video_info']['variants']:
if tweet["extended_entities"]["media"][0]["video_info"]["variants"]:
best_bitrate = 0
thumb = tweet['extended_entities']['media'][0]['media_url']
for video in tweet['extended_entities']['media'][0]['video_info']['variants']:
if video['content_type'] == "video/mp4" and video['bitrate'] > best_bitrate:
url = video['url']
thumb = tweet["extended_entities"]["media"][0]["media_url"]
for video in tweet["extended_entities"]["media"][0]["video_info"][
"variants"
]:
if (
video["content_type"] == "video/mp4"
and video["bitrate"] > best_bitrate
):
url = video["url"]
elif tweetType(tweet) == "Text":
url = ""
thumb = ""
else:
imgs = ["", "", "", "", ""]
i = 0
for media in tweet['extended_entities']['media']:
imgs[i] = media['media_url_https']
for media in tweet["extended_entities"]["media"]:
imgs[i] = media["media_url_https"]
i = i + 1
# print(imgs)
imgs[4] = str(i)
url = ""
images = imgs
thumb = tweet['extended_entities']['media'][0]['media_url_https']
thumb = tweet["extended_entities"]["media"][0]["media_url_https"]
qrt = {}
if 'quoted_status' in tweet:
qrt['desc'] = tweet['quoted_status']['full_text']
qrt['handle'] = tweet['quoted_status']['user']['name']
qrt['screen_name'] = tweet['quoted_status']['user']['screen_name']
if "quoted_status" in tweet:
qrt["desc"] = tweet["quoted_status"]["full_text"]
qrt["handle"] = tweet["quoted_status"]["user"]["name"]
qrt["screen_name"] = tweet["quoted_status"]["user"]["screen_name"]
text = tweet['full_text']
text = tweet["full_text"]
if 'possibly_sensitive' in tweet:
nsfw = tweet['possibly_sensitive']
if "possibly_sensitive" in tweet:
nsfw = tweet["possibly_sensitive"]
else:
nsfw = False
vnf = tweetInfo(
url,
video_link,
text, thumb,
tweet['user']['name'],
tweet['user']['screen_name'],
tweet['user']['profile_image_url'],
text,
thumb,
tweet["user"]["name"],
tweet["user"]["screen_name"],
tweet["user"]["profile_image_url"],
tweetType(tweet),
likes=tweet['favorite_count'],
rts=tweet['retweet_count'],
time=tweet['created_at'],
likes=tweet["favorite_count"],
rts=tweet["retweet_count"],
time=tweet["created_at"],
qrt=qrt,
images=imgs,
nsfw=nsfw
nsfw=nsfw,
)
return vnf
def link_to_vnf_from_youtubedl(video_link):
print(" ➤ [ X ] Attempting to download tweet info via YoutubeDL: " + video_link)
with youtube_dl.YoutubeDL({'outtmpl': '%(id)s.%(ext)s'}) as ydl:
with youtube_dl.YoutubeDL({"outtmpl": "%(id)s.%(ext)s"}) as ydl:
result = ydl.extract_info(video_link, download=False)
vnf = tweetInfo(result['url'], video_link, result['description'].rsplit(' ',1)[0], result['thumbnail'], result['uploader'])
vnf = tweetInfo(
result["url"],
video_link,
result["description"].rsplit(" ", 1)[0],
result["thumbnail"],
result["uploader"],
)
return vnf
def link_to_vnf(video_link): # Return a VideoInfo object or die trying
if config['config']['method'] == 'hybrid':
if config["config"]["method"] == "hybrid":
try:
return link_to_vnf_from_api(video_link)
except Exception as e:
print(" ➤ [ !!! ] API Failed")
print(e)
return link_to_vnf_from_youtubedl(video_link)
elif config['config']['method'] == 'api':
elif config["config"]["method"] == "api":
try:
return link_to_vnf_from_api(video_link)
except Exception as e:
print(" ➤ [ X ] API Failed")
print(e)
return None
elif config['config']['method'] == 'youtube-dl':
elif config["config"]["method"] == "youtube-dl":
try:
return link_to_vnf_from_youtubedl(video_link)
except Exception as e:
@ -505,21 +702,29 @@ def link_to_vnf(video_link): # Return a VideoInfo object or die trying
print(e)
return None
else:
print("Please set the method key in your config file to 'api' 'youtube-dl' or 'hybrid'")
print(
"Please set the method key in your config file to 'api' 'youtube-dl' or 'hybrid'"
)
return None
def getVnfFromLinkCache(video_link):
if link_cache_system == "db":
collection = db.linkCache
vnf = collection.find_one({'tweet': video_link})
vnf = collection.find_one({"tweet": video_link})
# print(vnf)
if vnf != None:
hits = ( vnf['hits'] + 1 )
print(" ➤ [ ✔ ] Link located in DB cache. " + "hits on this link so far: [" + str(hits) + "]")
query = { 'tweet': video_link }
hits = vnf["hits"] + 1
print(
" ➤ [ ✔ ] Link located in DB cache. "
+ "hits on this link so far: ["
+ str(hits)
+ "]"
)
query = {"tweet": video_link}
change = {"$set": {"hits": hits}}
out = db.linkCache.update_one(query, change)
addToStat('embeds')
addToStat("embeds")
return vnf
else:
print(" ➤ [ X ] Link not in DB cache")
@ -533,12 +738,13 @@ def getVnfFromLinkCache(video_link):
print(" ➤ [ X ] Link not in json cache")
return None
def addVnfToLinkCache(video_link, vnf):
if link_cache_system == "db":
try:
out = db.linkCache.insert_one(vnf)
print(" ➤ [ + ] Link added to DB cache ")
addToStat('linksCached')
addToStat("linksCached")
return True
except Exception:
print(" ➤ [ X ] Failed to add link to DB cache")
@ -549,79 +755,105 @@ def addVnfToLinkCache(video_link, vnf):
json.dump(link_cache, outfile, indent=4, sort_keys=True)
return None
def message(text):
return render_template(
'default.html',
"default.html",
message=text,
color = config['config']['color'],
appname = config['config']['appname'],
repo = config['config']['repo'],
url = config['config']['url'] )
color=config["config"]["color"],
appname=config["config"]["appname"],
repo=config["config"]["repo"],
url=config["config"]["url"],
)
def embed(video_link, vnf, image):
print(" ➤ [ E ] Embedding " + vnf['type'] + ": " + vnf['url'])
desc = re.sub(r' http.*t\.co\S+', '', vnf['description'])
urlUser = urllib.parse.quote(vnf['uploader'])
def embed(video_link, vnf, image, raw=False):
print(" ➤ [ E ] Embedding " + vnf["type"] + ": " + vnf["url"])
desc = re.sub(r" http.*t\.co\S+", "", vnf["description"])
urlUser = urllib.parse.quote(vnf["uploader"])
urlDesc = urllib.parse.quote(desc)
urlLink = urllib.parse.quote(video_link)
likeDisplay = ("\n\n💖 " + str(vnf['likes']) + " 🔁 " + str(vnf['rts']) + "\n")
likeDisplay = "\n\n💖 " + str(vnf["likes"]) + " 🔁 " + str(vnf["rts"]) + "\n"
imagecount = "Twitter"
try:
if vnf['type'] == "":
if vnf["type"] == "":
desc = desc
elif vnf['type'] == "Video":
elif vnf["type"] == "Video":
desc = desc
elif vnf['qrt'] == {}: # Check if this is a QRT and modify the description
desc = (desc + likeDisplay)
elif vnf["qrt"] == {}: # Check if this is a QRT and modify the description
desc = desc + likeDisplay
else:
qrtDisplay = ("\n─────────────\n ➤ QRT of " + vnf['qrt']['handle'] + " (@" + vnf['qrt']['screen_name'] + "):\n─────────────\n'" + vnf['qrt']['desc'] + "'")
desc = (desc + qrtDisplay + likeDisplay)
qrtDisplay = (
"\n─────────────\n ➤ QRT of "
+ vnf["qrt"]["handle"]
+ " (@"
+ vnf["qrt"]["screen_name"]
+ "):\n─────────────\n'"
+ vnf["qrt"]["desc"]
+ "'"
)
desc = desc + qrtDisplay + likeDisplay
except:
vnf['likes'] = 0; vnf['rts'] = 0; vnf['time'] = 0
print(' ➤ [ X ] Failed QRT check - old VNF object')
vnf["likes"] = 0
vnf["rts"] = 0
vnf["time"] = 0
print(" ➤ [ X ] Failed QRT check - old VNF object")
if vnf['type'] == "Text": # Change the template based on tweet type
template = 'text.html'
if vnf['type'] == "Image":
image = vnf['images'][image]
template = 'image.html'
if vnf['type'] == "Video":
urlDesc = urllib.parse.quote(textwrap.shorten(desc, width=220, placeholder="..."))
template = 'video.html'
if vnf['type'] == "":
urlDesc = urllib.parse.quote(textwrap.shorten(desc, width=220, placeholder="..."))
template = 'video.html'
if vnf["type"] == "Text": # Change the template based on tweet type
template = "text.html"
if vnf["type"] == "Image":
image = vnf["images"][image]
if vnf["images"][4] != "1":
imagecount = "Twitter (" + vnf["images"][4] + " images in post)"
if raw == True:
template = "img.html"
else:
template = "image.html"
if vnf["type"] == "Video":
urlDesc = urllib.parse.quote(
textwrap.shorten(desc, width=220, placeholder="...")
)
template = "video.html"
if vnf["type"] == "":
urlDesc = urllib.parse.quote(
textwrap.shorten(desc, width=220, placeholder="...")
)
template = "video.html"
color = "#7FFFD4" # Green
if vnf['nsfw'] == True:
if vnf["nsfw"] == True:
color = "#800020" # Red
return render_template(
template,
likes = vnf['likes'],
rts = vnf['rts'],
time = vnf['time'],
screenName = vnf['screen_name'],
vidlink = vnf['url'],
pfp = vnf['pfp'],
vidurl = vnf['url'],
likes=vnf["likes"],
rts=vnf["rts"],
time=vnf["time"],
screenName=vnf["screen_name"],
vidlink=vnf["url"],
pfp=vnf["pfp"],
vidurl=vnf["url"],
desc=desc,
pic=image,
user = vnf['uploader'],
imagecount=imagecount,
user=vnf["uploader"],
video_link=video_link,
color=color,
appname = config['config']['appname'],
repo = config['config']['repo'],
url = config['config']['url'],
appname=config["config"]["appname"],
repo=config["config"]["repo"],
url=config["config"]["url"],
urlDesc=urlDesc,
urlUser=urlUser,
urlLink = urlLink )
urlLink=urlLink,
)
def tweetType(tweet): # Are we dealing with a Video, Image, or Text tweet?
if 'extended_entities' in tweet:
if 'video_info' in tweet['extended_entities']['media'][0]:
if "extended_entities" in tweet:
if "video_info" in tweet["extended_entities"]["media"][0]:
out = "Video"
else:
out = "Image"
@ -635,15 +867,16 @@ def oEmbedGen(description, user, video_link, ttype):
out = {
"type": ttype,
"version": "1.0",
"provider_name" : config['config']['appname'],
"provider_url" : config['config']['repo'],
"provider_name": config["config"]["appname"],
"provider_url": config["config"]["repo"],
"title": description,
"author_name": user,
"author_url" : video_link
"author_url": video_link,
}
return out
if __name__ == "__main__":
app.config['SERVER_NAME']='localhost:80'
app.run(host='0.0.0.0')
app.config["SERVER_NAME"] = "localhost:80"
app.run(host="0.0.0.0")

2
wsgi.py
View File

@ -2,4 +2,4 @@ from twitfix import app
if __name__ == "__main__":
# listen on 0.0.0.0 to facilitate testing with real services
app.run(host='0.0.0.0')
app.run(host="0.0.0.0")