fix: sync backup production numbers to date-only orders

routes/purchase.py

@@ -5,7 +5,7 @@ import re
 import unicodedata
 from datetime import datetime
 from flask import Blueprint, request, jsonify, session, redirect, url_for, render_template
-from services.db import query, execute, add_operation_log
+from services.db import query, execute, add_operation_log, get_db
 from services.field_perms import filter_rows as filter_field_rows
 from services.sales_calc import (compute_product_sales_cached,
                                  compute_product_sales_extended_cached)
@@ -3879,7 +3879,8 @@ def _get_ordered_backup_time(product_code, backup_id=None, ordered_at=None):
     其次按传入 ordered_at，最后回退到该商品最近一次下单。
     """
     if backup_id:
-        rows = query("SELECT ordered_at FROM pending_orders WHERE id=%s AND status='ordered'", (backup_id,))
+        rows = query("SELECT ordered_at FROM pending_orders WHERE id=%s AND product_code=%s AND status='ordered'",
+                     (backup_id, product_code))
     elif ordered_at:
         return str(ordered_at)[:19]
     else:
@@ -3893,38 +3894,97 @@ def _get_ordered_backup_time(product_code, backup_id=None, ordered_at=None):
     return value.strftime('%Y-%m-%d %H:%M:%S') if hasattr(value, 'strftime') else str(value)[:19]
 
 
+def _backup_linked_order_ids(read, table, product_code, ordered_at):
+    """Prefer the exact order time; date-only legacy rows need a unique source."""
+    code_field, time_field = {
+        'domestic_orders': ('货品编号', '计划下单时间'),
+        'overseas_orders': ('产品编码', '计划下单时间'),
+        'packaging_orders': ('来源产品编码', '下单时间'),
+    }[table]
+    rows = read(f'SELECT id FROM {table} WHERE `{code_field}`=%s AND `{time_field}`=%s',
+                (product_code, ordered_at))
+    if rows:
+        return [row['id'] for row in rows]
+
+    day = ordered_at[:10]
+    legacy = read(f'SELECT id FROM {table} WHERE `{code_field}`=%s AND TRIM(`{time_field}`)=%s',
+                  (product_code, day))
+    if not legacy:
+        return []
+    sources = read(
+        "SELECT id FROM pending_orders WHERE product_code=%s "
+        "AND status IN ('ordered','rolled_back') "
+        "AND ordered_at >= %s AND ordered_at < DATE_ADD(%s, INTERVAL 1 DAY)",
+        (product_code, day, day),
+    )
+    # 成品订单须唯一；同一备份可以生成多条包材订单。
+    if len(sources) != 1 or (table != 'packaging_orders' and len(legacy) != 1):
+        raise ValueError('同一商品当天有多笔记录，日期不足以确定对应订单，请核对计划下单时间')
+    return [row['id'] for row in legacy]
+
+
 @purchase_bp.route('/api/purchase/backup/update_prod_no', methods=['POST'])
 @login_required
 def backup_update_prod_no():
     denied = _require_purchase_operation('purchase_ready', 'edit')
     if denied:
         return denied
-    """Update only order rows created by this exact backup order time."""
-    data = request.json
+    """Save the backup and uniquely linked order numbers in one transaction."""
+    data = request.get_json(silent=True) or {}
     product_code = data.get('product_code', '')
     prod_no = data.get('production_order_no', '')
     backup_id = data.get('backup_id')
     ordered_at = data.get('ordered_at', '')
     if not product_code:
         return jsonify({"code": 400, "msg": "参数错误"})
-    ordered_at = _get_ordered_backup_time(product_code, backup_id=backup_id, ordered_at=ordered_at)
-    if not ordered_at:
-        return jsonify({"code": 404, "msg": "未找到对应下单时间"})
-    if backup_id:
-        query("""UPDATE pending_orders SET production_order_no=%s WHERE id=%s AND status='ordered'""",
-              (prod_no, backup_id))
-    else:
-        query("""UPDATE pending_orders SET production_order_no=%s
-                 WHERE product_code=%s AND status='ordered' AND ordered_at=%s""",
-              (prod_no, product_code, ordered_at))
-    query("""UPDATE domestic_orders SET `旺店通生产单号`=%s
-             WHERE `货品编号`=%s AND `计划下单时间`=%s""", (prod_no, product_code, ordered_at))
-    query("""UPDATE overseas_orders SET `旺店通生产单号`=%s
-             WHERE `产品编码`=%s AND `计划下单时间`=%s""", (prod_no, product_code, ordered_at))
-    query("""UPDATE packaging_orders SET `生产单号`=%s
-             WHERE `来源产品编码`=%s AND `下单时间`=%s""", (prod_no, product_code, ordered_at))
+    conn = get_db()
+    counts = {}
+    try:
+        conn.begin()
+        with conn.cursor() as cur:
+            def read(sql, params):
+                cur.execute(sql, params)
+                return cur.fetchall()
+
+            if backup_id:
+                backups = read("SELECT id, ordered_at FROM pending_orders WHERE id=%s "
+                               "AND product_code=%s AND status='ordered' FOR UPDATE", (backup_id, product_code))
+            elif ordered_at:
+                backups = read("SELECT id, ordered_at FROM pending_orders WHERE product_code=%s "
+                               "AND status='ordered' AND ordered_at=%s FOR UPDATE", (product_code, ordered_at))
+            else:
+                backups = read("SELECT id, ordered_at FROM pending_orders WHERE product_code=%s "
+                               "AND status='ordered' ORDER BY ordered_at DESC,id DESC LIMIT 1 FOR UPDATE", (product_code,))
+            if len(backups) != 1 or not backups[0].get('ordered_at'):
+                conn.rollback()
+                return jsonify({"code": 404, "msg": "未找到唯一对应的下单备份"})
+            backup = backups[0]
+            time = backup['ordered_at']
+            ordered_at = time.strftime('%Y-%m-%d %H:%M:%S') if hasattr(time, 'strftime') else str(time)[:19]
+            targets = {table: _backup_linked_order_ids(read, table, product_code, ordered_at)
+                       for table in ('domestic_orders', 'overseas_orders', 'packaging_orders')}
+            cur.execute('UPDATE pending_orders SET production_order_no=%s WHERE id=%s', (prod_no, backup['id']))
+            for table, ids in targets.items():
+                counts[table] = len(ids)
+                if ids:
+                    field = '生产单号' if table == 'packaging_orders' else '旺店通生产单号'
+                    placeholders = ','.join(['%s'] * len(ids))
+                    cur.execute(f'UPDATE {table} SET `{field}`=%s WHERE id IN ({placeholders})',
+                                (prod_no, *ids))
+        conn.commit()
+    except ValueError as exc:
+        conn.rollback()
+        return jsonify({"code": 409, "msg": str(exc)})
+    except Exception:
+        conn.rollback()
+        raise
+    finally:
+        conn.close()
     add_operation_log('生产单号更新', f'{product_code}/{ordered_at}: {prod_no}')
-    return jsonify({"code": 0, "msg": "保存成功"})
+    msg = '生产单号已保存，已同步对应订单'
+    if not any(counts.values()):
+        msg = '生产单号已保存，但未找到可同步的订单，请核对订单记录'
+    return jsonify({"code": 0, "msg": msg, "data": {"synced": counts}})
 
 
 @purchase_bp.route('/api/purchase/backup/audit_material_price', methods=['POST'])

static/js/app.js

@@ -19483,25 +19483,13 @@ window.saveBackupProdNo = function(bid, code) {
             inp.blur();
             // 更新缓存数据
             _bkItems().forEach(function(it) {
-                if (it.product_code === code) it.production_order_no = val;
+                if (String(it.id) === String(bid)) it.production_order_no = val;
             });
+            _renderFilteredBackup();
             _refreshLinkedOrderLists();
-            // 同步更新表格中的生产单号列
-            var tbody = _bkEl('Tbody');
-            if (tbody) {
-                tbody.querySelectorAll('tr').forEach(function(tr) {
-                    var codeEl = tr.querySelector('code');
-                    if (codeEl && codeEl.textContent.trim() === code) {
-                        var tds = tr.querySelectorAll('td');
-                        // 生产单号是第7列（序号0，产品类型1，货品编号2，货品名称3，下单时间4，下单人5，生产单号6）
-                        if (tds.length > 6) {
-                            tds[6].textContent = val || '-';
-                        }
-                    }
-                });
-            }
-            showCenteredAlert('保存成功', '✅ 生产单号已保存');
+            showCenteredAlert('保存成功', '✅ ' + (res.msg || '生产单号已保存'));
         } else {
+            showCenteredAlert('保存失败', res.msg || '生产单号保存失败');
             inp.style.borderColor = '#f5222d';
             setTimeout(function(){ inp.style.borderColor = ''; }, 1500);
         }
@@ -19610,12 +19598,12 @@ window.saveBackupAll = function(bid, code, saveBtn) {
         if (saveBtn) { saveBtn.disabled = false; saveBtn.textContent = '💾 保存'; }
         var hasError = results.some(function(r) { return r.code !== 0; });
         if (hasError) {
-            showCenteredAlert('保存失败', '❌ 部分数据保存失败，请重试');
+            showCenteredAlert('保存失败', '❌ ' + results.filter(function(r) { return r.code !== 0; }).map(function(r) { return r.msg || '部分数据保存失败'; }).join('；'));
         } else {
             // ---- 同步更新前端缓存，确保关闭后再打开看到新数据 ----
             var newProdNo = prodNoInp ? prodNoInp.value.trim() : '';
             _bkItems().forEach(function(it) {
-                if (it.product_code === code) {
+                if (String(it.id) === String(bid)) {
                     // 更新生产单号
                     it.production_order_no = newProdNo;
                     // 更新包材采购单号（bom_items JSON内）
@@ -19636,27 +19624,15 @@ window.saveBackupAll = function(bid, code, saveBtn) {
                     }
                 }
             });
-            // 同步更新表格中的生产单号列
-            var tbody = _bkEl('Tbody');
-            if (tbody) {
-                tbody.querySelectorAll('tr').forEach(function(tr) {
-                    var codeEl = tr.querySelector('code');
-                    if (codeEl && codeEl.textContent.trim() === code) {
-                        var tds = tr.querySelectorAll('td');
-                        if (tds.length > 6) {
-                            tds[6].textContent = newProdNo || '-';
-                        }
-                    }
-                });
-            }
+            _renderFilteredBackup();
             // 后台刷新列表数据（用户下次查看时拉取最新）
             if (typeof loadBackupList === 'function') {
-                loadBackupList();
+                loadBackupList(false);
             }
             _refreshLinkedOrderLists();
             // 包材订单管理数据已同步（旺店通单号等），保留当前筛选并重新拉取数据
             var detailOverlay = saveBtn && saveBtn.closest('.modal-overlay');
-            showCenteredAlert('保存成功', '✅ 生产单号、采购单号已保存');
+            showCenteredAlert('保存成功', '✅ ' + (prodNoInp && results[0] ? results[0].msg : '生产单号、采购单号已保存'));
             if (detailOverlay) detailOverlay.remove();
         }
     }).catch(function() {

tests/test_backup_production_no_sync.py

@@ -0,0 +1,120 @@
+"""Keep production numbers linked to one order, including date-only history."""
+import unittest
+from datetime import datetime
+from unittest import mock
+from flask import Flask
+
+import routes.purchase as purchase
+
+
+class BackupProductionNoSyncTest(unittest.TestCase):
+    def test_exact_time_wins_without_using_date_fallback(self):
+        read = mock.Mock(return_value=[{'id': 9}])
+        self.assertEqual(purchase._backup_linked_order_ids(
+            read, 'domestic_orders', 'P1', '2026-09-27 18:29:13'), [9])
+        read.assert_called_once()
+        self.assertEqual(read.call_args.args[1], ('P1', '2026-09-27 18:29:13'))
+
+    def test_legacy_date_matches_only_unique_backup_and_product_order(self):
+        read = mock.Mock(side_effect=[[], [{'id': 2804}], [{'id': 1555}]])
+        self.assertEqual(purchase._backup_linked_order_ids(
+            read, 'domestic_orders', 'P1', '2026-09-27 18:29:13'), [2804])
+        self.assertIn('TRIM(`计划下单时间`)=%s', read.call_args_list[1].args[0])
+        self.assertEqual(read.call_args_list[1].args[1], ('P1', '2026-09-27'))
+
+    def test_ambiguous_same_day_backups_are_never_broadly_updated(self):
+        read = mock.Mock(side_effect=[[], [{'id': 9}], [{'id': 1}, {'id': 2}]])
+        with self.assertRaises(ValueError):
+            purchase._backup_linked_order_ids(read, 'domestic_orders', 'P1', '2026-09-27 18:29:13')
+
+    def test_ambiguous_legacy_product_orders_are_rejected(self):
+        read = mock.Mock(side_effect=[[], [{'id': 9}, {'id': 10}], [{'id': 1}]])
+        with self.assertRaises(ValueError):
+            purchase._backup_linked_order_ids(read, 'overseas_orders', 'P1', '2026-09-27 18:29:13')
+
+    def test_multiple_packaging_rows_can_belong_to_one_backup(self):
+        read = mock.Mock(side_effect=[[], [{'id': 9}, {'id': 10}], [{'id': 1}]])
+        self.assertEqual(purchase._backup_linked_order_ids(
+            read, 'packaging_orders', 'P1', '2026-09-27 18:29:13'), [9, 10])
+
+    def call_save(self, replies, execute_error=None):
+        app = Flask(__name__)
+        app.secret_key = 'test-only'
+        conn = mock.MagicMock()
+        cur = conn.cursor.return_value.__enter__.return_value
+        cur.fetchall.side_effect = replies
+        if execute_error:
+            cur.execute.side_effect = execute_error
+        with app.test_request_context(json={
+            'product_code': 'P1', 'backup_id': 1555, 'production_order_no': 'PS2026093029'
+        }), mock.patch.object(purchase, '_require_purchase_operation', return_value=None), \
+                mock.patch.object(purchase, 'get_db', return_value=conn), \
+                mock.patch.object(purchase, 'add_operation_log'):
+            response = purchase.backup_update_prod_no.__wrapped__()
+        return response.get_json(), conn, cur
+
+    def test_save_and_sync_commit_together_using_target_ids(self):
+        body, conn, cur = self.call_save([
+            [{'id': 1555, 'ordered_at': datetime(2026, 9, 27, 18, 29, 13)}],
+            [], [{'id': 2804}], [{'id': 1555}],  # domestic legacy date
+            [], [],  # no overseas order
+            [{'id': 3}, {'id': 4}],  # exact packaging rows
+        ])
+        self.assertEqual(body['code'], 0)
+        self.assertEqual(body['data']['synced'], {
+            'domestic_orders': 1, 'overseas_orders': 0, 'packaging_orders': 2,
+        })
+        writes = [call.args for call in cur.execute.call_args_list if call.args[0].startswith('UPDATE')]
+        self.assertEqual([args[1] for args in writes], [
+            ('PS2026093029', 1555), ('PS2026093029', 2804), ('PS2026093029', 3, 4)
+        ])
+        for sql, params in writes:
+            self.assertEqual(sql.count('%s'), len(params))
+        conn.commit.assert_called_once()
+        conn.rollback.assert_not_called()
+        conn.close.assert_called_once()
+
+    def test_ambiguity_prevents_even_the_backup_write(self):
+        body, conn, cur = self.call_save([
+            [{'id': 1555, 'ordered_at': datetime(2026, 9, 27, 18, 29, 13)}],
+            [], [{'id': 2804}], [{'id': 1555}, {'id': 1556}],
+        ])
+        self.assertEqual(body['code'], 409)
+        self.assertFalse(any(call.args[0].startswith('UPDATE') for call in cur.execute.call_args_list))
+        conn.commit.assert_not_called()
+        conn.rollback.assert_called_once()
+
+    def test_wrong_product_or_missing_backup_is_not_saved(self):
+        body, conn, cur = self.call_save([[]])
+        self.assertEqual(body['code'], 404)
+        self.assertIn('product_code=%s', cur.execute.call_args.args[0])
+        self.assertEqual(cur.execute.call_args.args[1], (1555, 'P1'))
+        conn.commit.assert_not_called()
+        conn.rollback.assert_called_once()
+
+    def test_write_failure_rolls_back_all_related_updates(self):
+        app = Flask(__name__)
+        conn = mock.MagicMock()
+        cur = conn.cursor.return_value.__enter__.return_value
+        cur.fetchall.side_effect = [
+            [{'id': 1555, 'ordered_at': datetime(2026, 9, 27, 18, 29, 13)}],
+            [{'id': 2804}], [], [], [], [],
+        ]
+        def execute(sql, params):
+            if sql.startswith('UPDATE domestic_orders'):
+                raise RuntimeError('database write failure')
+        cur.execute.side_effect = execute
+        with app.test_request_context(json={'product_code': 'P1', 'backup_id': 1555, 'production_order_no': 'PS1'}), \
+                mock.patch.object(purchase, '_require_purchase_operation', return_value=None), \
+                mock.patch.object(purchase, 'get_db', return_value=conn), \
+                mock.patch.object(purchase, 'add_operation_log') as audit:
+            with self.assertRaises(RuntimeError):
+                purchase.backup_update_prod_no.__wrapped__()
+            audit.assert_not_called()
+        conn.commit.assert_not_called()
+        conn.rollback.assert_called_once()
+        conn.close.assert_called_once()
+
+
+if __name__ == '__main__':
+    unittest.main()