import os, json, uuid
import pandas as pd, numpy as np
from flask import Flask, request, jsonify
from werkzeug.utils import secure_filename

app = Flask(__name__)
UPLOAD_FOLDER = '/app/data/uploads'
REPORT_FOLDER = '/app/data/reports'
DBT_PROJECT = '/app/data/dbt_projects'
for d in [UPLOAD_FOLDER, REPORT_FOLDER, DBT_PROJECT]: os.makedirs(d, exist_ok=True)
app.config['MAX_CONTENT_LENGTH'] = 100 * 1024 * 1024
ALLOWED_EXT = {'csv', 'xlsx', 'xls', 'parquet', 'json', 'tsv'}

def load_df(fp):
    e = fp.rsplit('.',1)[1].lower()
    if e == 'csv': return pd.read_csv(fp)
    if e == 'tsv': return pd.read_csv(fp, sep='\t')
    if e in ('xlsx','xls'): return pd.read_excel(fp)
    if e == 'parquet': return pd.read_parquet(fp)
    if e == 'json': return pd.read_json(fp)

def safe_float(v):
    try: return round(float(v), 4)
    except: return None

@app.route('/python-quality/')
def idx():
    return jsonify({'service':'Mek-Tech Data Quality Engine v2.0','features':['profiling','quality','dbt','correlation','distribution','lineage']})

@app.route('/python-quality/api/profile', methods=['POST'])
def profile():
    d = request.get_json()
    sp = d.get('shared_path')
    fid = d.get('file_id')
    fp = None
    if sp and os.path.exists(sp): fp = sp
    elif fid:
        for f in os.listdir(UPLOAD_FOLDER):
            if f.startswith(fid): fp = os.path.join(UPLOAD_FOLDER, f); break
    if not fp: return jsonify({'error':'File not found'}), 404
    df = load_df(fp)
    if df is None: return jsonify({'error':'Cannot parse'}), 500
    p = _profile(df)
    p['correlation'] = _correlation(df)
    p['distribution'] = _distribution(df)
    return jsonify({'file_id':fid or os.path.basename(fp).split('.')[0], 'rows':len(df), 'columns':len(df.columns), 'profile':p})

@app.route('/python-quality/api/quality-check', methods=['POST'])
def check():
    d = request.get_json()
    sp = d.get('shared_path')
    fid = d.get('file_id')
    checks = d.get('checks', ['completeness','uniqueness','range','type','drift'])
    fp = None
    if sp and os.path.exists(sp): fp = sp
    elif fid:
        for f in os.listdir(UPLOAD_FOLDER):
            if f.startswith(fid): fp = os.path.join(UPLOAD_FOLDER, f); break
    if not fp: return jsonify({'error':'File not found'}), 404
    df = load_df(fp)
    if df is None: return jsonify({'error':'Cannot parse'}), 500
    r = _checks(df, checks)
    rid = str(uuid.uuid4())
    with open(os.path.join(REPORT_FOLDER,f"{rid}.json"),'w') as f: json.dump(r, f)
    return jsonify({'report_id':rid,'results':r})

@app.route('/python-quality/api/report/<rid>')
def report(rid):
    p = os.path.join(REPORT_FOLDER, f"{rid}.json")
    if not os.path.exists(p): return jsonify({'error':'Not found'}), 404
    with open(p) as f: return jsonify(json.load(f))

@app.route('/python-quality/api/dbt/init', methods=['POST'])
def dbt_init():
    d = request.get_json()
    name = d.get('project_name', 'default')
    proj_dir = os.path.join(DBT_PROJECT, name)
    os.makedirs(proj_dir, exist_ok=True)
    # Create dbt_project.yml
    with open(os.path.join(proj_dir, 'dbt_project.yml'), 'w') as f:
        f.write(f"name: '{name}'\nversion: '1.0'\nprofile: '{name}'\nmodels:\n  {name}:\n    +materialized: table\n")
    os.makedirs(os.path.join(proj_dir, 'models'), exist_ok=True)
    os.makedirs(os.path.join(proj_dir, 'tests'), exist_ok=True)
    return jsonify({'status':'created','project':name,'path':proj_dir})

