from odoo import fields, models, api, _ from odoo.exceptions import AccessError, UserError, ValidationError from odoo.tools.float_utils import float_is_zero from collections import defaultdict from datetime import timedelta, datetime from datetime import timedelta, datetime as waktu from itertools import groupby import pytz, requests, json, requests from dateutil import parser import datetime import hmac import hashlib import base64 import requests import time import logging import re from hashlib import sha256 _logger = logging.getLogger(__name__) Request_URI = '/openapi/order/v2/list-order' ACCESS_KEY = '24bb6a1ec618ec6a' SECRET_KEY = '32e4a78ad05ee230' class ShipmentGroup(models.Model): _name = "shipment.group" _description = "Shipment Group" _inherit = ['mail.thread'] _rec_name = 'number' number = fields.Char(string='Document No', index=True, copy=False, readonly=True, tracking=True) shipment_line = fields.One2many('shipment.group.line', 'shipment_id', string='Shipment Group Lines', auto_join=True) picking_lines = fields.One2many('picking.line', 'shipment_id', string='Picking Lines', auto_join=True) related_count = fields.Integer(compute='_compute_related_count', string='Related Count') receipt = fields.Char(string='Receipt', related='picking_lines.scan_receipt') def sync_product_to_picking_line(self): for shipment in self: for picking_line in shipment.picking_lines: shipment_lines = self.env['shipment.group.line'].search([ ('shipment_id', '=', shipment.id), ('invoice_marketplace', '=', picking_line.invoice_marketplace), ('picking_id', '=', picking_line.picking_id.id) ]) picking_line.product_shipment_lines.unlink() for sl in shipment_lines: self.env['product.shipment.line'].create({ 'picking_line_id': picking_line.id, 'product_id': sl.product_id.id, 'carrier': sl.carrier or picking_line.carrier, 'invoice_marketplace': sl.invoice_marketplace, 'picking_id': sl.picking_id.id, 'order_reference': sl.order_reference, }) def _compute_related_count(self): for record in self: record.related_count = len(record.picking_lines) def action_view_related_line(self): self.ensure_one() pickingLines = self.env['picking.line'] base_bu = pickingLines.search([ ('shipment_id', '=', self.id) ]) all_bu = base_bu return { 'name': 'Related', 'type': 'ir.actions.act_window', 'res_model': 'picking.line', 'view_mode': 'tree,form', 'target': 'current', 'domain': [('id', 'in', list(all_bu.ids))], } @api.constrains('picking_lines') def _check_picking_lines(self): for record in self: for picking_line in record.picking_lines: if not picking_line.picking_id: continue for move in picking_line.picking_id.move_ids_without_package: existing_line = record.shipment_line.filtered( lambda l: l.product_id == move.product_id and l.picking_id.tracking_number == picking_line.scan_receipt and l.carrier == picking_line.carrier ) if not existing_line: self.env['shipment.group.line'].create({ 'shipment_id': record.id, 'product_id': move.product_id.id, 'carrier': picking_line.picking_id.carrier, 'invoice_marketplace': picking_line.picking_id.invoice_mp, 'order_reference': picking_line.order_reference, 'picking_id': picking_line.picking_id.id }) def get_status(self): for picking_line in self.picking_lines: try: order_id = picking_line.invoice_marketplace authorization = self.sign_request() headers = { 'Content-Type': 'application/json', 'X-Advai-Country': 'ID', 'Authorization': authorization } payload = { "orderNumbers": [order_id], } url = "https://api.ginee.com/openapi/order/v2/list-order" response = requests.post( url, headers=headers, data=json.dumps(payload) ) if response.status_code == 200: data = response.json() if data.get('code') == 'SUCCESS' and data.get('message') == 'OK': content = data.get('data', {}).get('content', []) if not content: raise UserError(_("No List Order information found in response")) content_info = content[0] if content_info.get('orderStatus') == 'CANCELLED': picking_line.status = content_info.get('orderStatus') else: picking_line.status = '' else: raise UserError(_("API Error: %s - %s") % (data.get('code', 'UNKNOWN'), data.get('message', 'No error message'))) else: raise UserError(_("API request failed with status code: %s") % response.status_code) except Exception as e: raise UserError(_("Error: %s") % str(e)) def sign_request(self): signData = '$'.join(['POST', Request_URI]) + '$' authorization = ACCESS_KEY + ':' + base64.b64encode( hmac.new(SECRET_KEY.encode('utf-8'), signData.encode('utf-8'), digestmod=sha256).digest() ).decode('ascii') return authorization @api.model def create(self, vals): vals['number'] = self.env['ir.sequence'].next_by_code('shipment.group') or '0' result = super(ShipmentGroup, self).create(vals) return result class ShipmentGroupLine(models.Model): _name = 'shipment.group.line' _description = 'Shipment Group Line' _order = 'shipment_id, id' shipment_id = fields.Many2one('shipment.group', string='Shipment Ref', required=True, ondelete='cascade') product_id = fields.Many2one('product.product', string='Product') carrier = fields.Char(string='Shipping Method') invoice_marketplace = fields.Char(string='Invoice Marketplace') picking_id = fields.Many2one('stock.picking', string='Picking') order_reference = fields.Char(string='Order Reference') class PickingLine(models.Model): _name = 'picking.line' _description = 'Picking Line' _order = 'shipment_id, id' shipment_id = fields.Many2one('shipment.group', string='Shipment Ref', required=True, ondelete='cascade') product_shipment_lines = fields.One2many('product.shipment.line', 'picking_line_id', string='Product Shipment Lines', auto_join=True) picking_id = fields.Many2one('stock.picking', string='Picking') scan_receipt = fields.Char(string="Scan Receipt") invoice_marketplace = fields.Char(string='Invoice Marketplace') carrier = fields.Char(string='Ekspedisi') order_reference = fields.Char(string='Order Reference') status = fields.Char(string='Status') @api.constrains('scan_receipt') def _check_duplicate_scan(self): for line in self: if not line.scan_receipt: continue # Cari duplikat selain dirinya dup = self.search([ ('scan_receipt', '=', line.scan_receipt), ('id', '!=', line.id) ], limit=1) if not dup: continue # 1. Dup di shipment yang sama (lagi create / edit) if dup.shipment_id.id == line.shipment_id.id: raise ValidationError( "Receipt '%s' sudah ada di Shipment ini." % line.scan_receipt ) # 2. Dup di shipment lain (data lama) raise ValidationError( "Receipt '%s' sudah digunakan di Shipment %s." % (line.scan_receipt, dup.shipment_id.number or '-') ) @api.onchange('scan_receipt') def _onchange_scan_receipt(self): for line in self: if not line.scan_receipt: continue # Cari picking picking = self.env['stock.picking'].search([ ('tracking_number', '=', line.scan_receipt) ], limit=1) if not picking: raise UserError("Receipt '%s' not found." % line.scan_receipt) # Isi field otomatis line.picking_id = picking line.carrier = picking.carrier line.order_reference = picking.order_reference line.invoice_marketplace = picking.invoice_mp class ProductShipmentLine(models.Model): _name = 'product.shipment.line' _description = 'Product Shipment Line' _order = 'picking_line_id, id' picking_line_id = fields.Many2one('picking.line', string='Picking Line', required=True, ondelete='cascade') product_id = fields.Many2one('product.product', string='Product') carrier = fields.Char(string='Shipping Method') invoice_marketplace = fields.Char(string='Invoice Marketplace') picking_id = fields.Many2one('stock.picking', string='Picking') order_reference = fields.Char(string='Order Reference')