【系统架构设计百科】遗留系统现代化：评估框架与渐进策略

from dataclasses import dataclass, field
from enum import Enum
from typing import List


class TimeStrategy(Enum):
    TOLERATE = "容忍"
    INVEST = "投资"
    MIGRATE = "迁移"
    ELIMINATE = "淘汰"


@dataclass
class AssessmentDimension:
    name: str
    weight: float
    score: int  # 1-5
    category: str  # "business" or "technical"


@dataclass
class SystemAssessment:
    system_name: str
    dimensions: List[AssessmentDimension] = field(default_factory=list)

    def business_fitness(self) -> float:
        biz_dims = [d for d in self.dimensions if d.category == "business"]
        if not biz_dims:
            return 0.0
        total_weight = sum(d.weight for d in biz_dims)
        return sum(d.weight * d.score for d in biz_dims) / total_weight

    def technical_fitness(self) -> float:
        tech_dims = [d for d in self.dimensions if d.category == "technical"]
        if not tech_dims:
            return 0.0
        total_weight = sum(d.weight for d in tech_dims)
        return sum(d.weight * d.score for d in tech_dims) / total_weight

    def recommend_strategy(self, threshold: float = 3.0) -> TimeStrategy:
        biz = self.business_fitness()
        tech = self.technical_fitness()
        if biz >= threshold and tech >= threshold:
            return TimeStrategy.INVEST
        elif biz >= threshold and tech < threshold:
            return TimeStrategy.MIGRATE
        elif biz < threshold and tech >= threshold:
            return TimeStrategy.TOLERATE
        else:
            return TimeStrategy.ELIMINATE


# 示例：评估银行核心账务系统
core_banking = SystemAssessment(system_name="核心账务系统")
core_banking.dimensions = [
    AssessmentDimension("业务覆盖度", 0.20, 5, "business"),
    AssessmentDimension("收入贡献度", 0.15, 5, "business"),
    AssessmentDimension("技术栈活力", 0.15, 1, "technical"),
    AssessmentDimension("运维成本", 0.10, 2, "technical"),
    AssessmentDimension("人才可得性", 0.15, 1, "technical"),
    AssessmentDimension("集成能力", 0.10, 2, "technical"),
    AssessmentDimension("安全合规", 0.10, 3, "technical"),
    AssessmentDimension("可扩展性", 0.05, 1, "technical"),
]

strategy = core_banking.recommend_strategy()
print(f"系统：{core_banking.system_name}")
print(f"业务适合度：{core_banking.business_fitness():.2f}")
print(f"技术适合度：{core_banking.technical_fitness():.2f}")
print(f"建议策略：{strategy.value}")
# 输出：建议策略：迁移

/**
 * 封装遗留 COBOL 交易处理系统的 REST API 网关
 */
@RestController
@RequestMapping("/api/v1/transactions")
public class TransactionGateway {

    private final LegacyCobolBridge cobolBridge;
    private final MessageConverter converter;

    public TransactionGateway(LegacyCobolBridge cobolBridge,
                               MessageConverter converter) {
        this.cobolBridge = cobolBridge;
        this.converter = converter;
    }

    @PostMapping("/transfer")
    public ResponseEntity<TransferResult> transfer(
            @RequestBody TransferRequest request) {
        // 将现代 JSON 请求转换为 COBOL copybook 格式
        CobolCopybook copybook = converter.toCobolFormat(request);

        // 通过 MQ 调用大型机交易
        CobolResponse response = cobolBridge.executeTransaction(
            "TRFR0100",  // COBOL 程序名
            copybook
        );

        // 将 COBOL 响应转换回 JSON
        TransferResult result = converter.fromCobolFormat(response);
        return ResponseEntity.ok(result);
    }
}

# 将遗留的 Java EE 应用容器化
# Dockerfile
FROM websphere-liberty:24.0.0.3-full-java17

COPY --chown=1001:0 server.xml /config/server.xml
COPY --chown=1001:0 legacy-app.ear /config/apps/

# 保留原有 JNDI 数据源配置
COPY --chown=1001:0 jvm.options /config/jvm.options