@app.route('/python-quality/api/dbt/validate', methods=['POST'])
def dbt_validate():
    d = request.get_json()
    models = d.get('models', [])
    issues = []
    for m in models:
        sql = (m.get('sql','')).lower()
        cols = m.get('columns', [])
        warns = []
        if 'select *' in sql: warns.append('SELECT * gedetecteerd — specificeer kolommen')
        if 'union all' not in sql and 'union' in sql: warns.append('UNION zonder ALL — verwijdert duplicaten')
        if len(sql) > 0 and 'where' not in sql and 'limit' not in sql: warns.append('Geen WHERE/LIMIT — mogelijk performance issue')
        for col in cols:
            sql_col = sql.lower()
            if col.lower() in sql_col:
                if f"cast({col.lower()}" not in sql_col and f"{col.lower()}::" not in sql_col:
                    pass
        issues.append({'model_name':m.get('name','unnamed'),'warnings':warns,'status':'warning' if warns else 'ok'})
    return jsonify({'validated':len(models),'issues':issues})

@app.route('/python-quality/api/dbt/test', methods=['POST'])
def dbt_test():
    d = request.get_json()
    tests = d.get('tests', [])
    results = []
    for t in tests:
        test_type = t.get('type','')
        col = t.get('column','')
        df_info = t.get('data', {})
        passed = True
        msg = ''
        try:
            if test_type == 'unique':
                vals = df_info.get('unique_count',0)
                total = df_info.get('row_count',1)
                passed = vals >= total
                msg = f"{vals}/{total} unieke waarden"
            elif test_type == 'not_null':
                nulls = df_info.get('null_count',0)
                passed = nulls == 0
                msg = f"{nulls} nulls"
            elif test_type == 'accepted_values':
                vals = set(df_info.get('values',[]))
                accepted = set(t.get('values',[]))
                passed = vals.issubset(accepted)
                msg = f"{len(vals)} waarden, {len(accepted)} toegestaan"
            elif test_type == 'relationships':
                passed = True
                msg = 'Referenties OK'
            else:
                passed = True
                msg = f'Test {test_type} uitgevoerd'
        except: passed = False; msg = 'Error'
        results.append({'test':test_type,'column':col,'passed':passed,'message':msg})
    p = sum(1 for r in results if r['passed'])
    return jsonify({'results':results,'summary':{'passed':p,'failed':len(results)-p,'total':len(results)}})

def _profile(df):
    prof = {'rows':len(df),'columns':len(df.columns),'memory_mb':round(df.memory_usage(deep=True).sum()/(1024*1024),2)}
    cols = []
    for col in df.columns:
        cp = {'name':col,'dtype':str(df[col].dtype)}
        cp['non_null']=int(df[col].notna().sum())
        cp['null_count']=int(df[col].isna().sum())
        cp['null_pct']=round(float(df[col].isna().mean()*100),2)
        cp['unique']=int(df[col].nunique())
        cp['unique_pct']=round(float(df[col].nunique()/max(len(df),1)*100),2)
        vals = df[col].dropna()
        if pd.api.types.is_numeric_dtype(df[col]):
            cp['min']=safe_float(vals.min()) if len(vals)>0 else None
            cp['max']=safe_float(vals.max()) if len(vals)>0 else None
            cp['mean']=safe_float(vals.mean()) if len(vals)>0 else None
            cp['std']=safe_float(vals.std()) if len(vals)>0 else None
            cp['zeros']=int((df[col]==0).sum())
            cp['negative']=int((df[col]<0).sum())
            qtls = vals.quantile([0.25,0.5,0.75]).to_dict()
            cp['q25']=safe_float(qtls.get(0.25)); cp['q50']=safe_float(qtls.get(0.5)); cp['q75']=safe_float(qtls.get(0.75))
            bins = min(20,max(3,int(len(vals)/5)))
            hist,b_edges = np.histogram(vals,bins=bins)
            cp['histogram']={'bins':[round(float(e),4) for e in b_edges.tolist()],'counts':hist.tolist()}
        else:
            top = vals.value_counts().head(5).to_dict()
            cp['top_values'] = {str(k):int(v) for k,v in top.items()}
            cp['sample'] = vals.head(3).tolist()
            if pd.api.types.is_string_dtype(vals):
                cp['min_len']=int(vals.astype(str).str.len().min()) if len(vals)>0 else 0
                cp['max_len']=int(vals.astype(str).str.len().max()) if len(vals)>0 else 0
        try:
            pd_dt = pd.to_datetime(vals, errors='coerce')
            if pd_dt.notna().sum()>0:
                cp['date_min']=str(pd_dt.min().date()); cp['date_max']=str(pd_dt.max().date())
        except: pass
        cols.append(cp)
    prof['columns_profile']=cols
    prof['overall_quality_score']=_score(cols)
    prof['suggestions']=_suggest(cols)
    return prof

