from __future__ import annotations
import hashlib, json, re, unicodedata
from difflib import SequenceMatcher
from pathlib import Path
from typing import Any
import pandas as pd
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models.entities import (DataCorrectionPolicyV031, DataQualityScorecardV031, Dataset,
    AdvancedQuarantineRecordV031, MasterEntity, MasterMergeProposalV031, MasterSourceRecord, QualityCorrectionRunV031, utcnow)
from app.services.analytics import latest_frame

EMAIL_RE=re.compile(r'^[^\s@]+@[^\s@]+\.[^\s@]+$')
PHONE_RE=re.compile(r'^\+?[0-9]{8,15}$')

def digits(value:Any)->str: return re.sub(r'\D','',str(value or ''))
def normalize_text(value:Any)->str:
    text=unicodedata.normalize('NFKD',str(value or '')).encode('ascii','ignore').decode().strip().lower()
    return re.sub(r'\s+',' ',text)
def valid_cpf(value:Any)->bool:
    n=digits(value)
    if len(n)!=11 or n==n[0]*11: return False
    for size in (9,10):
        total=sum(int(n[i])*(size+1-i) for i in range(size)); d=(total*10)%11; d=0 if d==10 else d
        if d!=int(n[size]): return False
    return True
def valid_cnpj(value:Any)->bool:
    n=digits(value)
    if len(n)!=14 or n==n[0]*14: return False
    for size,weights in ((12,[5,4,3,2,9,8,7,6,5,4,3,2]),(13,[6,5,4,3,2,9,8,7,6,5,4,3,2])):
        total=sum(int(n[i])*weights[i] for i in range(size)); rem=total%11; d=0 if rem<2 else 11-rem
        if d!=int(n[size]): return False
    return True
def valid_email(value:Any)->bool: return bool(EMAIL_RE.match(str(value or '').strip().lower()))
def valid_phone(value:Any)->bool: return bool(PHONE_RE.match(digits(value)))
def apply_correction(value:Any,kind:str,cfg:dict)->Any:
    if value is None: return value
    if kind=='trim': return str(value).strip()
    if kind=='lowercase': return str(value).strip().lower()
    if kind=='uppercase': return str(value).strip().upper()
    if kind=='digits_only': return digits(value)
    if kind=='normalize_text': return normalize_text(value)
    if kind=='replace': return str(value).replace(str(cfg.get('old','')),str(cfg.get('new','')))
    if kind=='default_if_blank': return cfg.get('value') if str(value).strip()=='' else value
    if kind=='date_format': return pd.to_datetime(value,errors='coerce').strftime(cfg.get('format','%Y-%m-%d'))
    return value

def validate_value(value:Any,rule_type:str,cfg:dict)->bool:
    if rule_type=='cpf': return valid_cpf(value)
    if rule_type=='cnpj': return valid_cnpj(value)
    if rule_type=='email': return valid_email(value)
    if rule_type=='phone': return valid_phone(value)
    if rule_type=='required': return value is not None and str(value).strip()!=''
    if rule_type=='regex': return bool(re.match(cfg.get('pattern','.*'),str(value or '')))
    return True

def run_corrections(db:Session,tenant_id:str,dataset:Dataset,user_id:str,dry_run:bool=True)->QualityCorrectionRunV031:
    run=QualityCorrectionRunV031(tenant_id=tenant_id,dataset_id=dataset.id,dry_run=dry_run,status='running',initiated_by=user_id,started_at=utcnow())
    db.add(run); db.flush()
    try:
        df=latest_frame(db,dataset).copy(); run.rows_read=len(df); changes=[]; quarantined=[]
        policies=db.scalars(select(DataCorrectionPolicyV031).where(DataCorrectionPolicyV031.tenant_id==tenant_id,DataCorrectionPolicyV031.dataset_id==dataset.id,DataCorrectionPolicyV031.enabled.is_(True)).order_by(DataCorrectionPolicyV031.priority)).all()
        for p in policies:
            if p.column_name not in df.columns: continue
            before=df[p.column_name].copy(); df[p.column_name]=df[p.column_name].map(lambda v: apply_correction(v,p.correction_type,p.config or {}))
            mask=before.astype(str)!=df[p.column_name].astype(str); count=int(mask.sum())
            if count: changes.append({'policy_id':p.id,'column':p.column_name,'changed':count})
            validator=(p.config or {}).get('validator')
            if validator:
                bad=~df[p.column_name].map(lambda v: validate_value(v,validator,p.config or {}))
                for idx,row in df.loc[bad].head(500).iterrows(): quarantined.append({'policy_id':p.id,'row_reference':str(idx),'row_data':{k:(None if pd.isna(v) else v) for k,v in row.to_dict().items()},'reason':f'{p.name}: {validator}'})
        run.rows_changed=sum(x['changed'] for x in changes); run.rows_quarantined=len(quarantined); run.summary={'changes':changes,'policies':len(policies)}
        if not dry_run:
            root=Path('storage/data')/tenant_id/'quality-corrections'; root.mkdir(parents=True,exist_ok=True)
            path=root/f'{run.id}.parquet'; df.to_parquet(path,index=False); run.output_object_key=str(path); run.checksum=hashlib.sha256(path.read_bytes()).hexdigest()
            for q in quarantined: db.add(AdvancedQuarantineRecordV031(tenant_id=tenant_id,dataset_id=dataset.id,correction_run_id=run.id,**q))
        run.status='success'; run.finished_at=utcnow()
    except Exception as exc:
        run.status='failed'; run.error=str(exc); run.finished_at=utcnow()
    db.commit(); db.refresh(run); return run

