120 lines
3.9 KiB
Python
120 lines
3.9 KiB
Python
""" Process emails """
|
|
import sys
|
|
import os
|
|
|
|
import zipfile
|
|
from zipfile import BadZipFile
|
|
import re
|
|
import time
|
|
import traceback
|
|
from time import strftime
|
|
|
|
import requests
|
|
import json
|
|
|
|
import io
|
|
|
|
from django.core.management.base import BaseCommand
|
|
|
|
from django.urls import reverse
|
|
from django.conf import settings
|
|
|
|
from django.utils import timezone
|
|
from rowers.models import User, Workout, Rower
|
|
|
|
from rowingdata import rower as rrower
|
|
from rowingdata import rowingdata as rrdata
|
|
|
|
import rowers.uploads as uploads
|
|
|
|
from rowers.opaque import encoder
|
|
from rowers.integrations import *
|
|
from rowers.rower_rules import user_is_not_basic, user_is_coachee
|
|
from rowers.utils import dologging
|
|
|
|
# If you find a solution that does not need the two paths, please comment!
|
|
sys.path.append('$path_to_root_of_project$')
|
|
sys.path.append('$path_to_root_of_project$/$project_name$')
|
|
|
|
os.environ['DJANGO_SETTINGS_MODULE'] = '$project_name$.settings'
|
|
|
|
PY3K = sys.version_info >= (3, 0)
|
|
|
|
if not getattr(__builtins__, "WindowsError", None):
|
|
class WindowsError(OSError):
|
|
pass
|
|
|
|
|
|
def rdata(file_obj, rower=rrower()): # pragma: no cover
|
|
""" Read rowing data file and return 0 if file doesn't exist"""
|
|
try:
|
|
result = rrdata(file_obj, rower=rower)
|
|
except IOError:
|
|
result = 0
|
|
|
|
return result
|
|
|
|
|
|
class Command(BaseCommand):
|
|
def add_arguments(self, parser):
|
|
parser.add_argument(
|
|
'--testing',
|
|
action='store_true',
|
|
dest='testing',
|
|
default=False,
|
|
help="Run in testing mode, don't send emails", )
|
|
parser.add_argument(
|
|
'--mailbox',
|
|
action='store_true',
|
|
dest='mailbox',
|
|
default='workouts',
|
|
help="Changing mailbox name", )
|
|
|
|
"""Run the Email processing command """
|
|
def handle(self, *args, **options):
|
|
# Polar
|
|
try:
|
|
polarintegration = PolarIntegration(None)
|
|
|
|
_ = polarintegration.get_workouts()
|
|
except: # pragma: no cover
|
|
exc_type, exc_value, exc_traceback = sys.exc_info()
|
|
lines = traceback.format_exception(exc_type, exc_value, exc_traceback)
|
|
dologging('processemail.log', ''.join('!! ' + line for line in lines))
|
|
|
|
# Concept2
|
|
rowers = Rower.objects.filter(c2_auto_import=True)
|
|
for r in rowers: # pragma: no cover
|
|
try:
|
|
c2integration = C2Integration(r.user)
|
|
_ = c2integration.get_workouts()
|
|
except: # pragma: no cover
|
|
exc_type, exc_value, exc_traceback = sys.exc_info()
|
|
lines = traceback.format_exception(exc_type, exc_value, exc_traceback)
|
|
dologging('processemail.log', ''.join('!! ' + line for line in lines))
|
|
|
|
#rowers = Rower.objects.filter(rp3_auto_import=True)
|
|
#for r in rowers: # pragma: no cover
|
|
# try:
|
|
# rp3_integration = RP3Integration(r.user)
|
|
# _ = rp3_integration.get_workouts()
|
|
# except: # pragma: no cover
|
|
# exc_type, exc_value, exc_traceback = sys.exc_info()
|
|
# lines = traceback.format_exception(exc_type, exc_value, exc_traceback)
|
|
# dologging('processemail.log', ''.join('!! ' + line for line in lines))
|
|
|
|
rowers = Rower.objects.filter(nk_auto_import=True)
|
|
for r in rowers: # pragma: no cover
|
|
try:
|
|
dologging("nklog.log","NK Auto import set for rower {id}".format(id=r.user.id))
|
|
nk_integration = NKIntegration(r.user)
|
|
_ = nk_integration.get_workouts()
|
|
except: # pragma: no cover
|
|
exc_type, exc_value, exc_traceback = sys.exc_info()
|
|
lines = traceback.format_exception(exc_type, exc_value, exc_traceback)
|
|
dologging('processemail.log', ''.join('!! ' + line for line in lines))
|
|
|
|
|
|
self.stdout.write(self.style.SUCCESS(
|
|
'Successfully processed email attachments'))
|