113 lines
6.0 KiB
Python
113 lines
6.0 KiB
Python
"""Additive legacy SQLite migration. Never rewrites existing order numbers."""
|
|
import sqlite3
|
|
from datetime import datetime
|
|
from decimal import Decimal, ROUND_HALF_UP
|
|
from pathlib import Path
|
|
|
|
|
|
def upgrade_database(path):
|
|
path = Path(path)
|
|
if not path.exists():
|
|
return
|
|
with sqlite3.connect(path) as con:
|
|
tables = {r[0] for r in con.execute("SELECT name FROM sqlite_master WHERE type='table'")}
|
|
if 'orders' not in tables:
|
|
return
|
|
version = con.execute('PRAGMA user_version').fetchone()[0]
|
|
if version >= 5:
|
|
return
|
|
duplicates = con.execute('SELECT order_no, COUNT(*) FROM orders GROUP BY order_no HAVING COUNT(*) > 1').fetchall()
|
|
if duplicates:
|
|
raise RuntimeError('升级前请处理重复单号（原数据未修改）：%s' % duplicates)
|
|
backup = path.with_name(path.name + '.pre-upgrade-' + datetime.now().strftime('%Y%m%d-%H%M%S-%f') + '.bak')
|
|
with sqlite3.connect(backup) as dest:
|
|
con.backup(dest)
|
|
con.execute('BEGIN IMMEDIATE')
|
|
try:
|
|
if version < 2:
|
|
upgrade_fixed_amounts(con)
|
|
if version < 3:
|
|
merge_approval(con, tables)
|
|
if version < 4:
|
|
add_tracking(con)
|
|
if 'suppliers' in tables:
|
|
columns = {row[1] for row in con.execute('PRAGMA table_info(suppliers)')}
|
|
if 'short_name' not in columns:
|
|
con.execute("ALTER TABLE suppliers ADD COLUMN short_name VARCHAR(60) NOT NULL DEFAULT ''")
|
|
con.execute('PRAGMA user_version=5')
|
|
con.commit()
|
|
except Exception:
|
|
con.rollback()
|
|
raise
|
|
|
|
|
|
def add_tracking(con):
|
|
# Historical invoices, payments and deliveries are unknown until recorded.
|
|
additions = {
|
|
'needs_prepayment': 'BOOLEAN NOT NULL DEFAULT 0',
|
|
'invoice_status': "VARCHAR(20) NOT NULL DEFAULT 'unknown'",
|
|
'delivery_status': "VARCHAR(20) NOT NULL DEFAULT 'unknown'",
|
|
'payment_status': "VARCHAR(20) NOT NULL DEFAULT 'unknown'",
|
|
'payment_date': "VARCHAR(10) NOT NULL DEFAULT ''",
|
|
'fulfillment_note': "TEXT NOT NULL DEFAULT ''",
|
|
'payment_note': "TEXT NOT NULL DEFAULT ''",
|
|
'fulfillment_revision': 'INTEGER NOT NULL DEFAULT -1',
|
|
'payment_revision': 'INTEGER NOT NULL DEFAULT -1',
|
|
}
|
|
existing = {row[1] for row in con.execute('PRAGMA table_info(orders)')}
|
|
for name, definition in additions.items():
|
|
if name not in existing:
|
|
con.execute('ALTER TABLE orders ADD COLUMN %s %s' % (name, definition))
|
|
|
|
|
|
def upgrade_fixed_amounts(con):
|
|
additions = {
|
|
'users': {'session_version': 'INTEGER NOT NULL DEFAULT 1', 'must_change_password': 'BOOLEAN NOT NULL DEFAULT 1'},
|
|
'orders': {'reviewed_by_id': 'INTEGER', 'approved_by_id': 'INTEGER', 'revision': 'INTEGER NOT NULL DEFAULT 0', 'row_version': 'INTEGER NOT NULL DEFAULT 1'},
|
|
'approval_logs': {'user_id': 'INTEGER', 'revision': 'INTEGER NOT NULL DEFAULT 0', 'from_status': 'VARCHAR(20)', 'to_status': 'VARCHAR(20)'},
|
|
}
|
|
for table, columns in additions.items():
|
|
existing = {r[1] for r in con.execute('PRAGMA table_info(%s)' % table)}
|
|
for name, definition in columns.items():
|
|
if name not in existing:
|
|
con.execute('ALTER TABLE %s ADD COLUMN %s %s' % (table, name, definition))
|
|
for table, old, new, places in [
|
|
('materials', 'price', 'price_units', 4), ('orders', 'total_amount', 'total_cents', 2),
|
|
('order_items', 'qty', 'qty_units', 4), ('order_items', 'price', 'price_units', 4),
|
|
('order_items', 'amount', 'amount_cents', 2),
|
|
]:
|
|
columns = {r[1] for r in con.execute('PRAGMA table_info(%s)' % table)}
|
|
if new in columns:
|
|
continue
|
|
con.execute('ALTER TABLE %s ADD COLUMN %s INTEGER DEFAULT 0' % (table, new))
|
|
for row_id, value in con.execute('SELECT id, %s FROM %s' % (old, table)).fetchall():
|
|
val = Decimal(str(value or 0))
|
|
if not val.is_finite() or abs(val) * 10 ** places > 2 ** 63 - 1:
|
|
raise RuntimeError('无法迁移 %s.id=%s 的金额/数量，请检查原始数据' % (table, row_id))
|
|
scaled = int((val * 10 ** places).quantize(Decimal('1'), rounding=ROUND_HALF_UP))
|
|
con.execute('UPDATE %s SET %s=? WHERE id=?' % (table, new), (scaled, row_id))
|
|
con.execute('CREATE UNIQUE INDEX IF NOT EXISTS uq_orders_order_no ON orders(order_no)')
|
|
|
|
|
|
def merge_approval(con, tables):
|
|
# Check queued tasks against the OLD status before consolidating stages.
|
|
# Completed deliveries/logs and archived PDF bytes must remain untouched.
|
|
if 'notifications' in tables:
|
|
con.execute("""UPDATE notifications SET status='cancelled',
|
|
last_error='审批流程已合并，旧通知已失效', lease_token=NULL, lease_until=NULL
|
|
WHERE channel IN ('review','approval') AND status IN ('pending','retry','sending','failed')
|
|
AND NOT EXISTS (SELECT 1 FROM orders o WHERE o.id=notifications.order_id
|
|
AND o.status=notifications.expected_status AND o.revision=notifications.revision
|
|
AND o.status IN ('pending_review','pending_approval'))""")
|
|
con.execute("""UPDATE notifications SET channel='approval', expected_status='pending_approval',
|
|
event='待审批（流程已合并）' WHERE channel IN ('review','approval')
|
|
AND status IN ('pending','retry','sending','failed')""")
|
|
if 'notification_settings' in tables:
|
|
con.execute("""UPDATE notification_settings SET approval_webhook=
|
|
CASE WHEN COALESCE(approval_webhook,'')='' THEN COALESCE(review_webhook,'')
|
|
ELSE approval_webhook END, review_webhook=''""")
|
|
con.execute("""UPDATE orders SET status='pending_approval', row_version=row_version+1
|
|
WHERE status IN ('pending_review','pending_approval')""")
|
|
con.execute("""UPDATE users SET role='approver', session_version=session_version+1
|
|
WHERE role='reviewer'""")
|