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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
# frozen_string_literal: true
# Copyright (c) 2010-2012, Diaspora Inc. This file is
# licensed under the Affero General Public License version 3 or later. See
# the COPYRIGHT file.
module Diaspora; module Fetcher; class Public
include Diaspora::Logging
# various states that can be assigned to a person to describe where
# in the process of fetching their public posts we're currently at
Status_Initial = 0
Status_Running = 1
Status_Fetched = 2
Status_Processed = 3
Status_Done = 4
Status_Failed = 5
Status_Unfetchable = 6
def self.queue_for(person)
Workers::FetchPublicPosts.perform_async(person.diaspora_handle) unless person.fetch_status > Status_Initial
end
# perform all actions necessary to fetch the public posts of a person
# with the given diaspora_id
def fetch! diaspora_id
@person = Person.by_account_identifier diaspora_id
return unless qualifies_for_fetching?
begin
retrieve_and_process_posts
rescue => e
set_fetch_status Public::Status_Failed
raise e
end
set_fetch_status Public::Status_Done
end
private
# checks, that public posts for the person can be fetched,
# if it is reasonable to do so, and that they have not been fetched already
def qualifies_for_fetching?
raise ActiveRecord::RecordNotFound unless @person.present?
return false if @person.fetch_status == Public::Status_Unfetchable
# local users don't need to be fetched
if @person.local?
set_fetch_status Public::Status_Unfetchable
return false
end
# this record is already being worked on
return false if @person.fetch_status > Public::Status_Initial
# ok, let's go
@person.remote? &&
@person.fetch_status == Public::Status_Initial
end
# call the methods to fetch and process the public posts for the person
# does some error logging, in case of an exception
def retrieve_and_process_posts
begin
retrieve_posts
rescue => e
logger.error "unable to retrieve public posts for #{@person.diaspora_handle}"
raise e
end
begin
process_posts
rescue => e
logger.error "unable to process public posts for #{@person.diaspora_handle}"
raise e
end
end
# fetch the public posts of the person from their server and save the
# JSON response to `@data`
def retrieve_posts
set_fetch_status Public::Status_Running
logger.info "fetching public posts for #{@person.diaspora_handle}"
resp = Faraday.get("#{@person.url}people/#{@person.guid}/stream") do |req|
req.headers['Accept'] = 'application/json'
req.headers['User-Agent'] = 'diaspora-fetcher'
end
logger.debug "fetched response: #{resp.body.to_s[0..250]}"
@data = JSON.parse resp.body
set_fetch_status Public::Status_Fetched
end
# process the public posts that were previously fetched with `retrieve_posts`
# adds posts, which pass some basic sanity-checking
# @see validate
def process_posts
@data.each do |post|
next unless validate(post)
logger.info "saving fetched post (#{post['guid']}) to database"
logger.debug "post: #{post.to_s[0..250]}"
entry = StatusMessage.diaspora_initialize(
author: @person,
public: true,
guid: post["guid"],
text: post["text"],
provider_display_name: post["provider_display_name"],
created_at: ActiveSupport::TimeZone.new("UTC").parse(post["created_at"]).to_datetime,
)
entry.save
end
set_fetch_status Public::Status_Processed
end
# set and save the fetch status for the current person
def set_fetch_status status
return if @person.nil?
@person.fetch_status = status
@person.save
end
# perform various validations to make sure the post can be saved without
# troubles
# @see check_existing
# @see check_author
# @see check_public
# @see check_type
def validate post
check_existing(post) && check_author(post) && check_public(post) && check_type(post)
end
# hopefully there is no post with the same guid somewhere already...
def check_existing post
new_post = (Post.find_by_guid(post['guid']).blank?)
logger.warn "a post with that guid (#{post['guid']}) already exists" unless new_post
new_post
end
# checks if the author of the given post is actually from the person
# we're currently processing
def check_author post
guid = post['author']['guid']
equal = (guid == @person.guid)
unless equal
logger.warn "the author (#{guid}) does not match the person currently being processed (#{@person.guid})"
end
equal
end
# returns wether the given post is public
def check_public post
ispublic = (post['public'] == true)
logger.warn "the post (#{post['guid']}) is not public, this is not intended..." unless ispublic
ispublic
end
# see, if the type of the given post is something we can handle
def check_type post
type_ok = (post['post_type'] == "StatusMessage")
logger.warn "the post (#{post['guid']}) has a type, which cannot be handled (#{post['post_type']})" unless type_ok
type_ok
end
end; end; end
|
