"""Staff order browsing and reversible import batch records.""" import hashlib import json import sqlite3 from datetime import datetime from . import order_data MAX_ORDER_ID = 9223372036854775807 ORDER_LIST_FIELDS = ('tenant_id', 'order_no', 'status', 'inbound_code', 'shipment_id', 'country_code', 'transport_mode', 'warehouse_code', 'source_updated_at', 'snapshot_at') def init_order_admin_tables(connection): connection.execute('''CREATE TABLE IF NOT EXISTS order_import_batches ( id INTEGER PRIMARY KEY AUTOINCREMENT, filename TEXT NOT NULL, imported_by TEXT NOT NULL, imported_at TEXT NOT NULL, total_rows INTEGER NOT NULL, imported_rows INTEGER NOT NULL, preserved_rows INTEGER NOT NULL, state TEXT NOT NULL DEFAULT 'active' CHECK(state IN ('active','undone')), undone_by TEXT, undone_at TEXT, deleted_rows INTEGER NOT NULL DEFAULT 0, preserved_on_undo INTEGER NOT NULL DEFAULT 0 )''') connection.execute('''CREATE TABLE IF NOT EXISTS order_import_batch_items ( batch_id INTEGER NOT NULL, order_id INTEGER NOT NULL, row_fingerprint TEXT NOT NULL, PRIMARY KEY(batch_id, order_id) )''') connection.execute('CREATE INDEX IF NOT EXISTS idx_order_import_batches_time ' 'ON order_import_batches(imported_at DESC, id DESC)') def _row_fingerprint(row): values = [row[field] for field in order_data.FIELDS] encoded = json.dumps(values, ensure_ascii=False, separators=(',', ':'), default=str) return hashlib.sha256(encoded.encode('utf-8')).hexdigest() def _safe_filename(filename): return str(filename or '').replace('\\', '/').split('/')[-1][:255] or 'orders.csv' def create_import_batch(connection, filename, username, parsed, result, existing_keys): """Persist the import audit and fingerprints for newly inserted records.""" now = datetime.now().isoformat(timespec='seconds') cursor = connection.execute('''INSERT INTO order_import_batches (filename,imported_by,imported_at,total_rows,imported_rows,preserved_rows) VALUES(?,?,?,?,?,?)''', (_safe_filename(filename), username, now, parsed['total_rows'], result['imported'], result['preserved'])) batch_id = cursor.lastrowid for item in parsed['rows']: key = (item['tenant_id'], item['order_no'].casefold()) if key in existing_keys: continue row = connection.execute('SELECT id,' + ','.join(order_data.FIELDS) + ' FROM order_test_data WHERE tenant_id=? AND order_no=?', (item['tenant_id'], item['order_no'])).fetchone() if row is None: continue record = dict(zip(('id', *order_data.FIELDS), row)) connection.execute('INSERT INTO order_import_batch_items(batch_id,order_id,row_fingerprint) VALUES(?,?,?)', (batch_id, record['id'], _row_fingerprint(record))) return batch_id def _conditions(query, tenant, status, before): clauses, values = [], [] if query: escaped = query.replace('\\', '\\\\').replace('%', '\\%').replace('_', '\\_') clauses.append("(order_no LIKE ? ESCAPE '\\' OR inbound_code LIKE ? ESCAPE '\\' " "OR shipment_id LIKE ? ESCAPE '\\' OR source_order_id LIKE ? ESCAPE '\\')") values.extend(['%' + escaped + '%'] * 4) if tenant: clauses.append('tenant_id=?') values.append(tenant) if status: clauses.append('status=?') values.append(status) if before is not None: clauses.append('id 128 or len(tenant) > 128: raise ValueError('搜索内容不能超过128个字符。') if status and status not in order_data.STATUS_LABELS: raise ValueError('订单状态无效。') try: limit = int(limit) before = int(before) if before is not None else None except (TypeError, ValueError): raise ValueError('分页参数无效。') from None if not 1 <= limit <= 100 or (before is not None and not 0 < before <= MAX_ORDER_ID): raise ValueError('分页参数无效。') where, values = _conditions(query, tenant, status, None) total = connection.execute('SELECT COUNT(*) FROM order_test_data' + where, values).fetchone()[0] page_where, page_values = _conditions(query, tenant, status, before) rows = connection.execute('SELECT id,' + ','.join(ORDER_LIST_FIELDS) + ' FROM order_test_data' + page_where + ' ORDER BY id DESC LIMIT ?', (*page_values, limit + 1)).fetchall() has_more = len(rows) > limit rows = rows[:limit] orders = [dict(zip(('id', *ORDER_LIST_FIELDS), row)) for row in rows] return {'orders': orders, 'total': total, 'next_before': rows[-1][0] if has_more and rows else None, 'limit': limit} def list_import_batches(connection, limit=50): limit = max(1, min(100, int(limit))) rows = connection.execute('''SELECT id,filename,imported_by,imported_at,total_rows,imported_rows, preserved_rows,state,undone_by,undone_at,deleted_rows,preserved_on_undo FROM order_import_batches ORDER BY id DESC LIMIT ?''', (limit,)).fetchall() fields = ('id','filename','imported_by','imported_at','total_rows','imported_rows', 'preserved_rows','state','undone_by','undone_at','deleted_rows','preserved_on_undo') return [dict(zip(fields, row)) for row in rows] def undo_import_batch(connection, batch_id, username): try: batch_id = int(batch_id) except (TypeError, ValueError): raise ValueError('导入批次编号无效。') from None if not 0 < batch_id <= MAX_ORDER_ID: raise ValueError('导入批次编号无效。') batch = connection.execute('SELECT state,deleted_rows,preserved_on_undo FROM order_import_batches WHERE id=?', (batch_id,)).fetchone() if not batch: return None if batch[0] == 'undone': return {'batch_id': batch_id, 'deleted': batch[1], 'preserved': batch[2], 'already_undone': True} deleted = preserved = 0 items = connection.execute('SELECT order_id,row_fingerprint FROM order_import_batch_items WHERE batch_id=?', (batch_id,)).fetchall() for order_id, fingerprint in items: current = connection.execute('SELECT ' + ','.join(order_data.FIELDS) + ' FROM order_test_data WHERE id=?', (order_id,)).fetchone() if current is None: continue record = dict(zip(order_data.FIELDS, current)) if _row_fingerprint(record) != fingerprint: preserved += 1 continue deleted += connection.execute('DELETE FROM order_test_data WHERE id=?', (order_id,)).rowcount now = datetime.now().isoformat(timespec='seconds') connection.execute('''UPDATE order_import_batches SET state='undone',undone_by=?,undone_at=?, deleted_rows=?,preserved_on_undo=? WHERE id=?''', (username, now, deleted, preserved, batch_id)) return {'batch_id': batch_id, 'deleted': deleted, 'preserved': preserved, 'already_undone': False}