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

Added logging to /var/log/???.log and Initial message to Rocket

Browse Source
This commit is contained in:
root 2025-01-08 19:25:43 +00:00
parent dd9c93de82
commit d7f2b71533
3 changed files with 133 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored Normal file
View File

@ -0,0 +1 @@
*.db

65
bugzilla rss to rocket json.py
View File

@ -1,16 +1,52 @@
import time
import feedparser
import requests
import sqlite3
import logging
import os
# Bugzilla RSS feed URL
BUGZILLA_RSS_URL = "https://bugs.koozali.org/buglist.cgi?chfield=%5BBug%20creation%5D&chfieldfrom=7d&ctype=atom&title=Bugs%20reported%20in%20the%20last%207%20days"
# Updated Rocket.Chat webhook URL
ROCKET_CHAT_URL = "https://chat.koozali.org/hooks/677e97a73ddf8049989dbc8c/r9uiYpTRAXo3mkFKxHnoTwGCdtKpYaDemCpHArgz89knkwLo"
# Set to keep track of sent bug IDs
sent_bugs = set()
# Set up logging
log_file = "/var/log/BugzillaToRocket.log"
logging.basicConfig(filename=log_file, level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# Function to send message to Rocket.Chat
# Ensure the log directory exists
os.makedirs(os.path.dirname(log_file), exist_ok=True)
# Database setup
def setup_database():
conn = sqlite3.connect('sent_bugs.db')
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS sent_bugs (
id TEXT PRIMARY KEY
)
''')
conn.commit()
conn.close()
# Function to check if a bug ID has been sent
def has_bug_been_sent(bug_id):
conn = sqlite3.connect('sent_bugs.db')
cursor = conn.cursor()
cursor.execute('SELECT * FROM sent_bugs WHERE id = ?', (bug_id,))
exists = cursor.fetchone() is not None
conn.close()
return exists
# Function to mark a bug ID as sent
def mark_bug_as_sent(bug_id):
conn = sqlite3.connect('sent_bugs.db')
cursor = conn.cursor()
cursor.execute('INSERT OR IGNORE INTO sent_bugs (id) VALUES (?)', (bug_id,))
conn.commit()
conn.close()
# Function to send a message to Rocket.Chat
def send_to_rocket_chat(bug_title, bug_link, bug_id):
payload = {
"alias": "Bugzilla",
@ -25,9 +61,21 @@ def send_to_rocket_chat(bug_title, bug_link, bug_id):
}
response = requests.post(ROCKET_CHAT_URL, json=payload)
if response.status_code == 200:
print(f"Bug report sent successfully: {bug_title} (ID: {bug_id})")
logging.info(f"Bug notification sent successfully: {bug_title} (ID: {bug_id})")
else:
print(f"Failed to send bug report: {response.status_code} - {response.text}")
logging.error(f"Failed to send bug notification: {response.status_code} - {response.text}")
# Function to send a startup message to Rocket.Chat
def send_startup_message():
payload = {
"alias": "Bugzilla",
"text": "Bugzilla 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 Bugzilla RSS feed
def fetch_bugzilla_feed():
@ -36,17 +84,20 @@ def fetch_bugzilla_feed():
# Main polling loop
def main():
setup_database() # Initialize the database
send_startup_message() # Send the startup message
while True:
entries = fetch_bugzilla_feed()
for entry in entries:
bug_id = entry.id.split('=')[-1] # Extract the bug number from the URL
if bug_id not in sent_bugs:
if not has_bug_been_sent(bug_id): # Check if the bug has been sent
title = entry.title.strip()
link = entry.link
# Send the bug title, link, and ID
send_to_rocket_chat(title, link, bug_id)
sent_bugs.add(bug_id) # Track the sent bug ID
mark_bug_as_sent(bug_id) # Mark the bug ID as sent
# Wait for 1 minute before polling again
time.sleep(60)

92
koji rss to rocket json.py
View File

@ -1,54 +1,104 @@
import time
import feedparser
import requests
import sqlite3
import logging
import os
# Koji RSS feed URL
KOJI_RSS_URL = "http://koji.koozali.org/koji/recentbuilds?feed=rss"
# Rocket.Chat webhook URL
# Updated Rocket.Chat webhook URL
ROCKET_CHAT_URL = "https://chat.koozali.org/hooks/66d24441effca216c2ca309f/KJLaNwc5vyHwqz5MhRDpBkKWnQuAfsCX3xZMHxPhpuqmFgBn"
# Set up logging
log_file = "/var/log/Koji2Rocket.log"
logging.basicConfig(filename=log_file, level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# Ensure the log directory exists
os.makedirs(os.path.dirname(log_file), exist_ok=True)
# Database setup
def setup_database():
conn = sqlite3.connect('sent_builds.db')
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS sent_builds (
id TEXT PRIMARY KEY
)
''')
conn.commit()
conn.close()
# Function to check if a build ID has been sent
def has_build_been_sent(build_id):
conn = sqlite3.connect('sent_builds.db')
cursor = conn.cursor()
cursor.execute('SELECT * FROM sent_builds WHERE id = ?', (build_id,))
exists = cursor.fetchone() is not None
conn.close()
return exists
# Function to mark a build ID as sent
def mark_build_as_sent(build_id):
conn = sqlite3.connect('sent_builds.db')
cursor = conn.cursor()
cursor.execute('INSERT OR IGNORE INTO sent_builds (id) VALUES (?)', (build_id,))
conn.commit()
conn.close()
# Function to send message to Rocket.Chat
def send_to_rocket_chat(message_title, message_link, message_description):
def send_to_rocket_chat(build_title, build_link, build_id):
payload = {
"alias": "Koji",
"text": message_title,
"text": f"Build Notification - ID: {build_id}",
"attachments": [
{
"title": message_title,
"title_link": message_link,
"text": message_description,
# "image_url": "https://chat.koozali.org/images/integration-attachment-example.png",
"title": build_title,
"title_link": build_link,
"color": "#764FA5"
}
]
}
response = requests.post(ROCKET_CHAT_URL, json=payload)
if response.status_code == 200:
print(f"Message ({title}) sent successfully!")
logging.info(f"Build notification sent successfully: {build_title} (ID: {build_id})")
else:
print(f"Failed to send message: {response.status_code} - {response.text}")
logging.error(f"Failed to send build notification: {response.status_code} - {response.text}")
# Function to send startup message to Rocket.Chat
def send_startup_message():
payload = {
"alias": "Koji",
"text": "Koji 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 Koji RSS feed
def fetch_koji_feed():
feed = feedparser.parse(KOJI_RSS_URL)
if feed.entries:
latest_build = feed.entries[0]
title = latest_build.title.strip()
link = latest_build.link
description = latest_build.description.strip().replace("<pre>", "").replace("</pre>", "")
return title, link, description
return None, None, None
return feed.entries
# Main polling loop
def main():
last_title = None
setup_database() # Initialize the database
send_startup_message() # Send startup message
while True:
title, link, description = fetch_koji_feed()
entries = fetch_koji_feed()
if title and title != last_title:
send_to_rocket_chat(title, link, description)
last_title = title
for entry in entries:
build_title = entry.title.strip()
build_link = entry.link
# Extracting the build ID from the title
build_id = build_title.split(':')[1].split(',')[0].strip() # Extract the relevant part from the title
if not has_build_been_sent(build_id): # Check if the build has been sent
send_to_rocket_chat(build_title, build_link, build_id)
mark_build_as_sent(build_id) # Mark the build ID as sent
# Wait for 1 minute before polling again
time.sleep(60)