ENV LEGACY_DB_HOST=oracle-legacy.internal
ENV LEGACY_DB_PORT=1521
ENV LEGACY_DB_SID=PRODDB

EXPOSE 9080 9443

// 重构前：紧耦合的单体处理函数
func processOrder(db *sql.DB, orderData []byte) error {
    // 解析订单、校验库存、计算价格、扣减库存、
    // 生成物流单、发送通知...全部在一个函数里
    // 500+ 行代码
    return nil
}

// 重构后：按领域拆分为独立服务接口
type OrderService interface {
    CreateOrder(ctx context.Context, req CreateOrderRequest) (*Order, error)
}

type InventoryService interface {
    CheckStock(ctx context.Context, sku string, qty int) (bool, error)
    Reserve(ctx context.Context, sku string, qty int) (*Reservation, error)
}

type PricingService interface {
    Calculate(ctx context.Context, items []OrderItem) (*PriceResult, error)
}

type ShippingService interface {
    CreateShipment(ctx context.Context, order *Order) (*Shipment, error)
}

type NotificationService interface {
    Send(ctx context.Context, event Event) error
}

// ============================================
// 新系统的领域模型（干净的、不受遗留系统污染的）
// ============================================

interface Customer {
  id: string;
  fullName: string;
  email: string;
  phoneNumber: string;
  kycStatus: KycStatus;
  riskLevel: RiskLevel;
  accounts: Account[];
  createdAt: Date;
}

enum KycStatus {
  PENDING = "PENDING",
  VERIFIED = "VERIFIED",
  EXPIRED = "EXPIRED",
  REJECTED = "REJECTED",
}

enum RiskLevel {
  LOW = "LOW",
  MEDIUM = "MEDIUM",
  HIGH = "HIGH",
}

interface Account {
  accountId: string;
  accountType: AccountType;
  currency: string;
  balance: number;
  status: AccountStatus;
}

// ============================================
// 遗留系统的数据结构（来自大型机 COBOL copybook）
// ============================================

interface LegacyCustRecord {
  CUST_NO: string;       // 客户编号（数字，左补零）
  CUST_NM_LAST: string;  // 姓（大写，右补空格）
  CUST_NM_FIRST: string; // 名（大写，右补空格）
  CUST_EMAIL: string;    // 可能为空或"N/A"
  CUST_PHONE: string;    // 格式不统一
  KYC_FLAG: string;      // "Y"/"N"/"E"/"R"
  RISK_CD: string;       // "01"/"02"/"03"
  ACCT_LIST: LegacyAcctRecord[];
  CUST_OPEN_DT: string;  // YYYYMMDD 格式
}

interface LegacyAcctRecord {
  ACCT_NO: string;
  ACCT_TYP_CD: string;   // "SA"/"CA"/"TD"/"LN"
  CCY_CD: string;
  ACCT_BAL: string;      // 字符串表示的数字，最后两位为小数
  ACCT_STS: string;      // "A"/"C"/"F"/"D"
}

// ============================================
// ACL 翻译器：遗留数据 → 领域模型
// ============================================

class CustomerTranslator {
  translateCustomer(legacy: LegacyCustRecord): Customer {
    return {
      id: this.normalizeCustomerId(legacy.CUST_NO),
      fullName: this.translateName(legacy.CUST_NM_FIRST, legacy.CUST_NM_LAST),
      email: this.translateEmail(legacy.CUST_EMAIL),
      phoneNumber: this.normalizePhone(legacy.CUST_PHONE),
      kycStatus: this.translateKycStatus(legacy.KYC_FLAG),
      riskLevel: this.translateRiskLevel(legacy.RISK_CD),
      accounts: legacy.ACCT_LIST.map((a) => this.translateAccount(a)),
      createdAt: this.parseDate(legacy.CUST_OPEN_DT),
    };
  }

  private normalizeCustomerId(custNo: string): string {
    return `CUST-${custNo.replace(/^0+/, "")}`;
  }

