pax_global_header 0000666 0000000 0000000 00000000064 13414137563 0014521 g ustar 00root root 0000000 0000000 52 comment=130005535d1d257a835857d4e42b2541ce490b40
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/ 0000775 0000000 0000000 00000000000 13414137563 0022710 5 ustar 00root root 0000000 0000000 woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/ 0000775 0000000 0000000 00000000000 13414137563 0024165 5 ustar 00root root 0000000 0000000 woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/ 0000775 0000000 0000000 00000000000 13414137563 0026616 5 ustar 00root root 0000000 0000000 woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/__init__.py 0000664 0000000 0000000 00000000270 13414137563 0030726 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
from .base import UserError, NotLoaded, NotAvailable, BaseObject, Capability
__all__ = ['UserError', 'NotLoaded', 'NotAvailable', 'BaseObject', 'Capability']
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/account.py 0000664 0000000 0000000 00000006445 13414137563 0030635 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2011 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 .
from .base import Capability, BaseObject, StringField, IntField, Field, UserError
__all__ = ['AccountRegisterError', 'Account', 'StatusField', 'CapAccount']
class AccountRegisterError(UserError):
"""
Raised when there is an error during registration.
"""
class Account(BaseObject):
"""
Describe an account and its properties.
"""
login = StringField('Login')
password = StringField('Password')
properties = Field('List of key/value properties', dict)
def __init__(self, id=None, url=None):
super(Account, self).__init__(id, url)
class StatusField(BaseObject):
"""
Field of an account staeobjectus.
"""
FIELD_TEXT = 0x001 # the value is a long text
FIELD_HTML = 0x002 # the value is HTML formated
key = StringField('Key')
label = StringField('Label')
value = StringField('Value')
flags = IntField('Flags')
def __init__(self, key, label, value, flags=0, url=None):
super(StatusField, self).__init__(key, url)
self.key = key
self.label = label
self.value = value
self.flags = flags
class CapAccount(Capability):
"""
Capability for websites when you can create and manage accounts.
:var ACCOUNT_REGISTER_PROPERTIES: This class constant may be a list of
:class:`weboob.tools.value.Value` objects.
If the value remains None, weboob considers
that :func:`register_account` isn't supported.
"""
ACCOUNT_REGISTER_PROPERTIES = None
@staticmethod
def register_account(account):
"""
Register an account on website
This is a static method, it would be called even if the backend is
instancied.
:param account: describe the account to create
:type account: :class:`Account`
:raises: :class:`AccountRegisterError`
"""
raise NotImplementedError()
def confirm_account(self, mail):
"""
From an email go to the confirm link.
"""
raise NotImplementedError()
def get_account(self):
"""
Get the current account.
"""
raise NotImplementedError()
def update_account(self, account):
"""
Update the current account.
"""
raise NotImplementedError()
def get_account_status(self):
"""
Get status of the current account.
:returns: a list of fields
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/audio.py 0000664 0000000 0000000 00000011072 13414137563 0030272 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2013 Pierre Mazière
#
# 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 .
import re
from weboob.tools.compat import unicode
from .image import Thumbnail
from .base import Field, StringField, IntField, BaseObject
from .date import DeltaField
from .file import CapFile, BaseFile
__all__ = ['BaseAudio', 'CapAudio']
def decode_id(decode_id):
def wrapper(func):
def inner(self, *args, **kwargs):
arg = unicode(args[0])
_id = decode_id(arg)
if _id is None:
return None
new_args = [_id]
new_args.extend(args[1:])
return func(self, *new_args, **kwargs)
return inner
return wrapper
class Album(BaseObject):
"""
Represent an album
"""
title = StringField('album name')
author = StringField('artist name')
year = IntField('release year')
thumbnail = Field('Image associated to the album', Thumbnail)
tracks_list = Field('list of tracks', list)
@classmethod
def decode_id(cls, _id):
if _id:
m = re.match('^(album)\.(.*)', _id)
if m:
return m.group(2)
return _id
class Playlist(BaseObject):
"""
Represent a playlist
"""
title = StringField('playlist name')
tracks_list = Field('list of tracks', list)
@classmethod
def decode_id(cls, _id):
if _id:
m = re.match('^(playlist)\.(.*)', _id)
if m:
return m.group(2)
return _id
class BaseAudio(BaseFile):
"""
Represent an audio file
"""
duration = DeltaField('file duration')
bitrate = IntField('file bit rate in Kbps')
format = StringField('file format')
thumbnail = Field('Image associated to the file', Thumbnail)
@classmethod
def decode_id(cls, _id):
if _id:
m = re.match('^(audio)\.(.*)', _id)
if m:
return m.group(2)
return _id
class CapAudio(CapFile):
"""
Audio file provider
"""
@classmethod
def get_object_method(cls, _id):
m = re.match('^(\w+)\.(.*)', _id)
if m:
if m.group(1) == 'album':
return 'get_album'
elif m.group(1) == 'playlist':
return 'get_playlist'
else:
return 'get_audio'
def search_audio(self, pattern, sortby=CapFile.SEARCH_RELEVANCE):
"""
search for a audio file
:param pattern: pattern to search on
:type pattern: str
:param sortby: sort by ...(use SEARCH_* constants)
:rtype: iter[:class:`BaseAudio`]
"""
return self.search_file(pattern, sortby)
def search_album(self, pattern, sortby=CapFile.SEARCH_RELEVANCE):
"""
search for an album
:param pattern: pattern to search on
:type pattern: str
:rtype: iter[:class:`Album`]
"""
raise NotImplementedError()
def search_playlist(self, pattern, sortby=CapFile.SEARCH_RELEVANCE):
"""
search for an album
:param pattern: pattern to search on
:type pattern: str
:rtype: iter[:class:`Playlist`]
"""
raise NotImplementedError()
@decode_id(BaseAudio.decode_id)
def get_audio(self, _id):
"""
Get an audio file from an ID.
:param id: audio file ID
:type id: str
:rtype: :class:`BaseAudio`]
"""
return self.get_file(_id)
@decode_id(Playlist.decode_id)
def get_playlist(self, _id):
"""
Get a playlist from an ID.
:param id: playlist ID
:type id: str
:rtype: :class:`Playlist`]
"""
raise NotImplementedError()
@decode_id(Album.decode_id)
def get_album(self, _id):
"""
Get an album from an ID.
:param id: album ID
:type id: str
:rtype: :class:`Album`]
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/audiostream.py0000664 0000000 0000000 00000003552 13414137563 0031512 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2013 Pierre Mazière
#
# 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 .
from weboob.tools.capabilities.streaminfo import StreamInfo
from .base import Field
from .file import CapFile
from .audio import CapAudio, BaseAudio
__all__ = ['BaseAudioStream', 'CapAudioStream']
class BaseAudioStream(BaseAudio):
"""
Audio stream object
"""
current = Field('Information related to current broadcast', StreamInfo)
def __unicode__(self):
return u'%s (%s)' % (self.title, self.url)
def __repr__(self):
return '%r (%r)' % (self.title, self.url)
class CapAudioStream(CapAudio):
"""
Audio streams provider
"""
def search_audiostreams(self, pattern, sortby=CapFile.SEARCH_RELEVANCE):
"""
Search an audio stream
:param pattern: pattern to search
:type pattern: str
:param sortby: sort by ... (use SEARCH_* constants)
:rtype: iter[:class:`BaseAudioStream`]
"""
return self.search_audio(pattern, sortby)
def get_audiostream(self, _id):
"""
Get an audio stream
:param _id: Audiostream ID
:type id: str
:rtype: :class:`BaseAudioStream`
"""
return self.get_audio(_id)
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/bank.py 0000664 0000000 0000000 00000062030 13414137563 0030104 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2016 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 .
from datetime import date, datetime
from binascii import crc32
import re
from unidecode import unidecode
from weboob.capabilities.base import empty, find_object
from weboob.exceptions import BrowserQuestion
from weboob.tools.capabilities.bank.iban import is_iban_valid
from weboob.tools.compat import unicode
from .base import BaseObject, Field, StringField, DecimalField, IntField, \
UserError, Currency, NotAvailable, EnumField, Enum
from .date import DateField
from .collection import CapCollection
__all__ = [
'CapBank', 'BaseAccount', 'Account', 'Loan', 'Transaction', 'AccountNotFound',
'AccountType',
'CapBankWealth', 'Investment', 'CapBankPockets', 'Pocket',
'CapBankTransfer', 'Transfer', 'Recipient',
'TransferError', 'TransferBankError', 'TransferInvalidAmount', 'TransferInsufficientFunds',
'TransferInvalidCurrency', 'TransferInvalidLabel',
'TransferInvalidEmitter', 'TransferInvalidRecipient',
'TransferStep',
'CapBankTransferAddRecipient',
'RecipientNotFound', 'AddRecipientError', 'AddRecipientBankError', 'AddRecipientTimeout',
'AddRecipientStep', 'RecipientInvalidIban', 'RecipientInvalidLabel',
'Rate', 'CapCurrencyRate',
]
class AccountNotFound(UserError):
"""
Raised when an account is not found.
"""
def __init__(self, msg='Account not found'):
super(AccountNotFound, self).__init__(msg)
class RecipientNotFound(UserError):
"""
Raised when a recipient is not found.
"""
def __init__(self, msg='Recipient not found'):
super(RecipientNotFound, self).__init__(msg)
class TransferError(UserError):
"""
A transfer has failed.
"""
code = 'transferError'
def __init__(self, description=None, message=None):
"""
:param message: error message from the bank, if any
"""
super(TransferError, self).__init__(message or description)
self.message = message
self.description = description
class TransferBankError(TransferError):
"""The transfer was rejected by the bank with a message."""
code = 'bankMessage'
class TransferInvalidLabel(TransferError):
"""The transfer label is invalid."""
code = 'invalidLabel'
class TransferInvalidEmitter(TransferError):
"""The emitter account cannot be used for transfers."""
code = 'invalidEmitter'
class TransferInvalidRecipient(TransferError):
"""The emitter cannot transfer to this recipient."""
code = 'invalidRecipient'
class TransferInvalidAmount(TransferError):
"""This amount is not allowed."""
code = 'invalidAmount'
class TransferInvalidCurrency(TransferInvalidAmount):
"""The transfer currency is invalid."""
code = 'invalidCurrency'
class TransferInsufficientFunds(TransferInvalidAmount):
"""Not enough funds on emitter account."""
code = 'insufficientFunds'
class TransferInvalidDate(TransferError):
"""This execution date cannot be used."""
code = 'invalidDate'
class AddRecipientError(UserError):
"""
Failed trying to add a recipient.
"""
code = 'AddRecipientError'
def __init__(self, description=None, message=None):
"""
:param message: error message from the bank, if any
"""
super(AddRecipientError, self).__init__(message or description)
self.message = message
self.description = description
class AddRecipientBankError(AddRecipientError):
"""The new recipient was rejected by the bank with a message."""
code = 'bankMessage'
class AddRecipientTimeout(AddRecipientError):
"""Add new recipient request has timeout"""
code = 'timeout'
class RecipientInvalidIban(AddRecipientError):
code = 'invalidIban'
class RecipientInvalidLabel(AddRecipientError):
code = 'invalidLabel'
class BaseAccount(BaseObject, Currency):
"""
Generic class aiming to be parent of :class:`Recipient` and
:class:`Account`.
"""
label = StringField('Pretty label')
currency = StringField('Currency', default=None)
bank_name = StringField('Bank Name', mandatory=False)
def __init__(self, id='0', url=None):
super(BaseAccount, self).__init__(id, url)
@property
def currency_text(self):
return Currency.currency2txt(self.currency)
@property
def ban(self):
""" Bank Account Number part of IBAN"""
if not self.iban:
return NotAvailable
return self.iban[4:]
class Recipient(BaseAccount):
"""
Recipient of a transfer.
"""
enabled_at = DateField('Date of availability')
category = StringField('Recipient category')
iban = StringField('International Bank Account Number')
# Needed for multispaces case
origin_account_id = StringField('Account id which recipient belong to')
class AccountType(Enum):
UNKNOWN = 0
CHECKING = 1
"Transaction, everyday transactions"
SAVINGS = 2
"Savings/Deposit, can be used for every banking"
DEPOSIT = 3
"Term of Fixed Deposit, has time/amount constraints"
LOAN = 4
"Loan account"
MARKET = 5
"Stock market or other variable investments"
JOINT = 6
"Joint account"
CARD = 7
"Card account"
LIFE_INSURANCE = 8
"Life insurances"
PEE = 9
"Employee savings PEE"
PERCO = 10
"Employee savings PERCO"
ARTICLE_83 = 11
"Article 83"
RSP = 12
"Employee savings RSP"
PEA = 13
"Share savings"
CAPITALISATION = 14
"Life Insurance capitalisation"
PERP = 15
"Retirement savings"
MADELIN = 16
"Complementary retirement savings"
MORTGAGE = 17
"Mortgage"
CONSUMER_CREDIT = 18
"Consumer credit"
REVOLVING_CREDIT = 19
"Revolving credit"
class Account(BaseAccount):
"""
Bank account.
"""
TYPE_UNKNOWN = AccountType.UNKNOWN
TYPE_CHECKING = AccountType.CHECKING
TYPE_SAVINGS = AccountType.SAVINGS
TYPE_DEPOSIT = AccountType.DEPOSIT
TYPE_LOAN = AccountType.LOAN
TYPE_MARKET = AccountType.MARKET
TYPE_JOINT = AccountType.JOINT
TYPE_CARD = AccountType.CARD
TYPE_LIFE_INSURANCE = AccountType.LIFE_INSURANCE
TYPE_PEE = AccountType.PEE
TYPE_PERCO = AccountType.PERCO
TYPE_ARTICLE_83 = AccountType.ARTICLE_83
TYPE_RSP = AccountType.RSP
TYPE_PEA = AccountType.PEA
TYPE_CAPITALISATION = AccountType.CAPITALISATION
TYPE_PERP = AccountType.PERP
TYPE_MADELIN = AccountType.MADELIN
TYPE_MORTGAGE = AccountType.MORTGAGE
TYPE_CONSUMER_CREDIT = AccountType.CONSUMER_CREDIT
TYPE_REVOLVING_CREDIT = AccountType.REVOLVING_CREDIT
type = EnumField('Type of account', AccountType, default=TYPE_UNKNOWN)
balance = DecimalField('Balance on this bank account')
coming = DecimalField('Sum of coming movements')
iban = StringField('International Bank Account Number', mandatory=False)
# card attributes
paydate = DateField('For credit cards. When next payment is due.')
paymin = DecimalField('For credit cards. Minimal payment due.')
cardlimit = DecimalField('For credit cards. Credit limit.')
number = StringField('Shown by the bank to identify your account ie XXXXX7489')
# market and lifeinssurance accounts
valuation_diff = DecimalField('+/- values total')
valuation_diff_percent = DecimalField('+/- values ratio')
# parent account
# - A checking account parent of a card account
# - A checking account parent of a recurring loan account
# - An investment account parent of a liquidity account
# - ...
parent = Field('Parent account', BaseAccount)
opening_date = DateField('Date when the account contract was created on the bank')
def __repr__(self):
return "<%s id=%r label=%r>" % (type(self).__name__, self.id, self.label)
class Loan(Account):
"""
Account type dedicated to loans and credits.
"""
name = StringField('Person name')
account_label = StringField('Label of the debited account')
insurance_label = StringField('Label of the insurance')
total_amount = DecimalField('Total amount loaned')
available_amount = DecimalField('Amount available') # only makes sense for revolving credit
used_amount = DecimalField('Amount already used') # only makes sense for revolving credit
subscription_date = DateField('Date of subscription of the loan')
maturity_date = DateField('Estimated end date of the loan')
duration = IntField('Duration of the loan given in months')
rate = DecimalField('Monthly rate of the loan')
nb_payments_left = IntField('Number of payments still due')
nb_payments_done = IntField('Number of payments already done')
nb_payments_total = IntField('Number total of payments')
last_payment_amount = DecimalField('Amount of the last payment done')
last_payment_date = DateField('Date of the last payment done')
next_payment_amount = DecimalField('Amount of next payment')
next_payment_date = DateField('Date of the next payment')
class TransactionType(Enum):
UNKNOWN = 0
TRANSFER = 1
ORDER = 2
CHECK = 3
DEPOSIT = 4
PAYBACK = 5
WITHDRAWAL = 6
CARD = 7
LOAN_PAYMENT = 8
BANK = 9
CASH_DEPOSIT = 10
CARD_SUMMARY = 11
DEFERRED_CARD = 12
class Transaction(BaseObject):
"""
Bank transaction.
"""
TYPE_UNKNOWN = TransactionType.UNKNOWN
TYPE_TRANSFER = TransactionType.TRANSFER
TYPE_ORDER = TransactionType.ORDER
TYPE_CHECK = TransactionType.CHECK
TYPE_DEPOSIT = TransactionType.DEPOSIT
TYPE_PAYBACK = TransactionType.PAYBACK
TYPE_WITHDRAWAL = TransactionType.WITHDRAWAL
TYPE_CARD = TransactionType.CARD
TYPE_LOAN_PAYMENT = TransactionType.LOAN_PAYMENT
TYPE_BANK = TransactionType.BANK
TYPE_CASH_DEPOSIT = TransactionType.CASH_DEPOSIT
TYPE_CARD_SUMMARY = TransactionType.CARD_SUMMARY
TYPE_DEFERRED_CARD = TransactionType.DEFERRED_CARD
date = DateField('Debit date on the bank statement')
rdate = DateField('Real date, when the payment has been made; usually extracted from the label or from credit card info')
vdate = DateField('Value date, or accounting date; usually for professional accounts')
type = EnumField('Type of transaction, use TYPE_* constants', TransactionType, default=TYPE_UNKNOWN)
raw = StringField('Raw label of the transaction')
category = StringField('Category of the transaction')
label = StringField('Pretty label')
amount = DecimalField('Amount of the transaction')
card = StringField('Card number (if any)')
commission = DecimalField('Commission part on the transaction (in account currency)')
# International
original_amount = DecimalField('Original amount (in another currency)')
original_currency = StringField('Currency of the original amount')
country = StringField('Country of transaction')
# Financial arbitrations
investments = Field('List of investments related to the transaction', list, default=[])
def __repr__(self):
return "" % (self.date, self.label, self.amount)
def unique_id(self, seen=None, account_id=None):
"""
Get an unique ID for the transaction based on date, amount and raw.
:param seen: if given, the method uses this dictionary as a cache to
prevent several transactions with the same values to have the same
unique ID.
:type seen: :class:`dict`
:param account_id: if given, add the account ID in data used to create
the unique ID. Can be useful if you want your ID to be unique across
several accounts.
:type account_id: :class:`str`
:returns: an unique ID encoded in 8 length hexadecimal string (for example ``'a64e1bc9'``)
:rtype: :class:`str`
"""
crc = crc32(unicode(self.date).encode('utf-8'))
crc = crc32(unicode(self.amount).encode('utf-8'), crc)
if not empty(self.raw):
label = self.raw
else:
label = self.label
crc = crc32(re.sub('[ ]+', ' ', label).encode("utf-8"), crc)
if account_id is not None:
crc = crc32(unicode(account_id).encode('utf-8'), crc)
if seen is not None:
while crc in seen:
crc = crc32(b"*", crc)
seen.add(crc)
return "%08x" % (crc & 0xffffffff)
class Investment(BaseObject):
"""
Investment in a financial market.
"""
CODE_TYPE_ISIN = u'ISIN'
CODE_TYPE_AMF = u'AMF'
label = StringField('Label of stocks')
code = StringField('Identifier of the stock')
code_type = StringField('Type of stock code (ISIN or AMF)')
description = StringField('Short description of the stock')
quantity = DecimalField('Quantity of stocks')
unitprice = DecimalField('Buy price of one stock')
unitvalue = DecimalField('Current value of one stock')
valuation = DecimalField('Total current valuation of the Investment')
vdate = DateField('Value date of the valuation amount')
diff = DecimalField('Difference between the buy cost and the current valuation')
diff_percent = DecimalField('Difference in ratio (1 meaning 100%) between the buy cost and the current valuation')
portfolio_share = DecimalField('Ratio (1 meaning 100%) of the current amount relative to the total')
# International
original_currency = StringField('Currency of the original amount')
original_valuation = DecimalField('Original valuation (in another currency)')
original_unitvalue = DecimalField('Original unitvalue (in another currency)')
original_unitprice = DecimalField('Original unitprice (in another currency)')
original_diff = DecimalField('Original diff (in another currency)')
def __repr__(self):
return '' % (self.label, self.code, self.valuation)
class PocketCondition(Enum):
UNKNOWN = 0
DATE = 1
AVAILABLE = 2
RETIREMENT = 3
WEDDING = 4
DEATH = 5
INDEBTEDNESS = 6
DIVORCE = 7
DISABILITY = 8
BUSINESS_CREATION = 9
BREACH_EMPLOYMENT_CONTRACT = 10
UNLOCKING_EXCEPTIONAL = 11
THIRD_CHILD = 12
EXPIRATION_UNEMPLOYMENT = 13
PURCHASE_APARTMENT = 14
class Pocket(BaseObject):
"""
Pocket
"""
CONDITION_UNKNOWN = PocketCondition.UNKNOWN
CONDITION_DATE = PocketCondition.DATE
CONDITION_AVAILABLE = PocketCondition.AVAILABLE
CONDITION_RETIREMENT = PocketCondition.RETIREMENT
CONDITION_WEDDING = PocketCondition.WEDDING
CONDITION_DEATH = PocketCondition.DEATH
CONDITION_INDEBTEDNESS = PocketCondition.INDEBTEDNESS
CONDITION_DIVORCE = PocketCondition.DIVORCE
CONDITION_DISABILITY = PocketCondition.DISABILITY
CONDITION_BUSINESS_CREATION = PocketCondition.BUSINESS_CREATION
CONDITION_BREACH_EMPLOYMENT_CONTRACT = PocketCondition.BREACH_EMPLOYMENT_CONTRACT
CONDITION_UNLOCKING_EXCEPTIONAL = PocketCondition.UNLOCKING_EXCEPTIONAL
CONDITION_THIRD_CHILD = PocketCondition.THIRD_CHILD
CONDITION_EXPIRATION_UNEMPLOYMENT = PocketCondition.EXPIRATION_UNEMPLOYMENT
CONDITION_PURCHASE_APARTMENT = PocketCondition.PURCHASE_APARTMENT
label = StringField('Label of pocket')
amount = DecimalField('Amount of the pocket')
quantity = DecimalField('Quantity of stocks')
availability_date = DateField('Availability date of the pocket')
condition = EnumField('Withdrawal condition of the pocket', PocketCondition, default=CONDITION_UNKNOWN)
investment = Field('Reference to the investment of the pocket', Investment)
class TransferStep(BrowserQuestion):
def __init__(self, transfer, *values):
super(TransferStep, self).__init__(*values)
self.transfer = transfer
class AddRecipientStep(BrowserQuestion):
def __init__(self, recipient, *values):
super(AddRecipientStep, self).__init__(*values)
self.recipient = recipient
class Transfer(BaseObject, Currency):
"""
Transfer from an account to a recipient.
"""
amount = DecimalField('Amount to transfer')
currency = StringField('Currency', default=None)
fees = DecimalField('Fees', default=None)
exec_date = Field('Date of transfer', date, datetime)
account_id = StringField('ID of origin account')
account_iban = StringField('International Bank Account Number')
account_label = StringField('Label of origin account')
account_balance = DecimalField('Balance of origin account before transfer')
recipient_id = StringField('ID of recipient account')
recipient_iban = StringField('International Bank Account Number')
recipient_label = StringField('Label of recipient account')
label = StringField('Reason')
class CapBank(CapCollection):
"""
Capability of bank websites to see accounts and transactions.
"""
def iter_resources(self, objs, split_path):
"""
Iter resources.
Default implementation of this method is to return on top-level
all accounts (by calling :func:`iter_accounts`).
:param objs: type of objects to get
:type objs: tuple[:class:`BaseObject`]
:param split_path: path to discover
:type split_path: :class:`list`
:rtype: iter[:class:`BaseObject`]
"""
if Account in objs:
self._restrict_level(split_path)
return self.iter_accounts()
def iter_accounts(self):
"""
Iter accounts.
:rtype: iter[:class:`Account`]
"""
raise NotImplementedError()
def get_account(self, id):
"""
Get an account from its ID.
:param id: ID of the account
:type id: :class:`str`
:rtype: :class:`Account`
:raises: :class:`AccountNotFound`
"""
return find_object(self.iter_accounts(), id=id, error=AccountNotFound)
def iter_history(self, account):
"""
Iter history of transactions on a specific account.
:param account: account to get history
:type account: :class:`Account`
:rtype: iter[:class:`Transaction`]
:raises: :class:`AccountNotFound`
"""
raise NotImplementedError()
def iter_coming(self, account):
"""
Iter coming transactions on a specific account.
:param account: account to get coming transactions
:type account: :class:`Account`
:rtype: iter[:class:`Transaction`]
:raises: :class:`AccountNotFound`
"""
raise NotImplementedError()
class CapCgp(CapBank):
"""
Capability of cgp website to see accounts and transactions.
"""
class CapBankWealth(CapBank):
"""
Capability of bank websites to see investment.
"""
def iter_investment(self, account):
"""
Iter investment of a market account
:param account: account to get investments
:type account: :class:`Account`
:rtype: iter[:class:`Investment`]
:raises: :class:`AccountNotFound`
"""
raise NotImplementedError()
class CapBankPockets(CapBankWealth):
"""
Capability of bank websites to see pockets.
"""
def iter_pocket(self, account):
"""
Iter pocket
:param account: account to get pockets
:type account: :class:`Account`
:rtype: iter[:class:`Pocket`]
:raises: :class:`AccountNotFound`
"""
raise NotImplementedError()
class CapBankTransfer(CapBank):
def iter_transfer_recipients(self, account):
"""
Iter recipients availables for a transfer from a specific account.
:param account: account which initiate the transfer
:type account: :class:`Account`
:rtype: iter[:class:`Recipient`]
:raises: :class:`AccountNotFound`
"""
raise NotImplementedError()
def init_transfer(self, transfer, **params):
"""
Initiate a transfer.
:param :class:`Transfer`
:rtype: :class:`Transfer`
:raises: :class:`TransferError`
"""
raise NotImplementedError()
def execute_transfer(self, transfer, **params):
"""
Execute a transfer.
:param :class:`Transfer`
:rtype: :class:`Transfer`
:raises: :class:`TransferError`
"""
raise NotImplementedError()
def transfer(self, transfer, **params):
"""
Do a transfer from an account to a recipient.
:param :class:`Transfer`
:rtype: :class:`Transfer`
:raises: :class:`TransferError`
"""
if not transfer.amount or transfer.amount <= 0:
raise TransferInvalidAmount('amount must be strictly positive')
t = self.init_transfer(transfer, **params)
for key, value in t.iter_fields():
if hasattr(transfer, key) and key != 'id':
transfer_val = getattr(transfer, key)
try:
if hasattr(self, 'transfer_check_%s' % key):
assert getattr(self, 'transfer_check_%s' % key)(transfer_val, value)
else:
assert transfer_val == value or empty(transfer_val)
except AssertionError:
raise TransferError('%s changed during transfer processing (from %s to %s)' % (key, transfer_val, value))
return self.execute_transfer(t, **params)
def transfer_check_label(self, old, new):
old = re.sub(r'\s+', ' ', old).strip()
new = re.sub(r'\s+', ' ', new).strip()
return unidecode(old) == unidecode(new)
class CapBankTransferAddRecipient(CapBankTransfer):
def new_recipient(self, recipient, **params):
raise NotImplementedError()
def add_recipient(self, recipient, **params):
"""
Add a recipient to the connection.
:param iban: iban of the new recipient.
:type iban: :class:`str`
:param label: label of the new recipient.
:type label: :class`str`
:raises: :class:`BrowserQuestion`
:raises: :class:`AddRecipientError`
:rtype: :class:`Recipient`
"""
if not is_iban_valid(recipient.iban):
raise RecipientInvalidIban('Iban is not valid.')
if not recipient.label:
raise RecipientInvalidLabel('Recipient label is mandatory.')
return self.new_recipient(recipient, **params)
class Rate(BaseObject, Currency):
"""
Currency exchange rate.
"""
currency_from = StringField('The currency to which exchange rates are relative to. When converting 1 EUR to X HUF, currency_fom is EUR.)', default=None)
currency_to = StringField('The currency is converted to. When converting 1 EUR to X HUF, currency_to is HUF.)', default=None)
value = DecimalField('Exchange rate')
datetime = DateField('Collection date and time')
class CapCurrencyRate(CapBank):
"""
Capability of bank websites to get currency exchange rates.
"""
def iter_currencies(self):
"""
Iter available currencies.
:rtype: iter[:class:`Currency`]
"""
raise NotImplementedError()
def get_rate(self, currency_from, currency_to):
"""
Get exchange rate.
:param currency_from: currency to which exchange rate is relative to
:type currency_from: :class:`Currency`
:param currency_to: currency is converted to
:type currency_to: :class`Currency`
:rtype: :class:`Rate`
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/base.py 0000664 0000000 0000000 00000046531 13414137563 0030113 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2013 Christophe Benz, Romain Bignon, Julien Hebert
#
# 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 .
from collections import OrderedDict
import warnings
import re
from decimal import Decimal
from copy import deepcopy, copy
import sys
from weboob.tools.compat import unicode, long, with_metaclass, StrConv
from weboob.tools.misc import to_unicode
__all__ = ['UserError', 'FieldNotFound', 'NotAvailable', 'FetchError',
'NotLoaded', 'Capability', 'Field', 'IntField', 'DecimalField',
'FloatField', 'StringField', 'BytesField', 'BoolField',
'Enum', 'EnumField',
'empty', 'BaseObject']
class EnumMeta(type):
@classmethod
def __prepare__(mcs, name, bases, **kwargs):
# in python3.6, default namespace keeps declaration order
# in python>=3 but <3.6, force ordered namespace
# doesn't work in python2
return OrderedDict()
def __init__(cls, name, bases, attrs, *args, **kwargs):
super(EnumMeta, cls).__init__(name, bases, attrs, *args, **kwargs)
attrs = [(k, v) for k, v in attrs.items() if not callable(v) and not k.startswith('__')]
if sys.version_info.major < 3:
# can't have original declaration order, at least sort by value
attrs.sort(key=lambda kv: kv[1])
cls.__members__ = OrderedDict(attrs)
def __setattr__(cls, name, value):
super(EnumMeta, cls).__setattr__(name, value)
if not callable(value) and not name.startswith('__'):
cls.__members__[name] = value
def __call__(cls, *args, **kwargs):
raise ValueError("Enum type can't be instanciated")
@property
def _items(cls):
return cls.__members__.items()
@property
def _keys(cls):
return cls.__members__.keys()
@property
def _values(cls):
return cls.__members__.values()
@property
def _types(cls):
return set(map(type, cls._values))
def __iter__(cls):
return iter(cls.__members__.values())
def __len__(cls):
return len(cls.__members__)
def __contains__(cls, value):
return value in cls.__members__.values()
def __getitem__(cls, k):
return cls.__members__[k]
class Enum(with_metaclass(EnumMeta, object)):
pass
def empty(value):
"""
Checks if a value is empty (None, NotLoaded or NotAvailable).
:rtype: :class:`bool`
"""
return value is None or isinstance(value, EmptyType)
def find_object(mylist, error=None, **kwargs):
"""
Very simple tools to return an object with the matching parameters in
kwargs.
"""
for a in mylist:
for key, value in kwargs.items():
if getattr(a, key) != value:
break
else:
return a
if error is not None:
raise error()
return None
def strict_find_object(mylist, error=None, **kwargs):
"""
Tools to return an object with the matching parameters in kwargs.
Parameters with empty value are skipped
"""
kwargs = {k: v for k, v in kwargs.items() if not empty(v)}
if kwargs:
return find_object(mylist, error=error, **kwargs)
if error is not None:
raise error()
class UserError(Exception):
"""
Exception containing an error message for user.
"""
class FieldNotFound(Exception):
"""
A field isn't found.
:param obj: object
:type obj: :class:`BaseObject`
:param field: field not found
:type field: :class:`Field`
"""
def __init__(self, obj, field):
super(FieldNotFound, self).__init__(u'Field "%s" not found for object %s' % (field, obj))
class ConversionWarning(UserWarning):
"""
A field's type was changed when setting it.
Ideally, the module should use the right type before setting it.
"""
pass
class AttributeCreationWarning(UserWarning):
"""
A non-field attribute has been created with a name not
prefixed with a _.
"""
class EmptyType(object):
"""
Parent class for NotAvailableType, NotLoadedType and FetchErrorType.
"""
def __str__(self):
return repr(self)
def __copy__(self):
return self
def __deepcopy__(self, memo):
return self
def __nonzero__(self):
return False
__bool__ = __nonzero__
class NotAvailableType(EmptyType):
"""
NotAvailable is a constant to use on non available fields.
"""
def __repr__(self):
return 'NotAvailable'
def __unicode__(self):
return u'Not available'
NotAvailable = NotAvailableType()
class NotLoadedType(EmptyType):
"""
NotLoaded is a constant to use on not loaded fields.
When you use :func:`weboob.tools.backend.Module.fillobj` on a object based on :class:`BaseObject`,
it will request all fields with this value.
"""
def __repr__(self):
return 'NotLoaded'
def __unicode__(self):
return u'Not loaded'
NotLoaded = NotLoadedType()
class FetchErrorType(EmptyType):
"""
FetchError is a constant to use when parsing a non-mandatory field raises an exception.
"""
def __repr__(self):
return 'FetchError'
def __unicode__(self):
return u'Not mandatory'
FetchError = FetchErrorType()
class Capability(object):
"""
This is the base class for all capabilities.
A capability may define abstract methods (which raise :class:`NotImplementedError`)
with an explicit docstring to tell backends how to implement them.
Also, it may define some *objects*, using :class:`BaseObject`.
"""
class Field(object):
"""
Field of a :class:`BaseObject` class.
:param doc: docstring of the field
:type doc: :class:`str`
:param args: list of types accepted
:param default: default value of this field. If not specified, :class:`NotLoaded` is used.
"""
_creation_counter = 0
def __init__(self, doc, *args, **kwargs):
self.types = ()
self.value = kwargs.get('default', NotLoaded)
self.doc = doc
self.mandatory = kwargs.get('mandatory', True)
for arg in args:
if isinstance(arg, type) or isinstance(arg, str):
self.types += (arg,)
else:
raise TypeError('Arguments must be types or strings of type name')
self._creation_counter = Field._creation_counter
Field._creation_counter += 1
def convert(self, value):
"""
Convert value to the wanted one.
"""
return value
class IntField(Field):
"""
A field which accepts only :class:`int` and :class:`long` types.
"""
def __init__(self, doc, **kwargs):
super(IntField, self).__init__(doc, int, long, **kwargs)
def convert(self, value):
return int(value)
class BoolField(Field):
"""
A field which accepts only :class:`bool` type.
"""
def __init__(self, doc, **kwargs):
super(BoolField, self).__init__(doc, bool, **kwargs)
def convert(self, value):
return bool(value)
class DecimalField(Field):
"""
A field which accepts only :class:`decimal` type.
"""
def __init__(self, doc, **kwargs):
super(DecimalField, self).__init__(doc, Decimal, **kwargs)
def convert(self, value):
if isinstance(value, Decimal):
return value
return Decimal(value)
class FloatField(Field):
"""
A field which accepts only :class:`float` type.
"""
def __init__(self, doc, **kwargs):
super(FloatField, self).__init__(doc, float, **kwargs)
def convert(self, value):
return float(value)
class StringField(Field):
"""
A field which accepts only :class:`unicode` strings.
"""
def __init__(self, doc, **kwargs):
super(StringField, self).__init__(doc, unicode, **kwargs)
def convert(self, value):
return to_unicode(value)
class BytesField(Field):
"""
A field which accepts only :class:`bytes` strings.
"""
def __init__(self, doc, **kwargs):
super(BytesField, self).__init__(doc, bytes, **kwargs)
def convert(self, value):
if isinstance(value, unicode):
value = value.encode('utf-8')
return bytes(value)
class EnumField(Field):
def __init__(self, doc, enum, **kwargs):
if not issubclass(enum, Enum):
raise TypeError('invalid enum type: %r' % enum)
super(EnumField, self).__init__(doc, *enum._types, **kwargs)
self.enum = enum
def convert(self, value):
if value not in self.enum._values:
raise ValueError('value %r does not belong to enum %s' % (value, self.enum))
return value
class _BaseObjectMeta(type):
def __new__(cls, name, bases, attrs):
fields = [(field_name, attrs.pop(field_name)) for field_name, obj in list(attrs.items()) if isinstance(obj, Field)]
fields.sort(key=lambda x: x[1]._creation_counter)
new_class = super(_BaseObjectMeta, cls).__new__(cls, name, bases, attrs)
if new_class._fields is None:
new_class._fields = OrderedDict()
else:
new_class._fields = deepcopy(new_class._fields)
new_class._fields.update(fields)
if new_class.__doc__ is None:
new_class.__doc__ = ''
for name, field in fields:
doc = '(%s) %s' % (', '.join([':class:`%s`' % v.__name__ if isinstance(v, type) else v for v in field.types]), field.doc)
if field.value is not NotLoaded:
doc += ' (default: %s)' % field.value
new_class.__doc__ += '\n:var %s: %s' % (name, doc)
return new_class
class BaseObject(with_metaclass(_BaseObjectMeta, StrConv, object)):
"""
This is the base class for a capability object.
A capability interface may specify to return several kind of objects, to formalise
retrieved information from websites.
As python is a flexible language where variables are not typed, we use a system to
force backends to set wanted values on all fields. To do that, we use the :class:`Field`
class and all derived ones.
For example::
class Transfer(BaseObject):
" Transfer from an account to a recipient. "
amount = DecimalField('Amount to transfer')
date = Field('Date of transfer', basestring, date, datetime)
origin = Field('Origin of transfer', int, long, basestring)
recipient = Field('Recipient', int, long, basestring)
The docstring is mandatory.
"""
id = None
backend = None
url = StringField('url')
_fields = None
def __init__(self, id=u'', url=NotLoaded, backend=None):
self.id = to_unicode(id)
self.backend = backend
self._fields = deepcopy(self._fields)
self.__setattr__('url', url)
@property
def fullid(self):
"""
Full ID of the object, in form '**ID@backend**'.
"""
return '%s@%s' % (self.id, self.backend)
def __iscomplete__(self):
"""
Return True if the object is completed.
It is useful when the object is a field of an other object which is
going to be filled.
The default behavior is to iter on fields (with iter_fields) and if
a field is NotLoaded, return False.
"""
for key, value in self.iter_fields():
if value is NotLoaded:
return False
return True
def copy(self):
obj = copy(self)
obj._fields = copy(self._fields)
for k in obj._fields:
obj._fields[k] = copy(obj._fields[k])
return obj
def __deepcopy__(self, memo):
return self.copy()
def set_empty_fields(self, value, excepts=()):
"""
Set the same value on all empty fields.
:param value: value to set on all empty fields
:param excepts: if specified, do not change fields listed
"""
for key, old_value in self.iter_fields():
if empty(old_value) and key not in excepts:
setattr(self, key, value)
def iter_fields(self):
"""
Iterate on the fields keys and values.
Can be overloaded to iterate on other things.
:rtype: iter[(key, value)]
"""
if hasattr(self, 'id') and self.id is not None:
yield 'id', self.id
for name, field in self._fields.items():
yield name, field.value
def __eq__(self, obj):
if isinstance(obj, BaseObject):
return self.backend == obj.backend and self.id == obj.id
else:
return False
def __getattr__(self, name):
if self._fields is not None and name in self._fields:
return self._fields[name].value
else:
raise AttributeError("'%s' object has no attribute '%s'" % (
self.__class__.__name__, name))
def __setattr__(self, name, value):
try:
attr = (self._fields or {})[name]
except KeyError:
if name not in dir(self) and not name.startswith('_'):
warnings.warn('Creating a non-field attribute %s. Please prefix it with _' % name,
AttributeCreationWarning, stacklevel=2)
object.__setattr__(self, name, value)
else:
if not empty(value):
try:
# Try to convert value to the wanted one.
nvalue = attr.convert(value)
# If the value was converted
if nvalue is not value:
warnings.warn('Value %s was converted from %s to %s' %
(name, type(value), type(nvalue)),
ConversionWarning, stacklevel=2)
value = nvalue
except Exception:
# error during conversion, it will probably not
# match the wanted following types, so we'll
# raise ValueError.
pass
from collections import deque
actual_types = ()
for v in attr.types:
if isinstance(v, str):
# the following is a (almost) copy/paste from
# https://stackoverflow.com/questions/11775460/lexical-cast-from-string-to-type
q = deque([object])
while q:
t = q.popleft()
if t.__name__ == v:
actual_types += (t,)
else:
try:
# keep looking!
q.extend(t.__subclasses__())
except TypeError:
# type.__subclasses__ needs an argument for
# whatever reason.
if t is type:
continue
else:
raise
else:
actual_types += (v,)
if not isinstance(value, actual_types) and not empty(value):
raise ValueError(
'Value for "%s" needs to be of type %r, not %r' % (
name, actual_types, type(value)))
attr.value = value
def __delattr__(self, name):
try:
self._fields.pop(name)
except KeyError:
object.__delattr__(self, name)
def to_dict(self):
def iter_decorate(d):
for key, value in d:
if key == 'id' and self.backend is not None:
value = self.fullid
yield key, value
fields_iterator = self.iter_fields()
return OrderedDict(iter_decorate(fields_iterator))
def __getstate__(self):
d = self.to_dict()
d.update((k, v) for k, v in self.__dict__.items() if k != '_fields')
return d
@classmethod
def from_dict(cls, values, backend=None):
self = cls()
for attr in values:
setattr(self, attr, values[attr])
return self
def __setstate__(self, state):
self._fields = deepcopy(self._fields) # because yaml does not call __init__
for k in state:
setattr(self, k, state[k])
if sys.version_info.major >= 3:
def __dir__(self):
return list(super(BaseObject, self).__dir__()) + list(self._fields.keys())
class Currency(object):
CURRENCIES = OrderedDict([
(u'EUR', (u'€', u'EURO', u'EUROS')),
(u'CHF', (u'CHF',)),
(u'USD', (u'$',)),
(u'GBP', (u'£',)),
(u'LBP', (u'ل.ل',)),
(u'AED', (u'AED',)),
(u'XOF', (u'XOF',)),
(u'RUB', (u'руб',)),
(u'SGD', (u'SGD',)),
(u'BRL', (u'R$',)),
(u'MXN', (u'$',)),
(u'JPY', (u'¥',)),
(u'TRY', (u'₺', u'TRY')),
(u'RON', (u'lei',)),
(u'COP', (u'$',)),
(u'NOK', (u'kr',)),
(u'CNY', (u'¥',)),
(u'RSD', (u'din',)),
(u'ZAR', (u'rand',)),
(u'MYR', (u'RM',)),
(u'HUF', (u'Ft',)),
(u'HKD', (u'HK$',)),
(u'TWD', (u'NT$',)),
(u'QAR', (u'QR',)),
(u'MAD', (u'MAD',)),
(u'ARS', (u'ARS',)),
(u'AUD', (u'AUD',)),
(u'CAD', (u'CAD',)),
(u'NZD', (u'NZD',)),
(u'BHD', (u'BHD',)),
(u'SEK', (u'SEK',)),
(u'DKK', (u'DKK',)),
(u'LUF', (u'LUF',)),
(u'KZT', (u'KZT',)),
(u'PLN', (u'PLN',)),
(u'ILS', (u'ILS',)),
(u'THB', (u'THB',)),
(u'INR', (u'₹', u'INR')),
(u'PEN', (u'S/',)),
(u'IDR', (u'Rp',)),
(u'KWD', (u'KD',)),
(u'KRW', (u'₩',)),
(u'CZK', (u'Kč',)),
(u'EGP', (u'E£',)),
(u'ISK', (u'Íkr', u'kr')),
(u'XPF', (u'XPF',)),
])
EXTRACTOR = re.compile(r'[()\d\s,\.\-]', re.UNICODE)
@classmethod
def get_currency(klass, text):
u"""
>>> Currency.get_currency(u'42')
None
>>> Currency.get_currency(u'42 €')
u'EUR'
>>> Currency.get_currency(u'$42')
u'USD'
>>> Currency.get_currency(u'42.000,00€')
u'EUR'
>>> Currency.get_currency(u'$42 USD')
u'USD'
>>> Currency.get_currency(u'%42 USD')
u'USD'
>>> Currency.get_currency(u'US1D')
None
"""
curtexts = klass.EXTRACTOR.sub(' ', text.upper()).split()
for currency, symbols in klass.CURRENCIES.items():
for curtext in curtexts:
if curtext == currency:
return currency
for symbol in symbols:
if curtext == symbol:
return currency
return None
@classmethod
def currency2txt(klass, currency):
_currency = klass.CURRENCIES.get(currency, (u'',))
return _currency[0]
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/bill.py 0000664 0000000 0000000 00000017130 13414137563 0030114 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2012 Romain Bignon, Florent Fourcot
#
# 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 .
from .base import BaseObject, StringField, DecimalField, BoolField, UserError, Currency, Field
from .date import DateField
from .collection import CapCollection
__all__ = ['SubscriptionNotFound', 'DocumentNotFound', 'DocumentTypes', 'Detail', 'Document', 'Bill', 'Subscription', 'CapDocument']
class SubscriptionNotFound(UserError):
"""
Raised when a subscription is not found.
"""
def __init__(self, msg='Subscription not found'):
super(SubscriptionNotFound, self).__init__(msg)
class DocumentNotFound(UserError):
"""
Raised when a document is not found.
"""
def __init__(self, msg='Document not found'):
super(DocumentNotFound, self).__init__(msg)
class DocumentTypes(object):
RIB = u'RIB'
STATEMENT = u'statement'
CONTRACT = u'contract'
NOTICE = u'notice'
REPORT = u'report'
BILL = u'bill'
OTHER = u'other'
class Detail(BaseObject, Currency):
"""
Detail of a subscription
"""
label = StringField('label of the detail line')
infos = StringField('information')
datetime = DateField('date information')
price = DecimalField('Total price, taxes included')
vat = DecimalField('Value added Tax')
currency = StringField('Currency', default=None)
quantity = DecimalField('Number of units consumed')
unit = StringField('Unit of the consumption')
class Document(BaseObject):
"""
Document.
"""
date = DateField('The day the document has been sent to the subscriber')
format = StringField('file format of the document')
label = StringField('label of document')
type = StringField('type of document')
transactions = Field('List of transaction ID related to the document', list, default=[])
class Bill(Document, Currency):
"""
Bill.
"""
price = DecimalField('Price to pay')
currency = StringField('Currency', default=None)
vat = DecimalField('VAT included in the price')
duedate = DateField('The day the bill must be paid')
startdate = DateField('The first day the bill applies to')
finishdate = DateField('The last day the bill applies to')
income = BoolField('Boolean to set if bill is income or invoice', default=False)
def __init__(self, *args, **kwargs):
super(Bill, self).__init__(*args, **kwargs)
self.type = u'bill'
class Subscription(BaseObject):
"""
Subscription to a service.
"""
label = StringField('label of subscription')
subscriber = StringField('Subscriber name or identifier (for companies)')
validity = DateField('End validity date of the subscription (if any)')
renewdate = DateField('Reset date of consumption, for time based suscription (monthly, yearly, etc)')
class CapDocument(CapCollection):
accepted_doc_types = ()
"""
Tuple of document types handled by the module (:class:`DocumentTypes`)
"""
def iter_subscription(self):
"""
Iter subscriptions.
:rtype: iter[:class:`Subscription`]
"""
raise NotImplementedError()
def get_subscription(self, _id):
"""
Get a subscription.
:param _id: ID of subscription
:rtype: :class:`Subscription`
:raises: :class:`SubscriptionNotFound`
"""
raise NotImplementedError()
def iter_documents_history(self, subscription):
"""
Iter history of a subscription.
:param subscription: subscription to get history
:type subscription: :class:`Subscription`
:rtype: iter[:class:`Detail`]
"""
raise NotImplementedError()
def iter_bills_history(self, subscription):
"""
Iter history of a subscription.
:param subscription: subscription to get history
:type subscription: :class:`Subscription`
:rtype: iter[:class:`Detail`]
"""
return self.iter_documents_history(subscription)
def get_document(self, id):
"""
Get a document.
:param id: ID of document
:rtype: :class:`Document`
:raises: :class:`DocumentNotFound`
"""
raise NotImplementedError()
def download_document(self, id):
"""
Download a document.
:param id: ID of document
:rtype: bytes
:raises: :class:`DocumentNotFound`
"""
raise NotImplementedError()
def download_document_pdf(self, id):
"""
Download a document, convert it to PDF if it isn't the document format.
:param id: ID of document
:rtype: bytes
:raises: :class:`DocumentNotFound`
"""
if not isinstance(id, Document):
id = self.get_document(id)
if id.format == 'pdf':
return self.download_document(id)
else:
raise NotImplementedError()
def iter_documents(self, subscription):
"""
Iter documents.
:param subscription: subscription to get documents
:type subscription: :class:`Subscription`
:rtype: iter[:class:`Document`]
"""
raise NotImplementedError()
def iter_documents_by_types(self, subscription, accepted_types):
"""
Iter documents with specific types.
:param subscription: subscription to get documents
:type subscription: :class:`Subscription`
:param accepted_types: list of document types that should be returned
:type accepted_types: [:class:`DocumentTypes`]
:rtype: iter[:class:`Document`]
"""
accepted_types = frozenset(accepted_types)
for document in self.iter_documents(subscription):
if document.type in accepted_types:
yield document
def iter_bills(self, subscription):
"""
Iter bills.
:param subscription: subscription to get bills
:type subscription: :class:`Subscription`
:rtype: iter[:class:`Bill`]
"""
documents = self.iter_documents(subscription)
return [doc for doc in documents if doc.type == "bill"]
def get_details(self, subscription):
"""
Get details of a subscription.
:param subscription: subscription to get details
:type subscription: :class:`Subscription`
:rtype: iter[:class:`Detail`]
"""
raise NotImplementedError()
def get_balance(self, subscription):
"""
Get the balance of a subscription.
:param subscription: subscription to get balance
:type subscription: :class:`Subscription`
:rtype: class:`Detail`
"""
raise NotImplementedError()
def iter_resources(self, objs, split_path):
"""
Iter resources. Will return :func:`iter_subscriptions`.
"""
if Subscription in objs:
self._restrict_level(split_path)
return self.iter_subscription()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/bugtracker.py 0000664 0000000 0000000 00000022477 13414137563 0031335 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2011 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 .
from weboob.tools.compat import unicode
from .base import Capability, BaseObject, Field, StringField,\
IntField, UserError, Enum
from .date import DateField, DeltaField
__all__ = ['IssueError', 'Project', 'User', 'Version', 'Status', 'Attachment',
'Change', 'Update', 'Issue', 'Query', 'CapBugTracker']
class IssueError(UserError):
"""
Raised when there is an error with an issue.
"""
class Project(BaseObject):
"""
Represents a project.
"""
name = StringField('Name of the project')
members = Field('Members of projects', list)
versions = Field('List of versions available for this project', list)
trackers = Field('All trackers', list)
categories = Field('All categories', list)
statuses = Field('Available statuses for issues', list)
priorities = Field('Available priorities for issues', list)
tags = Field('All tags', tuple, list)
fields = Field('Custom fields names', list)
def __init__(self, id, name, url=None):
super(Project, self).__init__(id, url)
self.name = unicode(name)
def __repr__(self):
return '' % self.name
def find_user(self, id, name):
"""
Find a user from its ID.
If not found, create a :class:`User` with the specified name.
:param id: ID of user
:type id: str
:param name: Name of user
:type name: str
:rtype: :class:`User`
"""
for user in self.members:
if user.id == id:
return user
if name is None:
return None
return User(id, name)
def find_version(self, id, name):
"""
Find a version from an ID.
If not found, create a :class:`Version` with the specified name.
:param id: ID of version
:type id: str
:param name: Name of version
:type name: str
:rtype: :class:`Version`
"""
for version in self.versions:
if version.id == id:
return version
if name is None:
return None
return Version(id, name)
def find_status(self, name):
"""
Find a status from a name.
:param name: Name of status
:type name: str
:rtype: :class:`Status`
"""
for status in self.statuses:
if status.name == name:
return status
if name is None:
return None
return None
class User(BaseObject):
"""
User.
"""
name = StringField('Name of user')
def __init__(self, id, name, url=None):
super(User, self).__init__(id, url)
self.name = unicode(name)
def __repr__(self):
return '' % self.name
class Version(BaseObject):
"""
Version of a project.
"""
name = StringField('Name of version')
def __init__(self, id, name, url=None):
super(Version, self).__init__(id, url)
self.name = unicode(name)
def __repr__(self):
return '' % self.name
class StatusType(Enum):
NEW = 0
PROGRESS = 1
RESOLVED = 2
REJECTED = 3
class Status(BaseObject):
"""
Status of an issue.
**VALUE_** constants are the primary status
types.
"""
VALUE_NEW = StatusType.NEW
VALUE_PROGRESS = StatusType.PROGRESS
VALUE_RESOLVED = StatusType.RESOLVED
VALUE_REJECTED = StatusType.REJECTED
name = StringField('Name of status')
value = IntField('Value of status (constants VALUE_*)')
def __init__(self, id, name, value, url=None):
super(Status, self).__init__(id, url)
self.name = unicode(name)
self.value = value
def __repr__(self):
return '' % self.name
class Attachment(BaseObject):
"""
Attachment of an issue.
"""
filename = StringField('Filename')
def __repr__(self):
return '' % self.filename
class Change(BaseObject):
"""
A change of an update.
"""
field = StringField('What field has been changed')
last = StringField('Last value of field')
new = StringField('New value of field')
def __repr__(self):
return '<%s %r: %r (old: %r)>' % (type(self).__name__, self.field, self.new, self.last)
class Update(BaseObject):
"""
Represents an update of an issue.
"""
author = Field('Author of update', User)
date = DateField('Date of update')
hours = DeltaField('Time activity')
message = StringField('Log message')
attachments = Field('Files attached to update', list, tuple)
changes = Field('List of changes', list, tuple)
def __repr__(self):
return '' % self.id
class Issue(BaseObject):
"""
Represents an issue.
"""
project = Field('Project of this issue', Project)
title = StringField('Title of issue')
body = StringField('Text of issue')
creation = DateField('Date when this issue has been created')
updated = DateField('Date when this issue has been updated for the last time')
start = DateField('Date when this issue starts')
due = DateField('Date when this issue is due for')
attachments = Field('List of attached files', list, tuple)
history = Field('History of updates', list, tuple)
author = Field('Author of this issue', User)
assignee = Field('User assigned to this issue', User)
tracker = StringField('Name of the tracker')
category = StringField('Name of the category')
version = Field('Target version of this issue', Version)
status = Field('Status of this issue', Status)
fields = Field('Custom fields (key,value)', dict)
priority = StringField('Priority of the issue') #XXX
tags = Field('Categories/Tags of the issue', tuple, list)
related_issues = Field('Related issues', list)
class Query(BaseObject):
"""
Query to find an issue.
"""
project = Field('Filter on projects', str, unicode, Project)
title = StringField('Filter on titles')
author = Field('Filter on authors', str, unicode, User)
assignee = Field('Filter on assignees', str, unicode, User)
version = Field('Filter on versions', str, unicode, Version)
category = StringField('Filter on categories')
status = Field('Filter on statuses', str, unicode, Status)
tags = Field('Filter on tags', tuple, list)
fields = Field('Filter on custom fields', dict)
def __init__(self, id='', url=None):
super(Query, self).__init__(id, url)
class CapBugTracker(Capability):
"""
Bug trackers websites.
"""
(SORT_RELEVANCE,
SORT_RATING,
SORT_PRIORITY,
SORT_DATE) = range(4)
def iter_issues(self, query, sortby=SORT_RELEVANCE, ascending=False):
"""
Iter issues with optionnal patterns.
:param query: query
:type query: :class:`Query`
:rtype: iter[:class:`Issue`]
"""
raise NotImplementedError()
def get_issue(self, id):
"""
Get an issue from its ID.
:param id: ID of issue
:rtype: :class:`Issue`
"""
raise NotImplementedError()
def create_issue(self, project):
"""
Create an empty issue on the given project.
:param project: project
:type project: :class:`Project`
:returns: the created issue
:rtype: :class:`Issue`
"""
raise NotImplementedError()
def post_issue(self, issue):
"""
Post an issue to create or update it.
:param issue: issue to create or update
:type issue: :class:`Issue`
"""
raise NotImplementedError()
def update_issue(self, issue, update):
"""
Add an update to an issue.
:param issue: issue or id of issue
:type issue: :class:`Issue`
:param update: an Update object
:type update: :class:`Update`
"""
raise NotImplementedError()
def remove_issue(self, issue):
"""
Remove an issue.
:param issue: issue
:type issue: :class:`Issue`
"""
raise NotImplementedError()
def iter_projects(self):
"""
Iter projects.
:rtype: iter[:class:`Project`]
"""
raise NotImplementedError()
def get_project(self, id):
"""
Get a project from its ID.
:rtype: :class:`Project`
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/calendar.py 0000664 0000000 0000000 00000014440 13414137563 0030744 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2013 Bezleputh
#
# 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 .
from .base import (
BaseObject, StringField, IntField, FloatField, Field, EnumField,
Enum,
)
from .collection import CapCollection, CollectionNotFound, Collection
from .date import DateField
from datetime import time, datetime
from weboob.tools.date import parse_date
__all__ = ['BaseCalendarEvent', 'CapCalendarEvent']
class CATEGORIES(Enum):
CONCERT = u'Concert'
CINE = u'Cinema'
THEATRE = u'Theatre'
TELE = u'Television'
CONF = u'Conference'
AUTRE = u'Autre'
EXPO = u'Exposition'
SPECTACLE = u'Spectacle'
FEST = u'Festival'
SPORT = u'Sport'
#the following elements deal with ICalendar stantdards
#see http://fr.wikipedia.org/wiki/ICalendar#Ev.C3.A9nements_.28VEVENT.29
class TRANSP(Enum):
OPAQUE = u'OPAQUE'
TRANSPARENT = u'TRANSPARENT'
class STATUS(Enum):
TENTATIVE = u'TENTATIVE'
CONFIRMED = u'CONFIRMED'
CANCELLED = u'CANCELLED'
class TICKET(Enum):
AVAILABLE = u'Available'
NOTAVAILABLE = u'Not available'
CLOSED = u'Closed'
class BaseCalendarEvent(BaseObject):
"""
Represents a calendar event
"""
start_date = DateField('Start date of the event')
end_date = DateField('End date of the event')
timezone = StringField('Timezone of the event in order to convert to utc time', default='Etc/UCT')
summary = StringField('Title of the event')
city = StringField('Name of the city in witch event will take place')
location = StringField('Location of the event')
category = EnumField('Category of the event', CATEGORIES)
description = StringField('Description of the event')
price = FloatField('Price of the event')
booked_entries = IntField('Entry number')
max_entries = IntField('Max entry number')
event_planner = StringField('Name of the event planner')
#the following elements deal with ICalendar stantdards
#see http://fr.wikipedia.org/wiki/ICalendar#Ev.C3.A9nements_.28VEVENT.29
sequence = IntField('Number of updates, the first is number 1')
# (TENTATIVE, CONFIRMED, CANCELLED)
status = EnumField('Status of the event', STATUS)
# (OPAQUE, TRANSPARENT)
transp = EnumField('Describes if event is available', TRANSP)
# (AVAILABLE, NOTAVAILABLE, CLOSED)
ticket = EnumField('Describes if tickets are available', TICKET)
@classmethod
def id2url(cls, _id):
"""Overloaded in child classes provided by backends."""
raise NotImplementedError()
@property
def page_url(self):
"""
Get page URL of the announce.
"""
return self.id2url(self.id)
class Query(BaseObject):
"""
Query to find events
"""
start_date = DateField('Start date of the event')
end_date = DateField('End date of the event')
city = StringField('Name of the city in witch event will take place')
categories = Field('List of categories of the event', list, tuple, default=list(CATEGORIES))
ticket = Field('List of status of the tickets sale', list, tuple, default=list(TICKET))
summary = StringField('Title of the event')
class CapCalendarEvent(CapCollection):
"""
Capability of calendar event type sites
"""
ASSOCIATED_CATEGORIES = 'ALL'
def has_matching_categories(self, query):
if self.ASSOCIATED_CATEGORIES == 'ALL':
return True
for category in query.categories:
if category in self.ASSOCIATED_CATEGORIES:
return True
return False
def search_events(self, query):
"""
Search event
:param query: search query
:type query: :class:`Query`
:rtype: iter[:class:`BaseCalendarEvent`]
"""
raise NotImplementedError()
def list_events(self, date_from, date_to=None):
"""
list coming event.
:param date_from: date of beguinning of the events list
:type date_from: date
:param date_to: date of ending of the events list
:type date_to: date
:rtype: iter[:class:`BaseCalendarEvent`]
"""
raise NotImplementedError()
def get_event(self, _id):
"""
Get an event from an ID.
:param _id: id of the event
:type _id: str
:rtype: :class:`BaseCalendarEvent` or None is fot found.
"""
raise NotImplementedError()
def attends_event(self, event, is_attending=True):
"""
Attends or not to an event
:param event : the event
:type event : BaseCalendarEvent
:param is_attending : is attending to the event or not
:type is_attending : bool
"""
raise NotImplementedError()
def iter_resources(self, objs, split_path):
"""
Iter events by category
"""
if len(split_path) == 0 and self.ASSOCIATED_CATEGORIES != 'ALL':
for category in self.ASSOCIATED_CATEGORIES:
collection = Collection([category], category)
yield collection
elif len(split_path) == 1 and split_path[0] in self.ASSOCIATED_CATEGORIES:
query = Query()
query.categories = split_path
query.start_date = datetime.combine(parse_date('today'), time.min)
query.end_date = parse_date('')
query.city = u''
for event in self.search_events(query):
yield event
def validate_collection(self, objs, collection):
"""
Validate Collection
"""
if collection.path_level == 0:
return
if collection.path_level == 1 and collection.split_path[0] in CATEGORIES.values:
return
raise CollectionNotFound(collection.split_path)
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/captcha.py 0000664 0000000 0000000 00000010766 13414137563 0030605 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2018 Vincent A
#
# 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 .
from time import sleep
from .base import Capability, BaseObject, StringField, UserError, BytesField
from ..exceptions import RecaptchaQuestion, NocaptchaQuestion, ImageCaptchaQuestion
__all__ = [
'CapCaptchaSolver',
'SolverJob', 'RecaptchaJob', 'NocaptchaJob', 'ImageCaptchaJob',
'CaptchaError', 'UnsolvableCaptcha', 'InvalidCaptcha', 'InsufficientFunds',
'exception_to_job',
]
class SolverJob(BaseObject):
solution = StringField('CAPTCHA solution')
class RecaptchaJob(SolverJob):
site_url = StringField('Site URL for ReCaptcha service')
site_key = StringField('Site key for ReCaptcha service')
solution_challenge = StringField('Challenge ID of the solution (output value)')
class NocaptchaJob(SolverJob):
site_url = StringField('Site URL for NoCaptcha service')
site_key = StringField('Site key for NoCaptcha service')
class ImageCaptchaJob(SolverJob):
image = BytesField('data of the image to solve')
class CaptchaError(UserError):
"""Generic solving error"""
class InvalidCaptcha(CaptchaError):
"""CAPTCHA cannot be used (e.g. invalid image format)"""
class UnsolvableCaptcha(CaptchaError):
"""CAPTCHA is too hard or impossible"""
class InsufficientFunds(CaptchaError):
"""Not enough funds to pay solution"""
def exception_to_job(exc):
if isinstance(exc, RecaptchaQuestion):
job = RecaptchaJob()
job.site_url = exc.website_url
job.site_key = exc.website_key
elif isinstance(exc, NocaptchaQuestion):
job = NocaptchaJob()
job.site_url = exc.website_url
job.site_key = exc.website_key
elif isinstance(exc, ImageCaptchaQuestion):
job = ImageCaptchaJob()
job.image = exc.image_data
else:
raise NotImplementedError()
return job
class CapCaptchaSolver(Capability):
"""
Provide CAPTCHA solving
"""
RETRIES = 30
WAIT_TIME = 2
def create_job(self, job):
"""Start a CAPTCHA solving job
The `job.id` shall be filled. The CAPTCHA is not solved yet when the method returns.
:param job: job to start
:type job: :class:`SolverJob`
:raises: :class:`NotImplementedError` if CAPTCHA type is not supported
:raises: :class:`CaptchaError` in case of other error
"""
raise NotImplementedError()
def poll_job(self, job):
"""Check if a job was solved
If `job` is solved, return True and fill `job.solution`.
Return False if solution is still pending.
In case of solving problem, an exception may be raised.
It should not wait for the solution but return the current state.
:param job: job to check and to fill when solved
:type job: :class:`SolverJob`
:returns: True if the job was solved
:rtype: bool
:raises: :class:`CaptchaError`
"""
raise NotImplementedError()
def solve_catpcha_blocking(self, job):
"""Start a CAPTCHA solving job and wait for its solution
:param job: job to start and solve
:type job: :class:`SolverJob`
:raises: :class:`CaptchaError`
"""
self.create_job(job)
for i in range(self.RETRIES):
sleep(self.WAIT_TIME)
if self.poll_job(job):
return job
def report_wrong_solution(self, job):
"""Report a solved job as a wrong solution
Sometimes, jobs are solved, but the solution is rejected by the CAPTCHA
site because the solution is wrong.
This method reports the solution as wrong to the CAPTCHA solver.
:param job: job to flag
:type job: :class:`SolverJob`
"""
raise NotImplementedError()
def get_balance(self):
"""Get the prepaid balance left
:rtype: float
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/chat.py 0000664 0000000 0000000 00000004443 13414137563 0030114 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2011 Christophe Benz
#
# 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 .
import datetime
from .base import Capability, BaseObject, StringField, UserError
from .date import DateField
__all__ = ['ChatException', 'ChatMessage', 'CapChat']
class ChatException(UserError):
"""
Exception raised when there is a problem with the chat.
"""
class ChatMessage(BaseObject):
"""
Message on the chat.
"""
id_from = StringField('ID of sender')
id_to = StringField('ID of recipient')
message = StringField('Content of message')
date = DateField('Date when the message has been sent')
def __init__(self, id_from, id_to, message, date=None, url=None):
super(ChatMessage, self).__init__('%s.%s' % (id_from, id_to), url)
self.id_from = id_from
self.id_to = id_to
self.message = message
self.date = date
if self.date is None:
self.date = datetime.datetime.utcnow()
class CapChat(Capability):
"""
Websites with a chat system.
"""
def iter_chat_messages(self, _id=None):
"""
Iter messages.
:param _id: optional parameter to only get messages
from a given contact.
:type _id: str
:rtype: iter[:class:`ChatMessage`]
"""
raise NotImplementedError()
def send_chat_message(self, _id, message):
"""
Send a message to a contact.
:param _id: ID of recipient
:type _id: str
:param message: message to send
:type message: str
:raises: :class:`ChatException`
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/cinema.py 0000664 0000000 0000000 00000014277 13414137563 0030437 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2013 Julien Veyssier
#
# 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 .
from .base import Capability, BaseObject, StringField, IntField, Field
from .date import DateField
__all__ = ['Movie', 'Person', 'CapCinema']
class Movie(BaseObject):
"""
Movie object.
"""
original_title = StringField('Original title of the movie')
other_titles = Field('Titles in other countries', list)
release_date = DateField('Release date of the movie')
all_release_dates = StringField('Release dates list of the movie')
duration = IntField('Duration of the movie in minutes')
short_description = StringField('Short description of the movie')
genres = Field('Genres of the movie', list)
pitch = StringField('Short story description of the movie')
country = StringField('Origin country of the movie')
note = StringField('Notation of the movie')
roles = Field('Lists of Persons related to the movie indexed by roles', dict)
thumbnail_url = StringField('Url of movie thumbnail')
def __init__(self, id, original_title, url=None):
super(Movie, self).__init__(id, url)
self.original_title = original_title
def get_roles_by_person_name(self, name):
for role in self.roles.keys():
if name.lower() in [person[1].lower() for person in self.roles[role]]:
return role
return None
def get_roles_by_person_id(self, id):
result = []
for role in self.roles.keys():
if id in [person[0] for person in self.roles[role]]:
result.append(role)
return result
class Person(BaseObject):
"""
Person object.
"""
name = StringField('Star name of a person')
real_name = StringField('Real name of a person')
birth_date = DateField('Birth date of a person')
death_date = DateField('Death date of a person')
birth_place = StringField('City and country of birth of a person')
gender = StringField('Gender of a person')
nationality = StringField('Nationality of a person')
short_biography = StringField('Short biography of a person')
biography = StringField('Full biography of a person')
short_description = StringField('Short description of a person')
roles = Field('Lists of movies related to the person indexed by roles', dict)
thumbnail_url = StringField('Url of person thumbnail')
def __init__(self, id, name, url=None):
super(Person, self).__init__(id, url)
self.name = name
def get_roles_by_movie_title(self, title):
for role in self.roles.keys():
for mt in [movie[1] for movie in self.roles[role]]:
# title we have is included ?
if title.lower() in mt.lower():
return role
return None
def get_roles_by_movie_id(self, id):
result = []
for role in self.roles.keys():
if id in [movie[0] for movie in self.roles[role]]:
result.append(role)
return result
class CapCinema(Capability):
"""
Cinema databases.
"""
def iter_movies(self, pattern):
"""
Search movies and iterate on results.
:param pattern: pattern to search
:type pattern: str
:rtype: iter[:class:`Movies`]
"""
raise NotImplementedError()
def get_movie(self, _id):
"""
Get a movie object from an ID.
:param _id: ID of movie
:type _id: str
:rtype: :class:`Movie`
"""
raise NotImplementedError()
def get_movie_releases(self, _id, country=None):
"""
Get a list of a movie releases from an ID.
:param _id: ID of movie
:type _id: str
:rtype: :class:`String`
"""
raise NotImplementedError()
def iter_movie_persons(self, _id, role=None):
"""
Get the list of persons who are related to a movie.
:param _id: ID of movie
:type _id: str
:rtype: iter[:class:`Person`]
"""
raise NotImplementedError()
def iter_persons(self, pattern):
"""
Search persons and iterate on results.
:param pattern: pattern to search
:type pattern: str
:rtype: iter[:class:`persons`]
"""
raise NotImplementedError()
def get_person(self, _id):
"""
Get a person object from an ID.
:param _id: ID of person
:type _id: str
:rtype: :class:`Person`
"""
raise NotImplementedError()
def iter_person_movies(self, _id, role=None):
"""
Get the list of movies related to a person.
:param _id: ID of person
:type _id: str
:rtype: iter[:class:`Movie`]
"""
raise NotImplementedError()
def iter_person_movies_ids(self, _id):
"""
Get the list of movie ids related to a person.
:param _id: ID of person
:type _id: str
:rtype: iter[str]
"""
raise NotImplementedError()
def iter_movie_persons_ids(self, _id):
"""
Get the list of person ids related to a movie.
:param _id: ID of movie
:type _id: str
:rtype: iter[str]
"""
raise NotImplementedError()
def get_person_biography(self, id):
"""
Get the person full biography.
:param _id: ID of person
:type _id: str
:rtype: str
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/collection.py 0000664 0000000 0000000 00000013223 13414137563 0031324 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2012 Nicolas Duhamel, Laurent Bachelier
#
# 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 .
from collections import OrderedDict
from .base import Capability, BaseObject, UserError, StringField, Field
__all__ = ['CapCollection', 'BaseCollection', 'Collection', 'CollectionNotFound']
class CollectionNotFound(UserError):
def __init__(self, split_path=None):
if split_path is not None:
msg = 'Collection not found: %s' % '/'.join(split_path)
else:
msg = 'Collection not found'
super(CollectionNotFound, self).__init__(msg)
class BaseCollection(BaseObject):
"""
Inherit from this if you want to create an object that is *also* a Collection.
However, this probably will not work properly for now.
"""
def __init__(self, split_path, id=None, url=None):
super(BaseCollection, self).__init__(id, url)
self.split_path = split_path
@property
def basename(self):
return self.split_path[-1] if self.path_level else None
@property
def parent_path(self):
return self.split_path[0:-1] if self.path_level else None
@property
def path_level(self):
return len(self.split_path)
def to_dict(self):
def iter_decorate(d):
for key, value in d:
if key == 'id' and self.backend is not None:
value = u'%s@%s' % (self.basename, self.backend)
yield key, value
if key == 'split_path':
yield key, '/'.join(value)
fields_iterator = self.iter_fields()
return OrderedDict(iter_decorate(fields_iterator))
class Collection(BaseCollection):
"""
A Collection is a "fake" object returned in results, which shows you can get
more results if you go into its path.
It is a dumb object, it must not contain callbacks to a backend.
Do not inherit from this class if you want to make a regular BaseObject
a Collection, use BaseCollection instead.
"""
title = StringField('Collection title')
split_path = Field('Full collection path', list)
def __init__(self, split_path=None, title=None, id=None, url=None):
self.title = title
super(Collection, self).__init__(split_path, id, url)
def __unicode__(self):
if self.title and self.basename:
return u'%s (%s)' % (self.basename, self.title)
elif self.basename:
return u'%s' % self.basename
else:
return u'Unknown collection'
class CapCollection(Capability):
def iter_resources_flat(self, objs, split_path, clean_only=False):
"""
Call iter_resources() to fetch all resources in the tree.
If clean_only is True, do not explore paths, only remove them.
split_path is used to set the starting path.
"""
for resource in self.iter_resources(objs, split_path):
if isinstance(resource, Collection):
if not clean_only:
for res in self.iter_resources_flat(objs, resource.split_path):
yield res
else:
yield resource
def iter_resources(self, objs, split_path):
"""
split_path is a list, either empty (root path) or with one or many
components.
"""
raise NotImplementedError()
def get_collection(self, objs, split_path):
"""
Get a collection for a given split path.
If the path is invalid (i.e. can't be handled by this module),
it should return None.
"""
collection = Collection(split_path, None)
return self.validate_collection(objs, collection) or collection
def validate_collection(self, objs, collection):
"""
Tests if a collection is valid.
For compatibility reasons, and to provide a default way, it checks if
the collection has at least one object in it. However, it is not very
efficient or exact, and you are encouraged to override this method.
You can replace the collection object entirely by returning a new one.
"""
# Root
if collection.path_level == 0:
return
try:
i = self.iter_resources(objs, collection.split_path)
next(i)
except StopIteration:
raise CollectionNotFound(collection.split_path)
def _restrict_level(self, split_path, lmax=0):
if len(split_path) > lmax:
raise CollectionNotFound(split_path)
def test():
c = Collection([])
assert c.basename is None
assert c.parent_path is None
assert c.path_level == 0
c = Collection([u'lol'])
assert c.basename == u'lol'
assert c.parent_path == []
assert c.path_level == 1
c = Collection([u'lol', u'cat'])
assert c.basename == u'cat'
assert c.parent_path == [u'lol']
assert c.path_level == 2
c = Collection([u'w', u'e', u'e', u'b', u'o', u'o', u'b'])
assert c.basename == u'b'
assert c.parent_path == [u'w', u'e', u'e', u'b', u'o', u'o']
assert c.path_level == 7
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/contact.py 0000664 0000000 0000000 00000024534 13414137563 0030633 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2011 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 .
from collections import OrderedDict
from datetime import datetime
from weboob.tools.compat import unicode, basestring
from dateutil import rrule
from .base import Capability, BaseObject, Field, StringField, BytesField, IntField, \
BoolField, UserError
__all__ = [
'ProfileNode', 'ContactPhoto', 'Contact', 'QueryError', 'Query', 'CapContact',
'BaseContact', 'PhysicalEntity', 'Person', 'Place', 'OpeningHours',
'OpeningRule', 'RRuleField', 'CapDirectory',
]
class ProfileNode(object):
"""
Node of a :class:`Contact` profile.
"""
HEAD = 0x01
SECTION = 0x02
def __init__(self, name, label, value, sufix=None, flags=0):
self.name = name
self.label = label
self.value = value
self.sufix = sufix
self.flags = flags
def __getitem__(self, key):
return self.value[key]
class ContactPhoto(BaseObject):
"""
Photo of a contact.
"""
name = StringField('Name of the photo')
data = BytesField('Data of photo')
thumbnail_url = StringField('Direct URL to thumbnail')
thumbnail_data = BytesField('Data of thumbnail')
hidden = BoolField('True if the photo is hidden on website')
def __init__(self, name, url=None):
super(ContactPhoto, self).__init__(name, url)
self.name = name
def __iscomplete__(self):
return (self.data and (not self.thumbnail_url or self.thumbnail_data))
def __unicode__(self):
return self.url
def __repr__(self):
return '' % (self.id,
len(self.data) if self.data else 0,
len(self.thumbnail_data) if self.thumbnail_data else 0)
class BaseContact(BaseObject):
"""
This is the blase class for a contact.
"""
name = StringField('Name of contact')
phone = StringField('Phone number')
email = StringField('Contact email')
website = StringField('Website URL of the contact')
class Advisor(BaseContact):
"""
An advisor.
"""
email = StringField('Mail of advisor')
phone = StringField('Phone number of advisor')
mobile = StringField('Mobile number of advisor')
fax = StringField('Fax number of advisor')
agency = StringField('Name of agency')
address = StringField('Address of agency')
role = StringField('Role of advisor', default="bank")
class Contact(BaseContact):
"""
A contact.
"""
STATUS_ONLINE = 0x001
STATUS_AWAY = 0x002
STATUS_OFFLINE = 0x004
STATUS_ALL = 0xfff
status = IntField('Status of contact (STATUS_* constants)')
status_msg = StringField('Message of status')
summary = StringField('Description of contact')
photos = Field('List of photos', dict, default=OrderedDict())
profile = Field('Contact profile', dict, default=OrderedDict())
def __init__(self, id, name, status, url=None):
super(Contact, self).__init__(id, url)
self.name = name
self.status = status
def set_photo(self, name, **kwargs):
"""
Set photo of contact.
:param name: name of photo
:type name: str
:param kwargs: See :class:`ContactPhoto` to know what other parameters you can use
"""
if name not in self.photos:
self.photos[name] = ContactPhoto(name)
photo = self.photos[name]
for key, value in kwargs.items():
setattr(photo, key, value)
def get_text(self):
def print_node(node, level=1):
result = u''
if node.flags & node.SECTION:
result += u'\t' * level + node.label + '\n'
for sub in node.value.values():
result += print_node(sub, level + 1)
else:
if isinstance(node.value, (tuple, list)):
value = ', '.join(unicode(v) for v in node.value)
elif isinstance(node.value, float):
value = '%.2f' % node.value
else:
value = node.value
result += u'\t' * level + u'%-20s %s\n' % (node.label + ':', value)
return result
result = u'Nickname: %s\n' % self.name
if self.status & Contact.STATUS_ONLINE:
s = 'online'
elif self.status & Contact.STATUS_OFFLINE:
s = 'offline'
elif self.status & Contact.STATUS_AWAY:
s = 'away'
else:
s = 'unknown'
result += u'Status: %s (%s)\n' % (s, self.status_msg)
result += u'URL: %s\n' % self.url
result += u'Photos:\n'
for name, photo in self.photos.items():
result += u'\t%s%s\n' % (photo, ' (hidden)' if photo.hidden else '')
result += u'\nProfile:\n'
for head in self.profile.values():
result += print_node(head)
result += u'Description:\n'
for s in self.summary.split('\n'):
result += u'\t%s\n' % s
return result
class QueryError(UserError):
"""
Raised when unable to send a query to a contact.
"""
class Query(BaseObject):
"""
Query to send to a contact.
"""
message = StringField('Message received')
def __init__(self, id, message, url=None):
super(Query, self).__init__(id, url)
self.message = message
class CapContact(Capability):
def iter_contacts(self, status=Contact.STATUS_ALL, ids=None):
"""
Iter contacts
:param status: get only contacts with the specified status
:type status: Contact.STATUS_*
:param ids: if set, get the specified contacts
:type ids: list[str]
:rtype: iter[:class:`Contact`]
"""
raise NotImplementedError()
def get_contact(self, id):
"""
Get a contact from his id.
The default implementation only calls iter_contacts()
with the proper values, but it might be overloaded
by backends.
:param id: the ID requested
:type id: str
:rtype: :class:`Contact` or None if not found
"""
l = self.iter_contacts(ids=[id])
try:
return l[0]
except IndexError:
return None
def send_query(self, id):
"""
Send a query to a contact
:param id: the ID of contact
:type id: str
:rtype: :class:`Query`
:raises: :class:`QueryError`
"""
raise NotImplementedError()
def get_notes(self, id):
"""
Get personal notes about a contact
:param id: the ID of the contact
:type id: str
:rtype: unicode
"""
raise NotImplementedError()
def save_notes(self, id, notes):
"""
Set personal notes about a contact
:param id: the ID of the contact
:type id: str
:returns: the unicode object to save as notes
"""
raise NotImplementedError()
class PhysicalEntity(BaseContact):
"""
Contact which has a physical address.
"""
country = StringField('Country')
postcode = StringField('Post code')
city = StringField('City')
address = StringField('Address of the contact')
address_notes = StringField('Extra address info')
class Person(PhysicalEntity):
pass
class OpeningHours(BaseObject):
"""
Definition of times when a place is open or closed.
Consists in a list of :class:`OpeningRule`.
Rules should be ordered by priority.
If no rule matches the given date, it is considered closed by default.
"""
rules = Field('Rules of opening/closing', list)
def is_open_at(self, query):
for rule in self.rules:
if query in rule:
return rule.is_open
return False
@property
def is_open_now(self):
return self.is_open_at(datetime.now())
class RRuleField(Field):
def __init__(self, doc, **kargs):
super(RRuleField, self).__init__(doc, rrule.rrulebase)
def convert(self, v):
if isinstance(v, basestring):
return rrule.rrulestr(v)
return v
class OpeningRule(BaseObject):
"""
Single rule defining a (recurrent) time interval when a place is open or closed.
"""
dates = RRuleField('Dates on which this rule applies')
times = Field('Times of the day this rule applies', list)
is_open = BoolField('Is it an opening rule or closing rule?')
def __contains__(self, dt):
date = dt.date()
time = dt.time()
# check times before dates because there are probably fewer entries
for start, end in self.times:
if start <= time <= end:
break
else:
return False
# can't use "date in self.dates" because rrule only matches datetimes
for occ in self.dates:
occ = occ.date()
if occ > date:
break
elif occ == date:
return True
return False
class Place(PhysicalEntity):
opening = Field('Opening hours', OpeningHours)
class SearchQuery(BaseObject):
"""
Parameters to search for contacts.
"""
name = StringField('Name to search for')
address = StringField('Address where to look') # optional fields
city = StringField('City where to look')
class CapDirectory(Capability):
def search_contacts(self, query, sortby):
"""
Search contacts matching a query.
:param query: search parameters
:type query: :class:`SearchQuery`
:rtype: iter[:class:`PhysicalEntity`]
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/content.py 0000664 0000000 0000000 00000005227 13414137563 0030650 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2011 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 .
from .base import Capability, BaseObject, BoolField, StringField
from .date import DateField
__all__ = ['Content', 'Revision', 'CapContent']
class Content(BaseObject):
"""
Content object.
"""
title = StringField('Title of content')
author = StringField('Original author of content')
content = StringField('Body')
revision = StringField('ID of revision')
class Revision(BaseObject):
"""
Revision of a change on a content.
"""
author = StringField('Author of revision')
comment = StringField('Comment log about revision')
timestamp = DateField('Date of revision')
minor = BoolField('Is this change minor?')
class CapContent(Capability):
def get_content(self, id, revision=None):
"""
Get a content from an ID.
:param id: ID of content
:type id: str
:param revision: if given, get the content at this revision
:type revision: :class:`Revision`
:rtype: :class:`Content`
"""
raise NotImplementedError()
def iter_revisions(self, id):
"""
Iter revisions of a content.
:param id: id of content
:type id: str
:rtype: iter[:class:`Revision`]
"""
raise NotImplementedError()
def push_content(self, content, message=None, minor=False):
"""
Push a new revision of a content.
:param content: object to push
:type content: :class:`Content`
:param message: log message to associate to new revision
:type message: str
:param minor: this is a minor revision
:type minor: bool
"""
raise NotImplementedError()
def get_content_preview(self, content):
"""
Get a HTML preview of a content.
:param content: content object
:type content: :class:`Content`
:rtype: str
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/date.py 0000664 0000000 0000000 00000004356 13414137563 0030115 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2013 Christophe Benz, Romain Bignon, Julien Hebert
#
# 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 .
import datetime
from weboob.tools.date import new_date, new_datetime
from weboob.capabilities.base import Field
from weboob.tools.compat import long
__all__ = ['DateField', 'TimeField', 'DeltaField']
class DateField(Field):
"""
A field which accepts only :class:`datetime.date` and :class:`datetime.datetime` types.
"""
def __init__(self, doc, **kwargs):
super(DateField, self).__init__(doc, datetime.date, datetime.datetime, **kwargs)
def __setattr__(self, name, value):
if name == 'value':
# Force use of our date and datetime types, to fix bugs in python2
# with strftime on year<1900.
if type(value) is datetime.datetime:
value = new_datetime(value)
if type(value) is datetime.date:
value = new_date(value)
return object.__setattr__(self, name, value)
class TimeField(Field):
"""
A field which accepts only :class:`datetime.time` and :class:`datetime.time` types.
"""
def __init__(self, doc, **kwargs):
super(TimeField, self).__init__(doc, datetime.time, datetime.datetime, **kwargs)
class DeltaField(Field):
"""
A field which accepts only :class:`datetime.timedelta` type.
"""
def __init__(self, doc, **kwargs):
super(DeltaField, self).__init__(doc, datetime.timedelta, **kwargs)
def convert(self, value):
if isinstance(value, (int, long)):
value = datetime.timedelta(seconds=value)
return value
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/dating.py 0000664 0000000 0000000 00000007464 13414137563 0030451 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-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 .
from .base import Capability, BaseObject, Field, StringField, UserError
from .date import DateField
from .contact import Contact
__all__ = ['OptimizationNotFound', 'Optimization', 'Event', 'CapDating']
class OptimizationNotFound(UserError):
"""
Raised when an optimization is not found.
"""
class Optimization(BaseObject):
"""
Optimization.
:var CONFIG: Configuration of optim can be made by
:class:`weboob.tools.value.Value` objects
in this dict.
"""
CONFIG = {}
def start(self):
"""
Start optimization.
"""
raise NotImplementedError()
def stop(self):
"""
Stop optimization.
"""
raise NotImplementedError()
def is_running(self):
"""
Know if the optimization is currently running.
:rtype: bool
"""
raise NotImplementedError()
def get_config(self):
"""
Get config of this optimization.
:rtype: dict
"""
return None
def set_config(self, params):
"""
Set config of this optimization.
:param params: parameters
:type params: dict
"""
raise NotImplementedError()
class Event(BaseObject):
"""
A dating event (for example a visite, a query received, etc.)
"""
date = DateField('Date of event')
contact = Field('Contact related to this event', Contact)
type = StringField('Type of event')
message = StringField('Message of the event')
class CapDating(Capability):
"""
Capability for dating websites.
"""
def init_optimizations(self):
"""
Initialization of optimizations.
"""
raise NotImplementedError()
def add_optimization(self, name, optim):
"""
Add an optimization.
:param name: name of optimization
:type name: str
:param optim: optimization
:type optim: :class:`Optimization`
"""
optim.id = name
setattr(self, 'OPTIM_%s' % name, optim)
def iter_optimizations(self):
"""
Iter optimizations.
:rtype: iter[:class:`Optimization`]
"""
for attr_name in dir(self):
if not attr_name.startswith('OPTIM_'):
continue
attr = getattr(self, attr_name)
if attr is None:
continue
yield attr
def get_optimization(self, optim):
"""
Get an optimization from a name.
:param optim: name of optimization
:type optim: str
:rtype: :class:`Optimization`
"""
optim = optim.upper()
if not hasattr(self, 'OPTIM_%s' % optim):
raise OptimizationNotFound()
return getattr(self, 'OPTIM_%s' % optim)
def iter_events(self):
"""
Iter events.
:rtype: iter[:class:`Event`]
"""
raise NotImplementedError()
def iter_new_contacts(self):
"""
Iter new contacts.
:rtype: iter[:class:`Contact`]
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/file.py 0000664 0000000 0000000 00000006267 13414137563 0030122 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2013 Pierre Mazière
#
# 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 .
from weboob.tools.compat import long
from .base import Capability, BaseObject, NotAvailable, Field, StringField, IntField, Enum
from .date import DateField
__all__ = ['BaseFile', 'CapFile']
class LICENSES(Enum):
OTHER = u'Other license'
PD = u'Public Domain'
COPYRIGHT = u'All rights reserved'
CCBY = u'Creative Commons BY'
CCBYSA = u'Creative Commons BY-SA'
CCBYNC = u'Creative Commons BY-NC'
CCBYND = u'Creative Commons BY-ND'
CCBYNCSA = u'Creative Commons BY-NC-SA'
CCBYNCND = u'Creative Commons BY-NC-ND'
GFDL = u'GNU Free Documentation License'
class BaseFile(BaseObject):
"""
Represent a file.
"""
title = StringField('File title')
ext = StringField('File extension')
author = StringField('File author')
description = StringField('File description')
date = DateField('File publication date')
size = IntField('File size in bytes', default=NotAvailable)
rating = Field('Rating', int, long, float, default=NotAvailable)
rating_max = Field('Maximum rating', int, long, float, default=NotAvailable)
license = StringField('License name')
def __unicode__(self):
return self.url or u''
def __repr__(self):
return '<%s title=%r url=%r>' % (type(self).__name__, self.title, self.url)
@classmethod
def id2url(cls, _id):
"""
Overloaded in child classes provided by backends.
"""
raise NotImplementedError()
@property
def page_url(self):
"""
Get file page URL
"""
return self.id2url(self.id)
class SearchSort(Enum):
RELEVANCE = 0
RATING = 1
VIEWS = 2
DATE = 3
class CapFile(Capability):
"""
Provide file download
"""
SEARCH_RELEVANCE = SearchSort.RELEVANCE
SEARCH_RATING = SearchSort.RATING
SEARCH_VIEWS = SearchSort.VIEWS
SEARCH_DATE = SearchSort.DATE
def search_file(self, pattern, sortby=SEARCH_RELEVANCE):
"""
:param pattern: pattern to search on
:type pattern: str
:param sortby: sort by ... (user SEARCH_* constants)
:rtype: iter[:class:`BaseFile`]
"""
raise NotImplementedError()
def get_file(self, _id):
"""
Get a file from an ID
:param _id: the file id. I can be a numeric ID, or a page url
:type _id: str
:rtype: :class:`BaseFile` or None if not found.
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/gallery.py 0000664 0000000 0000000 00000010364 13414137563 0030633 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2011 Romain Bignon, Christophe Benz, Noé Rubinstein
#
# 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 .
from weboob.capabilities.image import BaseImage as CIBaseImage, Thumbnail
from weboob.tools.compat import unicode
from .base import Capability, BaseObject, NotLoaded, Field, StringField, \
IntField, FloatField, Enum
from .date import DateField
__all__ = ['BaseGallery', 'BaseImage', 'CapGallery']
class BaseGallery(BaseObject):
"""
Represents a gallery.
This object has to be inherited to specify how to calculate the URL of the gallery from its ID.
"""
title = StringField('Title of gallery')
description = StringField('Description of gallery')
cardinality = IntField('Cardinality of gallery')
date = DateField('Date of gallery')
rating = FloatField('Rating of this gallery')
rating_max = FloatField('Max rating available')
thumbnail = Field('Thumbnail', Thumbnail)
def __init__(self, _id, title=NotLoaded, url=NotLoaded, cardinality=NotLoaded, date=NotLoaded,
rating=NotLoaded, rating_max=NotLoaded, thumbnail=NotLoaded, thumbnail_url=None, nsfw=False):
super(BaseGallery, self).__init__(unicode(_id), url)
self.title = title
self.date = date
self.rating = rating
self.rating_max = rating_max
self.thumbnail = thumbnail
@classmethod
def id2url(cls, _id):
"""Overloaded in child classes provided by backends."""
raise NotImplementedError()
@property
def page_url(self):
"""
Get URL to page of this gallery.
"""
return self.id2url(self.id)
def iter_image(self):
"""
Iter images.
"""
raise NotImplementedError()
class BaseImage(CIBaseImage):
"""
Base class for images.
"""
index = IntField('Usually page number')
gallery = Field('Reference to the Gallery object', BaseGallery)
def __init__(self, _id=u'', index=None, thumbnail=NotLoaded, url=NotLoaded,
ext=NotLoaded, gallery=None):
super(BaseImage, self).__init__(unicode(_id), url)
self.index = index
self.thumbnail = thumbnail
self.ext = ext
self.gallery = gallery
def __unicode__(self):
return self.url
def __repr__(self):
return '' % self.url
def __iscomplete__(self):
return self.data is not NotLoaded
class SearchSort(Enum):
RELEVANCE = 0
RATING = 1
VIEWS = 2
DATE = 3
class CapGallery(Capability):
"""
This capability represents the ability for a website backend to provide videos.
"""
SEARCH_RELEVANCE = SearchSort.RELEVANCE
SEARCH_RATING = SearchSort.RATING
SEARCH_VIEWS = SearchSort.VIEWS
SEARCH_DATE = SearchSort.DATE
def search_galleries(self, pattern, sortby=SEARCH_RELEVANCE):
"""
Iter results of a search on a pattern.
:param pattern: pattern to search on
:type pattern: str
:param sortby: sort by...
:type sortby: SEARCH_*
:rtype: :class:`BaseGallery`
"""
raise NotImplementedError()
def get_gallery(self, _id):
"""
Get gallery from an ID.
:param _id: the gallery id. It can be a numeric ID, or a page url, or so.
:type _id: str
:rtype: :class:`Gallery`
"""
raise NotImplementedError()
def iter_gallery_images(self, gallery):
"""
Iterate images from a Gallery.
:type gallery: BaseGallery
:rtype: iter(BaseImage)
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/gauge.py 0000664 0000000 0000000 00000006545 13414137563 0030272 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2012 Romain Bignon, Florent Fourcot
#
# 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 .
from .base import Capability, BaseObject, StringField, DecimalField, Field, UserError, empty
from .date import DateField
__all__ = ['Gauge', 'GaugeSensor', 'GaugeMeasure', 'CapGauge', 'SensorNotFound']
class SensorNotFound(UserError):
"""
Not found a sensor
"""
class Gauge(BaseObject):
"""
Gauge class.
"""
name = StringField('Name of gauge')
city = StringField('City of the gauge')
object = StringField('What is evaluate') # For example, name of a river
sensors = Field('List of sensors on the gauge', list)
class GaugeMeasure(BaseObject):
"""
Measure of a gauge sensor.
"""
level = DecimalField('Level of measure')
date = DateField('Date of measure')
alarm = StringField('Alarm level')
def __repr__(self):
if empty(self.level):
return "" % self.level
else:
return "" % (self.level, self.alarm, self.date)
class GaugeSensor(BaseObject):
"""
GaugeSensor class.
"""
name = StringField('Name of the sensor')
unit = StringField('Unit of values')
forecast = StringField('Forecast')
address = StringField('Address')
latitude = DecimalField('Latitude')
longitude = DecimalField('Longitude')
lastvalue = Field('Last value', GaugeMeasure)
history = Field('Value history', list) # lastvalue not included
gaugeid = StringField('Id of the gauge')
def __repr__(self):
return "" % (self.id, self.name)
class CapGauge(Capability):
def iter_gauges(self, pattern=None):
"""
Iter gauges.
:param pattern: if specified, used to search gauges.
:type pattern: str
:rtype: iter[:class:`Gauge`]
"""
raise NotImplementedError()
def iter_sensors(self, id, pattern=None):
"""
Iter instrument of a gauge.
:param: ID of the gauge
:param pattern: if specified, used to search sensors.
:type pattern: str
:rtype: iter[:class:`GaugeSensor`]
"""
raise NotImplementedError()
def iter_gauge_history(self, id):
"""
Get history of a gauge sensor.
:param id: ID of the gauge sensor
:type id: str
:rtype: iter[:class:`GaugeMeasure`]
"""
raise NotImplementedError()
def get_last_measure(self, id):
"""
Get last measures of a sensor.
:param id: ID of the sensor.
:type id: str
:rtype: :class:`GaugeMeasure`
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/geolocip.py 0000664 0000000 0000000 00000003264 13414137563 0030776 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2011 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 .
from .base import Capability, BaseObject, StringField, FloatField
__all__ = ['IpLocation', 'CapGeolocIp']
class IpLocation(BaseObject):
"""
Represents the location of an IP address.
"""
city = StringField('City')
region = StringField('Region')
zipcode = StringField('Zip code')
country = StringField('Country')
lt = FloatField('Latitude')
lg = FloatField('Longitude')
osmlink = StringField('Link to OpenStreetMap location page')
host = StringField('Hostname')
tld = StringField('Top Level Domain')
isp = StringField('Internet Service Provider')
class CapGeolocIp(Capability):
"""
Access information about IP addresses database.
"""
def get_location(self, ipaddr):
"""
Get location of an IP address.
:param ipaddr: IP address
:type ipaddr: str
:rtype: :class:`IpLocation`
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/housing.py 0000664 0000000 0000000 00000012303 13414137563 0030643 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2012 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 .
from .base import Capability, BaseObject, Field, IntField, DecimalField, \
StringField, BytesField, Enum, EnumField, UserError
from .date import DateField
__all__ = [
'CapHousing', 'Housing', 'Query', 'City', 'UTILITIES', 'ENERGY_CLASS',
'POSTS_TYPES', 'ADVERT_TYPES', 'HOUSE_TYPES', 'TypeNotSupported',
'HousingPhoto',
]
class TypeNotSupported(UserError):
"""
Raised when query type is not supported
"""
def __init__(self,
msg='This type of house is not supported by this module'):
super(TypeNotSupported, self).__init__(msg)
class HousingPhoto(BaseObject):
"""
Photo of a housing.
"""
data = BytesField('Data of photo')
def __init__(self, url):
super(HousingPhoto, self).__init__(url.split('/')[-1], url)
def __iscomplete__(self):
return self.data
def __unicode__(self):
return self.url
def __repr__(self):
return '' % (self.id, len(self.data) if self.data else 0, self.url)
class UTILITIES(Enum):
INCLUDED = u'C.C.'
EXCLUDED = u'H.C.'
UNKNOWN = u''
class ENERGY_CLASS(Enum):
A = u'A'
B = u'B'
C = u'C'
D = u'D'
E = u'E'
F = u'F'
G = u'G'
class POSTS_TYPES(Enum):
RENT=u'RENT'
SALE = u'SALE'
SHARING = u'SHARING'
FURNISHED_RENT = u'FURNISHED_RENT'
VIAGER = u'VIAGER'
class ADVERT_TYPES(Enum):
PROFESSIONAL = u'Professional'
PERSONAL = u'Personal'
class HOUSE_TYPES(Enum):
APART = u'Apartment'
HOUSE = u'House'
PARKING = u'Parking'
LAND = u'Land'
OTHER = u'Other'
UNKNOWN = u'Unknown'
class Housing(BaseObject):
"""
Content of a housing.
"""
type = EnumField('Type of housing (rent, sale, sharing)',
POSTS_TYPES)
advert_type = EnumField('Type of advert (professional or personal)',
ADVERT_TYPES)
house_type = EnumField(u'Type of house (apartment, house, parking, …)',
HOUSE_TYPES)
title = StringField('Title of housing')
area = DecimalField('Area of housing, in m2')
cost = DecimalField('Cost of housing')
price_per_meter = DecimalField('Price per meter ratio')
currency = StringField('Currency of cost')
utilities = EnumField('Utilities included or not', UTILITIES)
date = DateField('Date when the housing has been published')
location = StringField('Location of housing')
station = StringField('What metro/bus station next to housing')
text = StringField('Text of the housing')
phone = StringField('Phone number to contact')
photos = Field('List of photos', list)
rooms = DecimalField('Number of rooms')
bedrooms = DecimalField('Number of bedrooms')
details = Field('Key/values of details', dict)
DPE = EnumField('DPE (Energy Performance Certificate)', ENERGY_CLASS)
GES = EnumField('GES (Greenhouse Gas Emissions)', ENERGY_CLASS)
class Query(BaseObject):
"""
Query to find housings.
"""
type = EnumField('Type of housing to find (POSTS_TYPES constants)',
POSTS_TYPES)
cities = Field('List of cities to search in', list, tuple)
area_min = IntField('Minimal area (in m2)')
area_max = IntField('Maximal area (in m2)')
cost_min = IntField('Minimal cost')
cost_max = IntField('Maximal cost')
nb_rooms = IntField('Number of rooms')
house_types = Field('List of house types', list, tuple, default=list(HOUSE_TYPES))
advert_types = Field('List of advert types to filter on', list, tuple,
default=list(ADVERT_TYPES))
class City(BaseObject):
"""
City.
"""
name = StringField('Name of city')
class CapHousing(Capability):
"""
Capability of websites to search housings.
"""
def search_housings(self, query):
"""
Search housings.
:param query: search query
:type query: :class:`Query`
:rtype: iter[:class:`Housing`]
"""
raise NotImplementedError()
def get_housing(self, housing):
"""
Get an housing from an ID.
:param housing: ID of the housing
:type housing: str
:rtype: :class:`Housing` or None if not found.
"""
raise NotImplementedError()
def search_city(self, pattern):
"""
Search a city from a pattern.
:param pattern: pattern to search
:type pattern: str
:rtype: iter[:class:`City`]
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/image.py 0000664 0000000 0000000 00000005706 13414137563 0030262 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2011 Romain Bignon, Christophe Benz, Noé Rubinstein
# Copyright(C) 2013 Pierre Mazière
#
# 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 .
from collections import OrderedDict
from .base import NotLoaded, Field, BoolField, BytesField
from .file import CapFile, BaseFile
__all__ = ['BaseImage', 'Thumbnail', 'CapImage']
class _BaseImage(BaseFile):
"""
Fake class to allow the inclusion of a BaseImage property within
the real BaseImage class
"""
pass
class Thumbnail(_BaseImage):
"""
Thumbnail of an image.
"""
data = BytesField('Data')
def __init__(self, url):
super(Thumbnail, self).__init__(url)
self.url = url.replace(u' ', u'%20')
def __unicode__(self):
return self.url
def __repr__(self):
return '' % self.url
def __iscomplete__(self):
return self.data is not NotLoaded
class BaseImage(_BaseImage):
"""
Represents an image file.
"""
nsfw = BoolField('Is this Not Safe For Work', default=False)
thumbnail = Field('Thumbnail of the image', Thumbnail)
data = BytesField('Data of image')
def __iscomplete__(self):
return self.data is not NotLoaded
def to_dict(self):
def iter_decorate(d):
for key, value in d:
if key == 'data':
continue
if key == 'id' and self.backend is not None:
value = self.fullid
yield key, value
fields_iterator = self.iter_fields()
return OrderedDict(iter_decorate(fields_iterator))
class CapImage(CapFile):
"""
Image file provider
"""
def search_image(self, pattern, sortby=CapFile.SEARCH_RELEVANCE, nsfw=False):
"""
search for an image file
:param pattern: pattern to search on
:type pattern: str
:param sortby: sort by ...(use SEARCH_* constants)
:param nsfw: include non-suitable for work images if True
:type nsfw: bool
:rtype: iter[:class:`BaseImage`]
"""
return self.search_file(pattern, sortby)
def get_image(self, _id):
"""
Get an image file from an ID.
:param id: image file ID
:type id: str
:rtype: :class:`BaseImage`]
"""
return self.get_file(_id)
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/job.py 0000664 0000000 0000000 00000006616 13414137563 0027753 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2013 Bezleputh
#
# 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 .
from .base import BaseObject, Capability, StringField
from .date import DateField
__all__ = ['BaseJobAdvert', 'CapJob']
class BaseJobAdvert(BaseObject):
"""
Represents a job announce.
"""
publication_date = DateField('Date when the announce has been published')
society_name = StringField('Name of the society taht published the announce')
place = StringField('Place where the job take place')
job_name = StringField('Name of the job')
title = StringField('Title of the announce')
contract_type = StringField('Type of the contrat : CDI, CDD')
pay = StringField('Amount of the salary')
description = StringField('Description of the job')
formation = StringField('Required formation')
experience = StringField('Required experience')
def __unicode__(self):
message = u'\r\n-- Advert --\r\n'
message += u'id : %s\r\n' % self.id
message += u'url : %s\r\n' % self.url
message += u'publication_date : %s\r\n' % self.publication_date
message += u'society_name : %s\r\n' % self.society_name
message += u'place : %s\r\n' % self.place
message += u'job_name : %s\r\n' % self.job_name
message += u'title : %s\r\n' % self.title
message += u'contract_type : %s\r\n' % self.contract_type
message += u'pay : %s\r\n' % self.pay
message += u'description : %s\r\n' % self.description
message += u'formation : %s\r\n' % self.formation
message += u'experience : %s\r\n' % self.experience
return message
@classmethod
def id2url(cls, _id):
"""Overloaded in child classes provided by backends."""
raise NotImplementedError()
@property
def page_url(self):
"""
Get page URL of the announce.
"""
return self.id2url(self.id)
class CapJob(Capability):
"""
Capability of job annouce websites.
"""
def search_job(self, pattern=None):
"""
Iter results of a search on a pattern.
:param pattern: pattern to search on
:type pattern: str
:rtype: iter[:class:`BaseJobAdvert`]
"""
raise NotImplementedError()
def advanced_search_job(self):
"""
Iter results of an advanced search
:rtype: iter[:class:`BaseJobAdvert`]
"""
def get_job_advert(self, _id, advert=None):
"""
Get an announce from an ID.
:param _id: id of the advert
:type _id: str
:param advert: the advert
:type advert: BaseJobAdvert
:rtype: :class:`BaseJobAdvert` or None if not found.
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/library.py 0000664 0000000 0000000 00000004243 13414137563 0030637 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2012 Jeremy Monnet
#
# 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 .
from .collection import CapCollection
from .base import BaseObject, BoolField, StringField
from .date import DateField
__all__ = ['Book', 'Renew', 'CapBook']
class Book(BaseObject):
"""
Describes a book.
"""
name = StringField('Name of the book')
author = StringField('Author of the book')
location = StringField('Location')
date = DateField('The due date')
late = BoolField('Are you late?')
class Renew(BaseObject):
"""
A renew message.
"""
message = StringField('Message')
class CapBook(CapCollection):
"""
Library websites.
"""
def iter_resources(self, objs, split_path):
"""
Iter resources. It retuns :func:`iter_books`.
"""
if Book in objs:
self._restrict_level(split_path)
return self.iter_books()
def iter_books(self):
"""
Iter books.
:rtype: iter[:class:`Book`]
"""
raise NotImplementedError()
def get_book(self, _id):
"""
Get a book from an ID.
:param _id: ID of the book
:type _id: str
:rtype: :class:`Book`
"""
raise NotImplementedError()
def iter_booked(self):
raise NotImplementedError()
def renew_book(self, _id):
raise NotImplementedError()
def iter_rented(self):
raise NotImplementedError()
def search_books(self, _string):
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/lyrics.py 0000664 0000000 0000000 00000003221 13414137563 0030473 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2013
#
# 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 .
from .base import BaseObject, Capability, StringField
__all__ = ['SongLyrics', 'CapLyrics']
class SongLyrics(BaseObject):
"""
Song lyrics object.
"""
title = StringField('Title of the song')
artist = StringField('Artist of the song')
content = StringField('Lyrics of the song')
class CapLyrics(Capability):
"""
Lyrics websites.
"""
def iter_lyrics(self, criteria, pattern):
"""
Search lyrics by artist or by song
and iterate on results.
:param criteria: 'artist' or 'song'
:type criteria: str
:param pattern: pattern to search
:type pattern: str
:rtype: iter[:class:`SongLyrics`]
"""
raise NotImplementedError()
def get_lyrics(self, _id):
"""
Get a lyrics object from an ID.
:param _id: ID of lyrics
:type _id: str
:rtype: :class:`SongLyrics`
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/messages.py 0000664 0000000 0000000 00000014306 13414137563 0031003 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2015 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 .
import datetime
import time
from weboob.tools.compat import unicode
from .base import Capability, BaseObject, NotLoaded, Field, StringField, \
IntField, UserError
from .date import DateField
__all__ = ['Thread', 'Message', 'CapMessages', 'CantSendMessage', 'CapMessagesPost']
class Message(BaseObject):
"""
Represents a message read or to send.
"""
IS_HTML = 0x001
"The content is HTML formatted"
IS_UNREAD = 0x002
"The message is unread"
IS_RECEIVED = 0x004
"The receiver has read this message"
IS_NOT_RECEIVED = 0x008
"The receiver hass not read this message"
thread = Field('Reference to the thread', 'Thread')
title = StringField('Title of message')
sender = StringField('Author of this message')
receivers = Field('Receivers of the message', list)
date = DateField('Date when the message has been sent')
content = StringField('Body of message')
signature = StringField('Optional signature')
parent = Field('Parent message', 'Message')
children = Field('Children fields', list)
flags = IntField('Flags (IS_* constants)', default=0)
def __init__(self, thread=NotLoaded,
id=NotLoaded,
title=NotLoaded,
sender=NotLoaded,
receivers=NotLoaded,
date=None,
parent=NotLoaded,
content=NotLoaded,
signature=NotLoaded,
children=NotLoaded,
flags=0,
url=None):
super(Message, self).__init__(id, url)
self.thread = thread
self.title = title
self.sender = sender
self.receivers = receivers
self.content = content
self.signature = signature
self.children = children
self.flags = flags
if date is None:
date = datetime.datetime.utcnow()
self.date = date
if isinstance(parent, Message):
self.parent = parent
else:
self.parent = NotLoaded
self._parent_id = parent
@property
def date_int(self):
"""
Date of message as an integer.
"""
return int(time.strftime('%Y%m%d%H%M%S', self.date.timetuple()))
@property
def full_id(self):
"""
Full ID of message (in form '**THREAD_ID.MESSAGE_ID**')
"""
return '%s.%s' % (self.thread.id, self.id)
@property
def full_parent_id(self):
"""
Get the full ID of the parent message (in form '**THREAD_ID.MESSAGE_ID**').
"""
if self.parent:
return self.parent.full_id
elif self._parent_id is None:
return ''
elif self._parent_id is NotLoaded:
return NotLoaded
else:
return '%s.%s' % (self.thread.id, self._parent_id)
def __eq__(self, msg):
if not isinstance(msg, Message):
return False
if self.thread:
return unicode(self.thread.id) == unicode(msg.thread.id) and \
unicode(self.id) == unicode(msg.id)
else:
return unicode(self.id) == unicode(msg.id)
def __repr__(self):
return '' % (
self.full_id, self.title, self.date, self.sender)
class Thread(BaseObject):
"""
Thread containing messages.
"""
IS_THREADS = 0x001
IS_DISCUSSION = 0x002
root = Field('Root message', Message)
title = StringField('Title of thread')
date = DateField('Date of thread')
flags = IntField('Flags (IS_* constants)', default=IS_THREADS)
def iter_all_messages(self):
"""
Iter all messages of the thread.
:rtype: iter[:class:`Message`]
"""
if self.root:
yield self.root
for m in self._iter_all_messages(self.root):
yield m
def _iter_all_messages(self, message):
if message.children:
for child in message.children:
yield child
for m in self._iter_all_messages(child):
yield m
class CapMessages(Capability):
"""
Capability to read messages.
"""
def iter_threads(self):
"""
Iterates on threads, from newers to olders.
:rtype: iter[:class:`Thread`]
"""
raise NotImplementedError()
def get_thread(self, id):
"""
Get a specific thread.
:rtype: :class:`Thread`
"""
raise NotImplementedError()
def iter_unread_messages(self):
"""
Iterates on messages which hasn't been marked as read.
:rtype: iter[:class:`Message`]
"""
raise NotImplementedError()
def set_message_read(self, message):
"""
Set a message as read.
:param message: message read (or ID)
:type message: :class:`Message` or str
"""
raise NotImplementedError()
class CantSendMessage(UserError):
"""
Raised when a message can't be send.
"""
class CapMessagesPost(Capability):
"""
This capability allow user to send a message.
"""
def post_message(self, message):
"""
Post a message.
:param message: message to send
:type message: :class:`Message`
:raises: :class:`CantSendMessage`
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/parcel.py 0000664 0000000 0000000 00000004155 13414137563 0030443 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2013 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 .
from .base import Capability, BaseObject, Field, StringField, UserError, Enum
from .date import DateField
class Event(BaseObject):
date = DateField('Date')
activity = StringField('Activity')
location = StringField('Location')
def __repr__(self):
return '' % (self.date, self.activity, self.location)
class ParcelState(Enum):
UNKNOWN = 0
PLANNED = 1
IN_TRANSIT = 2
ARRIVED = 3
class Parcel(BaseObject):
STATUS_UNKNOWN = ParcelState.UNKNOWN
STATUS_PLANNED = ParcelState.PLANNED
STATUS_IN_TRANSIT = ParcelState.IN_TRANSIT
STATUS_ARRIVED = ParcelState.ARRIVED
arrival = DateField('Scheduled arrival date')
status = Field('Status of parcel', int, default=STATUS_UNKNOWN)
info = StringField('Information about parcel status')
history = Field('History', list)
class CapParcel(Capability):
def get_parcel_tracking(self, id):
"""
Get information abouut a parcel.
:param id: ID of the parcel
:type id: :class:`str`
:rtype: :class:`Parcel`
:raises: :class:`ParcelNotFound`
"""
raise NotImplementedError()
class ParcelNotFound(UserError):
"""
Raised when a parcell is not found.
It can be an user error, or an expired parcel
"""
def __init__(self, msg='Account not found'):
super(ParcelNotFound, self).__init__(msg)
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/paste.py 0000664 0000000 0000000 00000007215 13414137563 0030311 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2011 Laurent Bachelier
#
# 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 .
from weboob.tools.compat import unicode
from .base import Capability, BaseObject, NotLoaded, StringField, BoolField, UserError
__all__ = ['PasteNotFound', 'BasePaste', 'CapPaste']
class PasteNotFound(UserError):
"""
Raised when a paste is not found.
"""
def __init__(self):
return super(PasteNotFound, self).__init__("Paste not found")
class BasePaste(BaseObject):
"""
Represents a pasted text.
"""
title = StringField('Title of paste')
language = StringField('Language of the paste')
contents = StringField('Content of the paste')
public = BoolField('Is this paste public?')
def __init__(self, _id, title=NotLoaded, language=NotLoaded, contents=NotLoaded,
public=NotLoaded, url=None):
super(BasePaste, self).__init__(unicode(_id), url)
self.title = title
self.language = language
self.contents = contents
self.public = public
@classmethod
def id2url(cls, _id):
"""Overloaded in child classes provided by backends."""
raise NotImplementedError()
@property
def page_url(self):
"""
Get URL to page of this paste.
"""
return self.id2url(self.id)
class CapPaste(Capability):
"""
This capability represents the ability for a website backend to store plain text.
"""
def new_paste(self, *args, **kwargs):
"""
Get a new paste object for posting it with the backend.
The parameters should be passed to the object init.
:rtype: :class:`BasePaste`
"""
raise NotImplementedError()
def can_post(self, contents, title=None, public=None, max_age=None):
"""
Checks if the paste can be pasted by this backend.
Some properties are considered required (public/private, max_age) while others
are just bonuses (language).
contents: Can be used to check encodability, maximum length, etc.
title: Can be used to check length, allowed characters. Should not be required.
public: True must be public, False must be private, None do not care.
max_age: Maximum time to live in seconds.
A score of 0 means the backend is not suitable.
A score of 1 means the backend is suitable.
Higher scores means it is more suitable than others with a lower score.
:rtype: int
:returns: score
"""
raise NotImplementedError()
def get_paste(self, url):
"""
Get a Paste from an ID or URL.
:param _id: the paste id. It can be an ID or a page URL.
:type _id: str
:rtype: :class:`BasePaste`
:raises: :class:`PasteNotFound`
"""
raise NotImplementedError()
def post_paste(self, paste, max_age=None):
"""
Post a paste.
:param paste: a Paste object
:type paste: :class:`BasePaste`
"""
raise NotImplementedError()
pricecomparison.py 0000664 0000000 0000000 00000005135 13414137563 0032312 0 ustar 00root root 0000000 0000000 woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities # -*- coding: utf-8 -*-
# Copyright(C) 2012 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 .
from .base import Capability, BaseObject, Field, DecimalField, \
StringField, UserError
from .date import DateField
__all__ = ['Shop', 'Price', 'Product', 'CapPriceComparison']
class PriceNotFound(UserError):
"""
Raised when a price is not found
"""
def __init__(self, msg='Price not found'):
super(PriceNotFound, self).__init__(msg)
class Product(BaseObject):
"""
A product.
"""
name = StringField('Name of product')
class Shop(BaseObject):
"""
A shop where the price is.
"""
name = StringField('Name of shop')
location = StringField('Location of the shop')
info = StringField('Information about the shop')
class Price(BaseObject):
"""
Price.
"""
date = DateField('Date when this price has been published')
cost = DecimalField('Cost of the product in this shop')
currency = StringField('Currency of the price')
message = StringField('Message related to this price')
shop = Field('Shop information', Shop)
product = Field('Product', Product)
class CapPriceComparison(Capability):
"""
Capability for price comparison websites.
"""
def search_products(self, pattern=None):
"""
Search products from a pattern.
:param pattern: pattern to search
:type pattern: str
:rtype: iter[:class:`Product`]
"""
raise NotImplementedError()
def iter_prices(self, products):
"""
Iter prices for a product.
:param product: product to search
:type product: :class:`Product`
:rtype: iter[:class:`Price`]
"""
raise NotImplementedError()
def get_price(self, id):
"""
Get a price from an ID
:param id: ID of price
:type id: str
:rtype: :class:`Price`
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/profile.py 0000664 0000000 0000000 00000005646 13414137563 0030643 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2016 Edouard Lambert
#
# 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 .
from .base import Capability, BaseObject, DecimalField, StringField, UserError
from .date import DateField
__all__ = ['Profile', 'Person', 'Company', 'CapProfile']
class ProfileMissing(UserError):
"""
Raised when profile is not accessible
"""
class Profile(BaseObject):
"""
Profile.
"""
name = StringField('Full name or company name')
address = StringField('Full Address')
country = StringField('Country of owner')
phone = StringField('Phone number')
email = StringField('Mail of owner')
main_bank = StringField('Main bank of owner')
class Person(Profile):
"""
Person.
"""
birth_date = DateField('Birth date')
nationality = StringField('Nationality of owner')
mobile = StringField('Mobile number of owner')
spouse_name = StringField('Name of spouse')
children = DecimalField('Number of dependent children')
family_situation = StringField('Family situation')
matrimonial = StringField('Matrimonial status')
housing_status = StringField('Housing status')
job = StringField('Profession')
job_start_date = DateField('Start date of current job')
job_activity_area = StringField('Activity area of company')
job_contract_type = StringField('Contract type of current job')
company_name = StringField('Name of company')
company_siren = StringField('SIREN Number of company')
socioprofessional_category = StringField('Socio-Professional Category')
class Company(Profile):
"""
Company.
"""
siren = StringField('SIREN Number')
registration_date = DateField('Registration date')
activity_area = StringField('Activity area')
class CapProfile(Capability):
def get_profile(self):
"""
Get profile.
:rtype: :class:`Person` or :class:`Company`
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/radio.py 0000664 0000000 0000000 00000003275 13414137563 0030275 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2013 Romain Bignon
# Copyright(C) 2013 Pierre Mazière
#
# 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 .
from .base import Capability, BaseObject, Field, StringField
from weboob.tools.capabilities.streaminfo import StreamInfo
__all__ = ['Radio', 'CapRadio']
class Radio(BaseObject):
"""
Radio object.
"""
title = StringField('Title of radio')
description = StringField('Description of radio')
current = Field('Current emission', StreamInfo)
streams = Field('List of streams', list)
class CapRadio(Capability):
"""
Capability of radio websites.
"""
def iter_radios_search(self, pattern):
"""
Search a radio.
:param pattern: pattern to search
:type pattern: str
:rtype: iter[:class:`Radio`]
"""
raise NotImplementedError()
def get_radio(self, id):
"""
Get a radio from an ID.
:param id: ID of radio
:type id: str
:rtype: :class:`Radio`
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/recipe.py 0000664 0000000 0000000 00000005316 13414137563 0030444 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2013 Julien Veyssier
#
# 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 .
from .base import Capability, BaseObject, StringField, IntField, Field
from .image import BaseImage
__all__ = ['CapRecipe', 'Recipe', 'Comment']
class Comment(BaseObject):
author = StringField('Author of the comment')
rate = StringField('Rating')
text = StringField('Comment')
def __unicode__(self):
result = u''
if self.author:
result += u'author: %s, ' % self.author
if self.rate:
result += u'note: %s, ' % self.rate
if self.text:
result += u'comment: %s' % self.text
return result
class Recipe(BaseObject):
"""
Recipe object.
"""
title = StringField('Title of the recipe')
author = StringField('Author name of the recipe')
picture = Field('Picture of the dish', BaseImage)
short_description = StringField('Short description of a recipe')
nb_person = Field('The recipe was made for this amount of persons', list)
preparation_time = IntField('Preparation time of the recipe in minutes')
cooking_time = IntField('Cooking time of the recipe in minutes')
ingredients = Field('Ingredient list necessary for the recipe', list)
instructions = StringField('Instruction step list of the recipe')
comments = Field('User comments about the recipe', list)
def __init__(self, id='', title=u'', url=None):
super(Recipe, self).__init__(id, url)
self.title = title
class CapRecipe(Capability):
"""
Recipe providers.
"""
def iter_recipes(self, pattern):
"""
Search recipes and iterate on results.
:param pattern: pattern to search
:type pattern: str
:rtype: iter[:class:`Recipe`]
"""
raise NotImplementedError()
def get_recipe(self, _id):
"""
Get a recipe object from an ID.
:param _id: ID of recipe
:type _id: str
:rtype: :class:`Recipe`
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/shop.py 0000664 0000000 0000000 00000007533 13414137563 0030151 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2014 Oleg Plakhotniuk
#
# 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 .
from .base import BaseObject, StringField, DecimalField, UserError
from .date import DateField
from .collection import CapCollection
__all__ = ['OrderNotFound', 'Order', 'Payment', 'Item', 'CapShop']
class OrderNotFound(UserError):
"""
Raised when an order is not found.
"""
def __init__(self, msg='Order not found'):
super(OrderNotFound, self).__init__(msg)
class Order(BaseObject):
"""
Purchase order.
"""
date = DateField('Date when the order was placed')
shipping = DecimalField('Shipping price')
discount = DecimalField('Discounts')
tax = DecimalField('Tax')
total = DecimalField('Total')
def __repr__(self):
return "" % (self.id, self.date)
class Payment(BaseObject):
"""
Payment for an order.
"""
date = DateField('The date when payment was applied')
method = StringField('Payment method; e.g. "VISA 1234"')
amount = DecimalField('Payment amount')
def __repr__(self):
return "" % \
(self.date, self.method, self.amount)
class Item(BaseObject):
"""
Purchased item within an order.
"""
label = StringField('Item label')
price = DecimalField('Item price')
def __repr__(self):
return "- " % (self.label, self.price)
class CapShop(CapCollection):
"""
Capability of online shops to see orders history.
"""
def iter_resources(self, objs, split_path):
"""
Iter resources.
Default implementation of this method is to return on top-level
all orders (by calling :func:`iter_accounts`).
:param objs: type of objects to get
:type objs: tuple[:class:`BaseObject`]
:param split_path: path to discover
:type split_path: :class:`list`
:rtype: iter[:class:`BaseObject`]
"""
if Order in objs:
self._restrict_level(split_path)
return self.iter_orders()
def get_currency(self):
"""
Get the currency this shop uses.
:rtype: :class:`str`
"""
raise NotImplementedError()
def iter_orders(self):
"""
Iter history of orders.
:rtype: iter[:class:`Order`]
"""
raise NotImplementedError()
def get_order(self, id):
"""
Get an order from its ID.
:param id: ID of the order
:type id: :class:`str`
:rtype: :class:`Order`
:raises: :class:`OrderNotFound`
"""
raise NotImplementedError()
def iter_payments(self, order):
"""
Iter payments of a specific order.
:param order: order to get payments
:type order: :class:`Order`
:rtype: iter[:class:`Payment`]
:raises: :class:`OrderNotFound`
"""
raise NotImplementedError()
def iter_items(self, order):
"""
Iter items of a specific order.
:param order: order to get items
:type order: :class:`Order`
:rtype: iter[:class:`Item`]
:raises: :class:`OrderNotFound`
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/subtitle.py 0000664 0000000 0000000 00000004406 13414137563 0031027 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2013 Julien Veyssier
#
# 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 .
from .base import Capability, BaseObject, StringField, IntField, UserError
__all__ = ['Subtitle', 'CapSubtitle']
class LanguageNotSupported(UserError):
"""
Raised when the language is not supported
"""
def __init__(self, msg='language is not supported'):
super(LanguageNotSupported, self).__init__(msg)
class Subtitle(BaseObject):
"""
Subtitle object.
"""
name = StringField('Name of subtitle')
ext = StringField('Extension of file')
nb_cd = IntField('Number of cd or files')
language = StringField('Language of the subtitle')
description=StringField('Description of corresponding video')
def __init__(self, id='', name='', url=None):
super(Subtitle, self).__init__(id, url)
self.name = name
class CapSubtitle(Capability):
"""
Subtitle providers.
"""
def iter_subtitles(self, pattern):
"""
Search subtitles and iterate on results.
:param pattern: pattern to search
:type pattern: str
:rtype: iter[:class:`Subtitle`]
"""
raise NotImplementedError()
def get_subtitle(self, _id):
"""
Get a subtitle object from an ID.
:param _id: ID of subtitle
:type _id: str
:rtype: :class:`Subtitle`
"""
raise NotImplementedError()
def get_subtitle_file(self, _id):
"""
Get the content of the subtitle file.
:param _id: ID of subtitle
:type _id: str
:rtype: str
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/torrent.py 0000664 0000000 0000000 00000004711 13414137563 0030670 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2012 Romain Bignon, Laurent Bachelier
#
# 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 .
from .base import Capability, BaseObject, Field, StringField, FloatField, \
IntField, UserError
from .date import DateField
__all__ = ['MagnetOnly', 'Torrent', 'CapTorrent']
class MagnetOnly(UserError):
"""
Raised when trying to get URL to torrent but only magnet is available.
"""
def __init__(self, magnet):
self.magnet = magnet
super(MagnetOnly, self).__init__('Only magnet URL is available')
class Torrent(BaseObject):
"""
Torrent object.
"""
name = StringField('Name of torrent')
size = FloatField('Size of torrent')
date = DateField('Date when torrent has been published')
magnet = StringField('URI of magnet')
seeders = IntField('Number of seeders')
leechers = IntField('Number of leechers')
files = Field('Files in torrent', list)
description = StringField('Description of torrent')
filename = StringField('Name of .torrent file')
class CapTorrent(Capability):
"""
Torrent trackers.
"""
def iter_torrents(self, pattern):
"""
Search torrents and iterate on results.
:param pattern: pattern to search
:type pattern: str
:rtype: iter[:class:`Torrent`]
"""
raise NotImplementedError()
def get_torrent(self, _id):
"""
Get a torrent object from an ID.
:param _id: ID of torrent
:type _id: str
:rtype: :class:`Torrent`
"""
raise NotImplementedError()
def get_torrent_file(self, _id):
"""
Get the content of the .torrent file.
:param _id: ID of torrent
:type _id: str
:rtype: bytes
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/translate.py 0000664 0000000 0000000 00000003774 13414137563 0031200 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2012 Lucien Loiseau
#
# 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 .
from .base import Capability, BaseObject, StringField, UserError
__all__ = ['TranslationFail', 'LanguageNotSupported', 'CapTranslate']
class LanguageNotSupported(UserError):
"""
Raised when the language is not supported
"""
def __init__(self, msg='language is not supported'):
super(LanguageNotSupported, self).__init__(msg)
class TranslationFail(UserError):
"""
Raised when no translation matches the given request
"""
def __init__(self, msg='No Translation Available'):
super(TranslationFail, self).__init__(msg)
class Translation(BaseObject):
"""
Translation.
"""
lang_src = StringField('Source language')
lang_dst = StringField('Destination language')
text = StringField('Translation')
class CapTranslate(Capability):
"""
Capability of online translation website to translate word or sentence
"""
def translate(self, source_language, destination_language, request):
"""
Perfom a translation.
:param source_language: language in which the request is written
:param destination_language: language to translate the request into
:param request: the sentence to be translated
:rtype: Translation
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/travel.py 0000664 0000000 0000000 00000010356 13414137563 0030472 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2011 Romain Bignon, Julien Hebert
#
# 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 .
import datetime
from .base import Capability, BaseObject, StringField, DecimalField, UserError
from .date import TimeField, DeltaField, DateField
__all__ = ['Station', 'Departure', 'RoadStep', 'RoadmapError', 'RoadmapFilters', 'CapTravel']
class Station(BaseObject):
"""
Describes a station.
"""
name = StringField('Name of station')
def __init__(self, id=None, name=None, url=None):
super(Station, self).__init__(id, url)
self.name = name
def __repr__(self):
return "" % (self.id, self.name)
class Departure(BaseObject):
"""
Describes a departure.
"""
type = StringField('Type of train')
time = TimeField('Departure time')
departure_station = StringField('Departure station')
arrival_station = StringField('Destination of the train')
arrival_time = TimeField('Arrival time')
late = TimeField('Optional late', default=datetime.time())
information = StringField('Informations')
plateform = StringField('Where the train will leave')
price = DecimalField('Price of ticket')
currency = StringField('Currency', default=None)
def __init__(self, id=None, _type=None, _time=None, url=None):
super(Departure, self).__init__(id, url)
self.type = _type
self.time = _time
def __repr__(self):
return "" % (
self.id, self.type, self.time.strftime('%H:%M'), self.departure_station, self.arrival_station)
class RoadStep(BaseObject):
"""
A step on a roadmap.
"""
line = StringField('When line')
start_time = TimeField('Start of step')
end_time = TimeField('End of step')
departure = StringField('Departure station')
arrival = StringField('Arrival station')
duration = DeltaField('Duration of this step')
class RoadmapError(UserError):
"""
Raised when the roadmap is unable to be calculated.
"""
class RoadmapFilters(BaseObject):
"""
Filters to get a roadmap.
"""
departure_time = DateField('Wanted departure time')
arrival_time = DateField('Wanted arrival time')
def __init__(self, id='', url=None):
super(RoadmapFilters, self).__init__(id, url)
class CapTravel(Capability):
"""
Travel websites.
"""
def iter_station_search(self, pattern):
"""
Iterates on search results of stations.
:param pattern: the search pattern
:type pattern: str
:rtype: iter[:class:`Station`]
"""
raise NotImplementedError()
def iter_station_departures(self, station_id, arrival_id=None, date=None):
"""
Iterate on departures.
:param station_id: the station ID
:type station_id: str
:param arrival_id: optionnal arrival station ID
:type arrival_id: str
:param date: optional date
:type date: datetime.datetime
:rtype: iter[:class:`Departure`]
"""
raise NotImplementedError()
def iter_roadmap(self, departure, arrival, filters):
"""
Get a roadmap.
:param departure: name of departure station
:type departure: str
:param arrival: name of arrival station
:type arrival: str
:param filters: filters on search
:type filters: :class:`RoadmapFilters`
:rtype: iter[:class:`RoadStep`]
"""
raise NotImplementedError()
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/video.py 0000664 0000000 0000000 00000003527 13414137563 0030305 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2013 Romain Bignon, Christophe Benz
# Copyright(C) 2013 Pierre Mazière
#
# 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 .
from .date import DeltaField
from .image import CapImage, BaseImage
__all__ = ['BaseVideo', 'CapVideo']
class BaseVideo(BaseImage):
"""
Represents a video.
This object has to be inherited to specify how to calculate the URL of the video from its ID.
"""
duration = DeltaField('file duration')
class CapVideo(CapImage):
"""
Video file provider.
"""
def search_videos(self, pattern, sortby=CapImage.SEARCH_RELEVANCE, nsfw=False):
"""
search for a video file
:param pattern: pattern to search on
:type pattern: str
:param sortby: sort by... (use SEARCH_* constants)
:param nsfw: include non-suitable for work videos if True
:type nsfw: bool
:rtype: iter[:class:`BaseVideo`]
"""
return self.search_image(pattern, sortby, nsfw)
def get_video(self, _id):
"""
Get a video file from an ID.
:param _id: video file ID
:type _id: str
:rtype: :class:`BaseVideo` or None is fot found.
"""
return self.get_image(_id)
woob-130005535d1d257a835857d4e42b2541ce490b40-weboob-capabilities/weboob/capabilities/weather.py 0000664 0000000 0000000 00000012550 13414137563 0030632 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2010-2011 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 .
from datetime import datetime, date
from weboob.tools.compat import basestring, unicode
from .base import Capability, BaseObject, Field, FloatField, \
StringField, IntField, UserError, NotLoaded, EnumField, Enum
from .date import DateField
__all__ = [
'Forecast', 'Current', 'City', 'CityNotFound', 'Temperature', 'CapWeather',
'BaseWeather', 'Direction', 'Precipitation',
]
class Direction(Enum):
S = 'South'
N = 'North'
E = 'East'
W = 'West'
SE = 'Southeast'
SW = 'Southwest'
NW = 'Northwest'
NE = 'Northeast'
SSE = 'South-Southeast'
SSW = 'South-Southwest'
NNW = 'North-Northwest'
NNE = 'North-Northeast'
ESE = 'East-Southeast'
ENE = 'East-Northeast'
WSW = 'West-Southwest'
WNW = 'West-Northwest'
# METAR keys
class Precipitation(Enum):
RA = 'Rain'
SN = 'Snow'
GR = 'Hail'
PL = 'Ice pellets'
GS = 'Small hail'
DZ = 'Drizzle'
IC = 'Ice cristals'
SG = 'Small grains'
UP = 'Unknown precipiation'
class Temperature(BaseObject):
value = FloatField('Temperature value')
unit = StringField('Input unit')
def __init__(self, value=NotLoaded, unit = u'', url=None):
super(Temperature, self).__init__(unicode(value), url)
self.value = value
if unit not in [u'C', u'F']:
unit = u''
self.unit = unit
def asfahrenheit(self):
if not self.unit:
return u'%s' % int(round(self.value))
elif self.unit == 'F':
return u'%s °F' % int(round(self.value))
else:
return u'%s °F' % int(round((self.value * 9.0 / 5.0) + 32))
def ascelsius(self):
if not self.unit:
return u'%s' % int(round(self.value))
elif self.unit == 'C':
return u'%s °C' % int(round(self.value))
else:
return u'%s °C' % int(round((self.value - 32.0) * 5.0 / 9.0))
def __repr__(self):
if self.value is not None and self.unit:
return '%r %r' % (self.value, self.unit)
return ''
class BaseWeather(BaseObject):
precipitation = EnumField('Precipitation type', Precipitation)
precipitation_probability = FloatField('Probability of precipitation (ratio)')
wind_direction = EnumField('Wind direction', Direction)
wind_speed = FloatField('Wind speed (in km/h)')
humidity = FloatField('Relative humidity (ratio)')
pressure = FloatField('Atmospheric pressure (in hPa)')
visibility = FloatField('Horizontal visibility distance (in km)')
cloud = IntField('Cloud coverage (in okta (0-8))')
class Forecast(BaseWeather):
"""
Weather forecast.
"""
date = Field('Date for the forecast', datetime, date, basestring)
low = Field('Low temperature', Temperature)
high = Field('High temperature', Temperature)
text = StringField('Comment on forecast')
def __init__(self, date=NotLoaded, low=None, high=None, text=None, unit=None, url=None):
super(Forecast, self).__init__(unicode(date), url)
self.date = date
self.low = Temperature(low, unit)
self.high = Temperature(high, unit)
self.text = text
class Current(BaseWeather):
"""
Current weather.
"""
date = DateField('Date of measure')
text = StringField('Comment about current weather')
temp = Field('Current temperature', Temperature)
def __init__(self, date=NotLoaded, temp=None, text=None, unit=None, url=None):
super(Current, self).__init__(unicode(date), url)
self.date = date
self.text = text
self.temp = Temperature(temp, unit)
class City(BaseObject):
"""
City where to find weather.
"""
name = StringField('Name of city')
def __init__(self, id='', name=None, url=None):
super(City, self).__init__(id, url)
self.name = name
class CityNotFound(UserError):
"""
Raised when a city is not found.
"""
class CapWeather(Capability):
"""
Capability for weather websites.
"""
def iter_city_search(self, pattern):
"""
Look for a city.
:param pattern: pattern to search
:type pattern: str
:rtype: iter[:class:`City`]
"""
raise NotImplementedError()
def get_current(self, city_id):
"""
Get current weather.
:param city_id: ID of the city
:rtype: :class:`Current`
"""
raise NotImplementedError()
def iter_forecast(self, city_id):
"""
Iter forecasts of a city.
:param city_id: ID of the city
:rtype: iter[:class:`Forecast`]
"""
raise NotImplementedError()