git2edx/edxStudio.py at master · mitodl/git2edx · GitHub

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
#!/usr/bin/python
#
# File:   edxStudio.py
# Date:   12-Mar-14
# Author: I. Chuang <ichuang@mit.edu>
#
# upload / download course .tar.gz files to/from edX Studio
#
# usage:
#
# python edxStudio.py course_id tar_file_name
#
# Put username and passsword in config.py; eg
#    username = "me"
#    password = "my password"


import os, sys
import time
import requests

try:
    from config import *
except:
    pass

class edxStudio(object):

    def __init__(self, base="https://studio.edge.edx.org", username='', password=''):
        self.ses = requests.session()
        self.BASE = base
        self.login(username, password)

    def login(self, username, pw):
        url = '%s/signin' % self.BASE
        r1 = self.ses.get(url)
        csrf = self.ses.cookies['csrftoken']
        url2 = '%s/login_post' % self.BASE
        headers = {'X-CSRFToken':csrf,
                   'Referer': '%s/signin' % self.BASE}
        r2 = self.ses.post(url2, data={'email': username, 'password': pw}, headers=headers)

        # print "login ret = ", r2
        if not r2.status_code==200:
            print "Login failed!"
            print r2.text

    def do_download(self, course_id):

        print "Downloading tar.gz for %s" % (course_id)

        (org, num, sem) = course_id.split('/')
        # url = '%s/export/%s/branch/draft/block/%s?_accept=application/x-tgz' % (self.BASE, course_id.replace('/','.'), sem)
        url = '%s/export/%s?_accept=application/x-tgz' % (self.BASE, course_id)#.replace('/','.'))
        r3 = self.ses.get(url)

        if not r3.ok or (r3.status_code==404):
            url = '%s/export/slashes:%s+%s?_accept=application/x-tgz' % (self.BASE, course_id.replace('/','+'), sem)
            r3 = self.ses.get(url)

        dt = time.ctime(time.time()).replace(' ','_').replace(':','')
        fn = 'COURSE_DATA/COURSE-%s___%s.tar.gz' % (course_id.replace('/','__'),dt)

        open(fn, 'w').write(r3.content)
        print "--> %s" % (fn)
        return fn

    def do_upload(self, course_id, tfn, nwait=20):

        print "Uploading %s for %s" % (tfn, course_id)

        tfnbn = os.path.basename(tfn)
        # (org, num, sem) = course_id.split('/')
        url = '%s/import/%s' % (self.BASE, course_id)
        # url = '%s/import/%s/branch/draft/block/%s' % (self.BASE, course_id.replace('/','.'), sem)

        files = {'course-data': (tfnbn, open(tfn, 'rb'), 'application/x-gzip')}

        csrf = self.ses.cookies['csrftoken']
        print "csrf=%s" % csrf
        headers = {'X-CSRFToken':csrf,
                   'Referer': url,
                   'Accept': 'application/json, text/javascript, */*; q=0.01',
               }

        #r2 = self.ses.get(url)
        #if not r2.ok or (r2.status_code==404):
        #    url = '%s/import/slashes:%s' % (self.BASE, course_id.replace('/','+'))
        #    mode = 2
        #    r2 = self.ses.get(url)

        #print r2.status_code

        print url

        try:
            r3 = self.ses.post(url, files=files, headers=headers)
        except Exception as err:
            print "Error %s" % str(err)
            print "url=%s, files=%s, headers=%s" % (url, files, headers)
            sys.stdout.flush()
            sys.exit(-1)

        # print r3.headers
        print "r3 = ", r3.status_code
        print "--> %s" % (r3.content)

        url = '%s/import_status/%s/%s' % (self.BASE, course_id, tfnbn.replace('/','-'))
        #url = '%s/import/%s' % (self.BASE, course_id)
        print url

        for k in range(nwait):
            r4 = self.ses.get(url)
            if r4.ok:
                print r4.content
                if r4.json()["ImportStatus"]==4:
                    print "Done!"
                    return
            else:
                print r4
            sys.stdout.flush()
            time.sleep(2)

        dt = time.ctime(time.time()).replace(' ','_').replace(':','')


#-----------------------------------------------------------------------------

if __name__=='__main__':

    if len(sys.argv)>2:
        cid = sys.argv[1]
        tfn = sys.argv[2]
    else:
        print "Usage: python %s course_id tar_file_name.tgz" % (sys.argv[0])
        exit(0)

    base = "https://studio.edge.edx.org"
    if len(sys.argv)>3:
        base = sys.argv[3]

    es = edxStudio(base=base, username=username, password=password)
    es.do_upload(cid, tfn)
    #es.do_download(cid)