  private translateName(first: string, last: string): string {
    const capitalize = (s: string) =>
      s.trim().charAt(0).toUpperCase() +
      s.trim().slice(1).toLowerCase();
    return `${capitalize(first)} ${capitalize(last)}`;
  }

  private translateEmail(email: string): string {
    if (!email || email.trim() === "" || email.trim() === "N/A") {
      return "";
    }
    return email.trim().toLowerCase();
  }

  private normalizePhone(phone: string): string {
    return phone.replace(/[^\d+]/g, "");
  }

  private translateKycStatus(flag: string): KycStatus {
    const mapping: Record<string, KycStatus> = {
      Y: KycStatus.VERIFIED,
      N: KycStatus.PENDING,
      E: KycStatus.EXPIRED,
      R: KycStatus.REJECTED,
    };
    return mapping[flag] ?? KycStatus.PENDING;
  }

  private translateRiskLevel(code: string): RiskLevel {
    const mapping: Record<string, RiskLevel> = {
      "01": RiskLevel.LOW,
      "02": RiskLevel.MEDIUM,
      "03": RiskLevel.HIGH,
    };
    return mapping[code] ?? RiskLevel.MEDIUM;
  }

  private translateAccount(legacy: LegacyAcctRecord): Account {
    return {
      accountId: legacy.ACCT_NO.trim(),
      accountType: this.translateAccountType(legacy.ACCT_TYP_CD),
      currency: legacy.CCY_CD.trim(),
      balance: this.parseBalance(legacy.ACCT_BAL),
      status: this.translateAccountStatus(legacy.ACCT_STS),
    };
  }

  private translateAccountType(code: string): AccountType {
    const mapping: Record<string, AccountType> = {
      SA: AccountType.SAVINGS,
      CA: AccountType.CHECKING,
      TD: AccountType.TERM_DEPOSIT,
      LN: AccountType.LOAN,
    };
    return mapping[code] ?? AccountType.CHECKING;
  }

  private parseBalance(balStr: string): number {
    const cleaned = balStr.replace(/[^\d-]/g, "");
    return parseInt(cleaned, 10) / 100;
  }

  private translateAccountStatus(code: string): AccountStatus {
    const mapping: Record<string, AccountStatus> = {
      A: AccountStatus.ACTIVE,
      C: AccountStatus.CLOSED,
      F: AccountStatus.FROZEN,
      D: AccountStatus.DORMANT,
    };
    return mapping[code] ?? AccountStatus.ACTIVE;
  }

  private parseDate(dateStr: string): Date {
    const year = parseInt(dateStr.substring(0, 4), 10);
    const month = parseInt(dateStr.substring(4, 6), 10) - 1;
    const day = parseInt(dateStr.substring(6, 8), 10);
    return new Date(year, month, day);
  }
}

enum AccountType {
  SAVINGS = "SAVINGS",
  CHECKING = "CHECKING",
  TERM_DEPOSIT = "TERM_DEPOSIT",
  LOAN = "LOAN",
}

enum AccountStatus {
  ACTIVE = "ACTIVE",
  CLOSED = "CLOSED",
  FROZEN = "FROZEN",
  DORMANT = "DORMANT",
}

// ============================================
// ACL 门面：封装与遗留系统的通信
// ============================================

class LegacyCustomerFacade {
  private mqClient: MQClient;
  private translator: CustomerTranslator;

  constructor(mqClient: MQClient, translator: CustomerTranslator) {
    this.mqClient = mqClient;
    this.translator = translator;
  }

  async getCustomer(customerId: string): Promise<Customer> {
    const legacyId = this.toLegacyId(customerId);
    const request = this.buildCicsRequest("CINQ0100", legacyId);

    const response = await this.mqClient.sendAndReceive(
      "LEGACY.CUST.REQ",
      "LEGACY.CUST.RESP",
      request,
      { timeout: 5000 }
    );

    const legacyRecord = this.parseCobolResponse(response);
    return this.translator.translateCustomer(legacyRecord);
  }

  private toLegacyId(customerId: string): string {
    const numericId = customerId.replace("CUST-", "");
    return numericId.padStart(10, "0");
  }

