This repository was archived by the owner on Nov 29, 2019. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 34
Expand file tree
/
Copy pathutils.py
More file actions
209 lines (161 loc) · 7.07 KB
/
Copy pathutils.py
File metadata and controls
209 lines (161 loc) · 7.07 KB
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
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
# -*- coding: utf-8 -*-
from sqlalchemy import desc
from sqlalchemy.orm import contains_eager, joinedload, subqueryload
from assembl import models
from assembl.models.idea import MessageView
from assembl.models.post import deleted_publication_states
from assembl.models.timeline import Phases
from assembl.graphql.utils import get_root_thematic_for_phase
def format_date(datetime_to_format):
return datetime_to_format.strftime('%d/%m/%Y')
def get_posts(idea, start=None, end=None,
include_deleted=None,
publication_states=None,
only_top_posts=False,
include_moderating=False):
"""
Get all posts given a given idea filtered by start and end dates.
@param: idea Idea
@param: start datetime
@param: end datetime
@param: publication_states: PublicationState[] filter on post publication states
@param: include_moderating: bool if true, includes posts currently under moderation
@param: only_top_post: bool
"""
Post = models.Post
related = idea.get_related_posts_query(True, include_moderating=include_moderating, include_deleted=include_deleted)
query = Post.query.join(
related, Post.id == related.c.post_id
).order_by(desc(Post.creation_date), Post.id
).options(subqueryload(Post.creator),
subqueryload(Post.creator, models.AgentProfile.accounts))
if start is not None:
query = query.filter(Post.creation_date >= start)
if end is not None:
query = query.filter(Post.creation_date <= end)
if publication_states:
assert isinstance(publication_states, (tuple, list, set))
if len(publication_states) == 1:
query = query.filter(Post.publication_state == publication_states[0])
else:
query = query.filter(Post.publication_state.in_(publication_states))
if only_top_posts:
query = query.filter(Post.parent_id == None) # noqa: E711
return query
def get_published_top_posts(idea, start=None, end=None):
query = get_posts(idea, start=start, end=end,
include_deleted=False,
publication_states=[models.PublicationStates.PUBLISHED],
only_top_posts=True)
return query
def get_deleted_posts(idea, start=None, end=None):
"""Get deleted posts for the given idea filtered by start and end dates.
@param: idea Idea
@param: start datetime
@param: end datetime
"""
# Don't use include_deleted=False here. A post can be deleted (publication_state) but not tombstoned
query = get_posts(idea, start, end, include_deleted=None, publication_states=deleted_publication_states)
return query
def get_related_extracts(content):
extracts = content.db.query(models.Extract
).filter(models.Extract.content_id == content.id)
return extracts
def get_all_phase_root_ideas(discussion):
root_ideas = []
for phase in discussion.timeline_phases:
if phase.root_idea:
root_ideas.append(phase.root_idea)
return root_ideas
def get_descendants(ideas):
if not ideas:
return []
descendants = []
for idea in ideas:
descendants.extend(idea.get_all_descendants(id_only=True, inclusive=True))
return descendants
def get_multicolumns_ideas(discussion, start=None, end=None):
return get_ideas_for_export(discussion, MessageView.messageColumns.value, start=start, end=end)
def get_survey_ideas(discussion, start=None, end=None):
return get_ideas_for_export(discussion, MessageView.survey.value,
start=start, end=end)
def get_thread_ideas(discussion, start=None, end=None):
return get_ideas_for_export(discussion, MessageView.thread.value, start=start, end=end)
def get_bright_mirror_ideas(discussion, start=None, end=None):
return get_ideas_for_export(discussion, MessageView.brightMirror.value, start=start, end=end)
def get_vote_session_ideas(discussion, start=None, end=None):
return get_ideas_for_export(discussion, MessageView.voteSession.value, start=start, end=end)
def get_ideas_for_export(discussion, module_type=None, start=None, end=None):
ideas = []
for phase in discussion.timeline_events:
# If [start, end] and [phase.start, phase.end] don't overlap,
# don't export the ideas.
if phase.end < start or phase.start > end:
continue
root_thematic = get_root_thematic_for_phase(phase)
if root_thematic is not None:
ideas_query = get_ideas(phase)
if module_type is None:
ideas.extend(ideas_query.all())
else:
ideas.extend(ideas_query.filter(
models.Idea.message_view_override == module_type).all())
return ideas
def get_ideas(phase, options=None):
root_thematic = get_root_thematic_for_phase(phase)
if root_thematic is None:
return []
model = models.Idea
query = model.query
descendants_query = root_thematic.get_descendants_query(inclusive=False)
query = query.filter(model.id.in_(descendants_query))
query = query.outerjoin(
models.Idea.source_links
).filter(
~model.sqla_type.in_(('question', 'vote_proposal')),
model.hidden == False, # noqa: E712
model.tombstone_date == None # noqa: E711
).options(
contains_eager(models.Idea.source_links),
joinedload(models.Idea.title).joinedload("entries"),
joinedload(models.Idea.description).joinedload("entries"),
).order_by(models.IdeaLink.order, models.Idea.creation_date)
if options is not None:
query = query.options(*options)
return query
def get_posts_for_phases(
discussion, identifiers, include_deleted=False, include_moderating=None):
"""Return related posts for the ideas with module type in `identifiers` on `discussion`.
"""
if not discussion:
return None
module_types = identifiers[:]
if Phases.multiColumns.value in module_types:
# backward compatibility with bluenove-actionable
module_types.remove(Phases.multiColumns.value)
module_types.append(MessageView.messageColumns.value)
ideas = []
for phase in discussion.timeline_events:
root_thematic = get_root_thematic_for_phase(phase)
if root_thematic is not None:
ideas_query = get_ideas(phase)
ideas.extend(ideas_query.filter(
models.Idea.message_view_override.in_(module_types)).all())
if not ideas:
return None
model = models.AssemblPost
query = discussion.db.query(model)
queries = []
for idea in ideas:
# Note we are not sending user_id
related = idea.get_related_posts_query(
True, include_deleted, include_moderating)
related_query = query.join(
related, model.id == related.c.post_id
)
queries.append(related_query)
query = queries[0].union_all(*queries[1:])
if not include_deleted:
return query.filter(
model.publication_state == models.PublicationStates.PUBLISHED)
return query