Commit 4bb66891 authored by Romain Bignon's avatar Romain Bignon

Update of modules

parent ac7f9f36
# -*- coding: utf-8 -*-
# Copyright(C) 2014 Romain Bignon
#
# This file is part of weboob.
#
# weboob is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# weboob is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with weboob. If not, see <http://www.gnu.org/licenses/>.
from weboob.browser import LoginBrowser, URL, need_login
from weboob.exceptions import BrowserIncorrectPassword
from .pages import LoginPage, AccountsPage
__all__ = ['AlloRestoBrowser']
class AlloRestoBrowser(LoginBrowser):
BASEURL = 'https://www.alloresto.fr'
login = URL('/identification-requise.*', LoginPage)
accounts = URL('/chez-moi/releve-compte-miams', AccountsPage)
def do_login(self):
assert isinstance(self.username, basestring)
assert isinstance(self.password, basestring)
self.accounts.stay_or_go()
self.page.login(self.username, self.password)
if self.login.is_here():
raise BrowserIncorrectPassword()
@need_login
def get_accounts_list(self):
return self.accounts.stay_or_go().iter_accounts()
@need_login
def get_account(self, id):
assert isinstance(id, basestring)
for a in self.get_accounts_list():
if a.id == id:
return a
return None
@need_login
def get_history(self, account):
return self.accounts.stay_or_go().get_transactions(type='consommable')
@need_login
def get_coming(self, account):
return self.accounts.stay_or_go().get_transactions(type='acquisition')
# -*- coding: utf-8 -*-
# Copyright(C) 2014 Romain Bignon
#
# This file is part of weboob.
#
# weboob is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# weboob is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with weboob. If not, see <http://www.gnu.org/licenses/>.
from weboob.capabilities.bank import CapBank, AccountNotFound
from weboob.tools.backend import Module, BackendConfig
from weboob.tools.value import ValueBackendPassword
from .browser import AlloRestoBrowser
__all__ = ['AlloRestoModule']
class AlloRestoModule(Module, CapBank):
NAME = 'alloresto'
MAINTAINER = u'Romain Bignon'
EMAIL = 'romain@weboob.org'
VERSION = '1.4'
DESCRIPTION = u'Allo Resto'
LICENSE = 'AGPLv3+'
CONFIG = BackendConfig(ValueBackendPassword('login', label='Identifiant', masked=False),
ValueBackendPassword('password', label='Mot de passe'))
BROWSER = AlloRestoBrowser
def create_default_browser(self):
return self.create_browser(self.config['login'].get(),
self.config['password'].get())
def iter_accounts(self):
return self.browser.get_accounts_list()
def get_account(self, _id):
account = self.browser.get_account(_id)
if account:
return account
else:
raise AccountNotFound()
def iter_history(self, account):
return self.browser.get_history(account)
def iter_coming(self, account):
return self.browser.get_coming(account)
# -*- coding: utf-8 -*-
# Copyright(C) 2014 Romain Bignon
#
# This file is part of weboob.
#
# weboob is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# weboob is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with weboob. If not, see <http://www.gnu.org/licenses/>.
import datetime
from decimal import Decimal
from weboob.browser.pages import HTMLPage, LoggedPage
from weboob.browser.elements import ItemElement, method
from weboob.browser.filters.standard import CleanDecimal, CleanText, Filter, TableCell
from weboob.capabilities.bank import Account
from weboob.tools.capabilities.bank.transactions import FrenchTransaction as Transaction
class LoginPage(HTMLPage):
def login(self, username, password):
form = self.get_form(xpath='//form[has-class("form_o")]')
form['uname'] = username
form['pass'] = password
form.submit()
class AccountsPage(LoggedPage, HTMLPage):
@method
class iter_accounts(ItemElement):
def __call__(self):
return self
klass = Account
obj_id = '0'
obj_label = u'Compte miams'
obj_balance = CleanDecimal('//div[@class="compteur"]//strong', replace_dots=True)
obj_currency = u'MIAM'
obj_coming = CleanDecimal('//table[@id="solde_acquisition_lignes"]//th[@class="col_montant"]', default=Decimal('0'), replace_dots=True)
class MyDate(Filter):
MONTHS = ['janv', u'févr', u'mars', u'avr', u'mai', u'juin', u'juil', u'août', u'sept', u'oct', u'nov', u'déc']
def filter(self, txt):
day, month, year = txt.split(' ')
day = int(day)
year = int(year)
month = self.MONTHS.index(month.rstrip('.')) + 1
return datetime.date(year, month, day)
def get_transactions(self, type='consommable'):
class get_history(Transaction.TransactionsElement):
head_xpath = '//table[@id="solde_%s_lignes"]//thead//tr/th/text()' % type
item_xpath = '//table[@id="solde_%s_lignes"]//tbody/tr' % type
col_date = u"Date de valeur"
col_raw = u"Motif"
class item(Transaction.TransactionElement):
obj_amount = Transaction.Amount('./td[last()]')
obj_date = AccountsPage.MyDate(CleanText(TableCell('date')))
return get_history(self)()
# -*- coding: utf-8 -*-
# Copyright(C) 2013 Romain Bignon
# Copyright(C) 2017 Vincent Ardisson
#
# This file is part of weboob.
#
......@@ -17,8 +17,10 @@
# You should have received a copy of the GNU Affero General Public License
# along with weboob. If not, see <http://www.gnu.org/licenses/>.
from __future__ import unicode_literals
from .module import VoyagesSNCFModule
from .module import AsanaModule
__all__ = ['VoyagesSNCFModule']
__all__ = ['AsanaModule']
# -*- coding: utf-8 -*-
# Copyright(C) 2017 Vincent Ardisson
#
# This file is part of weboob.
#
# weboob is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# weboob is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with weboob. If not, see <http://www.gnu.org/licenses/>.
from __future__ import unicode_literals
import time
from weboob.browser.browsers import APIBrowser
from weboob.browser.exceptions import ClientError
from weboob.capabilities.base import NotAvailable
from weboob.capabilities.bugtracker import User, Project, Issue, Status, Update
from weboob.exceptions import BrowserIncorrectPassword
from dateutil.parser import parse as parse_date
class AsanaBrowser(APIBrowser):
BASEURL = 'https://app.asana.com/api/1.0/'
STATUS_OPEN = Status(0, 'Open', Status.VALUE_NEW)
STATUS_CLOSED = Status(1, 'Closed', Status.VALUE_RESOLVED)
def __init__(self, token, *args, **kwargs):
super(AsanaBrowser, self).__init__(*args, **kwargs)
self.token = token
self.session.headers['Authorization'] = 'Bearer %s' % token
def open(self, *args, **kwargs):
try:
return super(AsanaBrowser, self).open(*args, **kwargs)
except ClientError as e:
if e.response.status_code == 401:
raise BrowserIncorrectPassword()
elif e.response.status_code == 429:
self.logger.warning('reached requests quota...')
waiting = int(e.response.headers['Retry-After'])
if waiting <= 60:
self.logger.warning('waiting %s seconds', waiting)
time.sleep(waiting)
return super(AsanaBrowser, self).open(*args, **kwargs)
else:
self.logger.warning('not waiting %s seconds, just fuck it', waiting)
raise
def _make_user(self, data):
u = User(data['id'], None)
if 'name' in data:
u.name = data['name']
return u
def _make_project(self, data):
p = Project(str(data['id']), data['name'])
p.url = 'https://app.asana.com/0/%s' % p.id
if 'members' in data:
p.members = [self._make_user(u) for u in data['members']]
p.statuses = [self.STATUS_OPEN, self.STATUS_CLOSED]
p._workspace = data['workspace']['id']
# these fields don't exist in asana
p.priorities = []
p.versions = []
return p
def _make_issue(self, data):
if data['name'].endswith(':'):
# section, not task
return None
i = Issue(str(data['id']))
i.url = 'https://app.asana.com/0/0/%s/f' % i.id
i.title = data['name']
if 'notes' in data:
i.body = data['notes']
if data.get('assignee'):
i.assignee = self._make_user(data['assignee'])
if data.get('created_by'):
# created_by is not documented
i.author = self._make_user(data['created_by'])
if 'created_at' in data:
i.creation = parse_date(data['created_at'])
if 'modified_at' in data:
i.updated = parse_date(data['modified_at'])
if 'due_at' in data:
if data['due_at']:
i.due = parse_date(data['due_at'])
else:
i.due = NotAvailable
if 'due_on' in data:
if data['due_on']:
i.due = parse_date(data['due_on'])
else:
i.due = NotAvailable
if data.get('projects'):
i.project = self._make_project(data['projects'][0])
if 'completed' in data:
i.status = self.STATUS_CLOSED if data['completed'] else self.STATUS_OPEN
if 'custom_fields' in data:
def get(d):
for k in ('string_value', 'number_value', 'enum_value', 'text_value'):
if k in d:
return d[k]
assert False, 'custom type not handled'
i.fields = {d['name']: get(d) for d in data['custom_fields']}
if 'tags' in data:
i.tags = [d['name'] for d in data['tags']]
if data.get('memberships') and data['memberships'][0]['section']:
i.category = data['memberships'][0]['section']['name']
i.version = NotAvailable
i.priority = NotAvailable
return i
def _make_update(self, data):
u = Update(str(data['id']))
if 'created_at' in data:
u.date = parse_date(data['created_at'])
u.message = '%s: %s' % (data['type'], data['text'])
if 'created_by' in data:
u.author = self._make_user(data['created_by'])
return u
def paginate(self, *args, **kwargs):
params = kwargs.setdefault('params', {})
params['limit'] = 20
reply = self.request(*args, **kwargs)
for d in reply['data']:
yield d
while reply.get('next_page') and reply['next_page'].get('uri'):
reply = self.request(reply['next_page']['uri'])
for d in reply['data']:
yield d
# -*- coding: utf-8 -*-
# Copyright(C) 2017 Vincent Ardisson
#
# This file is part of weboob.
#
# weboob is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# weboob is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with weboob. If not, see <http://www.gnu.org/licenses/>.
from __future__ import unicode_literals
from weboob.capabilities.base import empty
from weboob.capabilities.bugtracker import CapBugTracker, Project, Issue, User
from weboob.tools.backend import Module, BackendConfig
from weboob.tools.value import ValueBackendPassword
from .browser import AsanaBrowser
__all__ = ['AsanaModule']
class AsanaModule(Module, CapBugTracker):
NAME = 'asana'
DESCRIPTION = 'Asana'
MAINTAINER = 'Vincent A'
EMAIL = 'dev@indigo.re'
LICENSE = 'AGPLv3+'
VERSION = '1.4'
CONFIG = BackendConfig(ValueBackendPassword('token', label='Personal access token'))
BROWSER = AsanaBrowser
def create_default_browser(self):
return self.create_browser(self.config['token'].get())
## read-only issues and projects
def iter_issues(self, query):
query = query.copy()
params = {}
if query.title:
params['text'] = query.title
if query.project:
if not isinstance(query.project, Project):
query.project = next(p for p in self.iter_projects() if query.project.lower() in p.name.lower())
params['project'] = query.project.id
if query.tags:
params['tags.all'] = ','.join(query.project._tagdict[tag] for tag in query.tags)
if query.assignee:
if isinstance(query.assignee, User):
params['assignee'] = query.assignee.id
else:
params['assignee'] = query.assignee
if query.author:
if isinstance(query.author, User):
params['created_by'] = query.author.id
else:
params['created_by'] = query.author
if query.status:
if query.status.lower() == 'closed':
params['completed'] = 'true'
else:
params['completed'] = 'false'
params['completed_since'] = 'now' # completed=false is not enough...
if not query.project:
workspaces = list(self._iter_workspaces())
if len(workspaces) == 1:
params['workspace'] = workspaces[0]
if query.project and query.assignee:
# asana's wtf api doesn't allow more than 1 filter...
del params['project']
params['workspace'] = query.project._workspace
opt = '?opt_fields=%s' % ','.join([
'name', 'completed', 'due_at', 'due_on', 'created_at', 'modified_at',
'notes', 'custom_fields', 'tags.name', 'assignee.name', 'created_by.name',
'projects.name', 'projects.workspace',
])
if params:
data = self.browser.paginate('tasks%s' % opt, params=params)
else:
data = []
for issue in data:
issue = self.browser._make_issue(issue)
if issue is None: # section
continue
# post-filter because many server-side filters don't always work...
if query.title and query.title.lower() not in issue.title.lower():
self.logger.debug('"title" filter failed on issue %r', issue)
continue
if query.status and query.status.lower() != issue.status.name.lower():
self.logger.debug('"completed" filter failed on issue %r', issue)
continue
if query.tags and not (set(query.tags) <= set(issue.tags)):
self.logger.debug('"tags" filter failed on issue %r', issue)
continue
if query.author:
if isinstance(query.author, User):
if query.author.id != issue.author.id:
continue
else:
if query.author.lower() != issue.author.name.lower():
continue
yield issue
def _set_stories(self, issue):
ds = self.browser.request('tasks/%s/stories' % issue.id)['data']
issue.history = [self.browser._make_update(d) for d in ds]
def get_issue(self, id):
if not id.isdigit():
return
data = self.browser.request('tasks/%s' % id)['data']
return self.browser._make_issue(data)
def iter_projects(self):
for w in self._iter_workspaces():
tags = self.browser.request('tags?workspace=%s' % w)['data']
data = self.browser.paginate('projects?opt_fields=name,members.name,workspace&workspace=%s' % w)
for p in data:
project = self.browser._make_project(p)
self._assign_tags(tags, project)
yield project
def _assign_tags(self, data, project):
project._tagdict = {d['name']: str(d['id']) for d in data}
project.tags = list(project._tagdict)
def get_project(self, id):
if not id.isdigit():
return
data = self.browser.request('projects/%s' % id)['data']
return self.browser._make_project(data)
def _iter_workspaces(self):
return (d['id'] for d in self.browser.paginate('workspaces'))
## writing issues
def create_issue(self, project):
issue = Issue(0)
issue._project = project
return issue
def post_issue(self, issue):
data = {}
if issue.title:
data['name'] = issue.title
if issue.body:
data['notes'] = issue.body
if issue.due:
data['due_at'] = issue.due.strftime('%Y-%m-%d')
if issue.assignee:
data['assignee'] = issue.assignee.id
if issue.id and issue.id != '0':
data['projects'] = issue._project
self.browser.request('tasks', data=data)
if issue.tags:
self._set_tag_list(issue, True)
else:
self.browser.request('tasks/%s' % issue.id, data=data, method='PUT')
if not empty(issue.tags):
self._set_tag_list(issue)
def _set_tag_list(self, issue, add=False):
to_remove = set()
to_add = set(issue.tags)
if not add:
existing = set(self.get_issue(issue.id).tags)
to_add = to_add - existing
to_remove = existing - to_add
for old in to_remove:
tagid = issue.project._tagdict[old]
self.browser.request('tasks/%s/removeTag', data={'tag': tagid})
for new in to_add:
tagid = issue.project._tagdict[new]
self.browser.request('tasks/%s/addTag', data={'tag': tagid})
def update_issue(self, issue, update):
assert not update.changes, 'changes are not supported yet'
assert update.message
self.browser.request('tasks/%s/stories' % issue.id, data={'text': update.message})
def remove_issue(self, issue):
self.browser.request('tasks/%s' % issue.id, method='DELETE')
## filling
def fill_project(self, project, fields):
if set(['members']) & set(fields):
return self.get_project(project.id)
def fill_issue(self, issue, fields):
if set(['body', 'assignee', 'due', 'creation', 'updated', 'project']) & set(fields):
new = self.get_issue(issue.id)
for f in fields:
if getattr(new, f):
setattr(issue, f, getattr(new, f))
if 'history' in fields:
self._set_stories(issue)
OBJECTS = {
Project: fill_project,
Issue: fill_issue,
}
# -*- coding: utf-8 -*-
# Copyright(C) 2017 Vincent Ardisson
#
# This file is part of weboob.
#
# weboob is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# weboob is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with weboob. If not, see <http://www.gnu.org/licenses/>.
from __future__ import unicode_literals
from weboob.capabilities.base import empty
from weboob.capabilities.bugtracker import Query
from weboob.tools.test import BackendTest
class AsanaTest(BackendTest):
MODULE = 'asana'
def test_iter_projects(self):
projects = list(self.backend.iter_projects())
self.assertTrue(projects)
self.assertTrue(projects[0].statuses)
self.assertTrue(projects[0].members)
def test_find_any_issues(self):
projects = list(self.backend.iter_projects())
self.assertTrue(projects)
q = Query()
q.project = projects[0]
issues = [issue for issue, _ in zip(self.backend.iter_issues(q), range(30))]
self.assertTrue(issues)
for issue in issues:
self.assertTrue(issue.project)
self.assertEquals(issue.project.id, projects[0].id)
self.assertTrue(issue.title)
self.assertFalse(empty(issue.body))
self.assertTrue(issue.creation)
self.assertTrue(issue.author, issue.title)
def _test_find_by_criterion(self, attr, first_cb=None, matcher_cb=None):
if matcher_cb is None:
def matcher_cb(expected, actual):
self.assertEquals(expected.id, actual.id, 'different id on: %s != %s' % (expected, actual))
if first_cb is None:
def first_cb(obj):
return bool(obj)
projects = list(self.backend.iter_projects())
self.assertTrue(projects)
q = Query()
q.project = projects[0]
for issue, _ in zip(self.backend.iter_issues(q), range(30)):
criterion_obj = getattr(issue, attr)
if first_cb(criterion_obj):
break
else:
assert False, 'not a single issue has this criterion'
setattr(q, attr, criterion_obj)
some = False
for issue, _ in zip(self.backend.iter_issues(q), range(30)):
some = True
fetched_obj = getattr(issue, attr)
matcher_cb(criterion_obj, fetched_obj)
assert some, 'the issue searched for was not found'
def test_find_by_assignee(self):
self._test_find_by_criterion('assignee')
def test_find_by_author(self):
self._test_find_by_criterion('author')