from __future__ import annotations
import hashlib, json, os, shutil
from datetime import datetime, timezone, timedelta
from pathlib import Path
from typing import Any
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models.entities import LakehouseCatalogV035, LakehouseTableV035, LakehouseSnapshotV035, LakehouseDataFileV035, LakehouseMaintenanceRunV035

ALLOWED_ZONES={'bronze','silver','gold'}
ALLOWED_CATALOGS={'native','iceberg_rest','hive','glue'}
ALLOWED_MAINTENANCE={'compact','expire_snapshots','rewrite_manifests','verify'}

def safe_segment(value:str)->str:
    value=''.join(c if c.isalnum() or c in {'-','_','.'} else '_' for c in str(value)).strip('._')
    if not value or value in {'.','..'}: raise ValueError('Nome de caminho inválido')
    return value[:180]

def warehouse_root(catalog:LakehouseCatalogV035, tenant_id:str)->Path:
    uri=(catalog.warehouse_uri or 'storage/lakehouse').strip()
    if uri.startswith(('s3://','http://','https://')):
        raise ValueError('O modo nativo requer caminho local. Use adaptador Iceberg/S3 homologado para URI remota.')
    root=Path(uri).expanduser().resolve()/safe_segment(tenant_id)
    root.mkdir(parents=True,exist_ok=True)
    return root

def table_root(catalog:LakehouseCatalogV035, table:LakehouseTableV035)->Path:
    root=warehouse_root(catalog,table.tenant_id)/safe_segment(table.namespace)/safe_segment(table.name)
    (root/'data').mkdir(parents=True,exist_ok=True); (root/'metadata').mkdir(parents=True,exist_ok=True)
    return root

def validate_schema(schema:list[dict])->list[dict]:
    names=set(); out=[]
    for col in schema:
        name=str(col.get('name','')).strip()
        if not name or name in names: raise ValueError('Schema contém coluna vazia ou duplicada')
        names.add(name)
        out.append({'name':name,'type':str(col.get('type','string')).lower(),'nullable':bool(col.get('nullable',True)),'description':str(col.get('description',''))})
    if not out: raise ValueError('Schema deve possuir pelo menos uma coluna')
    return out

def evolve_schema(current:list[dict], requested:list[dict])->list[dict]:
    current_map={c['name']:c for c in current}; requested=validate_schema(requested)
    for col in requested:
        old=current_map.get(col['name'])
        if old and old.get('type')!=col.get('type'):
            raise ValueError(f"Alteração incompatível de tipo: {col['name']} ({old.get('type')} -> {col.get('type')})")
    missing=[c['name'] for c in current if c['name'] not in {x['name'] for x in requested}]
    if missing: raise ValueError('Remoção de colunas exige migração explícita: '+', '.join(missing))
    return requested

def create_table(db:Session,catalog:LakehouseCatalogV035,tenant_id:str,user_id:str,*,namespace:str,name:str,friendly_name:str,zone:str,schema:list,partition_spec:list,properties:dict,dataset_id:str|None=None):
    if zone not in ALLOWED_ZONES: raise ValueError('Zona inválida')
    item=LakehouseTableV035(tenant_id=tenant_id,catalog_id=catalog.id,dataset_id=dataset_id,namespace=safe_segment(namespace),name=safe_segment(name),friendly_name=friendly_name,zone=zone,schema_json=validate_schema(schema),partition_spec=partition_spec or [],properties=properties or {},created_by=user_id)
    db.add(item); db.flush(); table_root(catalog,item); db.commit(); db.refresh(item); return item

def commit_files(db:Session,catalog:LakehouseCatalogV035,table:LakehouseTableV035,user_id:str,files:list[dict],operation:str='append',summary:dict|None=None):
    if not files: raise ValueError('Informe ao menos um arquivo Parquet')
    root=table_root(catalog,table); metadata=root/'metadata'; normalized=[]; total_rows=total_bytes=0
    for raw in files:
        src=Path(str(raw.get('path',''))).expanduser().resolve()
        if not src.exists() or not src.is_file(): raise ValueError(f'Arquivo não encontrado: {src}')
        if src.suffix.lower() not in {'.parquet','.pq'}: raise ValueError('Somente arquivos Parquet são aceitos no commit nativo')
        data=src.read_bytes(); checksum=hashlib.sha256(data).hexdigest(); target=root/'data'/f"{checksum[:16]}-{safe_segment(src.name)}"
        if not target.exists(): shutil.copy2(src,target)
        rows=int(raw.get('row_count',0)); size=target.stat().st_size
        normalized.append({'path':str(target),'partition_values':raw.get('partition_values') or {},'row_count':rows,'size_bytes':size,'checksum_sha256':checksum,'column_stats':raw.get('column_stats') or {}})
        total_rows+=rows; total_bytes+=size
    snapshot=LakehouseSnapshotV035(tenant_id=table.tenant_id,table_id=table.id,parent_snapshot_id=table.current_snapshot_id,operation=operation,schema_json=table.schema_json,partition_spec=table.partition_spec,summary=summary or {},row_count=total_rows,file_count=len(normalized),total_bytes=total_bytes,committed_by=user_id)
    db.add(snapshot); db.flush()
    manifest={'format-version':1,'table-id':table.id,'snapshot-id':snapshot.id,'parent-snapshot-id':snapshot.parent_snapshot_id,'committed-at':datetime.now(timezone.utc).isoformat(),'operation':operation,'schema':table.schema_json,'partition-spec':table.partition_spec,'files':normalized,'summary':summary or {}}
    blob=json.dumps(manifest,ensure_ascii=False,sort_keys=True,indent=2).encode(); path=metadata/f'snapshot-{snapshot.id}.json'; path.write_bytes(blob)
    snapshot.manifest_path=str(path); snapshot.checksum_sha256=hashlib.sha256(blob).hexdigest()
    for f in normalized: db.add(LakehouseDataFileV035(tenant_id=table.tenant_id,table_id=table.id,snapshot_id=snapshot.id,file_path=f['path'],partition_values=f['partition_values'],row_count=f['row_count'],size_bytes=f['size_bytes'],checksum_sha256=f['checksum_sha256'],column_stats=f['column_stats']))
    table.current_snapshot_id=snapshot.id; db.commit(); db.refresh(snapshot); return snapshot

