1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
|
#!/usr/bin/env python3
# File: sniffa.py
# Location: https://gitlab.com/bgstack15/sniffa
# Author: danielmitterdorfer
# SPDX-License-Identifer: GPL-3.0
# Startdate: 2016-07-16
# Title: Discourse sniffer
# Purpose: Send alerts for saved searches on Discourse forum websites
# History:
# 2022-01-07 forked by bgstack15 to send discord messages
# Homepage: https://github.com/danielmitterdorfer/sniffa
# Usage:
# ./sniffa.py aoe2
# Reference:
# https://meta.discourse.org/t/watching-and-sending-notifications-for-keywords/59286
# Improve:
# Documentation: See README.md
# Dependencies:
# req-fedora: python3-certifi
# req-pip3: discord
# A webhook given to me by discord "server" admin
import os, sys, errno, json, configparser, urllib3, urllib.parse, datetime, certifi, requests
from discord import Webhook, RequestsWebhookAdapter
DOMAIN_SECTION_KEY = "sniffa.domain"
def ensure_dir(directory):
try:
os.makedirs(directory)
except OSError as exception:
if exception.errno != errno.EEXIST:
raise
def creation_date(item):
return datetime.datetime.strptime(item["created_at"], "%Y-%m-%dT%H:%M:%S.%fZ").timestamp()
def main():
if not len(sys.argv) == 2:
print("usage: %s domain_key" % sys.argv[0], file=sys.stderr)
exit(1)
domain_key = sys.argv[1]
http = urllib3.PoolManager(cert_reqs="CERT_REQUIRED", ca_certs=certifi.where())
watches_dir = "%s/.sniffa" % os.getenv("HOME")
watches_file = "%s/watches-%s.ini" % (watches_dir, domain_key)
ensure_dir(watches_dir)
config = configparser.ConfigParser()
config.read(watches_file)
if not DOMAIN_SECTION_KEY in config or not "url" in config[DOMAIN_SECTION_KEY]:
print("Invalid configuration in [%s]" % watches_file, file=sys.stderr)
print("", file=sys.stderr)
print("Please add a domain to query according to the following example in [%s]" % watches_file, file=sys.stderr)
print("", file=sys.stderr)
print(" [%s]" % DOMAIN_SECTION_KEY, file=sys.stderr)
print(" url=https://discuss.example.org", file=sys.stderr)
print("", file=sys.stderr)
exit(1)
domain = config[DOMAIN_SECTION_KEY]["url"]
print("Checking for new posts on %s" % domain_key)
for keyword in config.sections():
if keyword == DOMAIN_SECTION_KEY:
continue
if "ids" in config[keyword]:
ids = config[keyword]["ids"]
if len(ids) > 0:
known_ids = set([int(id) for id in config[keyword]["ids"].split(",")])
else:
known_ids = set()
else:
known_ids = set()
query_string = urllib.parse.quote_plus("%s order:latest" % keyword)
r = http.request("GET", "%s/search?q=%s" % (domain, query_string), headers={"Accept": "application/json"})
results = json.loads(r.data.decode("utf-8"))
topics_by_id = {}
for topic in results["topics"]:
topics_by_id[topic["id"]] = topic
posts = []
for post in results["posts"]:
post_id = post["id"]
if post_id not in known_ids:
known_ids.add(post_id)
posts.append(post)
config[keyword]["ids"] = ",".join([str(id) for id in known_ids])
sorted(posts, key=creation_date, reverse=True)
if len(posts) > 0:
webhook = Webhook.from_url(config[DOMAIN_SECTION_KEY]["webhook"], adapter=RequestsWebhookAdapter())
for post in posts:
topic = topics_by_id[post["topic_id"]]
message = f"New post mentioning \"{keyword}\". Url=\"{domain}/t/{topic['slug']}/{topic['id']}\". group={str(topic['id'])}."
print(message)
webhook.send(f"{domain}/t/{topic['slug']}/{topic['id']}", username=config[DOMAIN_SECTION_KEY]["discord_user"])
with open(watches_file, "w") as f:
config.write(f)
print("Finished")
if __name__ == "__main__":
main()
|
