feat: add material deletion confirmation with checkbox UI
- Add MaterialsToBeDeletedDAO for managing material deletion records by MaterialCode - Add MaterialValidationResult dataclass for enhanced validation results - Add CheckboxTreeview component with selectable checkbox functionality - Refactor material validation UI with new columns: select, material name, code, spec, model, manager - Add double-click to edit manager name functionality - Add select all/deselect all buttons - Add confirm deletion button to write selected records to database - Prioritize MaterialsToBeDeleted.ManagerName over type-based matching when displaying - Export results to Excel with selection state Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Misaka_Company
390
db/materials_to_be_deleted_records_dao.py
Normal file
390
db/materials_to_be_deleted_records_dao.py
Normal file
@@ -0,0 +1,390 @@
|
||||
"""
|
||||
Data Access Object for MaterialsToBeDeleted table.
|
||||
|
||||
This module provides CRUD operations for the MaterialsToBeDeleted table,
|
||||
which tracks specific material records (by MaterialCode) that need to be deleted.
|
||||
This table is different from MaterialsTypeToBeDeleted which matches by MaterialName.
|
||||
"""
|
||||
|
||||
from typing import List, Dict, Any, Set, Optional
|
||||
from db.connection import get_connection
|
||||
|
||||
|
||||
class MaterialsToBeDeletedDAO:
|
||||
"""Data Access Object for MaterialsToBeDeleted table CRUD operations
|
||||
|
||||
This table stores material records identified by MaterialCode (exact match),
|
||||
unlike MaterialsTypeToBeDeleted which uses MaterialName (partial match).
|
||||
"""
|
||||
|
||||
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.disconnect()
|
||||
|
||||
def close(self):
|
||||
"""Close database connection"""
|
||||
if self.db:
|
||||
self.db.disconnect()
|
||||
|
||||
# ==================== UPSERT (MERGE) ====================
|
||||
|
||||
def upsert_material(self, material_code: str, manager_name: str) -> bool:
|
||||
"""
|
||||
Insert or update a single material record using MERGE.
|
||||
|
||||
Args:
|
||||
material_code: Material code (exact match key)
|
||||
manager_name: Manager name
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
if not material_code or not material_code.strip():
|
||||
print("[ERROR] MaterialCode cannot be empty")
|
||||
return False
|
||||
|
||||
sql = """
|
||||
MERGE [dbo].[MaterialsToBeDeleted] AS target
|
||||
USING (SELECT ? AS MaterialCode, ? AS ManagerName) AS source
|
||||
ON (target.MaterialCode = source.MaterialCode)
|
||||
WHEN MATCHED THEN
|
||||
UPDATE SET ManagerName = source.ManagerName
|
||||
WHEN NOT MATCHED THEN
|
||||
INSERT (MaterialCode, ManagerName)
|
||||
VALUES (source.MaterialCode, source.ManagerName);
|
||||
"""
|
||||
try:
|
||||
with get_connection() as db:
|
||||
db.execute_update(sql, (material_code.strip(), manager_name.strip() if manager_name else None))
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Error upserting material: {e}")
|
||||
return False
|
||||
|
||||
def upsert_batch(self, materials: List[Dict[str, str]]) -> Dict[str, int]:
|
||||
"""
|
||||
Insert or update multiple material records in batch.
|
||||
|
||||
Args:
|
||||
materials: List of dictionaries with 'material_code' and 'manager_name' keys
|
||||
|
||||
Returns:
|
||||
Dictionary with statistics: {'total': int, 'success': int, 'failed': int}
|
||||
"""
|
||||
if not materials:
|
||||
return {'total': 0, 'success': 0, 'failed': 0}
|
||||
|
||||
sql = """
|
||||
MERGE [dbo].[MaterialsToBeDeleted] AS target
|
||||
USING (SELECT ? AS MaterialCode, ? AS ManagerName) AS source
|
||||
ON (target.MaterialCode = source.MaterialCode)
|
||||
WHEN MATCHED THEN
|
||||
UPDATE SET ManagerName = source.ManagerName
|
||||
WHEN NOT MATCHED THEN
|
||||
INSERT (MaterialCode, ManagerName)
|
||||
VALUES (source.MaterialCode, source.ManagerName);
|
||||
"""
|
||||
|
||||
stats = {'total': len(materials), 'success': 0, 'failed': 0}
|
||||
|
||||
try:
|
||||
with get_connection() as db:
|
||||
for material in materials:
|
||||
material_code = material.get('material_code', '').strip()
|
||||
manager_name = material.get('manager_name', '')
|
||||
|
||||
if not material_code:
|
||||
stats['failed'] += 1
|
||||
continue
|
||||
|
||||
try:
|
||||
db.execute_update(sql, (material_code, manager_name.strip() if manager_name else None))
|
||||
stats['success'] += 1
|
||||
except Exception as e:
|
||||
print(f"Error upserting material {material_code}: {e}")
|
||||
stats['failed'] += 1
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error in batch upsert: {e}")
|
||||
stats['failed'] = stats['total'] - stats['success']
|
||||
|
||||
return stats
|
||||
|
||||
# ==================== READ ====================
|
||||
|
||||
def get_all_material_codes(self) -> Set[str]:
|
||||
"""
|
||||
Get all material codes as a set for fast lookup.
|
||||
|
||||
Returns:
|
||||
Set of material codes
|
||||
"""
|
||||
sql = """
|
||||
SELECT [MaterialCode]
|
||||
FROM [dbo].[MaterialsToBeDeleted]
|
||||
WHERE [MaterialCode] IS NOT NULL
|
||||
"""
|
||||
try:
|
||||
with get_connection() as db:
|
||||
results = db.execute_query(sql)
|
||||
return {r['MaterialCode'] for r in results if r.get('MaterialCode')}
|
||||
except Exception as e:
|
||||
print(f"Error getting material codes: {e}")
|
||||
return set()
|
||||
|
||||
def get_all_records(self) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get all material records.
|
||||
|
||||
Returns:
|
||||
List of all material records with all fields
|
||||
"""
|
||||
sql = """
|
||||
SELECT [ID], [MaterialCode], [ManagerName]
|
||||
FROM [dbo].[MaterialsToBeDeleted]
|
||||
WHERE [MaterialCode] IS NOT NULL
|
||||
ORDER BY [ManagerName], [MaterialCode]
|
||||
"""
|
||||
with get_connection() as db:
|
||||
return db.execute_query(sql)
|
||||
|
||||
def get_materials_by_manager(self, manager_name: str) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get all materials for a specific manager.
|
||||
|
||||
Args:
|
||||
manager_name: Manager name
|
||||
|
||||
Returns:
|
||||
List of materials for the specified manager
|
||||
"""
|
||||
sql = """
|
||||
SELECT [ID], [MaterialCode], [ManagerName]
|
||||
FROM [dbo].[MaterialsToBeDeleted]
|
||||
WHERE [ManagerName] = ? AND [MaterialCode] IS NOT NULL
|
||||
ORDER BY [MaterialCode]
|
||||
"""
|
||||
with get_connection() as db:
|
||||
return db.execute_query(sql, (manager_name,))
|
||||
|
||||
def get_managers(self) -> List[str]:
|
||||
"""
|
||||
Get list of unique manager names.
|
||||
|
||||
Returns:
|
||||
List of unique manager names
|
||||
"""
|
||||
sql = """
|
||||
SELECT DISTINCT [ManagerName]
|
||||
FROM [dbo].[MaterialsToBeDeleted]
|
||||
WHERE [ManagerName] IS NOT NULL
|
||||
ORDER BY [ManagerName]
|
||||
"""
|
||||
with get_connection() as db:
|
||||
results = db.execute_query(sql)
|
||||
return [r['ManagerName'] for r in results if r.get('ManagerName')]
|
||||
|
||||
def get_record_by_material_code(self, material_code: str) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Get a specific record by material code.
|
||||
|
||||
Args:
|
||||
material_code: Material code
|
||||
|
||||
Returns:
|
||||
Dictionary representing the record, or None if not found
|
||||
"""
|
||||
sql = """
|
||||
SELECT [ID], [MaterialCode], [ManagerName]
|
||||
FROM [dbo].[MaterialsToBeDeleted]
|
||||
WHERE [MaterialCode] = ?
|
||||
"""
|
||||
with get_connection() as db:
|
||||
results = db.execute_query(sql, (material_code.strip(),))
|
||||
return results[0] if results else None
|
||||
|
||||
# ==================== DELETE ====================
|
||||
|
||||
def delete_by_material_code(self, material_code: str) -> bool:
|
||||
"""
|
||||
Delete a specific material record by material code.
|
||||
|
||||
Args:
|
||||
material_code: Material code
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
sql = """
|
||||
DELETE FROM [dbo].[MaterialsToBeDeleted]
|
||||
WHERE [MaterialCode] = ?
|
||||
"""
|
||||
try:
|
||||
with get_connection() as db:
|
||||
affected = db.execute_update(sql, (material_code.strip(),))
|
||||
return affected > 0
|
||||
except Exception as e:
|
||||
print(f"Error deleting material: {e}")
|
||||
return False
|
||||
|
||||
def delete_by_manager(self, manager_name: str) -> int:
|
||||
"""
|
||||
Delete all materials for a specific manager.
|
||||
|
||||
Args:
|
||||
manager_name: Manager name
|
||||
|
||||
Returns:
|
||||
Number of records deleted
|
||||
"""
|
||||
sql = """
|
||||
DELETE FROM [dbo].[MaterialsToBeDeleted]
|
||||
WHERE [ManagerName] = ?
|
||||
"""
|
||||
try:
|
||||
with get_connection() as db:
|
||||
return db.execute_update(sql, (manager_name,))
|
||||
except Exception as e:
|
||||
print(f"Error deleting by manager: {e}")
|
||||
return 0
|
||||
|
||||
def delete_all_materials(self) -> int:
|
||||
"""
|
||||
Delete all material records.
|
||||
|
||||
Returns:
|
||||
Number of records deleted
|
||||
"""
|
||||
sql = "DELETE FROM [dbo].[MaterialsToBeDeleted]"
|
||||
try:
|
||||
with get_connection() as db:
|
||||
return db.execute_update(sql)
|
||||
except Exception as e:
|
||||
print(f"Error deleting all materials: {e}")
|
||||
return 0
|
||||
|
||||
def delete_by_material_codes(self, material_codes: List[str]) -> int:
|
||||
"""
|
||||
Delete multiple materials by material codes.
|
||||
|
||||
Args:
|
||||
material_codes: List of material codes to delete
|
||||
|
||||
Returns:
|
||||
Number of records deleted
|
||||
"""
|
||||
if not material_codes:
|
||||
return 0
|
||||
|
||||
batch_size = 1000
|
||||
total_deleted = 0
|
||||
|
||||
for i in range(0, len(material_codes), batch_size):
|
||||
batch = material_codes[i:i + batch_size]
|
||||
placeholders = ','.join(['?' for _ in batch])
|
||||
sql = f"DELETE FROM [dbo].[MaterialsToBeDeleted] WHERE [MaterialCode] IN ({placeholders})"
|
||||
|
||||
try:
|
||||
with get_connection() as db:
|
||||
deleted = db.execute_update(sql, tuple(batch))
|
||||
total_deleted += deleted
|
||||
except Exception as e:
|
||||
print(f"Error deleting batch: {e}")
|
||||
|
||||
return total_deleted
|
||||
|
||||
# ==================== UTILITIES ====================
|
||||
|
||||
def material_exists(self, material_code: str) -> bool:
|
||||
"""
|
||||
Check if a material exists in the table.
|
||||
|
||||
Args:
|
||||
material_code: Material code
|
||||
|
||||
Returns:
|
||||
True if material exists, False otherwise
|
||||
"""
|
||||
sql = """
|
||||
SELECT COUNT(*) as count
|
||||
FROM [dbo].[MaterialsToBeDeleted]
|
||||
WHERE [MaterialCode] = ?
|
||||
"""
|
||||
with get_connection() as db:
|
||||
result = db.execute_query(sql, (material_code.strip(),))
|
||||
return result[0]['count'] > 0 if result else False
|
||||
|
||||
def count_all(self) -> int:
|
||||
"""
|
||||
Count all material records.
|
||||
|
||||
Returns:
|
||||
Total number of records
|
||||
"""
|
||||
sql = "SELECT COUNT(*) as count FROM [dbo].[MaterialsToBeDeleted]"
|
||||
with get_connection() as db:
|
||||
result = db.execute_query(sql)
|
||||
return result[0]['count'] if result else 0
|
||||
|
||||
def count_by_manager(self, manager_name: str) -> int:
|
||||
"""
|
||||
Count materials for a specific manager.
|
||||
|
||||
Args:
|
||||
manager_name: Manager name
|
||||
|
||||
Returns:
|
||||
Number of materials for the manager
|
||||
"""
|
||||
sql = """
|
||||
SELECT COUNT(*) as count
|
||||
FROM [dbo].[MaterialsToBeDeleted]
|
||||
WHERE [ManagerName] = ?
|
||||
"""
|
||||
with get_connection() as db:
|
||||
result = db.execute_query(sql, (manager_name,))
|
||||
return result[0]['count'] if result else 0
|
||||
|
||||
def get_statistics(self) -> Dict[str, Any]:
|
||||
"""
|
||||
Get comprehensive statistics about the data.
|
||||
|
||||
Returns:
|
||||
Dictionary with statistics including total materials,
|
||||
unique managers, and materials per manager
|
||||
"""
|
||||
sql = """
|
||||
SELECT
|
||||
COUNT(*) as total_materials,
|
||||
COUNT(DISTINCT ManagerName) as unique_managers
|
||||
FROM [dbo].[MaterialsToBeDeleted]
|
||||
WHERE [MaterialCode] IS NOT NULL
|
||||
"""
|
||||
with get_connection() as db:
|
||||
result = db.execute_query(sql)
|
||||
stats = result[0] if result else {}
|
||||
|
||||
# Get materials per manager
|
||||
manager_sql = """
|
||||
SELECT [ManagerName], COUNT(*) as count
|
||||
FROM [dbo].[MaterialsToBeDeleted]
|
||||
WHERE [ManagerName] IS NOT NULL
|
||||
GROUP BY [ManagerName]
|
||||
ORDER BY count DESC
|
||||
"""
|
||||
manager_results = db.execute_query(manager_sql)
|
||||
stats['materials_per_manager'] = [
|
||||
{r['ManagerName']: r['count']} for r in manager_results
|
||||
]
|
||||
|
||||
return stats
|
||||
Reference in New Issue
Block a user