def _correlation(df):
    nums = df.select_dtypes(include=[np.number])
    if len(nums.columns)<2: return []
    corr = nums.corr().round(3)
    rels = []
    for i,ci in enumerate(corr.columns):
        for j,cj in enumerate(corr.columns):
            if i<j:
                v = float(corr.iloc[i,j])
                if not np.isnan(v):
                    rels.append({'from':ci,'to':cj,'value':v,'strength':'high' if abs(v)>0.7 else 'medium' if abs(v)>0.4 else 'low','direction':'positive' if v>0 else 'negative'})
    rels.sort(key=lambda x: abs(x['value']), reverse=True)
    return rels[:15]

def _distribution(df):
    dist = {}
    for col in df.select_dtypes(include=[np.number]).columns[:5]:
        vals = df[col].dropna()
        if len(vals)<2: continue
        dist[col] = {
            'min':safe_float(vals.min()),'max':safe_float(vals.max()),'mean':safe_float(vals.mean()),
            'median':safe_float(vals.median()),'std':safe_float(vals.std()),
            'skew':safe_float(vals.skew()) if len(vals)>2 else 0,
            'kurtosis':safe_float(vals.kurtosis()) if len(vals)>3 else 0,
            'outliers_iqr':int(((vals<(vals.quantile(0.25)-1.5*(vals.quantile(0.75)-vals.quantile(0.25))))|(vals>(vals.quantile(0.75)+1.5*(vals.quantile(0.75)-vals.quantile(0.25))))).sum())
        }
    return dist

def _score(profiles):
    if not profiles: return 100
    scs = []
    for p in profiles:
        s = 100
        if p['null_pct']>50: s-=35
        elif p['null_pct']>20: s-=20
        elif p['null_pct']>5: s-=8
        if p['unique_pct']>99 and p['null_pct']<50: s-=10
        if p['unique_pct']<1: s-=25
        scs.append(max(0,s))
    return round(sum(scs)/len(scs),1)

