287 lines
14 KiB
Python
287 lines
14 KiB
Python
|
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)
|