smedev/RSS-to-Rocket
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add in IP cache for Wiki and just the def for bugzilla

Browse Source
This commit is contained in:
Brian Read
2025-03-16 10:56:06 +00:00
parent 1793812809
commit 7c9bfbb55a
2 changed files with 104 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
bugzilla rss to rocket json.py
View File

@@ -28,6 +28,33 @@ FEED_TO_CHAT_MAP = {
}
}
def get_ip_address(domain, retries=10, delay=1):
"""
Resolves the IP address of a domain, retrying up to `retries` times if it fails.
Args:
domain (str): The domain name to resolve.
retries (int): Number of retry attempts (default is 10).
delay (int): Delay between retries in seconds (default is 1 second).
Returns:
str: The IP address if resolved successfully.
Raises:
RuntimeError: If unable to resolve the domain after all attempts.
"""
for attempt in range(1, retries + 1):
try:
ip_address = socket.gethostbyname(domain)
logging.info(f"Successfully resolved {domain} to {ip_address}")
return ip_address
except socket.gaierror:
logging.warning(f"Attempt {attempt} failed. Retrying...")
time.sleep(delay)
raise RuntimeError(f"Unable to resolve domain '{domain}' after {retries} attempts.")
# Set up logging to the current directory
log_file = "/var/log/BugzillaToRocket.log"
logging.basicConfig(filename=log_file, level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
@@ -247,4 +274,4 @@ def main():
break # Exit loop after one iteration if --one-off is set
if __name__ == "__main__":
main()
main()

162
wiki rss to rocket chat.py
View File

@@ -4,52 +4,34 @@ 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"
# }
# }
import socket
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
def get_ip_address(domain, retries=10, delay=1):
for attempt in range(1, retries + 1):
try:
ip_address = socket.gethostbyname(domain)
logging.info(f"Successfully resolved {domain} to {ip_address}")
return ip_address
except socket.gaierror:
logging.warning(f"Attempt {attempt} failed. Retrying...")
time.sleep(delay)
raise RuntimeError(f"Unable to resolve domain '{domain}' after {retries} attempts.")
WIKI_DOMAIN = "wiki.koozali.org"
Wiki_RSS_URL = None
ROCKET_CHAT_URL = "https://chat.koozali.org/hooks/67b8a0fc1f56a124fa47dbec/Dee9hFcASKvvBsW9uRyy7Xjg2m3NTxDprt2HQKTRNuWv8SFr"
# Set up logging
log_file = "/var/log/Wiki2Rocket.log"
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()
@@ -61,7 +43,6 @@ def setup_database():
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()
@@ -70,7 +51,6 @@ def has_wiki_been_sent(wiki_id):
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()
@@ -78,9 +58,7 @@ def mark_wiki_as_sent(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}",
@@ -98,7 +76,6 @@ def send_to_rocket_chat(wiki_title, wiki_link, wiki_id, description):
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",
@@ -110,12 +87,19 @@ def send_startup_message():
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
def fetch_Wiki_feed(original_domain):
try:
response = requests.get(
Wiki_RSS_URL,
headers={'Host': original_domain},
verify=True
)
feed = feedparser.parse(response.content)
return feed.entries
except Exception as e:
logging.error(f"RSS request to wiki failed: {str(e)}")
return []
# Function to clear the sent wiki database
def clear_sent_wiki_database():
conn = sqlite3.connect('sent_wikis.db')
cursor = conn.cursor()
@@ -124,56 +108,62 @@ def clear_sent_wiki_database():
conn.close()
logging.info("Cleared the sent database.")
# Main polling loop
def main(one_shot=False, empty_db=False):
setup_database() # Initialize the database
global Wiki_RSS_URL
try:
wiki_ip = get_ip_address(WIKI_DOMAIN)
Wiki_RSS_URL = f"http://{wiki_ip}/api.php?hidebots=1&urlversion=2&days=7&limit=50&action=feedrecentchanges&feedformat=rss"
logging.info(f"Resolved Wiki URL to: {Wiki_RSS_URL}")
except RuntimeError as e:
logging.error(str(e))
return
if empty_db:
clear_sent_wiki_database()
setup_database()
send_startup_message() # Send startup message
if empty_db:
clear_sent_wiki_database()
while True:
entries = fetch_Wiki_feed()
send_startup_message()
for entry in entries:
json_data = json.dumps(entry, indent=2)
data = json.loads(json_data)
while True:
entries = fetch_Wiki_feed(WIKI_DOMAIN)
title = data["title"]
published_date = data["published"]
author = data["author"]
revision_link = data["link"]
for entry in entries:
json_data = json.dumps(entry, indent=2)
data = json.loads(json_data)
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"])
title = data["title"]
published_date = data["published"]
author = data["author"]
revision_link = data["link"]
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}")
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 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 match:
wiki_link = match.group(1) + title
wiki_id = match.group(2)
else:
wiki_id = "Unexpected link"
wiki_link = "Unexpected link"
if not one_shot:
# Wait for 1 minute before polling again
time.sleep(60)
if not has_wiki_been_sent(wiki_id):
send_to_rocket_chat(title, wiki_link, wiki_id, formatted_string)
mark_wiki_as_sent(wiki_id)
if one_shot:
logging.info("One-shot mode activated; exiting after sending one message.")
return
if not one_shot:
time.sleep(60)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Wiki to Rocket.Chat integration.")
@@ -182,4 +172,4 @@ if __name__ == "__main__":
args = parser.parse_args()
main(one_shot=args.one_shot, empty_db=args.empty_db)
main(one_shot=args.one_shot, empty_db=args.empty_db)