107 lines
3.5 KiB
Python
107 lines
3.5 KiB
Python
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 up logging
|
|
log_file = "/var/log/BugzillaToRocket.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_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",
|
|
"text": f"Bug Report - ID: {bug_id}",
|
|
"attachments": [
|
|
{
|
|
"title": bug_title,
|
|
"title_link": bug_link,
|
|
"color": "#764FA5"
|
|
}
|
|
]
|
|
}
|
|
response = requests.post(ROCKET_CHAT_URL, json=payload)
|
|
if response.status_code == 200:
|
|
logging.info(f"Bug notification sent successfully: {bug_title} (ID: {bug_id})")
|
|
else:
|
|
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():
|
|
feed = feedparser.parse(BUGZILLA_RSS_URL)
|
|
return feed.entries
|
|
|
|
# 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 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)
|
|
mark_bug_as_sent(bug_id) # Mark the bug ID as sent
|
|
|
|
# Wait for 1 minute before polling again
|
|
time.sleep(60)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|