BIPAuto/docs/discrete_material_plan_extractor_api.md

Discrete Material Plan Extractor - API Reference

Pure Function Data Extraction for Yonyou BIP

Stateless functions for extracting discrete material plan data from Yonyou BIP ERP system with batch processing and Excel post-processing capabilities.

---

Overview

The extractor module provides high-level pure functions for extracting material plan data from the Yonyou BIP ERP system. All functions are stateless and explicitly accept required parameters, following the same design patterns as extractor_core.py and auth.py.

Architecture Overview

graph TB
    subgraph User["User Code"]
        A[Main Script]
    end
    
    subgraph Auth["utils/auth.py"]
        B[login]
        C[logout]
    end
    
    subgraph Extractor["utils/discrete_material_plan/extractor.py"]
        D[extract_and_post_process]
        E[extract_batches]
        F[post_process_downloads]
        G[read_order_ids_from_file]
        H[chunk_order_ids]
    end
    
    subgraph Core["utils/discrete_material_plan/extractor_core.py"]
        I[navigate_to_discrete_material_page]
        J[setup_query_interface]
        K[execute_batch_download_workflow]
    end
    
    subgraph Converter["utils/discrete_material_plan/excel_converter.py"]
        L[ExcelConverter.convert]
    end
    
    subgraph External["External"]
        M[(Yonyou BIP ERP)]
        N[(Downloaded Excel Files)]
        O[(Merged Output Excel)]
    end
    
    A --> B
    A --> D
    A --> G
    
    B --> M
    D --> E
    D --> F
    E --> H
    E --> K
    F --> L
    
    B --> I
    I --> M
    K --> M
    K --> N
    
    L --> N
    L --> O
    
    style User fill:#e1f5ff
    style Auth fill:#fff4e1
    style Extractor fill:#e8f5e9
    style Core fill:#fce4ec
    style Converter fill:#f3e5f5
    style External fill:#ffebee

Key Features:

• ✅ Stateless pure functions (no class instances)
• ✅ Explicit parameter passing
• ✅ Batch processing for large order lists
• ✅ Excel conversion and merging
• ✅ Progress reporting with verbose logging
• ✅ Caller-managed session lifecycle

Module Location: utils/discrete_material_plan/extractor.py

---

Quick Start

Basic Usage - Complete Workflow

from utils.discrete_material_plan import (
    extract_and_post_process,
    get_login_url,
)
from utils.auth import login
from utils.discrete_material_plan import navigate_to_discrete_material_page
from playwright.sync_api import sync_playwright

# 1. Setup browser session
with sync_playwright() as playwright:
    browser, context, page, main_frame = login(
        playwright=playwright,
        username="your_username",
        password="your_password",
        url=get_login_url("https://erp.example.com"),
        headless=True,
        ignore_https_errors=True,
    )
    
    try:
        # 2. Navigate to discrete material page
        work_frame, page1 = navigate_to_discrete_material_page(main_frame, page)
        
        # 3. Extract and process data
        output_path, df = extract_and_post_process(
            work_frame=work_frame,
            page=page1,
            order_ids=["SC70202603240001", "SC70202603240002"],
            download_dir="./downloads",
            output_file="./output/merged_data.xlsx",
            batch_size=10,
            verbose=True,
        )
        
        print(f"Extracted {len(df)} records to {output_path}")
        
    finally:
        # 4. Caller manages browser lifecycle
        context.close()
        browser.close()

From File - Simplest Approach

from utils.discrete_material_plan import read_order_ids_from_file

# Read order IDs from file
order_ids = read_order_ids_from_file("order_ids.txt")
print(f"Loaded {len(order_ids)} order IDs")

---

API Reference

Helper Functions

chunk_order_ids(order_ids: List[str], batch_size: int) -> List[List[str]]

Split order IDs into batches for processing.

Parameters:

Name	Type	Description
order_ids	List[str]	List of order IDs to process
batch_size	int	Maximum order IDs per batch

Returns: List[List[str]] - List of batches

Example:

