refactor: restructure main_clean.py with dependency injection and interfaces
Refactor discrete material cleaner to improve maintainability and testability: - Introduce dependency injection pattern for service components - Add interfaces (IPageNavigator, IMaterialExtractor, IDeletionChecker) - Extract PageNavigator service for page navigation logic - Extract MaterialExtractor service for data extraction - Extract DatabaseDeletionChecker service for deletion logic - Add MaterialInfo data model for structured data - Centralize configuration (CleanerConfig, UIConstants) - Implement separation of concerns across services layer Also includes comprehensive execution mechanism documentation. Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Misaka_Company
7
services/__init__.py
Normal file
7
services/__init__.py
Normal file
@@ -0,0 +1,7 @@
|
||||
"""服务模块 - 实现业务逻辑"""
|
||||
|
||||
from services.page_navigator import PageNavigator
|
||||
from services.material_extractor import MaterialExtractor
|
||||
from services.deletion_checker import DatabaseDeletionChecker
|
||||
|
||||
__all__ = ["PageNavigator", "MaterialExtractor", "DatabaseDeletionChecker"]
|
||||
45
services/deletion_checker.py
Normal file
45
services/deletion_checker.py
Normal file
@@ -0,0 +1,45 @@
|
||||
"""
|
||||
删除判断服务实现
|
||||
基于数据库查询判断物料是否需要删除
|
||||
"""
|
||||
|
||||
from interfaces.i_deletion_checker import IDeletionChecker
|
||||
from models.material_info import MaterialInfo
|
||||
from db.materials_to_delete import should_delete_material
|
||||
|
||||
|
||||
class DatabaseDeletionChecker(IDeletionChecker):
|
||||
"""基于数据库的删除判断实现"""
|
||||
|
||||
def __init__(self, manager_name: str, verbose: bool = True):
|
||||
"""
|
||||
初始化删除检查器
|
||||
|
||||
Args:
|
||||
manager_name: 负责人姓名
|
||||
verbose: 是否打印详细日志
|
||||
"""
|
||||
self.manager_name = manager_name
|
||||
self.verbose = verbose
|
||||
|
||||
def _print(self, *args, **kwargs):
|
||||
"""打印日志（如果 verbose=True）"""
|
||||
if self.verbose:
|
||||
print(*args, **kwargs)
|
||||
|
||||
def should_delete(self, material: MaterialInfo) -> bool:
|
||||
"""
|
||||
通过查询数据库判断是否需要删除
|
||||
|
||||
Args:
|
||||
material: 物料信息对象
|
||||
|
||||
Returns:
|
||||
是否需要删除
|
||||
"""
|
||||
result = should_delete_material(self.manager_name, material.code)
|
||||
if self.verbose and result:
|
||||
self._print(f">>> 需要清理：{material.name}【{material.code}】")
|
||||
elif self.verbose and not result:
|
||||
self._print(f"保留：{material.name}【{material.code}】无需清理")
|
||||
return result
|
||||
166
services/material_extractor.py
Normal file
166
services/material_extractor.py
Normal file
@@ -0,0 +1,166 @@
|
||||
"""
|
||||
物料信息提取服务实现
|
||||
实现从页面提取物料信息的具体逻辑
|
||||
"""
|
||||
|
||||
import re
|
||||
from playwright.sync_api import TimeoutError as PlaywrightTimeoutError
|
||||
from interfaces.i_material_extractor import IMaterialExtractor
|
||||
from config.ui_constants import UIConstants
|
||||
from models.material_info import MaterialInfo
|
||||
|
||||
|
||||
class MaterialExtractor(IMaterialExtractor):
|
||||
"""物料信息提取实现"""
|
||||
|
||||
def __init__(self, config: UIConstants, verbose: bool = True):
|
||||
"""
|
||||
初始化物料提取器
|
||||
|
||||
Args:
|
||||
config: UI 配置
|
||||
verbose: 是否打印详细日志
|
||||
"""
|
||||
self.config = config
|
||||
self.verbose = verbose
|
||||
|
||||
def _print(self, *args, **kwargs):
|
||||
"""打印日志（如果 verbose=True）"""
|
||||
if self.verbose:
|
||||
print(*args, **kwargs)
|
||||
|
||||
def extract_detail_count(self, frame) -> int:
|
||||
"""
|
||||
提取详细信息数量
|
||||
|
||||
Args:
|
||||
frame: 目标 iframe
|
||||
|
||||
Returns:
|
||||
详细信息数量
|
||||
"""
|
||||
detail_element = frame.get_by_text(re.compile(r"^详细信息 \(\d+\)$"))
|
||||
detail_text = detail_element.inner_text()
|
||||
match = re.search(r"详细信息 \((\d+)\)", detail_text)
|
||||
if match:
|
||||
detail_count = int(match.group(1))
|
||||
self._print(f"详细信息数量: {detail_count}")
|
||||
return detail_count
|
||||
return 0
|
||||
|
||||
def extract_detail_status(self, frame) -> str:
|
||||
"""
|
||||
提取备料状态
|
||||
|
||||
Args:
|
||||
frame: 目标 iframe
|
||||
|
||||
Returns:
|
||||
备料状态文本
|
||||
"""
|
||||
detail_element = frame.get_by_text(re.compile(r"^备料状态:.+$"))
|
||||
detail_text = detail_element.inner_text().replace("\n", "")
|
||||
match = re.search(r"^备料状态:(.+)$", detail_text)
|
||||
if match:
|
||||
detail_status = match.group(1)
|
||||
self._print(f"备料状态: {detail_status}")
|
||||
return detail_status
|
||||
return ""
|
||||
|
||||
def extract_materials(self, frame, count: int):
|
||||
"""
|
||||
提取所有物料信息
|
||||
|
||||
Args:
|
||||
frame: 目标 iframe
|
||||
count: 物料数量
|
||||
|
||||
Returns:
|
||||
物料信息列表
|
||||
"""
|
||||
materials = []
|
||||
|
||||
# 获取展开后的父容器
|
||||
child_form = frame.locator(self.config.selectors.CARD_TABLE_SIDE_BOX)
|
||||
child_form.wait_for(state="visible", timeout=self.config.timeouts.FORM_VISIBLE)
|
||||
self._print(f"父容器 {self.config.selectors.CARD_TABLE_SIDE_BOX} 已找到")
|
||||
|
||||
for i in range(count):
|
||||
serial_number = i + 1
|
||||
id_label_locator = child_form.get_by_text("序号 " + str(serial_number))
|
||||
id_label_locator.wait_for(
|
||||
state="visible", timeout=self.config.timeouts.SERIAL_VISIBLE
|
||||
)
|
||||
self._print(f"处理 {id_label_locator.inner_text()}")
|
||||
|
||||
# 提取物料信息
|
||||
code = self._extract_material_code(child_form)
|
||||
name = self._extract_material_name(child_form)
|
||||
pending_quantity = self._extract_pending_quantity(child_form)
|
||||
shipped_quantity = self._extract_shipped_quantity(child_form)
|
||||
|
||||
material = MaterialInfo(
|
||||
serial_number=serial_number,
|
||||
code=code,
|
||||
name=name,
|
||||
pending_quantity=pending_quantity,
|
||||
shipped_quantity=shipped_quantity,
|
||||
)
|
||||
materials.append(material)
|
||||
|
||||
# 导航到下一个物料
|
||||
self._navigate_to_next_material(child_form, i, count)
|
||||
|
||||
return materials
|
||||
|
||||
def _extract_material_code(self, child_form) -> str:
|
||||
"""提取物料编码"""
|
||||
input_box = (
|
||||
child_form.locator("div")
|
||||
.filter(has_text=re.compile(r"^材料编码\d{11}$", re.MULTILINE))
|
||||
.locator("input")
|
||||
.first
|
||||
)
|
||||
code = input_box.input_value()
|
||||
self._print(f"材料编码: {code}")
|
||||
return code
|
||||
|
||||
def _extract_material_name(self, child_form) -> str:
|
||||
"""提取物料名称"""
|
||||
input_box = (
|
||||
child_form.locator("div")
|
||||
.filter(has_text=re.compile(r"^材料名称$"))
|
||||
.locator("input[type='text']")
|
||||
)
|
||||
name = input_box.input_value()
|
||||
self._print(f"材料名称: {name}")
|
||||
return name
|
||||
|
||||
def _extract_pending_quantity(self, child_form) -> str:
|
||||
"""提取累计待发数量"""
|
||||
input_box = (
|
||||
child_form.locator("div")
|
||||
.filter(has_text=re.compile(r"^累计待发数量$"))
|
||||
.locator("input[type='text']")
|
||||
)
|
||||
quantity = input_box.input_value()
|
||||
self._print(f"累计待发数量: {quantity}")
|
||||
return quantity
|
||||
|
||||
def _extract_shipped_quantity(self, child_form) -> str:
|
||||
"""提取累计出库数量"""
|
||||
input_box = (
|
||||
child_form.locator("div")
|
||||
.filter(has_text=re.compile(r"^累计出库数量$"))
|
||||
.locator("input[type='text']")
|
||||
)
|
||||
quantity = input_box.input_value()
|
||||
self._print(f"累计出库数量: {quantity}")
|
||||
return quantity
|
||||
|
||||
def _navigate_to_next_material(self, child_form, current_index: int, total_count: int):
|
||||
"""导航到下一个物料"""
|
||||
if current_index != total_count - 1:
|
||||
child_form.get_by_role("button").filter(has_text=re.compile(r"^$")).nth(2).click()
|
||||
else:
|
||||
child_form.get_by_role("button").filter(has_text=re.compile(r"^$")).nth(4).click()
|
||||
171
services/page_navigator.py
Normal file
171
services/page_navigator.py
Normal file
@@ -0,0 +1,171 @@
|
||||
"""
|
||||
页面导航服务实现
|
||||
实现页面导航、iframe 操作等具体逻辑
|
||||
"""
|
||||
|
||||
import re
|
||||
import time
|
||||
from playwright.sync_api import TimeoutError as PlaywrightTimeoutError
|
||||
from interfaces.i_page_navigator import IPageNavigator
|
||||
from config.ui_constants import UIConstants
|
||||
|
||||
|
||||
class PageNavigator(IPageNavigator):
|
||||
"""页面导航实现"""
|
||||
|
||||
def __init__(self, config: UIConstants, verbose: bool = True):
|
||||
"""
|
||||
初始化页面导航器
|
||||
|
||||
Args:
|
||||
config: UI 配置
|
||||
verbose: 是否打印详细日志
|
||||
"""
|
||||
self.config = config
|
||||
self.verbose = verbose
|
||||
|
||||
def _print(self, *args, **kwargs):
|
||||
"""打印日志（如果 verbose=True）"""
|
||||
if self.verbose:
|
||||
print(*args, **kwargs)
|
||||
|
||||
def navigate_to_main_page(self, page) -> tuple:
|
||||
"""
|
||||
导航到主页面 iframe
|
||||
|
||||
实际返回 (page, main_frame, inner_frame) 的元组
|
||||
"""
|
||||
# 点击打开"功能菜单"
|
||||
main_frame = page.locator(self.config.selectors.FORWARD_FRAME).content_frame
|
||||
main_frame.locator("i").first.click()
|
||||
|
||||
# 点击打开"离散生产订单维护"
|
||||
with page.expect_popup() as page1_info:
|
||||
main_frame.get_by_title("离散生产订单维护", exact=True).first.click()
|
||||
page1 = page1_info.value
|
||||
|
||||
# 获取 nested iframe
|
||||
main_frame = page1.locator(self.config.selectors.FORWARD_FRAME).content_frame
|
||||
inner_frame_locator = main_frame.locator(self.config.selectors.MAIN_IFRAME)
|
||||
inner_frame_locator.wait_for(
|
||||
state="visible", timeout=self.config.timeouts.IFRAME_VISIBLE
|
||||
)
|
||||
inner_frame = inner_frame_locator.content_frame
|
||||
|
||||
return page1, main_frame, inner_frame
|
||||
|
||||
def navigate_to_order_page(self, main_frame, page):
|
||||
"""导航到订单页面"""
|
||||
# 这个方法在当前实现中与 navigate_to_main_page 类似
|
||||
# 返回新的页面对象和内部 iframe
|
||||
return self.navigate_to_main_page(page)
|
||||
|
||||
def navigate_to_material_plan_page(self, page):
|
||||
"""
|
||||
导航到备料计划页面
|
||||
|
||||
Args:
|
||||
page: 订单页面对象 (page1)
|
||||
|
||||
Returns:
|
||||
(page2, inner_frame) - 新页面和内部 iframe
|
||||
"""
|
||||
# 获取主 iframe
|
||||
main_frame = page.locator(self.config.selectors.FORWARD_FRAME).content_frame
|
||||
inner_frame = main_frame.locator(self.config.selectors.MAIN_IFRAME).content_frame
|
||||
|
||||
# 点击"更多"
|
||||
inner_frame.locator(self.config.selectors.HOT_KEY_HEAD).get_by_text(
|
||||
self.config.selectors.MORE_BUTTON
|
||||
).click()
|
||||
|
||||
# 点击"备料计划"并等待新页面
|
||||
with page.expect_popup() as page2_info:
|
||||
inner_frame.get_by_text(self.config.selectors.MATERIAL_PLAN).click()
|
||||
page2 = page2_info.value
|
||||
|
||||
# 获取 nested iframe
|
||||
main_frame = page2.locator(self.config.selectors.FORWARD_FRAME).content_frame
|
||||
inner_frame_locator = main_frame.locator(self.config.selectors.MAIN_IFRAME)
|
||||
inner_frame_locator.wait_for(
|
||||
state="visible", timeout=self.config.timeouts.IFRAME_VISIBLE
|
||||
)
|
||||
inner_frame = inner_frame_locator.content_frame
|
||||
|
||||
# 等待备料计划页面加载完成
|
||||
self._wait_for_plan_page_loaded(inner_frame)
|
||||
|
||||
return page2, inner_frame
|
||||
|
||||
def setup_query_interface(self, frame):
|
||||
"""设置查询界面"""
|
||||
# 点击图标按钮打开查询界面
|
||||
frame.locator(self.config.selectors.SEARCH_ICON).click()
|
||||
frame.get_by_text(self.config.selectors.QUERY_BY_ORDER).click()
|
||||
frame.get_by_role("tab", name=self.config.selectors.TAB_ALL).click()
|
||||
|
||||
# 填充并验证显示数量，如果失败则重试
|
||||
max_retries = self.config.retry.DISPLAY_COUNT_MAX_RETRIES
|
||||
expected_value = self.config.retry.EXPECTED_DISPLAY_COUNT
|
||||
|
||||
for attempt in range(max_retries):
|
||||
frame.locator(self.config.selectors.DISPLAY_COUNT_INPUT).fill(expected_value)
|
||||
frame.locator(self.config.selectors.DISPLAY_COUNT_INPUT).press("Enter")
|
||||
# 检查填充是否成功
|
||||
actual_value = frame.locator(self.config.selectors.DISPLAY_COUNT_INPUT).input_value()
|
||||
if actual_value == expected_value:
|
||||
self._print(f"文本框填充成功: {expected_value}")
|
||||
break
|
||||
else:
|
||||
self._print(
|
||||
f"第 {attempt + 1} 次填充失败，实际值: {actual_value}，重试..."
|
||||
)
|
||||
if attempt == max_retries - 1:
|
||||
self._print(
|
||||
f"警告: {max_retries} 次尝试后仍未成功填充，继续执行..."
|
||||
)
|
||||
|
||||
def wait_for_page_loaded(self, frame):
|
||||
"""等待页面加载完成"""
|
||||
loading_locator = frame.locator("div").filter(has_text="加载中").nth(1)
|
||||
try:
|
||||
loading_locator.wait_for(
|
||||
state="visible", timeout=self.config.timeouts.LOADING_VISIBLE_WAIT
|
||||
)
|
||||
loading_locator.wait_for(
|
||||
state="hidden", timeout=self.config.timeouts.LOADING_HIDDEN_WAIT
|
||||
)
|
||||
except PlaywrightTimeoutError:
|
||||
# 加载很快完成，或者没有出现加载提示
|
||||
pass
|
||||
return True
|
||||
|
||||
def _wait_for_plan_page_loaded(self, inner_frame):
|
||||
"""等待备料计划页面加载完成（内部方法）"""
|
||||
self._print("等待备料计划页面加载完成...")
|
||||
plan_code_locator = inner_frame.get_by_text(
|
||||
re.compile(r"^离散备料计划维护：")
|
||||
)
|
||||
plan_code_locator.wait_for(
|
||||
state="visible", timeout=self.config.timeouts.PLAN_CODE_VISIBLE
|
||||
)
|
||||
|
||||
# 循环检查编码是否已加载
|
||||
max_wait = self.config.timeouts.PLAN_CODE_MAX_WAIT
|
||||
wait_interval = self.config.timeouts.PLAN_CHECK_INTERVAL
|
||||
waited = 0
|
||||
plan_code = None
|
||||
|
||||
while waited < max_wait:
|
||||
plan_text = plan_code_locator.inner_text()
|
||||
match = re.search(r"离散备料计划维护：(.+)", plan_text)
|
||||
if match and match.group(1).strip():
|
||||
plan_code = match.group(1).strip()
|
||||
break
|
||||
time.sleep(wait_interval)
|
||||
waited += wait_interval
|
||||
|
||||
if plan_code:
|
||||
self._print(f"备料计划页面加载完成，编码: {plan_code}")
|
||||
else:
|
||||
self._print(f"警告: 备料计划页面加载超时")
|
||||
Reference in New Issue
Block a user