Compare commits

...

6 Commits

Author SHA1 Message Date
Misaka_Company
53a1e33e45 feat: add SQL Server database persistence for extracted material plan data
Add optional database persistence feature that automatically saves extracted
discrete material plan data to SQL Server. Users can enable this feature in
the settings tab.

Changes:
- Add enable_db_persistence flag to ExtractionConfig (default: disabled)
- Create DiscreteMaterialPlanDAO for database operations with REPLACE pattern
- Update progress tracking to include database persistence stage (90-100%)
- Add database persistence checkbox in settings UI
- Remove verbose logging checkbox from data extraction UI (config-only now)
- Update extraction workflow to save merged DataFrame to database

Progress weights adjusted:
- download: 65% -> 60%
- database: 10% (new stage)
- Other stages adjusted accordingly

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-06 12:21:58 +08:00
Misaka_Company
3180ccacb8 feat: enhance progress reporting granularity in data extraction workflow
- Break down query phase into 3 detailed steps (read file, query database, complete)
- Split download batch process into 7 granular steps per batch (clear, fill, search, wait, select, configure, download)
- Divide conversion phase into 2 steps per file plus 3 final steps (merge, cleanup)
- Add 'action' parameter to all progress reports for better categorization
- Improve user feedback with more detailed status messages and completion indicators

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-05 15:17:18 +08:00
Misaka_Company
2ca53f6a55 style: normalize code formatting across the codebase
- Standardize quote style (single to double quotes)
- Improve code formatting consistency
- Apply formatting to utilities, GUI components, and tools
- Update imports and docstrings for consistency

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-05 14:56:35 +08:00
Misaka_Company
c34a300f0b docs: add progress callback mechanism documentation
Add comprehensive documentation for _report_progress mechanism:
- System architecture with mermaid diagrams
- Complete sequence diagram showing data flow
- Progress calculation logic with stage weights
- Code examples and usage patterns
- Design considerations (thread safety, decoupling, fault tolerance)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-05 14:38:08 +08:00
Misaka_Company
13bc4520bf refactor: centralize configuration management with type-safe schema
Major changes:
- Add dataclass-based configuration schema with validation (config/schema.py)
- Create centralized config loader and default values (config/defaults.py, config/loader.py)
- Remove duplicate database_config.py, merge into unified structure
- Consolidate browser settings into ERP config
- Add batch_size parameter support to extractor

Bug fixes:
- Fix settings save error by updating config paths (browser.* → erp.*)
- Fix batch_size not being applied in data extraction

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-05 14:27:28 +08:00
Misaka_Company
885d87dff8 feat: add progress callback mechanism for real-time updates in data extraction
- Add ProgressInfo and ProgressCalculator classes for structured progress tracking
- Implement RealtimeOutput for immediate log display without buffering
- Add progress callback support to DiscreteMaterialPlanExtractor
- Use queue-based thread-safe communication for GUI progress updates
- Fix log output issue where all content appeared at once after completion

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-05 14:06:30 +08:00
31 changed files with 2304 additions and 572 deletions

View File

@@ -1,9 +0,0 @@
# ================= SQL Server 配置 =================
SQL_SERVER_CONFIG = {
'driver': 'ODBC Driver 18 for SQL Server',
'server': '192.168.110.114',
'database': 'CompanyDB',
'username': 'peng',
'password': 'Cqbld123456.',
'TrustServerCertificate': 'yes'
}

52
config/defaults.py Normal file
View File

@@ -0,0 +1,52 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
默认配置值
定义所有配置项的默认值。
"""
from config.schema import (
ERPConfig,
DatabaseConfig,
PathConfig,
ExtractionConfig,
AppConfig,
)
# 默认配置
DEFAULT_APP_CONFIG = AppConfig(
erp=ERPConfig(
url="https://68.11.34.30:8082/",
username="BLDpengqiangqiang",
password="Cqbld123456.",
headless=True,
ignore_https_errors=True,
auto_close_browser=True,
),
database=DatabaseConfig(
server="192.168.110.114",
database="CompanyDB",
username="peng",
password="Cqbld123456.",
driver="ODBC Driver 18 for SQL Server",
trust_server_certificate="yes",
),
paths=PathConfig(
data_dir="D:/python/playwrite/data/",
production_id_file="ProductionID.txt",
default_output="离散备料计划维护_合并.xlsx",
validation_output="物料状态校验结果.xlsx",
),
extraction=ExtractionConfig(
batch_size=100,
verbose=True,
auto_convert=True,
merge_batches=True,
enable_db_persistence=False, # Disabled by default
),
)
# 兼容旧版本的字典格式
DEFAULT_SETTINGS_DICT = DEFAULT_APP_CONFIG.to_dict()

150
config/loader.py Normal file
View File

@@ -0,0 +1,150 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
配置加载器
负责加载、合并和验证配置。
"""
import json
import os
from typing import Any, Dict
from config.schema import AppConfig
from config.defaults import DEFAULT_APP_CONFIG, DEFAULT_SETTINGS_DICT
class ConfigLoader:
"""配置加载器"""
@staticmethod
def load(config_file: str = "config/user_settings.json") -> AppConfig:
"""
加载配置文件
Args:
config_file: 配置文件路径
Returns:
应用配置对象
"""
if os.path.exists(config_file):
try:
with open(config_file, "r", encoding="utf-8") as f:
loaded_settings = json.load(f)
# 合并默认配置和加载的配置
merged_settings = ConfigLoader._merge_settings(
DEFAULT_SETTINGS_DICT, loaded_settings
)
return ConfigLoader._dict_to_config(merged_settings)
except (json.JSONDecodeError, IOError) as e:
print(f"加载配置文件失败: {e},使用默认配置")
return DEFAULT_APP_CONFIG
else:
# 首次运行,创建默认配置文件
ConfigLoader.save(DEFAULT_APP_CONFIG, config_file)
return DEFAULT_APP_CONFIG
@staticmethod
def save(config: AppConfig, config_file: str = "config/user_settings.json") -> bool:
"""
保存配置到文件
Args:
config: 应用配置对象
config_file: 配置文件路径
Returns:
保存是否成功
"""
try:
# 确保配置目录存在
os.makedirs(os.path.dirname(config_file), exist_ok=True)
with open(config_file, "w", encoding="utf-8") as f:
json.dump(config.to_dict(), f, ensure_ascii=False, indent=2)
return True
except IOError as e:
print(f"保存配置文件失败: {e}")
return False
@staticmethod
def _merge_settings(defaults: Dict, loaded: Dict) -> Dict:
"""
合并默认配置和加载的配置
Args:
defaults: 默认配置
loaded: 加载的配置
Returns:
合并后的配置
"""
result = defaults.copy()
for key, value in loaded.items():
if (
key in result
and isinstance(result[key], dict)
and isinstance(value, dict)
):
result[key] = ConfigLoader._merge_settings(result[key], value)
else:
result[key] = value
return result
@staticmethod
def _dict_to_config(settings: Dict) -> AppConfig:
"""
将字典转换为配置对象
Args:
settings: 配置字典
Returns:
应用配置对象
"""
erp_dict = settings.get("erp", {})
database_dict = settings.get("database", {})
paths_dict = settings.get("paths", {})
extraction_dict = settings.get("extraction", {})
return AppConfig(
erp=ERPConfig(
url=erp_dict.get("url", ""),
username=erp_dict.get("username", ""),
password=erp_dict.get("password", ""),
headless=erp_dict.get("headless", True),
ignore_https_errors=erp_dict.get("ignore_https_errors", True),
auto_close_browser=erp_dict.get("auto_close_browser", True),
),
database=DatabaseConfig(
server=database_dict.get("server", ""),
database=database_dict.get("database", ""),
username=database_dict.get("username", ""),
password=database_dict.get("password", ""),
driver=database_dict.get("driver", "ODBC Driver 18 for SQL Server"),
trust_server_certificate=database_dict.get(
"trust_server_certificate", "yes"
),
),
paths=PathConfig(
data_dir=paths_dict.get("data_dir", ""),
production_id_file=paths_dict.get("production_id_file", ""),
default_output=paths_dict.get(
"default_output", "离散备料计划维护_合并.xlsx"
),
validation_output=paths_dict.get(
"validation_output", "物料状态校验结果.xlsx"
),
),
extraction=ExtractionConfig(
batch_size=extraction_dict.get("batch_size", 100),
verbose=extraction_dict.get("verbose", True),
auto_convert=extraction_dict.get("auto_convert", True),
merge_batches=extraction_dict.get("merge_batches", True),
),
)
# 为了兼容旧代码,导入必要的类型
from config.schema import ERPConfig, DatabaseConfig, PathConfig, ExtractionConfig

150
config/schema.py Normal file
View File

