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

Add in forever loop and remove redundant def

Browse Source
This commit is contained in:
Brian Read 2025-02-27 10:17:53 +00:00
parent 72dd1383ee
commit 985aa30648
1 changed files with 36 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

97
wiki rss to rocket chat.py
View File

@ -40,32 +40,6 @@ import argparse
def join_lines_with_newline(lines):
return "\n".join(lines)
def extract_changelog_top(Wiki_link):
response = requests.get(Wiki_link)
soup = BeautifulSoup(response.text, 'html.parser')
changelog_td = soup.find('td', class_='changelog')
if changelog_td:
changelog_text = changelog_td.get_text(strip=False)
lines = changelog_text.split('\n')
result = []
seen = set()
for line in lines:
stripped_line = line.strip()
if not stripped_line:
if result: # Stop at the first blank line after content
break
continue
if stripped_line not in seen:
seen.add(stripped_line)
if not result or stripped_line.startswith('-'):
result.append(stripped_line)
return join_lines_with_newline(result)
else:
return []
# 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
@ -159,46 +133,47 @@ def main(one_shot=False, empty_db=False):
send_startup_message() # Send startup message
entries = fetch_Wiki_feed()
while True
entries = fetch_Wiki_feed()
for entry in entries:
json_data = json.dumps(entry, indent=2)
data = json.loads(json_data)
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"]
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"
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}"
)
#logging.info(f"{wiki_link} {wiki_id} {title}")
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 # 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 one_shot:
# Wait for 1 minute before polling again
time.sleep(60)
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.")
@ -207,4 +182,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)