Import Prices

modules/purchase_trade/pricing.py

@@ -15,6 +15,7 @@ from sql.functions import CurrentTimestamp, DateTrunc
 from trytond.wizard import Button, StateTransition, StateView, Wizard
 from itertools import chain, groupby
 from operator import itemgetter
+import calendar
 import datetime
 import logging
 import re
@@ -162,10 +163,8 @@ class ImportPrices(Wizard):
                 if prices:
                     price = prices[0]
                 elif create_missing_price_index:
-                    price, = Price.create([{
-                        'price_index': price_index,
-                        'price_desc': price_index,
-                    }])
+                    price, = Price.create([
+                        cls._price_index_values(price_index)])
                     stats['created_indexes'].append(
                         cls._result_line(
                             row_number, price_index, None,
@@ -210,6 +209,70 @@ class ImportPrices(Wizard):
 
         return stats
 
+    @classmethod
+    def _price_index_values(cls, price_index):
+        values = {
+            'price_index': price_index,
+            'price_desc': price_index,
+            'price_curve_type': 'future',
+        }
+        references = [
+            ('price_type', 'price.fixtype', [('name', '=', 'Market price')]),
+            ('price_currency', 'currency.currency', [('code', '=', 'USD')]),
+            ('price_calendar', 'price.calendar', [('name', '=', 'Argus EU')]),
+            ('price_unit', 'product.uom', [('name', '=', 'Mt')]),
+        ]
+        for field, model_name, domain in references:
+            record = cls._first_record(model_name, domain)
+            if record:
+                values[field] = record.id
+
+        period = cls._period_from_price_index(price_index)
+        if period:
+            values['price_period'] = period.id
+        return values
+
+    @classmethod
+    def _period_from_price_index(cls, price_index):
+        match = re.search(r'(?<!\d)(20\d{2})[-_/\. ](0[1-9]|1[0-2])(?!\d)',
+            price_index)
+        if not match:
+            return None
+
+        year = int(match.group(1))
+        month = int(match.group(2))
+        month_name = cls._period_month_name(year, month)
+        periods = cls._period_model().search(
+            [('month_name', '=', month_name)], limit=1)
+        if periods:
+            return periods[0]
+
+        beg_date = datetime.date(year, month, 1)
+        end_date = datetime.date(
+            year, month, calendar.monthrange(year, month)[1])
+        period, = cls._period_model().create([{
+            'month_name': month_name,
+            'description': month_name,
+            'beg_date': beg_date,
+            'end_date': end_date,
+            'is_cotation': True,
+        }])
+        return period
+
+    @staticmethod
+    def _period_month_name(year, month):
+        return '%s%s' % (
+            calendar.month_abbr[month].upper(), str(year)[-2:])
+
+    @staticmethod
+    def _period_model():
+        return Pool().get('product.month')
+
+    @staticmethod
+    def _first_record(model_name, domain):
+        records = Pool().get(model_name).search(domain, limit=1)
+        return records[0] if records else None
+
     @classmethod
     def _price_value_values(cls, price, row, price_date):
         return {

modules/purchase_trade/tests/test_module.py

@@ -106,6 +106,15 @@ class PurchaseTradeTestCase(ModuleTestCase):
             pool.return_value.get.side_effect = {
                 'price.price': price_model,
                 'price.price_value': price_value_model,
+                'price.fixtype': _FakeRecordModel([
+                    SimpleNamespace(id=10, name='Market price')]),
+                'currency.currency': _FakeRecordModel([
+                    SimpleNamespace(id=20, code='USD')]),
+                'price.calendar': _FakeRecordModel([
+                    SimpleNamespace(id=30, name='Argus EU')]),
+                'product.uom': _FakeRecordModel([
+                    SimpleNamespace(id=40, name='Mt')]),
+                'product.month': _FakeRecordModel(),
             }.__getitem__
 
             stats = ImportPrices._import_rows([{
@@ -122,6 +131,74 @@ class PurchaseTradeTestCase(ModuleTestCase):
         self.assertEqual(price_model.records[0].price_index, 'LME Copper')
         self.assertEqual(price_value_model.records[0].price_value, 98.0)
 
+    def test_import_prices_defaults_created_price_index(self):
+        'import prices defaults newly created price index fields'
+        price_model = _FakePriceModel()
+        price_value_model = _FakePriceValueModel()
+        period_model = _FakeRecordModel()
+
+        with patch('trytond.modules.purchase_trade.pricing.Pool') as pool:
+            pool.return_value.get.side_effect = {
+                'price.price': price_model,
+                'price.price_value': price_value_model,
+                'price.fixtype': _FakeRecordModel([
+                    SimpleNamespace(id=10, name='Market price')]),
+                'currency.currency': _FakeRecordModel([
+                    SimpleNamespace(id=20, code='USD')]),
+                'price.calendar': _FakeRecordModel([
+                    SimpleNamespace(id=30, name='Argus EU')]),
+                'product.uom': _FakeRecordModel([
+                    SimpleNamespace(id=40, name='Mt')]),
+                'product.month': period_model,
+            }.__getitem__
+
+            ImportPrices._import_rows([{
+                'price_index': 'Argus Copper 2026-07',
+                'price_date': date(2026, 3, 27),
+                'price_value': '98',
+            }], create_missing_price_index=True)
+
+        price = price_model.records[0]
+        period = period_model.records[0]
+        self.assertEqual(price.price_type, 10)
+        self.assertEqual(price.price_currency, 20)
+        self.assertEqual(price.price_calendar, 30)
+        self.assertEqual(price.price_curve_type, 'future')
+        self.assertEqual(price.price_unit, 40)
+        self.assertEqual(price.price_period, period.id)
+        self.assertEqual(period.month_name, 'JUL26')
+        self.assertEqual(period.beg_date, date(2026, 7, 1))
+        self.assertEqual(period.end_date, date(2026, 7, 31))
+        self.assertTrue(period.is_cotation)
+
+    def test_import_prices_uses_existing_period_from_price_index(self):
+        'import prices reuses existing period parsed from the price index'
+        price_model = _FakePriceModel()
+        price_value_model = _FakePriceValueModel()
+        existing_period = SimpleNamespace(
+            id=55, month_name='APR25', description='APR25')
+        period_model = _FakeRecordModel([existing_period])
+
+        with patch('trytond.modules.purchase_trade.pricing.Pool') as pool:
+            pool.return_value.get.side_effect = {
+                'price.price': price_model,
+                'price.price_value': price_value_model,
+                'price.fixtype': _FakeRecordModel(),
+                'currency.currency': _FakeRecordModel(),
+                'price.calendar': _FakeRecordModel(),
+                'product.uom': _FakeRecordModel(),
+                'product.month': period_model,
+            }.__getitem__
+
+            ImportPrices._import_rows([{
+                'price_index': 'Argus Copper 2025-04',
+                'price_date': date(2026, 3, 27),
+                'price_value': '98',
+            }], create_missing_price_index=True)
+
+        self.assertEqual(price_model.records[0].price_period, 55)
+        self.assertEqual(period_model.records, [existing_period])
+
     def test_import_prices_skips_existing_date_without_overwrite(self):
         'import prices skips existing price dates unless overwrite is enabled'
         price = SimpleNamespace(id=1, price_index='LME Copper')
@@ -278,4 +355,32 @@ class _FakePriceValueModel:
                 setattr(record, name, value)
 
 
+class _FakeRecordModel:
+
+    def __init__(self, records=None):
+        self.records = records or []
+
+    def search(self, domain, limit=None):
+        records = self.records
+        for name, operator, value in domain:
+            self.assert_operator(operator)
+            records = [
+                record for record in records
+                if getattr(record, name, None) == value]
+        return records[:limit] if limit else records
+
+    def create(self, values):
+        records = []
+        for value in values:
+            record = SimpleNamespace(id=len(self.records) + 1, **value)
+            self.records.append(record)
+            records.append(record)
+        return records
+
+    @staticmethod
+    def assert_operator(operator):
+        if operator != '=':
+            raise AssertionError('Unsupported operator: %s' % operator)
+
+
 del ModuleTestCase