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
|
from jb.decorators import LOG
from jb.flow.monitoring import emit_assignment_event, emit_error_event
from jb.managers.amt import (
REJECT_MESSAGE_UNKNOWN_ASSIGNMENT,
AMTManager,
)
from jb.managers.assignment import AssignmentManager
from jb.managers.bonus import BonusManager
from jb.managers.hit import HitManager
from jb.models.assignment import Assignment
from jb.models.definitions import AssignmentStatus
from jb.models.event import MTurkEvent
def process_assignment_submitted(
amtm: AMTManager,
am: AssignmentManager,
hm: HitManager,
bm: BonusManager,
event: MTurkEvent,
) -> None:
"""
Reject any submitted assignments
"""
#
# Step 1: Attempt to get the Assignment out of the API
#
LOG.info(f"{event=}")
# This is the assignment model from AMT. In the DB, we should only have
# the AssignmentStub
# This call is hitting AMT, not our db.
# The API won't even return is unless it has been submitted. So if this
# fails, we don't need to do anything (i.e. reject it).
assignment = amtm.get_assignment_if_exists(
amt_assignment_id=event.amt_assignment_id
)
if assignment is None:
# It is not found in amt, either it is invalid, not yet submitted, or
# already been approved/rejected, so we just do nothing ...
# todo: maybe we confirm its state matches what we have in the db
LOG.warning(f"No assignment found on AMT: {event.amt_assignment_id}")
emit_error_event(
event_type="assignment_not_found_in_amt",
amt_hit_type_id=event.amt_hit_type_id,
)
return
# Even if the assignment doesn't exist, the hit must ...
hit = hm.get_from_amt_id(amt_hit_id=assignment.amt_hit_id)
reject_assignment(
amtm=amtm,
am=am,
hm=hm,
amt_assignment_id=assignment.amt_assignment_id,
msg=REJECT_MESSAGE_UNKNOWN_ASSIGNMENT,
amt_hit_type_id=hit.amt_hit_type_id,
)
review_hit(amtm=amtm, hm=hm, assignment=assignment)
return
def review_hit(amtm: AMTManager, hm: HitManager, assignment: Assignment) -> None:
# Reviewable to Reviewing
amtm.update_hit_review_status(amt_hit_id=assignment.amt_hit_id, revert=False)
hit, _ = amtm.get_hit_if_exists(amt_hit_id=assignment.amt_hit_id)
if hit is None:
LOG.warning(f"Hit not found when trying to review hit: {assignment.amt_hit_id}")
return
# Update the db
hm.update_hit(hit)
return
def reject_assignment(
amtm: AMTManager,
am: AssignmentManager,
hm: HitManager,
amt_assignment_id: str,
msg: str,
amt_hit_type_id: str,
) -> Assignment:
# Reject in AMT, update db
res = amtm.reject_assignment_if_possible(
amt_assignment_id=amt_assignment_id, msg=msg
)
if res is None:
# We failed to reject this assignment. This happens if the assignment
# is already approved or already rejected.
emit_error_event(
event_type="failed_to_reject_assignment",
amt_hit_type_id=amt_hit_type_id,
)
LOG.exception(f"Failed to reject assignment: {amt_assignment_id}")
# We just rejected this assignment, get it from amazon again
assignment = amtm.get_assignment(amt_assignment_id=amt_assignment_id)
assert assignment.status == AssignmentStatus.Rejected
# And update the db. The assignment may not actually exist in the db (if
# a baddie intercepted it and is trying to game us). So, we might
# need to create as assignment first ...
stub = am.get_stub_if_exists(amt_assignment_id=assignment.amt_assignment_id)
if stub is None:
LOG.warning(
f"Rejected assignment doesn't exist in DB. Creating ... : {amt_assignment_id}"
)
# Even if the assignment doesn't exist, the hit must ...
hit = hm.get_from_amt_id(amt_hit_id=assignment.amt_hit_id)
assignment.hit_id = hit.id
am.create(assignment=assignment)
am.reject(assignment=assignment)
emit_assignment_event(
status=AssignmentStatus.Rejected, amt_hit_type_id=amt_hit_type_id, reason=msg
)
LOG.warning(f"Rejected assignment: {amt_assignment_id}")
return assignment
|
