RPA自动化迁移机器人流程自动化在数据迁移中的应用

{
    "robot": {
        "name": "数据采集机器人",
        "description": "从遗留系统采集数据",
        "modules": [
            {
                "name": "登录模块",
                "steps": [
                    { "action": "click", "selector": "#loginBtn" },
                    { "action": "input", "selector": "#username", "value": "{{username}}" },
                    { "action": "input", "selector": "#password", "value": "{{password}}" },
                    { "action": "click", "selector": "#submitBtn" }
                ]
            },
            {
                "name": "数据采集模块",
                "steps": [
                    { "action": "navigate", "url": "{{dataPageUrl}}" },
                    { "action": "input", "selector": "#startDate", "value": "{{startDate}}" },
                    { "action": "input", "selector": "#endDate", "value": "{{endDate}}" },
                    { "action": "click", "selector": "#searchBtn" },
                    { "action": "extract", "selector": "#dataTable", "output": "{{data}}" }
                ]
            }
        ],
        "errorHandling": {
            "retry": 3,
            "timeout": 30,
            "onError": "logAndContinue"
        }
    }
}

from playwright.sync_api import sync_playwright
import json
import time

class DataExtractionRobot:
    def __init__(self, config):
        self.config = config
        self.playwright = None
        self.browser = None
        self.page = None
    
    def init_browser(self):
        self.playwright = sync_playwright().start()
        self.browser = self.playwright.chromium.launch(headless=False)
        self.page = self.browser.new_page()
    
    def login(self):
        self.page.goto(self.config['login_url'])
        self.page.fill('#username', self.config['username'])
        self.page.fill('#password', self.config['password'])
        self.page.click('#submitBtn')
        self.page.wait_for_load_state('networkidle')
    
    def navigate_to_data_page(self):
        self.page.click('#dataMenu')
        self.page.wait_for_load_state('networkidle')
    
    def set_query_condition(self, start_date, end_date):
        self.page.fill('#startDate', start_date)
        self.page.fill('#endDate', end_date)
        self.page.click('#searchBtn')
        self.page.wait_for_load_state('networkidle')
    
    def extract_data(self):
        data = []
        rows = self.page.query_selector_all('#dataTable tbody tr')
        
        for row in rows:
            record = {
                'id': row.query_selector('td:nth-child(1)').text_content(),
                'name': row.query_selector('td:nth-child(2)').text_content(),
                'email': row.query_selector('td:nth-child(3)').text_content(),
                'phone': row.query_selector('td:nth-child(4)').text_content(),
                'createdAt': row.query_selector('td:nth-child(5)').text_content()
            }
            data.append(record)
        
        return data
    
    def has_more_data(self):
        next_btn = self.page.query_selector('#nextPageBtn')
        return next_btn is not None and next_btn.is_enabled()
    
    def go_to_next_page(self):
        self.page.click('#nextPageBtn')
        self.page.wait_for_load_state('networkidle')
    
    def save_data(self, data, filename):
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
    
    def run(self, start_date, end_date, output_file):
        try:
            self.init_browser()
            self.login()
            self.navigate_to_data_page()
            self.set_query_condition(start_date, end_date)
            
            all_data = []
            while True:
                page_data = self.extract_data()
                all_data.extend(page_data)
                print(f"Extracted {len(page_data)} records, total {len(all_data)}")
                
                if self.has_more_data():
                    self.go_to_next_page()
                else:
                    break
            
            self.save_data(all_data, output_file)
            print(f"Data extraction complete. Saved {len(all_data)} records to {output_file}")
            
        finally:
            if self.browser:
                self.browser.close()
            if self.playwright:
                self.playwright.stop()

import json
import re

class DataProcessingRobot:
    def __init__(self):
        self.mapping_rules = {
            'id': 'id',
            'name': 'name',
            'email': 'email',
            'phone': 'phone',
            'createdAt': 'created_at'
        }
    
    def load_data(self, input_file):
        with open(input_file, 'r', encoding='utf-8') as f:
            return json.load(f)
    
    def clean_data(self, data):
        cleaned = []
        for record in data:
            cleaned_record = {
                'id': record.get('id', '').strip(),
                'name': record.get('name', '').strip(),
                'email': record.get('email', '').strip().lower(),
                'phone': self.clean_phone(record.get('phone', '')),
                'createdAt': record.get('createdAt', '')
            }
            cleaned.append(cleaned_record)
        return cleaned
    
    def clean_phone(self, phone):
        return re.sub(r'\D', '', phone)
    
    def transform_data(self, data):
        transformed = []
        for record in data:
            transformed_record = {
                self.mapping_rules[key]: value
                for key, value in record.items()
                if key in self.mapping_rules
            }
            transformed_record['status'] = 'active'
            transformed.append(transformed_record)
        return transformed
    
    def validate_data(self, data):
        valid = []
        errors = []
        
        for record in data:
            record_errors = []
            
            if not record.get('id'):
                record_errors.append('ID is required')
            if not record.get('name'):
                record_errors.append('Name is required')
            if not record.get('email'):
                record_errors.append('Email is required')
            elif not re.match(r'^[\w\.-]+@[\w\.-]+\.\w+$', record['email']):
                record_errors.append('Invalid email format')
            
            if record_errors:
                errors.append({'record': record, 'errors': record_errors})
            else:
                valid.append(record)
        
        return valid, errors
    
    def save_data(self, data, output_file):
        with open(output_file, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
    
    def save_errors(self, errors, error_file):
        with open(error_file, 'w', encoding='utf-8') as f:
            json.dump(errors, f, ensure_ascii=False, indent=2)
    
    def run(self, input_file, output_file, error_file):
        print(f"Loading data from {input_file}")
        raw_data = self.load_data(input_file)
        print(f"Loaded {len(raw_data)} records")
        
        print("Cleaning data...")
        cleaned_data = self.clean_data(raw_data)
        
        print("Transforming data...")
        transformed_data = self.transform_data(cleaned_data)
        
        print("Validating data...")
        valid_data, errors = self.validate_data(transformed_data)
        
        print(f"Validation complete: {len(valid_data)} valid, {len(errors)} errors")
        
        self.save_data(valid_data, output_file)
        print(f"Saved {len(valid_data)} records to {output_file}")
        
        if errors:
            self.save_errors(errors, error_file)
            print(f"Saved {len(errors)} errors to {error_file}")
        
        return valid_data, errors