サンプルコード

四種類のコマンドラインスクリプト (Python) を実装してみます。

  • OAuth で認証を実行し、認証トークンを取得する。
  • 新着情報を取得する。
  • スケジュールの情報を iCal 形式に変換する。
  • ToDo の情報を CSV 形式に変換する。

Python の実行環境と、依存するライブラリをインストールしてください。

パッケージ管理に easy_install (setuptools) を使う場合は次のコマンドでインストールできます。

$ sudo easy_install httplib2
$ sudo easy_install oauth2
$ sudo easy_install icalendar

認証

認証処理を実行し、認証トークンを取得します。 OAuth の認証フローは3ステップ必要ですので、少し長めです。

 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
#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""Sample script to execute OAuth three-legged dance flow for 'cybozulive'.
"""

# Additional modules from PyPI.
import oauth2

CYBOZULIVE_OAUTH = {
    'STEP_1': 'https://api.cybozulive.com/oauth/initiate',
    'STEP_2': 'https://api.cybozulive.com/oauth/authorize',
    'STEP_3': 'https://api.cybozulive.com/oauth/token',
}

CONSUMER_TOKEN = {
    'key'   : '<CONSUMER_KEY>',
    'secret': '<CONSUMER_SECRET>'
}


def main():
    consumer = oauth2.Consumer(CONSUMER_TOKEN['key'], CONSUMER_TOKEN['secret'])

    # Step 1 to retrieve temporal token.
    client1 = oauth2.Client(consumer)
    header, body = client1.request(CYBOZULIVE_OAUTH['STEP_1'])
    if int(header['status']) != 200:
        msg = ["Failed at authentication step 1.", body]
        raise SystemExit('\n'.join(msg))
    ''' 'body' is something like
    oauth_token=XXXX&oauth_token_secret=XXXX&oauth_callback_confirmed=true
    '''
    tmp_token = oauth2.Token.from_string(body)

    # Step 2 to authenticate user on service site, and get verifier code.
    url = '%s?oauth_token=%s' % (CYBOZULIVE_OAUTH['STEP_2'], tmp_token.key)
    print "Go this URL and get verifier code."
    print url
    try:
        verifier = raw_input("Verifier >")
        if not verifier:
            raise SystemExit(1)
    except KeyboardInterrupt:
        raise SystemExit(1)
    except EOFError:
        raise SystemExit(1)
    tmp_token.set_verifier(verifier)

    # Step 3 to retrieve access token.
    client2 = oauth2.Client(consumer, tmp_token)
    header, body = client2.request(CYBOZULIVE_OAUTH['STEP_3'])
    if int(header['status']) != 200:
        msg = ["Failed at authentication step 3.", body]
        raise SystemExit('\n'.join(msg))
    token = oauth2.Token.from_string(body)
    print "Save your token."
    print "Key   : ", token.key
    print "Secret: ", token.secret

if __name__ == '__main__':
    main()

# vim: set et ts=4 sw=4 cindent fileencoding=utf-8 :

以降のサンプルでは、次の設定ファイル (settings.py) を使います。 上記のスクリプトの CONSUMER_TOKEN と、取得した認証トークンのキー/バリューのペアを登録して保存しておきます。

1
2
3
4
OAUTH_TOKEN = {
    'CONSUMER': ('<CONSUMER_KEY>', '<CONSUMER_SECRET>'),
    'ACCESSOR': ('<ACCESS_TOKEN_KEY>', '<ACCESS_TOKEN_SECRET>')
}

新着情報

認証トークンを使って新着情報をダウンロードします。 レスポンスが Atom 形式ですので、そのまま XML ファイルとして保存します。 スクリプトを定期的に実行させ、XML ファイルを別途 RSS リーダーなどに登録しておくことで、 定期的に新着情報を確認できます。

 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
#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""Sample script to fetch a OAuth-protected feed.
"""

# Additional modules from PyPI.
import oauth2

URL = 'https://api.cybozulive.com/api/notification/V2'


def main():
    from settings import OAUTH_TOKEN
    consumer = oauth2.Consumer(*OAUTH_TOKEN['CONSUMER'])
    token = oauth2.Token(*OAUTH_TOKEN['ACCESSOR'])
    client = oauth2.Client(consumer, token)
    header, body = client.request(URL)
    if int(header['status']) == 200:
        fp = open('notification.xml', 'wb')
        fp.write(body)
        fp.close()

if __name__ == '__main__':
    main()

# vim: set et ts=4 sw=4 cindent fileencoding=utf-8 :

スケジュール

スケジュールの情報を取得し、XML を解析して iCal 形式に変換します。 Microsoft Outlook や Google Calendar などで iCal 形式を読み込めます。

 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