  private buildCicsRequest(programName: string, data: string): Buffer {
    // 构建 CICS 交易请求的二进制报文
    const header = Buffer.alloc(32);
    header.write(programName, 0, 8, "ascii");
    header.writeUInt32BE(data.length, 8);
    return Buffer.concat([header, Buffer.from(data, "ascii")]);
  }

  private parseCobolResponse(response: Buffer): LegacyCustRecord {
    // 按照 COBOL copybook 定义的固定长度字段解析
    // 这里简化处理，实际需要严格按字段偏移量解析
    return JSON.parse(response.toString());
  }
}

import { describe, it, expect } from "vitest";

describe("CustomerTranslator", () => {
  const translator = new CustomerTranslator();

  describe("translateCustomer", () => {
    it("应正确转换标准客户记录", () => {
      const legacy: LegacyCustRecord = {
        CUST_NO: "0000012345",
        CUST_NM_LAST: "ZHANG         ",
        CUST_NM_FIRST: "SAN           ",
        CUST_EMAIL: "zhang.san@example.com",
        CUST_PHONE: "(021) 1234-5678",
        KYC_FLAG: "Y",
        RISK_CD: "01",
        ACCT_LIST: [
          {
            ACCT_NO: "6222021234567890",
            ACCT_TYP_CD: "SA",
            CCY_CD: "CNY",
            ACCT_BAL: "0000012345",
            ACCT_STS: "A",
          },
        ],
        CUST_OPEN_DT: "20050315",
      };

      const customer = translator.translateCustomer(legacy);

      expect(customer.id).toBe("CUST-12345");
      expect(customer.fullName).toBe("San Zhang");
      expect(customer.email).toBe("zhang.san@example.com");
      expect(customer.phoneNumber).toBe("02112345678");
      expect(customer.kycStatus).toBe(KycStatus.VERIFIED);
      expect(customer.riskLevel).toBe(RiskLevel.LOW);
      expect(customer.accounts).toHaveLength(1);
      expect(customer.accounts[0].balance).toBe(123.45);
      expect(customer.accounts[0].accountType).toBe(AccountType.SAVINGS);
    });

    it("应处理缺失的电子邮件", () => {
      const legacy = createLegacyRecord({ CUST_EMAIL: "N/A" });
      const customer = translator.translateCustomer(legacy);
      expect(customer.email).toBe("");
    });

    it("应处理未知的 KYC 标识", () => {
      const legacy = createLegacyRecord({ KYC_FLAG: "X" });
      const customer = translator.translateCustomer(legacy);
      expect(customer.kycStatus).toBe(KycStatus.PENDING);
    });
  });
});

# Debezium CDC 连接器配置
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: legacy-db-connector
spec:
  class: io.debezium.connector.oracle.OracleConnector
  tasksMax: 1
  config:
    database.hostname: legacy-oracle.internal
    database.port: 1521
    database.dbname: LEGACYDB
    database.user: cdc_reader
    database.password: ${CDC_PASSWORD}
    schema.include.list: LEGACY_SCHEMA
    table.include.list: >
      LEGACY_SCHEMA.CUSTOMER,
      LEGACY_SCHEMA.ACCOUNT,
      LEGACY_SCHEMA.TRANSACTION
    topic.prefix: legacy-cdc
    snapshot.mode: schema_only
    log.mining.strategy: online_catalog
    transforms: route
    transforms.route.type: >
      io.debezium.transforms.ByLogicalTableRouter
    transforms.route.topic.regex: (.*)
    transforms.route.topic.replacement: legacy-changes.$1

import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from typing import Iterator, Optional

logger = logging.getLogger(__name__)


@dataclass
class MigrationRecord:
    source_id: str
    source_table: str
    target_table: str
    status: str  # "pending" | "migrated" | "failed" | "validated"
    error_message: Optional[str] = None
    migrated_at: Optional[datetime] = None


