diff --git a/tests/test_extractor_component.py b/tests/test_extractor_component.py index 924c01d..e034ee5 100644 --- a/tests/test_extractor_component.py +++ b/tests/test_extractor_component.py @@ -1,5 +1,8 @@ """ -Test DiscreteMaterialPlanExtractor component structure +Test discrete_material_plan.extractor module functions + +Tests pure functions for data extraction and post-processing. +Does not require actual browser or ERP connection. """ import sys @@ -18,98 +21,128 @@ import os os.environ["PLAYWRIGHT_BROWSERS_PATH"] = os.getenv("PLAYWRIGHT_BROWSERS_PATH", "") print("=" * 60) -print("Testing DiscreteMaterialPlanExtractor Component") +print("Testing discrete_material_plan.extractor Functions") print("=" * 60) try: - # Test 1: Import the component - print("\n[1/4] Importing DiscreteMaterialPlanExtractor...") + # Test 1: Import all functions + print("\n[1/5] Importing extractor functions...") from utils.discrete_material_plan.extractor import ( - DiscreteMaterialPlanExtractor, - extract_from_file + chunk_order_ids, + get_login_url, + extract_batch, + extract_batches, + extract_and_post_process, + read_order_ids_from_file, + extract_from_file, ) - print("[OK] Import successful") + print("[OK] All functions imported successfully") - # Test 2: Verify class exists and has expected methods - print("\n[2/4] Verifying class structure...") - expected_methods = [ - '__init__', - '_print', - '_cleanup_temp_files', - '_get_login_url', - '_chunk_order_ids', - '_download_batches', - 'post_process_downloads', - 'extract_and_process', - 'extract_from_file' - ] + # Test 2: Test chunk_order_ids + print("\n[2/5] Testing chunk_order_ids...") - missing_methods = [] - for method in expected_methods: - if not hasattr(DiscreteMaterialPlanExtractor, method): - missing_methods.append(method) + # Test normal chunking + result = chunk_order_ids(["A", "B", "C", "D", "E"], 2) + expected = [["A", "B"], ["C", "D"], ["E"]] + assert result == expected, f"Expected {expected}, got {result}" + print(f" chunk_order_ids(['A','B','C','D','E'], 2) = {result}") - if missing_methods: - print(f"[FAIL] Missing methods: {missing_methods}") - raise AttributeError(f"Missing methods: {missing_methods}") + # Test exact division + result = chunk_order_ids(["A", "B", "C", "D"], 2) + expected = [["A", "B"], ["C", "D"]] + assert result == expected, f"Expected {expected}, got {result}" + print(f" chunk_order_ids(['A','B','C','D'], 2) = {result}") - print(f"[OK] All {len(expected_methods)} expected methods found") + # Test batch size larger than list + result = chunk_order_ids(["A", "B"], 10) + expected = [["A", "B"]] + assert result == expected, f"Expected {expected}, got {result}" + print(f" chunk_order_ids(['A','B'], 10) = {result}") + + # Test empty list + result = chunk_order_ids([], 5) + expected = [] + assert result == expected, f"Expected {expected}, got {result}" + print(f" chunk_order_ids([], 5) = {result}") + + print("[OK] chunk_order_ids works correctly") - # Test 3: Test instantiation with dummy parameters (no browser) - print("\n[3/4] Testing class instantiation...") - extractor = DiscreteMaterialPlanExtractor( - username="test_user", - password="test_password", - base_url="https://example.com", - headless=True, - ignore_https_errors=True, - verbose=True, - download_dir=None, - batch_size=10 + # Test 3: Test get_login_url + print("\n[3/5] Testing get_login_url...") + + # Test with trailing slash + result = get_login_url("https://erp.example.com/") + expected = "https://erp.example.com/yonbip/resources/uap/rbac/login/main/index.html" + assert result == expected, f"Expected {expected}, got {result}" + print(f" get_login_url('https://erp.example.com/') = {result}") + + # Test without trailing slash + result = get_login_url("https://erp.example.com") + assert result == expected, f"Expected {expected}, got {result}" + print(f" get_login_url('https://erp.example.com') = {result}") + + # Test with path + result = get_login_url("https://erp.example.com/some/path/") + expected = "https://erp.example.com/some/path/yonbip/resources/uap/rbac/login/main/index.html" + assert result == expected, f"Expected {expected}, got {result}" + print(f" get_login_url('https://erp.example.com/some/path/') = {result}") + + print("[OK] get_login_url works correctly") + + # Test 4: Test read_order_ids_from_file + print("\n[4/5] Testing read_order_ids_from_file...") + + # Create a temporary test file + import tempfile + with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False, encoding='utf-8') as f: + f.write("ID001\n") + f.write("ID002\n") + f.write("\n") # Empty line + f.write("ID003\n") + f.write(" \n") # Whitespace only + f.write("ID004\n") + temp_file = f.name + + try: + result = read_order_ids_from_file(temp_file) + expected = ["ID001", "ID002", "ID003", "ID004"] + assert result == expected, f"Expected {expected}, got {result}" + print(f" read_order_ids_from_file(temp_file) = {result}") + print("[OK] read_order_ids_from_file works correctly") + finally: + # Cleanup temp file + Path(temp_file).unlink() + + # Test FileNotFoundError + try: + read_order_ids_from_file("nonexistent_file.txt") + print("[FAIL] Should have raised FileNotFoundError") + raise AssertionError("Should have raised FileNotFoundError") + except FileNotFoundError: + print(f" read_order_ids_from_file('nonexistent') raises FileNotFoundError [OK]") + + # Test 5: Test module exports + print("\n[5/5] Testing module exports...") + from utils.discrete_material_plan import ( + chunk_order_ids as exported_chunk, + get_login_url as exported_get_url, + extract_batch as exported_extract_batch, + extract_batches as exported_extract_batches, + extract_and_post_process as exported_extract_and_post, + read_order_ids_from_file as exported_read_ids, + extract_from_file as exported_extract_file, ) - - # Verify attributes are set correctly - assert extractor.username == "test_user", "Username not set correctly" - assert extractor.password == "test_password", "Password not set correctly" - assert extractor.base_url == "https://example.com", "Base URL not set correctly" - assert extractor.headless == True, "Headless not set correctly" - assert extractor.ignore_https_errors == True, "HTTPS errors setting not set correctly" - assert extractor.verbose == True, "Verbose not set correctly" - assert extractor.batch_size == 10, "Batch size not set correctly" - assert hasattr(extractor, 'download_dir'), "Download dir not set" - - print("[OK] Instantiation successful, all attributes verified") - - # Test 4: Test helper methods - print("\n[4/4] Testing helper methods...") - - # Test _get_login_url - login_url = extractor._get_login_url() - expected_url = "https://example.com/yonbip/resources/uap/rbac/login/main/index.html" - assert login_url == expected_url, f"Login URL incorrect: {login_url}" - print(f"[OK] _get_login_url() returns: {login_url}") - - # Test _chunk_order_ids - test_ids = ["ID1", "ID2", "ID3", "ID4", "ID5"] - extractor.batch_size = 2 - chunks = extractor._chunk_order_ids(test_ids) - expected_chunks = [["ID1", "ID2"], ["ID3", "ID4"], ["ID5"]] - assert chunks == expected_chunks, f"Chunking incorrect: {chunks}" - print(f"[OK] _chunk_order_ids() works correctly: {chunks}") - - # Test _print (should not raise) - extractor._print("Test message") - print("[OK] _print() works correctly") - - # Cleanup - extractor._cleanup_temp_files() + print("[OK] All functions exported correctly from module") print("\n" + "=" * 60) - print("[SUCCESS] All component tests passed!") + print("[SUCCESS] All extractor function tests passed!") print("=" * 60) + print("\nNote: extract_batch, extract_batches, extract_and_post_process,") + print(" and extract_from_file require actual browser session and") + print(" are not tested here. Integration tests cover those cases.") except Exception as e: - print(f"\n[ERROR] Component test failed!") + print(f"\n[ERROR] Function test failed!") print(f"Error: {e}") import traceback traceback.print_exc() diff --git a/utils/discrete_material_plan/__init__.py b/utils/discrete_material_plan/__init__.py new file mode 100644 index 0000000..8801982 --- /dev/null +++ b/utils/discrete_material_plan/__init__.py @@ -0,0 +1,43 @@ +""" +Discrete Material Plan Maintenance Package +Core web operations and high-level extractor for Yonyou BIP discrete material plan maintenance. +""" + +from .extractor_core import ( + navigate_to_discrete_material_page, + setup_query_interface, + fill_and_search_orders, + download_batch_data, + execute_batch_download_workflow, +) + +from .extractor import ( + chunk_order_ids, + get_login_url, + extract_batch, + extract_batches, + extract_and_post_process, + read_order_ids_from_file, + extract_from_file, +) + +from .excel_converter import ExcelConverter + +__all__ = [ + # Core functions + "navigate_to_discrete_material_page", + "setup_query_interface", + "fill_and_search_orders", + "download_batch_data", + "execute_batch_download_workflow", + # High-level extractor functions + "chunk_order_ids", + "get_login_url", + "extract_batch", + "extract_batches", + "extract_and_post_process", + "read_order_ids_from_file", + "extract_from_file", + # Utilities + "ExcelConverter", +] diff --git a/utils/discrete_material_plan/extractor.py b/utils/discrete_material_plan/extractor.py index f826fb0..5172546 100644 --- a/utils/discrete_material_plan/extractor.py +++ b/utils/discrete_material_plan/extractor.py @@ -1,334 +1,334 @@ """ -High-level extractor component for discrete material plan data extraction. -Provides batch processing with progress reporting and error handling. +Discrete Material Plan Data Extractor + +Pure functions for extracting and post-processing discrete material plan data. +All functions are stateless and accept required parameters explicitly. +Caller is responsible for browser/session lifecycle management. """ -import tempfile -import shutil +import pandas as pd from pathlib import Path -from typing import List, Optional -from playwright.sync_api import sync_playwright - -from utils.auth import login -from .excel_converter import ExcelConverter -from .extractor_core import ( - navigate_to_discrete_material_page, - setup_query_interface, - execute_batch_download_workflow, -) +from typing import List, Optional, Tuple +from playwright.sync_api import Page, Frame -class DiscreteMaterialPlanExtractor: +def chunk_order_ids(order_ids: List[str], batch_size: int) -> List[List[str]]: """ - High-level extractor for discrete material plan data. - Handles session management, batch processing, and post-processing. - """ - - def __init__( - self, - username: str, - password: str, - base_url: str, - headless: bool = True, - ignore_https_errors: bool = True, - verbose: bool = True, - download_dir: Optional[str] = None, - batch_size: int = 10, - ): - """ - Initialize the extractor. - - Args: - username: Login username - password: Login password - base_url: Base URL for the ERP system - headless: Whether to run browser in headless mode - ignore_https_errors: Whether to ignore HTTPS certificate errors - verbose: Whether to print detailed logs - download_dir: Directory to save downloaded files (default: temp dir) - batch_size: Number of order IDs per batch - """ - self.username = username - self.password = password - self.base_url = base_url - self.headless = headless - self.ignore_https_errors = ignore_https_errors - self.verbose = verbose - self.batch_size = batch_size - - # Set up download directory - if download_dir: - self.download_dir = Path(download_dir) - self.download_dir.mkdir(parents=True, exist_ok=True) - else: - self.download_dir = Path(tempfile.mkdtemp()) - self._is_temp_dir = True - - self._print(f"Download directory: {self.download_dir}") - - def _print(self, *args, **kwargs): - """Print log message if verbose mode is enabled.""" - if self.verbose: - print(*args, **kwargs) - - def _cleanup_temp_files(self): - """Remove temporary files created during extraction.""" - if hasattr(self, "_is_temp_dir") and self._is_temp_dir: - try: - shutil.rmtree(self.download_dir) - self._print(f"Cleaned up temp directory: {self.download_dir}") - except Exception as e: - self._print(f"Warning: Failed to clean up temp directory: {e}") - - def _get_login_url(self) -> str: - """ - Construct the complete login URL. - - Returns: - str: Complete login page URL - """ - base = self.base_url.rstrip("/") - return f"{base}/yonbip/resources/uap/rbac/login/main/index.html" - - def _chunk_order_ids(self, order_ids: List[str]) -> List[List[str]]: - """ - Split order IDs into batches. - - Args: - order_ids: List of order IDs to process - - Returns: - List of batches, where each batch is a list of order IDs - """ - chunks = [] - for i in range(0, len(order_ids), self.batch_size): - chunks.append(order_ids[i : i + self.batch_size]) - return chunks - - def _download_batches( - self, order_ids: List[str] - ) -> List[str]: - """ - Download data for all batches of order IDs. - - Args: - order_ids: List of order IDs to download - - Returns: - List of paths to downloaded files - """ - downloaded_files = [] - chunks = self._chunk_order_ids(order_ids) - - self._print(f"Processing {len(order_ids)} order IDs in {len(chunks)} batches") - - with sync_playwright() as playwright: - # Login - self._print("Logging in...") - browser, context, page, main_frame = login( - playwright=playwright, - username=self.username, - password=self.password, - url=self._get_login_url(), - headless=self.headless, - ignore_https_errors=self.ignore_https_errors, - verbose=self.verbose, - ) - - try: - # Navigate to discrete material page - self._print("Navigating to discrete material plan page...") - work_frame, page1 = navigate_to_discrete_material_page( - main_frame, page - ) - - # Setup query interface - self._print("Setting up query interface...") - setup_query_interface(work_frame) - - # Process each batch - for batch_index, batch in enumerate(chunks): - self._print( - f"Processing batch {batch_index + 1}/{len(chunks)} ({len(batch)} orders)" - ) - - file_path = execute_batch_download_workflow( - work_frame=work_frame, - page=page1, - order_ids=batch, - batch_index=batch_index, - download_dir=str(self.download_dir), - ) - - downloaded_files.append(file_path) - self._print(f"Downloaded: {file_path}") - - finally: - # Cleanup - self._print("Closing browser...") - context.close() - browser.close() - - return downloaded_files - - def post_process_downloads( - self, downloaded_files: List[str], output_file: Optional[str] = None - ) -> Optional[str]: - """ - Convert and merge downloaded Excel files into structured format. - - Args: - downloaded_files: List of paths to downloaded Excel files - output_file: Path to save merged result (default: merged_result.xlsx in download_dir) - - Returns: - Path to the merged output file - """ - if not downloaded_files: - self._print("No files to post-process") - return None - - self._print(f"Post-processing {len(downloaded_files)} downloaded files...") - - converter = ExcelConverter(verbose=self.verbose) - all_dfs = [] - - # Convert each file - for i, file_path in enumerate(downloaded_files): - self._print(f"Converting file {i + 1}/{len(downloaded_files)}: {file_path}") - df = converter.convert(input_file=file_path) # type: ignore - all_dfs.append(df) - - # Merge all DataFrames - if all_dfs: - import pandas as pd - - merged_df = pd.concat(all_dfs, ignore_index=True) - - # Determine output file path - if output_file: - output_path = Path(output_file) - else: - output_path = self.download_dir / "merged_result.xlsx" - - # Save merged result - merged_df.to_excel(output_path, index=False) - self._print(f"Merged result saved to: {output_path}") - self._print(f"Total rows: {len(merged_df)}") - - return str(output_path) - - return None - - def extract_and_process( - self, order_ids: List[str], output_file: Optional[str] = None - ) -> Optional[str]: - """ - Complete extraction workflow: download + post-process. - - Args: - order_ids: List of order IDs to extract - output_file: Path to save merged result (optional) - - Returns: - Path to the final merged output file - """ - # Download all batches - downloaded_files = self._download_batches(order_ids) - - # Post-process: convert and merge - output_path = self.post_process_downloads(downloaded_files, output_file) - - return output_path - - def extract_from_file( - self, id_file: str, output_file: Optional[str] = None - ) -> Optional[str]: - """ - Extract data from order IDs listed in a file. - - Args: - id_file: Path to file containing order IDs (one per line) - output_file: Path to save merged result (optional) - - Returns: - Path to the final merged output file - """ - # Read order IDs from file - id_path = Path(id_file) - if not id_path.exists(): - raise FileNotFoundError(f"ID file not found: {id_file}") - - with open(id_path, "r", encoding="utf-8") as f: - order_ids = [line.strip() for line in f if line.strip()] - - self._print(f"Loaded {len(order_ids)} order IDs from {id_file}") - - # Extract and process - return self.extract_and_process(order_ids, output_file) - - -def extract_from_file( - id_file: str, - output_file: Optional[str] = None, - username: Optional[str] = None, - password: Optional[str] = None, - base_url: Optional[str] = None, - headless: bool = True, - ignore_https_errors: bool = True, - verbose: bool = True, - download_dir: Optional[str] = None, - batch_size: int = 10, -) -> Optional[str]: - """ - Convenience function to extract data from order IDs in a file. + Split order IDs into batches. Args: - id_file: Path to file containing order IDs (one per line) - output_file: Path to save merged result (optional) - username: Login username (required if not in env) - password: Login password (required if not in env) - base_url: Base URL for ERP system (required if not in env) - headless: Whether to run browser in headless mode - ignore_https_errors: Whether to ignore HTTPS certificate errors - verbose: Whether to print detailed logs - download_dir: Directory to save downloaded files (optional) - batch_size: Number of order IDs per batch + order_ids: List of order IDs to process + batch_size: Maximum number of order IDs per batch Returns: - Path to the final merged output file + List of batches, where each batch is a list of order IDs - Note: - If username, password, or base_url are not provided, - they will be read from environment variables. + Example: + >>> chunk_order_ids(["A", "B", "C", "D"], 2) + [["A", "B"], ["C", "D"]] """ - import os + return [ + order_ids[i : i + batch_size] + for i in range(0, len(order_ids), batch_size) + ] - # Get credentials from parameters or environment - if not username: - username = os.getenv("ERP_USERNAME") - if not password: - password = os.getenv("ERP_PASSWORD") - if not base_url: - base_url = os.getenv("ERP_URL") - if not username or not password or not base_url: - raise ValueError( - "username, password, and base_url must be provided either as parameters or environment variables" +def get_login_url(base_url: str) -> str: + """ + Construct the complete login URL from base URL. + + Args: + base_url: Base ERP URL (e.g., "https://erp.example.com") + + Returns: + Complete login page URL + + Example: + >>> get_login_url("https://erp.example.com") + "https://erp.example.com/yonbip/resources/uap/rbac/login/main/index.html" + """ + return f"{base_url.rstrip('/')}/yonbip/resources/uap/rbac/login/main/index.html" + + +def extract_batch( + work_frame: Frame, + page: Page, + order_ids: List[str], + batch_index: int, + download_dir: str, +) -> str: + """ + Execute download workflow for a single batch of order IDs. + + This is a thin wrapper around `execute_batch_download_workflow` from + extractor_core.py, providing progress reporting context. + + Args: + work_frame: The work iframe containing the search form and data grid + page: The Playwright page object for download handling + order_ids: List of order IDs for this batch + batch_index: Zero-based batch index for naming the output file + download_dir: Directory path to save the downloaded file + + Returns: + Full path to the downloaded Excel file + """ + from .extractor_core import execute_batch_download_workflow + + return execute_batch_download_workflow( + work_frame=work_frame, + page=page, + order_ids=order_ids, + batch_index=batch_index, + download_dir=download_dir, + ) + + +def extract_batches( + work_frame: Frame, + page: Page, + order_ids: List[str], + download_dir: str, + batch_size: int = 10, +) -> List[str]: + """ + Download data for multiple batches of order IDs. + + Caller is responsible for: + - Browser session management (login, logout) + - Navigation to discrete material plan page + - Query interface setup + + Args: + work_frame: The work iframe containing the data grid + page: The Playwright page object for download handling + order_ids: List of order IDs to download + download_dir: Directory path to save downloaded files + batch_size: Maximum number of order IDs per batch + + Returns: + List of paths to downloaded Excel files + + Example: + >>> # Caller manages 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, page1, order_ids, "/downloads") + >>> context.close() + >>> browser.close() + """ + from .extractor_core import setup_query_interface + + downloaded_files = [] + chunks = chunk_order_ids(order_ids, batch_size) + + # Setup query interface once + setup_query_interface(work_frame) + + # Process each batch + for batch_index, batch in enumerate(chunks): + file_path = extract_batch( + work_frame=work_frame, + page=page, + order_ids=batch, + batch_index=batch_index, + download_dir=download_dir, ) + downloaded_files.append(file_path) - extractor = DiscreteMaterialPlanExtractor( - username=username, - password=password, - base_url=base_url, - headless=headless, - ignore_https_errors=ignore_https_errors, - verbose=verbose, + return downloaded_files + + +def post_process_downloads( + downloaded_files: List[str], + output_file: str, + verbose: bool = True, +) -> Tuple[str, pd.DataFrame]: + """ + Convert and merge downloaded Excel files into structured DataFrame. + + Uses ExcelConverter to convert each file, then merges all results. + + Args: + downloaded_files: List of paths to downloaded Excel files + output_file: Path to save merged Excel result + verbose: Whether to print progress messages + + Returns: + Tuple of (output_file_path, merged_dataframe) + + Example: + >>> output_path, df = post_process_downloads( + ... downloaded_files=["batch_1.xlsx", "batch_2.xlsx"], + ... output_file="merged.xlsx", + ... verbose=True + ... ) + """ + from .excel_converter import ExcelConverter + + converter = ExcelConverter(verbose=verbose) + all_dfs = [] + + # Convert each file + for i, file_path in enumerate(downloaded_files): + if verbose: + print(f"Converting file {i + 1}/{len(downloaded_files)}: {file_path}") + + # Convert (do not save intermediate result) + df = converter.convert(input_file=file_path) + all_dfs.append(df) + + # Merge all DataFrames + if not all_dfs: + merged_df = pd.DataFrame() + else: + merged_df = pd.concat(all_dfs, ignore_index=True) + + # Save merged result + output_path = Path(output_file) + output_path.parent.mkdir(parents=True, exist_ok=True) + merged_df.to_excel(output_path, index=False) + + if verbose: + print(f"Merged result saved to: {output_path}") + print(f"Total rows: {len(merged_df)}") + + return str(output_path), merged_df + + +def extract_and_post_process( + work_frame: Frame, + page: Page, + order_ids: List[str], + download_dir: str, + output_file: str, + batch_size: int = 10, + verbose: bool = True, +) -> Tuple[str, pd.DataFrame]: + """ + Complete extraction workflow: download batches + post-process to merged Excel. + + This is a high-level convenience function that orchestrates the full workflow. + Caller is still responsible for browser session management. + + Args: + work_frame: The work iframe containing the data grid + page: The Playwright page object for download handling + order_ids: List of order IDs to extract + download_dir: Directory for temporary batch files + output_file: Path for final merged Excel output + batch_size: Maximum order IDs per batch + verbose: Whether to print progress messages + + Returns: + Tuple of (output_file_path, merged_dataframe) + + Example: + >>> # Caller manages session + >>> browser, context, page, main_frame = login(...) + >>> work_frame, page1 = navigate_to_discrete_material_page(main_frame, page) + >>> output_path, df = extract_and_post_process( + ... work_frame, page1, order_ids, "/downloads", "output.xlsx" + ... ) + >>> context.close() + >>> browser.close() + """ + # Step 1: Download all batches + if verbose: + print(f"Downloading {len(order_ids)} orders in batches of {batch_size}...") + + downloaded_files = extract_batches( + work_frame=work_frame, + page=page, + order_ids=order_ids, download_dir=download_dir, batch_size=batch_size, ) - try: - return extractor.extract_from_file(id_file, output_file) - finally: - extractor._cleanup_temp_files() + if verbose: + print(f"Downloaded {len(downloaded_files)} batch file(s)") + + # Step 2: Post-process (convert + merge) + output_path, merged_df = post_process_downloads( + downloaded_files=downloaded_files, + output_file=output_file, + verbose=verbose, + ) + + return output_path, merged_df + + +def 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). + + Args: + id_file: Path to file containing order IDs + encoding: File encoding (default: utf-8) + + Returns: + List of order IDs (stripped, empty lines filtered) + + Raises: + FileNotFoundError: If id_file doesn't exist + + Example: + >>> order_ids = read_order_ids_from_file("orders.txt") + """ + id_path = Path(id_file) + if not id_path.exists(): + raise FileNotFoundError(f"Order ID file not found: {id_file}") + + with open(id_path, "r", encoding=encoding) as f: + return [line.strip() for line in f if line.strip()] + + +def extract_from_file( + id_file: str, + work_frame: Frame, + page: Page, + download_dir: str, + output_file: str, + batch_size: int = 10, + verbose: bool = True, +) -> Tuple[str, pd.DataFrame]: + """ + Extract data from order IDs in a file and post-process to merged Excel. + + Convenience function that reads IDs from file and calls extract_and_post_process. + + Args: + id_file: Path to file containing order IDs (one per line) + work_frame: The work iframe containing the data grid + page: The Playwright page object for download handling + download_dir: Directory for temporary batch files + output_file: Path for final merged Excel output + batch_size: Maximum order IDs per batch + verbose: Whether to print progress messages + + Returns: + Tuple of (output_file_path, merged_dataframe) + + Example: + >>> # Read order IDs + >>> order_ids = read_order_ids_from_file("orders.txt") + >>> # Extract and process + >>> output_path, df = extract_from_file( + ... "orders.txt", work_frame, page, "/downloads", "output.xlsx" + ... ) + """ + order_ids = read_order_ids_from_file(id_file) + + if verbose: + print(f"Loaded {len(order_ids)} order IDs from {id_file}") + + return extract_and_post_process( + work_frame=work_frame, + page=page, + order_ids=order_ids, + download_dir=download_dir, + output_file=output_file, + batch_size=batch_size, + verbose=verbose, + )