Unified_Authentication_Platform/backend/app/services/mapping_service.py

import io
import pandas as pd
from datetime import datetime
from typing import List, Tuple
from sqlalchemy.orm import Session
from fastapi import HTTPException
from app.models.user import User, UserRole, UserStatus
from app.models.mapping import AppUserMapping
from app.schemas.mapping import (
    MappingRowPreview, 
    MappingRowStatus, 
    MappingPreviewResponse, 
    MappingStrategy, 
    MappingImportSummary,
    ImportLogResponse,
    ImportLogItem
)
from app.core import security
from app.services.sms_service import SmsService

class MappingService:
    @staticmethod
    def process_excel_file(file_content: bytes) -> pd.DataFrame:
        """
        Reads excel file bytes into DataFrame.
        Expected columns: 'mobile'/'手机号', 'mapped_key'/'映射账号'
        """
        try:
            # Try reading as Excel first
            try:
                df = pd.read_excel(io.BytesIO(file_content))
            except:
                # Fallback to CSV
                df = pd.read_csv(io.BytesIO(file_content))

            # Normalize headers to support Chinese and English
            # Map: 手机号 -> mobile, 映射账号 -> mapped_key, 映射邮箱 -> mapped_email
            column_map = {
                '手机号': 'mobile',
                '映射账号': 'mapped_key',
                '映射邮箱': 'mapped_email',
                'user mobile': 'mobile',
                'mapped key': 'mapped_key',
                'mapped email': 'mapped_email'
            }
            
            new_columns = []
            for col in df.columns:
                col_str = str(col).strip()
                col_lower = col_str.lower()
                if col_str in column_map:
                    new_columns.append(column_map[col_str])
                elif col_lower in column_map:
                    new_columns.append(column_map[col_lower])
                else:
                    new_columns.append(col_lower)
            
            df.columns = new_columns
            
            return df
        except Exception as e:
            raise ValueError(f"文件格式无效: {str(e)}")

    @staticmethod
    def preview_import(db: Session, app_id: int, file_content: bytes, filename: str = "") -> MappingPreviewResponse:
        df = MappingService.process_excel_file(file_content)
        
        required_cols = {'mobile', 'mapped_key'}
        if not required_cols.issubset(df.columns):
            raise ValueError(f"缺少必要列。请确保文件包含: 手机号(mobile), 映射账号(mapped_key)")

        preview_rows: List[MappingRowPreview] = []
        stats = {"valid": 0, "error": 0, "new": 0, "update": 0}

        # 1. Batch query all users involved
        mobiles = df['mobile'].astype(str).unique().tolist()
        users = db.query(User).filter(User.mobile.in_(mobiles)).all()
        user_map = {u.mobile: u.id for u in users}

        # 2. Batch query existing mappings for this app
        user_ids = list(user_map.values())
        existing_mappings = db.query(AppUserMapping).filter(
            AppUserMapping.app_id == app_id,
            AppUserMapping.user_id.in_(user_ids)
        ).all()
        mapping_map = {m.user_id: m for m in existing_mappings}

        # 3. Iterate rows
        for index, row in df.iterrows():
            mobile = str(row['mobile']).strip()
            mapped_key = str(row['mapped_key']).strip()
            mapped_email = None
            if 'mapped_email' in df.columns and not pd.isna(row['mapped_email']):
                 val = str(row['mapped_email']).strip()
                 mapped_email = val if val else None
            
            row_preview = MappingRowPreview(
                row_index=index + 1, # 1-based index for UI
                mobile=mobile, 
                mapped_key=mapped_key, 
                mapped_email=mapped_email,
                status=MappingRowStatus.ERROR
            )

            if not mobile or not mapped_key or mobile == 'nan' or mapped_key == 'nan':
                row_preview.message = "手机号或映射账号为空"
                stats["error"] += 1
            elif mobile not in user_map:
                row_preview.status = MappingRowStatus.AUTO_CREATE_USER
                row_preview.message = "用户不存在，将自动创建"
                stats["new"] += 1
                stats["valid"] += 1
            else:
                user_id = user_map[mobile]
                row_preview.user_id = user_id
                
                if user_id in mapping_map:
                    row_preview.status = MappingRowStatus.UPDATE
                    row_preview.message = f"当前映射: {mapping_map[user_id].mapped_key}"
                    stats["update"] += 1
                    stats["valid"] += 1
                else:
                    row_preview.status = MappingRowStatus.NEW
                    stats["new"] += 1
                    stats["valid"] += 1
            
            preview_rows.append(row_preview)

        return MappingPreviewResponse(
            total_rows=len(df),
            valid_count=stats["valid"],
            error_count=stats["error"],
            new_count=stats["new"],
            update_count=stats["update"],
            preview_rows=preview_rows
        )

    @staticmethod
    def execute_import(
        db: Session, 
        app_id: int, 
        file_content: bytes, 
        filename: str,
        strategy: MappingStrategy,
        current_user_mobile: str,
        verification_code: str
    ) -> ImportLogResponse:
        
        # 0. Verify SMS Code
        if not SmsService.verify_code(current_user_mobile, verification_code):
            raise HTTPException(status_code=400, detail="验证码无效或已过期")

        preview = MappingService.preview_import(db, app_id, file_content, filename)
        
        summary = MappingImportSummary(total_processed=0, inserted=0, updated=0, failed=0)
        logs: List[ImportLogItem] = []
        import_time = datetime.now()
        
        for row in preview.preview_rows:
            summary.total_processed += 1
            
            log_item = ImportLogItem(
                import_time=import_time,
                mobile=row.mobile,
                mapped_key=row.mapped_key,
                mapped_email=row.mapped_email,
                status="Failed",
                message="",
                is_unified_account_created=False
            )

            if row.status == MappingRowStatus.ERROR:
                summary.failed += 1
                log_item.status = "Failed"
                log_item.message = row.message or "数据错误"
                logs.append(log_item)
                continue
                
            try:
                if row.status == MappingRowStatus.NEW or row.status == MappingRowStatus.AUTO_CREATE_USER:
                    user_id = row.user_id
                    
                    if row.status == MappingRowStatus.AUTO_CREATE_USER:
                        # Create User
                        pwd = security.generate_alphanumeric_password(8)
                        new_user = User(
                            mobile=row.mobile,
                            password_hash=security.get_password_hash(pwd),
                            status=UserStatus.ACTIVE,
                            role=UserRole.ORDINARY_USER
                        )
                        db.add(new_user)
                        db.flush() # flush to get ID
                        user_id = new_user.id
                        
                        log_item.is_unified_account_created = True
                        log_item.generated_password = pwd
                    
                    # Check uniqueness before insert (Concurrency edge case or just double check)
                    # We rely on unique constraint but catching it is cleaner
                    
                    new_mapping = AppUserMapping(
                        app_id=app_id,
                        user_id=user_id,
                        mapped_key=row.mapped_key,
                        mapped_email=row.mapped_email
                    )
                    db.add(new_mapping)
                    summary.inserted += 1
                    log_item.status = "Success"
                    log_item.message = "映射创建成功"
                    
                elif row.status == MappingRowStatus.UPDATE:
                    if strategy == MappingStrategy.OVERWRITE:
                        # Find and update
                        existing = db.query(AppUserMapping).filter(
                            AppUserMapping.app_id == app_id,
                            AppUserMapping.user_id == row.user_id
                        ).first()
                        if existing:
                            existing.mapped_key = row.mapped_key
                            existing.mapped_email = row.mapped_email
                            db.add(existing)
                            summary.updated += 1
                            log_item.status = "Success"
                            log_item.message = "映射更新成功"
                    else:
                        # SKIP
                        log_item.status = "Skipped"
                        log_item.message = "映射已存在 (已跳过)"
            
            except Exception as e:
                summary.failed += 1
                log_item.status = "Failed"
                log_item.message = str(e)
            
            logs.append(log_item)
        
        db.commit()
        return ImportLogResponse(summary=summary, logs=logs)