diff --git a/docs/discrete_material_plan_extractor_core.md b/docs/discrete_material_plan_extractor_core.md new file mode 100644 index 0000000..230c98d --- /dev/null +++ b/docs/discrete_material_plan_extractor_core.md @@ -0,0 +1,426 @@ +# extractor_core.py - API Reference + +> **Core Web Operations for Yonyou BIP** +> +> Low-level pure functions for interacting with the discrete material plan maintenance page. + +--- + +## Overview + +The `extractor_core.py` module provides low-level pure functions for web operations on the Yonyou BIP ERP system. All functions are **stateless** and accept an optional `logger` parameter for observability. + +**Design Principles:** +- ✅ **Silent by default**: Functions produce no output when `logger=None` +- ✅ **Optional logging**: Pass a logger to get debug/info messages +- ✅ **Stateless**: No internal state, all dependencies explicit +- ✅ **Pure functions**: Same inputs → same outputs, no side effects + +**Module Location:** `utils/discrete_material_plan/extractor_core.py` + +--- + +## Functions + +### `navigate_to_discrete_material_page(main_frame, page, logger=None)` + +Navigate to the discrete material plan maintenance page. + +**Parameters:** +| Name | Type | Default | Description | +|------|------|---------|-------------| +| `main_frame` | `Frame` | - | The main forwardFrame iframe | +| `page` | `Page` | - | The Playwright page object | +| `logger` | `Optional[logging.Logger]` | `None` | Optional logger for debug output | + +**Returns:** `tuple[FrameLocator, Page]` - (work_frame, page1) + +**Behavior:** +- Clicks menu icon to open navigation +- Opens popup and clicks "离散备料计划维护" menu item +- Navigates through nested iframes to return work frame + +**Logging (when logger provided):** +- `DEBUG`: "Navigated to discrete material plan page" + +**Example:** +```python +from utils.discrete_material_plan.extractor_core import navigate_to_discrete_material_page +import logging + +# Silent mode (default) +work_frame, page1 = navigate_to_discrete_material_page(main_frame, page) + +# With logging +logging.basicConfig(level=logging.DEBUG) +logger = logging.getLogger(__name__) +work_frame, page1 = navigate_to_discrete_material_page(main_frame, page, logger) +``` + +--- + +### `setup_query_interface(work_frame, logger=None)` + +Initialize the query interface by selecting order number query tab. + +**Parameters:** +| Name | Type | Default | Description | +|------|------|---------|-------------| +| `work_frame` | `FrameLocator` | - | The inner work iframe | +| `logger` | `Optional[logging.Logger]` | `None` | Optional logger for debug output | + +**Returns:** `None` + +**Behavior:** +- Opens search panel +- Selects "订单号查询" (order number query) tab +- Selects "全部" (All) tab +- Sets page size to 5000 + +**Logging (when logger provided):** +- `DEBUG`: "Query interface setup complete" + +**Example:** +```python +from utils.discrete_material_plan.extractor_core import setup_query_interface + +# Silent mode +setup_query_interface(work_frame) + +# With logging +setup_query_interface(work_frame, logger) +``` + +--- + +### `fill_and_search_orders(work_frame, order_ids, logger=None)` + +Fill order IDs into the search textbox and trigger search. + +**Parameters:** +| Name | Type | Default | Description | +|------|------|---------|-------------| +| `work_frame` | `FrameLocator` | - | The work iframe containing search form | +| `order_ids` | `List[str]` | - | List of order IDs to search for | +| `logger` | `Optional[logging.Logger]` | `None` | Optional logger for debug output | + +**Returns:** `None` + +**Behavior:** +- Clears and fills order IDs into "来源生产订单号" textbox +- Clicks search button +- Waits for loading indicator (3s timeout, continues if not found) + +**Logging (when logger provided):** +- `DEBUG`: "Loading indicator timeout - continuing anyway" (if timeout occurs) +- `DEBUG`: "Searched for {n} order IDs" (after search) + +**Example:** +```python +from utils.discrete_material_plan.extractor_core import fill_and_search_orders + +# Silent mode +fill_and_search_orders(work_frame, ["SC70202603240001", "SC70202603240002"]) + +# With logging +fill_and_search_orders(work_frame, order_ids, logger) +``` + +--- + +### `download_batch_data(work_frame, page, order_ids, batch_index, download_dir, logger=None)` + +Execute the download workflow for a single batch of order IDs. + +**Parameters:** +| Name | Type | Default | Description | +|------|------|---------|-------------| +| `work_frame` | `FrameLocator` | - | The work iframe containing data grid | +| `page` | `Page` | - | Playwright page for download handling | +| `order_ids` | `List[str]` | - | List of order IDs (for context) | +| `batch_index` | `int` | - | Zero-based batch index | +| `download_dir` | `str` | - | Directory to save downloaded file | +| `logger` | `Optional[logging.Logger]` | `None` | Optional logger for info output | + +**Returns:** `str` - Full path to downloaded file + +**Behavior:** +1. Selects first row in grid +2. Hovers over "更多" (More) button +3. Clicks "输出" (Export) +4. Sets row threshold to 300000 +5. Triggers download and saves as `temp_batch_{n}.xlsx` + +**Logging (when logger provided):** +- `INFO`: "Downloaded batch {n} to {path}" + +**Raises:** +- `TimeoutError`: If UI elements not found or operations timeout + +**Example:** +```python +from utils.discrete_material_plan.extractor_core import download_batch_data + +# Silent mode +file_path = download_batch_data( + work_frame, page, order_ids, batch_index=0, download_dir="./downloads" +) + +# With logging +file_path = download_batch_data( + work_frame, page, order_ids, 0, "./downloads", logger +) +# INFO: Downloaded batch 1 to ./downloads/temp_batch_1.xlsx +``` + +--- + +### `execute_batch_download_workflow(work_frame, page, order_ids, batch_index, download_dir, logger=None)` + +Complete workflow: fill orders, search, and download for a single batch. + +**Parameters:** +| Name | Type | Default | Description | +|------|------|---------|-------------| +| `work_frame` | `FrameLocator` | - | Work iframe with search form and data grid | +| `page` | `Page` | - | Playwright page for download | +| `order_ids` | `List[str]` | - | List of order IDs for this batch | +| `batch_index` | `int` | - | Zero-based batch index | +| `download_dir` | `str` | - | Directory to save downloaded file | +| `logger` | `Optional[logging.Logger]` | `None` | Optional logger for debug output | + +**Returns:** `str` - Full path to downloaded file + +**Behavior:** +- Calls `fill_and_search_orders()` then `download_batch_data()` +- Propagates logger to both inner calls + +**Logging (when logger provided):** +- All logs from `fill_and_search_orders()` and `download_batch_data()` + +**Example:** +```python +from utils.discrete_material_plan.extractor_core import execute_batch_download_workflow + +# Silent mode +file_path = execute_batch_download_workflow( + work_frame, page, order_ids, batch_index=0, download_dir="./downloads" +) + +# With logging +file_path = execute_batch_download_workflow( + work_frame, page, order_ids, 0, "./downloads", logger +) +``` + +--- + +## Usage Patterns + +### Pattern 1: Silent Mode (Default) + +```python +from utils.discrete_material_plan.extractor_core import ( + navigate_to_discrete_material_page, + setup_query_interface, + execute_batch_download_workflow, +) + +# No output, completely silent +work_frame, page1 = navigate_to_discrete_material_page(main_frame, page) +setup_query_interface(work_frame) +file_path = execute_batch_download_workflow( + work_frame, page, order_ids, 0, "./downloads" +) +``` + +### Pattern 2: With Logging + +```python +import logging +from utils.discrete_material_plan.extractor_core import ( + navigate_to_discrete_material_page, + setup_query_interface, + execute_batch_download_workflow, +) + +# Configure logging +logging.basicConfig( + level=logging.DEBUG, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger(__name__) + +# Verbose output +work_frame, page1 = navigate_to_discrete_material_page(main_frame, page, logger) +setup_query_interface(work_frame, logger) +file_path = execute_batch_download_workflow( + work_frame, page, order_ids, 0, "./downloads", logger +) + +# Output: +# DEBUG:__main__:Navigated to discrete material plan page +# DEBUG:__main__:Query interface setup complete +# DEBUG:__main__:Searched for 10 order IDs +# INFO:__main__:Downloaded batch 1 to ./downloads/temp_batch_1.xlsx +``` + +### Pattern 3: Complete Workflow + +```python +from playwright.sync_api import sync_playwright +from utils.auth import login +from utils.discrete_material_plan.extractor_core import ( + navigate_to_discrete_material_page, + setup_query_interface, + execute_batch_download_workflow, +) +import logging + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +with sync_playwright() as playwright: + browser, context, page, main_frame = login( + playwright, "user", "pass", + url="https://erp.example.com/...", + headless=True, + ignore_https_errors=True, + ) + + try: + # Navigate and setup + work_frame, page1 = navigate_to_discrete_material_page(main_frame, page, logger) + setup_query_interface(work_frame, logger) + + # Download batches + order_ids = ["SC70202603240001", "SC70202603240002", ...] + file_path = execute_batch_download_workflow( + work_frame, page1, order_ids, 0, "./downloads", logger + ) + + finally: + context.close() + browser.close() +``` + +--- + +## Error Handling + +### TimeoutError on Loading Indicator + +The `fill_and_search_orders()` function handles loading indicator timeouts gracefully: + +```python +try: + loading_locator.wait_for(state="visible", timeout=3000) + loading_locator.wait_for(state="hidden", timeout=0) +except TimeoutError: + if logger: + logger.debug("Loading indicator timeout - continuing anyway") + # Continues execution - not a fatal error +``` + +### When to Raise vs Log + +- **Timeout on loading indicator**: Log at DEBUG, continue (non-fatal) +- **Missing UI elements**: Raise TimeoutError (fatal) +- **Download failures**: Raise exception (fatal) + +--- + +## Architecture + +### Function Call Flow + +```mermaid +graph TD + A[User Script] --> B[navigate_to_discrete_material_page] + A --> C[setup_query_interface] + A --> D[execute_batch_download_workflow] + + D --> E[fill_and_search_orders] + D --> F[download_batch_data] + + E --> G[fill_and_search_orders implementation] + F --> H[download_batch_data implementation] + + style A fill:#e1f5ff + style B fill:#fff4e1 + style C fill:#fff4e1 + style D fill:#e8f5e9 + style E fill:#fce4ec + style F fill:#fce4ec +``` + +### Type Hierarchy + +``` +Page (from playwright) + └─> main_frame: Frame + └─> navigate_to_discrete_material_page() + └─> Returns: FrameLocator (work_frame) + └─> setup_query_interface(work_frame: FrameLocator) + └─> fill_and_search_orders(work_frame: FrameLocator) + └─> download_batch_data(work_frame: FrameLocator) +``` + +--- + +## Testing + +### Import Test + +```bash +python -c "from utils.discrete_material_plan.extractor_core import fill_and_search_orders; print('[OK] Import works')" +``` + +### Silent Mode Test + +```python +from utils.discrete_material_plan.extractor_core import fill_and_search_orders + +# Should produce ZERO output +fill_and_search_orders(mock_frame, ["ID1", "ID2"]) +print("Silent mode: OK") +``` + +### With Logger Test + +```python +import logging +from utils.discrete_material_plan.extractor_core import fill_and_search_orders + +logging.basicConfig(level=logging.DEBUG) +logger = logging.getLogger(__name__) + +# Should produce debug output +fill_and_search_orders(mock_frame, ["ID1", "ID2"], logger) +``` + +--- + +## Related Documentation + +- [`discrete_material_plan_extractor_api.md`](discrete_material_plan_extractor_api.md) - High-level extractor API +- [`extractor_post_processing.md`](extractor_post_processing.md) - Excel conversion +- [`authentication.md`](authentication.md) - Auth and session management + +--- + +## Version + +**Current:** v2.1.0 (added optional logger parameter) + +**Changes in v2.1.0:** +- Added optional `logger` parameter to all functions +- Functions remain silent when `logger=None` +- TimeoutError handling now logs at DEBUG level +- No breaking changes - backward compatible + +--- + +## License + +Part of BIPAuto project - see project root for license information. diff --git a/tests/test_auth_config.py b/tests/test_auth_config.py index c053074..1e67e5f 100644 --- a/tests/test_auth_config.py +++ b/tests/test_auth_config.py @@ -12,7 +12,10 @@ PROJECT_ROOT = Path(__file__).resolve().parent.parent load_dotenv(PROJECT_ROOT / ".env") # Configure browser path -os.environ["PLAYWRIGHT_BROWSERS_PATH"] = os.getenv("PLAYWRIGHT_BROWSERS_PATH") +browser_path = os.getenv("PLAYWRIGHT_BROWSERS_PATH") +if not browser_path: + raise ValueError("PLAYWRIGHT_BROWSERS_PATH environment variable is required") +os.environ["PLAYWRIGHT_BROWSERS_PATH"] = browser_path print("=" * 50) print("ERP System Configuration Check") diff --git a/tests/test_extractor_real.py b/tests/test_extractor_real.py index 09edfa2..e4520fd 100644 --- a/tests/test_extractor_real.py +++ b/tests/test_extractor_real.py @@ -86,12 +86,20 @@ try: # Setup browser and extract data print("\n[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=os.getenv("ERP_USERNAME"), - password=os.getenv("ERP_PASSWORD"), - url=get_login_url(os.getenv("ERP_URL")), + 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, diff --git a/tests/test_login.py b/tests/test_login.py index 8dc4b70..2e1515c 100644 --- a/tests/test_login.py +++ b/tests/test_login.py @@ -18,7 +18,10 @@ from dotenv import load_dotenv load_dotenv(PROJECT_ROOT / ".env") # Configure browser path -os.environ["PLAYWRIGHT_BROWSERS_PATH"] = os.getenv("PLAYWRIGHT_BROWSERS_PATH") +browser_path = os.getenv("PLAYWRIGHT_BROWSERS_PATH") +if not browser_path: + raise ValueError("PLAYWRIGHT_BROWSERS_PATH environment variable is required") +os.environ["PLAYWRIGHT_BROWSERS_PATH"] = browser_path print("=" * 60) print("Testing Yonyou BIP Login and Logout") @@ -38,15 +41,24 @@ if not base_url: url = f"{base_url.rstrip('/')}/yonbip/resources/uap/rbac/login/main/index.html" try: + # Get required environment variables + username = os.getenv('ERP_USERNAME') + password = os.getenv('ERP_PASSWORD') + headless = os.getenv('ERP_HEADLESS', 'false').lower() in ('true', '1', 'yes') + ignore_https_errors = os.getenv('ERP_IGNORE_HTTPS_ERRORS', 'true').lower() in ('true', '1', 'yes') + + if not username or not password: + raise ValueError("ERP_USERNAME and ERP_PASSWORD environment variables are required") + with sync_playwright() as p: print("\n[1/5] Starting browser...") browser, context, page, main_frame = login( playwright=p, - username=os.getenv('ERP_USERNAME'), - password=os.getenv('ERP_PASSWORD'), + username=username, + password=password, url=url, - headless=os.getenv('ERP_HEADLESS', 'false').lower() in ('true', '1', 'yes'), - ignore_https_errors=os.getenv('ERP_IGNORE_HTTPS_ERRORS', 'true').lower() in ('true', '1', 'yes'), + headless=headless, + ignore_https_errors=ignore_https_errors, verbose=True ) diff --git a/utils/__init__.py b/utils/__init__.py index 6423ece..fcb0013 100644 --- a/utils/__init__.py +++ b/utils/__init__.py @@ -3,8 +3,11 @@ Utils package for BIPAuto automation framework. """ from .auth import login, logout +from .logging import get_logger __all__ = [ + # Logging module (foundation) + "get_logger", # Auth module "login", "logout", diff --git a/utils/auth.py b/utils/auth.py index 14f8ee2..319ed42 100644 --- a/utils/auth.py +++ b/utils/auth.py @@ -2,7 +2,9 @@ Authentication module - Responsible for Yonyou BIP system login and logout operations """ -from playwright.sync_api import Playwright, Browser, BrowserContext, Page, Frame +import logging +from typing import Optional +from playwright.sync_api import Playwright, Browser, BrowserContext, Page, Frame, FrameLocator def login( @@ -13,7 +15,8 @@ def login( headless: bool, ignore_https_errors: bool, verbose: bool = True, -) -> tuple[Browser, BrowserContext, Page, Frame]: + logger: Optional[logging.Logger] = None, +) -> tuple[Browser, BrowserContext, Page, FrameLocator]: """ Login to Yonyou BIP system @@ -24,7 +27,8 @@ def login( url: Complete login page URL (required) headless: Whether to use headless mode (required) ignore_https_errors: Whether to ignore HTTPS errors (required) - verbose: Whether to print detailed logs (default: True) + verbose: Whether to print detailed logs (default: True). Deprecated, use logger instead. + logger: Optional logging.Logger instance. If None and verbose=True, creates default logger. Returns: tuple: (browser, context, page, main_frame) @@ -34,6 +38,13 @@ def login( - main_frame: Main iframe after login (forwardFrame) """ import time + from utils.logging import get_logger + + # Create default logger if needed + if logger is None and verbose: + logger = get_logger('bipauto.auth') + elif logger is None: + logger = None # Silent mode, no logging # Validate required parameters if not username or not username.strip(): @@ -71,27 +82,35 @@ def login( confirm_btn = main_frame.get_by_role("button", name="确定") if confirm_btn.count() > 0: confirm_btn.click() - if verbose: - print("Force login detected") + if logger: + logger.info("Force login detected") else: - if verbose: - print("Normal login") + if logger: + logger.debug("Normal login") return browser, context, page, main_frame -def logout(main_frame: Frame, verbose: bool = True) -> None: +def logout(main_frame: FrameLocator, verbose: bool = True, logger: Optional[logging.Logger] = None) -> None: """ Execute account logout Args: main_frame: Main iframe object (forwardFrame) - verbose: Whether to print detailed logs + verbose: Whether to print detailed logs. Deprecated, use logger instead. + logger: Optional logging.Logger instance. If None and verbose=True, creates default logger. """ import time + from utils.logging import get_logger - if verbose: - print("Clicking account menu button...") + # Create default logger if needed + if logger is None and verbose: + logger = get_logger('bipauto.auth') + elif logger is None: + logger = None # Silent mode, no logging + + if logger: + logger.info("Clicking account menu button...") # Element 1: Account menu button (logo icon) main_frame.get_by_role("img", name="logo").click() @@ -99,8 +118,8 @@ def logout(main_frame: Frame, verbose: bool = True) -> None: # Wait for menu to appear time.sleep(1) - if verbose: - print("Clicking logout button...") + if logger: + logger.info("Clicking logout button...") # Element 2: "Logout" button main_frame.get_by_text("退出登录").click() @@ -108,19 +127,19 @@ def logout(main_frame: Frame, verbose: bool = True) -> None: # Wait for confirmation dialog to appear time.sleep(1) - if verbose: - print("Waiting for logout confirmation dialog...") + if logger: + logger.info("Waiting for logout confirmation dialog...") # Element 3: Logout confirmation dialog (check if appears) try: confirm_text = main_frame.get_by_text("退出确定要退出当前账号吗?") - if verbose: - print("Found confirmation dialog, clicking confirm button") + if logger: + logger.info("Found confirmation dialog, clicking confirm button") # Element 4: Confirm button main_frame.get_by_role("button", name="确定(Y)").click() except: - if verbose: - print("Confirmation dialog not found, may have auto-logged out") + if logger: + logger.warning("Confirmation dialog not found, may have auto-logged out") time.sleep(2) # Wait for logout to complete diff --git a/utils/discrete_material_plan/excel_converter.py b/utils/discrete_material_plan/excel_converter.py index 43f5ea3..5eb2486 100644 --- a/utils/discrete_material_plan/excel_converter.py +++ b/utils/discrete_material_plan/excel_converter.py @@ -1,139 +1,150 @@ """ -Excel 报表数据转换工具组件 -将 Excel 报表数据转换为数据库记录形式 +Excel Report Data Conversion Utility +Converts Excel report data to database record format """ import pandas as pd import openpyxl from typing import List, Dict, Optional import os +import logging +from utils.logging import get_logger class ExcelConverter: - """Excel 报表数据转换器""" + """Excel Report Data Converter""" - # 字段名称映射(解决字段名冲突) - FIELD_NAME_MAPPING = {"计划数量": "产品计划数量", "单位": "产品单位"} + # Field name mapping (resolves field name conflicts) + FIELD_NAME_MAPPING = {"计划数量": "Product planned quantity", "单位": "Product unit"} - def __init__(self, verbose: bool = True): + def __init__(self, verbose: bool = True, logger: Optional[logging.Logger] = None): """ - 初始化转换器 + Initialize the converter Args: - verbose: 是否打印详细日志 + verbose: Whether to print detailed logs (default: True). Deprecated, use logger instead. + logger: Optional logging.Logger instance. If None and verbose=True, creates default logger. """ self.verbose = verbose + + # Create default logger if needed + if logger is None and verbose: + self.logger = get_logger('bipauto.converter') + elif logger is None: + # Silent mode - create a logger but disable output + silent_logger = logging.getLogger('bipauto.converter.silent') + silent_logger.setLevel(logging.CRITICAL + 1) # Higher than critical, never logs + self.logger = silent_logger + else: + self.logger = logger - def _print(self, *args, **kwargs): - """打印日志(如果 verbose=True)""" - if self.verbose: - print(*args, **kwargs) - - def convert(self, input_file: str, output_file: str = None) -> pd.DataFrame: + def convert(self, input_file: str, output_file: Optional[str] = None) -> pd.DataFrame: """ - 转换 Excel 文件 + Convert Excel file Args: - input_file: 输入文件路径 - output_file: 输出文件路径(可选,不指定则不保存) + input_file: Input file path + output_file: Output file path (optional, not saved if not specified) Returns: - 转换后的 DataFrame + Converted DataFrame """ - # 处理输出文件名 + # Handle output file if output_file: output_file = self._handle_output_file(output_file) - # 读取工作表 + # Read worksheet wb = openpyxl.load_workbook(input_file) ws = wb.active - # 解析订单数据 + # Parse order data orders = self._parse_sheet(ws) - # 转换为 DataFrame + # Convert to DataFrame df = self._convert_to_dataframe(orders) if not df.empty: - # 保存文件 + # Save file if output_file: df.to_excel(output_file, index=False) - # 打印汇总报告 - self._print("=" * 60) - self._print("转换完成") - self._print("=" * 60) - self._print(f"订单数: {len(orders)}") - self._print(f"数据行数: {len(df)}") - self._print(f"输出文件: {output_file if output_file else 'N/A'}") - self._print("=" * 60) + # Print summary report (using logger) + self.logger.info("=" * 60) + self.logger.info("Conversion complete") + self.logger.info("=" * 60) + self.logger.info(f"Order count: {len(orders)}") + self.logger.info(f"Data row count: {len(df)}") + self.logger.info(f"Output file: {output_file if output_file else 'N/A'}") + self.logger.info("=" * 60) return df def _handle_output_file(self, output_file: str) -> str: """ - 处理输出文件,如果文件存在则尝试删除 + Handle output file, attempt to delete if it exists Args: - output_file: 输出文件路径 + output_file: Output file path Returns: - 实际使用的输出文件路径 + Actual output file path used """ if os.path.exists(output_file): try: os.remove(output_file) except PermissionError: - self._print(f"警告: 无法删除 {output_file},可能文件被其他程序打开") - # 修改文件名 + self.logger.warning( + f"Warning: Could not delete {output_file}, file may be open by another program" + ) + # Modify filename base, ext = os.path.splitext(output_file) output_file = f"{base}_new{ext}" return output_file def _parse_sheet(self, ws) -> List[Dict]: """ - 解析一个工作表,返回所有订单的数据 + Parse a worksheet and return data for all orders - 每个订单包含: - - order_info: 订单头信息(包括页脚) - - materials: 物料数据列表 + Each order contains: + - order_info: Order header information (including footer) + - materials: List of material data Args: - ws: openpyxl 工作表对象 + ws: openpyxl worksheet object Returns: - 订单列表 + Order list """ orders = [] all_rows = list(ws.iter_rows(values_only=True)) - # 逐行扫描,按订单结构解析 + # Scan row by row, parse by order structure i = 0 while i < len(all_rows): row = all_rows[i] - # 检查是否是订单标题行 + # Check if this is the order title row if row and "离散备料计划" in str(row[0]): - # 解析订单头信息(接下来的4行) + # Parse order header information (next 4 rows) order_info = {} for j in range(1, 5): if i + j < len(all_rows) and all_rows[i + j]: self._parse_header_row(all_rows[i + j], order_info) - # 跳过空行,找到表格标题行 + # Skip empty rows, find table header row table_row = i + 5 while table_row < len(all_rows) and ( not all_rows[table_row] or not all_rows[table_row][0] ): table_row += 1 - # 检查是否是表格标题行 + # Check if this is the table header row if ( table_row < len(all_rows) and all_rows[table_row] and all_rows[table_row][0] == "序号" ): - # 检查表头下一行是否为空,判断是否存在数据 + # Check if the row below the header is empty to determine if data exists next_row = table_row + 1 is_empty_row = ( next_row < len(all_rows) @@ -145,7 +156,7 @@ class ExcelConverter: ) if is_empty_row: - # 没有数据,查找页脚信息 + # No data, find footer information materials = [] footer_info = {} data_row = next_row + 1 @@ -172,18 +183,18 @@ class ExcelConverter: } ) else: - # 有数据,开始提取物料 + # Has data, start extracting materials materials = [] - footer_info = {} # 页脚信息 + footer_info = {} # Footer information data_row = table_row + 1 while data_row < len(all_rows) and all_rows[data_row]: - # 检查是否是页脚信息(制单人、打印人) + # Check if this is footer information (creator, printer) if all_rows[data_row + 1][0] and "制单人" in str( all_rows[data_row + 1][0] ): - # 解析页脚信息 + # Parse footer information self._parse_header_row(all_rows[data_row], footer_info) - # 检查下一行是否也是页脚信息 + # Check if the next row is also footer information if ( data_row + 1 < len(all_rows) and all_rows[data_row + 1] @@ -193,7 +204,7 @@ class ExcelConverter: ) break - # 提取物料数据 + # Extract material data material_row = all_rows[data_row] material = { "序号": material_row[0], @@ -227,45 +238,45 @@ class ExcelConverter: def _parse_header_row(self, row: tuple, info: Dict): """ - 解析订单头信息的一行(字段名和值交错排列) + Parse a row of order header information (field names and values interleaved) Args: - row: 行数据 - info: 存储解析结果的字典 + row: Row data + info: Dictionary to store parsing results """ i = 0 while i < len(row): cell = row[i] if cell and str(cell).strip() and ":" in str(cell): - # 找到字段名 - field_name = str(cell).replace(":", "").strip() + # Find field name + field_name = str(cell).replace(":", "").strip() - # 应用字段名映射 + # Apply field name mapping if field_name in self.FIELD_NAME_MAPPING: field_name = self.FIELD_NAME_MAPPING[field_name] - # 跳过空单元格,找到第一个非字段名的值 + # Skip empty cells, find the first non-field-name value j = i + 1 while j < len(row) and ( - not row[j] or not str(row[j]).strip() or ":" in str(row[j]) + not row[j] or not str(row[j]).strip() or ":" in str(row[j]) ): j += 1 - if j < len(row) and row[j] and not ":" in str(row[j]): + if j < len(row) and row[j] and ":" not in str(row[j]): info[field_name] = str(row[j]).strip() - # 跳过已处理的值,继续找下一个字段名 + # Skip processed value, continue to find next field name i = j + 1 else: i += 1 def _convert_to_dataframe(self, orders: List[Dict]) -> pd.DataFrame: """ - 将订单数据转换为扁平化的 DataFrame + Convert order data to a flattened DataFrame Args: - orders: 订单列表 + orders: Order list Returns: - 扁平化的 DataFrame + Flattened DataFrame """ all_records = [] diff --git a/utils/discrete_material_plan/extractor.py b/utils/discrete_material_plan/extractor.py index df3eaa4..77e65b8 100644 --- a/utils/discrete_material_plan/extractor.py +++ b/utils/discrete_material_plan/extractor.py @@ -9,7 +9,9 @@ Caller is responsible for browser/session lifecycle management. import pandas as pd from pathlib import Path from typing import List, Optional, Tuple -from playwright.sync_api import Page, Frame +from playwright.sync_api import Page, Frame, FrameLocator +import logging +from utils.logging import get_logger def chunk_order_ids(order_ids: List[str], batch_size: int) -> List[List[str]]: @@ -51,11 +53,12 @@ def get_login_url(base_url: str) -> str: def extract_batch( - work_frame: Frame, + work_frame: FrameLocator, page: Page, order_ids: List[str], batch_index: int, download_dir: str, + logger: Optional[logging.Logger] = None, ) -> str: """ Execute download workflow for a single batch of order IDs. @@ -69,6 +72,7 @@ def extract_batch( 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 + logger: Optional logger for debug output (silent if None) Returns: Full path to the downloaded Excel file @@ -81,15 +85,17 @@ def extract_batch( order_ids=order_ids, batch_index=batch_index, download_dir=download_dir, + logger=logger, ) def extract_batches( - work_frame: Frame, + work_frame: FrameLocator, page: Page, order_ids: List[str], download_dir: str, batch_size: int = 10, + logger: Optional[logging.Logger] = None, ) -> List[str]: """ Download data for multiple batches of order IDs. @@ -105,6 +111,7 @@ def extract_batches( 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 + logger: Optional logger for debug output (silent if None) Returns: List of paths to downloaded Excel files @@ -124,7 +131,7 @@ def extract_batches( chunks = chunk_order_ids(order_ids, batch_size) # Setup query interface once - setup_query_interface(work_frame) + setup_query_interface(work_frame, logger) # Process each batch for batch_index, batch in enumerate(chunks): @@ -134,6 +141,7 @@ def extract_batches( order_ids=batch, batch_index=batch_index, download_dir=download_dir, + logger=logger, ) downloaded_files.append(file_path) @@ -145,6 +153,7 @@ def post_process_downloads( output_file: str, verbose: bool = True, cleanup_temp_files: bool = True, + logger: Optional[logging.Logger] = None, ) -> Tuple[str, pd.DataFrame]: """ Convert and merge downloaded Excel files into structured DataFrame. @@ -154,8 +163,9 @@ def post_process_downloads( Args: downloaded_files: List of paths to downloaded Excel files output_file: Path to save merged Excel result - verbose: Whether to print progress messages + verbose: Whether to print progress messages (deprecated, use logger instead) cleanup_temp_files: Whether to delete temporary downloaded files after processing (default: True) + logger: Optional logger for progress output. If None and verbose=True, creates default logger. Returns: Tuple of (output_file_path, merged_dataframe) @@ -170,13 +180,26 @@ def post_process_downloads( """ from .excel_converter import ExcelConverter - converter = ExcelConverter(verbose=verbose) + # Create default logger if needed + if logger is None and verbose: + logger = logging.getLogger('bipauto.extractor.post_process') + 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 + elif logger is None: + # Silent mode + logger = logging.getLogger('bipauto.extractor.post_process.silent') + logger.setLevel(logging.CRITICAL + 1) + + converter = ExcelConverter(verbose=verbose, logger=logger) 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}") + logger.info(f"Converting file {i + 1}/{len(downloaded_files)}: {file_path}") # Convert (do not save intermediate result) df = converter.convert(input_file=file_path) @@ -193,43 +216,55 @@ def post_process_downloads( 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)}") + logger.info(f"Merged result saved to: {output_path}") + logger.info(f"Total rows: {len(merged_df)}") # Cleanup temporary downloaded files if cleanup_temp_files: - _cleanup_temp_files(downloaded_files, verbose) + _cleanup_temp_files(downloaded_files, logger=logger) return str(output_path), merged_df -def _cleanup_temp_files(downloaded_files: List[str], verbose: bool = True) -> int: +def _cleanup_temp_files(downloaded_files: List[str], logger: Optional[logging.Logger] = None, verbose: bool = True) -> int: """ Remove temporary downloaded files. Args: downloaded_files: List of file paths to delete - verbose: Whether to print progress messages + logger: Optional logger for progress output. If None and verbose=True, creates default logger. + verbose: Whether to print progress messages (deprecated, use logger instead) Returns: Number of files successfully deleted """ + # Create default logger if needed + if logger is None and verbose: + logger = logging.getLogger('bipauto.extractor.cleanup') + 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 + elif logger is None: + # Silent mode + logger = logging.getLogger('bipauto.extractor.cleanup.silent') + logger.setLevel(logging.CRITICAL + 1) + deleted_count = 0 for file_path in downloaded_files: try: Path(file_path).unlink() deleted_count += 1 - if verbose: - print(f"Deleted temp file: {file_path}") + logger.debug(f"Deleted temp file: {file_path}") except Exception as e: - if verbose: - print(f"Warning: Could not delete {file_path}: {e}") + logger.warning(f"Warning: Could not delete {file_path}: {e}") return deleted_count def extract_and_post_process( - work_frame: Frame, + work_frame: FrameLocator, page: Page, order_ids: List[str], download_dir: str, @@ -237,6 +272,7 @@ def extract_and_post_process( batch_size: int = 10, verbose: bool = True, cleanup_temp_files: bool = True, + logger: Optional[logging.Logger] = None, ) -> Tuple[str, pd.DataFrame]: """ Complete extraction workflow: download batches + post-process to merged Excel. @@ -251,8 +287,9 @@ def extract_and_post_process( 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 + verbose: Whether to print progress messages (deprecated, use logger instead) cleanup_temp_files: Whether to delete temporary downloaded files after processing (default: True) + logger: Optional logger for debug output. If None and verbose=True, creates default logger. Returns: Tuple of (output_file_path, merged_dataframe) @@ -268,9 +305,22 @@ def extract_and_post_process( >>> context.close() >>> browser.close() """ + # Create default logger if needed + if logger is None and verbose: + logger = logging.getLogger('bipauto.extractor') + 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 + elif logger is None: + # Silent mode + logger = logging.getLogger('bipauto.extractor.silent') + logger.setLevel(logging.CRITICAL + 1) + # Step 1: Download all batches - if verbose: - print(f"Downloading {len(order_ids)} orders in batches of {batch_size}...") + logger.info(f"Downloading {len(order_ids)} orders in batches of {batch_size}...") downloaded_files = extract_batches( work_frame=work_frame, @@ -278,10 +328,10 @@ def extract_and_post_process( order_ids=order_ids, download_dir=download_dir, batch_size=batch_size, + logger=logger, ) - if verbose: - print(f"Downloaded {len(downloaded_files)} batch file(s)") + logger.info(f"Downloaded {len(downloaded_files)} batch file(s)") # Step 2: Post-process (convert + merge) output_path, merged_df = post_process_downloads( @@ -289,6 +339,7 @@ def extract_and_post_process( output_file=output_file, verbose=verbose, cleanup_temp_files=cleanup_temp_files, + logger=logger, ) return output_path, merged_df @@ -321,13 +372,14 @@ def read_order_ids_from_file(id_file: str, encoding: str = "utf-8") -> List[str] def extract_from_file( id_file: str, - work_frame: Frame, + work_frame: FrameLocator, page: Page, download_dir: str, output_file: str, batch_size: int = 10, verbose: bool = True, cleanup_temp_files: bool = True, + logger: Optional[logging.Logger] = None, ) -> Tuple[str, pd.DataFrame]: """ Extract data from order IDs in a file and post-process to merged Excel. @@ -341,8 +393,9 @@ def extract_from_file( 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 + verbose: Whether to print progress messages (deprecated, use logger instead) cleanup_temp_files: Whether to delete temporary downloaded files after processing (default: True) + logger: Optional logger for debug output. If None and verbose=True, creates default logger. Returns: Tuple of (output_file_path, merged_dataframe) @@ -358,8 +411,22 @@ def extract_from_file( """ order_ids = read_order_ids_from_file(id_file) - if verbose: - print(f"Loaded {len(order_ids)} order IDs from {id_file}") + # Create default logger if needed (for this function's own logging) + func_logger = logger + if func_logger is None and verbose: + func_logger = logging.getLogger('bipauto.extractor.file') + func_logger.setLevel(logging.INFO) + if not func_logger.handlers: + handler = logging.StreamHandler() + handler.setFormatter(logging.Formatter("[%(levelname)s] %(name)s: %(message)s")) + func_logger.addHandler(handler) + func_logger.propagate = False + elif func_logger is None: + # Silent mode + func_logger = logging.getLogger('bipauto.extractor.file.silent') + func_logger.setLevel(logging.CRITICAL + 1) + + func_logger.info(f"Loaded {len(order_ids)} order IDs from {id_file}") return extract_and_post_process( work_frame=work_frame, @@ -370,4 +437,5 @@ def extract_from_file( batch_size=batch_size, verbose=verbose, cleanup_temp_files=cleanup_temp_files, + logger=logger, ) diff --git a/utils/discrete_material_plan/extractor_core.py b/utils/discrete_material_plan/extractor_core.py index af5ea3b..2fc43b0 100644 --- a/utils/discrete_material_plan/extractor_core.py +++ b/utils/discrete_material_plan/extractor_core.py @@ -6,17 +6,19 @@ All functions are stateless and accept required parameters explicitly. import re import os -from typing import List -from playwright.sync_api import Page, Frame, TimeoutError +import logging +from typing import List, Optional +from playwright.sync_api import Page, Frame, FrameLocator, TimeoutError -def navigate_to_discrete_material_page(main_frame: Frame, page: Page) -> tuple[Frame, Page]: +def navigate_to_discrete_material_page(main_frame: FrameLocator, page: Page, logger: Optional[logging.Logger] = None) -> tuple[FrameLocator, Page]: """ Navigate to the discrete material plan maintenance page. Args: - main_frame: The main forwardFrame iframe + main_frame: The main forwardFrame iframe (FrameLocator) page: The Playwright page object + logger: Optional logger for debug output (silent if None) Returns: tuple: (work_frame, page1) - The work iframe and the new popup page @@ -35,16 +37,20 @@ def navigate_to_discrete_material_page(main_frame: Frame, page: Page) -> tuple[F inner_frame_locator = f_frame.locator("#mainiframe") inner_frame_locator.wait_for(state="visible", timeout=15000) work_frame = inner_frame_locator.content_frame + + if logger: + logger.debug("Navigated to discrete material plan page") return work_frame, page1 -def setup_query_interface(work_frame: Frame) -> None: +def setup_query_interface(work_frame: FrameLocator, logger: Optional[logging.Logger] = None) -> None: """ Initialize the query interface by selecting order number query tab. Args: work_frame: The inner work iframe containing the query interface + logger: Optional logger for debug output (silent if None) """ # Open search panel work_frame.locator(".search-name-wrapper > .iconfont").click() @@ -59,15 +65,19 @@ def setup_query_interface(work_frame: Frame) -> None: input_box = work_frame.locator("#rc_select_0") input_box.fill("5000") input_box.press("Enter") + + if logger: + logger.debug("Query interface setup complete") -def fill_and_search_orders(work_frame: Frame, order_ids: List[str]) -> None: +def fill_and_search_orders(work_frame: FrameLocator, order_ids: List[str], logger: Optional[logging.Logger] = None) -> None: """ Fill order IDs into the search textbox and trigger search. Args: work_frame: The work iframe containing the search form order_ids: List of order IDs to search for + logger: Optional logger for debug output (silent if None) """ textbox = work_frame.get_by_role("textbox", name="来源生产订单号") @@ -84,15 +94,20 @@ def fill_and_search_orders(work_frame: Frame, order_ids: List[str]) -> None: loading_locator.wait_for(state="visible", timeout=3000) loading_locator.wait_for(state="hidden", timeout=0) except TimeoutError: - pass + if logger: + logger.debug("Loading indicator timeout - continuing anyway") + + if logger: + logger.debug(f"Searched for {len(order_ids)} order IDs") def download_batch_data( - work_frame: Frame, + work_frame: FrameLocator, page: Page, order_ids: List[str], batch_index: int, - download_dir: str + download_dir: str, + logger: Optional[logging.Logger] = None ) -> str: """ Execute the download workflow for a single batch of order IDs. @@ -103,6 +118,7 @@ def download_batch_data( order_ids: List of order IDs to download batch_index: Zero-based batch index for naming the output file download_dir: Directory path to save the downloaded file + logger: Optional logger for info output (silent if None) Returns: str: Full path to the downloaded file @@ -130,22 +146,26 @@ def download_batch_data( # Step 5: Trigger download and save file download_filename = f"temp_batch_{batch_index + 1}.xlsx" download_path = os.path.join(download_dir, download_filename) - + with page.expect_download() as download_info: work_frame.get_by_role("button", name="确定(Y)").click() download = download_info.value download.save_as(download_path) + + if logger: + logger.info(f"Downloaded batch {batch_index + 1} to {download_path}") return download_path def execute_batch_download_workflow( - work_frame: Frame, + work_frame: FrameLocator, page: Page, order_ids: List[str], batch_index: int, - download_dir: str + download_dir: str, + logger: Optional[logging.Logger] = None ) -> str: """ Complete workflow: fill orders, search, and download for a single batch. @@ -156,9 +176,10 @@ def execute_batch_download_workflow( 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 + logger: Optional logger for debug output (silent if None) Returns: str: Full path to the downloaded file """ - fill_and_search_orders(work_frame, order_ids) - return download_batch_data(work_frame, page, order_ids, batch_index, download_dir) + fill_and_search_orders(work_frame, order_ids, logger) + return download_batch_data(work_frame, page, order_ids, batch_index, download_dir, logger) diff --git a/utils/logging.py b/utils/logging.py new file mode 100644 index 0000000..95614d4 --- /dev/null +++ b/utils/logging.py @@ -0,0 +1,81 @@ +""" +Centralized logging utility module for BIPAuto project. + +Provides a standardized logger configuration for consistent logging across +all BIPAuto components. Supports console output with optional file logging. + +Basic usage examples: + + # Basic usage - console logging at INFO level + from utils.logging import get_logger + logger = get_logger('bipauto.auth') + logger.info('Login successful') + + # With debug level for detailed output + logger = get_logger('bipauto.extractor', level=logging.DEBUG) + logger.debug('Processing batch 1 of 5...') + + # With file output for persistent logs + logger = get_logger('bipauto.app', level=logging.INFO, log_file='app.log') + logger.info('Application started') + +Logger naming convention: + Use hierarchical names with 'bipauto.' prefix: + - 'bipauto.auth' - Authentication module + - 'bipauto.extractor' - Material plan extractor + - 'bipauto.converter' - Excel converter + - 'bipauto.utils' - Utility functions +""" + +import logging +from typing import Optional + + +def get_logger( + name: str, + level: int = logging.INFO, + log_file: Optional[str] = None +) -> logging.Logger: + """ + Create and configure a logger with console and optional file handlers. + + Args: + name: Logger name (use hierarchical naming, e.g., 'bipauto.auth') + level: Logging level (default: logging.INFO) + log_file: Optional path to log file. If None, only console output. + + Returns: + Configured logging.Logger instance + + Example: + >>> logger = get_logger('bipauto.auth') + >>> logger.info('User logged in') + [INFO] bipauto.auth: User logged in + """ + logger = logging.getLogger(name) + logger.setLevel(level) + + # Avoid adding duplicate handlers if logger already configured + if logger.handlers: + return logger + + # Create formatter + formatter = logging.Formatter("[%(levelname)s] %(name)s: %(message)s") + + # Console handler (always added) + console_handler = logging.StreamHandler() + console_handler.setLevel(level) + console_handler.setFormatter(formatter) + logger.addHandler(console_handler) + + # File handler (optional) + if log_file: + file_handler = logging.FileHandler(log_file) + file_handler.setLevel(level) + file_handler.setFormatter(formatter) + logger.addHandler(file_handler) + + # Prevent log propagation to root logger (avoids duplicate output) + logger.propagate = False + + return logger