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>
This commit is contained in:
Misaka_Company
313
db/discrete_material_plan_dao.py
Normal file
313
db/discrete_material_plan_dao.py
Normal 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 {}
|
||||
Reference in New Issue
Block a user