Business Application Migration

+----------------------------------------------------------------+
|                    MIGRATION PHASES                            |
+----------------------------------------------------------------+
|                                                                |
|  +------------+    +------------+    +------------+            |
|  |            |    |            |    |            |            |
|  | EXTRACTION +--->+ TRANSFORM  +--->+   LOAD     |            |
|  |            |    |            |    |            |            |
|  | Week 1-2   |    | Week 2-3   |    | Week 3-4   |            |
|  +------------+    +------------+    +------+-----+            |
|                                             |                  |
|       +-------------------------------------+                  |
|       |                                                        |
|       v                                                        |
|  +------------+    +------------+                              |
|  |            |    |            |                              |
|  | VALIDATION +--->+  CUTOVER   |                              |
|  |            |    |            |                              |
|  | Week 4-5   |    | Weekend    |                              |
|  +------------+    +------------+                              |
|                                                                |
+----------------------------------------------------------------+

   -- Example: Extract status codes from source
   SELECT status_code, status_name, is_active, sort_order
   FROM ref_status_codes
   WHERE object_type = 'CONTACT'
   ORDER BY sort_order;

   SELECT *
   FROM contacts
   WHERE created_date >= '2019-01-01'  -- Per historical scope decision
   ORDER BY contact_id;

   SELECT *
   FROM gl_transactions
   WHERE fiscal_year >= 2020
   ORDER BY fiscal_year, fiscal_period, transaction_date;

   -- Source system counts
   SELECT 'contacts' as object, COUNT(*) as cnt FROM contacts
   WHERE created_date >= '2019-01-01'
   UNION ALL
   SELECT 'organisations', COUNT(*) FROM organisations
   WHERE created_date >= '2019-01-01';

   import pandas as pd
   from pathlib import Path

   EXTRACT_DIR = Path('/data/migration/extracts')
   TRANSFORM_DIR = Path('/data/migration/transforms')

   # Load reference data first
   status_map = pd.read_csv(EXTRACT_DIR / 'ref_status_codes.csv')
   status_lookup = dict(zip(status_map['status_code'],
                           status_map['status_name']))

   def transform_contacts(df):
       result = pd.DataFrame()

       # Direct copy with rename
       result['external_id'] = 'LEG-' + df['CONTACT_ID'].astype(str)

       # Trim whitespace
       result['first_name'] = df['FIRST_NM'].str.strip()
       result['last_name'] = df['LAST_NM'].str.strip()

       # Validate and transform email
       email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
       result['email'] = df['EMAIL_ADDR'].where(
           df['EMAIL_ADDR'].str.match(email_pattern),
           other=None
       )

       # Normalise phone to E.164
       result['phone'] = df['PHONE_1'].apply(normalise_phone)

       # Map status codes
       result['status'] = df['STATUS_CD'].map(
           {'A': 'Active', 'I': 'Inactive', 'D': 'Deleted'}
       )

       return result

   def transform_with_exceptions(df, transform_func, object_name):
       exceptions = []
       results = []

       for idx, row in df.iterrows():
           try:
               results.append(transform_func(row))
           except Exception as e:
               exceptions.append({
                   'object': object_name,
                   'source_id': row.get('id', idx),
                   'error': str(e),
                   'row_data': row.to_dict()
               })

       # Write exceptions for review
       if exceptions:
           pd.DataFrame(exceptions).to_csv(
               TRANSFORM_DIR / f'{object_name}_exceptions.csv',
               index=False
           )

       return pd.DataFrame(results)

   audit_records = []

   for source_row, target_row in zip(source_df.itertuples(),
                                      target_df.itertuples()):
       audit_records.append({
           'source_id': source_row.CONTACT_ID,
           'target_external_id': target_row.external_id,
           'transform_timestamp': datetime.utcnow().isoformat(),
           'rules_applied': 'standard_contact_transform_v2'
       })

   pd.DataFrame(audit_records).to_csv(
       TRANSFORM_DIR / 'contacts_audit.csv',
       index=False
   )

   def validate_for_target(df, schema):
       errors = []

       for field, spec in schema.items():
           if spec['required'] and df[field].isna().any():
               errors.append(f"{field}: {df[field].isna().sum()} null values")

           if spec['max_length']:
               over_length = df[field].str.len() > spec['max_length']
               if over_length.any():
                   errors.append(
                       f"{field}: {over_length.sum()} values exceed "
                       f"{spec['max_length']} chars"
                   )

       return errors

   Disabled for migration load:
   - Workflow: New Contact Welcome Email (disabled 2024-03-15 09:00)
   - Workflow: Opportunity Stage Change Notification (disabled 2024-03-15 09:00)
   - Validation: Email Domain Check (disabled 2024-03-15 09:05)
   - Trigger: Contact Duplicate Check (disabled 2024-03-15 09:05)

   import requests
   from time import sleep

   def load_batch(records, endpoint, batch_size=500):
       results = {'success': 0, 'failed': 0, 'errors': []}

       for i in range(0, len(records), batch_size):
           batch = records[i:i + batch_size]

           response = requests.post(
               f'https://target-system.example.org/api/v2/{endpoint}/batch',
               headers={'Authorization': f'Bearer {API_TOKEN}'},
               json={'records': batch}
           )

           if response.status_code == 200:
               batch_result = response.json()
               results['success'] += batch_result['created']
               results['failed'] += len(batch_result.get('errors', []))
               results['errors'].extend(batch_result.get('errors', []))
           else:
               results['failed'] += len(batch)
               results['errors'].append({
                   'batch_start': i,
                   'error': response.text
               })

           # Respect rate limits
           sleep(1)

       return results

   id_mapping = {}

   for source_id, target_response in load_results:
       id_mapping[source_id] = target_response['id']

   # Save mapping for relationship linking
   pd.DataFrame([
       {'source_id': k, 'target_id': v}
       for k, v in id_mapping.items()
   ]).to_csv(TRANSFORM_DIR / 'contacts_id_mapping.csv', index=False)

   # Load opportunities referencing contacts
   opportunities_df = pd.read_csv(
       TRANSFORM_DIR / 'opportunities_transformed.csv'
   )

   # Map contact foreign keys
   contact_id_map = pd.read_csv(
       TRANSFORM_DIR / 'contacts_id_mapping.csv'
   ).set_index('source_id')['target_id'].to_dict()

   opportunities_df['contact_id'] = (
       opportunities_df['source_contact_id'].map(contact_id_map)
   )

   # Handle unmapped references
   unmapped = opportunities_df['contact_id'].isna()
   if unmapped.any():
       print(f"Warning: {unmapped.sum()} opportunities have unmapped contacts")
       # Log for manual review

   def load_attachment(file_path, parent_type, parent_id):
       with open(file_path, 'rb') as f:
           response = requests.post(
               f'https://target-system.example.org/api/v2/{parent_type}/'
               f'{parent_id}/attachments',
               headers={'Authorization': f'Bearer {API_TOKEN}'},
               files={'file': f}
           )
       return response.status_code == 201

   import random

   sample_ids = random.sample(list(contact_id_map.keys()), 500)

   validation_results = []
   for source_id in sample_ids:
       target_id = contact_id_map[source_id]

       source_record = fetch_from_source(source_id)
       target_record = fetch_from_target(target_id)

       for field_mapping in mapping_document:
           source_val = source_record.get(field_mapping['source_field'])
           target_val = target_record.get(field_mapping['target_field'])
           expected = apply_transform(source_val, field_mapping)

           validation_results.append({
               'source_id': source_id,
               'field': field_mapping['target_field'],
               'expected': expected,
               'actual': target_val,
               'match': expected == target_val
           })

   # Calculate pass rate
   pass_rate = sum(r['match'] for r in validation_results) / len(validation_results)
   print(f"Field validation pass rate: {pass_rate:.2%}")

   -- Find orphaned opportunity records
   SELECT o.id, o.external_id, o.contact_id
   FROM opportunities o
   LEFT JOIN contacts c ON o.contact_id = c.id
   WHERE c.id IS NULL;
   -- Expected: 0 rows

   -- Compare organisation opportunity totals
   SELECT
       o.external_id,
       o.opportunity_total as migrated_total,
       (SELECT SUM(amount) FROM opportunities
        WHERE organisation_id = o.id) as calculated_total
   FROM organisations o
   WHERE o.opportunity_total != (
       SELECT SUM(amount) FROM opportunities
       WHERE organisation_id = o.id
   );

   Integration test: Finance system sync

   1. Create test transaction in target CRM: Invoice INV-TEST-001, £1,500.00
   2. Trigger sync to finance system
   3. Verify transaction appears in finance system within 5 minutes
   4. Verify amount, date, reference match
   5. Delete test transaction from both systems

   Result: PASS / FAIL
   Tester: _______________
   Date: _______________

