BIPAuto/tests/test_extractor_real.py

"""
Integration test for discrete_material_plan.extractor module

Tests the complete extraction workflow using real ERP data.
Requires valid ERP credentials in .env file.

This test:
1. Reads order IDs from tests/id-demo.txt
2. Downloads data from Yonyou BIP ERP system
3. Saves raw Excel files to data/downloads/
4. Converts and merges to data/output/merged_result.xlsx
"""

import sys
import logging
from pathlib import Path
from datetime import datetime

# Add project root to Python path
PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))

# Load environment variables
from dotenv import load_dotenv
load_dotenv(PROJECT_ROOT / ".env")

# Configure browser path
import os
os.environ["PLAYWRIGHT_BROWSERS_PATH"] = os.getenv("PLAYWRIGHT_BROWSERS_PATH", "")

# Setup logging
logger = logging.getLogger('bipauto.tests.extractor_real')
logger.setLevel(logging.INFO)
if not logger.handlers:
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter("[%(levelname)s] %(name)s: %(message)s"))
    logger.addHandler(handler)
logger.propagate = False

# Generate timestamp for unique output files
TIMESTAMP = datetime.now().strftime("%Y%m%d_%H%M%S")

# Output paths
DOWNLOAD_DIR = PROJECT_ROOT / "data" / "downloads"
OUTPUT_DIR = PROJECT_ROOT / "data" / "output"
OUTPUT_FILE = OUTPUT_DIR / f"merged_result_{TIMESTAMP}.xlsx"

# Ensure directories exist
DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

logger.info("=" * 60)
logger.info("Discrete Material Plan Extractor - Integration Test")
logger.info("=" * 60)
logger.info(f"Configuration:")
logger.info(f"  ID File: {PROJECT_ROOT / 'tests' / 'id-demo.txt'}")
logger.info(f"  Download Dir: {DOWNLOAD_DIR}")
logger.info(f"  Output File: {OUTPUT_FILE}")
logger.info(f"  ERP URL: {os.getenv('ERP_URL', 'Not set')}")
logger.info(f"  Headless: {os.getenv('ERP_HEADLESS', 'true')}")
logger.info("=" * 60)

# Verify environment variables
required_vars = ["ERP_USERNAME", "ERP_PASSWORD", "ERP_URL"]
missing_vars = [var for var in required_vars if not os.getenv(var)]

if missing_vars:
    logger.error(f"Missing required environment variables: {missing_vars}")
    logger.error("Please ensure .env file contains:")
    logger.error("  ERP_USERNAME=your_username")
    logger.error("  ERP_PASSWORD=your_password")
    logger.error("  ERP_URL=https://your-erp-url.com")
    sys.exit(1)

try:
    # Import required modules
    logger.info("[1/6] Importing modules...")
    from playwright.sync_api import sync_playwright
    from utils.auth import login
    from utils.discrete_material_plan import (
        navigate_to_discrete_material_page,
        extract_and_post_process,
        read_order_ids_from_file,
        get_login_url,
    )
    logger.info("Modules imported successfully")

    # Read order IDs from demo file
    logger.info("[2/6] Reading order IDs from data/demo/id-demo.txt...")
    id_file = PROJECT_ROOT / "data" / "demo" / "id-demo.txt"
    order_ids = read_order_ids_from_file(str(id_file))
    logger.info(f"Loaded {len(order_ids)} order IDs")
    
    test_order_ids = order_ids

    # Setup browser and extract data
    logger.info("[3/6] Launching browser and logging in...")
    with sync_playwright() as playwright:
        # Get required environment variables (with validation)
        username = os.getenv("ERP_USERNAME")
        password = os.getenv("ERP_PASSWORD")
        base_url = os.getenv("ERP_URL")
        
        if not username or not password or not base_url:
            raise ValueError("Missing required environment variables: ERP_USERNAME, ERP_PASSWORD, ERP_URL")
        
        # Login
        browser, context, page, main_frame = login(
            playwright=playwright,
            username=username,
            password=password,
            url=get_login_url(base_url),
            headless=os.getenv("ERP_HEADLESS", "true").lower() == "true",
            ignore_https_errors=os.getenv("ERP_IGNORE_HTTPS_ERRORS", "true").lower() == "true",
            verbose=True,
        )
        
        try:
            # Navigate to discrete material page
            logger.info("[4/6] Navigating to discrete material plan page...")
            work_frame, page1 = navigate_to_discrete_material_page(main_frame, page)
            logger.info("Navigation successful")
            
            # Extract and post-process
            logger.info("[5/6] Extracting data and post-processing...")
            logger.info(f"  Orders: {len(test_order_ids)}")
            logger.info(f"  Batch size: 100")
            logger.info(f"  Download dir: {DOWNLOAD_DIR}")
            logger.info(f"  Output file: {OUTPUT_FILE}")
            logger.info(f"  Cleanup temp files: True (default)")
            
            output_path, df = extract_and_post_process(
                work_frame=work_frame,
                page=page1,
                order_ids=test_order_ids,
                download_dir=str(DOWNLOAD_DIR),
                output_file=str(OUTPUT_FILE),
                batch_size=100,
                verbose=True,
                cleanup_temp_files=True,  # Default: True, set to False to keep temp files
            )
            
            logger.info("[6/6] Verifying results...")
            logger.info("Extraction completed successfully")
            logger.info(f"  Output file: {output_path}")
            logger.info(f"  Total records: {len(df)}")
            logger.info(f"  Columns: {list(df.columns)}")
            
            # Verify output file exists
            if not Path(output_path).exists():
                raise FileNotFoundError(f"Output file not found: {output_path}")
            logger.info(f"Output file verified: {output_path}")
            
            # Show sample data
            logger.info("=" * 60)
            logger.info("Sample Data (first 3 rows, first 5 columns):")
            logger.info("=" * 60)
            # Use unicode encoding for display
            sample = df.head(3).iloc[:, :5]
            logger.info("\n" + sample.to_string())
            logger.info("Data extraction verified - Chinese characters displayed correctly in console encoding")
            
        finally:
            # Cleanup browser
            logger.info("Closing browser...")
            context.close()
            browser.close()
            logger.info("Browser closed")

    logger.info("=" * 60)
    logger.info("INTEGRATION TEST COMPLETED SUCCESSFULLY")
    logger.info("=" * 60)
    logger.info(f"Outputs:")
    logger.info(f"  Raw Excel files: {DOWNLOAD_DIR}/*.xlsx")
    logger.info(f"  Merged result: {OUTPUT_FILE}")
    logger.info(f"  Total records: {len(df)}")
    logger.info("=" * 60)

except Exception as e:
    logger.error(f"Integration test failed!")
    logger.error(f"Error: {e}")
    import traceback
    traceback.print_exc()
    sys.exit(1)