1326 lines
56 KiB
Python
1326 lines
56 KiB
Python
"""
|
|
Enhanced Time Attendance Import Service with Duplicate Review
|
|
============================================================
|
|
|
|
Added functionality to detect and present duplicates for user review.
|
|
"""
|
|
|
|
import pandas as pd
|
|
import uuid
|
|
from datetime import datetime, time
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from typing import Dict, List, Any, Optional, Tuple
|
|
import traceback
|
|
import hashlib
|
|
|
|
class TimeAttendanceImportService:
|
|
"""Enhanced service with duplicate detection and review"""
|
|
|
|
def __init__(self, db, logger_handler=None):
|
|
"""Initialize the import service with database and logger"""
|
|
self.db = db
|
|
self.logger = logger_handler
|
|
|
|
def _update_progress(self, current: int, total: int, status: str = "Processing"):
|
|
"""
|
|
Update progress information for real-time tracking
|
|
|
|
Args:
|
|
current: Current record number being processed
|
|
total: Total number of records
|
|
status: Status message
|
|
"""
|
|
if hasattr(self, 'progress_callback') and self.progress_callback:
|
|
percentage = int((current / total) * 100) if total > 0 else 0
|
|
self.progress_callback({
|
|
'current': current,
|
|
'total': total,
|
|
'percentage': percentage,
|
|
'status': status
|
|
})
|
|
|
|
def _read_excel_with_formulas(self, file_path: str) -> pd.DataFrame:
|
|
"""
|
|
Read Excel file preserving HYPERLINK formulas in Recorded Address column
|
|
Uses openpyxl to extract formulas, then creates DataFrame
|
|
|
|
Args:
|
|
file_path: Path to the Excel file
|
|
|
|
Returns:
|
|
DataFrame with formulas preserved
|
|
"""
|
|
from openpyxl import load_workbook
|
|
|
|
# Load workbook with openpyxl to get formulas (data_only=False preserves formulas)
|
|
wb = load_workbook(file_path, data_only=False)
|
|
ws = wb.active
|
|
|
|
# Get header row
|
|
headers = []
|
|
for cell in ws[1]:
|
|
headers.append(cell.value)
|
|
|
|
# Find the index of 'Recorded Address' column
|
|
recorded_address_idx = None
|
|
try:
|
|
recorded_address_idx = headers.index('Recorded Address')
|
|
except ValueError:
|
|
pass # Column doesn't exist
|
|
|
|
# Read all data rows
|
|
data_rows = []
|
|
for row in ws.iter_rows(min_row=2, values_only=False):
|
|
row_data = []
|
|
for col_idx, cell in enumerate(row):
|
|
# For Recorded Address column, preserve the formula if it exists
|
|
if col_idx == recorded_address_idx and cell.value:
|
|
# Check if cell contains a formula
|
|
if isinstance(cell.value, str) and cell.value.startswith('='):
|
|
# This is a formula, keep it as-is
|
|
row_data.append(cell.value)
|
|
if self.logger:
|
|
self.logger.logger.debug(f"Found formula in Recorded Address: {cell.value[:60]}...")
|
|
else:
|
|
# Regular value
|
|
row_data.append(cell.value)
|
|
else:
|
|
# For other columns, just get the value
|
|
row_data.append(cell.value)
|
|
|
|
# Skip completely empty rows
|
|
if any(val is not None for val in row_data):
|
|
data_rows.append(row_data)
|
|
|
|
# Create DataFrame
|
|
df = pd.DataFrame(data_rows, columns=headers)
|
|
|
|
if self.logger:
|
|
self.logger.logger.info(f"Read Excel with formulas preserved: {len(df)} rows, {len(headers)} columns")
|
|
|
|
return df
|
|
|
|
def _parse_excel_hyperlink(self, cell_value: str) -> str:
|
|
"""
|
|
Parse Excel HYPERLINK formula to extract the display text (address)
|
|
|
|
Handles formats like:
|
|
- =HYPERLINK("http://maps.google.com/maps?q=38.8769894000,-77.2220616000","2815 Hartland Road, Falls Church, VA 22043")
|
|
- Regular text (no formula)
|
|
|
|
Args:
|
|
cell_value: The cell value which may contain a HYPERLINK formula
|
|
|
|
Returns:
|
|
Extracted address text or original value if not a hyperlink
|
|
"""
|
|
if not cell_value or not isinstance(cell_value, str):
|
|
return cell_value
|
|
|
|
cell_value = cell_value.strip()
|
|
|
|
# Check if it's a HYPERLINK formula
|
|
if cell_value.startswith('=HYPERLINK('):
|
|
try:
|
|
import re
|
|
|
|
# Match the display text (second quoted string)
|
|
# Pattern: =HYPERLINK("url","display_text")
|
|
pattern = r'=HYPERLINK\s*\(\s*"[^"]*"\s*,\s*"([^"]*)"\s*\)'
|
|
match = re.search(pattern, cell_value)
|
|
|
|
if match:
|
|
address_text = match.group(1).strip()
|
|
if self.logger:
|
|
self.logger.logger.debug(f"Parsed HYPERLINK: '{cell_value[:60]}...' -> '{address_text}'")
|
|
return address_text
|
|
else:
|
|
# Fallback: try to extract text between last pair of quotes
|
|
# Find all quoted strings
|
|
quoted_strings = re.findall(r'"([^"]*)"', cell_value)
|
|
if len(quoted_strings) >= 2:
|
|
# The address is typically the last quoted string
|
|
address_text = quoted_strings[-1].strip()
|
|
if self.logger:
|
|
self.logger.logger.debug(f"Parsed HYPERLINK (fallback): '{cell_value[:60]}...' -> '{address_text}'")
|
|
return address_text
|
|
else:
|
|
if self.logger:
|
|
self.logger.logger.warning(f"Could not parse HYPERLINK formula: {cell_value[:60]}...")
|
|
return None
|
|
|
|
except Exception as e:
|
|
if self.logger:
|
|
self.logger.logger.error(f"Failed to parse HYPERLINK formula: {cell_value[:60]}... Error: {e}")
|
|
return None
|
|
|
|
# Not a hyperlink formula, return as-is
|
|
return cell_value if cell_value else None
|
|
|
|
def _parse_distance_field(self, row) -> Optional[float]:
|
|
"""
|
|
Parse Distance field from Excel (optional column)
|
|
|
|
Args:
|
|
row: DataFrame row containing the 'Distance' column
|
|
|
|
Returns:
|
|
Distance value as float in miles, or None if not present/invalid
|
|
"""
|
|
# Check if Distance column exists
|
|
if 'Distance' not in row.index:
|
|
return None
|
|
|
|
distance_value = row.get('Distance')
|
|
|
|
# Check if value exists and is not NaN
|
|
if pd.isna(distance_value):
|
|
return None
|
|
|
|
# Try to parse as float
|
|
try:
|
|
# Handle string values
|
|
if isinstance(distance_value, str):
|
|
distance_str = distance_value.strip()
|
|
|
|
# Skip empty strings
|
|
if not distance_str or distance_str.lower() in ['', 'n/a', 'na', 'none']:
|
|
return None
|
|
|
|
# Remove any text like "miles", "mi", "m"
|
|
distance_str = distance_str.lower()
|
|
distance_str = distance_str.replace('miles', '').replace('mile', '').replace('mi', '').replace('m', '').strip()
|
|
|
|
# Parse the number
|
|
distance_float = float(distance_str)
|
|
else:
|
|
# Already a number
|
|
distance_float = float(distance_value)
|
|
|
|
# Validate the distance is reasonable (0 to 1000 miles)
|
|
if 0 <= distance_float <= 1000:
|
|
return distance_float
|
|
else:
|
|
if self.logger:
|
|
self.logger.logger.warning(f"Distance value {distance_float} is out of reasonable range")
|
|
return None
|
|
|
|
except (ValueError, TypeError) as e:
|
|
if self.logger:
|
|
self.logger.logger.warning(f"Could not parse distance value '{distance_value}': {e}")
|
|
return None
|
|
|
|
def _process_recorded_address(self, row) -> Optional[str]:
|
|
"""
|
|
Process Recorded Address field from Excel - handles HYPERLINK formulas
|
|
|
|
Args:
|
|
row: DataFrame row containing the 'Recorded Address' column
|
|
|
|
Returns:
|
|
Cleaned address text, or None if not present/invalid
|
|
"""
|
|
# Check if Recorded Address column exists
|
|
if 'Recorded Address' not in row.index:
|
|
return None
|
|
|
|
address_value = row.get('Recorded Address')
|
|
|
|
# Check if value exists and is not NaN
|
|
if pd.isna(address_value):
|
|
return None
|
|
|
|
# Parse HYPERLINK formula if present, or return raw value
|
|
parsed_address = self._parse_excel_hyperlink(address_value)
|
|
|
|
# Clean and return
|
|
if parsed_address:
|
|
return str(parsed_address).strip()
|
|
else:
|
|
return None
|
|
|
|
def _generate_record_hash(self, record_data: Dict[str, Any]) -> str:
|
|
"""
|
|
Generate unique hash for a time attendance record
|
|
|
|
Args:
|
|
record_data: Dictionary containing record data
|
|
|
|
Returns:
|
|
SHA-256 hash string
|
|
"""
|
|
hash_string = (
|
|
f"{record_data['employee_id']}-"
|
|
f"{record_data['attendance_date']}-"
|
|
f"{record_data['attendance_time']}-"
|
|
f"{record_data['location_name']}-"
|
|
f"{record_data['action_description']}"
|
|
)
|
|
|
|
return hashlib.sha256(hash_string.encode()).hexdigest()
|
|
|
|
def _get_existing_record_hashes(self) -> set:
|
|
"""
|
|
Get hashes of all existing time attendance records
|
|
|
|
Returns:
|
|
Set of hash strings
|
|
"""
|
|
try:
|
|
from models.time_attendance import TimeAttendance
|
|
|
|
records = TimeAttendance.query.all()
|
|
hashes = set()
|
|
|
|
for record in records:
|
|
record_data = {
|
|
'employee_id': record.employee_id,
|
|
'attendance_date': record.attendance_date,
|
|
'attendance_time': record.attendance_time,
|
|
'location_name': record.location_name,
|
|
'action_description': record.action_description
|
|
}
|
|
hashes.add(self._generate_record_hash(record_data))
|
|
|
|
return hashes
|
|
|
|
except Exception as e:
|
|
if self.logger:
|
|
self.logger.logger.error(f"Failed to get existing record hashes: {e}")
|
|
return set()
|
|
|
|
def _get_existing_record_hashes_with_data(self) -> Dict[str, Dict]:
|
|
"""
|
|
Get hashes with corresponding record data for duplicate comparison
|
|
|
|
Returns:
|
|
Dictionary mapping hash to record data
|
|
"""
|
|
try:
|
|
from models.time_attendance import TimeAttendance
|
|
|
|
records = TimeAttendance.query.all()
|
|
hash_map = {}
|
|
|
|
for record in records:
|
|
record_data = {
|
|
'employee_id': record.employee_id,
|
|
'employee_name': record.employee_name,
|
|
'attendance_date': record.attendance_date,
|
|
'attendance_time': record.attendance_time,
|
|
'location_name': record.location_name,
|
|
'action_description': record.action_description
|
|
}
|
|
record_hash = self._generate_record_hash(record_data)
|
|
hash_map[record_hash] = record_data
|
|
|
|
return hash_map
|
|
|
|
except Exception as e:
|
|
if self.logger:
|
|
self.logger.logger.error(f"Failed to get existing record hashes with data: {e}")
|
|
return {}
|
|
|
|
def analyze_for_duplicates(self, file_path: str, project_id: int = None) -> Dict[str, Any]:
|
|
"""
|
|
Analyze file for potential duplicates WITHOUT importing.
|
|
|
|
Args:
|
|
file_path: Path to the Excel file
|
|
project_id: Optional project ID — if supplied, every Location Name
|
|
in the file is validated against that project's QR code
|
|
locations before the duplicate analysis proceeds.
|
|
|
|
Returns:
|
|
Dictionary containing duplicate analysis
|
|
"""
|
|
analysis_result = {
|
|
'success': False,
|
|
'total_records': 0,
|
|
'new_records': 0,
|
|
'duplicate_records': 0,
|
|
'duplicates': [],
|
|
'errors': []
|
|
}
|
|
|
|
try:
|
|
# Read Excel file
|
|
df = self._read_excel_with_formulas(file_path)
|
|
|
|
# Validate required columns
|
|
required_columns = ['ID', 'Date', 'Time', 'Location Name', 'Action Description']
|
|
missing_columns = [col for col in required_columns if col not in df.columns]
|
|
|
|
if missing_columns:
|
|
analysis_result['errors'].append(f"Missing columns: {', '.join(missing_columns)}")
|
|
return analysis_result
|
|
|
|
# Remove empty rows
|
|
df = df.dropna(how='all')
|
|
analysis_result['total_records'] = len(df)
|
|
|
|
# ── Project-location validation ──────────────────────────────
|
|
if project_id:
|
|
try:
|
|
from models.qrcode import QRCode
|
|
from models.project import Project
|
|
file_locations = set(
|
|
str(loc).strip()
|
|
for loc in df['Location Name'].dropna().unique()
|
|
if str(loc).strip()
|
|
)
|
|
project_locations = set(
|
|
qr.location
|
|
for qr in QRCode.query.filter_by(project_id=project_id)
|
|
.with_entities(QRCode.location).all()
|
|
)
|
|
unmatched = next(
|
|
(loc for loc in sorted(file_locations) if loc not in project_locations),
|
|
None
|
|
)
|
|
if unmatched:
|
|
project_obj = self.db.session.get(Project, project_id)
|
|
project_name = project_obj.name if project_obj else f'ID {project_id}'
|
|
error_msg = (
|
|
f"The data in the file does not belong to the project '{project_name}'. "
|
|
f"Please verify the selected project or correct the file."
|
|
)
|
|
analysis_result['errors'].append(error_msg)
|
|
if self.logger:
|
|
self.logger.logger.warning(f"Project-location mismatch during duplicate analysis: {error_msg}")
|
|
return analysis_result
|
|
if self.logger:
|
|
self.logger.logger.info(
|
|
f"Project-location validation passed (duplicate analysis): "
|
|
f"all {len(file_locations)} location(s) belong to project ID {project_id}."
|
|
)
|
|
except Exception as e:
|
|
if self.logger:
|
|
self.logger.logger.warning(f"Could not perform project-location validation (duplicate analysis): {e}")
|
|
# ── End project-location validation ──────────────────────────
|
|
|
|
# Get existing record hashes
|
|
existing_hashes = self._get_existing_record_hashes_with_data()
|
|
|
|
# Process each row
|
|
duplicates_list = []
|
|
new_records_count = 0
|
|
|
|
# Process each row with enhanced validation
|
|
for index, row in df.iterrows():
|
|
# Update progress every 10 records or on last record
|
|
if (index + 1) % 10 == 0 or (index + 1) == len(df):
|
|
self._update_progress(
|
|
index + 1,
|
|
len(df),
|
|
f"Processing row {index + 2} of {len(df) + 1}"
|
|
)
|
|
|
|
try:
|
|
# Skip empty rows - only ID is required
|
|
if pd.isna(row['ID']):
|
|
continue
|
|
|
|
# Clean the employee ID first (handles float issues like '1234.0')
|
|
clean_id = self._clean_employee_id(row['ID'])
|
|
|
|
# Get employee name - either from Excel or lookup from employee table
|
|
employee_name = None
|
|
if 'Name' in df.columns and pd.notna(row.get('Name')):
|
|
employee_name = str(row['Name']).strip()
|
|
else:
|
|
# Lookup employee name from employee table using cleaned ID
|
|
employee_name = self._get_employee_name(clean_id)
|
|
|
|
# Parse date and time
|
|
attendance_date = pd.to_datetime(row['Date']).date()
|
|
attendance_time = self._parse_time_field(row['Time'])
|
|
|
|
# Prepare record data
|
|
record_data = {
|
|
'employee_id': clean_id,
|
|
'employee_name': employee_name,
|
|
'platform': str(row.get('Platform', '')).strip() if pd.notna(row.get('Platform')) else None,
|
|
'attendance_date': attendance_date,
|
|
'attendance_time': attendance_time,
|
|
'location_name': str(row['Location Name']).strip(),
|
|
'action_description': str(row['Action Description']).strip(),
|
|
'event_description': str(row.get('Event Description', '')).strip() if pd.notna(row.get('Event Description')) else None,
|
|
'recorded_address': self._process_recorded_address(row),
|
|
}
|
|
|
|
# Check for duplicates
|
|
record_hash = self._generate_record_hash(record_data)
|
|
|
|
if record_hash in existing_hashes:
|
|
# Found duplicate - get existing record details
|
|
existing_record = existing_hashes[record_hash]
|
|
|
|
duplicates_list.append({
|
|
'row_number': index + 2,
|
|
'new_record': record_data,
|
|
'existing_record': existing_record,
|
|
'hash': record_hash
|
|
})
|
|
else:
|
|
new_records_count += 1
|
|
|
|
except Exception as e:
|
|
if self.logger:
|
|
self.logger.logger.warning(f"Error analyzing row {index + 2}: {e}")
|
|
continue
|
|
|
|
analysis_result['success'] = True
|
|
analysis_result['new_records'] = new_records_count
|
|
analysis_result['duplicate_records'] = len(duplicates_list)
|
|
analysis_result['duplicates'] = duplicates_list
|
|
|
|
if self.logger:
|
|
self.logger.logger.info(
|
|
f"Duplicate analysis complete - Total: {analysis_result['total_records']}, "
|
|
f"New: {new_records_count}, Duplicates: {len(duplicates_list)}"
|
|
)
|
|
|
|
except Exception as e:
|
|
analysis_result['errors'].append(f"Analysis failed: {str(e)}")
|
|
if self.logger:
|
|
self.logger.logger.error(f"Duplicate analysis error: {e}")
|
|
|
|
return analysis_result
|
|
|
|
def analyze_for_invalid_rows(self, file_path: str, project_id: int = None) -> Dict[str, Any]:
|
|
"""
|
|
Analyze file for invalid rows with detailed error information.
|
|
|
|
Args:
|
|
file_path: Path to the Excel file
|
|
project_id: Optional project ID — if supplied, every Location Name
|
|
in the file is validated against that project's QR code
|
|
locations before the row analysis proceeds.
|
|
|
|
Returns:
|
|
Dictionary containing invalid row analysis
|
|
"""
|
|
analysis_result = {
|
|
'success': False,
|
|
'total_rows': 0,
|
|
'valid_rows': 0,
|
|
'invalid_rows': 0,
|
|
'invalid_details': [],
|
|
'errors': []
|
|
}
|
|
|
|
try:
|
|
# Read Excel file
|
|
df = self._read_excel_with_formulas(file_path)
|
|
|
|
# Validate required columns
|
|
required_columns = ['ID', 'Date', 'Time', 'Location Name', 'Action Description']
|
|
missing_columns = [col for col in required_columns if col not in df.columns]
|
|
|
|
if missing_columns:
|
|
analysis_result['errors'].append(f"Missing columns: {', '.join(missing_columns)}")
|
|
return analysis_result
|
|
|
|
# Remove empty rows
|
|
df = df.dropna(how='all')
|
|
analysis_result['total_rows'] = len(df)
|
|
|
|
# ── Project-location validation ──────────────────────────────
|
|
if project_id:
|
|
try:
|
|
from models.qrcode import QRCode
|
|
from models.project import Project
|
|
file_locations = set(
|
|
str(loc).strip()
|
|
for loc in df['Location Name'].dropna().unique()
|
|
if str(loc).strip()
|
|
)
|
|
project_locations = set(
|
|
qr.location
|
|
for qr in QRCode.query.filter_by(project_id=project_id)
|
|
.with_entities(QRCode.location).all()
|
|
)
|
|
unmatched = next(
|
|
(loc for loc in sorted(file_locations) if loc not in project_locations),
|
|
None
|
|
)
|
|
if unmatched:
|
|
project_obj = self.db.session.get(Project, project_id)
|
|
project_name = project_obj.name if project_obj else f'ID {project_id}'
|
|
error_msg = (
|
|
f"The data in the file does not belong to the project '{project_name}'. "
|
|
f"Please verify the selected project or correct the file."
|
|
)
|
|
analysis_result['errors'].append(error_msg)
|
|
if self.logger:
|
|
self.logger.logger.warning(f"Project-location mismatch during invalid-row analysis: {error_msg}")
|
|
return analysis_result
|
|
if self.logger:
|
|
self.logger.logger.info(
|
|
f"Project-location validation passed (invalid-row analysis): "
|
|
f"all {len(file_locations)} location(s) belong to project ID {project_id}."
|
|
)
|
|
except Exception as e:
|
|
if self.logger:
|
|
self.logger.logger.warning(f"Could not perform project-location validation (invalid-row analysis): {e}")
|
|
# ── End project-location validation ──────────────────────────
|
|
|
|
# Analyze each row
|
|
invalid_list = []
|
|
valid_count = 0
|
|
|
|
for index, row in df.iterrows():
|
|
row_errors = []
|
|
row_data = {
|
|
'row_number': index + 2,
|
|
'employee_id': self._clean_employee_id(row['ID']) if pd.notna(row['ID']) else None,
|
|
}
|
|
|
|
# Check ID
|
|
if pd.isna(row['ID']):
|
|
row_errors.append("Missing ID")
|
|
else:
|
|
row_data['employee_id'] = self._clean_employee_id(row['ID'])
|
|
# Get employee name
|
|
if 'Name' in df.columns and pd.notna(row.get('Name')):
|
|
row_data['employee_name'] = str(row['Name']).strip()
|
|
else:
|
|
row_data['employee_name'] = self._get_employee_name(row_data['employee_id'])
|
|
|
|
# Check and parse Date
|
|
if pd.isna(row['Date']):
|
|
row_errors.append("Missing Date")
|
|
else:
|
|
try:
|
|
attendance_date = pd.to_datetime(row['Date']).date()
|
|
row_data['attendance_date'] = attendance_date
|
|
except Exception:
|
|
row_errors.append(f"Invalid date format: {row['Date']}")
|
|
|
|
# Check and parse Time
|
|
if pd.isna(row['Time']):
|
|
row_errors.append("Missing Time")
|
|
else:
|
|
try:
|
|
attendance_time = self._parse_time_field(row['Time'])
|
|
row_data['attendance_time'] = attendance_time
|
|
except Exception as e:
|
|
row_errors.append(f"Invalid time format: {row['Time']}")
|
|
|
|
# Check Location Name
|
|
if pd.isna(row['Location Name']):
|
|
row_errors.append("Missing Location Name")
|
|
else:
|
|
row_data['location_name'] = str(row['Location Name']).strip()
|
|
|
|
# Check Action Description
|
|
if pd.isna(row['Action Description']):
|
|
row_errors.append("Missing Action Description")
|
|
else:
|
|
row_data['action_description'] = str(row['Action Description']).strip()
|
|
|
|
# Optional fields
|
|
if pd.notna(row.get('Platform')):
|
|
row_data['platform'] = str(row['Platform']).strip()
|
|
|
|
if pd.notna(row.get('Event Description')):
|
|
row_data['event_description'] = str(row['Event Description']).strip()
|
|
|
|
row_data['recorded_address'] = self._process_recorded_address(row)
|
|
|
|
# If row has errors, add to invalid list
|
|
if row_errors:
|
|
invalid_list.append({
|
|
'row_number': index + 2, # +2 for header and 0-based index
|
|
'row_data': row_data,
|
|
'errors': row_errors
|
|
})
|
|
else:
|
|
valid_count += 1
|
|
|
|
analysis_result['success'] = True
|
|
analysis_result['valid_rows'] = valid_count
|
|
analysis_result['invalid_rows'] = len(invalid_list)
|
|
analysis_result['invalid_details'] = invalid_list
|
|
|
|
if self.logger:
|
|
self.logger.logger.info(
|
|
f"Invalid row analysis complete - Total: {analysis_result['total_rows']}, "
|
|
f"Valid: {valid_count}, Invalid: {len(invalid_list)}"
|
|
)
|
|
|
|
except Exception as e:
|
|
analysis_result['errors'].append(f"Analysis failed: {str(e)}")
|
|
if self.logger:
|
|
self.logger.logger.error(f"Invalid row analysis error: {e}")
|
|
|
|
return analysis_result
|
|
|
|
def import_from_excel(self, file_path: str, created_by: int = None,
|
|
import_source: str = None, skip_duplicates: bool = True,
|
|
force_import_hashes: List[str] = None, project_id: int = None,
|
|
progress_callback=None) -> Dict[str, Any]:
|
|
"""
|
|
Import time attendance data from Excel file with enhanced duplicate handling.
|
|
|
|
Args:
|
|
file_path: Path to the Excel file
|
|
created_by: User ID who initiated the import
|
|
import_source: Description of import source
|
|
skip_duplicates: Whether to skip duplicate records
|
|
force_import_hashes: List of hashes to force import (user confirmed duplicates)
|
|
progress_callback: Optional callable(current, total, message) for real-time progress
|
|
|
|
Returns:
|
|
Dictionary containing import results
|
|
"""
|
|
batch_id = str(uuid.uuid4())
|
|
import_results = {
|
|
'batch_id': batch_id,
|
|
'total_records': 0,
|
|
'imported_records': 0,
|
|
'failed_records': 0,
|
|
'duplicate_records': 0,
|
|
'skipped_records': 0,
|
|
'forced_duplicates': 0,
|
|
'errors': [],
|
|
'warnings': [],
|
|
'success': False,
|
|
'import_date': datetime.utcnow()
|
|
}
|
|
|
|
force_import_hashes = force_import_hashes or []
|
|
|
|
try:
|
|
# Log import start
|
|
if self.logger:
|
|
self.logger.logger.info(
|
|
f"Starting enhanced time attendance import from {file_path} by user {created_by} "
|
|
f"(skip_duplicates={skip_duplicates}, force_import={len(force_import_hashes)})"
|
|
)
|
|
|
|
# Read Excel file
|
|
try:
|
|
df = self._read_excel_with_formulas(file_path)
|
|
|
|
if self.logger:
|
|
self.logger.logger.info(f"Read Excel file with {len(df)} rows and formulas preserved")
|
|
except Exception as e:
|
|
error_msg = f"Failed to read Excel file: {str(e)}"
|
|
import_results['errors'].append(error_msg)
|
|
if self.logger:
|
|
self.logger.logger.error(error_msg)
|
|
return import_results
|
|
|
|
# Validate required columns
|
|
required_columns = ['ID', 'Date', 'Time', 'Location Name', 'Action Description']
|
|
missing_columns = [col for col in required_columns if col not in df.columns]
|
|
|
|
if missing_columns:
|
|
error_msg = f"Missing required columns: {', '.join(missing_columns)}"
|
|
import_results['errors'].append(error_msg)
|
|
if self.logger:
|
|
self.logger.logger.error(error_msg)
|
|
return import_results
|
|
|
|
# Remove completely empty rows
|
|
df = df.dropna(how='all')
|
|
import_results['total_records'] = len(df)
|
|
|
|
if import_results['total_records'] == 0:
|
|
error_msg = "No valid data rows found in Excel file"
|
|
import_results['errors'].append(error_msg)
|
|
return import_results
|
|
|
|
# ── Project-location validation ──────────────────────────────────────
|
|
# If a project_id is provided, verify that every unique Location Name
|
|
# in the file belongs to that project. Return immediately on the first
|
|
# mismatch so the user can correct the file or project selection.
|
|
if project_id:
|
|
try:
|
|
from models.qrcode import QRCode
|
|
from models.project import Project
|
|
|
|
# Collect all unique, non-empty location names from the file
|
|
file_locations = set(
|
|
str(loc).strip()
|
|
for loc in df['Location Name'].dropna().unique()
|
|
if str(loc).strip()
|
|
)
|
|
|
|
# Fetch all location names that belong to the selected project.
|
|
# Strip each value — DB entries may have trailing whitespace.
|
|
project_locations = set(
|
|
qr.location.strip()
|
|
for qr in QRCode.query.filter_by(project_id=project_id)
|
|
.with_entities(QRCode.location).all()
|
|
if qr.location
|
|
)
|
|
|
|
if self.logger:
|
|
self.logger.logger.info(
|
|
f"Project-location validation: project_id={project_id}, "
|
|
f"file_locations={sorted(file_locations)}, "
|
|
f"project_locations={sorted(project_locations)}"
|
|
)
|
|
|
|
# Find the first location in the file that is not in the project
|
|
unmatched = next(
|
|
(loc for loc in sorted(file_locations) if loc not in project_locations),
|
|
None
|
|
)
|
|
|
|
if unmatched:
|
|
project_obj = self.db.session.get(Project, project_id)
|
|
project_name = project_obj.name if project_obj else f'ID {project_id}'
|
|
error_msg = (
|
|
f"The data in the file does not belong to the project '{project_name}'. "
|
|
f"Location '{unmatched}' was not found in this project. "
|
|
f"Please verify the selected project or correct the file."
|
|
)
|
|
import_results['errors'].append(error_msg)
|
|
if self.logger:
|
|
self.logger.logger.warning(
|
|
f"Project-location mismatch: {error_msg}"
|
|
)
|
|
return import_results
|
|
|
|
if self.logger:
|
|
self.logger.logger.info(
|
|
f"Project-location validation passed: all {len(file_locations)} "
|
|
f"location(s) belong to project ID {project_id}."
|
|
)
|
|
|
|
except Exception as e:
|
|
# Fail closed: if validation cannot be performed, block the import.
|
|
# This prevents a DB or import error from silently bypassing the check.
|
|
error_msg = f"Project-location validation could not be completed: {e}"
|
|
import_results['errors'].append(error_msg)
|
|
if self.logger:
|
|
self.logger.logger.error(
|
|
f"Project-location validation error (failing closed): {e}",
|
|
exc_info=True
|
|
)
|
|
return import_results
|
|
# ── End project-location validation ──────────────────────────────────
|
|
|
|
# Track duplicates using hash
|
|
duplicate_hashes = set()
|
|
if skip_duplicates:
|
|
duplicate_hashes = self._get_existing_record_hashes()
|
|
|
|
# Process each row with enhanced validation
|
|
for index, row in df.iterrows():
|
|
try:
|
|
# Skip empty rows - only ID is required
|
|
if pd.isna(row['ID']):
|
|
import_results['skipped_records'] += 1
|
|
import_results['warnings'].append(f"Row {index + 2}: Skipped due to missing ID")
|
|
continue
|
|
|
|
# Clean the employee ID first (handles float issues like '1234.0')
|
|
clean_id = self._clean_employee_id(row['ID'])
|
|
|
|
# Get employee name - either from Excel or lookup from employee table
|
|
employee_name = None
|
|
if 'Name' in df.columns and pd.notna(row.get('Name')):
|
|
employee_name = str(row['Name']).strip()
|
|
else:
|
|
# Lookup employee name from employee table using cleaned ID
|
|
employee_name = self._get_employee_name(clean_id)
|
|
|
|
# Validate and parse date
|
|
try:
|
|
attendance_date = pd.to_datetime(row['Date']).date()
|
|
except Exception as date_error:
|
|
import_results['failed_records'] += 1
|
|
import_results['errors'].append(f"Row {index + 2}: Invalid date format - {str(date_error)}")
|
|
continue
|
|
|
|
# Validate and parse time
|
|
try:
|
|
attendance_time = self._parse_time_field(row['Time'])
|
|
except Exception as time_error:
|
|
import_results['failed_records'] += 1
|
|
import_results['errors'].append(f"Row {index + 2}: Invalid time format - {str(time_error)}")
|
|
continue
|
|
|
|
# Prepare record data
|
|
record_data = {
|
|
'employee_id': clean_id,
|
|
'employee_name': employee_name,
|
|
'platform': str(row.get('Platform', '')).strip() if pd.notna(row.get('Platform')) else None,
|
|
'attendance_date': attendance_date,
|
|
'attendance_time': attendance_time,
|
|
'location_name': str(row['Location Name']).strip(),
|
|
'action_description': str(row['Action Description']).strip(),
|
|
'event_description': str(row.get('Event Description', '')).strip() if pd.notna(row.get('Event Description')) else None,
|
|
'recorded_address': self._process_recorded_address(row),
|
|
'distance': self._parse_distance_field(row),
|
|
}
|
|
|
|
# Check for duplicates
|
|
if skip_duplicates:
|
|
record_hash = self._generate_record_hash(record_data)
|
|
|
|
# If duplicate and NOT in force import list, skip it
|
|
if record_hash in duplicate_hashes and record_hash not in force_import_hashes:
|
|
import_results['duplicate_records'] += 1
|
|
import_results['warnings'].append(
|
|
f"Row {index + 2}: Duplicate record for {record_data['employee_name']} "
|
|
f"on {attendance_date} at {attendance_time} - Skipped"
|
|
)
|
|
continue
|
|
|
|
# If in force import list, track it
|
|
if record_hash in force_import_hashes:
|
|
import_results['forced_duplicates'] += 1
|
|
|
|
duplicate_hashes.add(record_hash)
|
|
|
|
# Create TimeAttendance record
|
|
from models.time_attendance import TimeAttendance
|
|
|
|
time_attendance_record = TimeAttendance(
|
|
**record_data,
|
|
import_batch_id=batch_id,
|
|
import_source=import_source or f"Excel Import - {file_path}",
|
|
created_by=created_by,
|
|
project_id=project_id
|
|
)
|
|
|
|
self.db.session.add(time_attendance_record)
|
|
import_results['imported_records'] += 1
|
|
|
|
# Commit in batches of 50 to prevent memory buildup on large files
|
|
if import_results['imported_records'] % 50 == 0:
|
|
self.db.session.commit()
|
|
|
|
# Report progress via callback if provided
|
|
if progress_callback:
|
|
processed = (import_results['imported_records'] +
|
|
import_results['failed_records'] +
|
|
import_results['duplicate_records'] +
|
|
import_results['skipped_records'])
|
|
progress_callback(
|
|
processed,
|
|
import_results['total_records'],
|
|
f"Importing record {processed} of {import_results['total_records']}..."
|
|
)
|
|
|
|
except Exception as e:
|
|
import_results['failed_records'] += 1
|
|
error_msg = f"Row {index + 2}: {str(e)}"
|
|
import_results['errors'].append(error_msg)
|
|
|
|
if self.logger:
|
|
self.logger.logger.warning(f"Failed to import row {index + 2}: {e}")
|
|
|
|
continue
|
|
|
|
# Final commit
|
|
self.db.session.commit()
|
|
import_results['success'] = import_results['imported_records'] > 0
|
|
|
|
# Log successful import
|
|
if self.logger:
|
|
self.logger.logger.info(
|
|
f"Time attendance import completed - Batch: {batch_id}, "
|
|
f"Total: {import_results['total_records']}, "
|
|
f"Imported: {import_results['imported_records']}, "
|
|
f"Failed: {import_results['failed_records']}, "
|
|
f"Duplicates: {import_results['duplicate_records']}, "
|
|
f"Forced: {import_results['forced_duplicates']}, "
|
|
f"Skipped: {import_results['skipped_records']}"
|
|
)
|
|
|
|
except SQLAlchemyError as e:
|
|
self.db.session.rollback()
|
|
error_msg = f"Database error during import: {str(e)}"
|
|
import_results['errors'].append(error_msg)
|
|
|
|
if self.logger:
|
|
self.logger.log_database_error('time_attendance_import', e)
|
|
|
|
except Exception as e:
|
|
self.db.session.rollback()
|
|
error_msg = f"Unexpected error during import: {str(e)}"
|
|
import_results['errors'].append(error_msg)
|
|
import_results['traceback'] = traceback.format_exc()
|
|
|
|
if self.logger:
|
|
self.logger.logger.error(f"Time attendance import failed: {e}")
|
|
self.logger.logger.error(f"Traceback: {traceback.format_exc()}")
|
|
|
|
return import_results
|
|
|
|
def _parse_time_field(self, time_value) -> time:
|
|
"""
|
|
Parse various time formats from Excel
|
|
|
|
Handles:
|
|
- datetime.time objects
|
|
- datetime.datetime objects
|
|
- String formats (HH:MM, HH:MM:SS, HH:MM AM/PM)
|
|
|
|
Args:
|
|
time_value: Time value from Excel
|
|
|
|
Returns:
|
|
time object
|
|
"""
|
|
if pd.isna(time_value):
|
|
raise ValueError("Time value is empty")
|
|
|
|
# If already a time object
|
|
if isinstance(time_value, time):
|
|
return time_value
|
|
|
|
# If datetime object, extract time
|
|
if isinstance(time_value, datetime):
|
|
return time_value.time()
|
|
|
|
# If string, parse it
|
|
if isinstance(time_value, str):
|
|
time_str = time_value.strip()
|
|
|
|
# Try parsing with pandas
|
|
try:
|
|
dt = pd.to_datetime(time_str)
|
|
return dt.time()
|
|
except:
|
|
# Try manual parsing for common formats
|
|
try:
|
|
# Format: HH:MM or HH:MM:SS
|
|
parts = time_str.split(':')
|
|
if len(parts) >= 2:
|
|
hour = int(parts[0])
|
|
minute = int(parts[1])
|
|
second = int(parts[2]) if len(parts) > 2 else 0
|
|
return time(hour, minute, second)
|
|
except:
|
|
pass
|
|
|
|
raise ValueError(f"Could not parse time value: {time_value}")
|
|
|
|
def _get_employee_name(self, employee_id: str) -> str:
|
|
"""
|
|
Get employee name from database, formatted as 'lastname, firstname'
|
|
|
|
Args:
|
|
employee_id: Employee ID to lookup
|
|
|
|
Returns:
|
|
Employee name in 'lastname, firstname' format
|
|
"""
|
|
try:
|
|
from models.employee import Employee
|
|
|
|
# CRITICAL: Clean the employee_id to handle float values like '1234.0'
|
|
# Remove '.0' suffix if present and convert to integer
|
|
cleaned_id = str(employee_id).strip()
|
|
|
|
# If it's a float string like '1234.0', remove the decimal part
|
|
if '.' in cleaned_id:
|
|
try:
|
|
# Convert to float first, then to int to handle '1234.0' -> 1234
|
|
cleaned_id = str(int(float(cleaned_id)))
|
|
except (ValueError, TypeError):
|
|
pass # Keep original if conversion fails
|
|
|
|
# Now lookup the employee
|
|
employee = Employee.get_by_employee_id(int(cleaned_id))
|
|
if employee:
|
|
# Format name as "lastname, firstname"
|
|
return f"{employee.lastName}, {employee.firstName}"
|
|
else:
|
|
if self.logger:
|
|
self.logger.logger.warning(f"Employee ID {cleaned_id} not found in employee table")
|
|
return f"Employee {cleaned_id}"
|
|
except Exception as e:
|
|
if self.logger:
|
|
self.logger.logger.warning(f"Could not lookup employee name for ID {employee_id}: {e}")
|
|
# Return cleaned ID in error message too
|
|
try:
|
|
cleaned_id = str(int(float(str(employee_id).strip())))
|
|
return f"Employee {cleaned_id}"
|
|
except:
|
|
return f"Employee {employee_id}"
|
|
|
|
def _clean_employee_id(self, employee_id) -> str:
|
|
"""
|
|
Clean employee ID to handle various formats
|
|
|
|
Args:
|
|
employee_id: Raw employee ID value
|
|
|
|
Returns:
|
|
Cleaned employee ID string
|
|
"""
|
|
try:
|
|
# Convert to string first
|
|
id_str = str(employee_id).strip()
|
|
|
|
# Handle float values like 1234.0 or '1234.0'
|
|
if '.' in id_str:
|
|
# Convert to float, then to int, then back to string
|
|
# This removes the decimal part: 1234.0 -> 1234
|
|
id_str = str(int(float(id_str)))
|
|
|
|
return id_str
|
|
except (ValueError, TypeError) as e:
|
|
# If conversion fails, return original string
|
|
if self.logger:
|
|
self.logger.logger.warning(f"Could not clean employee ID '{employee_id}': {e}")
|
|
return str(employee_id).strip()
|
|
|
|
def validate_excel_file(self, file_path: str) -> Dict[str, Any]:
|
|
"""
|
|
Validate Excel file structure and content before import
|
|
|
|
Args:
|
|
file_path: Path to the Excel file
|
|
|
|
Returns:
|
|
Dictionary containing validation results
|
|
"""
|
|
validation_results = {
|
|
'valid': False,
|
|
'errors': [],
|
|
'warnings': [],
|
|
'total_rows': 0,
|
|
'valid_rows': 0,
|
|
'invalid_rows': 0
|
|
}
|
|
|
|
try:
|
|
# Try to read the Excel file
|
|
df = pd.read_excel(file_path)
|
|
validation_results['total_rows'] = len(df)
|
|
|
|
# Check required columns
|
|
required_columns = ['ID', 'Date', 'Time', 'Location Name', 'Action Description']
|
|
missing_columns = [col for col in required_columns if col not in df.columns]
|
|
|
|
if missing_columns:
|
|
validation_results['errors'].append(
|
|
f"Missing required columns: {', '.join(missing_columns)}"
|
|
)
|
|
|
|
# Check optional columns
|
|
optional_columns = ['Name', 'Platform', 'Event Description', 'Recorded Address', 'Distance']
|
|
present_optional = [col for col in optional_columns if col in df.columns]
|
|
|
|
if present_optional:
|
|
validation_results['warnings'].append(
|
|
f"Optional columns found: {', '.join(present_optional)}"
|
|
)
|
|
|
|
# Validate data in rows
|
|
if not missing_columns:
|
|
valid_row_count = 0
|
|
invalid_row_details = []
|
|
|
|
for index, row in df.iterrows():
|
|
is_valid = True
|
|
missing_fields = []
|
|
|
|
for col in required_columns:
|
|
if col in df.columns:
|
|
if pd.isna(row[col]):
|
|
is_valid = False
|
|
missing_fields.append(col)
|
|
else:
|
|
# Column doesn't exist in file
|
|
is_valid = False
|
|
missing_fields.append(f"{col} (column not found)")
|
|
|
|
if is_valid:
|
|
valid_row_count += 1
|
|
else:
|
|
# Track invalid row for detailed reporting
|
|
invalid_row_details.append({
|
|
'row': index + 2, # +2 for header and 0-based index
|
|
'missing': missing_fields
|
|
})
|
|
|
|
validation_results['valid_rows'] = valid_row_count
|
|
validation_results['invalid_rows'] = len(df) - valid_row_count
|
|
|
|
if validation_results['invalid_rows'] > 0:
|
|
# Provide detailed warning about invalid rows
|
|
validation_results['warnings'].append(
|
|
f"{validation_results['invalid_rows']} rows have missing required data"
|
|
)
|
|
|
|
# Add details about first few invalid rows for debugging
|
|
if invalid_row_details:
|
|
sample_invalid = invalid_row_details[:3] # Show first 3 invalid rows
|
|
details_msg = "Examples: "
|
|
for detail in sample_invalid:
|
|
details_msg += f"Row {detail['row']} (missing: {', '.join(detail['missing'])}); "
|
|
validation_results['warnings'].append(details_msg.rstrip('; '))
|
|
|
|
if 'Date' in df.columns:
|
|
invalid_dates = 0
|
|
for idx, date_val in df['Date'].items():
|
|
if pd.notna(date_val):
|
|
try:
|
|
pd.to_datetime(date_val)
|
|
except:
|
|
invalid_dates += 1
|
|
|
|
if invalid_dates > 0:
|
|
validation_results['warnings'].append(
|
|
f"{invalid_dates} rows have invalid date format"
|
|
)
|
|
|
|
if 'Time' in df.columns:
|
|
invalid_times = 0
|
|
for idx, time_val in df['Time'].items():
|
|
if pd.notna(time_val):
|
|
try:
|
|
self._parse_time_field(time_val)
|
|
except:
|
|
invalid_times += 1
|
|
|
|
if invalid_times > 0:
|
|
validation_results['warnings'].append(
|
|
f"{invalid_times} rows have invalid time format"
|
|
)
|
|
|
|
if all(col in df.columns for col in ['ID', 'Date', 'Time', 'Location Name']):
|
|
duplicate_check = df[['ID', 'Date', 'Time', 'Location Name']].duplicated()
|
|
duplicate_count = duplicate_check.sum()
|
|
|
|
if duplicate_count > 0:
|
|
validation_results['warnings'].append(
|
|
f"{duplicate_count} potential duplicate records detected"
|
|
)
|
|
|
|
validation_results['valid'] = (
|
|
len(validation_results['errors']) == 0 and
|
|
validation_results['valid_rows'] > 0
|
|
)
|
|
|
|
except Exception as e:
|
|
validation_results['errors'].append(f"Failed to validate Excel file: {str(e)}")
|
|
if self.logger:
|
|
self.logger.logger.error(f"Validation error: {e}")
|
|
|
|
return validation_results
|
|
|
|
def get_import_summary(self, batch_id: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Get detailed summary of imported data by batch ID
|
|
|
|
Args:
|
|
batch_id: Import batch identifier
|
|
|
|
Returns:
|
|
Dictionary containing import summary
|
|
"""
|
|
try:
|
|
from models.time_attendance import TimeAttendance
|
|
|
|
records = TimeAttendance.get_by_import_batch(batch_id)
|
|
|
|
if not records:
|
|
return None
|
|
|
|
# Calculate summary statistics
|
|
total_records = len(records)
|
|
unique_employees = len(set(record.employee_id for record in records))
|
|
unique_locations = len(set(record.location_name for record in records))
|
|
date_range = {
|
|
'start': min(record.attendance_date for record in records),
|
|
'end': max(record.attendance_date for record in records)
|
|
}
|
|
|
|
# Group by action description
|
|
actions = {}
|
|
for record in records:
|
|
action = record.action_description
|
|
actions[action] = actions.get(action, 0) + 1
|
|
|
|
# Group by employee
|
|
employee_summary = {}
|
|
for record in records:
|
|
emp_id = record.employee_id
|
|
if emp_id not in employee_summary:
|
|
employee_summary[emp_id] = {
|
|
'name': record.employee_name,
|
|
'count': 0
|
|
}
|
|
employee_summary[emp_id]['count'] += 1
|
|
|
|
return {
|
|
'batch_id': batch_id,
|
|
'total_records': total_records,
|
|
'unique_employees': unique_employees,
|
|
'unique_locations': unique_locations,
|
|
'date_range': date_range,
|
|
'actions': actions,
|
|
'employee_summary': employee_summary,
|
|
'import_date': records[0].import_date if records else None,
|
|
'import_source': records[0].import_source if records else None
|
|
}
|
|
|
|
except Exception as e:
|
|
if self.logger:
|
|
self.logger.logger.error(f"Failed to get import summary for batch {batch_id}: {e}")
|
|
return None
|
|
|
|
def delete_import_batch(self, batch_id: str, deleted_by: int = None) -> Dict[str, Any]:
|
|
"""
|
|
Delete all records from a specific import batch
|
|
|
|
Args:
|
|
batch_id: Import batch identifier
|
|
deleted_by: User ID who initiated the deletion
|
|
|
|
Returns:
|
|
Dictionary containing deletion results
|
|
"""
|
|
result = {
|
|
'success': False,
|
|
'deleted_count': 0,
|
|
'message': ''
|
|
}
|
|
|
|
try:
|
|
from models.time_attendance import TimeAttendance
|
|
|
|
records = TimeAttendance.query.filter_by(import_batch_id=batch_id).all()
|
|
deleted_count = len(records)
|
|
|
|
if deleted_count == 0:
|
|
result['message'] = 'No records found for this batch'
|
|
return result
|
|
|
|
# Delete records
|
|
for record in records:
|
|
self.db.session.delete(record)
|
|
|
|
self.db.session.commit()
|
|
|
|
# Log deletion
|
|
if self.logger:
|
|
self.logger.logger.info(
|
|
f"User {deleted_by} deleted import batch {batch_id} - "
|
|
f"Removed {deleted_count} records"
|
|
)
|
|
|
|
result['success'] = True
|
|
result['deleted_count'] = deleted_count
|
|
result['message'] = f'Successfully deleted {deleted_count} records'
|
|
|
|
except Exception as e:
|
|
self.db.session.rollback()
|
|
result['message'] = f'Error deleting batch: {str(e)}'
|
|
|
|
if self.logger:
|
|
self.logger.logger.error(f"Failed to delete batch {batch_id}: {e}")
|
|
|
|
return result |