mirror of
https://github.com/truh/SlackFeeder.git
synced 2025-08-03 04:23:32 +02:00
136 lines
3.6 KiB
Python
136 lines
3.6 KiB
Python
import json
|
|
import os
|
|
|
|
import bcrypt
|
|
import toml
|
|
|
|
from aiohttp import web
|
|
from aiohttp_basicauth import BasicAuthMiddleware
|
|
from feedgen.feed import FeedGenerator
|
|
from slacker import Slacker
|
|
|
|
|
|
def get_config():
|
|
file_path = os.environ.get("SLACKFEEDER_CONFIG", "/etc/slackfeeder.toml")
|
|
with open(file_path) as file:
|
|
_config = toml.load(file)
|
|
return _config
|
|
|
|
|
|
CONFIG = get_config()
|
|
SLACK = Slacker(
|
|
CONFIG["Slack"]["token"]
|
|
if "token" in CONFIG["Slack"]
|
|
else open(CONFIG["Slack"]["token_file"])
|
|
)
|
|
|
|
|
|
def get_personal_space_history():
|
|
whoami = SLACK.auth.test()
|
|
im_id = [
|
|
x for x in SLACK.im.list().body["ims"] if x["user"] == whoami.body["user_id"]
|
|
][0]["id"]
|
|
return SLACK.im.history(im_id).body
|
|
|
|
|
|
def slack_history_to_feedgen(history):
|
|
fg = FeedGenerator()
|
|
|
|
fg.title(CONFIG["Feed"]["title"])
|
|
fg.id(CONFIG["Feed"]["id"])
|
|
fg.link(CONFIG["Feed"]["link"])
|
|
fg.description(CONFIG["Feed"]["description"])
|
|
|
|
for message in history["messages"]:
|
|
message_id = message.get(
|
|
"client_msg_id", f'{message.get("user")}--{message.get("ts")}'
|
|
)
|
|
|
|
fe = fg.add_entry()
|
|
fe.id(message_id)
|
|
|
|
title = None
|
|
summary = None
|
|
link = []
|
|
|
|
for file in message.get("files", []):
|
|
title = title or file.get("title", None)
|
|
|
|
link_href = file.get("url_private", None)
|
|
link_rel = "alternate"
|
|
link.append(dict(href=link_href, rel=link_rel, title=title))
|
|
|
|
for attachment in message.get("attachments", []):
|
|
title = title or attachment.get("title", None)
|
|
summary = summary or attachment.get("text", None)
|
|
|
|
link_href = attachment.get("from_url", None)
|
|
link_rel = "alternate"
|
|
link.append(dict(href=link_href, rel=link_rel, title=title))
|
|
|
|
fe.title(title or "Untitled")
|
|
fe.summary(summary)
|
|
fe.content(message["text"])
|
|
fe.link(link)
|
|
|
|
return fg
|
|
|
|
|
|
class CustomBasicAuth(BasicAuthMiddleware):
|
|
async def check_credentials(self, username, password):
|
|
try:
|
|
stored_user, hash = (
|
|
CONFIG["Auth"]["htpasswd"]
|
|
if "htpasswd" in CONFIG["Auth"]
|
|
else open(CONFIG["Auth"]["htpasswd_file"]
|
|
).read()).split(":")
|
|
password_matches = bcrypt.checkpw(
|
|
password.encode("utf-8"), hash.encode("utf-8")
|
|
)
|
|
return password_matches and username == stored_user
|
|
except (ValueError, KeyError, TypeError):
|
|
return False
|
|
|
|
|
|
async def handle_rss(request):
|
|
history = get_personal_space_history()
|
|
feed_generator = slack_history_to_feedgen(history)
|
|
return web.Response(
|
|
body=feed_generator.rss_str(pretty=True),
|
|
charset="utf-8",
|
|
content_type="application/rss+xml",
|
|
)
|
|
|
|
|
|
async def handle_atom(request):
|
|
history = get_personal_space_history()
|
|
feed_generator = slack_history_to_feedgen(history)
|
|
return web.Response(
|
|
body=feed_generator.atom_str(pretty=True),
|
|
charset="utf-8",
|
|
content_type="application/atom+xml",
|
|
)
|
|
|
|
|
|
def webapp():
|
|
middlewares = []
|
|
auth_config = CONFIG.get("Auth", False)
|
|
if auth_config and auth_config.get("enable", False):
|
|
print('Enable basic auth middleware')
|
|
auth = CustomBasicAuth()
|
|
middlewares.append(auth)
|
|
|
|
app = web.Application(middlewares=middlewares)
|
|
|
|
app.add_routes([web.get("/rss.xml", handle_rss), web.get("/atom.xml", handle_atom)])
|
|
|
|
return app
|
|
|
|
|
|
def main():
|
|
app = webapp()
|
|
web.run_app(app, **CONFIG['Network'])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|