>>> chunk_order_ids(["A", "B", "C", "D", "E"], 2)
[["A", "B"], ["C", "D"], ["E"]]

---

get_login_url(base_url: str) -> str

Construct complete login URL from base ERP URL.

Parameters:

Name	Type	Description
base_url	str	Base ERP URL (e.g., "https://erp.example.com")

Returns: str - Complete login page URL

Example:

>>> get_login_url("https://erp.example.com")
"https://erp.example.com/yonbip/resources/uap/rbac/login/main/index.html"

---

read_order_ids_from_file(id_file: str, encoding: str = "utf-8") -> List[str]

Read order IDs from a text file (one ID per line).

Parameters:

Name	Type	Default	Description
id_file	str	-	Path to file containing order IDs
encoding	str	"utf-8"	File encoding

Returns: List[str] - List of order IDs (empty lines filtered)

Raises:

• FileNotFoundError - If id_file doesn't exist

Example:

>>> order_ids = read_order_ids_from_file("orders.txt")
>>> len(order_ids)
15

---

Core Extraction Functions

extract_batch(work_frame, page, order_ids, batch_index, download_dir) -> str

Execute download workflow for a single batch of order IDs.

Parameters:

Name	Type	Description
work_frame	Frame	Work iframe containing the data grid
page	Page	Playwright page object for download handling
order_ids	List[str]	Order IDs for this batch
batch_index	int	Zero-based batch index (for file naming)
download_dir	str	Directory to save downloaded file

Returns: str - Path to downloaded Excel file

Note: This is a thin wrapper around execute_batch_download_workflow from extractor_core.py.

---

extract_batches(work_frame, page, order_ids, download_dir, batch_size=10) -> List[str]

Download data for multiple batches of order IDs.

Parameters:

Name	Type	Default	Description
work_frame	Frame	Work iframe containing the data grid
page	Page	Playwright page object
order_ids	List[str]	List of order IDs to download
download_dir	str	Directory to save downloaded files
batch_size	int	10	Max order IDs per batch

Returns: List[str] - List of downloaded file paths

Example:

# Caller manages browser session
browser, context, page, main_frame = login(...)
work_frame, page1 = navigate_to_discrete_material_page(main_frame, page)
setup_query_interface(work_frame)

files = extract_batches(
    work_frame=work_frame,
    page=page1,
    order_ids=["ID1", "ID2", "ID3"],
    download_dir="./downloads",
    batch_size=5
)

context.close()
browser.close()

---

High-Level Workflow Functions

extract_and_post_process(work_frame, page, order_ids, download_dir, output_file, batch_size=10, verbose=True) -> Tuple[str, DataFrame]

Complete extraction workflow: download batches + post-process to merged Excel.

Workflow Diagram

sequenceDiagram
    participant User as User Code
    participant Auth as utils.auth.login
    participant Nav as navigate_to_discrete<br/>_material_page
    participant Setup as setup_query<br/>_interface
    participant Extract as extract_and_post<br/>_process
    participant Batches as extract_batches
    participant PostProcess as post_process<br/>_downloads
    participant Converter as ExcelConverter
    participant Browser as Browser
    participant ERP as Yonyou BIP ERP
    participant FS as File System

    User->>Auth: login(credentials)
    Auth->>Browser: Launch
    Auth->>ERP: Authenticate
    Auth-->>User: (browser, context, page, main_frame)

    User->>Nav: navigate(main_frame, page)
    Nav->>ERP: Load page
    Nav-->>User: (work_frame, page1)

    User->>Setup: setup_query_interface(work_frame)
    Setup->>ERP: Configure query panel

    User->>Extract: extract_and_post_process(params)
    
    Note over Extract,PostProcess: Phase 1: Download
    Extract->>Batches: extract_batches(work_frame, page, order_ids)
    
    loop For each batch
        Batches->>Batches: chunk_order_ids(order_ids, batch_size)
        Batches->>Setup: setup_query_interface()
        Batches->>ERP: Fill order IDs + Search
        Batches->>ERP: Click Export
        ERP-->>FS: Save batch_N.xlsx
        Batches-->>Extract: [batch_1.xlsx, batch_2.xlsx, ...]
    end

    Note over Extract,PostProcess: Phase 2: Post-Process
    Extract->>PostProcess: post_process_downloads(files, output_file)
    
    loop For each downloaded file
        PostProcess->>Converter: convert(batch_N.xlsx)
        Converter->>FS: Read Excel
        Converter->>Converter: Parse nested structure
        Converter-->>PostProcess: DataFrame
    end

    PostProcess->>PostProcess: pd.concat(all_dfs)
    PostProcess->>FS: Write merged.xlsx
    PostProcess-->>Extract: (output_path, merged_df)
    
    Extract-->>User: (output_path, DataFrame)

    User->>Browser: context.close()
    User->>Browser: browser.close()

