Mercurial > prosody-modules

comparison mod_pubsub_feeds/mod_pubsub_feeds.lua @ 5653:62c6e17a5e9d

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Merge
author Stephen Paul Weber <singpolyma@singpolyma.net>
date Mon, 18 Sep 2023 08:24:19 -0500
parents bc292c84f56c
children 3f75ac4311bf
comparison
equal deleted inserted replaced
5652:eade7ff9f52c 5653:62c6e17a5e9d
1 -- Fetches Atom feeds and publishes to PubSub nodes 1 -- Fetches Atom feeds and publishes to PubSub nodes
2 --
3 -- Config:
4 -- Component "pubsub.example.com" "pubsub"
5 -- modules_enabled = {
6 -- "pubsub_feeds";
7 -- }
8 -- feeds = { -- node -> url
9 -- prosody_blog = "http://blog.prosody.im/feed/atom.xml";
10 -- }
11 -- feed_pull_interval = 20 -- minutes
12 --
13 -- Reference
14 -- http://pubsubhubbub.googlecode.com/svn/trunk/pubsubhubbub-core-0.4.html
15 2
16 local pubsub = module:depends"pubsub"; 3 local pubsub = module:depends"pubsub";
17 4
18 local time = os.time; 5 local time = os.time;
19 local dt_parse, dt_datetime = require "util.datetime".parse, require "util.datetime".datetime; 6 local dt_parse, dt_datetime = require "util.datetime".parse, require "util.datetime".datetime;
34 return translate_rss(feed); 21 return translate_rss(feed);
35 end 22 end
36 return nil, "unsupported-format"; 23 return nil, "unsupported-format";
37 end 24 end
38 25
39 local use_pubsubhubub = module:get_option_boolean("use_pubsubhubub", true); 26 local use_pubsubhubub = module:get_option_boolean("use_pubsubhubub", false);
40 if use_pubsubhubub then 27 if use_pubsubhubub then
41 module:depends"http"; 28 module:depends"http";
42 end 29 end
43 30
44 local http = require "net.http"; 31 local http = require "net.http";
45 local formdecode = http.formdecode; 32 local formdecode = http.formdecode;
46 local formencode = http.formencode; 33 local formencode = http.formencode;
47 34
48 local feed_list = module:shared("feed_list"); 35 local feed_list = module:shared("feed_list");
49 local refresh_interval = module:get_option_number("feed_pull_interval", 15) * 60; 36 local legacy_refresh_interval = module:get_option_number("feed_pull_interval", 15);
37 local refresh_interval = module:get_option_number("feed_pull_interval_seconds", legacy_refresh_interval*60);
50 local lease_length = tostring(math.floor(module:get_option_number("feed_lease_length", 86400))); 38 local lease_length = tostring(math.floor(module:get_option_number("feed_lease_length", 86400)));
51 39
52 function module.load() 40 function module.load()
53 local config = module:get_option("feeds", { }); 41 local config = module:get_option("feeds", { });
54 local ok, nodes = pubsub.service:get_nodes(true); 42 local ok, nodes = pubsub.service:get_nodes(true);
58 if type(node) == "number" then 46 if type(node) == "number" then
59 node = url; 47 node = url;
60 end 48 end
61 new_feed_list[node] = true; 49 new_feed_list[node] = true;
62 if not feed_list[node] then 50 if not feed_list[node] then
63 feed_list[node] = { url = url; node = node; last_update = 0 }; 51 local ok, err = pubsub.service:create(node, true);
52 if ok or err == "conflict" then
53 feed_list[node] = { url = url; node = node; last_update = 0 };
54 else
55 module:log("error", "Could not create node %s: %s", node, err);
56 end
64 else 57 else
65 feed_list[node].url = url; 58 feed_list[node].url = url;
66 end 59 end
67 if not nodes[node] then 60 if not nodes[node] then
68 feed_list[node].last_update = 0; 61 feed_list[node].last_update = 0;
73 feed_list[node] = nil; 66 feed_list[node] = nil;
74 end 67 end
75 end 68 end
76 end 69 end
77 70
78 function update_entry(item) 71 function update_entry(item, data)
79 local node = item.node; 72 local node = item.node;
80 module:log("debug", "parsing %d bytes of data in node %s", #item.data or 0, node) 73 module:log("debug", "parsing %d bytes of data in node %s", #data or 0, node)
81 local feed, err = parse_feed(item.data); 74 local feed, err = parse_feed(data);
82 if not feed then 75 if not feed then
83 module:log("error", "Could not parse feed %q: %s", item.url, err); 76 module:log("error", "Could not parse feed %q: %s", item.url, err);
84 module:log("debug", "Feed data:\n%s\n.", item.data); 77 module:log("debug", "Feed data:\n%s\n.", data);
85 return; 78 return;
86 end 79 end
87 local entries = {}; 80 local entries = {};
88 for entry in feed:childtags("entry") do 81 for entry in feed:childtags("entry") do
89 table.insert(entries, entry); 82 table.insert(entries, entry);
90 end 83 end
91 local ok, items = pubsub.service:get_items(node, true); 84 local ok, last_id = pubsub.service:get_last_item(node, true);
92 if not ok then 85 if not ok then
93 local ok, err = pubsub.service:create(node, true); 86 module:log("error", "PubSub node %q missing: %s", node, last_id);
94 if not ok then 87 return
95 module:log("error", "Could not create node %s: %s", node, err); 88 end
96 return; 89
97 end 90 local start_from = #entries;
98 items = {}; 91 for i, entry in ipairs(entries) do
99 end 92 local id = entry:get_child_text("id");
100 for i = #entries, 1, -1 do -- Feeds are usually in reverse order 93 if not id then
94 local link = entry:get_child("link");
95 if link then
96 module:log("debug", "Feed %q item %s is missing an id, using <link> instead", item.url, entry:top_tag());
97 id = link and link.attr.href;
98 else
99 module:log("error", "Feed %q item %s is missing both id and link, this feed is unusable", item.url, entry:top_tag());
100 return;
101 end
102 entry:text_tag("id", id);
103 end
104
105 if last_id == id then
106 -- This should be the first item that we already have.
107 start_from = i-1;
108 break
109 end
110 end
111
112 for i = start_from, 1, -1 do -- Feeds are usually in reverse order
101 local entry = entries[i]; 113 local entry = entries[i];
102 entry.attr.xmlns = xmlns_atom; 114 entry.attr.xmlns = xmlns_atom;
103 115
104 local e_published = entry:get_child_text("published"); 116 local id = entry:get_child_text("id");
105 e_published = e_published and dt_parse(e_published); 117
106 local e_updated = entry:get_child_text("updated"); 118 local timestamp = dt_parse(entry:get_child_text("published"));
107 e_updated = e_updated and dt_parse(e_updated); 119 if not timestamp then
108 120 timestamp = time();
109 local timestamp = e_updated or e_published or nil; 121 entry:text_tag("published", dt_datetime(timestamp));
110 --module:log("debug", "timestamp is %s, item.last_update is %s", tostring(timestamp), tostring(item.last_update)); 122 end
123
111 if not timestamp or not item.last_update or timestamp > item.last_update then 124 if not timestamp or not item.last_update or timestamp > item.last_update then
112 local id = entry:get_child_text("id"); 125 local xitem = st.stanza("item", { id = id, xmlns = "http://jabber.org/protocol/pubsub" }):add_child(entry);
113 if not id then 126 -- TODO Put data from /feed into item/source
114 local link = entry:get_child("link"); 127
115 id = link and link.attr.href; 128 local ok, err = pubsub.service:publish(node, true, id, xitem);
116 end 129 if not ok then
117 if not id then 130 module:log("error", "Publishing to node %s failed: %s", node, err);
118 -- Sigh, no link? 131 elseif timestamp then
119 id = feed.url .. "#" .. hmac_sha1(feed.url, tostring(entry), true) .. "@" .. dt_datetime(timestamp); 132 item.last_update = timestamp;
120 end
121 if not items[id] then
122 local xitem = st.stanza("item", { id = id, xmlns = "http://jabber.org/protocol/pubsub" }):add_child(entry);
123 -- TODO Put data from /feed into item/source
124
125 --module:log("debug", "publishing to %s, id %s", node, id);
126 local ok, err = pubsub.service:publish(node, true, id, xitem);
127 if not ok then
128 module:log("error", "Publishing to node %s failed: %s", node, err);
129 end
130 end 133 end
131 end 134 end
132 end 135 end
133 136
134 if item.lease_expires and item.lease_expires > time() then 137 if item.lease_expires and item.lease_expires > time() then
146 end 149 end
147 end 150 end
148 end 151 end
149 152
150 function fetch(item, callback) -- HTTP Pull 153 function fetch(item, callback) -- HTTP Pull
151 local headers = { }; 154 local headers = {
152 if item.data and item.etag then 155 ["If-None-Match"] = item.etag;
153 headers["If-None-Match"] = item.etag; 156 ["Accept"] = "application/atom+xml, application/x-rss+xml, application/xml";
154 end 157 };
155 http.request(item.url, { headers = headers }, function(data, code, resp) 158 http.request(item.url, { headers = headers }, function(data, code, resp)
156 if code == 200 then 159 if code == 200 then
157 item.data = data; 160 if callback then callback(item, data) end
158 if callback then callback(item) end
159 item.last_update = time();
160 if resp.headers then 161 if resp.headers then
161 item.etag = resp.headers.etag 162 item.etag = resp.headers.etag
162 end 163 end
163 elseif code == 304 then 164 elseif code == 304 then
164 item.last_update = time(); 165 module:log("debug", "No updates to %q", item.url);
165 elseif code == 301 and resp.headers.location then 166 elseif code == 301 and resp.headers.location then
166 module:log("info", "Feed %q has moved to %q", item.url, resp.headers.location); 167 module:log("info", "Feed %q has moved to %q", item.url, resp.headers.location);
167 elseif code <= 100 then 168 elseif code <= 100 then
168 module:log("error", "Error fetching %q: %q[%d]", item.url, data, code); 169 module:log("error", "Error fetching %q: %q[%d]", item.url, data, code);
169 else 170 else
266 module:log("debug", "Invalid signature, got %s but wanted %s", tostring(signature), tostring(localsig)); 267 module:log("debug", "Invalid signature, got %s but wanted %s", tostring(signature), tostring(localsig));
267 return 401; 268 return 401;
268 end 269 end
269 module:log("debug", "Valid signature"); 270 module:log("debug", "Valid signature");
270 end 271 end
271 feed.data = body; 272 update_entry(feed, body);
272 update_entry(feed);
273 feed.last_update = time();
274 return 202; 273 return 202;
275 end 274 end
276 return 400; 275 return 400;
277 end 276 end
278 return 501; 277 return 501;