def _suggest(profiles):
    sug = []
    for p in profiles:
        if p['null_pct']>30: sug.append({'column':p['name'],'issue':f"Zeer hoge null-rate ({p['null_pct']}%)",'severity':'high','action':'Imputeer met mediaan/modus of overweeg kolom te verwijderen'})
        elif p['null_pct']>10: sug.append({'column':p['name'],'issue':f"Missende waarden ({p['null_pct']}%)",'severity':'medium','action':'Overweeg imputatie of markeer als optioneel'})
        elif p['null_pct']>2: sug.append({'column':p['name'],'issue':f"Enkele nulls ({p['null_pct']}%)",'severity':'low','action':'Controleer of nulls verwacht worden'})
        if p.get('zeros') and p['zeros']/(max(p['non_null'],1))>0.4: sug.append({'column':p['name'],'issue':f"Veel nullen (>{round(p['zeros']/max(p['non_null'],1)*100)}%)",'severity':'medium','action':'Controleer of 0 legitieme waarden of missende data zijn'})
        if p.get('negative') and p['negative']>0: sug.append({'column':p['name'],'issue':f"Negatieve waarden ({p['negative']}) in numerieke kolom",'severity':'medium','action':'Valideer of negatieve waarden zakelijk correct zijn'})
        if p['unique_pct']>95 and p['null_pct']<50: sug.append({'column':p['name'],'issue':'Nagenoeg uniek — vermoedelijk ID kolom','severity':'low','action':'Geschikt als primary key, niet voor aggregatie of model training'})
        if p['unique_pct']<1 and p['null_pct']<90: sug.append({'column':p['name'],'issue':'Constante kolom — geen variatie','severity':'low','action':'Overweeg deze kolom te verwijderen voor analyse'})
        if p.get('std') and p.get('mean') and p['mean']!=0 and p['std']/abs(p['mean'])>2: sug.append({'column':p['name'],'issue':'Hoge variabiliteit (CV > 200%)','severity':'medium','action':'Controleer op outliers, overweeg winsorization of log-transformatie'})
    return sug

def _checks(df, checks):
    res = {'checks':[],'summary':{'passed':0,'failed':0,'total':0}}
    for col in df.columns:
        if 'completeness' in checks:
            npct = float(df[col].isna().mean()*100)
            ok = npct<20
            res['checks'].append({'type':'completeness','column':col,'passed':ok,'value':round(npct,2),'threshold':'<20% null'})
            res['summary']['total']+=1
            if ok: res['summary']['passed']+=1
            else: res['summary']['failed']+=1
        if 'uniqueness' in checks:
            upct = float(df[col].nunique()/max(len(df),1)*100)
            ok = upct>1
            res['checks'].append({'type':'uniqueness','column':col,'passed':ok,'value':round(upct,2),'threshold':'>1% uniek'})
            res['summary']['total']+=1
            if ok: res['summary']['passed']+=1
            else: res['summary']['failed']+=1
        is_num = False
        try: is_num = np.issubdtype(df[col].dtype, np.number)
        except: pass
        if 'range' in checks and is_num:
            vals = df[col].dropna()
            if len(vals)>1:
                q1=float(vals.quantile(0.25)); q3=float(vals.quantile(0.75)); iqr=q3-q1
                out = int(((vals<(q1-1.5*iqr))|(vals>(q3+1.5*iqr))).sum())
                ok = out/max(len(vals),1)<0.1
                res['checks'].append({'type':'outliers','column':col,'passed':ok,'value':out,'threshold':'<10% outliers'})
                res['summary']['total']+=1
                if ok: res['summary']['passed']+=1
                else: res['summary']['failed']+=1
        if 'drift' in checks and is_num:
            vals = df[col].dropna()
            if len(vals)>1:
                m=float(vals.mean()); s=float(vals.std())
                ok = s/(abs(m)+1)<1.0 if m!=0 else True
                res['checks'].append({'type':'drift','column':col,'passed':ok,'value':round(s/max(abs(m),0.01),3),'threshold':'std/mean < 1.0'})
                res['summary']['total']+=1
                if ok: res['summary']['passed']+=1
                else: res['summary']['failed']+=1
        if 'type' in checks:
            res['checks'].append({'type':'type','column':col,'passed':True,'value':str(df[col].dtype),'threshold':'consistent'})
            res['summary']['total']+=1; res['summary']['passed']+=1
    sc = round((res['summary']['passed']/max(res['summary']['total'],1))*100,1)
    res['quality_score']=sc
    res['grade']='A' if sc>=90 else 'B' if sc>=75 else 'C' if sc>=50 else 'D' if sc>=25 else 'F'
    return res

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=int(os.environ.get('QS_PORT',3002)), debug=False)