Parameters:

Name	Type	Default	Description
work_frame	Frame	-	Work iframe containing the data grid
page	Page	-	Playwright page object for download handling
order_ids	List[str]	-	List of order IDs to extract
download_dir	str	-	Directory for temporary batch files
output_file	str	-	Path for final merged Excel output
batch_size	int	10	Max order IDs per batch
verbose	bool	True	Print progress messages

Returns: Tuple[str, pd.DataFrame] - (output_file_path, merged_dataframe)

Example:

from utils.discrete_material_plan import extract_and_post_process

# Caller manages session (see full example above)
output_path, df = extract_and_post_process(
    work_frame=work_frame,
    page=page1,
    order_ids=["SC70202603240001", "SC70202603240002"],
    download_dir="./downloads",
    output_file="./output/merged.xlsx",
    batch_size=10,
    verbose=True,
)

print(f"Saved {len(df)} records to {output_path}")

Workflow:

1. Downloads order data in batches to download_dir
2. Converts each downloaded Excel file to DataFrame
3. Merges all DataFrames into one
4. Saves merged result to output_file
5. Returns (path, dataframe) tuple

---

extract_from_file(id_file, work_frame, page, download_dir, output_file, batch_size=10, verbose=True) -> Tuple[str, DataFrame]

Extract data from order IDs in a file and post-process to merged Excel.

Parameters:

Name	Type	Default	Description
id_file	str	-	Path to file with order IDs (one per line)
work_frame	Frame	Work iframe containing the data grid
page	Page	Playwright page object for download
download_dir	str	Directory for temporary batch files
output_file	str	Path for final merged Excel output
batch_size	int	10	Max order IDs per batch
verbose	bool	True	Print progress messages

Returns: Tuple[str, pd.DataFrame] - (output_file_path, merged_dataframe)

Example:

from utils.discrete_material_plan import extract_from_file

# After setting up browser session and navigating to page
output_path, df = extract_from_file(
    id_file="order_ids.txt",
    work_frame=work_frame,
    page=page1,
    download_dir="./downloads",
    output_file="./output/results.xlsx",
    batch_size=10,
    verbose=True,
)

File Format:

SC70202603240001
SC70202603240002
SC70202603240003

---

Post-Processing Functions

post_process_downloads(downloaded_files, output_file, verbose=True) -> Tuple[str, DataFrame]

Convert and merge downloaded Excel files into structured DataFrame.

Parameters:

Name	Type	Default	Description
downloaded_files	List[str]	-	List of downloaded Excel file paths
output_file	str	-	Path to save merged Excel result
verbose	bool	True	Print progress messages

Returns: Tuple[str, pd.DataFrame] - (output_file_path, merged_dataframe)

Example:

from utils.discrete_material_plan import post_process_downloads

output_path, df = post_process_downloads(
    downloaded_files=[
        "./downloads/batch_1.xlsx",
        "./downloads/batch_2.xlsx",
    ],
    output_file="./output/merged.xlsx",
    verbose=True,
)

print(f"Merged {len(df)} records")

Process:

1. Uses ExcelConverter to convert each file
2. Merges all DataFrames with pd.concat()
3. Saves merged result to output_file

---

Architecture

Function Hierarchy

