import json,time
from datetime import datetime,timezone
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.security import encrypt_secret,decrypt_secret
from app.models.entities import ConnectorDefinition,TenantConnectorInstance,ConnectorDiscoveredObject

BUILTINS=[
 ('postgresql','PostgreSQL','database','sql',['catalog','full','incremental','watermark'],['password','ssl'],'postgresql+psycopg'),
 ('mysql','MySQL / MariaDB','database','sql',['catalog','full','incremental','watermark'],['password','ssl'],'mysql+pymysql'),
 ('sqlserver','Microsoft SQL Server','database','sql',['catalog','full','incremental','watermark'],['password','windows','ssl'],'mssql+pyodbc'),
 ('oracle','Oracle Database','database','sql',['catalog','full','incremental','watermark'],['password','wallet'],'oracle+oracledb'),
 ('db2','IBM DB2','database','sql',['catalog','full','incremental'],['password','ssl'],'ibm_db_sa'),
 ('firebird','Firebird','database','sql',['catalog','full','incremental'],['password'],'firebird+firebird'),
 ('snowflake','Snowflake','warehouse','sql',['catalog','full','incremental'],['password','keypair','oauth'],'snowflake-sqlalchemy'),
 ('bigquery','Google BigQuery','warehouse','sql',['catalog','full','incremental'],['service_account','oauth'],'sqlalchemy-bigquery'),
 ('redshift','Amazon Redshift','warehouse','sql',['catalog','full','incremental'],['password','iam'],'redshift-connector'),
 ('clickhouse','ClickHouse','database','sql',['catalog','full','incremental'],['password','ssl'],'clickhouse-connect'),
 ('rest','REST API','api','http',['openapi','pagination','incremental','webhook'],['api_key','bearer','oauth2','basic'],'httpx'),
 ('graphql','GraphQL','api','http',['schema','pagination'],['api_key','bearer','oauth2'],'httpx'),
 ('soap','SOAP / WSDL','api','xml',['wsdl','pagination'],['basic','certificate'],'zeep'),
 ('odata','OData','api','http',['metadata','pagination','incremental'],['basic','oauth2'],'httpx'),
 ('sftp','SFTP','file','sftp',['list','download','incremental'],['password','ssh_key'],'paramiko'),
 ('csv','CSV','file','file',['schema_inference','full','incremental'],['none'],'pandas'),
 ('excel','Excel','file','file',['schema_inference','sheets','full'],['none'],'openpyxl'),
 ('json','JSON / JSONL','file','file',['schema_inference','full','incremental'],['none'],'orjson'),
 ('xml','XML','file','file',['schema_inference','full'],['none'],'lxml'),
 ('parquet','Parquet','file','file',['schema_inference','full'],['none'],'pyarrow'),
]

def seed_connectors(db:Session):
    existing={x.code for x in db.scalars(select(ConnectorDefinition)).all()}
    for code,name,category,protocol,caps,auth,driver in BUILTINS:
        if code not in existing:
            db.add(ConnectorDefinition(code=code,name=name,category=category,protocol=protocol,capabilities=caps,auth_methods=auth,driver_package=driver,config_schema={'required':['name'],'secret_fields':['password','token','client_secret','private_key']}))
    db.commit()

def protect_config(config:dict):
    public={}; secret={}
    secret_names={'password','token','api_key','client_secret','private_key','connection_string'}
    for k,v in config.items(): (secret if k in secret_names else public)[k]=v
    return public,encrypt_secret(json.dumps(secret)) if secret else ''

def reveal_config(instance:TenantConnectorInstance):
    secret=json.loads(decrypt_secret(instance.encrypted_config)) if instance.encrypted_config else {}
    return {**(instance.public_config or {}),**secret}

def test_instance(instance:TenantConnectorInstance):
    cfg=reveal_config(instance); started=time.perf_counter(); code=instance.connector_code
    if code in {'postgresql','mysql','sqlserver','oracle'}:
        from app.connectors.sqlalchemy_connector import SQLAlchemyConnector
        c=SQLAlchemyConnector(code,cfg.get('host',''),int(cfg.get('port') or 0),cfg.get('database',''),cfg.get('username',''),cfg.get('password',''),cfg.get('options',{})); c.test()
    elif code in {'rest','graphql','odata'}:
        import httpx
        url=cfg.get('base_url') or cfg.get('url')
        if not url: raise ValueError('URL base obrigatória')
        r=httpx.get(url,timeout=10,follow_redirects=False); r.raise_for_status()
    elif code in {'csv','excel','json','xml','parquet'}:
        from pathlib import Path
        p=Path(cfg.get('path',''))
        if not p.exists() or not p.is_file(): raise ValueError('Arquivo não encontrado')
    return int((time.perf_counter()-started)*1000)

def discover(instance:TenantConnectorInstance):
    cfg=reveal_config(instance); code=instance.connector_code
    if code in {'postgresql','mysql','sqlserver','oracle'}:
        from app.connectors.sqlalchemy_connector import SQLAlchemyConnector
        c=SQLAlchemyConnector(code,cfg.get('host',''),int(cfg.get('port') or 0),cfg.get('database',''),cfg.get('username',''),cfg.get('password',''),cfg.get('options',{})); return c.catalog(cfg.get('schema'))
    if code=='csv':
        import pandas as pd
        df=pd.read_csv(cfg['path'],nrows=1000); return [{'type':'file','schema':'','name':cfg['path'].split('/')[-1],'qualified_name':cfg['path'],'columns':[{'name':str(k),'type':str(v),'nullable':True} for k,v in df.dtypes.items()]}]
    if code=='excel':
        import pandas as pd
        x=pd.ExcelFile(cfg['path']); out=[]
        for sheet in x.sheet_names:
            df=pd.read_excel(cfg['path'],sheet_name=sheet,nrows=100); out.append({'type':'sheet','schema':'','name':sheet,'qualified_name':f"{cfg['path']}#{sheet}",'columns':[{'name':str(k),'type':str(v),'nullable':True} for k,v in df.dtypes.items()]})
        return out
    if code in {'rest','graphql','odata','soap'}: return []
    return []

def persist_discovery(db:Session,tenant_id:str,instance_id:str,objects:list):
    count=0
    for obj in objects:
        qn=obj.get('qualified_name') or obj.get('name')
        row=db.scalar(select(ConnectorDiscoveredObject).where(ConnectorDiscoveredObject.tenant_id==tenant_id,ConnectorDiscoveredObject.connector_instance_id==instance_id,ConnectorDiscoveredObject.qualified_name==qn))
        if not row:
            row=ConnectorDiscoveredObject(tenant_id=tenant_id,connector_instance_id=instance_id,object_type=obj.get('type','table'),namespace=obj.get('schema',''),object_name=obj.get('name',qn),qualified_name=qn,columns_json=obj.get('columns',[])); db.add(row); count+=1
        else:
            row.columns_json=obj.get('columns',[]); row.discovered_at=datetime.now(timezone.utc)
    db.commit(); return count