"""SQL Server → PostgreSQL 全量迁移（文档 12.3、12.6）。目标库须为空；任一问题都整体回滚。""" import datetime as dt import uuid from dataclasses import dataclass, field from zoneinfo import ZoneInfo from sqlalchemy import Uuid, insert from sqlalchemy.orm import Session from flyway.domain.clock import NZ from flyway.migration.dates import ( Decision, apply, apply_date, decide_batch_date, decide_slip_date, decide_transfer_date, ) from flyway.migration.logs import parse_log_entries from flyway.migration.profile import parse_expand_attributes, parse_profile from flyway.migration.source import LegacySource from flyway.models import ADMIN_ROLE, AppUser, AppUserRole, LegacyValue from flyway.models.legacy import ( Client, ClientAccount, ClientAccountClaim, ClientAccountLogin, ClientAccountRole, ClientProduct, ClientRole, PackingSlip, Product, ProductBatch, ShippingParcel, Supplier, Template, Transfer, TransferItem, TransferRelation, ) COPY_ORDER = [Client, Product, ProductBatch, Supplier, Template, PackingSlip, Transfer, TransferItem, TransferRelation, ShippingParcel, ClientProduct, ClientRole, ClientAccount, ClientAccountRole, ClientAccountClaim, ClientAccountLogin] UNIQUE_KEYS = { "fw_ClientTransfer": lambda r: None if r.get("TransferFromId") else (r["Code"] or "").strip().lower(), "fw_PackingSlip": lambda r: (r["Reference"] or "").strip().lower(), "fw_ShippingParcels": lambda r: (r["Barcode"] or "").strip().lower(), "fw_ProductBatch": lambda r: (str(r["ProductId"]).lower(), (r["Number"] or "").rstrip(' ').lower()), } class MigrationBlocked(Exception): def __init__(self, pending: list[dict], duplicates: list[str]): super().__init__(f"{len(pending)} pending dates, {len(duplicates)} duplicates") self.pending, self.duplicates = pending, duplicates @dataclass class MigrationResult: pending: list[dict] = field(default_factory=list) duplicates: list[str] = field(default_factory=list) merged_batches: dict[str, str] = field(default_factory=dict) # 被合并批次 ID → 保留批次 ID renamed: list[str] = field(default_factory=list) def _merge_batches(tables: dict[str, list[dict]], legacy_values: list[dict]) -> dict[str, str]: """同一产品下批号去尾部空格、忽略大小写后相同的批次合并；前导空格按旧版保留为不同批次。 须经业务确认后才用于生产迁移（文档 10.6、第 11 节补充）。""" groups: dict[tuple, list[dict]] = {} for r in tables["fw_ProductBatch"]: groups.setdefault(UNIQUE_KEYS["fw_ProductBatch"](r), []).append(r) merged: dict[str, str] = {} for rows in groups.values(): if len(rows) < 2: continue keeper, *dups = sorted(rows, key=lambda r: r["Date"]) keeper["Number"] = keeper["Number"].rstrip(' ') for d in dups: merged[_key(d["ID"])] = _key(keeper["ID"]) legacy_values.append({"table_name": "fw_ProductBatch", "row_id": str(d["ID"]), "column_name": "ID", "raw_value": f'{d["ID"]} Number={d["Number"]!r}', "resolution": "merged", "reason": f"merged into {keeper['ID']}"}) if merged: tables["fw_ProductBatch"] = [r for r in tables["fw_ProductBatch"] if _key(r["ID"]) not in merged] for it in tables["fw_ClientTransferItem"]: it["BatchId"] = merged.get(_key(it["BatchId"]), it["BatchId"]) return merged RENAME_COLUMNS = {"fw_ClientTransfer": ("Code", "Date"), "fw_PackingSlip": ("Reference", "Date"), "fw_ShippingParcels": ("Barcode", "CreatedUtc")} def rename_duplicate_codes(tables: dict[str, list[dict]], legacy_values: list[dict]) -> list[str]: """单号 / 装箱单号 / 条码去首尾空格、忽略大小写后重复的不同记录：保留最早的一条，其余改为“原值-DUPn”，原值写入 migration_legacy_value。调拨转入单跟随转出单改名。须经业务确认后才用于生产迁移（旧库只差开头空格即视为不同值）。""" renamed = [] for table, (column, order) in RENAME_COLUMNS.items(): groups: dict = {} for r in tables[table]: k = UNIQUE_KEYS[table](r) if k is not None: groups.setdefault(k, []).append(r) for rows in groups.values(): if len(rows) < 2: continue keeper, *dups = sorted(rows, key=lambda r: r[order]) for n, d in enumerate(dups, start=1): old, new = d[column], f"{(d[column] or '').strip()}-DUP{n}" targets = [d] + [t for t in tables[table] if table == "fw_ClientTransfer" and _key(t.get("TransferFromId")) == _key(d["ID"])] for t in targets: t[column] = new legacy_values.append({"table_name": table, "row_id": str(t["ID"]), "column_name": column, "raw_value": old, "resolution": "renamed", "reason": f"duplicate of {keeper['ID']} ({keeper[column]!r})"}) renamed.append(f"{table}: {d['ID']} {old!r} -> {new!r} (kept {keeper['ID']})") return renamed def _key(value) -> str | None: return str(value).lower() if value else None def _find_duplicates(tables: dict[str, list[dict]]) -> list[str]: found = [] for table, key in UNIQUE_KEYS.items(): seen: dict = {} for row in tables.get(table, []): k = key(row) if k is None: continue if k in seen: found.append(f"{table}: {seen[k]} / {row['ID']} share {k!r}") seen.setdefault(k, row["ID"]) return found def _coerce(model, rows: list[dict]) -> list[dict]: """pyodbc 读出的 GUID 是字符串；按目标列类型转成 uuid.UUID。日期列由调用方先处理。""" uuid_cols = [c.name for c in model.__table__.columns if isinstance(c.type, Uuid)] for row in rows: for name in uuid_cols: if isinstance(row.get(name), str): row[name] = uuid.UUID(row[name]) return rows def _topological(rows: list[dict], key: str) -> list[dict]: """自引用外键（TransferFromId、RelatedTransferItemId）：被引用的行先插入。""" done, ordered, rest = set(), [], list(rows) while rest: ready = [r for r in rest if not r.get(key) or _key(r[key]) in done] if not ready: raise ValueError(f"cyclic or dangling {key}") done.update(_key(r["ID"]) for r in ready) ordered += ready ready_ids = {id(r) for r in ready} rest = [r for r in rest if id(r) not in ready_ids] return ordered def migrate(session: Session, source: LegacySource, *, server_tz: ZoneInfo, resolutions: dict[tuple[str, str, str], str], allow_pending: bool, merge_duplicate_batches: bool = False, rename_duplicates: bool = False) -> MigrationResult: tables = {m.__tablename__: source.rows(m.__tablename__) for m in COPY_ORDER} legacy_values: list[dict] = [] pending: list[dict] = [] merged = _merge_batches(tables, legacy_values) if merge_duplicate_batches else {} renamed = rename_duplicate_codes(tables, legacy_values) if rename_duplicates else [] duplicates = _find_duplicates(tables) if duplicates: raise MigrationBlocked([], duplicates) def decide(table: str, row: dict, column: str, decision: Decision) -> Decision: override = resolutions.get((table, str(row["ID"]), column)) if override: decision = Decision(override, "resolved by business") if decision.resolution == "pending": raw = row[column] # 文档 12.6 第 4 条：异常清单列出按两种来源换算后的新西兰时间，供业务对照实际单据判定 as_server = apply(raw, Decision("server", ""), server_tz).astimezone(NZ) as_input = apply(raw, Decision("user_nz", ""), server_tz) pending.append({"table": table, "row_id": str(row["ID"]), "column": column, "raw": raw.isoformat(), "code": row.get("Code") or row.get("Reference") or "", "reason": decision.reason, "nz_if_server_time": as_server.strftime("%Y-%m-%d %H:%M:%S"), "nz_if_user_input": as_input.strftime("%Y-%m-%d %H:%M:%S"), "resolution": ""}) # 业务填 server / user_nz 后，本文件可直接作为 --resolutions 传入 legacy_values.append({"table_name": table, "row_id": str(row["ID"]), "column_name": column, "raw_value": row[column].isoformat(), "resolution": decision.resolution, "reason": decision.reason}) return decision # 装箱单（先保存原始日期，供出库单“日期等于装箱单日期”比较） raw_slips = {_key(r["ID"]): {"Date": r["Date"]} for r in tables["fw_PackingSlip"]} slip_decisions = {} for r in tables["fw_PackingSlip"]: d = decide("fw_PackingSlip", r, "Date", decide_slip_date(r, server_tz)) slip_decisions[_key(r["ID"])] = d r["Date"] = apply(r["Date"], d, server_tz) if r.get("Completed"): done = decide("fw_PackingSlip", r, "Completed", Decision("server", "Completed is server time")) r["Completed"] = apply(r["Completed"], done, server_tz) # 单据：转出单先判定，转入单跟随 transfers = _topological(tables["fw_ClientTransfer"], "TransferFromId") out_decisions = {} for r in transfers: key = {**r, "PackingSlipId": _key(r.get("PackingSlipId")), "TransferFromId": _key(r.get("TransferFromId"))} d = decide("fw_ClientTransfer", r, "Date", decide_transfer_date(key, raw_slips, slip_decisions, out_decisions, server_tz)) out_decisions[_key(r["ID"])] = d r["Date"] = apply(r["Date"], d, server_tz) if r.get("DepartureDate"): dep = decide("fw_ClientTransfer", r, "DepartureDate", Decision("pending", "unused column")) r["DepartureDate"] = apply(r["DepartureDate"], dep, server_tz) tables["fw_ClientTransfer"] = transfers # 批次：与包含该批次的入库单 Logs 时间比对 load_in_ids = {_key(r["ID"]) for r in transfers if not r["IsLoadOut"]} logs_by_transfer = {_key(r["ID"]): [e.at for e in parse_log_entries(r.get("Logs"))] for r in transfers} times_by_batch: dict[str, list] = {} for it in tables["fw_ClientTransferItem"]: tid = _key(it["TransferId"]) if tid in load_in_ids: times_by_batch.setdefault(_key(it["BatchId"]), []).extend(logs_by_transfer.get(tid, [])) for r in tables["fw_ProductBatch"]: d = decide("fw_ProductBatch", r, "Date", decide_batch_date(r, times_by_batch.get(_key(r["ID"]), []), server_tz)) r['display_date'] = r['Date'] r["Date"] = apply_date(r["Date"], d, server_tz) if r.get("UsedByDate"): r["UsedByDate"] = apply_date(r["UsedByDate"], Decision("date", "calendar date"), server_tz) for r in tables["fw_ShippingParcels"]: for col in ("CreatedUtc", "ShippedUtc", "DeletedUtc"): if r.get(col): r[col] = apply(r[col], decide("fw_ShippingParcels", r, col, Decision("utc", "UTC column")), server_tz) for r in tables["fw_ClientProduct"]: for col in ("OpeningDate", "ClosingDate"): r['opening_display_date' if col == 'OpeningDate' else 'closing_display_date'] = r[col] decide('fw_ClientProduct', r, col, Decision('date', 'calendar date; original time retained for display')) r[col] = apply_date(r[col], Decision("date", "calendar date"), server_tz) if pending and not allow_pending: raise MigrationBlocked(pending, []) tables["fw_ClientTransferItem"] = _topological(tables["fw_ClientTransferItem"], "RelatedTransferItemId") for model in COPY_ORDER: rows = _coerce(model, tables[model.__tablename__]) if rows: session.execute(insert(model.__table__), rows) _migrate_users(session, source) if legacy_values: session.execute(insert(LegacyValue.__table__), legacy_values) session.flush() return MigrationResult(pending=pending, merged_batches=merged, renamed=renamed) def _utc(value): return value.replace(tzinfo=dt.UTC) if value else None def _migrate_users(session: Session, source: LegacySource) -> None: """员工：账号与状态迁入，密码不迁移（上线重设）；Membership 时间按 UTC（12.3 非业务数据处理表）。""" members = {_key(m["UserId"]): m for m in source.rows("aspnet_Membership")} roles = {_key(r["RoleId"]): r["RoleName"] for r in source.rows("aspnet_Roles")} user_roles: dict[str, list[str]] = {} for ur in source.rows("aspnet_UsersInRoles"): user_roles.setdefault(_key(ur["UserId"]), []).append(roles.get(_key(ur["RoleId"]), "")) profiles = {_key(p["UserId"]): parse_expand_attributes( parse_profile(p.get("PropertyNames"), p.get("PropertyValuesString")).get("ExpandAttribute")) for p in source.rows("aspnet_Profile")} for u in source.rows("aspnet_Users"): key = _key(u["UserId"]) m, prof = members.get(key, {}), profiles.get(key, {}) user = AppUser( id=uuid.UUID(key), username=u["UserName"], email=m.get("Email") or "", first_name=prof.get("FirstName", ""), last_name=prof.get("LastName", ""), mobile=prof.get("MobileNumber", ""), password_hash=None, is_active=bool(m.get("IsApproved", True)), is_locked_out=bool(m.get("IsLockedOut", False)), created_at=_utc(m.get("CreateDate")) or dt.datetime.now(dt.UTC), last_login_at=_utc(m.get("LastLoginDate")), last_activity_at=_utc(u.get("LastActivityDate")), ) user.roles = [AppUserRole(role=ADMIN_ROLE)] if ADMIN_ROLE in user_roles.get(key, []) else [] session.add(user)