graph LR
    subgraph Level1["Level 1: Entry Points"]
        A1[extract_and_post_process]
        A2[extract_from_file]
    end
    
    subgraph Level2["Level 2: Workflow Orchestration"]
        B1[extract_batches]
        B2[post_process_downloads]
        B3[read_order_ids_from_file]
    end
    
    subgraph Level3["Level 3: Core Operations"]
        C1[extract_batch]
        C2[chunk_order_ids]
        C3[setup_query_interface]
        C4[ExcelConverter.convert]
    end
    
    subgraph Level4["Level 4: Low-Level"]
        D1[execute_batch_download_workflow]
        D2[fill_and_search_orders]
        D3[download_batch_data]
    end
    
    A1 --> B1
    A1 --> B2
    A2 --> B3
    A2 --> A1
    
    B1 --> C1
    B1 --> C2
    B1 --> C3
    B2 --> C4
    
    C1 --> D1
    C3 --> D2
    D1 --> D2
    D1 --> D3
    
    style Level1 fill:#e3f2fd
    style Level2 fill:#fff3e0
    style Level3 fill:#f3e5f5
    style Level4 fill:#e8f5e9

Design Principles

1. Stateless Pure Functions

# ✅ Correct: Stateless, explicit parameters
output_path, df = extract_and_post_process(
    work_frame=work_frame,
    page=page,
    order_ids=order_ids,
    # ... explicit params
)

# ❌ Wrong: Stateful class (OLD approach - removed)
# extractor = DiscreteMaterialPlanExtractor(username, password, ...)

2. Caller-Managed Lifecycle

# Caller manages browser session
browser, context, page, main_frame = login(...)
try:
    work_frame, page1 = navigate_to_discrete_material_page(main_frame, page)
    result = extract_and_post_process(work_frame, page1, order_ids, ...)
finally:
    context.close()  # Caller closes
    browser.close()  # Caller closes

3. Explicit Dependencies

# ❌ No implicit state
def extract(order_ids):  # Missing required params
    ...

# ✅ All params explicit
def extract(work_frame, page, order_ids, download_dir, output_file):
    ...

Module Dependencies

graph TD
    subgraph Extractor["extractor.py"]
        E1[extract_and_post_process]
        E2[extract_batches]
        E3[post_process_downloads]
    end
    
    subgraph Core["extractor_core.py"]
        C1[execute_batch_download_workflow]
        C2[setup_query_interface]
        C3[fill_and_search_orders]
        C4[download_batch_data]
    end
    
    subgraph Converter["excel_converter.py"]
        K1[ExcelConverter]
        K2[convert]
    end
    
    subgraph External["External Libraries"]
        P1[pandas DataFrame]
        P2[openpyxl]
    end
    
    E1 --> E2
    E1 --> E3
    E2 --> C1
    E2 --> C2
    E3 --> K1
    K1 --> K2
    K2 --> P1
    K1 --> P2
    
    C1 --> C2
    C1 --> C3
    C1 --> C4
    
    style Extractor fill:#e8f5e9
    style Core fill:#fff3e0
    style Converter fill:#f3e5f5
    style External fill:#e3f2fd

Low Coupling:

• No direct dependency on utils.auth
• Session objects (work_frame, page) passed as parameters
• Caller controls lifecycle

---

Integration Patterns

Data Flow Diagram

flowchart LR
    subgraph Input["Input"]
        I1[Order IDs<br/>List or File]
    end
    
    subgraph Download["Download Phase<br/>Requires Browser"]
        D1[Chunk into<br/>Batches]
        D2[Search Orders<br/>in ERP]
        D3[Export to<br/>Excel]
        D4[(Batch Excel<br/>Files)]
    end
    
    subgraph Process["Post-Process Phase<br/>No Browser Needed"]
        P1[Read Excel<br/>Files]
        P2[Parse Nested<br/>Structure]
        P3[Flatten to<br/>DataFrame]
        P4[Concatenate<br/>All Batches]
        P5[(Merged<br/>Excel)]
    end
    
    subgraph Output["Output"]
        O1[DataFrame<br/>Object]
        O2[Excel File]
    end
    
    I1 --> D1
    D1 --> D2
    D2 --> D3
    D3 --> D4
    D4 --> P1
    P1 --> P2
    P2 --> P3
    P3 --> P4
    P4 --> P5
    P4 --> O1
    P5 --> O2
    
    style Input fill:#e1f5ff
    style Download fill:#fff4e1
    style Process fill:#e8f5e9
    style Output fill:#fce4ec