class DataMigrator(ABC):
    """数据迁移器基类：提取 → 转换 → 加载"""

    @abstractmethod
    def extract(self, batch_size: int) -> Iterator[dict]:
        """从源系统提取数据"""
        pass

    @abstractmethod
    def transform(self, record: dict) -> dict:
        """转换数据格式和语义"""
        pass

    @abstractmethod
    def load(self, record: dict) -> bool:
        """加载到目标系统"""
        pass

    @abstractmethod
    def validate(self, source: dict, target: dict) -> bool:
        """校验迁移结果"""
        pass

    def migrate(self, batch_size: int = 1000) -> dict:
        stats = {"total": 0, "success": 0, "failed": 0, "skipped": 0}

        for batch in self.extract(batch_size):
            stats["total"] += 1
            try:
                transformed = self.transform(batch)
                if transformed is None:
                    stats["skipped"] += 1
                    continue
                success = self.load(transformed)
                if success:
                    stats["success"] += 1
                else:
                    stats["failed"] += 1
            except Exception as e:
                stats["failed"] += 1
                logger.error(
                    "迁移失败：source_id=%s，错误=%s",
                    batch.get("id", "unknown"),
                    str(e),
                )

        return stats


class CustomerMigrator(DataMigrator):
    """客户数据迁移器"""

    def __init__(self, source_db, target_db, mapping_rules):
        self.source_db = source_db
        self.target_db = target_db
        self.mapping_rules = mapping_rules

    def extract(self, batch_size: int) -> Iterator[dict]:
        offset = 0
        while True:
            rows = self.source_db.query(
                """
                SELECT c.*, a.ACCT_NO, a.ACCT_BAL
                FROM CUSTOMER c
                LEFT JOIN ACCOUNT a ON c.CUST_NO = a.CUST_NO
                WHERE c.MIGRATION_FLAG != 'Y'
                ORDER BY c.CUST_NO
                OFFSET :offset ROWS FETCH NEXT :limit ROWS ONLY
                """,
                offset=offset,
                limit=batch_size,
            )
            if not rows:
                break
            for row in rows:
                yield dict(row)
            offset += batch_size

    def transform(self, record: dict) -> dict:
        # 数据清洗与格式转换
        return {
            "customer_id": f"CUST-{record['CUST_NO'].lstrip('0')}",
            "full_name": self._normalize_name(record),
            "email": self._clean_email(record.get("CUST_EMAIL")),
            "phone": self._normalize_phone(record.get("CUST_PHONE")),
            "created_at": self._parse_date(record.get("CUST_OPEN_DT")),
            "source_system": "LEGACY_CORE",
            "source_id": record["CUST_NO"],
        }

    def load(self, record: dict) -> bool:
        return self.target_db.upsert("customers", record, key="customer_id")

    def validate(self, source: dict, target: dict) -> bool:
        return (
            target is not None
            and source["CUST_NO"].lstrip("0") in target["customer_id"]
        )

    def _normalize_name(self, record: dict) -> str:
        first = record.get("CUST_NM_FIRST", "").strip().title()
        last = record.get("CUST_NM_LAST", "").strip().title()
        return f"{first} {last}".strip()

    def _clean_email(self, email: Optional[str]) -> str:
        if not email or email.strip() in ("", "N/A", "NONE"):
            return ""
        return email.strip().lower()

    def _normalize_phone(self, phone: Optional[str]) -> str:
        if not phone:
            return ""
        import re
        return re.sub(r"[^\d+]", "", phone)

    def _parse_date(self, date_str: Optional[str]) -> Optional[datetime]:
        if not date_str or len(date_str) != 8:
            return None
        try:
            return datetime.strptime(date_str, "%Y%m%d")
        except ValueError:
            return None

-- 第一层：记录数校验
SELECT
    'CUSTOMER' AS table_name,
    (SELECT COUNT(*) FROM legacy_schema.customer) AS source_count,
    (SELECT COUNT(*) FROM new_schema.customers) AS target_count,
    (SELECT COUNT(*) FROM legacy_schema.customer) -
    (SELECT COUNT(*) FROM new_schema.customers) AS diff
