RSS-to-Rocket/wiki rss to rocket chat.py

import time
import feedparser
import requests
import sqlite3
import logging
import os
from bs4 import BeautifulSoup
import json
import re
import argparse

# Example structure from Wiki
# {
# "title": "Koozali SME Server Debugging (French)",
# "recent_changes": {
# "last_updated": "February 16, 2025, at 20:26 GMT",
# "author": "Gieres"
# },
# "summary_of_changes": {
# "description": "The recent update to the page discusses using Perl with Visual Studio.",
# "server_commands": {
# "removed_commands": "yum --enablerepo=* install gcc gcc-c++ perl-App-cpanminus perl-AnyEvent-AIO perl-Coro",
# "updated_commands": "yum --enablerepo=* install gcc gcc-c++"
# },
# "package_changes": {
# "changed_from": "cpanm Class::Refresh",
# "changed_to": "cpan App::cpanminus",
# "additional_commands": [
# "cpanm Compiler::Lexer",
# "cpanm Perl::LanguageServer"
# ]
# }
# },
# "links": {
# "full_revision": "https://wiki.koozali.org/index.php?title=Koozali_SME_Server_Debugging/fr&diff=43309&oldid=43107",
# "comments": "https://wiki.koozali.org/Talk:Koozali_SME_Server_Debugging/fr"
# }
# }

def join_lines_with_newline(lines):
    return "\n".join(lines)

# Wiki RSS feed URL
Wiki_RSS_URL = "https://wiki.koozali.org/api.php?hidebots=1&urlversion=2&days=7&limit=50&action=feedrecentchanges&feedformat=rss"
# Updated Rocket.Chat webhook URL
ROCKET_CHAT_URL = "https://chat.koozali.org/hooks/67b8a0fc1f56a124fa47dbec/Dee9hFcASKvvBsW9uRyy7Xjg2m3NTxDprt2HQKTRNuWv8SFr"

# Set up logging
log_file = "/var/log/Wiki2Rocket.log" 
logging.basicConfig(filename=log_file, level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# Database setup
def setup_database():
    conn = sqlite3.connect('sent_wikis.db')
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS sent_wikis (
            id TEXT PRIMARY KEY
        )
    ''')
    conn.commit()
    conn.close()

# Function to check if a wiki ID has been sent
def has_wiki_been_sent(wiki_id):
    conn = sqlite3.connect('sent_wikis.db')
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM sent_wikis WHERE id = ?', (wiki_id,))
    exists = cursor.fetchone() is not None
    conn.close()
    return exists

# Function to mark a wiki ID as sent
def mark_wiki_as_sent(wiki_id):
    conn = sqlite3.connect('sent_wikis.db')
    cursor = conn.cursor()
    cursor.execute('INSERT OR IGNORE INTO sent_wikis (id) VALUES (?)', (wiki_id,))
    conn.commit()
    conn.close()

# Function to send message to Rocket.Chat
def send_to_rocket_chat(wiki_title, wiki_link, wiki_id, description):
    #logging.info(f"Logging link:{wiki_link}")
    payload = {
        "alias": "Wiki",
        "text": f"{description}",
        "attachments": [
            {
                "title": wiki_link,
                "title_link": wiki_link,
                "color": "#764FA5"
            }
        ]
    }
    response = requests.post(ROCKET_CHAT_URL, json=payload)
    if response.status_code == 200:
        logging.info(f"Wiki notification sent successfully: {wiki_title} (ID: {wiki_id})")
    else:
        logging.error(f"Failed to send wiki notification: {response.status_code} - {response.text}")

# Function to send startup message to Rocket.Chat
def send_startup_message():
    payload = {
        "alias": "Wiki",
        "text": "Wiki to Rocket.Chat integration started successfully.",
    }
    response = requests.post(ROCKET_CHAT_URL, json=payload)
    if response.status_code == 200:
        logging.info("Startup message sent successfully to Rocket.Chat.")
    else:
        logging.error(f"Failed to send startup message: {response.status_code} - {response.text}")

# Function to fetch and parse the Wiki RSS feed
def fetch_Wiki_feed():
    feed = feedparser.parse(Wiki_RSS_URL)
    return feed.entries

# Function to clear the sent wiki database
def clear_sent_wiki_database():
    conn = sqlite3.connect('sent_wikis.db')
    cursor = conn.cursor()
    cursor.execute('DELETE FROM sent_wikis')
    conn.commit()
    conn.close()
    logging.info("Cleared the sent database.")

# Main polling loop
def main(one_shot=False, empty_db=False):
	setup_database()  # Initialize the database

	if empty_db:
		clear_sent_wiki_database()

	send_startup_message()  # Send startup message

	while True:
		entries = fetch_Wiki_feed()

		for entry in entries:
			json_data = json.dumps(entry, indent=2)
			data = json.loads(json_data)

			title = data["title"]
			published_date = data["published"]
			author = data["author"]
			revision_link = data["link"]

			formatted_string = (
				f"Title: {title}\n"
				f"Published Date: {published_date}\n"
				f"Author: {author}\n"
				f"Link to Comparison (old vs new): {revision_link}"
			)
			
			match = re.search(r'(.*?title=.*?)(.*)', data["id"])

			if match:
				wiki_link = match.group(1)+title  # Everything up to and including "title=<whatever>"
				wiki_id = match.group(2)     # Everything from (and including) <whatever> to the end
			else:
				wiki_id = "Unexpected link"
				wiki_link = "Unexpected link"
				
			#logging.info(f"{wiki_link} {wiki_id} {title}")

			if not has_wiki_been_sent(wiki_id):  # Check if the wiki has been sent
				send_to_rocket_chat(title, wiki_link, wiki_id, formatted_string)
				mark_wiki_as_sent(wiki_id)  # Mark the wiki ID as sent
				
				if one_shot:
					logging.info("One-shot mode activated; exiting after sending one message.")
					return  # Exit after sending the first message

		if not one_shot:
			# Wait for 1 minute before polling again
			time.sleep(60)

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Wiki to Rocket.Chat integration.")
    parser.add_argument('--one-shot', action='store_true', help='Exit after sending one message.')
    parser.add_argument('--empty-db', action='store_true', help='Clear the database of sent entries.')

    args = parser.parse_args()
    
    main(one_shot=args.one_shot, empty_db=args.empty_db)