def entity_similarity(a:dict,b:dict,fields:list[str])->tuple[int,dict]:
    evidence={}; scores=[]
    for field in fields:
        av,bv=normalize_text(a.get(field)),normalize_text(b.get(field))
        score=100 if av and av==bv else round(SequenceMatcher(None,av,bv).ratio()*100) if av and bv else 0
        evidence[field]={'left':a.get(field),'right':b.get(field),'score':score}; scores.append(score)
    return (round(sum(scores)/len(scores)) if scores else 0,evidence)

def propose_merges(db:Session,tenant_id:str,entity_type:str,user_id:str,fields:list[str],threshold:int=85)->list[MasterMergeProposalV031]:
    entities=db.scalars(select(MasterEntity).where(MasterEntity.entity_type==entity_type,MasterEntity.lifecycle_status=='active')).all(); created=[]
    for i,left in enumerate(entities):
        for right in entities[i+1:]:
            score,evidence=entity_similarity(left.golden_data or {},right.golden_data or {},fields)
            if score<threshold: continue
            exists=db.scalar(select(MasterMergeProposalV031).where(MasterMergeProposalV031.tenant_id==tenant_id,MasterMergeProposalV031.survivor_id==left.id,MasterMergeProposalV031.status=='pending'))
            if exists and right.id in (exists.duplicate_ids or []): continue
            proposal=MasterMergeProposalV031(tenant_id=tenant_id,entity_type=entity_type,survivor_id=left.id,duplicate_ids=[right.id],confidence_score=score,evidence=evidence,proposed_golden_data=left.golden_data or {},proposed_by=user_id)
            db.add(proposal); created.append(proposal)
    db.commit(); return created

def approve_merge(db:Session,proposal:MasterMergeProposalV031,user_id:str)->None:
    survivor=db.get(MasterEntity,proposal.survivor_id)
    if not survivor: raise ValueError('Registro sobrevivente não encontrado')
    for duplicate_id in proposal.duplicate_ids or []:
        duplicate=db.get(MasterEntity,duplicate_id)
        if not duplicate: continue
        for src in db.scalars(select(MasterSourceRecord).where(MasterSourceRecord.master_entity_id==duplicate.id)).all(): src.master_entity_id=survivor.id
        duplicate.lifecycle_status='merged'; duplicate.status='merged'; duplicate.golden_data={**(duplicate.golden_data or {}),'merged_into':survivor.id}
    survivor.golden_data=proposal.proposed_golden_data or survivor.golden_data; proposal.status='approved'; proposal.reviewed_by=user_id; proposal.reviewed_at=utcnow(); db.commit()

def create_scorecard(db:Session,tenant_id:str,dataset_id:str,domain:str='general')->DataQualityScorecardV031:
    from app.models.entities import QualityRun, QuarantineRecord
    run=db.scalar(select(QualityRun).where(QualityRun.dataset_id==dataset_id,QualityRun.status=='completed').order_by(QualityRun.created_at.desc()))
    dims=(run.dimension_scores or {}) if run else {}; quarantined=db.query(QuarantineRecord).filter(QuarantineRecord.dataset_id==dataset_id,QuarantineRecord.status=='quarantined').count()
    item=DataQualityScorecardV031(tenant_id=tenant_id,dataset_id=dataset_id,domain=domain,overall_score=(run.overall_score if run else 0),completeness_score=dims.get('completeness',0),validity_score=dims.get('validity',0),uniqueness_score=dims.get('uniqueness',0),consistency_score=dims.get('consistency',0),freshness_score=dims.get('freshness',0),quarantined_rows=quarantined,details={'run_id':run.id if run else None})
    db.add(item); db.commit(); db.refresh(item); return item