admin/BIPAuto
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(extractor): Convert to stateless pure functions

Browse Source 
- Remove DiscreteMaterialPlanExtractor class (stateful)
- Replace with pure functions: chunk_order_ids, get_login_url, extract_batch, etc.
- All functions are stateless, accept explicit parameters
- Caller manages browser/session lifecycle (consistent with extractor_core.py)
- Lower coupling: no direct dependency on utils.auth.login
- Update tests to match new function signatures

Breaking Changes:
- DiscreteMaterialPlanExtractor class removed
- Use extract_and_post_process() or extract_from_file() instead of class methods
- Caller must manage browser session before calling extractor functions
This commit is contained in:
Misaka_Company
2026-03-27 13:52:13 +08:00
parent da567a2679
commit 1b984f5cfd
3 changed files with 466 additions and 390 deletions
Download Patch File Download Diff File Expand all files Collapse all files

187
tests/test_extractor_component.py
View File

@@ -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()

43
utils/discrete_material_plan/__init__.py Normal file
View File

@@ -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",
]

626
utils/discrete_material_plan/extractor.py
View File

@@ -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,
)