Caution: This is not a working plugin. Just some code for further discussion within the developer community. Main issue is that we don't (yet) have much of a proper interface to hook into, so this code pokes around in the database and misuses various functions to accomplish the task of posting and replying. Probably circumventing some other code which encompasses these things in regular PieFed for good reasons.
And then I hardcoded several magic numbers so it won't work unless those are replaced with correct instance-specific values. And no guarantee this will not end up in a some loop and post endlessly.
Anyway: use a dedicated user for bots, set the "bot account" flag and consider using a seperate community (preferrably local-only) for testing. And make sure you have database backups.
So: you've been warned not to deploy this "plugin".
Description: Experimenting with the plugin interface
Additional dependencies: None
codeberg_crossposter/init.py
import os
from datetime import datetime, timedelta, timezone
import logging
from app.plugins.hooks import hook
from flask import current_app, g
from app.utils import get_task_session, html_to_text
from app.models import User, Community, Post, Site, PostReply, Notification, CommunityFlair
from app.shared.post import make_post, edit_post
from app.shared.reply import make_reply, edit_reply
from app.shared.tasks import task_selector
from app.constants import POST_TYPE_ARTICLE, POST_TYPE_LINK, SRC_API
from sqlalchemy import or_
import httpx
from bs4 import BeautifulSoup
import re
PLUGIN_NAME = __name__.split('.')[-1]
logger = logging.getLogger(__name__)
@hook("cron_often")
def codeberg_import_loop(data):
"""Hook that runs on a periodic interval"""
from app import redis_client
with current_app.app_context(), get_task_session() as session:
g.site = session.query(Site).get(1)
user = session.query(User).filter_by(user_name="rss_bot", id=999122630, banned=False, deleted=False).one()
res = redis_client.get(f"plugins:{PLUGIN_NAME}:previous_time")
if res is not None:
previous_time = datetime.fromtimestamp(float(res))
else:
logger.warning("Couldn't determine last run")
previous_time = datetime(year=1970, month=1, day=1)
if datetime.now() - previous_time >= timedelta(minutes=15):
# Name, RSS Feed, post to community_id
urls = [("PieFed", "https://codeberg.org/rimu/pyfedi.rss", 999846),
("PieFed-Notifs", "https://codeberg.org/PieFed/piefed-notifs.rss", 999846)]
for feedname, url, community_id in urls:
logger.info(f"[{PLUGIN_NAME}] Fetching Codeberg RSS feed {url}")
community = session.query(Community).filter_by(id=community_id, banned=False).one()
response = httpx.get(url, timeout=5.0)
response.raise_for_status()
soup = BeautifulSoup(response.content, 'xml')
items = soup.find_all(['item','entry'])
for item in items[:30]:
title = item.find('title').text
link = item.find('link').get('href') or item.find('link').text
description = item.find('description').text
content = item.find(['content', 'content:encoded'])
content = content.text if content else None
author = item.find('author').text
published = item.find('pubDate').text
#pub_timestamp = datetime.strptime(published, "%a, %d %b %Y %H:%M:%S %z")
if user and community and title and description and content:
x_push = re.match(r"^(?P<author>\S+) pushed to (?P<branch>\S+) at (?P<repo>\S+)$", title)
x_issue = re.match(r"^(?P<author>\S+) (?P<action>created|opened|closed|merged|reopened|commented on) (?P<category>issue|pull request) (?P<repo>\S+)#(?P<issue>\d+)$", title)
if x_push:
pass
elif x_issue:
author = x_issue.group('author')
action = x_issue.group('action')
category = x_issue.group('category')
if category == "issue":
category = "Issue"
elif category == "pull request":
category = "PR"
repo = x_issue.group('repo')
issue = x_issue.group('issue')
if action == "created" or action == "opened":
y = re.match(r"^(?P<num>\d+)#(?P<title>.+)#$", description)
num = y.group('num')
title = y.group('title')
title = (feedname + " " + category + " #" + issue + ": " + title)[:220]
body = f"{action} {published} by {author}\n\n" + html_to_text(content) + f"\n\n(imported from [{feedname} Feed]({url}) by [rss_bot](user.public_url()))"
previous_posts = session.query(Post).filter(or_(Post.title.startswith(title), Post.url == link), Post.community_id == community.id).all()
if not previous_posts:
post_data = {
'title': title,
'body': body,
'url': link,
'nsfw': False,
'notify_author': True,
'language_id': 2,
'timezone': "Europe/Berlin",
}
uploaded_file = None
user_id, post = make_post(post_data, community, POST_TYPE_LINK if link else POST_TYPE_ARTICLE, SRC_API, auth="Bearer "+user.encode_jwt_token(), uploaded_file=uploaded_file)
session.commit()
logger.info(f"Posted {post.public_url()} {post.title}\n")
else:
for post in previous_posts:
if post.user_id == user.id:
post_data = {
'title': post.title,
'body': post.body,
'url': post.url,
'nsfw': post.nsfw,
'notify_author': post.notify_author,
'language_id': post.language_id,
'timezone': post.timezone,
}
modified = False
z = re.match(r"^" + re.escape(title) + r"( (\[merged\]|\[closed\]|\[reopened\]))*$", post.title)
if not z:
post_data['title'] = title
modified = True
logger.info(f"Post title edited: {post.public_url()} {title}")
if post.body != body:
post_data['body'] = body
modified = True
logger.info(f"Post body edited: {post.public_url()} {title}")
if modified:
user_id, post = edit_post(post_data, post, post.type, SRC_API, user, uploaded_file=None)
session.commit()
elif action == "merged" or action == "closed" or action == "reopened":
title = (feedname + " " + category + " #" + issue + ": " + description)[:220]
previous_posts = session.query(Post).filter(or_(Post.title.startswith(title), Post.url == link), Post.community_id == community.id, Post.user_id == user.id, Post.deleted == False)>
for post in previous_posts:
reply_data = {
'body': f"{action} {published} by {author}",
'notify_author': True,
'language_id': 2,
'distinguished': False,
}
previous_replies = post.replies.filter(PostReply.body == reply_data['body'], PostReply.user_id == user.id, PostReply.deleted == False).all()
if not previous_replies:
user_id, reply = make_reply(reply_data, post, None, SRC_API, auth="Bearer "+user.encode_jwt_token())
post.title = post.title + f" [{action}]";
flair = session.query(CommunityFlair).filter(CommunityFlair.flair == action, CommunityFlair.community_id == post.community_id).one_or_none()
if action == "merged" or action == "closed":
post.flair = list(filter(lambda f: f.flair != "open", post.flair))
elif action == "reopened":
post.flair = list(filter(lambda f: f.flair != "closed", post.flair))
post.title.removesuffix(' [closed] [reopened]')
flair = session.query(CommunityFlair).filter(CommunityFlair.flair == "open", CommunityFlair.community_id == post.community_id).one_or_none()
if flair:
post.flair.append(flair)
logger.info(f"flair {flair.flair} appended")
session.commit()
task_selector('edit_post', post_id=post.id)
logger.info(f"Post edited ({action}): {post.public_url()} {post.title}\n")
elif action == "commented on":
x_link = re.match(r"^(?P<parentlink>\S+)(?P<commentid>#.+)$", link)
parent_link = x_link.group('parentlink')
parent_post = session.query(Post).filter(Post.url == parent_link, Post.community_id == community.id, Post.user_id == user.id, Post.deleted == False).first()
#content.removeprefix("prefix")
content = f"{link}\n{published} by {author}\n\n" + content.split("\n",2)[2]
if parent_post:
reply_data = {
'body': content,
'notify_author': True,
'language_id': 2,
'distinguished': False,
}
previous_reply = parent_post.replies.filter(or_(PostReply.body == reply_data['body'], PostReply.body.startswith(link)), PostReply.user_id == user.id, PostReply.deleted == Fals>
if not previous_reply:
user_id, reply = make_reply(reply_data, parent_post, None, SRC_API, auth="Bearer "+user.encode_jwt_token())
session.commit()
logger.info(f"Reply {reply.public_url()} {reply.body[:50]}\n")
else:
if previous_reply.user_id == user.id:
if previous_reply.body != reply_data['body']:
user_id, reply = edit_reply(reply_data, previous_reply, parent_post, SRC_API, auth="Bearer "+user.encode_jwt_token())
session.commit()
logger.info(f"Comment edited: {previous_reply.public_url()} {previous_reply.body[:50]}\n")
else:
logger.warning(f"Unknown action \"{action}\"")
else:
logger.warning(f"Unknown feed item \"{title}\"")
redis_client.set(f"plugins:{PLUGIN_NAME}:previous_time", datetime.now().timestamp())
notifications = session.query(Notification).filter(Notification.user_id == user.id, Notification.read == False).all()
for notification in notifications:
if notification.targets:
post_id = notification.targets.get('post_id', 0)
comment_id = notification.targets.get('comment_id', 0)
if post_id and comment_id:
print(f"Notification: {post_id} {comment_id}")
post = session.query(Post).filter(Post.id == post_id, Post.deleted == False).one()
if post.user_id == user.id and post.community_id in [999846]:
parent_reply = post.replies.filter(PostReply.id == comment_id, PostReply.deleted == False).one()
if parent_reply:
reply_data = {
'body': "I'm a one-directional bot. Your replies will not be forwarded to Codeberg!",
'notify_author': True,
'language_id': 2,
'distinguished': False,
}
user_id, reply = make_reply(reply_data, post, parent_reply.id, SRC_API, auth="Bearer "+user.encode_jwt_token())
session.commit()
user.unread_notifications -= 1
notification.read = True
session.commit()
logger.info(f"Reply {reply.public_url()} {reply.body[:50]}\n")
return None
def plugin_info():
"""Plugin metadata"""
return {
"name": "Codeberg Crossposter",
"version": "0.0.1",
"description": "Experimenting with the plugin interface",
"license": "AGPL-3.0", # Must be AGPL-compatible
"source_url": "https://piefed.social/post/1306500", # Required
"author": "hendrik"
}