UNION ALL
SELECT
    'ACCOUNT',
    (SELECT COUNT(*) FROM legacy_schema.account),
    (SELECT COUNT(*) FROM new_schema.accounts),
    (SELECT COUNT(*) FROM legacy_schema.account) -
    (SELECT COUNT(*) FROM new_schema.accounts);

-- 第二层：金额校验（关键业务字段）
SELECT
    'ACCOUNT_BALANCE' AS check_name,
    (SELECT SUM(CAST(ACCT_BAL AS DECIMAL(18,2)) / 100)
     FROM legacy_schema.account
     WHERE ACCT_STS = 'A') AS source_total,
    (SELECT SUM(balance)
     FROM new_schema.accounts
     WHERE status = 'ACTIVE') AS target_total;

-- 第三层：抽样详细校验
SELECT
    l.CUST_NO AS legacy_id,
    n.customer_id AS new_id,
    l.CUST_NM_FIRST || ' ' || l.CUST_NM_LAST AS legacy_name,
    n.full_name AS new_name,
    CASE WHEN TRIM(l.CUST_NM_FIRST) || ' ' || TRIM(l.CUST_NM_LAST) = n.full_name
         THEN 'MATCH' ELSE 'MISMATCH' END AS name_check
FROM legacy_schema.customer l
JOIN new_schema.customers n
    ON 'CUST-' || LTRIM(l.CUST_NO, '0') = n.customer_id
WHERE l.CUST_NO IN (
    SELECT CUST_NO FROM legacy_schema.customer
    ORDER BY DBMS_RANDOM.VALUE
    FETCH FIRST 100 ROWS ONLY
);

import re
from dataclasses import dataclass, field
from typing import List


@dataclass
class BusinessRule:
    rule_id: str
    program: str
    paragraph: str
    condition: str
    action: str
    confidence: float
    source_lines: List[int] = field(default_factory=list)


class CobolRuleExtractor:
    """从 COBOL 源代码中提取业务规则"""

    # COBOL 条件语句模式
    IF_PATTERN = re.compile(
        r"IF\s+(.+?)\s+(THEN\s+)?(.+?)(?:ELSE\s+(.+?))?END-IF",
        re.DOTALL | re.IGNORECASE,
    )

    EVALUATE_PATTERN = re.compile(
        r"EVALUATE\s+(.+?)\s+((?:WHEN\s+.+?\s+.+?\s*)+)END-EVALUATE",
        re.DOTALL | re.IGNORECASE,
    )

    PERFORM_PATTERN = re.compile(
        r"PERFORM\s+(\S+)\s+(?:UNTIL|VARYING|THRU)",
        re.IGNORECASE,
    )

    def extract_rules(self, source_code: str, program_name: str) -> List[BusinessRule]:
        rules = []
        lines = source_code.split("\n")

        for i, line in enumerate(lines):
            # 跳过注释行（第 7 列为 *）
            if len(line) > 6 and line[6] == "*":
                continue

            # 提取 IF 条件中的业务规则
            for match in self.IF_PATTERN.finditer(source_code):
                condition = self._clean_condition(match.group(1))
                action = self._clean_action(match.group(3))

                if self._is_business_rule(condition):
                    rule = BusinessRule(
                        rule_id=f"{program_name}-R{len(rules)+1:04d}",
                        program=program_name,
                        paragraph=self._find_paragraph(lines, match.start()),
                        condition=condition,
                        action=action,
                        confidence=self._assess_confidence(condition, action),
                    )
                    rules.append(rule)

        return rules

    def _is_business_rule(self, condition: str) -> bool:
        """判断条件是否为业务规则（而非技术逻辑）"""
        business_indicators = [
            "ACCT-TYPE", "CUST-TYPE", "TXN-AMT", "RATE",
            "FEE", "LIMIT", "STATUS", "GRADE", "LEVEL",
            "BALANCE", "CREDIT", "DEBIT", "INTEREST",
        ]
        return any(ind in condition.upper() for ind in business_indicators)

    def _clean_condition(self, condition: str) -> str:
        return " ".join(condition.split())

    def _clean_action(self, action: str) -> str:
        return " ".join(action.split())

    def _find_paragraph(self, lines: List[str], char_offset: int) -> str:
        current_offset = 0
        current_paragraph = "UNKNOWN"
        for line in lines:
            if re.match(r"^\s{7}\S+\.\s*$", line):
                current_paragraph = line.strip().rstrip(".")
            current_offset += len(line) + 1
            if current_offset > char_offset:
                break
        return current_paragraph

    def _assess_confidence(self, condition: str, action: str) -> float:
        score = 0.5
        if any(kw in condition.upper() for kw in ["GREATER", "LESS", "EQUAL"]):
            score += 0.2
        if any(kw in action.upper() for kw in ["COMPUTE", "MOVE", "ADD"]):
            score += 0.1
        return min(score, 1.0)

