Last active
April 3, 2020 12:03
-
-
Save jev-odoo/04d5da155b7c267e118dd21662033f82 to your computer and use it in GitHub Desktop.
POS - Change Cash Basis Taxes On pos_order_lines
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # FOR VERSION 12 | |
| # Set sessions ids between the brackets, separated by a coma | |
| # I.E. SESSION_ID = [12, 18, 132] or SESSION_ID = [4] | |
| SESSION_IDS = [record.id] | |
| # DO NOT MODIFY | |
| VALID_VERSIONS = (12,) | |
| VALID_SESSION_STATES = ('opened', 'closing_control') | |
| LOG_LINE_ACTION_NAME = 'POS - Change Cash Basis Taxes' | |
| LOG_LINE_NAME = 'Support Intervention' | |
| ON_SUCCESS_LOG_NOTE = 'This session had some taxes due on payment. For performance purpose, they have been replace by the same taxes due on invoice.' | |
| SAVE_LOGS = True # Save logs in ir.logging table accessible from technical menu | |
| RAISE_LOGS = True # Show logs in a popup after the process | |
| PARTIAL_LOGS = [] | |
| FULL_LOGS = [] | |
| INFO = 'Info' | |
| WARNING = 'Warning' | |
| ERROR = 'Error' | |
| TAXES = {} | |
| # -----===== HELPERS =====------ | |
| def _add_log_line(level, log_message, log_details: list = None): | |
| if log_details: | |
| details = [] | |
| for detail in log_details: | |
| for key, value in detail.items(): | |
| if key == 'counter': | |
| details.append('{}/{}'.format(value[0], value[1])) | |
| else: | |
| details.append('{}: {}'.format(key.title(), value)) | |
| log_message = '{} ({})'.format(log_message, ' - '.join(details)) | |
| PARTIAL_LOGS.append((log_message, level)) | |
| def check_version(env, versions: float = None): | |
| latest_version = env['ir.module.module'].search([('name', '=', 'base')]).latest_version.replace('~', '-') | |
| actual_version = 0 | |
| versions_to_check = [versions] if versions else VALID_VERSIONS | |
| for version in versions_to_check: | |
| if latest_version.startswith(str(version)) or latest_version.startswith('saas-%s' % str(version)): | |
| actual_version = version | |
| if not versions and not actual_version: | |
| raise Warning('Invalid version. (Actual version: {} - Valid versions: {})'.format(latest_version, iter_to_string(VALID_VERSIONS))) | |
| return actual_version | |
| def commit(env, session): | |
| env.cr.commit() | |
| def format_logs(logs): | |
| lines = ['{} - {}'.format(x[1], x[0]) for x in logs] | |
| return '\n'.join(lines) | |
| def get_session(env, additional_domains: list = None, raise_exception=True): | |
| sessions = env['pos.session'] | |
| domain = [] | |
| if SESSION_IDS: | |
| domain.append(('id', 'in', SESSION_IDS)) | |
| else: | |
| if raise_exception: | |
| raise Warning('Please define the session id.') | |
| if additional_domains: | |
| domain += additional_domains | |
| sessions = sessions.search(domain, order='id ASC') | |
| if not sessions: | |
| raise Warning('No session found. ids: {}'.format(iter_to_string(SESSION_IDS))) | |
| return sessions | |
| def iter_to_string(iterable): | |
| return ', '.join(str(x) for x in iterable) | |
| def post_note(env, session): | |
| if not check_version(env, 12.0) and ON_SUCCESS_LOG_NOTE: | |
| message = '{}'.format(ON_SUCCESS_LOG_NOTE) | |
| session.message_post(body=message) | |
| def raise_logs(): | |
| if RAISE_LOGS: | |
| lines = format_logs(FULL_LOGS) | |
| raise Warning(lines) | |
| def save_log_lines(env): | |
| for line in PARTIAL_LOGS: | |
| FULL_LOGS.append(line) | |
| if SAVE_LOGS: | |
| lines = format_logs(PARTIAL_LOGS) | |
| data = ({ | |
| 'create_date': datetime.datetime.now(), | |
| 'create_uid': env.uid, | |
| 'type': 'server', | |
| 'dbname': env.cr.dbname, | |
| 'name': LOG_LINE_NAME, | |
| 'level': 'info', | |
| 'message': lines, | |
| 'path': 'action', | |
| 'line': 0, | |
| 'func': LOG_LINE_ACTION_NAME, | |
| }) | |
| env['ir.logging'].create(data) | |
| env.cr.commit() | |
| PARTIAL_LOGS.clear() | |
| # -----===== SPECIFIC FUNCTIONS =====------ | |
| def need_action(line): | |
| line_taxes = line.mapped('tax_ids_after_fiscal_position') | |
| return any([t.tax_exigibility == 'on_payment' for t in line_taxes]) | |
| def get_taxes_mapping(line): | |
| line_taxes = line.mapped('tax_ids_after_fiscal_position').filtered(lambda t: t.tax_exigibility == 'on_payment') | |
| taxes_mapping = {} | |
| for tax in line_taxes: | |
| if tax.id not in TAXES.keys(): | |
| new_tax = tax.search([('tax_exigibility', '=', 'on_invoice'), | |
| ('amount', '=', tax.amount), | |
| ('type_tax_use', '=', tax.type_tax_use), | |
| ('amount_type', '=', tax.amount_type), | |
| ('name', 'ilike', 'pos')]) | |
| if not new_tax: | |
| new_tax = tax.copy() | |
| new_tax.write({'name': 'POS - {}'.format(tax.name), | |
| 'tax_exigibility': 'on_invoice'}) | |
| TAXES.update({tax.id: new_tax.id}) | |
| taxes_mapping.update({tax.id: TAXES.get(tax.id)}) | |
| return taxes_mapping | |
| def update_tax(line, taxes_mapping): | |
| data = [] | |
| for old_tax, new_tax in taxes_mapping.items(): | |
| data.append((3, old_tax, 0)) | |
| data.append((4, new_tax, 0)) | |
| if data: | |
| line.write({ | |
| 'tax_ids': data | |
| }) | |
| line._get_tax_ids_after_fiscal_position() | |
| def change_taxes(session): | |
| lines = session.mapped('order_ids.lines') | |
| lines_filtered = list(filter(need_action, lines)) | |
| log_message = 'Processing {} lines needing action out of {}.'.format(len(lines_filtered), len(lines)) | |
| log_details = [{'session_id': session.id}] | |
| _add_log_line('info', log_message, log_details) | |
| for line in lines_filtered: | |
| taxes_mapping = get_taxes_mapping(line) | |
| if taxes_mapping: | |
| update_tax(line, taxes_mapping) | |
| log_message = 'Done.'.format(len(lines_filtered), len(lines)) | |
| log_details = [{'session_id': session.id}] | |
| _add_log_line('info', log_message, log_details) | |
| return True | |
| # -----===== MAIN PROCESS =====------ | |
| def process(env): | |
| check_version(env) | |
| additional_domain = [] | |
| sessions = get_session(env, additional_domain) | |
| for session in sessions: | |
| if session.state not in VALID_SESSION_STATES: | |
| log_message = 'Invalid session state, skipping.' | |
| log_details = [{'session_id': session.id}, | |
| {'actual state': session.state}, | |
| {'valid states': iter_to_string(VALID_SESSION_STATES)}] | |
| _add_log_line(WARNING, log_message, log_details) | |
| continue | |
| try: | |
| if change_taxes(session): | |
| commit(env, session) | |
| else: | |
| env.cr.rollback() | |
| except Exception as e: | |
| log_message = 'Unexpected Error occured.' | |
| log_details = [{'session_id': session.id}, | |
| {'error': str(e)}] | |
| _add_log_line(ERROR, log_message, log_details) | |
| env.cr.rollback() | |
| save_log_lines(env) | |
| raise_logs() | |
| process(env) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment