-
-
Save alt-odoo/5f1a90a1b887b5928cae20bd267f5b98 to your computer and use it in GitHub Desktop.
POS - Compare Orders
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, 12.3, 13 | |
| # Set sessions ids between the brackets, separated by a coma | |
| # I.E. SESSION_ID = [12, 18, 132] or SESSION_ID = [4] | |
| SESSION_IDS = [] | |
| FIELDS_TO_DISPLAY_IN_CSV = [ # Comment the field you don't want to see | |
| 'name', | |
| 'price_unit', | |
| 'tax_is', | |
| 'tax_should_be', | |
| 'diff_tax', | |
| 'tax_name', | |
| 'total_is', | |
| 'total_should_be', | |
| 'diff_line', | |
| 'order_ids', | |
| 'session_ids', | |
| 'recurrence' | |
| ] | |
| # DO NOT MODIFY | |
| VALID_VERSIONS = (12, 13,) | |
| VALID_SESSION_STATES = ('opened', 'closing_control') | |
| VALID_MODES = ('check', 'process') | |
| LOG_LINE_ACTION_NAME = 'POS - Compare Orders' | |
| LOG_LINE_NAME = 'Support Intervention' | |
| ON_SUCCESS_LOG_NOTE = '' | |
| 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' | |
| # -----===== 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 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 build_csv(data): | |
| result = '|'.join(FIELDS_TO_DISPLAY_IN_CSV) | |
| for value in data.values(): | |
| result += '\n{}'.format('|'.join(str(content) for field, content in value.items() if field in FIELDS_TO_DISPLAY_IN_CSV)) | |
| return result | |
| def _extract_tax_from_price(price_with_tax, price_without_tax): | |
| negative = price_with_tax < 0 | |
| result = abs(price_with_tax) - abs(price_without_tax) | |
| result = -result if negative else result | |
| return result | |
| def _get_order_currency(env, order): | |
| if check_version(env) < 13: | |
| return order.config_id.currency_id | |
| else: | |
| return order.currency_id | |
| def compare(env, session, product_resume): | |
| default_partner = env['res.partner'].create({ | |
| 'name': 'pos_test', | |
| 'email': '[email protected]', | |
| 'phone': '12341234' | |
| }) | |
| total_diff_tax = 0 | |
| total_diff_line = 0 | |
| for pos_order in session.order_ids: | |
| currency = _get_order_currency(env, pos_order) | |
| sale_order = env['sale.order'].create({ | |
| 'company_id': session.config_id.company_id.id, | |
| 'partner_id': pos_order.partner_id.id or default_partner.id, | |
| 'pricelist_id': pos_order.pricelist_id.id, | |
| }) | |
| for pol in pos_order.lines: | |
| sol = env['sale.order.line'].create({ | |
| 'order_id': sale_order.id, | |
| 'company_id': session.config_id.company_id.id, | |
| 'product_id': pol.product_id.id, | |
| 'tax_id': [(4, tax_id.id) for tax_id in pol.tax_ids_after_fiscal_position] | |
| }) | |
| sol.write({ | |
| 'price_unit': pol.price_unit, | |
| 'discount': pol.discount, | |
| 'product_uom_qty': pol.qty, | |
| }) | |
| pol_total = pol.price_subtotal_incl | |
| sol_total = sol.price_total | |
| pol_tax = _extract_tax_from_price(pol.price_subtotal_incl, pol.price_subtotal) | |
| sol_tax = sol.price_tax | |
| if not currency.is_zero(abs(pol_tax - sol_tax)) or not currency.is_zero(abs(pol_total - sol_total)): | |
| diff_tax = currency.round(pol_tax - sol_tax) | |
| diff_line = currency.round(pol_total - sol_total) | |
| total_diff_tax += diff_tax | |
| total_diff_line += diff_line | |
| key = '{}-{}'.format(pol.product_id, pol.price_unit) | |
| if key not in product_resume.keys(): | |
| data = { | |
| 'name': '{}({})'.format(pol.product_id.name, pol.product_id.id), | |
| 'price_unit': pol.price_unit, | |
| 'tax_is': currency.round(pol_tax), | |
| 'tax_should_be': currency.round(sol_tax), | |
| 'diff_tax': diff_tax, | |
| 'tax_name': [tax_id.name for tax_id in pol.tax_ids_after_fiscal_position], | |
| 'total_is': currency.round(pol_total), | |
| 'total_should_be': currency.round(sol_total), | |
| 'diff_line': diff_line, | |
| 'order_ids': [pos_order.id], | |
| 'session_ids': [session.id], | |
| 'recurrence': 1 | |
| } | |
| product_resume.update({key: data}) | |
| else: | |
| product_resume[key]['recurrence'] += 1 | |
| if pos_order.id not in product_resume[key]['order_ids']: | |
| product_resume[key]['order_ids'].append(pos_order.id) | |
| if session.id not in product_resume[key]['session_ids']: | |
| product_resume[key]['session_ids'].append(session.id) | |
| key = 'Session {}'.format(session.id) | |
| data = { | |
| 'name': key, | |
| 'price_unit': '-', | |
| 'tax_is': '-', | |
| 'tax_should_be': '-', | |
| 'diff_tax': total_diff_tax, | |
| 'tax_name': '-', | |
| 'total_is': '-', | |
| 'total_should_be': '-', | |
| 'diff_line': total_diff_line, | |
| 'order_ids': '-', | |
| 'session_ids': '-', | |
| 'recurrence': '-' | |
| } | |
| product_resume.update({key: data}) | |
| return product_resume | |
| # -----===== MAIN PROCESS =====------ | |
| def process(env): | |
| check_version(env) | |
| additional_domain = [] | |
| sessions = get_session(env, additional_domain) | |
| product_resume = {} | |
| 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: | |
| product_resume = compare(env, session, product_resume) | |
| 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) | |
| save_log_lines(env) | |
| if product_resume: | |
| output = ['-----=====BEGIN OF CSV=====-----'] | |
| output.append(build_csv(product_resume)) | |
| output.append('-----===== END OF CSV =====-----') | |
| log_message = '\n'.join(output) | |
| _add_log_line(INFO, log_message) | |
| 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