@Service
public class DualWriteTransactionService {

    private final LegacyTransactionGateway legacyGateway;
    private final NewTransactionService newService;
    private final ReconciliationService reconciliation;
    private final FeatureFlagService featureFlags;

    @Transactional
    public TransactionResult processTransfer(TransferRequest request) {
        String mode = featureFlags.getMode("transfer_migration");

        switch (mode) {
            case "LEGACY_PRIMARY":
                return legacyPrimaryWithShadowWrite(request);
            case "DUAL_WRITE_VERIFY":
                return dualWriteWithVerification(request);
            case "NEW_PRIMARY":
                return newPrimaryWithLegacySync(request);
            case "NEW_ONLY":
                return newService.processTransfer(request);
            default:
                return legacyGateway.processTransfer(request);
        }
    }

    private TransactionResult legacyPrimaryWithShadowWrite(
            TransferRequest request) {
        // 遗留系统为主，新系统异步影子写入
        TransactionResult result = legacyGateway.processTransfer(request);

        CompletableFuture.runAsync(() -> {
            try {
                newService.shadowWrite(request, result);
            } catch (Exception e) {
                log.warn("影子写入失败，记录待对账：{}", e.getMessage());
                reconciliation.recordDiscrepancy(request, result, e);
            }
        });

        return result;
    }

    private TransactionResult dualWriteWithVerification(
            TransferRequest request) {
        // 双写并比对结果
        TransactionResult legacyResult =
            legacyGateway.processTransfer(request);
        TransactionResult newResult =
            newService.processTransfer(request);

        if (!reconciliation.verify(legacyResult, newResult)) {
            log.error("双写结果不一致：legacy={}，new={}",
                legacyResult, newResult);
            reconciliation.recordDiscrepancy(
                request, legacyResult, newResult);
            // 以遗留系统结果为准，回滚新系统
            newService.compensate(request, newResult);
        }

        return legacyResult;
    }

    private TransactionResult newPrimaryWithLegacySync(
            TransferRequest request) {
        // 新系统为主，异步同步到遗留系统
        TransactionResult result = newService.processTransfer(request);

        CompletableFuture.runAsync(() -> {
            try {
                legacyGateway.syncTransaction(request, result);
            } catch (Exception e) {
                log.error("遗留系统同步失败：{}", e.getMessage());
                reconciliation.recordSyncFailure(request, result, e);
            }
        });

        return result;
    }
}

from dataclasses import dataclass
from enum import Enum
from typing import List


