392 lines
17 KiB
Python
392 lines
17 KiB
Python
"""
|
|
QR Code Import Service
|
|
=====================
|
|
|
|
Service for handling bulk QR code imports from Excel files.
|
|
Follows the same patterns as TimeAttendanceImportService.
|
|
|
|
IMPORTANT: This service does NOT import any models.
|
|
All model classes must be passed as parameters from app.py.
|
|
"""
|
|
|
|
import pandas as pd
|
|
import uuid
|
|
from datetime import datetime
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from typing import Dict, List, Any, Optional, Tuple
|
|
import traceback
|
|
|
|
|
|
class QRCodeImportService:
|
|
"""Service for bulk QR code import from Excel files"""
|
|
|
|
def __init__(self, db, logger_handler=None):
|
|
"""Initialize the import service with database and logger"""
|
|
self.db = db
|
|
self.logger = logger_handler
|
|
|
|
def validate_excel_file(self, file_path: str) -> Dict[str, Any]:
|
|
"""
|
|
Validate Excel file structure and data
|
|
|
|
Args:
|
|
file_path: Path to the Excel file
|
|
|
|
Returns:
|
|
Dictionary with validation results
|
|
"""
|
|
try:
|
|
# Read Excel file
|
|
df = pd.read_excel(file_path)
|
|
|
|
# Required columns
|
|
required_columns = [
|
|
'QR Code Name',
|
|
'QR Code Location',
|
|
'Project',
|
|
'Location Address',
|
|
'Event'
|
|
]
|
|
|
|
# Optional columns
|
|
optional_columns = [
|
|
'Latitude',
|
|
'Longitude'
|
|
]
|
|
|
|
# Check for required columns
|
|
missing_columns = []
|
|
for col in required_columns:
|
|
if col not in df.columns:
|
|
missing_columns.append(col)
|
|
|
|
if missing_columns:
|
|
return {
|
|
'success': False,
|
|
'error': f"Missing required columns: {', '.join(missing_columns)}",
|
|
'missing_columns': missing_columns
|
|
}
|
|
|
|
# Validate data
|
|
errors = []
|
|
warnings = []
|
|
valid_rows = []
|
|
invalid_rows = []
|
|
|
|
for index, row in df.iterrows():
|
|
row_num = index + 2 # Excel row number (header is row 1)
|
|
row_errors = []
|
|
|
|
# Validate QR Code Name
|
|
if pd.isna(row['QR Code Name']) or str(row['QR Code Name']).strip() == '':
|
|
row_errors.append(f"Row {row_num}: QR Code Name is required")
|
|
|
|
# Validate Location
|
|
if pd.isna(row['QR Code Location']) or str(row['QR Code Location']).strip() == '':
|
|
row_errors.append(f"Row {row_num}: QR Code Location is required")
|
|
|
|
# Validate Location Address
|
|
if pd.isna(row['Location Address']) or str(row['Location Address']).strip() == '':
|
|
row_errors.append(f"Row {row_num}: Location Address is required")
|
|
|
|
# Validate Event
|
|
if pd.isna(row['Event']) or str(row['Event']).strip() == '':
|
|
row_errors.append(f"Row {row_num}: Event is required")
|
|
|
|
# Validate Project (must be a string)
|
|
if pd.isna(row['Project']) or str(row['Project']).strip() == '':
|
|
row_errors.append(f"Row {row_num}: Project is required")
|
|
|
|
# Validate GPS coordinates if provided
|
|
has_latitude = 'Latitude' in df.columns and not pd.isna(row.get('Latitude'))
|
|
has_longitude = 'Longitude' in df.columns and not pd.isna(row.get('Longitude'))
|
|
|
|
if has_latitude and has_longitude:
|
|
try:
|
|
lat = float(row['Latitude'])
|
|
lon = float(row['Longitude'])
|
|
|
|
# Validate latitude range
|
|
if not (-90 <= lat <= 90):
|
|
row_errors.append(f"Row {row_num}: Latitude must be between -90 and 90")
|
|
|
|
# Validate longitude range
|
|
if not (-180 <= lon <= 180):
|
|
row_errors.append(f"Row {row_num}: Longitude must be between -180 and 180")
|
|
except (ValueError, TypeError):
|
|
row_errors.append(f"Row {row_num}: Invalid GPS coordinates format")
|
|
elif has_latitude or has_longitude:
|
|
warnings.append(f"Row {row_num}: Both Latitude and Longitude must be provided together")
|
|
|
|
if row_errors:
|
|
errors.extend(row_errors)
|
|
invalid_rows.append({
|
|
'row_number': row_num,
|
|
'data': row.to_dict(),
|
|
'errors': row_errors
|
|
})
|
|
else:
|
|
valid_rows.append({
|
|
'row_number': row_num,
|
|
'data': row.to_dict()
|
|
})
|
|
|
|
return {
|
|
'success': len(errors) == 0,
|
|
'total_rows': len(df),
|
|
'valid_rows': len(valid_rows),
|
|
'invalid_rows': len(invalid_rows),
|
|
'errors': errors,
|
|
'warnings': warnings,
|
|
'valid_data': valid_rows,
|
|
'invalid_data': invalid_rows
|
|
}
|
|
|
|
except Exception as e:
|
|
if self.logger:
|
|
self.logger.logger.error(f"Error validating Excel file: {e}")
|
|
return {
|
|
'success': False,
|
|
'error': f"Error reading Excel file: {str(e)}",
|
|
'total_rows': 0,
|
|
'valid_rows': 0,
|
|
'invalid_rows': 0,
|
|
'errors': [str(e)],
|
|
'warnings': []
|
|
}
|
|
|
|
def import_from_excel(
|
|
self,
|
|
file_path: str,
|
|
created_by: int,
|
|
generate_qr_code_func,
|
|
generate_qr_url_func,
|
|
request_url_root: str,
|
|
project_lookup: Dict[str, int] = None,
|
|
QRCode=None,
|
|
Project=None,
|
|
geocode_func=None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Import QR codes from Excel file
|
|
|
|
Args:
|
|
file_path: Path to the Excel file
|
|
created_by: User ID who initiated the import
|
|
generate_qr_code_func: Function to generate QR code image
|
|
generate_qr_url_func: Function to generate QR URL
|
|
request_url_root: Base URL for QR code destination
|
|
project_lookup: Dictionary mapping project names to IDs
|
|
QRCode: QRCode model class (passed from app.py)
|
|
Project: Project model class (passed from app.py)
|
|
geocode_func: Function to geocode addresses (optional, for auto-geocoding)
|
|
|
|
Returns:
|
|
Dictionary with import results
|
|
"""
|
|
# Models are now passed as parameters to avoid import issues
|
|
|
|
# Validate that model classes were passed
|
|
if QRCode is None or Project is None:
|
|
return {
|
|
'success': False,
|
|
'error': 'Model classes not provided. Please update your route to pass QRCode and Project models.',
|
|
'imported_records': 0,
|
|
'failed_records': 0,
|
|
'errors': ['Model classes missing']
|
|
}
|
|
|
|
try:
|
|
# First validate the file
|
|
validation_result = self.validate_excel_file(file_path)
|
|
|
|
if not validation_result['success']:
|
|
return {
|
|
'success': False,
|
|
'error': validation_result.get('error', 'Validation failed'),
|
|
'imported_records': 0,
|
|
'failed_records': validation_result['total_rows'],
|
|
'errors': validation_result['errors']
|
|
}
|
|
|
|
# Read Excel file
|
|
df = pd.read_excel(file_path)
|
|
|
|
# Track import statistics
|
|
imported_count = 0
|
|
failed_count = 0
|
|
geocoded_count = 0 # Track how many addresses were auto-geocoded
|
|
errors = []
|
|
imported_qr_codes = []
|
|
|
|
# If no project lookup provided, create one
|
|
if project_lookup is None:
|
|
projects = Project.query.filter_by(active_status=True).all()
|
|
project_lookup = {p.name: p.id for p in projects}
|
|
|
|
for index, row in df.iterrows():
|
|
row_num = index + 2
|
|
|
|
try:
|
|
# Extract data
|
|
name = str(row['QR Code Name']).strip()
|
|
location = str(row['QR Code Location']).strip()
|
|
location_address = str(row['Location Address']).strip()
|
|
location_event = str(row['Event']).strip()
|
|
project_name = str(row['Project']).strip()
|
|
|
|
# Get project ID
|
|
project_id = project_lookup.get(project_name)
|
|
if not project_id:
|
|
errors.append(f"Row {row_num}: Project '{project_name}' not found")
|
|
failed_count += 1
|
|
continue
|
|
|
|
# Extract GPS coordinates if provided
|
|
address_latitude = None
|
|
address_longitude = None
|
|
has_coordinates = False
|
|
coordinate_accuracy = None
|
|
|
|
if 'Latitude' in df.columns and 'Longitude' in df.columns:
|
|
if not pd.isna(row.get('Latitude')) and not pd.isna(row.get('Longitude')):
|
|
try:
|
|
address_latitude = float(row['Latitude'])
|
|
address_longitude = float(row['Longitude'])
|
|
has_coordinates = True
|
|
coordinate_accuracy = 'manual'
|
|
|
|
if self.logger:
|
|
self.logger.logger.info(f"Row {row_num}: Using provided coordinates ({address_latitude}, {address_longitude})")
|
|
except (ValueError, TypeError):
|
|
if self.logger:
|
|
self.logger.logger.warning(f"Row {row_num}: Invalid coordinate format, will attempt geocoding")
|
|
|
|
# Auto-geocode if coordinates not provided and geocode function available
|
|
if not has_coordinates and geocode_func and location_address:
|
|
try:
|
|
if self.logger:
|
|
self.logger.logger.info(f"Row {row_num}: Attempting to geocode address: {location_address[:50]}...")
|
|
|
|
# Call the geocoding function
|
|
geocoded_lat, geocoded_lng, geocoded_accuracy = geocode_func(location_address)
|
|
|
|
if geocoded_lat and geocoded_lng:
|
|
address_latitude = geocoded_lat
|
|
address_longitude = geocoded_lng
|
|
has_coordinates = True
|
|
coordinate_accuracy = geocoded_accuracy if geocoded_accuracy else 'geocoded'
|
|
geocoded_count += 1 # Increment geocoded counter
|
|
|
|
if self.logger:
|
|
self.logger.logger.info(
|
|
f"Row {row_num}: Successfully geocoded to ({address_latitude}, {address_longitude}) "
|
|
f"with accuracy: {coordinate_accuracy}"
|
|
)
|
|
else:
|
|
if self.logger:
|
|
self.logger.logger.warning(f"Row {row_num}: Geocoding returned no results for address")
|
|
except Exception as geocode_error:
|
|
if self.logger:
|
|
self.logger.logger.error(f"Row {row_num}: Geocoding error: {geocode_error}")
|
|
# Continue without coordinates - they're optional
|
|
|
|
# Check for duplicate QR code name
|
|
existing_qr = QRCode.query.filter_by(name=name, active_status=True).first()
|
|
if existing_qr:
|
|
errors.append(f"Row {row_num}: QR code with name '{name}' already exists")
|
|
failed_count += 1
|
|
continue
|
|
|
|
# Create new QR code record (without URL and image first)
|
|
new_qr_code = QRCode(
|
|
name=name,
|
|
location=location,
|
|
location_address=location_address,
|
|
location_event=location_event,
|
|
qr_code_image="",
|
|
qr_url="",
|
|
created_by=created_by,
|
|
project_id=project_id,
|
|
address_latitude=address_latitude,
|
|
address_longitude=address_longitude,
|
|
coordinate_accuracy=coordinate_accuracy,
|
|
coordinates_updated_date=datetime.utcnow() if has_coordinates else None,
|
|
fill_color='#000000',
|
|
back_color='#FFFFFF',
|
|
box_size=10,
|
|
border=4,
|
|
error_correction='H' # Highest error correction level (30% recovery)
|
|
)
|
|
|
|
# Add to session and flush to get ID
|
|
self.db.session.add(new_qr_code)
|
|
self.db.session.flush()
|
|
|
|
# Generate URL and QR code image
|
|
qr_url = generate_qr_url_func(name, new_qr_code.id)
|
|
qr_data = f"{request_url_root}qr/{qr_url}"
|
|
qr_image = generate_qr_code_func(
|
|
data=qr_data,
|
|
fill_color='#000000',
|
|
back_color='#FFFFFF',
|
|
box_size=10,
|
|
border=4,
|
|
error_correction='H' # Highest error correction level (30% recovery)
|
|
)
|
|
|
|
# Update QR code with URL and image
|
|
new_qr_code.qr_url = qr_url
|
|
new_qr_code.qr_code_image = qr_image
|
|
|
|
imported_count += 1
|
|
imported_qr_codes.append({
|
|
'name': name,
|
|
'location': location,
|
|
'project': project_name,
|
|
'id': new_qr_code.id
|
|
})
|
|
|
|
except Exception as row_error:
|
|
self.db.session.rollback()
|
|
error_msg = f"Row {row_num}: {str(row_error)}"
|
|
errors.append(error_msg)
|
|
failed_count += 1
|
|
|
|
if self.logger:
|
|
self.logger.logger.error(f"Error importing row {row_num}: {row_error}")
|
|
|
|
# Commit all successful imports
|
|
if imported_count > 0:
|
|
self.db.session.commit()
|
|
|
|
if self.logger:
|
|
self.logger.logger.info(
|
|
f"Bulk QR code import completed: {imported_count} imported, {failed_count} failed, "
|
|
f"{geocoded_count} addresses auto-geocoded"
|
|
)
|
|
|
|
return {
|
|
'success': True,
|
|
'imported_records': imported_count,
|
|
'failed_records': failed_count,
|
|
'geocoded_records': geocoded_count,
|
|
'total_rows': len(df),
|
|
'errors': errors,
|
|
'imported_qr_codes': imported_qr_codes
|
|
}
|
|
|
|
except Exception as e:
|
|
self.db.session.rollback()
|
|
|
|
if self.logger:
|
|
self.logger.logger.error(f"Error during QR code import: {e}")
|
|
self.logger.logger.error(traceback.format_exc())
|
|
|
|
return {
|
|
'success': False,
|
|
'error': str(e),
|
|
'imported_records': 0,
|
|
'failed_records': 0,
|
|
'errors': [str(e)]
|
|
} |