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

refactor: use database for production order queries

Browse Source 
- Add db/production_order_query.py component for querying production orders
- Replace file-based orderID.txt with database-driven approach
- Read ProductionID.txt (总排号) and query [26年压力表合同数据] table
- Update both extraction and cleaning scripts to use new component
- Change parameter: order_id_file → production_id_file

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Misaka_Company
2026-01-23 16:06:40 +08:00
parent 1e2eed76e2
commit 902ff9b831
3 changed files with 100 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
db/production_order_query.py Normal file
View File

@@ -0,0 +1,50 @@
"""
生产订单号查询组件
从 ProductionID.txt 读取总排号，查询数据库获取生产订单号
"""
from db.connection import get_connection
def read_production_ids(file_path):
"""
读取 ProductionID.txt 文件，获取总排号列表
Args:
file_path: ProductionID.txt 文件路径
Returns:
总排号列表
"""
with open(file_path, 'r', encoding='utf-8') as f:
# 去除空白行和空格
production_ids = [line.strip() for line in f if line.strip()]
return production_ids
def query_production_order_numbers(production_ids):
"""
根据总排号列表，从数据库查询生产订单号
Args:
production_ids: 总排号列表
Returns:
生产订单号列表
"""
if not production_ids:
return []
# 构建 IN 子句的占位符
placeholders = ','.join(['?' for _ in production_ids])
query = f"""
SELECT [生产订单号]
FROM [productionContractData].[26年压力表合同数据]
WHERE [总排号] IN ({placeholders})
"""
with get_connection() as conn:
results = conn.execute_query(query, tuple(production_ids))
# 提取生产订单号并去除空值
production_order_numbers = [row['生产订单号'] for row in results if row['生产订单号']]
return production_order_numbers

37
utils/离散备料计划维护数据提取.py
View File

@@ -7,6 +7,7 @@ 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
class DiscreteMaterialPlanExtractor:
@@ -33,11 +34,24 @@ class DiscreteMaterialPlanExtractor:
if self.verbose:
print(*args, **kwargs)
def read_order_ids(self, file_path):
"""读取订单号文件"""
with open(file_path, 'r', encoding='utf-8') as f:
# 去除空白行和空格
order_ids = [line.strip() for line in f if line.strip()]
def get_production_order_numbers(self, production_id_file):
"""
读取总排号文件并查询数据库获取生产订单号
Args:
production_id_file: ProductionID.txt 文件路径
Returns:
生产订单号列表
"""
# 读取总排号
production_ids = read_production_ids(production_id_file)
self._print(f"从文件读取到 {len(production_ids)} 个总排号")
# 查询数据库获取生产订单号
order_ids = query_production_order_numbers(production_ids)
self._print(f"查询到 {len(order_ids)} 个生产订单号")
return order_ids
def group_order_ids(self, order_ids, group_size=100):
@@ -171,14 +185,14 @@ class DiscreteMaterialPlanExtractor:
if attempt == max_retries - 1:
self._print(f"警告: {max_retries} 次尝试后仍未成功填充，继续执行...")
def extract(self, order_id_file, data_dir="D:/python/playwrite/data",
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:
order_id_file: 订单号文件路径
production_id_file: ProductionID.txt 文件路径
data_dir: 数据保存目录
output_file: 最终输出文件路径
debug_mode: 是否启用调试模式
@@ -219,9 +233,8 @@ class DiscreteMaterialPlanExtractor:
# 设置查询界面
self.setup_query_interface(inner_frame)
# 读取订单号文件
order_ids = self.read_order_ids(order_id_file)
self._print(f"共读取到 {len(order_ids)} 个订单号")
# 读取总排号并查询生产订单号
order_ids = self.get_production_order_numbers(production_id_file)
# 按批次下载
downloaded_files = []
@@ -264,10 +277,10 @@ def main():
verbose=True
)
order_id_file = os.path.join(os.path.dirname(__file__), "orderID.txt")
production_id_file = os.path.join(os.path.dirname(__file__), "productionID.txt")
output_file = "D:/python/playwrite/data/离散备料计划维护_合并.xlsx"
extractor.extract(order_id_file, output_file)
extractor.extract(production_id_file, output_file)
input("按回车退出...")

37
utils/离散备料计划维护数据清理.py
View File

@@ -5,6 +5,7 @@
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
class DiscreteMaterialPlanCleaner:
@@ -30,11 +31,24 @@ class DiscreteMaterialPlanCleaner:
if self.verbose:
print(*args, **kwargs)
def read_order_ids(self, file_path):
"""读取订单号文件"""
with open(file_path, 'r', encoding='utf-8') as f:
# 去除空白行和空格
order_ids = [line.strip() for line in f if line.strip()]
def get_production_order_numbers(self, production_id_file):
"""
读取总排号文件并查询数据库获取生产订单号
Args:
production_id_file: ProductionID.txt 文件路径
Returns:
生产订单号列表
"""
# 读取总排号
production_ids = read_production_ids(production_id_file)
self._print(f"从文件读取到 {len(production_ids)} 个总排号")
# 查询数据库获取生产订单号
order_ids = query_production_order_numbers(production_ids)
self._print(f"查询到 {len(order_ids)} 个生产订单号")
return order_ids
def process_order(self, inner_frame, order_id, order_index, page1, debug_mode=False, debug_order=None):
@@ -223,12 +237,12 @@ class DiscreteMaterialPlanCleaner:
if attempt == max_retries - 1:
self._print(f"警告: {max_retries} 次尝试后仍未成功填充，继续执行...")
def clean(self, order_id_file, debug_mode=False, debug_order=None):
def clean(self, production_id_file, debug_mode=False, debug_order=None):
"""
执行完整的数据清理流程
Args:
order_id_file: 订单号文件路径
production_id_file: ProductionID.txt 文件路径
debug_mode: 是否启用调试模式
debug_order: 调试订单索引
"""
@@ -264,9 +278,8 @@ class DiscreteMaterialPlanCleaner:
# 设置查询界面
self.setup_query_interface(inner_frame)
# 读取订单号文件
order_ids = self.read_order_ids(order_id_file)
self._print(f"共读取到 {len(order_ids)} 个订单号")
# 读取总排号并查询生产订单号
order_ids = self.get_production_order_numbers(production_id_file)
# 按订单清理
for order_index, order_id in enumerate(order_ids):
@@ -296,9 +309,9 @@ def main():
verbose=True
)
order_id_file = os.path.join(os.path.dirname(__file__), "orderID.txt")
production_id_file = os.path.join(os.path.dirname(__file__), "productionID.txt")
cleaner.clean(order_id_file)
cleaner.clean(production_id_file)
input("按回车退出...")