class RiskLevel(Enum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4


class RiskCategory(Enum):
    TECHNICAL = "技术风险"
    BUSINESS = "业务风险"
    ORGANIZATIONAL = "组织风险"
    DATA = "数据风险"
    COMPLIANCE = "合规风险"


@dataclass
class MigrationRisk:
    id: str
    category: RiskCategory
    description: str
    likelihood: RiskLevel
    impact: RiskLevel
    mitigation: str
    owner: str

    @property
    def risk_score(self) -> int:
        return self.likelihood.value * self.impact.value


class RiskRegistry:
    def __init__(self):
        self.risks: List[MigrationRisk] = []

    def add_risk(self, risk: MigrationRisk) -> None:
        self.risks.append(risk)

    def critical_risks(self) -> List[MigrationRisk]:
        return [r for r in self.risks if r.risk_score >= 9]

    def risks_by_category(self, category: RiskCategory) -> List[MigrationRisk]:
        return [r for r in self.risks if r.category == category]

    def generate_report(self) -> str:
        lines = ["迁移风险报告", "=" * 40]
        for risk in sorted(self.risks, key=lambda r: r.risk_score, reverse=True):
            lines.append(
                f"[{risk.id}] {risk.category.value} - "
                f"风险分={risk.risk_score} - {risk.description}"
            )
            lines.append(f"  缓解措施：{risk.mitigation}")
            lines.append(f"  负责人：{risk.owner}")
            lines.append("")
        return "\n".join(lines)


# 构建风险登记册
registry = RiskRegistry()

registry.add_risk(MigrationRisk(
    id="R001",
    category=RiskCategory.DATA,
    description="数据迁移过程中出现数据丢失或不一致",
    likelihood=RiskLevel.MEDIUM,
    impact=RiskLevel.CRITICAL,
    mitigation="实施多层数据校验，建立回滚机制，保留源数据至少 6 个月",
    owner="数据架构师",
))

registry.add_risk(MigrationRisk(
    id="R002",
    category=RiskCategory.TECHNICAL,
    description="COBOL 业务规则提取不完整导致功能缺失",
    likelihood=RiskLevel.HIGH,
    impact=RiskLevel.HIGH,
    mitigation="双写模式验证，自动化回归测试，资深工程师 Review",
    owner="技术负责人",
))

registry.add_risk(MigrationRisk(
    id="R003",
    category=RiskCategory.ORGANIZATIONAL,
    description="核心 COBOL 开发者在迁移完成前离职",
    likelihood=RiskLevel.MEDIUM,
    impact=RiskLevel.HIGH,
    mitigation="知识转移计划，关键人员留任激励，文档化所有隐式知识",
    owner="项目经理",
))

print(registry.generate_report())

from dataclasses import dataclass
from datetime import datetime
from typing import Dict, List


@dataclass
class ModuleMigrationStatus:
    module_name: str
    total_features: int
    migrated_features: int
    traffic_percentage: float  # 新系统承担的流量百分比
    start_date: datetime
    target_date: datetime
    status: str  # "not_started" | "in_progress" | "completed" | "blocked"

    @property
    def feature_progress(self) -> float:
        if self.total_features == 0:
            return 0.0
        return self.migrated_features / self.total_features * 100

    @property
    def is_on_track(self) -> bool:
        if self.status == "completed":
            return True
        elapsed = (datetime.now() - self.start_date).days
        total = (self.target_date - self.start_date).days
        if total == 0:
            return True
        expected_progress = elapsed / total * 100
        return self.feature_progress >= expected_progress * 0.9


class MigrationDashboard:
    def __init__(self, modules: List[ModuleMigrationStatus]):
        self.modules = modules

    def overall_progress(self) -> float:
        total = sum(m.total_features for m in self.modules)
        migrated = sum(m.migrated_features for m in self.modules)
        if total == 0:
            return 0.0
        return migrated / total * 100

    def overall_traffic_shift(self) -> float:
        if not self.modules:
            return 0.0
        return sum(m.traffic_percentage for m in self.modules) / len(self.modules)

    def blocked_modules(self) -> List[str]:
        return [m.module_name for m in self.modules if m.status == "blocked"]

    def at_risk_modules(self) -> List[str]:
        return [
            m.module_name
            for m in self.modules
            if m.status == "in_progress" and not m.is_on_track
        ]

    def summary(self) -> Dict:
        return {
            "总体功能迁移进度": f"{self.overall_progress():.1f}%",
            "总体流量切换进度": f"{self.overall_traffic_shift():.1f}%",
            "已完成模块": len([m for m in self.modules if m.status == "completed"]),
            "进行中模块": len([m for m in self.modules if m.status == "in_progress"]),
            "阻塞模块": self.blocked_modules(),
            "风险模块": self.at_risk_modules(),
        }