@@ -0,0 +1,150 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
配置结构定义
使用 dataclass 定义所有配置项的结构和类型。
"""
from dataclasses import dataclass, field
from typing import Optional
from pathlib import Path
@dataclass
class ERPConfig:
"""ERP 系统配置"""
url: str
username: str
password: str
headless: bool = True
ignore_https_errors: bool = True
auto_close_browser: bool = True
def validate(self) -> list[str]:
"""验证配置,返回错误列表"""
errors = []
if not self.url:
errors.append("ERP URL 不能为空")
if not self.username:
errors.append("ERP 用户名不能为空")
if not self.password:
errors.append("ERP 密码不能为空")
return errors
@dataclass
class DatabaseConfig:
"""数据库配置"""
server: str
database: str
username: str
password: str
driver: str = "ODBC Driver 18 for SQL Server"
trust_server_certificate: str = "yes"
def validate(self) -> list[str]:
"""验证配置,返回错误列表"""
errors = []
if not self.server:
errors.append("数据库服务器地址不能为空")
if not self.database:
errors.append("数据库名称不能为空")
if not self.username:
errors.append("数据库用户名不能为空")
if not self.password:
errors.append("数据库密码不能为空")
return errors
@dataclass
class PathConfig:
"""文件路径配置"""
data_dir: str
production_id_file: str
default_output: str = "离散备料计划维护_合并.xlsx"
validation_output: str = "物料状态校验结果.xlsx"
def validate(self) -> list[str]:
"""验证配置,返回错误列表"""
errors = []
if not self.data_dir:
errors.append("数据目录路径不能为空")
if not self.production_id_file:
errors.append("ProductionID 文件路径不能为空")
return errors
@dataclass
class ExtractionConfig:
"""数据提取配置"""
batch_size: int = 100
verbose: bool = True
auto_convert: bool = True
merge_batches: bool = True
enable_db_persistence: bool = False
def validate(self) -> list[str]:
"""验证配置,返回错误列表"""
errors = []
if self.batch_size <= 0:
errors.append("批次大小必须大于 0")
if self.batch_size > 1000:
errors.append("批次大小不应超过 1000")
return errors
@dataclass
class AppConfig:
"""应用总配置"""
erp: ERPConfig
database: DatabaseConfig
paths: PathConfig
extraction: ExtractionConfig
def validate(self) -> list[str]:
"""验证所有配置,返回错误列表"""
errors = []
errors.extend(self.erp.validate())
errors.extend(self.database.validate())
errors.extend(self.paths.validate())
errors.extend(self.extraction.validate())
return errors
def to_dict(self) -> dict:
"""转换为字典格式(用于保存到 JSON"""
return {
"erp": {
"url": self.erp.url,
"username": self.erp.username,
"password": self.erp.password,
"headless": self.erp.headless,
"ignore_https_errors": self.erp.ignore_https_errors,
"auto_close_browser": self.erp.auto_close_browser,
},
"database": {
"server": self.database.server,
"database": self.database.database,
"username": self.database.username,
"password": self.database.password,
"driver": self.database.driver,
"trust_server_certificate": self.database.trust_server_certificate,
},
"paths": {
"data_dir": self.paths.data_dir,
"production_id_file": self.paths.production_id_file,
"default_output": self.paths.default_output,
"validation_output": self.paths.validation_output,
},
"extraction": {
"batch_size": self.extraction.batch_size,
"verbose": self.extraction.verbose,
"auto_convert": self.extraction.auto_convert,
"merge_batches": self.extraction.merge_batches,
"enable_db_persistence": self.extraction.enable_db_persistence,
},
}

View File

@@ -1,40 +1,23 @@
"""
用户配置模板
用户配置模板(已废弃,保留用于向后兼容)
此文件包含用户配置的默认值和结构说明。
实际配置保存在 config/user_settings.json
此文件已迁移到:
- config/schema.py - 配置结构定义
- config/defaults.py - 默认配置值
请使用以下方式导入:
```python
from config.defaults import DEFAULT_APP_CONFIG
from config.schema import AppConfig, ERPConfig, DatabaseConfig
```
"""
DEFAULT_SETTINGS = {
"erp": {
"url": "https://68.11.34.30:8082/",
"username": "BLDpengqiangqiang",
"password": "Cqbld123456.",
"headless": True,
"ignore_https_errors": True,
"auto_close_browser": True
},
"database": {
"server": "192.168.110.114",
"database": "CompanyDB",
"username": "peng",
"password": "Cqbld123456."
},
"browser": {
"headless": True,
"ignore_https_errors": True,
"auto_close": True
},
"paths": {
"data_dir": "D:/python/playwrite/data/",
"production_id_file": "ProductionID.txt",
"default_output": "离散备料计划维护_合并.xlsx",
"validation_output": "物料状态校验结果.xlsx"
},
"extraction": {
"batch_size": 100,
"verbose": True,
"auto_convert": True,
"merge_batches": True
}
}
# 为了向后兼容,保留旧版本导入
from config.defaults import DEFAULT_SETTINGS_DICT as DEFAULT_SETTINGS
from config.schema import (
AppConfig,
ERPConfig,
DatabaseConfig,
PathConfig,
ExtractionConfig,
)

View File

@@ -3,6 +3,7 @@ SQL Server 数据库连接组件
提供数据库连接和查询接口
"""
import pyodbc
from typing import List, Dict, Any, Optional
import sys
@@ -13,7 +14,17 @@ project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if project_root not in sys.path:
sys.path.insert(0, project_root)
from config.database_config import SQL_SERVER_CONFIG
from config.defaults import DEFAULT_APP_CONFIG
# 从默认配置获取数据库配置
SQL_SERVER_CONFIG = {
"driver": DEFAULT_APP_CONFIG.database.driver,
"server": DEFAULT_APP_CONFIG.database.server,
"database": DEFAULT_APP_CONFIG.database.database,
"username": DEFAULT_APP_CONFIG.database.username,
"password": DEFAULT_APP_CONFIG.database.password,
"TrustServerCertificate": DEFAULT_APP_CONFIG.database.trust_server_certificate,
}
class DatabaseConnection:
@@ -51,7 +62,9 @@ class DatabaseConnection:
try:
self.connection = pyodbc.connect(conn_str)
print(f"成功连接到数据库: {self.config['server']}/{self.config['database']}")
print(
f"成功连接到数据库: {self.config['server']}/{self.config['database']}"
)
return self.connection
except pyodbc.Error as e:
print(f"数据库连接失败: {e}")
@@ -64,7 +77,9 @@ class DatabaseConnection:
self.connection = None
print("数据库连接已关闭")
def execute_query(self, sql: str, params: Optional[tuple] = None) -> List[Dict[str, Any]]:
def execute_query(
self, sql: str, params: Optional[tuple] = None
) -> List[Dict[str, Any]]:
"""
执行查询语句并返回结果
@@ -158,7 +173,7 @@ def query_production_orders(总排号_list: List[str]) -> List[Dict[str, Any]]:
db = DatabaseConnection()
# 构建占位符字符串
placeholders = ','.join(['?' for _ in 总排号_list])
placeholders = ",".join(["?" for _ in 总排号_list])
sql = f"""
SELECT [总排号], [生产订单号], [序号], [订单号], [客户名称], [产品型号]

View File

@@ -0,0 +1,313 @@
"""
Data Access Object for DiscreteMaterialPlanData table.
This module provides CRUD operations for persisting discrete material plan
data to SQL Server database. It handles mapping between Chinese DataFrame
columns (from ExcelConverter) and English database columns.
"""
from db.connection import get_connection
from typing import List, Dict, Any
import pandas as pd
class DiscreteMaterialPlanDAO:
"""Data Access Object for DiscreteMaterialPlanData table"""
def __init__(self):
self.db = None
def __enter__(self):
"""Enter context manager and establish database connection"""
self.db = get_connection()
self.db.connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Exit context manager and close database connection"""
if self.db:
self.db.close()
def close(self):
"""Close database connection"""
if self.db:
self.db.close()
def save_dataframe_with_replace(self, df: pd.DataFrame) -> Dict[str, int]:
"""
Save DataFrame using REPLACE strategy (DELETE + INSERT).
This method implements a replace strategy where existing records
matching the plan numbers in the DataFrame are deleted before
inserting new records.
Args:
df: DataFrame with discrete material plan data (Chinese column names)
Returns:
Dictionary with 'deleted' and 'inserted' counts
Example:
>>> dao = DiscreteMaterialPlanDAO()
>>> with dao:
... stats = dao.save_dataframe_with_replace(df)
... print(f"Deleted: {stats['deleted']}, Inserted: {stats['inserted']}")
"""
if df.empty:
return {'deleted': 0, 'inserted': 0}
# Remove duplicates based on PlanNumber and SequenceNumber
original_count = len(df)
df = df.drop_duplicates(subset=['备料计划单号', '序号'], keep='first')
duplicates_removed = original_count - len(df)
if duplicates_removed > 0:
print(f"[INFO] 检测到 {duplicates_removed} 条重复记录(相同计划单号和序号),已自动去重")
with get_connection() as db:
# Get unique plan numbers
plan_numbers = df['备料计划单号'].unique().tolist()
# Delete existing records
deleted = self._delete_by_plan_numbers(db, plan_numbers)
# Insert new records in batches
inserted = self._batch_insert(db, df)
return {'deleted': deleted, 'inserted': inserted}
def _delete_by_plan_numbers(self, db, plan_numbers: List[str]) -> int:
"""
Delete records by plan numbers.
Args:
db: Database connection object
plan_numbers: List of plan numbers to delete
Returns:
Number of records deleted
"""
if not plan_numbers:
return 0
# SQL Server has a limit on IN clause parameters
# Delete in batches to avoid exceeding the limit
batch_size = 1000 # Safe limit for IN clause
total_deleted = 0
for i in range(0, len(plan_numbers), batch_size):
batch = plan_numbers[i:i + batch_size]
placeholders = ','.join(['?' for _ in batch])
sql = f"DELETE FROM DiscreteMaterialPlanData WHERE PlanNumber IN ({placeholders})"
deleted = db.execute_update(sql, tuple(batch))
total_deleted += deleted
return total_deleted
def _batch_insert(self, db, df: pd.DataFrame, batch_size: int = 72) -> int:
"""
Batch insert records (max 72 per batch due to SQL Server 2100 param limit).
SQL Server has a limit of 2100 parameters per query. With 29 fields,
the maximum batch size is floor(2100 / 29) = 72 records per batch.
Args:
db: Database connection object
df: DataFrame to insert
batch_size: Number of records per batch (default: 72)
Returns:
Total number of records inserted
"""
sql = """
INSERT INTO DiscreteMaterialPlanData (
Factory, MaterialStatus, PlanNumber, SourceNumber, MaterialType,
ProductCode, ProductName, ProductUnit, ProductPlanQuantity,
UseDepartment, Remark, Creator, CreateDate, Approver, ApproveDate,
SequenceNumber, MaterialCode, MaterialName, Specification, Model,
DrawingNumber, MaterialQuality, PlanQuantity, Unit, RequiredDate,
Warehouse, UnitUsage, CumulativeOutputQuantity, BOMVersion
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
"""
total_inserted = 0
records = self._convert_df_to_records(df)
for i in range(0, len(records), batch_size):
batch = records[i:i + batch_size]
for record in batch:
db.execute_update(sql, record)
total_inserted += 1
return total_inserted
def _convert_df_to_records(self, df: pd.DataFrame) -> List[tuple]:
"""
Convert DataFrame to list of tuples for batch insert.
Maps Chinese DataFrame column names to English database column names
and converts each row to a tuple in the correct order.
Handles NaN/None values by converting them to None for NULL fields.
Args:
df: DataFrame with Chinese column names
Returns:
List of tuples, one per record
"""
# Column order must match INSERT statement
column_order = [
'工厂', '备料状态', '备料计划单号', '来源单号', '备料类型', '产品编码',
'产品名称', '产品单位', '产品计划数量', '用料部门', '备注', '制单人',
'制单日期', '审批人', '审批日期', '序号', '材料编码', '材料名称',
'规格', '型号', '图号', '物料材质', '计划数量', '单位', '需用日期',
'发料仓库', '单位用量', '累计出库数量', 'BOM版本'
]
# Numeric columns with their default values and data types
numeric_columns = {
'产品计划数量': (0, int),
'序号': (0, int),
'计划数量': (0, int),
'单位用量': (0.0, float),
'累计出库数量': (0, int),
}
records = []
for _, row in df.iterrows():
record = []
for col in column_order:
value = row.get(col)
# Handle NaN, None, or empty string values
if pd.isna(value) or value is None or (isinstance(value, str) and value.strip() == ''):
if col in numeric_columns:
# Use default value for numeric columns
record.append(numeric_columns[col][0])
else:
# Use None (NULL) for string columns
record.append(None)
else:
# Convert numeric columns to proper type
if col in numeric_columns:
try:
default_value, target_type = numeric_columns[col]
if target_type == float:
record.append(float(value))
else:
record.append(int(value))
except (ValueError, TypeError):
# If conversion fails, use default value
record.append(numeric_columns[col][0])
else:
# Keep string columns as is
record.append(value)
records.append(tuple(record))
return records
def query_by_plan_number(self, plan_number: str) -> List[Dict]:
"""
Query all records for a specific plan number.
Args:
plan_number: Plan number to query
Returns:
List of dictionaries representing records
"""
with get_connection() as db:
sql = "SELECT * FROM DiscreteMaterialPlanData WHERE PlanNumber = ?"
return db.execute_query(sql, (plan_number,))
def query_by_plan_numbers(self, plan_numbers: List[str]) -> List[Dict]:
"""
Query records for multiple plan numbers.
Args:
plan_numbers: List of plan numbers to query
Returns:
List of dictionaries representing records
"""
if not plan_numbers:
return []
placeholders = ','.join(['?' for _ in plan_numbers])
sql = f"SELECT * FROM DiscreteMaterialPlanData WHERE PlanNumber IN ({placeholders})"
with get_connection() as db:
return db.execute_query(sql, tuple(plan_numbers))
def query_by_production_order(self, order_id: str) -> List[Dict]:
"""
Query all records for a specific production order.
Args:
order_id: Production order ID (SourceNumber)
Returns:
List of dictionaries representing records
"""
with get_connection() as db:
sql = "SELECT * FROM DiscreteMaterialPlanData WHERE SourceNumber = ?"
return db.execute_query(sql, (order_id,))
def count_by_plan_number(self, plan_number: str) -> int:
"""
Count records for a specific plan number.
Args:
plan_number: Plan number to count
Returns:
Number of records
"""
with get_connection() as db:
sql = "SELECT COUNT(*) as count FROM DiscreteMaterialPlanData WHERE PlanNumber = ?"
result = db.execute_query(sql, (plan_number,))
return result[0]['count'] if result else 0
def count_all(self) -> int:
"""
Count all records in the table.
Returns:
Total number of records
"""
with get_connection() as db:
sql = "SELECT COUNT(*) as count FROM DiscreteMaterialPlanData"
result = db.execute_query(sql)
return result[0]['count'] if result else 0
def delete_by_plan_numbers(self, plan_numbers: List[str]) -> int:
"""
Delete all records for specified plan numbers.
Args:
plan_numbers: List of plan numbers to delete
Returns:
Number of records deleted
"""
with get_connection() as db:
return self._delete_by_plan_numbers(db, plan_numbers)
def get_statistics(self) -> Dict[str, Any]:
"""
Get comprehensive statistics about the data.
Returns:
Dictionary with statistics including total records,
unique plans, unique orders, and date range
"""
with get_connection() as db:
sql = """
SELECT
COUNT(*) as total_records,
COUNT(DISTINCT PlanNumber) as unique_plans,
COUNT(DISTINCT SourceNumber) as unique_orders,
MIN(CreateDate) as earliest_record,
MAX(CreateDate) as latest_record
FROM DiscreteMaterialPlanData
"""
result = db.execute_query(sql)
return result[0] if result else {}

View File

@@ -2,6 +2,7 @@
待删除物料查询组件
从数据库查询指定负责人需要删除的物料名称
"""
from typing import List, Dict, Any
from db.connection import get_connection
@@ -25,7 +26,7 @@ def get_materials_to_delete(manager_name):
with get_connection() as conn:
results = conn.execute_query(query, (manager_name,))
# 提取物料名称并去除空值
material_names = [row['MaterialName'] for row in results if row['MaterialName']]
material_names = [row["MaterialName"] for row in results if row["MaterialName"]]
return material_names

View File

@@ -2,6 +2,7 @@
生产订单号查询组件
从 ProductionID.txt 读取总排号,查询数据库获取生产订单号
"""
from db.connection import get_connection
@@ -15,7 +16,7 @@ def read_production_ids(file_path):
Returns:
总排号列表
"""
with open(file_path, 'r', encoding='utf-8') as f:
with open(file_path, "r", encoding="utf-8") as f:
# 去除空白行和空格
production_ids = [line.strip() for line in f if line.strip()]
return production_ids
@@ -40,8 +41,8 @@ def query_production_order_numbers(production_ids):
# 分批查询
for i in range(0, len(production_ids), BATCH_SIZE):
batch = production_ids[i:i + BATCH_SIZE]
placeholders = ','.join(['?' for _ in batch])
batch = production_ids[i : i + BATCH_SIZE]
placeholders = ",".join(["?" for _ in batch])
query = f"""
SELECT [生产订单号]
@@ -52,7 +53,7 @@ def query_production_order_numbers(production_ids):
with get_connection() as conn:
results = conn.execute_query(query, tuple(batch))
# 提取生产订单号并去除空值
batch_numbers = [row['生产订单号'] for row in results if row['生产订单号']]
batch_numbers = [row["生产订单号"] for row in results if row["生产订单号"]]
all_results.extend(batch_numbers)
return all_results

View File

@@ -0,0 +1,401 @@
# 进度回调机制详解
## 概述
`_report_progress` 是一个基于**回调函数模式**的进度报告系统,用于后台任务(数据提取)和 GUI 主线程之间的线程安全通信。
---
## 架构设计
### 系统架构图
```mermaid
flowchart TB
subgraph BG["后台线程 (Background Thread)"]
Extractor["DiscreteMaterialPlanExtractor"]
Report["_report_progress()"]
ProgressInfo["ProgressInfo 对象"]
end
subgraph Boundary["线程边界 (Thread Boundary)"]
Callback["progress_callback()"]
end
subgraph FG["主线程 (Main/GUI Thread)"]
Calc["ProgressCalculator<br/>计算总体百分比"]
Queue["queue.Queue<br/>线程安全队列"]
Poll["_poll_progress_queue()<br/>每50ms轮询"]
GUI["GUI 组件<br/>progress_bar<br/>status_label"]
end
Extractor -->|"调用"| Report
Report -->|"创建"| ProgressInfo
ProgressInfo -->|"触发"| Callback
Callback -->|"计算"| Calc
Calc -->|"put"| Queue
Queue -->|"get"| Poll
Poll -->|"更新"| GUI
style Callback fill:#ff9,stroke:#333,stroke-width:2px
style Queue fill:#9f9,stroke:#333,stroke-width:2px
style Boundary fill:#ddd,stroke:#333,stroke-dasharray: 5 5
```
### 组件职责
| 组件 | 职责 | 位置 |
|------|------|------|
| `DiscreteMaterialPlanExtractor` | 执行数据提取任务 | 后台线程 |
| `_report_progress()` | 报告进度到回调 | 后台线程 |
| `ProgressInfo` | 进度信息数据结构 | 跨线程 |
| `progress_callback()` | GUI 提供的回调函数 | 主线程定义,后台调用 |
| `ProgressCalculator` | 计算总体进度百分比 | 主线程 |
| `queue.Queue` | 线程安全的消息队列 | 主线程 |
| `_poll_progress_queue()` | 轮询队列并更新 GUI | 主线程 |
---
## 数据流程
### 完整时序图
```mermaid
sequenceDiagram
participant Bg as 后台线程<br/>(Extractor)
participant Report as _report_progress()
participant Callback as progress_callback()
participant Calc as ProgressCalculator
participant Queue as 进度队列
participant Poll as _poll_progress_queue()
participant GUI as GUI 组件
Bg->>Report: _report_progress('download', 2, 3, '下载中')
Report->>Report: 创建 ProgressInfo 对象
Report->>Callback: progress_callback(progress_info)
Note over Callback: 主线程定义的函数<br/>在后台线程中执行
Callback->>Calc: calculate_overall_percent(progress_info)
Note over Calc: download 阶段<br/>stage_offset=10%<br/>current=2, total=3<br/>weight=65%<br/>result = 10 + 67%×65 = 54%
Calc-->>Callback: 返回 54
Callback->>Queue: put((54, '下载中'))
Note over Queue: 线程安全队列<br/>缓冲区
loop 每 50ms
Poll->>Queue: get_nowait()
Queue-->>Poll: (54, '下载中')
Poll->>GUI: progress_bar['value'] = 54
Poll->>GUI: status_label['text'] = '下载中'
Poll->>Poll: after(50ms, 继续轮询)
end
```
### 进度计算逻辑
```mermaid
flowchart LR
A[ProgressInfo<br/>stage=download<br/>current=2<br/>total=3] --> B[ProgressCalculator]
subgraph Calc["计算过程"]
direction TB
B --> C["计算阶段内进度<br/>2/3 × 100 = 67%"]
C --> D["查找阶段权重<br/>download = 65%"]
D --> E["查找阶段偏移<br/>offset = 10%"]
E --> F["总体进度<br/>10 + 67%×65 = 54%"]
end
F --> G["更新进度条<br/>54%"]
```
---
## 阶段权重分配
### 进度阶段划分
```mermaid
pie title 各阶段权重分布
"登录 (5%)" : 5
"查询 (5%)" : 5
"下载 (65%)" : 65
"注销 (5%)" : 5
"转换 (15%)" : 15
"完成 (5%)" : 5
```
### 阶段详情表
| 阶段 | stage | 权重 | 进度范围 | 说明 |
|------|-------|------|----------|------|
| 登录 | `login` | 5% | 0-5% | ERP 系统登录 |
| 查询 | `query` | 5% | 5-10% | 查询数据库获取订单号 |
| 下载 | `download` | 65% | 10-75% | 批量下载数据(主要耗时) |
| 注销 | `logout` | 5% | 75-80% | 退出 ERP 系统 |
| 转换 | `convert` | 15% | 80-95% | 转换 Excel 格式并合并 |
| 完成 | `complete` | 5% | 95-100% | 任务完成 |
---
## 代码实现
### 1. 后台任务:报告进度
```python
# utils/离散备料计划维护数据提取.py
def _report_progress(self, stage: str, current: int, total: int, message: str, **detail):
"""
报告进度
Args:
stage: 阶段标识 ('login', 'query', 'download', 等)
current: 当前进度值 (1, 2, 3...)
total: 总量 (3, 100...)
message: 显示给用户的消息
**detail: 额外信息 (如 batch_index=1)
"""
if self.progress_callback:
try:
from gui.progress import ProgressInfo
progress_info = ProgressInfo(
stage=stage,
current=current,
total=total,
message=message,
detail=detail
)
# 调用 GUI 提供的回调函数
self.progress_callback(progress_info)
except Exception:
# 回调失败不影响主流程
pass
```
### 2. GUI设置回调
```python
# gui/data_extraction_tab.py
def _extraction_worker(self, input_file: str, output_file: str):
"""后台工作线程"""
# 创建进度回调函数
def progress_callback(progress_info: ProgressInfo):
# 1. 计算总体进度百分比
overall_percent = self.progress_calculator.calculate_overall_percent(progress_info)
# 2. 放入队列(线程安全)
self._update_progress(overall_percent, progress_info.message)
# 将回调传递给提取器
self.extractor.extract(
production_id_file=input_file,
output_file=output_file,
progress_callback=progress_callback
)
```
### 3. 线程安全:队列通信
```python
# gui/data_extraction_tab.py
def _update_progress(self, value: int, message: str):
"""后台线程调用,放入队列"""
try:
self.progress_queue.put_nowait((value, message))
except:
pass # 队列满时忽略
def _poll_progress_queue(self):
"""主线程轮询,更新 GUI"""
try:
while True:
# 非阻塞获取队列中的消息
progress_data = self.progress_queue.get_nowait()
value, message = progress_data
# 更新 GUI 组件
self.progress_bar['value'] = value
self.status_label.config(text=message)
except queue.Empty:
pass
finally:
# 继续轮询(每 50ms 检查一次)
self.after(50, self._poll_progress_queue)
```
### 4. 进度计算器
```python
# gui/progress.py
class ProgressCalculator:
# 各阶段在总进度中的占比
STAGE_WEIGHTS = {
'login': 5, # 0-5%
'query': 5, # 5-10%
'download': 65, # 10-75%
'logout': 5, # 75-80%
'convert': 15, # 80-95%
'complete': 5, # 95-100%
}
def calculate_overall_percent(self, progress: ProgressInfo) -> int:
"""计算总体进度百分比"""
stage = progress.stage
if stage == 'complete':
return 100
# 计算阶段起始百分比
stage_offset = self._stage_offsets[stage]
# 计算阶段内的进度百分比
stage_percent = progress.percent # current/total * 100
# 计算该阶段的权重
stage_weight = self.STAGE_WEIGHTS[stage]
# 总进度 = 阶段偏移 + (阶段内进度 × 阶段权重 / 100)
overall = stage_offset + int(stage_percent * stage_weight / 100)
return min(overall, 100)
```
---
## 设计要点
### 1. 线程安全
**问题**Tkinter 不是线程安全的,后台线程不能直接操作 GUI。
```mermaid
flowchart LR
A[后台线程] -->|"❌ 直接调用 GUI"| B[崩溃/未定义行为]
A -->|"✅ 写入队列"| C[queue.Queue]
C -->|"主线程读取"| D[GUI 更新]
```
**解决方案**:使用 `queue.Queue` 作为缓冲区。
```python
# 后台线程:只写入队列
self.progress_queue.put_nowait((value, message))
# 主线程:从队列读取并更新 GUI
progress_data = self.progress_queue.get_nowait()
self.progress_bar['value'] = progress_data[0]
```
### 2. 解耦设计
```mermaid
flowchart TB
A[提取器] -->|"不需要知道 GUI"| B[回调接口]
B -->|"由 GUI 提供"| C[实现]
C -->|"可以替换"| D[测试回调<br/>日志回调<br/>GUI 回调]
```
**好处**
- 提取器代码不依赖 GUI
- 易于测试(可以传入测试回调)
- 灵活扩展(不同场景使用不同回调)
### 3. 容错处理
```python
def _report_progress(self, ...):
if self.progress_callback:
try:
# 调用回调
self.progress_callback(progress_info)
except Exception:
# 回调失败不影响主流程
pass
```
**保证**:进度报告失败不会中断数据提取任务。
### 4. 准确的进度反映
**问题**:不同阶段耗时差异大(登录 3 秒,下载 60 秒)
**解决方案**:为每个阶段分配不同权重。
```mermaid
gantt
title 数据提取各阶段耗时示例
dateFormat X
axisFormat %s
section 任务
登录 :0, 3
查询 :3, 5
下载第1批 :5, 25
下载第2批 :25, 45
下载第3批 :45, 65
注销 :65, 68
转换 :68, 72
```
---
## 使用示例
### 在提取器中报告进度
```python
# 下载批次
for batch_index, order_ids_batch in enumerate(self.group_order_ids(order_ids, self.batch_size)):
# 报告批次开始
self._report_progress(
'download',
batch_index,
total_batches,
f'正在下载第 {batch_index + 1}/{total_batches}',
batch_index=batch_index + 1
)
# 执行下载
downloaded_file = self.download_batch(...)
# 报告批次完成
self._report_progress(
'download',
batch_index + 1,
total_batches,
f'{batch_index + 1} 批下载完成'
)
```
### 在 GUI 中接收进度
```python
from gui.progress import ProgressInfo, ProgressCalculator
class DataExtractionTab(ttk.Frame):
def __init__(self, ...):
self.progress_calculator = ProgressCalculator()
self.progress_queue = queue.Queue()
self._poll_progress_queue()
def progress_callback(self, progress_info: ProgressInfo):
"""后台任务调用的回调函数"""
overall_percent = self.progress_calculator.calculate_overall_percent(progress_info)
self._update_progress(overall_percent, progress_info.message)
```
---
## 总结
`_report_progress` 机制实现了:
1. **线程安全**:通过队列跨线程通信
2. **解耦设计**:提取器与 GUI 分离
3. **准确反映**:权重分配适配实际耗时
4. **容错能力**:回调失败不影响主流程
5. **易于测试**:可注入测试回调
这种模式适用于任何需要长时间运行任务并实时报告进度的场景。

View File

@@ -5,11 +5,15 @@
负责加载、保存和管理用户配置。
"""
import json
import os
from typing import Any, Dict
from config.user_settings import DEFAULT_SETTINGS
from typing import TYPE_CHECKING
from config.loader import ConfigLoader
from config.schema import AppConfig
from config.defaults import DEFAULT_SETTINGS_DICT
# 类型提示时导入,避免循环引用
if TYPE_CHECKING:
from config.schema import ERPConfig, DatabaseConfig, PathConfig, ExtractionConfig
class ConfigManager:
@@ -23,54 +27,25 @@ class ConfigManager:
config_file: 配置文件路径
"""
self.config_file = config_file
self.settings = self.load()
self.config: AppConfig = ConfigLoader.load(config_file)
def load(self) -> Dict[str, Any]:
"""
加载配置文件
# 验证配置
errors = self.config.validate()
if errors:
print("配置验证失败:")
for error in errors:
print(f" - {error}")
Returns:
配置字典,如果文件不存在则返回默认配置
"""
if os.path.exists(self.config_file):
try:
with open(self.config_file, 'r', encoding='utf-8') as f:
loaded_settings = json.load(f)
# 合并默认配置,确保所有必需的键都存在
return self._merge_settings(DEFAULT_SETTINGS, loaded_settings)
except (json.JSONDecodeError, IOError) as e:
print(f"加载配置文件失败: {e}")
return DEFAULT_SETTINGS.copy()
else:
# 首次运行,创建默认配置文件
self.save(DEFAULT_SETTINGS.copy())
return DEFAULT_SETTINGS.copy()
def save(self, settings: Dict[str, Any] = None) -> bool:
def save(self) -> bool:
"""
保存配置到文件
Args:
settings: 要保存的配置字典,如果为 None 则保存当前配置
Returns:
保存是否成功
"""
if settings is not None:
self.settings = settings
return ConfigLoader.save(self.config, self.config_file)
try:
# 确保配置目录存在
os.makedirs(os.path.dirname(self.config_file), exist_ok=True)
with open(self.config_file, 'w', encoding='utf-8') as f:
json.dump(self.settings, f, ensure_ascii=False, indent=2)
return True
except IOError as e:
print(f"保存配置文件失败: {e}")
return False
def get(self, key: str, default=None) -> Any:
def get(self, key: str, default=None):
"""
获取配置项
@@ -83,18 +58,17 @@ class ConfigManager:
Returns:
配置值
"""
keys = key.split('.')
value = self.settings
keys = key.split(".")
value = self.config
for k in keys:
if isinstance(value, dict) and k in value:
value = value[k]
else:
return default
try:
for k in keys:
value = getattr(value, k)
return value
except (AttributeError, TypeError):
return default
return value
def set(self, key: str, value: Any) -> None:
def set(self, key: str, value) -> None:
"""
设置配置项
@@ -104,38 +78,41 @@ class ConfigManager:
key: 配置键
value: 配置值
"""
keys = key.split('.')
settings = self.settings
keys = key.split(".")
obj = self.config
# 导航到父对象
for k in keys[:-1]:
if k not in settings:
settings[k] = {}
settings = settings[k]
obj = getattr(obj, k)
settings[keys[-1]] = value
# 设置最终值
setattr(obj, keys[-1], value)
def reset_to_defaults(self) -> None:
"""重置为默认配置"""
self.settings = DEFAULT_SETTINGS.copy()
self.config = ConfigLoader.load("default") # 重新加载默认配置
self.save()
def _merge_settings(self, defaults: Dict, loaded: Dict) -> Dict:
"""
合并默认配置和加载的配置
@property
def erp(self):
"""获取 ERP 配置"""
return self.config.erp
Args:
defaults: 默认配置
loaded: 加载的配置
@property
def database(self):
"""获取数据库配置"""
return self.config.database
Returns:
合并后的配置
"""
result = defaults.copy()
@property
def paths(self):
"""获取路径配置"""
return self.config.paths
for key, value in loaded.items():
if key in result and isinstance(result[key], dict) and isinstance(value, dict):
result[key] = self._merge_settings(result[key], value)
else:
result[key] = value
@property
def extraction(self):
"""获取提取配置"""
return self.config.extraction
return result
# 为了向后兼容,保留旧版本的导入
DEFAULT_SETTINGS = DEFAULT_SETTINGS_DICT

View File

@@ -9,13 +9,15 @@
import os
import sys
import threading
import queue
import tkinter as tk
from tkinter import ttk, filedialog, messagebox
from pathlib import Path
from io import StringIO
from contextlib import redirect_stdout
from gui.widgets import FileSelector, LogText
from gui.config_manager import ConfigManager
from gui.progress import ProgressInfo, ProgressCalculator
from gui.utils import RealtimeOutput
class DataExtractionTab(ttk.Frame):
@@ -34,6 +36,11 @@ class DataExtractionTab(ttk.Frame):
self.extracting = False
self.extractor = None
self.extraction_thread = None
self.progress_calculator = ProgressCalculator()
self.progress_queue = queue.Queue() # 进度更新队列
# 启动进度更新轮询
self._poll_progress_queue()
self.create_widgets()
@@ -71,12 +78,12 @@ class DataExtractionTab(ttk.Frame):
label_text="ProductionID 文件:",
file_type="file",
file_types=[("文本文件", "*.txt"), ("所有文件", "*.*")],
initial_dir="D:/python/playwrite/"
initial_dir="D:/python/playwrite/",
)
self.input_file_selector.pack(fill=tk.X)
# 设置默认文件
default_input = self.config.get('paths.production_id_file', 'ProductionID.txt')
default_input = self.config.get("paths.production_id_file", "ProductionID.txt")
if os.path.exists(default_input):
self.input_file_selector.set(default_input)
@@ -89,14 +96,14 @@ class DataExtractionTab(ttk.Frame):
label_text="保存为:",
file_type="file",
file_types=[("Excel 文件", "*.xlsx"), ("所有文件", "*.*")],
initial_dir=self.config.get('paths.data_dir', 'data/')
initial_dir=self.config.get("paths.data_dir", "data/"),
)
self.output_file_selector.pack(fill=tk.X)
# 设置默认输出文件
default_output = os.path.join(
self.config.get('paths.data_dir', 'data/'),
self.config.get('paths.default_output', '离散备料计划维护_合并.xlsx')
self.config.get("paths.data_dir", "data/"),
self.config.get("paths.default_output", "离散备料计划维护_合并.xlsx"),
)
self.output_file_selector.set(default_output)
@@ -104,30 +111,35 @@ class DataExtractionTab(ttk.Frame):
options_group = ttk.LabelFrame(parent, text="提取选项", padding=10)
options_group.pack(fill=tk.X, pady=5)
self.verbose_var = tk.BooleanVar(value=self.config.get('extraction.verbose', True))
ttk.Checkbutton(options_group, text="详细日志", variable=self.verbose_var).grid(row=0, column=0, sticky="w", padx=5)
self.headless_var = tk.BooleanVar(value=self.config.get('browser.headless', True))
ttk.Checkbutton(options_group, text="无头模式 (不显示浏览器)", variable=self.headless_var).grid(row=0, column=1, sticky="w", padx=5)
self.headless_var = tk.BooleanVar(value=self.config.get("erp.headless", True))
ttk.Checkbutton(
options_group, text="无头模式 (不显示浏览器)", variable=self.headless_var
).grid(row=0, column=0, sticky="w", padx=5)
# 进度显示
progress_group = ttk.LabelFrame(parent, text="进度", padding=10)
progress_group.pack(fill=tk.X, pady=5)
self.progress_bar = ttk.Progressbar(progress_group, mode='determinate')
self.progress_bar = ttk.Progressbar(progress_group, mode="determinate")
self.progress_bar.pack(fill=tk.X, pady=5)
self.status_label = ttk.Label(progress_group, text="就绪", relief=tk.SUNKEN, anchor=tk.W)
self.status_label = ttk.Label(
progress_group, text="就绪", relief=tk.SUNKEN, anchor=tk.W
)
self.status_label.pack(fill=tk.X)
# 控制按钮
button_frame = ttk.Frame(parent)
button_frame.pack(fill=tk.X, pady=10)
self.start_button = ttk.Button(button_frame, text="开始提取", command=self.start_extraction)
self.start_button = ttk.Button(
button_frame, text="开始提取", command=self.start_extraction
)
self.start_button.pack(side=tk.LEFT, padx=5)
self.stop_button = ttk.Button(button_frame, text="停止", command=self.stop_extraction, state=tk.DISABLED)
self.stop_button = ttk.Button(
button_frame, text="停止", command=self.stop_extraction, state=tk.DISABLED
)
self.stop_button.pack(side=tk.LEFT, padx=5)
def _create_log_panel(self, parent):
@@ -162,15 +174,14 @@ class DataExtractionTab(ttk.Frame):
self.extracting = True
self.start_button.config(state=tk.DISABLED)
self.stop_button.config(state=tk.NORMAL)
self.progress_bar['value'] = 0
self.progress_bar["value"] = 0
self.status_label.config(text="正在登录...")
self.log_text.clear()
self.log_text.info("开始数据提取...")
# 在后台线程中执行提取
self.extraction_thread = threading.Thread(
target=self._extraction_worker,
args=(input_file, output_file),
daemon=True
target=self._extraction_worker, args=(input_file, output_file), daemon=True
)
self.extraction_thread.start()
@@ -189,31 +200,36 @@ class DataExtractionTab(ttk.Frame):
# 创建提取器实例
self.extractor = DiscreteMaterialPlanExtractor(
username=self.config.get('erp.username'),
password=self.config.get('erp.password'),
username=self.config.get("erp.username"),
password=self.config.get("erp.password"),
headless=self.headless_var.get(),
verbose=self.verbose_var.get()
verbose=self.config.get("extraction.verbose", True),
batch_size=self.config.get("extraction.batch_size", 100),
enable_db_persistence=self.config.get("extraction.enable_db_persistence", False),
)
# 捕获 stdout 输出
captured_output = StringIO()
# 创建实时输出流,每次写入立即更新 GUI
realtime_output = RealtimeOutput(
lambda line: self._update_log(line, "INFO")
)
# 重定向 stdout 并执行提取
with redirect_stdout(captured_output):
# 创建进度回调函数
def progress_callback(progress_info: ProgressInfo):
# 计算总体进度百分比
overall_percent = self.progress_calculator.calculate_overall_percent(
progress_info
)
self._update_progress(overall_percent, progress_info.message)
# 重定向 stdout 并执行提取(带进度回调)
with redirect_stdout(realtime_output):
result = self.extractor.extract(
production_id_file=input_file,
output_file=output_file
output_file=output_file,
progress_callback=progress_callback,
)
# 获取捕获的输出并显示到日志
output_text = captured_output.getvalue()
if output_text:
for line in output_text.split('\n'):
if line.strip():
self._update_log(line, "INFO")
if result and self.extracting:
self._update_progress(100, "提取完成")
self._update_log(f"数据已保存到:{output_file}", "SUCCESS")
elif not self.extracting:
self._update_log("提取已取消", "WARNING")
@@ -233,17 +249,32 @@ class DataExtractionTab(ttk.Frame):
self.stop_button.config(state=tk.DISABLED)
self.extractor = None
def _update_progress(self, value: int, message: str):
"""线程安全的进度更新"""
def update():
if self.extracting:
self.progress_bar['value'] = value
self.status_label.config(text=message)
def _poll_progress_queue(self):
"""轮询进度队列,处理进度更新"""
try:
while True:
# 非阻塞地获取队列中的消息
try:
progress_data = self.progress_queue.get_nowait()
value, message = progress_data
self.progress_bar["value"] = value
self.status_label.config(text=message)
except queue.Empty:
break
finally:
# 继续轮询(每 50ms 检查一次)
self.after(50, self._poll_progress_queue)
self.after(0, update)
def _update_progress(self, value: int, message: str):
"""线程安全的进度更新(通过队列)"""
try:
self.progress_queue.put_nowait((value, message))
except:
pass # 队列满时忽略
def _update_log(self, message: str, level: str = "INFO"):
"""线程安全的日志更新"""
def update():
if self.extracting or level in ["ERROR", "WARNING", "SUCCESS"]:
if level == "INFO":

View File

@@ -67,7 +67,7 @@ class DataQueryTab(ttk.Frame):
info_label = ttk.Label(
parent,
text="输入总排号列表(每行一个),查询对应的生产订单号信息",
foreground="#666666"
foreground="#666666",
)
info_label.pack(anchor=tk.W, pady=(0, 5))
@@ -82,7 +82,9 @@ class DataQueryTab(ttk.Frame):
self.input_text = tk.Text(text_frame, height=10, wrap=tk.WORD)
self.input_text.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
text_scrollbar = ttk.Scrollbar(text_frame, orient=tk.VERTICAL, command=self.input_text.yview)
text_scrollbar = ttk.Scrollbar(
text_frame, orient=tk.VERTICAL, command=self.input_text.yview
)
text_scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
self.input_text.configure(yscrollcommand=text_scrollbar.set)
@@ -102,20 +104,31 @@ class DataQueryTab(ttk.Frame):
# 快捷按钮
ttk.Separator(right_frame, orient=tk.HORIZONTAL).pack(fill=tk.X, pady=10)
ttk.Button(right_frame, text="加载 ProductionID.txt", command=self._load_production_id).pack(fill=tk.X, pady=2)
ttk.Button(right_frame, text="清空输入", command=self._clear_input).pack(fill=tk.X, pady=2)
ttk.Button(
right_frame, text="加载 ProductionID.txt", command=self._load_production_id
).pack(fill=tk.X, pady=2)
ttk.Button(right_frame, text="清空输入", command=self._clear_input).pack(
fill=tk.X, pady=2
)
# 查询按钮
button_frame = ttk.Frame(parent)
button_frame.pack(fill=tk.X, pady=(10, 0))
self.query_button = ttk.Button(button_frame, text="执行查询", command=self.execute_query)
self.query_button = ttk.Button(
button_frame, text="执行查询", command=self.execute_query
)
self.query_button.pack(side=tk.LEFT, padx=5)
self.export_button = ttk.Button(button_frame, text="导出结果", command=self.export_results, state=tk.DISABLED)
self.export_button = ttk.Button(
button_frame,
text="导出结果",
command=self.export_results,
state=tk.DISABLED,
)
self.export_button.pack(side=tk.LEFT, padx=5)
self.progress = ttk.Progressbar(parent, mode='indeterminate')
self.progress = ttk.Progressbar(parent, mode="indeterminate")
self.progress.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=5)
def _create_result_table(self, parent):
@@ -135,9 +148,13 @@ class DataQueryTab(ttk.Frame):
# 添加滚动条
scrollbar_y = ttk.Scrollbar(parent, orient=tk.VERTICAL, command=self.tree.yview)
scrollbar_x = ttk.Scrollbar(parent, orient=tk.HORIZONTAL, command=self.tree.xview)
scrollbar_x = ttk.Scrollbar(
parent, orient=tk.HORIZONTAL, command=self.tree.xview
)
self.tree.configure(yscrollcommand=scrollbar_y.set, xscrollcommand=scrollbar_x.set)
self.tree.configure(
yscrollcommand=scrollbar_y.set, xscrollcommand=scrollbar_x.set
)
# 布局
self.tree.grid(row=0, column=0, sticky="nsew")
@@ -154,7 +171,7 @@ class DataQueryTab(ttk.Frame):
def _load_production_id(self):
"""加载 ProductionID.txt 文件"""
default_path = self.config.get('paths.production_id_file', 'ProductionID.txt')
default_path = self.config.get("paths.production_id_file", "ProductionID.txt")
# 检查默认路径
if os.path.exists(default_path):
@@ -162,16 +179,17 @@ class DataQueryTab(ttk.Frame):
else:
# 打开文件选择对话框
from tkinter import filedialog
file_path = filedialog.askopenfilename(
title="选择 ProductionID 文件",
filetypes=[("文本文件", "*.txt"), ("所有文件", "*.*")]
filetypes=[("文本文件", "*.txt"), ("所有文件", "*.*")],
)
if not file_path:
return
try:
with open(file_path, 'r', encoding='utf-8') as f:
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
self.input_text.delete("1.0", tk.END)
@@ -196,7 +214,9 @@ class DataQueryTab(ttk.Frame):
return
# 解析总排号
production_ids = [line.strip() for line in input_text.split('\n') if line.strip()]
production_ids = [
line.strip() for line in input_text.split("\n") if line.strip()
]
if not production_ids:
messagebox.showwarning("警告", "没有有效的总排号")
@@ -215,9 +235,7 @@ class DataQueryTab(ttk.Frame):
# 在后台线程中执行查询
query_thread = threading.Thread(
target=self._query_worker,
args=(production_ids,),
daemon=True
target=self._query_worker, args=(production_ids,), daemon=True
)
query_thread.start()
@@ -253,6 +271,7 @@ class DataQueryTab(ttk.Frame):
def _display_results(self, results: list):
"""在主线程中显示结果"""
def update():
for 总排号, 生产订单号 in results:
self.tree.insert("", tk.END, values=(总排号, 生产订单号, ""))
@@ -265,6 +284,7 @@ class DataQueryTab(ttk.Frame):
def _update_log(self, message: str, level: str = "INFO"):
"""线程安全的日志更新"""
def update():
if level == "INFO":
self.log_text.info(message)
@@ -287,7 +307,7 @@ class DataQueryTab(ttk.Frame):
title="导出查询结果",
defaultextension=".xlsx",
filetypes=[("Excel 文件", "*.xlsx"), ("所有文件", "*.*")],
initialfile="生产订单号查询结果.xlsx"
initialfile="生产订单号查询结果.xlsx",
)
if not output_file:

View File

@@ -92,13 +92,17 @@ class MainWindow:
# 状态文本
self.status_text = tk.StringVar()
self.status_text.set("就绪")
status_label = ttk.Label(self.status_bar, textvariable=self.status_text, anchor=tk.W)
status_label = ttk.Label(
self.status_bar, textvariable=self.status_text, anchor=tk.W
)
status_label.pack(side=tk.LEFT, padx=5)
# 配置状态指示
self.config_status = tk.StringVar()
self.config_status.set("配置已加载")
config_label = ttk.Label(self.status_bar, textvariable=self.config_status, anchor=tk.E)
config_label = ttk.Label(
self.status_bar, textvariable=self.config_status, anchor=tk.E
)
config_label.pack(side=tk.RIGHT, padx=5)
def _center_window(self):
@@ -113,6 +117,7 @@ class MainWindow:
def show_about(self):
"""显示关于对话框"""
from tkinter import messagebox
messagebox.showinfo(
"关于 ERP 自动化工具",
"ERP 自动化工具 v1.0\n\n"
@@ -120,6 +125,7 @@ class MainWindow:
"• 数据提取 - 从 ERP 系统提取备料计划数据\n"
"• 物料校验 - 校验物料状态并匹配待删除物料\n"
"• 数据查询 - 查询生产订单号等信息\n"
"• 设置管理 - 管理系统配置\n\n"
"基于 Playwright 和 Python 开发"
"• 设置管理 - 管理系统配置\n"
"• 数据库持久化 - 将提取的数据自动保存到 SQL Server\n\n"
"基于 Playwright 和 Python 开发",
)

View File

@@ -76,7 +76,7 @@ class MaterialValidationTab(ttk.Frame):
text="使用现有 Excel 文件",
variable=self.source_mode,
value="existing",
command=self._on_source_mode_change
command=self._on_source_mode_change,
).grid(row=0, column=0, sticky="w", padx=5)
ttk.Radiobutton(
@@ -84,7 +84,7 @@ class MaterialValidationTab(ttk.Frame):
text="完整工作流 (提取 + 校验)",
variable=self.source_mode,
value="full",
command=self._on_source_mode_change
command=self._on_source_mode_change,
).grid(row=0, column=1, sticky="w", padx=5)
# 文件选择
@@ -100,7 +100,7 @@ class MaterialValidationTab(ttk.Frame):
label_text="现有 Excel 文件:",
file_type="file",
file_types=[("Excel 文件", "*.xlsx"), ("所有文件", "*.*")],
initial_dir=self.config.get('paths.data_dir', 'data/')
initial_dir=self.config.get("paths.data_dir", "data/"),
)
self.existing_excel_selector.pack(fill=tk.X)
@@ -113,7 +113,7 @@ class MaterialValidationTab(ttk.Frame):
label_text="ProductionID 文件:",
file_type="file",
file_types=[("文本文件", "*.txt"), ("所有文件", "*.*")],
initial_dir="D:/python/playwrite/"
initial_dir="D:/python/playwrite/",
)
self.production_id_selector.pack(fill=tk.X)
@@ -126,14 +126,14 @@ class MaterialValidationTab(ttk.Frame):
label_text="输出文件:",
file_type="file",
file_types=[("Excel 文件", "*.xlsx"), ("所有文件", "*.*")],
initial_dir=self.config.get('paths.data_dir', 'data/')
initial_dir=self.config.get("paths.data_dir", "data/"),
)
self.output_file_selector.pack(fill=tk.X)
# 设置默认输出
default_output = os.path.join(
self.config.get('paths.data_dir', 'data/'),
self.config.get('paths.validation_output', '物料状态校验结果.xlsx')
self.config.get("paths.data_dir", "data/"),
self.config.get("paths.validation_output", "物料状态校验结果.xlsx"),
)
self.output_file_selector.set(default_output)
@@ -141,10 +141,17 @@ class MaterialValidationTab(ttk.Frame):
button_frame = ttk.Frame(parent)
button_frame.pack(fill=tk.X, pady=10)
self.start_button = ttk.Button(button_frame, text="开始校验", command=self.start_validation)
self.start_button = ttk.Button(
button_frame, text="开始校验", command=self.start_validation
)
self.start_button.pack(side=tk.LEFT, padx=5)
self.export_button = ttk.Button(button_frame, text="导出结果", command=self.export_results, state=tk.DISABLED)
self.export_button = ttk.Button(
button_frame,
text="导出结果",
command=self.export_results,
state=tk.DISABLED,
)
self.export_button.pack(side=tk.LEFT, padx=5)
def _create_result_table(self, parent):
@@ -166,9 +173,13 @@ class MaterialValidationTab(ttk.Frame):
# 添加滚动条
scrollbar_y = ttk.Scrollbar(parent, orient=tk.VERTICAL, command=self.tree.yview)
scrollbar_x = ttk.Scrollbar(parent, orient=tk.HORIZONTAL, command=self.tree.xview)
scrollbar_x = ttk.Scrollbar(
parent, orient=tk.HORIZONTAL, command=self.tree.xview
)
self.tree.configure(yscrollcommand=scrollbar_y.set, xscrollcommand=scrollbar_x.set)
self.tree.configure(
yscrollcommand=scrollbar_y.set, xscrollcommand=scrollbar_x.set
)
# 布局
self.tree.grid(row=0, column=0, sticky="nsew")
@@ -239,11 +250,13 @@ class MaterialValidationTab(ttk.Frame):
validation_thread = threading.Thread(
target=self._validation_worker,
args=(input_file, production_id_file, output_file),
daemon=True
daemon=True,
)
validation_thread.start()
def _validation_worker(self, input_file: str, production_id_file: str, output_file: str):
def _validation_worker(
self, input_file: str, production_id_file: str, output_file: str
):
"""校验工作线程"""
try:
# 导入校验器
@@ -251,10 +264,10 @@ class MaterialValidationTab(ttk.Frame):
# 创建校验器实例(需要 ERP 凭据,因为可能需要登录系统)
validator = MaterialStatusValidator(
username=self.config.get('erp.username'),
password=self.config.get('erp.password'),
headless=self.config.get('browser.headless', True),
verbose=True
username=self.config.get("erp.username"),
password=self.config.get("erp.password"),
headless=self.config.get("erp.headless", True),
verbose=True,
)
# 捕获 stdout 输出
@@ -264,20 +277,19 @@ class MaterialValidationTab(ttk.Frame):
with redirect_stdout(captured_output):
if self.source_mode.get() == "existing":
result = validator.validate_from_existing_excel(
excel_file=input_file,
output_file=output_file
excel_file=input_file, output_file=output_file
)
else:
result = validator.validate(
production_id_file=production_id_file,
merged_excel_file=None, # 将在内部生成
output_file=output_file
output_file=output_file,
)
# 获取捕获的输出并显示到日志
output_text = captured_output.getvalue()
if output_text:
for line in output_text.split('\n'):
for line in output_text.split("\n"):
if line.strip():
self._update_log(line, "INFO")
@@ -307,12 +319,16 @@ class MaterialValidationTab(ttk.Frame):
# 在主线程中更新表格
def update_table():
for _, row in df.iterrows():
self.tree.insert("", tk.END, values=(
row.get('材料名称', ''),
row.get('匹配的MaterialName', ''),
row.get('负责人', ''),
row.get('匹配状态', '')
))
self.tree.insert(
"",
tk.END,
values=(
row.get("材料名称", ""),
row.get("匹配的MaterialName", ""),
row.get("负责人", ""),
row.get("匹配状态", ""),
),
)
if len(df) > 0:
self.export_button.config(state=tk.NORMAL)
@@ -325,6 +341,7 @@ class MaterialValidationTab(ttk.Frame):
def _update_log(self, message: str, level: str = "INFO"):
"""线程安全的日志更新"""
def update():
if level == "INFO":
self.log_text.info(message)
@@ -345,7 +362,7 @@ class MaterialValidationTab(ttk.Frame):
output_file = filedialog.asksaveasfilename(
title="保存结果",
defaultextension=".xlsx",
filetypes=[("Excel 文件", "*.xlsx"), ("所有文件", "*.*")]
filetypes=[("Excel 文件", "*.xlsx"), ("所有文件", "*.*")],
)
if not output_file:
@@ -355,7 +372,7 @@ class MaterialValidationTab(ttk.Frame):
# 收集表格数据
data = []
for item in self.tree.get_children():
values = self.tree.item(item)['values']
values = self.tree.item(item)["values"]
data.append(values)
if not data:
@@ -363,7 +380,9 @@ class MaterialValidationTab(ttk.Frame):
return
# 创建 DataFrame 并保存
df = pd.DataFrame(data, columns=["材料名称", "匹配的MaterialName", "负责人", "匹配状态"])
df = pd.DataFrame(
data, columns=["材料名称", "匹配的MaterialName", "负责人", "匹配状态"]
)
df.to_excel(output_file, index=False)
messagebox.showinfo("成功", f"结果已导出到:{output_file}")

101
gui/progress.py Normal file
View File

@@ -0,0 +1,101 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
进度信息模块
定义用于进度回调的数据结构。
"""
from dataclasses import dataclass, field
from typing import Dict, Any, Optional
@dataclass
class ProgressInfo:
"""
进度信息
用于在后台任务和 GUI 之间传递进度信息。
"""
stage: (
str # 阶段标识: 'login', 'query', 'download', 'logout', 'convert', 'database', 'complete'
)
current: int # 当前进度值
total: int # 总量
message: str # 显示给用户的消息
detail: Dict[str, Any] = field(default_factory=dict) # 额外详细信息
@property
def percent(self) -> int:
"""计算进度百分比0-100"""
if self.total <= 0:
return 0
return int(self.current * 100 / self.total)
def __repr__(self) -> str:
return f"ProgressInfo(stage={self.stage}, {self.current}/{self.total}, {self.message})"
class ProgressCalculator:
"""
进度计算器
将各阶段的进度映射到总体进度百分比。
"""
# 各阶段在总进度中的占比
STAGE_WEIGHTS = {
"login": 5, # 登录: 0-5%
"query": 5, # 查询: 5-10%
"download": 60, # 下载: 10-70%
"logout": 5, # 注销: 70-75%
"convert": 15, # 转换: 75-90%
"database": 10, # 数据库持久化: 90-100%
"complete": 5, # 完成: 95-100%
}
def __init__(self):
"""初始化进度计算器"""
self._stage_offsets = self._calculate_stage_offsets()
def _calculate_stage_offsets(self) -> Dict[str, int]:
"""计算各阶段的起始偏移量(百分比)"""
offsets = {}
offset = 0
for stage, weight in self.STAGE_WEIGHTS.items():
offsets[stage] = offset
offset += weight
return offsets
def calculate_overall_percent(self, progress: ProgressInfo) -> int:
"""
计算总体进度百分比
Args:
progress: 进度信息
Returns:
总体进度百分比 (0-100)
"""
stage = progress.stage
if stage == "complete":
return 100
if stage not in self._stage_offsets:
return 0
# 计算阶段起始百分比
stage_offset = self._stage_offsets[stage]
# 计算阶段内的进度百分比
stage_percent = progress.percent
# 计算该阶段的权重
stage_weight = self.STAGE_WEIGHTS[stage]
# 总进度 = 阶段偏移 + (阶段内进度 * 阶段权重 / 100)
overall = stage_offset + int(stage_percent * stage_weight / 100)
return min(overall, 100)

View File

@@ -36,8 +36,7 @@ class SettingsTab(ttk.Frame):
scrollable_frame = ttk.Frame(canvas)
scrollable_frame.bind(
"<Configure>",
lambda e: canvas.configure(scrollregion=canvas.bbox("all"))
"<Configure>", lambda e: canvas.configure(scrollregion=canvas.bbox("all"))
)
canvas.create_window((0, 0), window=scrollable_frame, anchor="nw")
@@ -62,10 +61,18 @@ class SettingsTab(ttk.Frame):
button_frame = ttk.Frame(scrollable_frame)
button_frame.grid(row=5, column=0, columnspan=2, pady=20, sticky="ew")
ttk.Button(button_frame, text="测试 ERP 连接", command=self.test_erp_connection).pack(side="left", padx=5)
ttk.Button(button_frame, text="测试数据库连接", command=self.test_db_connection).pack(side="left", padx=5)
ttk.Button(button_frame, text="保存设置", command=self.save_settings).pack(side="left", padx=5)
ttk.Button(button_frame, text="恢复默认", command=self.reset_defaults).pack(side="left", padx=5)
ttk.Button(
button_frame, text="测试 ERP 连接", command=self.test_erp_connection
).pack(side="left", padx=5)
ttk.Button(
button_frame, text="测试数据库连接", command=self.test_db_connection
).pack(side="left", padx=5)
ttk.Button(button_frame, text="保存设置", command=self.save_settings).pack(
side="left", padx=5
)
ttk.Button(button_frame, text="恢复默认", command=self.reset_defaults).pack(
side="left", padx=5
)
# 布局
canvas.grid(row=0, column=0, sticky="nsew")
@@ -82,12 +89,16 @@ class SettingsTab(ttk.Frame):
# URL
ttk.Label(group, text="ERP URL:").grid(row=0, column=0, sticky="w", pady=5)
self.erp_url_var = tk.StringVar()
ttk.Entry(group, textvariable=self.erp_url_var, width=50).grid(row=0, column=1, pady=5, sticky="ew")
ttk.Entry(group, textvariable=self.erp_url_var, width=50).grid(
row=0, column=1, pady=5, sticky="ew"
)
# 用户名
ttk.Label(group, text="用户名:").grid(row=1, column=0, sticky="w", pady=5)
self.erp_username_var = tk.StringVar()
ttk.Entry(group, textvariable=self.erp_username_var, width=50).grid(row=1, column=1, pady=5, sticky="ew")
ttk.Entry(group, textvariable=self.erp_username_var, width=50).grid(
row=1, column=1, pady=5, sticky="ew"
)
# 密码
ttk.Label(group, text="密码:").grid(row=2, column=0, sticky="w", pady=5)
@@ -105,17 +116,23 @@ class SettingsTab(ttk.Frame):
# 服务器
ttk.Label(group, text="服务器:").grid(row=0, column=0, sticky="w", pady=5)
self.db_server_var = tk.StringVar()
ttk.Entry(group, textvariable=self.db_server_var, width=50).grid(row=0, column=1, pady=5, sticky="ew")
ttk.Entry(group, textvariable=self.db_server_var, width=50).grid(
row=0, column=1, pady=5, sticky="ew"
)
# 数据库名
ttk.Label(group, text="数据库:").grid(row=1, column=0, sticky="w", pady=5)
self.db_name_var = tk.StringVar()
ttk.Entry(group, textvariable=self.db_name_var, width=50).grid(row=1, column=1, pady=5, sticky="ew")
ttk.Entry(group, textvariable=self.db_name_var, width=50).grid(
row=1, column=1, pady=5, sticky="ew"
)
# 用户名
ttk.Label(group, text="用户名:").grid(row=2, column=0, sticky="w", pady=5)
self.db_username_var = tk.StringVar()
ttk.Entry(group, textvariable=self.db_username_var, width=50).grid(row=2, column=1, pady=5, sticky="ew")
ttk.Entry(group, textvariable=self.db_username_var, width=50).grid(
row=2, column=1, pady=5, sticky="ew"
)
# 密码
ttk.Label(group, text="密码:").grid(row=3, column=0, sticky="w", pady=5)
@@ -131,13 +148,19 @@ class SettingsTab(ttk.Frame):
group.grid(row=2, column=0, pady=10, padx=10, sticky="ew")
self.browser_headless_var = tk.BooleanVar()
ttk.Checkbutton(group, text="无头模式 (不显示浏览器)", variable=self.browser_headless_var).grid(row=0, column=0, sticky="w", pady=5)
ttk.Checkbutton(
group, text="无头模式 (不显示浏览器)", variable=self.browser_headless_var
).grid(row=0, column=0, sticky="w", pady=5)
self.browser_ignore_https_var = tk.BooleanVar()
ttk.Checkbutton(group, text="忽略 HTTPS 错误", variable=self.browser_ignore_https_var).grid(row=1, column=0, sticky="w", pady=5)
ttk.Checkbutton(
group, text="忽略 HTTPS 错误", variable=self.browser_ignore_https_var
).grid(row=1, column=0, sticky="w", pady=5)
self.browser_auto_close_var = tk.BooleanVar()
ttk.Checkbutton(group, text="操作完成后自动关闭浏览器", variable=self.browser_auto_close_var).grid(row=2, column=0, sticky="w", pady=5)
ttk.Checkbutton(
group, text="操作完成后自动关闭浏览器", variable=self.browser_auto_close_var
).grid(row=2, column=0, sticky="w", pady=5)
def _create_paths_group(self, parent):
"""创建路径配置组"""
@@ -152,14 +175,16 @@ class SettingsTab(ttk.Frame):
group,
label_text="",
file_type="directory",
initial_dir="D:/python/playwrite/data/"
initial_dir="D:/python/playwrite/data/",
)
self.data_dir_selector.grid(row=1, column=0, columnspan=2, sticky="ew", pady=5)
# 默认输出文件
ttk.Label(group, text="默认输出文件:").grid(row=2, column=0, sticky="w", pady=5)
self.default_output_var = tk.StringVar()
ttk.Entry(group, textvariable=self.default_output_var, width=40).grid(row=3, column=0, columnspan=2, sticky="ew", pady=5)
ttk.Entry(group, textvariable=self.default_output_var, width=40).grid(
row=3, column=0, columnspan=2, sticky="ew", pady=5
)
group.columnconfigure(0, weight=1)
@@ -171,75 +196,93 @@ class SettingsTab(ttk.Frame):
# 批次大小
ttk.Label(group, text="批次大小:").grid(row=0, column=0, sticky="w", pady=5)
self.batch_size_var = tk.IntVar(value=100)
ttk.Spinbox(group, from_=10, to=500, textvariable=self.batch_size_var, width=10).grid(row=0, column=1, sticky="w", pady=5)
ttk.Spinbox(
group, from_=10, to=500, textvariable=self.batch_size_var, width=10
).grid(row=0, column=1, sticky="w", pady=5)
# 详细日志
self.verbose_var = tk.BooleanVar()
ttk.Checkbutton(group, text="启用详细日志", variable=self.verbose_var).grid(row=1, column=0, columnspan=2, sticky="w", pady=5)
ttk.Checkbutton(group, text="启用详细日志", variable=self.verbose_var).grid(
row=1, column=0, columnspan=2, sticky="w", pady=5
)
# 自动转换
self.auto_convert_var = tk.BooleanVar()
ttk.Checkbutton(group, text="自动转换 Excel 格式", variable=self.auto_convert_var).grid(row=2, column=0, columnspan=2, sticky="w", pady=5)
ttk.Checkbutton(
group, text="自动转换 Excel 格式", variable=self.auto_convert_var
).grid(row=2, column=0, columnspan=2, sticky="w", pady=5)
# 合并批次
self.merge_batches_var = tk.BooleanVar()
ttk.Checkbutton(group, text="自动合并批次数据", variable=self.merge_batches_var).grid(row=3, column=0, columnspan=2, sticky="w", pady=5)
ttk.Checkbutton(
group, text="自动合并批次数据", variable=self.merge_batches_var
).grid(row=3, column=0, columnspan=2, sticky="w", pady=5)
# 数据库持久化
self.enable_db_persistence_var = tk.BooleanVar()
ttk.Checkbutton(
group, text="保存到数据库 (同时写入 SQL Server)", variable=self.enable_db_persistence_var
).grid(row=4, column=0, columnspan=2, sticky="w", pady=5)
def load_settings(self):
"""从配置加载设置到界面"""
# ERP 设置
self.erp_url_var.set(self.config.get('erp.url', ''))
self.erp_username_var.set(self.config.get('erp.username', ''))
self.erp_password_var.set(self.config.get('erp.password', ''))
self.erp_url_var.set(self.config.get("erp.url", ""))
self.erp_username_var.set(self.config.get("erp.username", ""))
self.erp_password_var.set(self.config.get("erp.password", ""))
# 数据库设置
self.db_server_var.set(self.config.get('database.server', ''))
self.db_name_var.set(self.config.get('database.database', ''))
self.db_username_var.set(self.config.get('database.username', ''))
self.db_password_var.set(self.config.get('database.password', ''))
self.db_server_var.set(self.config.get("database.server", ""))
self.db_name_var.set(self.config.get("database.database", ""))
self.db_username_var.set(self.config.get("database.username", ""))
self.db_password_var.set(self.config.get("database.password", ""))
# 浏览器设置
self.browser_headless_var.set(self.config.get('browser.headless', True))
self.browser_ignore_https_var.set(self.config.get('browser.ignore_https_errors', True))
self.browser_auto_close_var.set(self.config.get('browser.auto_close', True))
# 浏览器设置(已合并到 ERP 配置中)
self.browser_headless_var.set(self.config.get("erp.headless", True))
self.browser_ignore_https_var.set(
self.config.get("erp.ignore_https_errors", True)
)
self.browser_auto_close_var.set(self.config.get("erp.auto_close_browser", True))
# 路径设置
self.data_dir_selector.set(self.config.get('paths.data_dir', ''))
self.default_output_var.set(self.config.get('paths.default_output', ''))
self.data_dir_selector.set(self.config.get("paths.data_dir", ""))
self.default_output_var.set(self.config.get("paths.default_output", ""))
# 处理设置
self.batch_size_var.set(self.config.get('extraction.batch_size', 100))
self.verbose_var.set(self.config.get('extraction.verbose', True))
self.auto_convert_var.set(self.config.get('extraction.auto_convert', True))
self.merge_batches_var.set(self.config.get('extraction.merge_batches', True))
self.batch_size_var.set(self.config.get("extraction.batch_size", 100))
self.verbose_var.set(self.config.get("extraction.verbose", True))
self.auto_convert_var.set(self.config.get("extraction.auto_convert", True))
self.merge_batches_var.set(self.config.get("extraction.merge_batches", True))
self.enable_db_persistence_var.set(self.config.get("extraction.enable_db_persistence", False))
def save_settings(self):
"""保存界面设置到配置"""
# ERP 设置
self.config.set('erp.url', self.erp_url_var.get())
self.config.set('erp.username', self.erp_username_var.get())
self.config.set('erp.password', self.erp_password_var.get())
self.config.set("erp.url", self.erp_url_var.get())
self.config.set("erp.username", self.erp_username_var.get())
self.config.set("erp.password", self.erp_password_var.get())
# 数据库设置
self.config.set('database.server', self.db_server_var.get())
self.config.set('database.database', self.db_name_var.get())
self.config.set('database.username', self.db_username_var.get())
self.config.set('database.password', self.db_password_var.get())
self.config.set("database.server", self.db_server_var.get())
self.config.set("database.database", self.db_name_var.get())
self.config.set("database.username", self.db_username_var.get())
self.config.set("database.password", self.db_password_var.get())
# 浏览器设置
self.config.set('browser.headless', self.browser_headless_var.get())
self.config.set('browser.ignore_https_errors', self.browser_ignore_https_var.get())
self.config.set('browser.auto_close', self.browser_auto_close_var.get())
# 浏览器设置(已合并到 ERP 配置中)
self.config.set("erp.headless", self.browser_headless_var.get())
self.config.set("erp.ignore_https_errors", self.browser_ignore_https_var.get())
self.config.set("erp.auto_close_browser", self.browser_auto_close_var.get())
# 路径设置
self.config.set('paths.data_dir', self.data_dir_selector.get())
self.config.set('paths.default_output', self.default_output_var.get())
self.config.set("paths.data_dir", self.data_dir_selector.get())
self.config.set("paths.default_output", self.default_output_var.get())
# 处理设置
self.config.set('extraction.batch_size', self.batch_size_var.get())
self.config.set('extraction.verbose', self.verbose_var.get())
self.config.set('extraction.auto_convert', self.auto_convert_var.get())
self.config.set('extraction.merge_batches', self.merge_batches_var.get())
self.config.set("extraction.batch_size", self.batch_size_var.get())
self.config.set("extraction.verbose", self.verbose_var.get())
self.config.set("extraction.auto_convert", self.auto_convert_var.get())
self.config.set("extraction.merge_batches", self.merge_batches_var.get())
self.config.set("extraction.enable_db_persistence", self.enable_db_persistence_var.get())
# 保存到文件
if self.config.save():

39
gui/utils.py Normal file
View File

@@ -0,0 +1,39 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
GUI 工具模块
提供 GUI 相关的工具类和函数。
"""
class RealtimeOutput:
"""实时输出流,每次写入立即回调通知"""
def __init__(self, callback):
"""
初始化实时输出流
Args:
callback: 写入时的回调函数,接收文本内容
"""
self.callback = callback
self.buffer = []
def write(self, text):
"""写入文本"""
if text:
# 将文本按行分割,逐行回调
lines = text.split("\n")
for line in lines:
if line: # 忽略空行(由 split 产生)
self.callback(line)
return len(text)
def flush(self):
"""刷新(兼容性方法)"""
pass
def isatty(self):
"""返回 False表示不是终端"""
return False

View File

@@ -1,6 +1,7 @@
"""
主程序 - 使用离散备料计划维护数据提取工具
"""
import os
from utils.离散备料计划维护数据提取 import DiscreteMaterialPlanExtractor
@@ -11,7 +12,7 @@ def main():
username="BLDpengqiangqiang",
password="Cqbld123456.",
headless=True,
verbose=True
verbose=True,
)
# 设置文件路径
@@ -22,6 +23,5 @@ def main():
extractor.extract(order_id_file, output_file)
if __name__ == "__main__":
main()

View File

@@ -1,6 +1,7 @@
"""
主程序 - 使用离散备料计划维护数据清理工具
"""
import os
from utils.离散备料计划维护数据清理 import DiscreteMaterialPlanCleaner
@@ -12,7 +13,7 @@ def main():
password="Cqbld123456.",
manager_name="彭羽",
headless=True,
verbose=True
verbose=True,
)
# 设置文件路径
@@ -22,6 +23,5 @@ def main():
cleaner.clean(order_id_file)
if __name__ == "__main__":
main()

View File

@@ -1,42 +1,52 @@
import re
import time
from playwright.sync_api import Playwright, sync_playwright, expect, TimeoutError as PWTimeoutError
from playwright.sync_api import (
Playwright,
sync_playwright,
expect,
TimeoutError as PWTimeoutError,
)
from utils.auth import login
def click_button_until_disappear(frame, button_name="保存提交", interval=5, max_attempts=None, max_duration=None):
def click_button_until_disappear(
frame, button_name="保存提交", interval=5, max_attempts=None, max_duration=None
):
"""
持续点击按钮直到按钮消失
参数:
frame: iframe对象
button_name: 按钮名称
interval: 点击间隔时间(秒)
max_attempts: 最大点击次数None表示不限制
max_duration: 最大持续时间None表示不限制
返回:
dict: 包含点击次数、耗时等信息
"""
click_count = 0
start_time = time.time()
print(f"开始监控'{button_name}'按钮...")
while True:
try:
# 检查最大点击次数
if max_attempts and click_count >= max_attempts:
print(f"已达到最大点击次数{max_attempts}次,停止操作")
break
# 检查最大持续时间
if max_duration and (time.time() - start_time) >= max_duration:
elapsed = time.time() - start_time
print(f"已达到最大持续时间{max_duration}秒(实际{elapsed:.1f}秒),停止操作")
print(
f"已达到最大持续时间{max_duration}秒(实际{elapsed:.1f}秒),停止操作"
)
break
button = frame.get_by_role("button", name=button_name)
# 检查按钮是否存在
if button.count() == 0:
elapsed = time.time() - start_time
@@ -45,17 +55,17 @@ def click_button_until_disappear(frame, button_name="保存提交", interval=5,
return {
"success": True,
"click_count": click_count,
"elapsed_time": elapsed
"elapsed_time": elapsed,
}
# 点击按钮
button.click()
click_count += 1
print(f"[{time.strftime('%H:%M:%S')}] 第{click_count}次点击'{button_name}'")
# 等待指定时间
time.sleep(interval)
except Exception as e:
elapsed = time.time() - start_time
print(f"点击过程中出现异常: {e}")
@@ -63,20 +73,19 @@ def click_button_until_disappear(frame, button_name="保存提交", interval=5,
"success": False,
"click_count": click_count,
"elapsed_time": elapsed,
"error": str(e)
"error": str(e),
}
def get_input_by_label(frame, label_text: str, label_locator=None):
"""
通过标签文本获取对应的输入框对象
参数:
frame: iframe对象
label_text: 标签文本(用于查找和日志输出)
label_locator: 可选已经定位好的标签locator。如果为None则函数内部查找
返回:
成功返回输入框的locator对象
失败返回None
@@ -84,21 +93,25 @@ def get_input_by_label(frame, label_text: str, label_locator=None):
try:
# 如果没有传入label_locator则根据label_text查找
if label_locator is None:
label_locator = frame.locator("div").filter(has_text=re.compile(f"^{label_text}$")).first
label_locator = (
frame.locator("div")
.filter(has_text=re.compile(f"^{label_text}$"))
.first
)
print(f"找到{label_text}标签")
# 向上找到包含标签和输入框的共同父容器
parent_container = label_locator.locator("..") # 父元素
# 在父容器中查找输入框
input_box = parent_container.locator("input").first
# 如果父元素中没有,再向上一层
if input_box.count() == 0:
print(f"{label_text}: 在父元素中未找到,向上一层查找...")
grandparent = parent_container.locator("..") # 祖父元素
input_box = grandparent.locator("input").first
# 验证是否找到输入框
if input_box.count() > 0:
input_box.wait_for(state="visible", timeout=5000)
@@ -109,11 +122,12 @@ def get_input_by_label(frame, label_text: str, label_locator=None):
else:
print(f"{label_text}: 在父容器中找不到输入框")
return None
except Exception as e:
print(f"{label_text}失败: {e}")
return None
def run(playwright: Playwright) -> None:
# 1. 登录
browser, context, page, main_frame = login(
@@ -121,7 +135,7 @@ def run(playwright: Playwright) -> None:
username="BLDpengqiangqiang",
password="Cqbld123456.",
headless=False,
ignore_https_errors=True
ignore_https_errors=True,
)
# 3. 点击打开“补货安排”
@@ -138,7 +152,7 @@ def run(playwright: Playwright) -> None:
# 提取外层 forwardFrame
outer_frame = page1.locator("#forwardFrame").content_frame
# 关键:等待内层 #mainiframe 出现并加载
inner_frame_locator = outer_frame.locator("#mainiframe")
inner_frame_locator.wait_for(state="visible", timeout=15000) # 最多等15秒
@@ -153,7 +167,7 @@ def run(playwright: Playwright) -> None:
# 输入排产号
textbox = inner_frame.get_by_role("textbox", name="排产号")
textbox.fill("R") # 直接 fill不需要 press CapsLock
# 输入日期
inner_frame.get_by_role("textbox", name="单据日期结束日期").click()
inner_frame.get_by_text("今日").click()
@@ -170,7 +184,7 @@ def run(playwright: Playwright) -> None:
summary_locator = inner_frame.get_by_text(re.compile(r"合计:\s*\d+\s*行"))
# 获取元素的完整文本
summary_text = summary_locator.inner_text() # 例如 "合计: 135 行"
summary_text = summary_locator.inner_text() # 例如 "合计: 135 行"
# 用正则提取数字
match = re.search(r"\d+", summary_text)
@@ -181,14 +195,12 @@ def run(playwright: Playwright) -> None:
print("未匹配到行数")
row_count = 0
inner_frame.get_by_role("button").filter(has_text="补货安排").hover()
inner_frame.get_by_text("生产订单").click()
inner_frame.get_by_role("textbox", name="工厂").fill("10010705")
with page1.expect_popup(timeout=60000) as page2_info:
inner_frame.get_by_role("button", name="确定(Y)").click()
page2=page2_info.value
page2 = page2_info.value
# 新页面:等待页面加载完成 + 提取嵌套 iframe
print("新页面已打开,正在等待内层 iframe 加载...")
@@ -204,7 +216,6 @@ def run(playwright: Playwright) -> None:
# label_div = inner_frame.locator("div").filter(has_text=re.compile(r"^生产部门$")).first
# input_box = label_div.locator("..").locator(".wui-input-close > .wui-input")
time.sleep(25) # 等待页面完全加载
# pro_dep_input = get_input_by_label(inner_frame, "生产部门")
# if pro_dep_input:
@@ -216,18 +227,14 @@ def run(playwright: Playwright) -> None:
# if pro_SN_input:
# print(pro_SN_input.input_value())
#result = click_button_until_disappear(inner_frame, "保存提交", interval=5)
#print(result)
# result = click_button_until_disappear(inner_frame, "保存提交", interval=5)
# print(result)
input("操作完成,按回车关闭...")
context.close()
browser.close()
with sync_playwright() as playwright:
run(playwright)
run(playwright)

View File

@@ -1,14 +1,16 @@
"""
分析 Excel 文件的数据结构
"""
import pandas as pd
import openpyxl
import sys
# 设置输出编码
if sys.platform == 'win32':
if sys.platform == "win32":
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
# 读取 Excel 文件
file_path = "data/导出文件.xlsx"
@@ -52,12 +54,12 @@ for sheet_name in sheet_names:
for i, col in enumerate(df.columns):
print(f"{i}: {col}")
print(f"\n数据预览:")
pd.set_option('display.max_rows', 25)
pd.set_option('display.max_columns', 20)
pd.set_option('display.width', 200)
pd.set_option('display.max_colwidth', 30)
pd.set_option("display.max_rows", 25)
pd.set_option("display.max_columns", 20)
pd.set_option("display.width", 200)
pd.set_option("display.max_colwidth", 30)
print(df)
pd.reset_option('display.max_rows')
pd.reset_option('display.max_columns')
pd.reset_option('display.width')
pd.reset_option('display.max_colwidth')
pd.reset_option("display.max_rows")
pd.reset_option("display.max_columns")
pd.reset_option("display.width")
pd.reset_option("display.max_colwidth")

View File

@@ -15,12 +15,18 @@ def number_to_excel_col(n):
result = ""
while n > 0:
n -= 1
result = chr(n % 26 + ord('A')) + result
result = chr(n % 26 + ord("A")) + result
n //= 26
return result
def excel_to_markdown(input_file, output_file=None, sheet_name=0, include_row_numbers=True, include_col_numbers=True):
def excel_to_markdown(
input_file,
output_file=None,
sheet_name=0,
include_row_numbers=True,
include_col_numbers=True,
):
"""
将Excel文件转换为Markdown表格
@@ -39,7 +45,7 @@ def excel_to_markdown(input_file, output_file=None, sheet_name=0, include_row_nu
# 设置默认输出文件名
if output_file is None:
output_file = input_path.with_suffix('.md')
output_file = input_path.with_suffix(".md")
else:
output_file = Path(output_file)
@@ -80,7 +86,7 @@ def excel_to_markdown(input_file, output_file=None, sheet_name=0, include_row_nu
for sheet_key, df in dfs.items():
# 转换数据为字符串处理NaN值
df = df.fillna('')
df = df.fillna("")
df = df.astype(str)
# 生成Markdown表格
@@ -92,10 +98,14 @@ def excel_to_markdown(input_file, output_file=None, sheet_name=0, include_row_nu
# 添加列号行
if include_col_numbers:
col_headers = [''] if include_row_numbers else []
col_headers.extend(number_to_excel_col(i + 1) for i in range(len(df.columns)))
markdown_lines.append('| ' + ' | '.join(col_headers) + ' |')
markdown_lines.append('| ' + ' | '.join(['---' for _ in col_headers]) + ' |')
col_headers = [""] if include_row_numbers else []
col_headers.extend(
number_to_excel_col(i + 1) for i in range(len(df.columns))
)
markdown_lines.append("| " + " | ".join(col_headers) + " |")
markdown_lines.append(
"| " + " | ".join(["---" for _ in col_headers]) + " |"
)
# 添加数据行
for idx, row in df.iterrows():
@@ -103,15 +113,17 @@ def excel_to_markdown(input_file, output_file=None, sheet_name=0, include_row_nu
if include_row_numbers:
row_data.append(str(idx + 1))
row_data.extend(row)
markdown_lines.append('| ' + ' | '.join(row_data) + ' |')
markdown_lines.append("| " + " | ".join(row_data) + " |")
# 添加统计信息
markdown_lines.append(f"\n**统计信息:**")
markdown_lines.append(f"- 总行数: {len(df)}")
markdown_lines.append(f"- 总列数: {len(df.columns)}")
markdown_lines.append(f"- 数据范围: A1:{number_to_excel_col(len(df.columns))}{len(df)}")
markdown_lines.append(
f"- 数据范围: A1:{number_to_excel_col(len(df.columns))}{len(df)}"
)
all_sheets_content[sheet_key] = '\n'.join(markdown_lines)
all_sheets_content[sheet_key] = "\n".join(markdown_lines)
# 写入文件
if len(dfs) == 1:
@@ -120,7 +132,7 @@ def excel_to_markdown(input_file, output_file=None, sheet_name=0, include_row_nu
output_content = f"# {input_path.stem}\n\n"
output_content += f"从 `{input_file}` (工作表: {sheet_key}) 转换\n\n"
output_content += all_sheets_content[sheet_key]
output_file.write_text(output_content, encoding='utf-8')
output_file.write_text(output_content, encoding="utf-8")
print(f"✓ 转换成功!")
print(f" 输入文件: {input_file}")
@@ -133,11 +145,19 @@ def excel_to_markdown(input_file, output_file=None, sheet_name=0, include_row_nu
except Exception as e:
print(f"错误: {str(e)}")
import traceback
traceback.print_exc()
return False
def convert_multiple_sheets(input_file, output_file=None, sheet_names=None, include_row_numbers=True, include_col_numbers=True, merge_to_one_file=True):
def convert_multiple_sheets(
input_file,
output_file=None,
sheet_names=None,
include_row_numbers=True,
include_col_numbers=True,
merge_to_one_file=True,
):
"""
转换多个工作表
@@ -157,7 +177,7 @@ def convert_multiple_sheets(input_file, output_file=None, sheet_names=None, incl
# 设置默认输出文件名
if output_file is None:
output_file = input_path.with_suffix('.md')
output_file = input_path.with_suffix(".md")
else:
output_file = Path(output_file)
@@ -200,7 +220,7 @@ def convert_multiple_sheets(input_file, output_file=None, sheet_names=None, incl
for sheet_key, df in dfs.items():
# 转换数据为字符串处理NaN值
df = df.fillna('')
df = df.fillna("")
df = df.astype(str)
# 生成Markdown表格
@@ -212,10 +232,14 @@ def convert_multiple_sheets(input_file, output_file=None, sheet_names=None, incl
# 添加列号行
if include_col_numbers:
col_headers = [''] if include_row_numbers else []
col_headers.extend(number_to_excel_col(i + 1) for i in range(len(df.columns)))
markdown_lines.append('| ' + ' | '.join(col_headers) + ' |')
markdown_lines.append('| ' + ' | '.join(['---' for _ in col_headers]) + ' |')
col_headers = [""] if include_row_numbers else []
col_headers.extend(
number_to_excel_col(i + 1) for i in range(len(df.columns))
)
markdown_lines.append("| " + " | ".join(col_headers) + " |")
markdown_lines.append(
"| " + " | ".join(["---" for _ in col_headers]) + " |"
)
# 添加数据行
for idx, row in df.iterrows():
@@ -223,15 +247,17 @@ def convert_multiple_sheets(input_file, output_file=None, sheet_names=None, incl
if include_row_numbers:
row_data.append(str(idx + 1))
row_data.extend(row)
markdown_lines.append('| ' + ' | '.join(row_data) + ' |')
markdown_lines.append("| " + " | ".join(row_data) + " |")
# 添加统计信息
markdown_lines.append(f"\n**统计信息:**")
markdown_lines.append(f"- 总行数: {len(df)}")
markdown_lines.append(f"- 总列数: {len(df.columns)}")
markdown_lines.append(f"- 数据范围: A1:{number_to_excel_col(len(df.columns))}{len(df)}")
markdown_lines.append(
f"- 数据范围: A1:{number_to_excel_col(len(df.columns))}{len(df)}"
)
all_sheets_content[sheet_key] = '\n'.join(markdown_lines)
all_sheets_content[sheet_key] = "\n".join(markdown_lines)
# 写入文件
if merge_to_one_file:
@@ -244,7 +270,7 @@ def convert_multiple_sheets(input_file, output_file=None, sheet_names=None, incl
if sheet_key in all_sheets_content:
output_content += all_sheets_content[sheet_key] + "\n\n---\n\n"
output_file.write_text(output_content, encoding='utf-8')
output_file.write_text(output_content, encoding="utf-8")
print(f"✓ 转换成功!")
print(f" 输入文件: {input_file}")
@@ -252,7 +278,9 @@ def convert_multiple_sheets(input_file, output_file=None, sheet_names=None, incl
print(f" 工作表数量: {len(dfs)}")
for sheet_key in sheet_names:
if sheet_key in dfs:
print(f" - {sheet_key}: {len(dfs[sheet_key])}行 x {len(dfs[sheet_key].columns)}")
print(
f" - {sheet_key}: {len(dfs[sheet_key])}行 x {len(dfs[sheet_key].columns)}"
)
else:
# 分别输出到多个文件
output_stem = output_file.stem
@@ -271,7 +299,7 @@ def convert_multiple_sheets(input_file, output_file=None, sheet_names=None, incl
output_content += f"从 `{input_file}` (工作表: {sheet_key}) 转换\n\n"
output_content += all_sheets_content[sheet_key]
sheet_output_file.write_text(output_content, encoding='utf-8')
sheet_output_file.write_text(output_content, encoding="utf-8")
print(f"✓ 转换成功!")
print(f" 输入文件: {input_file}")
@@ -280,13 +308,16 @@ def convert_multiple_sheets(input_file, output_file=None, sheet_names=None, incl
for sheet_key in sheet_names:
if sheet_key in dfs:
sheet_filename = f"{output_stem}_{sheet_key}{output_suffix}"
print(f" - {sheet_key} -> {sheet_filename} ({len(dfs[sheet_key])}行 x {len(dfs[sheet_key].columns)}列)")
print(
f" - {sheet_key} -> {sheet_filename} ({len(dfs[sheet_key])}行 x {len(dfs[sheet_key].columns)}列)"
)
return True
except Exception as e:
print(f"错误: {str(e)}")
import traceback
traceback.print_exc()
return False
@@ -305,7 +336,7 @@ OUTPUT_FILE = None # 或指定 r"D:\path\to\output.md"
# 可以是单个值: 0 或 "Sheet1"
# 可以是列表: [0, 1, 2] 或 ["Sheet1", "Sheet2", "Sheet3"]
# 可以是 None 表示读取所有工作表
SHEET_NAMES = [0, 1,2,3] # 指定多个工作表索引
SHEET_NAMES = [0, 1, 2, 3] # 指定多个工作表索引
# 是否包含行号可选默认为True
INCLUDE_ROW_NUMBERS = True
@@ -332,19 +363,23 @@ def main():
sheet_names=SHEET_NAMES,
include_row_numbers=INCLUDE_ROW_NUMBERS,
include_col_numbers=INCLUDE_COL_NUMBERS,
merge_to_one_file=MULTI_SHEETS_TO_ONE_FILE
merge_to_one_file=MULTI_SHEETS_TO_ONE_FILE,
)
else:
# 单个工作表
sheet_name = SHEET_NAMES if isinstance(SHEET_NAMES, list) and len(SHEET_NAMES) == 1 else SHEET_NAMES
sheet_name = (
SHEET_NAMES
if isinstance(SHEET_NAMES, list) and len(SHEET_NAMES) == 1
else SHEET_NAMES
)
excel_to_markdown(
input_file=INPUT_FILE,
output_file=OUTPUT_FILE,
sheet_name=sheet_name,
include_row_numbers=INCLUDE_ROW_NUMBERS,
include_col_numbers=INCLUDE_COL_NUMBERS
include_col_numbers=INCLUDE_COL_NUMBERS,
)
if __name__ == '__main__':
if __name__ == "__main__":
main()

View File

@@ -1,6 +1,7 @@
"""
元素定位辅助工具 - 用于快速验证定位是否有效
"""
from playwright.sync_api import Page, Frame, Locator
@@ -38,7 +39,9 @@ def debug_locator(frame: Frame, locator: Locator, timeout: int = 5000):
try:
if element.is_visible(timeout=1000):
text = element.inner_text(timeout=1000)
print(f" 元素{i + 1}文本: {text[:100] if len(text) > 100 else text}")
print(
f" 元素{i + 1}文本: {text[:100] if len(text) > 100 else text}"
)
else:
print(f" 元素{i + 1}: 存在但不可见")
except:
@@ -53,7 +56,9 @@ def debug_locator(frame: Frame, locator: Locator, timeout: int = 5000):
return False
def try_multiple_locators(frame: Frame, selectors: list[str], timeout: int = 5000) -> Locator:
def try_multiple_locators(
frame: Frame, selectors: list[str], timeout: int = 5000
) -> Locator:
"""
尝试多个选择器,返回第一个有效的定位器
@@ -73,7 +78,9 @@ def try_multiple_locators(frame: Frame, selectors: list[str], timeout: int = 500
try:
locator = frame.locator(selector)
count = locator.count()
visible_count = sum(1 for j in range(count) if locator.nth(j).is_visible(timeout=1000))
visible_count = sum(
1 for j in range(count) if locator.nth(j).is_visible(timeout=1000)
)
print(f" 找到 {count} 个元素,其中 {visible_count} 个可见")
@@ -101,7 +108,7 @@ def interactive_locate(frame: Frame):
selector = input("\n>>> ")
selector = selector.strip()
if selector.lower() in ('q', 'quit'):
if selector.lower() in ("q", "quit"):
break
if not selector:
@@ -128,7 +135,9 @@ if __name__ == "__main__":
page = context.new_page()
# 登录
page.goto("https://68.11.34.30:8082/yonbip/resources/uap/rbac/login/main/index.html")
page.goto(
"https://68.11.34.30:8082/yonbip/resources/uap/rbac/login/main/index.html"
)
main_frame = page.locator("#forwardFrame").content_frame
main_frame.get_by_role("textbox", name="用户名").fill("BLDpengqiangqiang")
main_frame.get_by_role("textbox", name="密码").fill("Cqbld123456.")
@@ -158,10 +167,10 @@ if __name__ == "__main__":
# 询问是否继续
choice = input("\n是否继续下一轮调试?(y/n/q): ").strip().lower()
if choice in ('n', 'q', 'quit'):
if choice in ("n", "q", "quit"):
print("退出程序")
break
elif choice in ('y', ''): # 默认继续
elif choice in ("y", ""): # 默认继续
continue
else:
print("未知选项,退出程序")

View File

@@ -1,7 +1,8 @@
"""
工具组件包
"""
from .excel_converter import ExcelConverter
from .离散备料计划维护数据提取 import DiscreteMaterialPlanExtractor
__all__ = ['ExcelConverter', 'DiscreteMaterialPlanExtractor']
__all__ = ["ExcelConverter", "DiscreteMaterialPlanExtractor"]

View File

@@ -1,6 +1,7 @@
"""
认证模块 - 负责用友BIP系统的登录和退出操作
"""
from playwright.sync_api import Playwright, Browser, BrowserContext, Page, Frame
@@ -11,7 +12,7 @@ def login(
url: str = "https://68.11.34.30:8082/yonbip/resources/uap/rbac/login/main/index.html",
headless: bool = False,
ignore_https_errors: bool = True,
verbose: bool = True
verbose: bool = True,
) -> tuple[Browser, BrowserContext, Page, Frame]:
"""
登录用友BIP系统

View File

@@ -2,6 +2,7 @@
Excel 报表数据转换工具组件
将 Excel 报表数据转换为数据库记录形式
"""
import pandas as pd
import openpyxl
from typing import List, Dict, Optional
@@ -12,10 +13,7 @@ class ExcelConverter:
"""Excel 报表数据转换器"""
# 字段名称映射(解决字段名冲突)
FIELD_NAME_MAPPING = {
'计划数量': '产品计划数量',
'单位': '产品单位'
}
FIELD_NAME_MAPPING = {"计划数量": "产品计划数量", "单位": "产品单位"}
def __init__(self, verbose: bool = True):
"""
@@ -115,7 +113,7 @@ class ExcelConverter:
row = all_rows[i]
# 检查是否是订单标题行
if row and '离散备料计划' in str(row[0]):
if row and "离散备料计划" in str(row[0]):
# 解析订单头信息接下来的4行
order_info = {}
for j in range(1, 5):
@@ -124,16 +122,27 @@ class ExcelConverter:
# 跳过空行,找到表格标题行
table_row = i + 5
while table_row < len(all_rows) and (not all_rows[table_row] or not all_rows[table_row][0]):
while table_row < len(all_rows) and (
not all_rows[table_row] or not all_rows[table_row][0]
):
table_row += 1
# 检查是否是表格标题行
if table_row < len(all_rows) and all_rows[table_row] and all_rows[table_row][0] == '序号':
if (
table_row < len(all_rows)
and all_rows[table_row]
and all_rows[table_row][0] == "序号"
):
# 检查表头下一行是否为空,判断是否存在数据
next_row = table_row + 1
is_empty_row = (next_row < len(all_rows) and
all_rows[next_row] and
all(cell is None or str(cell).strip() == "" for cell in all_rows[next_row]))
is_empty_row = (
next_row < len(all_rows)
and all_rows[next_row]
and all(
cell is None or str(cell).strip() == ""
for cell in all_rows[next_row]
)
)
if is_empty_row:
# 没有数据,查找页脚信息
@@ -141,17 +150,27 @@ class ExcelConverter:
footer_info = {}
data_row = next_row + 1
while data_row < len(all_rows) and all_rows[data_row]:
if all_rows[data_row][0] and ('制单人' in str(all_rows[data_row][0]) or '打印人' in str(all_rows[data_row][0])):
if all_rows[data_row][0] and (
"制单人" in str(all_rows[data_row][0])
or "打印人" in str(all_rows[data_row][0])
):
self._parse_header_row(all_rows[data_row], footer_info)
if data_row + 1 < len(all_rows) and all_rows[data_row + 1]:
self._parse_header_row(all_rows[data_row + 1], footer_info)
if (
data_row + 1 < len(all_rows)
and all_rows[data_row + 1]
):
self._parse_header_row(
all_rows[data_row + 1], footer_info
)
break
data_row += 1
orders.append({
'order_info': {**order_info, **footer_info},
'materials': materials
})
orders.append(
{
"order_info": {**order_info, **footer_info},
"materials": materials,
}
)
else:
# 有数据,开始提取物料
materials = []
@@ -159,39 +178,48 @@ class ExcelConverter:
data_row = table_row + 1
while data_row < len(all_rows) and all_rows[data_row]:
# 检查是否是页脚信息(制单人、打印人)
if all_rows[data_row+1][0] and '制单人' in str(all_rows[data_row+1][0]) :
if all_rows[data_row + 1][0] and "制单人" in str(
all_rows[data_row + 1][0]
):
# 解析页脚信息
self._parse_header_row(all_rows[data_row], footer_info)
# 检查下一行是否也是页脚信息
if data_row + 1 < len(all_rows) and all_rows[data_row + 1]:
self._parse_header_row(all_rows[data_row + 1], footer_info)
if (
data_row + 1 < len(all_rows)
and all_rows[data_row + 1]
):
self._parse_header_row(
all_rows[data_row + 1], footer_info
)
break
# 提取物料数据
material_row = all_rows[data_row]
material = {
'序号': material_row[0],
'材料编码': material_row[1],
'材料名称': material_row[2],
'规格': material_row[3],
'型号': material_row[4],
'图号': material_row[5],
'物料材质': material_row[6],
'计划数量': material_row[7],
'单位': material_row[8],
'需用日期': material_row[9],
'发料仓库': material_row[10],
'单位用量': material_row[11],
'累计出库数量': material_row[12],
"序号": material_row[0],
"材料编码": material_row[1],
"材料名称": material_row[2],
"规格": material_row[3],
"型号": material_row[4],
"图号": material_row[5],
"物料材质": material_row[6],
"计划数量": material_row[7],
"单位": material_row[8],
"需用日期": material_row[9],
"发料仓库": material_row[10],
"单位用量": material_row[11],
"累计出库数量": material_row[12],
}
materials.append(material)
data_row += 1
orders.append({
'order_info': {**order_info, **footer_info},
'materials': materials
})
orders.append(
{
"order_info": {**order_info, **footer_info},
"materials": materials,
}
)
i += 1
@@ -208,9 +236,9 @@ class ExcelConverter:
i = 0
while i < len(row):
cell = row[i]
if cell and str(cell).strip() and '' in str(cell):
if cell and str(cell).strip() and "" in str(cell):
# 找到字段名
field_name = str(cell).replace('', '').strip()
field_name = str(cell).replace("", "").strip()
# 应用字段名映射
if field_name in self.FIELD_NAME_MAPPING:
@@ -218,9 +246,11 @@ class ExcelConverter:
# 跳过空单元格,找到第一个非字段名的值
j = i + 1
while j < len(row) and (not row[j] or not str(row[j]).strip() or '' in str(row[j])):
while j < len(row) and (
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()
# 跳过已处理的值,继续找下一个字段名
i = j + 1
@@ -240,14 +270,11 @@ class ExcelConverter:
all_records = []
for order in orders:
order_info = order['order_info']
materials = order['materials']
order_info = order["order_info"]
materials = order["materials"]
for material in materials:
record = {
**order_info,
**material
}
record = {**order_info, **material}
all_records.append(record)
return pd.DataFrame(all_records)

View File

@@ -2,6 +2,7 @@
物料状态校验工具
校验订单中的物料状态,匹配待删除物料
"""
import os
import pandas as pd
from typing import List, Dict, Any
@@ -49,8 +50,9 @@ class MaterialStatusValidator:
material_names = [str(name) for name in material_names]
return material_names
def match_materials(self, material_names: List[str],
db_materials: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
def match_materials(
self, material_names: List[str], db_materials: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
"""
匹配材料名称
@@ -66,22 +68,27 @@ class MaterialStatusValidator:
matched = None
for db_record in db_materials:
# 如果数据库的MaterialName出现在Excel的材料名称中
if db_record['MaterialName'] in material_name:
if db_record["MaterialName"] in material_name:
matched = db_record
break
results.append({
'材料名称': material_name,
'匹配的MaterialName': matched['MaterialName'] if matched else None,
'负责人': matched['ManagerName'] if matched else None,
'匹配状态': '匹配成功' if matched else '未匹配'
})
results.append(
{
"材料名称": material_name,
"匹配的MaterialName": matched["MaterialName"] if matched else None,
"负责人": matched["ManagerName"] if matched else None,
"匹配状态": "匹配成功" if matched else "未匹配",
}
)
return results
def validate(self, production_id_file: str,
merged_excel_file: str = None,
output_file: str = None) -> str:
def validate(
self,
production_id_file: str,
merged_excel_file: str = None,
output_file: str = None,
) -> str:
"""
执行完整的校验流程
@@ -106,7 +113,7 @@ class MaterialStatusValidator:
username=self.username,
password=self.password,
headless=self.headless,
verbose=self.verbose
verbose=self.verbose,
)
extractor.extract(production_id_file, output_file=merged_excel_file)
self._print(f"数据提取完成: {merged_excel_file}")
@@ -132,7 +139,7 @@ class MaterialStatusValidator:
self._print(f"结果已保存: {output_file}")
# 打印统计信息
matched_count = sum(1 for r in results if r['匹配状态'] == '匹配成功')
matched_count = sum(1 for r in results if r["匹配状态"] == "匹配成功")
self._print(f"\n统计信息:")
self._print(f" 总材料数: {len(results)}")
self._print(f" 匹配成功: {matched_count}")
@@ -140,7 +147,9 @@ class MaterialStatusValidator:
return output_file
def validate_from_existing_excel(self, excel_file: str, output_file: str = None) -> str:
def validate_from_existing_excel(
self, excel_file: str, output_file: str = None
) -> str:
"""
从已存在的Excel文件执行校验不需要重新提取数据
@@ -179,7 +188,7 @@ class MaterialStatusValidator:
self._print(f"结果已保存: {output_file}")
# 打印统计信息
matched_count = sum(1 for r in results if r['匹配状态'] == '匹配成功')
matched_count = sum(1 for r in results if r["匹配状态"] == "匹配成功")
self._print(f"\n统计信息:")
self._print(f" 总材料数: {len(results)}")
self._print(f" 匹配成功: {matched_count}")

View File

@@ -2,18 +2,26 @@
离散备料计划维护数据提取工具
负责登录、批量下载、转换数据
"""
import os
import pandas as pd
from playwright.sync_api import sync_playwright
from utils.excel_converter import ExcelConverter
from utils.auth import login, logout
from db.production_order_query import read_production_ids, query_production_order_numbers
from db.production_order_query import (
read_production_ids,
query_production_order_numbers,
)
from typing import Callable, Optional
class DiscreteMaterialPlanExtractor:
"""离散备料计划维护数据提取器"""
def __init__(self, username, password, headless=False, verbose=True):
def __init__(
self, username, password, headless=False, verbose=True, batch_size=100,
enable_db_persistence=False
):
"""
初始化提取器
@@ -22,61 +30,172 @@ class DiscreteMaterialPlanExtractor:
password: 登录密码
headless: 是否无头模式运行
verbose: 是否打印详细日志
batch_size: 批次大小
enable_db_persistence: 是否启用数据库持久化
"""
self.username = username
self.password = password
self.headless = headless
self.verbose = verbose
self.batch_size = batch_size
self.progress_callback = None
self.converter = ExcelConverter(verbose=verbose)
self.enable_db_persistence = enable_db_persistence
self.dao = None
if self.enable_db_persistence:
from db.discrete_material_plan_dao import DiscreteMaterialPlanDAO
self.dao = DiscreteMaterialPlanDAO()
self.dao.__enter__() # Enter context manager
def _print(self, *args, **kwargs):
"""打印日志(如果 verbose=True"""
if self.verbose:
print(*args, **kwargs)
def get_production_order_numbers(self, production_id_file):
def _report_progress(
self, stage: str, current: int, total: int, message: str, **detail
):
"""
报告进度
Args:
stage: 阶段标识
current: 当前进度值
total: 总量
message: 显示消息
**detail: 额外详细信息
"""
if self.progress_callback:
try:
from gui.progress import ProgressInfo
progress_info = ProgressInfo(
stage=stage,
current=current,
total=total,
message=message,
detail=detail,
)
self.progress_callback(progress_info)
except Exception:
# 如果进度回调失败,忽略错误,不影响主流程
pass
def get_production_order_numbers(self, production_id_file, report_progress=False):
"""
读取总排号文件并查询数据库获取生产订单号
Args:
production_id_file: ProductionID.txt 文件路径
report_progress: 是否报告进度
Returns:
生产订单号列表
"""
if report_progress:
self._report_progress(
"query",
1,
3,
"正在读取总排号文件...",
action="read_file",
)
# 读取总排号
production_ids = read_production_ids(production_id_file)
self._print(f"从文件读取到 {len(production_ids)} 个总排号")
if report_progress:
self._report_progress(
"query",
2,
3,
f"正在查询数据库({len(production_ids)} 个总排号)...",
action="query_database",
production_id_count=len(production_ids),
)
# 查询数据库获取生产订单号
order_ids = query_production_order_numbers(production_ids)
self._print(f"查询到 {len(order_ids)} 个生产订单号")
if report_progress:
self._report_progress(
"query",
3,
3,
f"查询完成:获取到 {len(order_ids)} 个生产订单号",
action="query_complete",
order_id_count=len(order_ids),
)
return order_ids
def group_order_ids(self, order_ids, group_size=100):
"""将订单号分组"""
for i in range(0, len(order_ids), group_size):
yield order_ids[i:i + group_size]
yield order_ids[i : i + group_size]
def download_batch(self, inner_frame, order_ids, batch_index, page1, debug_mode=False, debug_batch=None):
def download_batch(
self,
inner_frame,
order_ids,
batch_index,
total_batches,
page1,
debug_mode=False,
debug_batch=None,
):
"""下载一批订单号的数据"""
from playwright.sync_api import TimeoutError
import re
import time
# 清空文本框
# 步骤1清空文本框
self._report_progress(
"download",
batch_index * 7 + 1,
total_batches * 7,
f"{batch_index + 1}/{total_batches} 批 - 准备输入订单号",
batch_index=batch_index + 1,
action="clear_textbox",
)
textbox = inner_frame.get_by_role("textbox", name="来源生产订单号")
textbox.fill("")
# 填充订单号
# 步骤2填充订单号
self._report_progress(
"download",
batch_index * 7 + 2,
total_batches * 7,
f"{batch_index + 1}/{total_batches} 批 - 输入 {len(order_ids)} 个订单号",
batch_index=batch_index + 1,
action="fill_order_ids",
order_count=len(order_ids),
)
textbox.fill(",".join(order_ids))
# 点击查询
# 步骤3点击查询
self._report_progress(
"download",
batch_index * 7 + 3,
total_batches * 7,
f"{batch_index + 1}/{total_batches} 批 - 提交查询请求",
batch_index=batch_index + 1,
action="click_search",
)
inner_frame.locator(".search-component-searchBtn").click()
self._print(f"{batch_index + 1} 批查询完成,等待加载结果...")
# 等待加载完成
# 步骤4等待加载完成
self._report_progress(
"download",
batch_index * 7 + 4,
total_batches * 7,
f"{batch_index + 1}/{total_batches} 批 - 等待数据加载...",
batch_index=batch_index + 1,
action="wait_loading",
)
loading_locator = inner_frame.locator("div").filter(has_text="加载中").nth(1)
try:
loading_locator.wait_for(state="visible", timeout=3000)
@@ -91,18 +210,48 @@ class DiscreteMaterialPlanExtractor:
self._print(f"=== 调试暂停:第 {batch_index + 1} 批 ===")
page1.pause()
# 选择所有数据
# 步骤5选择所有数据
self._report_progress(
"download",
batch_index * 7 + 5,
total_batches * 7,
f"{batch_index + 1}/{total_batches} 批 - 选择所有数据行",
batch_index=batch_index + 1,
action="select_all_rows",
)
inner_frame.get_by_role("row", name="序号").get_by_label("").click()
# 步骤6配置并触发导出
self._report_progress(
"download",
batch_index * 7 + 6,
total_batches * 7,
f"{batch_index + 1}/{total_batches} 批 - 配置导出参数",
batch_index=batch_index + 1,
action="configure_export",
)
# 点击输出
inner_frame.get_by_role("button", name="更多").hover()
inner_frame.get_by_text("输出", exact=True).click()
# 设置行数阈值
input_box = inner_frame.locator("div").filter(has_text=re.compile(r"^行数阈值$")).locator("input[type='text']")
input_box = (
inner_frame.locator("div")
.filter(has_text=re.compile(r"^行数阈值$"))
.locator("input[type='text']")
)
input_box.fill("300000")
# 下载文件
# 步骤7下载文件
self._report_progress(
"download",
batch_index * 7 + 7,
total_batches * 7,
f"{batch_index + 1}/{total_batches} 批 - 正在下载文件...",
batch_index=batch_index + 1,
action="downloading_file",
)
download_path = f"D:/python/playwrite/data/temp_batch_{batch_index + 1}.xlsx"
with page1.expect_download() as download_info:
inner_frame.get_by_role("button", name="确定(Y)").click()
@@ -111,11 +260,16 @@ class DiscreteMaterialPlanExtractor:
download.save_as(download_path)
self._print(f"{batch_index + 1} 批下载完成: {download_path}")
# 关闭输出对话框(如果有的话)
# try:
# inner_frame.get_by_role("button", name="取消").click()
# except:
# pass
# 报告批次完成
self._report_progress(
"download",
(batch_index + 1) * 7,
total_batches * 7,
f"{batch_index + 1}/{total_batches} 批下载完成 ✓",
batch_index=batch_index + 1,
action="batch_complete",
file_path=download_path,
)
# 等待页面恢复,准备下一次查询
time.sleep(1)
@@ -123,7 +277,7 @@ class DiscreteMaterialPlanExtractor:
return download_path
def convert_and_merge_files(self, file_paths, output_path):
"""使用 ExcelConverter 转换并合并所有文件"""
"""使用 ExcelConverter 转换并合并所有文件,返回合并后的 DataFrame"""
# 确保输出文件路径是正确的格式
output_path = os.path.normpath(output_path)
output_dir = os.path.dirname(output_path)
@@ -133,140 +287,296 @@ class DiscreteMaterialPlanExtractor:
self._print(f"输出目录: {output_dir}")
self._print(f"输出文件名: {output_filename}")
# 确保输出文件的父目录存在
# 步骤1检查并创建输出目录
self._report_progress(
"convert",
1,
len(file_paths) * 2 + 3,
"准备转换:检查输出目录",
action="check_directory",
)
if output_dir and not os.path.exists(output_dir):
self._print(f"创建输出目录: {output_dir}")
os.makedirs(output_dir)
all_dataframes = []
# 步骤2-N转换每个文件
for i, file_path in enumerate(file_paths, 1):
self._print(f"转换第 {i} 个文件: {file_path}")
# 报告开始转换
self._report_progress(
"convert",
1 + (i - 1) * 2 + 1,
len(file_paths) * 2 + 3,
f"正在转换文件 {i}/{len(file_paths)}",
file_index=i,
file_path=file_path,
action="converting_file",
)
df = self.converter.convert(file_path, output_file=None) # 只转换,不保存
all_dataframes.append(df)
self._print(f" 提取到 {len(df)} 条记录")
# 报告转换完成
self._report_progress(
"convert",
1 + (i - 1) * 2 + 2,
len(file_paths) * 2 + 3,
f"文件 {i}/{len(file_paths)} 转换完成({len(df)} 条记录)",
file_index=i,
record_count=len(df),
action="file_converted",
)
merged_df = None
if all_dataframes:
# 步骤N+1合并数据
self._report_progress(
"convert",
len(file_paths) * 2 + 2,
len(file_paths) * 2 + 3,
f"正在合并 {len(all_dataframes)} 个文件的数据...",
action="merging_data",
file_count=len(all_dataframes),
)
self._print(f"\n合并 {len(all_dataframes)} 个文件的数据...")
merged_df = pd.concat(all_dataframes, ignore_index=True)
merged_df.to_excel(output_path, index=False)
self._print(f"合并完成: {output_path}, 总共 {len(merged_df)} 条记录")
# 删除临时文件
# 步骤N+2删除临时文件
self._report_progress(
"convert",
len(file_paths) * 2 + 3,
len(file_paths) * 2 + 3,
f"清理临时文件...",
action="cleanup",
total_records=len(merged_df),
)
for file_path in file_paths:
os.remove(file_path)
self._print(f"已删除临时文件: {file_path}")
return output_path
return None
return output_path, merged_df
return None, None
def _save_to_database(self, df: pd.DataFrame):
"""Save DataFrame to database with progress reporting"""
try:
self._report_progress(
"database", 0, 3, "准备保存到数据库...",
action="db_start"
)
stats = self.dao.save_dataframe_with_replace(df)
self._report_progress(
"database", 3, 3,
f"数据库保存完成: 删除 {stats['deleted']} 条, 新增 {stats['inserted']}",
action="db_complete",
stats=stats
)
self._print(f"\n数据库保存成功:")
self._print(f" 删除旧记录: {stats['deleted']}")
self._print(f" 新增记录: {stats['inserted']}")
except Exception as e:
self._print(f"\n警告: 数据库保存失败: {e}")
self._report_progress(
"database", 3, 3,
f"数据库保存失败: {str(e)}",
action="db_error",
error=str(e)
)
def setup_query_interface(self, inner_frame):
"""设置查询界面"""
"""设置查询界面(不报告进度,由 extract 统一报告)"""
import re
# 点击图标按钮打开查询界面
# 打开查询界面
inner_frame.locator(".search-name-wrapper > .iconfont").click()
inner_frame.get_by_text("订单号查询").click()
# 选择"全部"标签
inner_frame.get_by_role("tab", name="全部").click()
# 填充并验证,如果失败则重试
# 填充并验证
max_retries = 3
expected_value = "5000"
for attempt in range(max_retries):
inner_frame.locator("#rc_select_0").fill(expected_value)
inner_frame.locator("#rc_select_0").press("Enter")
# 检查填充是否成功
actual_value = inner_frame.locator("#rc_select_0").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 extract(self, production_id_file, data_dir="D:/python/playwrite/data",
output_file="D:/python/playwrite/data/离散备料计划维护_合并.xlsx",
debug_mode=False, debug_batch=None):
"""
执行完整的数据提取流程
Args:
production_id_file: ProductionID.txt 文件路径
data_dir: 数据保存目录
output_file: 最终输出文件路径
debug_mode: 是否启用调试模式
debug_batch: 调试批次号
Returns:
输出文件路径
"""
with sync_playwright() as playwright:
# 调用登录模块
browser, context, page, main_frame = login(
playwright=playwright,
username=self.username,
password=self.password,
headless=self.headless,
ignore_https_errors=True
)
self._print("=" * 80)
self._print("开始执行离散备料计划维护数据提取")
self._print("=" * 80)
# 登录成功后可以进行后续操作
# 点击打开"功能菜单"
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("#forwardFrame").content_frame
inner_frame_locator = main_frame.locator("#mainiframe")
inner_frame_locator.wait_for(state="visible", timeout=15000)
inner_frame = inner_frame_locator.content_frame
# 设置查询界面
self.setup_query_interface(inner_frame)
# 读取总排号并查询生产订单号
order_ids = self.get_production_order_numbers(production_id_file)
# 按批次下载
downloaded_files = []
for batch_index, order_ids_batch in enumerate(self.group_order_ids(order_ids, 100)):
self._print(f"\n=== 开始处理第 {batch_index + 1} 批,共 {len(order_ids_batch)} 个订单号 ===")
downloaded_file = self.download_batch(
inner_frame, order_ids_batch, batch_index, page1,
debug_mode=debug_mode, debug_batch=debug_batch
self._print(
f"{attempt + 1} 次填充失败,实际值: {actual_value},重试..."
)
downloaded_files.append(downloaded_file)
if attempt == max_retries - 1:
self._print(
f"警告: {max_retries} 次尝试后仍未成功填充,继续执行..."
)
# 执行账号注销
self._print("\n开始执行账号注销...")
logout(main_frame, verbose=self.verbose)
def extract(
self,
production_id_file,
data_dir="D:/python/playwrite/data",
output_file="D:/python/playwrite/data/离散备料计划维护_合并.xlsx",
debug_mode=False,
debug_batch=None,
progress_callback=None,
):
"""执行完整的数据提取流程"""
original_callback = self.progress_callback
self.progress_callback = progress_callback or self.progress_callback
# 转换并合并文件
if downloaded_files:
self._print(f"\n=== 开始转换并合并 {len(downloaded_files)} 个文件 ===")
self.convert_and_merge_files(downloaded_files, output_file)
else:
self._print("\n没有下载到任何文件")
try:
with sync_playwright() as playwright:
# 步骤1启动浏览器并登录
self._report_progress(
"login",
1,
3, # 保持 3 步
"启动浏览器并登录...",
action="launch_browser",
)
browser, context, page, main_frame = login(
playwright=playwright,
username=self.username,
password=self.password,
headless=self.headless,
ignore_https_errors=True,
)
self._print(f"\n=== 全部完成 ===")
self._print(f"最终文件: {output_file}")
# 步骤2打开功能页面
self._report_progress(
"login",
2,
3, # 保持 3 步
"登录成功,打开功能页面...",
action="open_function_page",
)
self._print("=" * 80)
self._print("开始执行离散备料计划维护数据提取")
self._print("=" * 80)
main_frame.locator("i").first.click()
# 关闭浏览器
context.close()
browser.close()
with page.expect_popup() as page1_info:
main_frame.get_by_title(
"离散备料计划维护", exact=True
).first.click()
page1 = page1_info.value
main_frame = page1.locator("#forwardFrame").content_frame
inner_frame_locator = main_frame.locator("#mainiframe")
inner_frame_locator.wait_for(state="visible", timeout=15000)
inner_frame = inner_frame_locator.content_frame
# 步骤3设置查询界面
self._report_progress(
"login",
3,
3, # 保持 3 步
"配置查询界面...",
action="setup_query_interface",
)
self.setup_query_interface(inner_frame)
# 后续代码保持不变...
order_ids = self.get_production_order_numbers(
production_id_file, report_progress=True
)
downloaded_files = []
total_batches = sum(
1 for _ in self.group_order_ids(order_ids, self.batch_size)
)
for batch_index, order_ids_batch in enumerate(
self.group_order_ids(order_ids, self.batch_size)
):
self._print(
f"\n=== 开始处理第 {batch_index + 1} 批,共 {len(order_ids_batch)} 个订单号 ==="
)
downloaded_file = self.download_batch(
inner_frame,
order_ids_batch,
batch_index,
total_batches,
page1,
debug_mode=debug_mode,
debug_batch=debug_batch,
)
downloaded_files.append(downloaded_file)
self._print("\n开始执行账号注销...")
self._report_progress(
"logout",
1,
2,
"正在注销账号...",
action="logout_start",
)
logout(main_frame, verbose=self.verbose)
self._report_progress(
"logout",
2,
2,
"注销完成 ✓",
action="logout_complete",
)
if downloaded_files:
self._print(
f"\n=== 开始转换并合并 {len(downloaded_files)} 个文件 ==="
)
output_path, merged_df = self.convert_and_merge_files(downloaded_files, output_file)
# 数据库保存步骤(独立阶段)
if self.enable_db_persistence and self.dao and merged_df is not None:
self._print(f"\n=== 开始保存数据到数据库 ===")
self._save_to_database(merged_df)
else:
self._print("\n没有下载到任何文件")
self._print(f"\n=== 全部完成 ===")
self._print(f"最终文件: {output_file}")
self._report_progress(
"complete", 1, 1, "数据提取完成 ✓",
output_file=output_file,
action="all_complete",
)
context.close()
browser.close()
return output_file
finally:
# Close database connection if open
if self.dao:
try:
self.dao.__exit__(None, None, None)
except Exception:
pass
self.progress_callback = original_callback
return output_file
def main():
"""测试函数"""
@@ -274,7 +584,7 @@ def main():
username="BLDpengqiangqiang",
password="Cqbld123456.",
headless=False,
verbose=True
verbose=True,
)
production_id_file = os.path.join(os.path.dirname(__file__), "productionID.txt")
@@ -286,4 +596,4 @@ def main():
if __name__ == "__main__":
main()
main()

View File

@@ -2,10 +2,14 @@
离散备料计划维护数据清理工具
负责登录、逐个清理订单数据
"""
import os
from playwright.sync_api import sync_playwright
from utils.auth import login, logout
from db.production_order_query import read_production_ids, query_production_order_numbers
from db.production_order_query import (
read_production_ids,
query_production_order_numbers,
)
from db.materials_to_delete import get_materials_to_delete
@@ -54,7 +58,16 @@ class DiscreteMaterialPlanCleaner:
return order_ids
def process_order(self, inner_frame, order_id, order_index, page1, materials_to_delete=None, debug_mode=False, debug_order=None):
def process_order(
self,
inner_frame,
order_id,
order_index,
page1,
materials_to_delete=None,
debug_mode=False,
debug_order=None,
):
"""清理单个订单的数据
Args:
@@ -97,9 +110,8 @@ class DiscreteMaterialPlanCleaner:
if debug_mode and (debug_order is None or order_index == debug_order):
self._print(f"=== 调试暂停:第 {order_index + 1} 个订单 ===")
page1.pause()
inner_frame.locator("#hot-key-head_list").get_by_text("更多").click()
with page1.expect_popup() as page2_info:
inner_frame.get_by_text("备料计划").click()
@@ -153,9 +165,7 @@ class DiscreteMaterialPlanCleaner:
detail_status = match.group(1)
self._print(f"备料状态: {detail_status}")
#page2.pause()
# page2.pause()
if detail_count > 0 and detail_status == "审批通过":
inner_frame.get_by_role("button", name="修改").click()
save_button_locator = inner_frame.get_by_role("button", name="保存")
@@ -163,7 +173,6 @@ class DiscreteMaterialPlanCleaner:
inner_frame.get_by_text("展开").first.click()
# 获取展开后的父容器,基于它定位子元素更加精确
# 父元素 class="card-table-side-box undefined"
child_form = inner_frame.locator(".card-table-side-box")
@@ -171,7 +180,6 @@ class DiscreteMaterialPlanCleaner:
child_form.wait_for(state="visible", timeout=5000)
self._print(f"父容器 .card-table-side-box 已找到")
page2.pause()
for id in range(detail_count):
id_lable_locator = child_form.get_by_text("序号 " + str(id + 1))
@@ -179,20 +187,37 @@ class DiscreteMaterialPlanCleaner:
self._print(f"处理 {id_lable_locator.inner_text()} ")
# 获取材料编码通过文本定位取第一个input
input_box = child_form.locator("div").filter(has_text=re.compile(r"^材料编码\d{11}$", re.MULTILINE)).locator("input").first
input_box = (
child_form.locator("div")
.filter(has_text=re.compile(r"^材料编码\d{11}$", re.MULTILINE))
.locator("input")
.first
)
self._print(f"材料编码:{input_box.input_value()}")
# 获取材料名称
input_box = child_form.locator("div").filter(has_text=re.compile(r"^材料名称$")).locator("input[type='text']")
input_box = (
child_form.locator("div")
.filter(has_text=re.compile(r"^材料名称$"))
.locator("input[type='text']")
)
material_name = input_box.input_value()
self._print(f"材料名称:{material_name}")
# 获取累计待发数量
input_box = child_form.locator("div").filter(has_text=re.compile(r"^累计待发数量$")).locator("input[type='text']")
input_box = (
child_form.locator("div")
.filter(has_text=re.compile(r"^累计待发数量$"))
.locator("input[type='text']")
)
self._print(f"累计待发数量:{input_box.input_value()}")
# 获取累计出库数量
input_box = child_form.locator("div").filter(has_text=re.compile(r"^累计出库数量$")).locator("input[type='text']")
input_box = (
child_form.locator("div")
.filter(has_text=re.compile(r"^累计出库数量$"))
.locator("input[type='text']")
)
self._print(f"累计出库数量:{input_box.input_value()}")
# 检查是否需要清理该物料
@@ -205,16 +230,22 @@ class DiscreteMaterialPlanCleaner:
break
if should_delete:
self._print(f">>> 需要清理:材料名称【{material_name}】匹配关键字【{matched_keyword}")
self._print(
f">>> 需要清理:材料名称【{material_name}】匹配关键字【{matched_keyword}"
)
# TODO: 执行删除操作
else:
self._print(f"保留:材料名称【{material_name}】无需清理")
if id != detail_count - 1:
child_form.get_by_role("button").filter(has_text=re.compile(r"^$")).nth(2).click()
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()
#page2.pause()
child_form.get_by_role("button").filter(
has_text=re.compile(r"^$")
).nth(4).click()
# page2.pause()
elif detail_count == 0:
self._print(f"{order_index + 1} 个订单无数据需要清理,跳过...")
@@ -225,11 +256,6 @@ class DiscreteMaterialPlanCleaner:
page2.close()
return
page2.close()
time.sleep(1)
pass
@@ -255,9 +281,13 @@ class DiscreteMaterialPlanCleaner:
self._print(f"文本框填充成功: {expected_value}")
break
else:
self._print(f"{attempt + 1} 次填充失败,实际值: {actual_value},重试...")
self._print(
f"{attempt + 1} 次填充失败,实际值: {actual_value},重试..."
)
if attempt == max_retries - 1:
self._print(f"警告: {max_retries} 次尝试后仍未成功填充,继续执行...")
self._print(
f"警告: {max_retries} 次尝试后仍未成功填充,继续执行..."
)
def clean(self, production_id_file, debug_mode=False, debug_order=None):
"""
@@ -280,7 +310,7 @@ class DiscreteMaterialPlanCleaner:
username=self.username,
password=self.password,
headless=self.headless,
ignore_https_errors=True
ignore_https_errors=True,
)
self._print("=" * 80)
@@ -310,10 +340,17 @@ class DiscreteMaterialPlanCleaner:
# 按订单清理
for order_index, order_id in enumerate(order_ids):
self._print(f"\n=== 开始处理第 {order_index + 1} 个订单,订单号: {order_id} ===")
self._print(
f"\n=== 开始处理第 {order_index + 1} 个订单,订单号: {order_id} ==="
)
self.process_order(
inner_frame, order_id, order_index, page1, materials_to_delete,
debug_mode=debug_mode, debug_order=debug_order
inner_frame,
order_id,
order_index,
page1,
materials_to_delete,
debug_mode=debug_mode,
debug_order=debug_order,
)
# 执行账号注销
@@ -334,7 +371,7 @@ def main():
password="Cqbld123456.",
manager_name="彭羽",
headless=False,
verbose=True
verbose=True,
)
production_id_file = os.path.join(os.path.dirname(__file__), "productionID.txt")

View File

@@ -2,6 +2,7 @@
物料状态校验脚本
校验订单中的物料状态,匹配待删除物料
"""
import os
import sys
@@ -20,7 +21,7 @@ def main():
username="BLDpengqiangqiang",
password="Cqbld123456.",
headless=True,
verbose=True
verbose=True,
)
# 设置文件路径