#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""Sample script to fetch schedule feed and convert it into iCal file.
"""

# Standard modules.
import datetime

try:
    from lxml import etree
except ImportError:
    import xml.etree.ElementTree as etree

# Additional modules from PyPI.
import oauth2
import icalendar

URL = 'https://api.cybozulive.com/api/schedule/V2'
ATOM_NAMESPACE = 'http://www.w3.org/2005/Atom'
CYBOZULIVE_COMMON_NAMESPACE = 'http://schemas.cybozulive.com/common/2010'


def _iso2datetime(string):
    if len(string) == 10:
        y, m, d = [int(f) for f in string.split('-')]
        return datetime.date(y, m, d)
    date, time = string[:-1].split('T')
    y, m, d = [int(f) for f in date.split('-')]
    H, M, S = [int(f) for f in time.split(':')]
    return datetime.datetime(y, m, d, H, M, S, tzinfo=icalendar.UTC)


def atom2ical(atom):

    def mapper(elements):
        ev = icalendar.Event()
        for e in elements:
            if e.tag.startswith('{%s}' % (ATOM_NAMESPACE,)):
                t = e.tag[len(ATOM_NAMESPACE) + 2:]
                if t == 'title':
                    ev.set('summary', e.text)
                elif t in ('content', 'summary'):
                    ev.set('description', e.text)
            elif e.tag.startswith('{%s}' % (CYBOZULIVE_COMMON_NAMESPACE,)):
                t = e.tag[len(CYBOZULIVE_COMMON_NAMESPACE) + 2:]
                if t == 'when':
                    start = _iso2datetime(e.get('startTime'))
                    end = _iso2datetime(e.get('endTime'))
                    if start:
                        if type(start) == datetime.date:
                            ev.set('dtstart', start)
                        else:
                            ev['dtstart'] = start.strftime('%Y%m%dT%H%M%SZ')
                    if end:
                        if type(end) == datetime.date:
                            ev.set('dtend', end)
                        else:
                            ev['dtend'] = end.strftime('%Y%m%dT%H%M%SZ')
        return ev

    ret = icalendar.Calendar()
    ret['PRODID'] = '-//cybozulive.com//Calendar Feed//JA'
    ret['VERSION'] = '2.0'

    feed = etree.fromstring(atom)
    for elem in feed.getchildren():
        if elem.tag == '{%s}entry' % (ATOM_NAMESPACE,):
            ev = mapper(elem.getchildren())
            ret.add_component(ev)
    return ret


def main():
    from settings import OAUTH_TOKEN
    consumer = oauth2.Consumer(*OAUTH_TOKEN['CONSUMER'])
    token = oauth2.Token(*OAUTH_TOKEN['ACCESSOR'])
    client = oauth2.Client(consumer, token)
    header, body = client.request(URL)
    if int(header['status']) == 200:
        ical = atom2ical(body)
        fp = open('cylive-schedule.ics', 'wb')
        fp.write(ical.as_string())
        fp.close()

if __name__ == '__main__':
    main()

# vim: set et ts=4 sw=4 cindent fileencoding=utf-8 :

ToDo リスト

ToDo リストの情報を取得し、XML を解析して CSV 形式に変換します。 エンコーディングに “cp932” を使用することで、Microsoft Excel で読み込むことが可能になります。

 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
#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""Sample script to fetch task feed, and convert it into CSV file.
"""

# Standard modules.
import csv
from datetime import datetime

try:
    from lxml import etree
except ImportError:
    import xml.etree.ElementTree as etree

# Additional modules from PyPI.
import oauth2

URL = 'https://api.cybozulive.com/api/task/V2'
ATOM_NAMESPACE = 'http://www.w3.org/2005/Atom'
CYBOZULIVE_COMMON_NAMESPACE = 'http://schemas.cybozulive.com/common/2010'
ENCODING = 'cp932'


def atom2csv(atom, writer):

    def mapper(elements):
        dt = {'title': u'Unknown',
              'group': u'Unknown',
              'number':u'Unknown',
              'status': u'Unknown',
              'description': u'',
              'deadline': u'Unknown'}
        for e in elements:
            if e.tag.startswith('{%s}' % (ATOM_NAMESPACE,)):
                t = e.tag[len(ATOM_NAMESPACE) + 2:]
                if t == 'title':
                    dt['title'] = e.text
                elif t in ('content', 'summary'):
                    dt['description'] = e.text
            elif e.tag.startswith('{%s}' % (CYBOZULIVE_COMMON_NAMESPACE,)):
                t = e.tag[len(CYBOZULIVE_COMMON_NAMESPACE) + 2:]
                if t == 'task':
                    dt['status'] = e.get('valueString')
                    dt['number'] = e.get('sequence')
                elif t == 'who':
                    pass
                elif t == 'group':
                    dt['group'] = e.get('valueString')
                elif t == 'when':
                    dt['deadline'] = e.get('endTime')
        ret = [dt['group'].encode(ENCODING),
               dt['number'],
               dt['title'].encode(ENCODING),
               dt['deadline'],
               dt['status'].encode(ENCODING),
               dt['description'].encode(ENCODING)]
        return ret

    def printer(row):
        writer.writerow(row)

    feed = etree.fromstring(atom)
    for elem in feed.getchildren():
        if elem.tag == '{%s}entry' % (ATOM_NAMESPACE,):
            row = mapper(elem.getchildren())
            printer(row)


def main():
    from settings import OAUTH_TOKEN
    consumer = oauth2.Consumer(*OAUTH_TOKEN['CONSUMER'])
    token = oauth2.Token(*OAUTH_TOKEN['ACCESSOR'])
    client = oauth2.Client(consumer, token)
    header, body = client.request(URL)
    if int(header['status']) == 200:
        fp = open('cylive-task.csv', 'wb')
        writer = csv.writer(fp)
        atom2csv(body, writer)
        fp.close()

if __name__ == '__main__':
    main()

# vim: set et ts=4 sw=4 cindent fileencoding=utf-8 :