+---------------------------------------------------------------------+
|                            CUTOVER SEQUENCE                         |
+---------------------------------------------------------------------+
|                                                                     |
|  Friday 18:00        Saturday           Sunday           Monday     |
|       |                  |                 |                |       |
|  +----v----+        +----v----+       +----v----+      +----v----+  |
|  | Source  |        | Final   |       | Go-Live |      | Support |  |
|  | freeze  |        | load    |       | verify  |      | & hyper |  |
|  |         |        |         |       |         |      | care    |  |
|  +---------+        +---------+       +---------+      +---------+  |
|                                                                     |
|  18:00 Freeze       00:00 Delta       06:00 UAT        06:00 Users  |
|  19:00 Delta        06:00 Complete    12:00 Prod       access       |
|        extract      08:00 Validate    18:00 Go/NoGo    production   |
|                                                                     |
+---------------------------------------------------------------------+

   Subject: CRM Migration - System freeze starting Friday 18:00

   The CRM system will be READ-ONLY from Friday 18:00 until Monday 06:00.

   During this time:
   - You can view existing records
   - You cannot create or edit records
   - Email logging is paused (emails will be logged after migration)

   On Monday morning, access the new system at: https://crm.example.org
   Training materials: https://intranet.example.org/crm-training

   Support contact: it-support@example.org or ext. 2200

   -- Verify no records modified after freeze
   SELECT COUNT(*)
   FROM contacts
   WHERE modified_date > '2024-03-22 18:00:00';
   -- Expected: 0

   SELECT * FROM contacts
   WHERE modified_date > '2024-03-15 00:00:00'  -- Last full extract
   ORDER BY contact_id;

-- Target system record counts
SELECT
    'contacts' as object, COUNT(*) as count FROM contacts
UNION ALL SELECT
    'organisations', COUNT(*) FROM organisations
UNION ALL SELECT
    'opportunities', COUNT(*) FROM opportunities;