def rollback_to_snapshot(db:Session,table:LakehouseTableV035,snapshot:LakehouseSnapshotV035,user_id:str):
    if snapshot.table_id!=table.id: raise ValueError('Snapshot não pertence à tabela')
    table.current_snapshot_id=snapshot.id; table.schema_json=snapshot.schema_json; table.partition_spec=snapshot.partition_spec; table.schema_version+=1
    marker=LakehouseSnapshotV035(tenant_id=table.tenant_id,table_id=table.id,parent_snapshot_id=snapshot.id,operation='rollback',manifest_path=snapshot.manifest_path,schema_json=snapshot.schema_json,partition_spec=snapshot.partition_spec,summary={'rolled_back_to':snapshot.id},row_count=snapshot.row_count,file_count=snapshot.file_count,total_bytes=snapshot.total_bytes,checksum_sha256=snapshot.checksum_sha256,committed_by=user_id)
    db.add(marker); db.flush(); table.current_snapshot_id=marker.id; db.commit(); db.refresh(marker); return marker

def run_maintenance(db:Session,catalog:LakehouseCatalogV035,table:LakehouseTableV035,user_id:str,kind:str,parameters:dict):
    if kind not in ALLOWED_MAINTENANCE: raise ValueError('Manutenção inválida')
    run=LakehouseMaintenanceRunV035(tenant_id=table.tenant_id,table_id=table.id,maintenance_type=kind,parameters=parameters or {},started_by=user_id); db.add(run); db.commit(); db.refresh(run)
    try:
        result={}
        if kind=='verify':
            bad=[]
            for f in db.scalars(select(LakehouseDataFileV035).where(LakehouseDataFileV035.table_id==table.id,LakehouseDataFileV035.status=='active')).all():
                p=Path(f.file_path)
                if not p.exists() or hashlib.sha256(p.read_bytes()).hexdigest()!=f.checksum_sha256: bad.append(f.id)
            result={'invalid_files':bad,'valid':not bad}
        elif kind=='expire_snapshots':
            keep=int(parameters.get('keep_last',5)); snaps=db.scalars(select(LakehouseSnapshotV035).where(LakehouseSnapshotV035.table_id==table.id).order_by(LakehouseSnapshotV035.committed_at.desc())).all(); expired=0
            for s in snaps[keep:]:
                if s.id!=table.current_snapshot_id:
                    for f in db.scalars(select(LakehouseDataFileV035).where(LakehouseDataFileV035.snapshot_id==s.id)).all(): f.status='expired'
                    expired+=1
            result={'snapshots_marked_expired':expired,'keep_last':keep}
        elif kind=='rewrite_manifests': result={'rewritten':True,'note':'Manifesto atual validado e mantido em formato nativo V1'}
        else: result={'planned':True,'note':'Compactação física requer pyarrow; o plano foi registrado sem apagar arquivos existentes'}
        run.status='success'; run.result=result
    except Exception as e: run.status='failed'; run.error_message=str(e)
    run.finished_at=datetime.now(timezone.utc); db.commit(); db.refresh(run); return run

def trino_sql(table:LakehouseTableV035,columns:list[str]|None=None,limit:int=1000,snapshot_id:str|None=None)->str:
    allowed={c['name'] for c in table.schema_json}; cols=columns or list(allowed)
    if not cols or any(c not in allowed for c in cols): raise ValueError('Coluna inválida')
    q='SELECT '+', '.join('"'+c.replace('"','')+'"' for c in cols)+f' FROM "{table.namespace}"."{table.name}"'
    if snapshot_id: q+=f" FOR VERSION AS OF '{snapshot_id}'"
    return q+f' LIMIT {max(1,min(int(limit),100000))}'