Private
Public Access
1
0

removing mailprocessing file

This commit is contained in:
Sander Roosendaal
2021-12-17 10:32:53 +01:00
parent 27a560c5a7
commit e1931065f9
14 changed files with 85 additions and 816 deletions

View File

@@ -29,7 +29,7 @@ from rowingdata import rower as rrower
from rowingdata import rowingdata as rrdata
import rowers.uploads as uploads
from rowers.mailprocessing import make_new_workout_from_email, send_confirm
import rowers.polarstuff as polarstuff
import rowers.c2stuff as c2stuff
import rowers.rp3stuff as rp3stuff
@@ -60,102 +60,6 @@ def rdata(file_obj, rower=rrower()): # pragma: no cover
return result
def processattachment(rower, fileobj, title, uploadoptions,testing=False):
try:
filename = fileobj.name
# filename = os.path.abspath(fileobj.name)
except AttributeError:
filename = fileobj[6:]
# test if file exists and is not empty
try:
with io.open('media/'+filename,'rb') as fop:
line = fop.readline()
except (IOError, UnicodeEncodeError): # pragma: no cover
return 0
# set user
if rower.user.is_staff and 'username' in uploadoptions:
users = User.objects.filter(username=uploadoptions['username'])
if len(users)==1:
therower = users[0].rower
elif uploadoptions['username'] == '': # pragma: no cover
therower = rower
else: # pragma: no cover
return 0
else:
therower = rower
uploadoptions['secret'] = settings.UPLOAD_SERVICE_SECRET
uploadoptions['user'] = therower.user.id
uploadoptions['file'] = 'media/'+filename
uploadoptions['title'] = title
url = settings.UPLOAD_SERVICE_URL
if not testing: # pragma: no cover
response = requests.post(url,data=uploadoptions)
# print("Upload response status code",response.status_code, response.json())
if response.status_code == 200:
response_json = response.json()
workoutid = [int(response_json['id'])]
else:
workoutid = [0]
# this is ugly and needs to be done better
if testing:
workoutid = [
make_new_workout_from_email(therower, filename, title,testing=testing)
]
if workoutid[0] and uploadoptions and not 'error' in uploadoptions:
workout = Workout.objects.get(id=workoutid[0])
uploads.make_private(workout, uploadoptions)
uploads.set_workouttype(workout, uploadoptions)
uploads.do_sync(workout, uploadoptions)
if 'raceid' in uploadoptions and workoutid[0] and rower.user.is_staff:
if testing and workoutid[0]:
w = Workout.objects.get(id = workoutid[0])
w.startdatetime = timezone.now()
w.date = timezone.now().date()
w.save()
try:
race = VirtualRace.objects.get(id=uploadoptions['raceid'])
if race.manager == rower.user:
result = email_submit_race(therower,race,workoutid[0])
except VirtualRace.DoesNotExist: # pragma: no cover
pass
return workoutid
def get_from_address(message):
from_address = message.from_address[0].lower()
if message.encoded: # pragma: no cover
body = message.text.splitlines()
else:
body = message.get_body().splitlines()
try:
first_line = body[0].lower()
except IndexError: # pragma: no cover
first_line = ''
try:
first_line = first_line.decode('utf-8')
except AttributeError: # pragma: no cover
pass
if "quiske" in first_line: # pragma: no cover
match = re.search(r'[\w\.-]+@[\w\.-]+', first_line)
return match.group(0)
return from_address
class Command(BaseCommand):