---

Integration Patterns

Pattern 1: Complete Workflow with Session Management

from playwright.sync_api import sync_playwright
from utils.auth import login, get_login_url
from utils.discrete_material_plan import (
    navigate_to_discrete_material_page,
    extract_and_post_process,
)

def main():
    with sync_playwright() as playwright:
        # Setup session
        browser, context, page, main_frame = login(
            playwright=playwright,
            username="admin",
            password="secret",
            url=get_login_url("https://erp.example.com"),
            headless=True,
        )
        
        try:
            # Navigate
            work_frame, page1 = navigate_to_discrete_material_page(main_frame, page)
            
            # Extract
            output_path, df = extract_and_post_process(
                work_frame=work_frame,
                page=page1,
                order_ids=["ID1", "ID2", "ID3"],
                download_dir="./downloads",
                output_file="./output/result.xlsx",
            )
            
            print(f"Done: {len(df)} records")
            
        finally:
            # Cleanup
            context.close()
            browser.close()

Pattern 2: Two-Phase (Download Then Process)

from utils.discrete_material_plan import (
    extract_batches,
    post_process_downloads,
    setup_query_interface,
    navigate_to_discrete_material_page,
)
from utils.auth import login

# Phase 1: Download
browser, context, page, main_frame = login(...)
work_frame, page1 = navigate_to_discrete_material_page(main_frame, page)

downloaded_files = extract_batches(
    work_frame=work_frame,
    page=page1,
    order_ids=order_ids,
    download_dir="./downloads",
    batch_size=10,
)

context.close()
browser.close()

# Phase 2: Process (can be done later, even without browser)
output_path, df = post_process_downloads(
    downloaded_files=downloaded_files,
    output_file="./output/merged.xlsx",
)

Pattern 3: Custom Batch Processing with Error Handling

from utils.discrete_material_plan import chunk_order_ids, extract_batch
from utils.auth import login

order_ids = [...]  # Large list
batches = chunk_order_ids(order_ids, batch_size=10)

success_files = []
failed_batches = []

for i, batch in enumerate(batches):
    try:
        file_path = extract_batch(
            work_frame=work_frame,
            page=page,
            order_ids=batch,
            batch_index=i,
            download_dir="./downloads",
        )
        success_files.append(file_path)
        print(f"Batch {i+1}/{len(batches)} OK")
    except Exception as e:
        failed_batches.append(i)
        print(f"Batch {i+1} failed: {e}")

print(f"Success: {len(success_files)}, Failed: {len(failed_batches)}")

---

Testing

Unit Tests

# Run component tests
source .venv/Scripts/activate && python tests/test_extractor_component.py

Test Coverage:

• ✅ chunk_order_ids() - Batch splitting logic
• ✅ get_login_url() - URL construction
• ✅ read_order_ids_from_file() - File reading
• ✅ Module exports verification

Integration Tests

# Run real data extraction test
source .venv/Scripts/activate && python tests/test_extractor_real.py

Note: Integration tests require:

• Valid ERP credentials in .env
• Playwright browser installed
• Network access to ERP system

---

Error Handling

Common Errors

FileNotFoundError:

>>> read_order_ids_from_file("nonexistent.txt")
FileNotFoundError: Order ID file not found: nonexistent.txt

Timeout during extraction:

try:
    output_path, df = extract_and_post_process(...)
except TimeoutError as e:
    print(f"Operation timed out: {e}")
    # Browser session may need re-initialization

Permission errors (file access):

try:
    post_process_downloads(downloaded_files, output_file)
