website_agent.rb 14KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385
  1. require 'nokogiri'
  2. require 'date'
  3. module Agents
  4. class WebsiteAgent < Agent
  5. include WebRequestConcern
  6. default_schedule "every_12h"
  7. UNIQUENESS_LOOK_BACK = 200
  8. UNIQUENESS_FACTOR = 3
  9. description <<-MD
  10. The WebsiteAgent scrapes a website, XML document, or JSON feed and creates Events based on the results.
  11. Specify a `url` and select a `mode` for when to create Events based on the scraped data, either `all` or `on_change`.
  12. `url` can be a single url, or an array of urls (for example, for multiple pages with the exact same structure but different content to scrape)
  13. The `type` value can be `xml`, `html`, `json`, or `text`.
  14. To tell the Agent how to parse the content, specify `extract` as a hash with keys naming the extractions and values of hashes.
  15. When parsing HTML or XML, these sub-hashes specify how each extraction should be done. The Agent first selects a node set from the document for each extraction key by evaluating either a CSS selector in `css` or an XPath expression in `xpath`. It then evaluates an XPath expression in `value` on each node in the node set, converting the result into string. Here's an example:
  16. "extract": {
  17. "url": { "css": "#comic img", "value": "@src" },
  18. "title": { "css": "#comic img", "value": "@title" },
  19. "body_text": { "css": "div.main", "value": ".//text()" }
  20. }
  21. "@_attr_" is the XPath expression to extract the value of an attribute named _attr_ from a node, and ".//text()" is to extract all the enclosed texts. You can also use [XPath functions](http://www.w3.org/TR/xpath/#section-String-Functions) like `normalize-space` to strip and squeeze whitespace, `substring-after` to extract part of a text, and `translate` to remove comma from a formatted number, etc. Note that these functions take a string, not a node set, so what you may think would be written as `normalize-space(.//text())` should actually be `normalize-space(.)`.
  22. When parsing JSON, these sub-hashes specify [JSONPaths](http://goessner.net/articles/JsonPath/) to the values that you care about. For example:
  23. "extract": {
  24. "title": { "path": "results.data[*].title" },
  25. "description": { "path": "results.data[*].description" }
  26. }
  27. When parsing text, each sub-hash should contain a `regexp` and `index`. Output text is matched against the regular expression repeatedly from the beginning through to the end, collecting a captured group specified by `index` in each match. Each index should be either an integer or a string name which corresponds to <code>(?&lt;<em>name</em>&gt;...)</code>. For example, to parse lines of <code><em>word</em>: <em>definition</em></code>, the following should work:
  28. "extract": {
  29. "word": { "regexp": "^(.+?): (.+)$", index: 1 },
  30. "definition": { "regexp": "^(.+?): (.+)$", index: 2 }
  31. }
  32. Or if you prefer names to numbers for index:
  33. "extract": {
  34. "word": { "regexp": "^(?<word>.+?): (?<definition>.+)$", index: 'word' },
  35. "definition": { "regexp": "^(?<word>.+?): (?<definition>.+)$", index: 'definition' }
  36. }
  37. To extract the whole content as one event:
  38. "extract": {
  39. "content": { "regexp": "\A(?m:.)*\z", index: 0 }
  40. }
  41. Beware that `.` does not match the newline character (LF) unless the `m` flag is in effect, and `^`/`$` basically match every line beginning/end. See [this document](http://ruby-doc.org/core-#{RUBY_VERSION}/doc/regexp_rdoc.html) to learn the regular expression variant used in this service.
  42. Note that for all of the formats, whatever you extract MUST have the same number of matches for each extractor. E.g., if you're extracting rows, all extractors must match all rows. For generating CSS selectors, something like [SelectorGadget](http://selectorgadget.com) may be helpful.
  43. Can be configured to use HTTP basic auth by including the `basic_auth` parameter with `"username:password"`, or `["username", "password"]`.
  44. Set `expected_update_period_in_days` to the maximum amount of time that you'd expect to pass between Events being created by this Agent. This is only used to set the "working" status.
  45. Set `uniqueness_look_back` to limit the number of events checked for uniqueness (typically for performance). This defaults to the larger of #{UNIQUENESS_LOOK_BACK} or #{UNIQUENESS_FACTOR}x the number of detected received results.
  46. Set `force_encoding` to an encoding name if the website does not return a Content-Type header with a proper charset.
  47. Set `user_agent` to a custom User-Agent name if the website does not like the default value (`#{default_user_agent}`).
  48. The `headers` field is optional. When present, it should be a hash of headers to send with the request.
  49. Set `disable_ssl_verification` to `true` to disable ssl verification.
  50. The WebsiteAgent can also scrape based on incoming events. It will scrape the url contained in the `url` key of the incoming event payload. If you specify `merge` as the mode, it will retain the old payload and update it with the new values.
  51. In Liquid templating, the following variable is available:
  52. * `_response_`: A response object with the following keys:
  53. * `status`: HTTP status as integer. (Almost always 200)
  54. * `headers`: Reponse headers; for example, `{{ _response_.headers.Content-Type }}` expands to the value of the Content-Type header. Keys are insentitive to cases and -/_.
  55. MD
  56. event_description do
  57. "Events will have the following fields:\n\n %s" % [
  58. Utils.pretty_print(Hash[options['extract'].keys.map { |key|
  59. [key, "..."]
  60. }])
  61. ]
  62. end
  63. def working?
  64. event_created_within?(interpolated['expected_update_period_in_days']) && !recent_error_logs?
  65. end
  66. def default_options
  67. {
  68. 'expected_update_period_in_days' => "2",
  69. 'url' => "http://xkcd.com",
  70. 'type' => "html",
  71. 'mode' => "on_change",
  72. 'extract' => {
  73. 'url' => { 'css' => "#comic img", 'value' => "@src" },
  74. 'title' => { 'css' => "#comic img", 'value' => "@alt" },
  75. 'hovertext' => { 'css' => "#comic img", 'value' => "@title" }
  76. }
  77. }
  78. end
  79. def validate_options
  80. # Check for required fields
  81. errors.add(:base, "url and expected_update_period_in_days are required") unless options['expected_update_period_in_days'].present? && options['url'].present?
  82. if !options['extract'].present? && extraction_type != "json"
  83. errors.add(:base, "extract is required for all types except json")
  84. end
  85. # Check for optional fields
  86. if options['mode'].present?
  87. errors.add(:base, "mode must be set to on_change, all or merge") unless %w[on_change all merge].include?(options['mode'])
  88. end
  89. if options['expected_update_period_in_days'].present?
  90. errors.add(:base, "Invalid expected_update_period_in_days format") unless is_positive_integer?(options['expected_update_period_in_days'])
  91. end
  92. if options['uniqueness_look_back'].present?
  93. errors.add(:base, "Invalid uniqueness_look_back format") unless is_positive_integer?(options['uniqueness_look_back'])
  94. end
  95. if (encoding = options['force_encoding']).present?
  96. case encoding
  97. when String
  98. begin
  99. Encoding.find(encoding)
  100. rescue ArgumentError
  101. errors.add(:base, "Unknown encoding: #{encoding.inspect}")
  102. end
  103. else
  104. errors.add(:base, "force_encoding must be a string")
  105. end
  106. end
  107. validate_web_request_options!
  108. end
  109. def check
  110. check_urls(interpolated['url'])
  111. end
  112. def check_urls(in_url)
  113. return unless in_url.present?
  114. Array(in_url).each do |url|
  115. check_url(url)
  116. end
  117. end
  118. def check_url(url, payload = {})
  119. log "Fetching #{url}"
  120. response = faraday.get(url)
  121. raise "Failed: #{response.inspect}" unless response.success?
  122. interpolation_context.stack {
  123. interpolation_context['_response_'] = ResponseDrop.new(response)
  124. body = response.body
  125. if (encoding = interpolated['force_encoding']).present?
  126. body = body.encode(Encoding::UTF_8, encoding)
  127. end
  128. doc = parse(body)
  129. if extract_full_json?
  130. if store_payload!(previous_payloads(1), doc)
  131. log "Storing new result for '#{name}': #{doc.inspect}"
  132. create_event payload: payload.merge(doc)
  133. end
  134. return
  135. end
  136. output =
  137. case extraction_type
  138. when 'json'
  139. extract_json(doc)
  140. when 'text'
  141. extract_text(doc)
  142. else
  143. extract_xml(doc)
  144. end
  145. num_unique_lengths = interpolated['extract'].keys.map { |name| output[name].length }.uniq
  146. if num_unique_lengths.length != 1
  147. raise "Got an uneven number of matches for #{interpolated['name']}: #{interpolated['extract'].inspect}"
  148. end
  149. old_events = previous_payloads num_unique_lengths.first
  150. num_unique_lengths.first.times do |index|
  151. result = {}
  152. interpolated['extract'].keys.each do |name|
  153. result[name] = output[name][index]
  154. if name.to_s == 'url'
  155. result[name] = (response.env[:url] + result[name]).to_s
  156. end
  157. end
  158. if store_payload!(old_events, result)
  159. log "Storing new parsed result for '#{name}': #{result.inspect}"
  160. create_event payload: payload.merge(result)
  161. end
  162. end
  163. }
  164. rescue => e
  165. error "Error when fetching url: #{e.message}\n#{e.backtrace.join("\n")}"
  166. end
  167. def receive(incoming_events)
  168. incoming_events.each do |event|
  169. interpolate_with(event) do
  170. url_to_scrape = event.payload['url']
  171. next unless url_to_scrape =~ /^https?:\/\//i
  172. check_url(url_to_scrape,
  173. interpolated['mode'].to_s == "merge" ? event.payload : {})
  174. end
  175. end
  176. end
  177. private
  178. # This method returns true if the result should be stored as a new event.
  179. # If mode is set to 'on_change', this method may return false and update an existing
  180. # event to expire further in the future.
  181. def store_payload!(old_events, result)
  182. case interpolated['mode'].presence
  183. when 'on_change'
  184. result_json = result.to_json
  185. old_events.each do |old_event|
  186. if old_event.payload.to_json == result_json
  187. old_event.expires_at = new_event_expiration_date
  188. old_event.save!
  189. return false
  190. end
  191. end
  192. true
  193. when 'all', 'merge', ''
  194. true
  195. else
  196. raise "Illegal options[mode]: #{interpolated['mode']}"
  197. end
  198. end
  199. def previous_payloads(num_events)
  200. if interpolated['uniqueness_look_back'].present?
  201. look_back = interpolated['uniqueness_look_back'].to_i
  202. else
  203. # Larger of UNIQUENESS_FACTOR * num_events and UNIQUENESS_LOOK_BACK
  204. look_back = UNIQUENESS_FACTOR * num_events
  205. if look_back < UNIQUENESS_LOOK_BACK
  206. look_back = UNIQUENESS_LOOK_BACK
  207. end
  208. end
  209. events.order("id desc").limit(look_back) if interpolated['mode'] == "on_change"
  210. end
  211. def extract_full_json?
  212. !interpolated['extract'].present? && extraction_type == "json"
  213. end
  214. def extraction_type
  215. (interpolated['type'] || begin
  216. case interpolated['url']
  217. when /\.(rss|xml)$/i
  218. "xml"
  219. when /\.json$/i
  220. "json"
  221. when /\.(txt|text)$/i
  222. "text"
  223. else
  224. "html"
  225. end
  226. end).to_s
  227. end
  228. def extract_each(doc, &block)
  229. interpolated['extract'].each_with_object({}) { |(name, extraction_details), output|
  230. output[name] = block.call(extraction_details)
  231. }
  232. end
  233. def extract_json(doc)
  234. extract_each(doc) { |extraction_details|
  235. result = Utils.values_at(doc, extraction_details['path'])
  236. log "Extracting #{extraction_type} at #{extraction_details['path']}: #{result}"
  237. result
  238. }
  239. end
  240. def extract_text(doc)
  241. extract_each(doc) { |extraction_details|
  242. regexp = Regexp.new(extraction_details['regexp'])
  243. result = []
  244. doc.scan(regexp) {
  245. result << Regexp.last_match[extraction_details['index']]
  246. }
  247. log "Extracting #{extraction_type} at #{regexp}: #{result}"
  248. result
  249. }
  250. end
  251. def extract_xml(doc)
  252. extract_each(doc) { |extraction_details|
  253. case
  254. when css = extraction_details['css']
  255. nodes = doc.css(css)
  256. when xpath = extraction_details['xpath']
  257. doc.remove_namespaces! # ignore xmlns, useful when parsing atom feeds
  258. nodes = doc.xpath(xpath)
  259. else
  260. raise '"css" or "xpath" is required for HTML or XML extraction'
  261. end
  262. case nodes
  263. when Nokogiri::XML::NodeSet
  264. result = nodes.map { |node|
  265. case value = node.xpath(extraction_details['value'])
  266. when Float
  267. # Node#xpath() returns any numeric value as float;
  268. # convert it to integer as appropriate.
  269. value = value.to_i if value.to_i == value
  270. end
  271. value.to_s
  272. }
  273. else
  274. raise "The result of HTML/XML extraction was not a NodeSet"
  275. end
  276. log "Extracting #{extraction_type} at #{xpath || css}: #{result}"
  277. result
  278. }
  279. end
  280. def parse(data)
  281. case extraction_type
  282. when "xml"
  283. Nokogiri::XML(data)
  284. when "json"
  285. JSON.parse(data)
  286. when "html"
  287. Nokogiri::HTML(data)
  288. when "text"
  289. data
  290. else
  291. raise "Unknown extraction type #{extraction_type}"
  292. end
  293. end
  294. def is_positive_integer?(value)
  295. Integer(value) >= 0
  296. rescue
  297. false
  298. end
  299. # Wraps Faraday::Response
  300. class ResponseDrop < LiquidDroppable::Drop
  301. def headers
  302. HeaderDrop.new(@object.headers)
  303. end
  304. # Integer value of HTTP status
  305. def status
  306. @object.status
  307. end
  308. end
  309. # Wraps Faraday::Utilsa::Headers
  310. class HeaderDrop < LiquidDroppable::Drop
  311. def before_method(name)
  312. @object[name.tr('_', '-')]
  313. end
  314. end
  315. end
  316. end