from __future__ import annotations
import hashlib, json, re, time
from datetime import datetime, timedelta, timezone
from sqlalchemy import select, text
from sqlalchemy.orm import Session
from app.core.security import decrypt_secret
from app.connectors.sqlalchemy_connector import SQLAlchemyConnector
from app.models.entities import (DataSource,Dataset,SemanticModel,SemanticExecution,SemanticQueryCache,
 SemanticCostPolicy,SemanticMaterialization)
from app.services.semantic_engine import semantic_plan
SAFE=re.compile(r'^[A-Za-z_][A-Za-z0-9_.$#\[\]-]*$')
OPS={'eq':'=','ne':'<>','gt':'>','gte':'>=','lt':'<','lte':'<=','in':'IN','like':'LIKE'}
def role_value(user): return user.role.value if hasattr(user.role,'value') else str(user.role)
def connector(source):
    return SQLAlchemyConnector(source.db_type.value,source.host,source.port,source.database,source.username,decrypt_secret(source.encrypted_password),source.options)
def policy_for(db,user):
    role=role_value(user)
    return db.scalar(select(SemanticCostPolicy).where(SemanticCostPolicy.enabled.is_(True),SemanticCostPolicy.subject_type=='user',SemanticCostPolicy.subject_value==user.id)) or db.scalar(select(SemanticCostPolicy).where(SemanticCostPolicy.enabled.is_(True),SemanticCostPolicy.subject_type=='role',SemanticCostPolicy.subject_value==role))
def quote_ident(value):
    if not SAFE.fullmatch(value): raise ValueError('Identificador físico inválido')
    return value
def estimate_cost(plan): return max(1,plan['limit'])*(max(1,len(plan['metrics']))+max(1,len(plan['dimensions'])))
def enforce_policy(db,user,plan):
    p=policy_for(db,user)
    limits={'max_rows':5000,'max_dimensions':8,'max_metrics':12,'max_cost_units':50000,'timeout_seconds':30,'cache_ttl_seconds':300}
    if p: limits.update({k:getattr(p,k) for k in limits})
    if plan['limit']>limits['max_rows']: raise PermissionError('Limite de linhas excede a política de custo')
    if len(plan['dimensions'])>limits['max_dimensions'] or len(plan['metrics'])>limits['max_metrics']: raise PermissionError('Complexidade semântica excede a política')
    cost=estimate_cost(plan)
    if cost>limits['max_cost_units']: raise PermissionError('Custo estimado excede a política')
    return limits,cost
def build_sql(plan,dialect='postgresql'):
    source=quote_ident(plan['source_object'])
    selects=[]; groups=[]
    for d in plan['dimensions']:
        expr=quote_ident(d['expression']); alias=quote_ident(d['name']); selects.append(f'{expr} AS {alias}'); groups.append(expr)
    aggs={'sum':'SUM','avg':'AVG','count':'COUNT','min':'MIN','max':'MAX','count_distinct':'COUNT'}
    for m in plan['metrics']:
        expr=quote_ident(m['expression']); alias=quote_ident(m['name']); agg=aggs.get(m['aggregation'])
        if not agg: raise ValueError('Agregação não autorizada')
        selects.append(f"COUNT(DISTINCT {expr}) AS {alias}" if m['aggregation']=='count_distinct' else f'{agg}({expr}) AS {alias}')
    if not selects: raise ValueError('Informe ao menos uma métrica ou dimensão')
    params={}; clauses=[]
    for i,(field,raw) in enumerate((plan.get('filters') or {}).items()):
        field=quote_ident(field); op='eq'; value=raw
        if isinstance(raw,dict): op=raw.get('op','eq'); value=raw.get('value')
        if op not in OPS: raise ValueError('Operador de filtro não autorizado')
        key=f'p{i}'
        if op=='in':
            if not isinstance(value,list) or not value: raise ValueError('Filtro IN inválido')
            names=[]
            for j,v in enumerate(value): params[f'{key}_{j}']=v; names.append(f':{key}_{j}')
            clauses.append(f"{field} IN ({','.join(names)})")
        else: params[key]=value; clauses.append(f'{field} {OPS[op]} :{key}')
    sql=f"SELECT {', '.join(selects)} FROM {source}"
    if clauses: sql+=' WHERE '+' AND '.join(clauses)
    if groups: sql+=' GROUP BY '+', '.join(groups)
    sql+=f" LIMIT {int(plan['limit'])}" if dialect not in {'oracle','mssql'} else ''
    if dialect=='oracle': sql+=f" FETCH FIRST {int(plan['limit'])} ROWS ONLY"
    if dialect=='mssql': sql=sql.replace('SELECT ','SELECT TOP '+str(int(plan['limit']))+' ',1)
    return sql,params
def execute_semantic(db,user,model,metrics,dimensions,filters,limit,use_cache=True):
    plan=semantic_plan(db,user,model,metrics,dimensions,filters,limit)
    limits,cost=enforce_policy(db,user,plan)
    key=hashlib.sha256(json.dumps({'u':user.id,'p':plan},sort_keys=True,default=str).encode()).hexdigest()
    now=datetime.now(timezone.utc)
    if use_cache:
        cached=db.scalar(select(SemanticQueryCache).where(SemanticQueryCache.cache_key==key,SemanticQueryCache.expires_at>now))
        if cached:
            run=SemanticExecution(user_id=user.id,model_id=model.id,plan=plan,status='success',row_count=cached.row_count,cache_hit=True,cost_units=cost,finished_at=now); db.add(run); db.commit(); db.refresh(run)
            return {'execution_id':run.id,'cache_hit':True,'columns':cached.payload.get('columns',[]),'rows':cached.payload.get('rows',[]),'row_count':cached.row_count,'plan':plan}
    dataset=db.get(Dataset,model.dataset_id) if model.dataset_id else None
    if not dataset: raise ValueError('Modelo sem dataset físico vinculado')
    source=db.get(DataSource,dataset.source_id)
    if not source or not source.enabled: raise ValueError('Fonte de dados indisponível')
    plan['source_object']=plan.get('source_object') or dataset.source_object
    c=connector(source); sql,params=build_sql(plan,c.engine.dialect.name)
    run=SemanticExecution(user_id=user.id,model_id=model.id,plan=plan,generated_sql=sql,status='running',cost_units=cost); db.add(run); db.commit(); db.refresh(run)
    start=time.perf_counter()
    try:
        with c.engine.connect() as conn:
            result=conn.execute(text(sql),params); columns=list(result.keys()); rows=[[v.isoformat() if hasattr(v,'isoformat') else v for v in r] for r in result.fetchall()]
        run.status='success'; run.row_count=len(rows); run.duration_ms=int((time.perf_counter()-start)*1000); run.finished_at=datetime.now(timezone.utc)
        payload={'columns':columns,'rows':rows}; db.add(SemanticQueryCache(cache_key=key,model_id=model.id,payload=payload,row_count=len(rows),expires_at=datetime.now(timezone.utc)+timedelta(seconds=limits['cache_ttl_seconds'])))
        db.commit(); return {'execution_id':run.id,'cache_hit':False,'columns':columns,'rows':rows,'row_count':len(rows),'duration_ms':run.duration_ms,'plan':plan}
    except Exception as exc:
        run.status='failed'; run.error=str(exc)[:2000]; run.duration_ms=int((time.perf_counter()-start)*1000); run.finished_at=datetime.now(timezone.utc); db.commit(); raise