except PermissionError:
    print(f"Cannot write to {output_file} - file may be open in another program")

Best Practices

# 1. Always cleanup browser sessions
try:
    # extraction logic
finally:
    context.close()
    browser.close()

# 2. Validate order IDs before extraction
order_ids = read_order_ids_from_file("orders.txt")
if not order_ids:
    raise ValueError("No order IDs to process")

# 3. Ensure output directory exists
from pathlib import Path
Path("./output").mkdir(parents=True, exist_ok=True)

---

Performance Considerations

Batch Size Tuning

Small batches (5-10):

• ✅ More frequent progress updates
• ✅ Easier to recover from failures
• ❌ More UI interactions (slower)

Large batches (50-100):

• ✅ Faster overall (fewer UI interactions)
• ✅ Better for large datasets
• ❌ Single failure affects more records

Recommendation: Start with batch_size=10, adjust based on:

• Total order count
• Network stability
• UI response time

Memory Usage

For very large extractions (1000+ orders):

# Process in stages to limit memory
all_dfs = []
for batch_files in batch_groups:
    _, df = post_process_downloads(batch_files, output_file)
    all_dfs.append(df)

# Final merge
merged_df = pd.concat(all_dfs, ignore_index=True)

---

Migration Guide

From Old Class-Based API (v1) to New Function API (v2)

API Comparison

mindmap
  root((API Comparison))
    v1 Old API
      Class-Based
        ::icon(fa fa-times-circle)
        Stateful
        Internal session management
        Hidden dependencies
      Usage
        extractor = DiscreteMaterialPlanExtractor(...)
        extractor.extract_from_file(...)
      Issues
        Hard to test
        Tight coupling
        Implicit state
    v2 New API
      Pure Functions
        ::icon(fa fa-check-circle)
        Stateless
        Caller-managed session
        Explicit dependencies
      Usage
        extract_and_post_process(params...)
        Return: (output_path, DataFrame)
      Benefits
        Easy to test
        Low coupling
        Clear data flow

Old (v1):

from utils.discrete_material_plan import DiscreteMaterialPlanExtractor

extractor = DiscreteMaterialPlanExtractor(
    username="admin",
    password="secret",
    base_url="https://erp.example.com",
)

output_path = extractor.extract_from_file("orders.txt", "output.xlsx")

New (v2):

from utils.discrete_material_plan import (
    extract_from_file,
    get_login_url,
    navigate_to_discrete_material_page,
)
from utils.auth import login

# Caller manages session
browser, context, page, main_frame = login(
    playwright,
    username="admin",
    password="secret",
    url=get_login_url("https://erp.example.com"),
)

try:
    work_frame, page1 = navigate_to_discrete_material_page(main_frame, page)
    
    output_path, df = extract_from_file(
        id_file="orders.txt",
        work_frame=work_frame,
        page=page1,
        download_dir="./downloads",
        output_file="./output/result.xlsx",
    )
finally:
    context.close()
    browser.close()

Version Comparison Table

Aspect	v1 (Class-Based)	v2 (Pure Functions)
Pattern	DiscreteMaterialPlanExtractor class	Standalone functions
State	Instance variables (self.username, etc.)	No state, explicit params
Session	Internal management	Caller manages
Return	str (output path only)	Tuple[str, DataFrame]
Testing	Hard (requires mocking class)	Easy (pure functions)
Coupling	High (depends on utils.auth)	Low (session passed as param)
Flexibility	Limited (fixed workflow)	High (can use individual functions)

Key Changes:

1. Class removed → pure functions
2. Session management moved to caller
3. Return value now includes DataFrame tuple
4. All dependencies explicit

---

Related Documentation

• extractor_core.py - Low-level web operations
• excel_converter.py - Excel conversion
• auth.py - Authentication and session management

---

Version

Current: v2.0.0 (stateless pure functions)

Breaking Changes in v2:

• Removed DiscreteMaterialPlanExtractor class
• Changed to caller-managed session lifecycle
• All functions now accept explicit parameters

---

License

Part of BIPAuto project - see project root for license information.