import io, os, tempfile from datetime import datetime, timezone from typing import Any import pandas as pd from fastapi import HTTPException from sqlalchemy import select from sqlalchemy.orm import Session from app.models.entities import Dataset, IngestionRun, JobStatus, DatasetPermission, PermissionEffect, User, Role, TeamMember, CalculatedField from app.services.storage import ObjectStorage ADMIN_ROLES={Role.admin,Role.director,Role.data_engineer,Role.data_steward,Role.auditor} ALLOWED_AGGS={'sum','avg','min','max','count','count_distinct'} ALLOWED_OPS={'eq','ne','gt','gte','lt','lte','contains','in','not_in','is_null','not_null'} def _active_permissions(db:Session,user:User,dataset_id:str): now=datetime.now(timezone.utc) candidates=[] team_ids={x for x in db.scalars(select(TeamMember.team_id).where(TeamMember.user_id==user.id)).all()} for p in db.scalars(select(DatasetPermission).where(DatasetPermission.dataset_id==dataset_id)).all(): matches=(p.subject_type=='user' and p.subject_id==user.id) or (p.subject_type=='role' and p.subject_id==user.role.value) or (p.subject_type=='department' and p.subject_id==user.department_id) or (p.subject_type=='team' and p.subject_id in team_ids) if matches and (not p.valid_from or p.valid_from<=now) and (not p.valid_until or p.valid_until>=now): candidates.append(p) return candidates def authorize_dataset(db:Session,user:User,dataset:Dataset,action='read'): if user.role in ADMIN_ROLES: return {'columns':None,'row_filters':[]} if dataset.owner_department_id and dataset.owner_department_id==user.department_id: return {'columns':None,'row_filters':[dataset.row_filter] if dataset.row_filter else []} permissions=_active_permissions(db,user,dataset.id) if any(p.effect==PermissionEffect.deny and action in p.actions for p in permissions): raise HTTPException(403,'Acesso negado por política') allows=[p for p in permissions if p.effect==PermissionEffect.allow and action in p.actions] if not allows: raise HTTPException(403,'Sem permissão para este dataset') allowed=set(); denied=set(); unrestricted=False; filters=[] for p in allows: if not p.allowed_columns: unrestricted=True allowed.update(p.allowed_columns or []) denied.update(p.denied_columns or []) filters.append(p.row_filter or {}) columns=None if unrestricted else sorted(allowed-denied) return {'columns':columns,'denied_columns':sorted(denied),'row_filters':filters} def latest_frame(db:Session,dataset:Dataset)->pd.DataFrame: run=db.scalar(select(IngestionRun).where(IngestionRun.dataset_id==dataset.id,IngestionRun.status==JobStatus.success,IngestionRun.object_key.is_not(None)).order_by(IngestionRun.finished_at.desc()).limit(1)) if not run or not run.object_key: raise HTTPException(409,'Dataset ainda não possui uma carga concluída') storage=ObjectStorage(); storage.ensure_bucket() with tempfile.NamedTemporaryFile(suffix='.parquet',delete=False) as f: path=f.name try: storage.client.download_file(storage.bucket,run.object_key,path) return pd.read_parquet(path) finally: if os.path.exists(path): os.unlink(path) def mask_frame(df:pd.DataFrame,dataset:Dataset)->pd.DataFrame: out=df.copy() for col in dataset.sensitive_columns or []: if col not in out.columns: continue policy=(dataset.masking_policy or {}).get(col,'redact') if policy=='hash': out[col]=out[col].astype(str).map(lambda v: __import__('hashlib').sha256(v.encode()).hexdigest()) elif policy=='last4': out[col]=out[col].astype(str).map(lambda v: ('*'*max(0,len(v)-4))+v[-4:]) elif policy=='email': out[col]=out[col].astype(str).map(lambda v: (v[:1]+'***@'+v.split('@',1)[1]) if '@' in v else '***') else: out[col]='[REDACTED]' return out def _apply_one(df,rule): col=rule.get('column'); op=rule.get('op'); value=rule.get('value') if col not in df.columns or op not in ALLOWED_OPS: raise HTTPException(400,'Filtro inválido') s=df[col] if op=='eq': return df[s==value] if op=='ne': return df[s!=value] if op=='gt': return df[s>value] if op=='gte': return df[s>=value] if op=='lt': return df[sdict[str,Any]: access=authorize_dataset(db,user,dataset,'read') df=latest_frame(db,dataset) allowed_cols=access['columns'] if allowed_cols is not None: df=df[[c for c in df.columns if c in allowed_cols]] denied=set(access.get('denied_columns') or []) if denied: df=df[[c for c in df.columns if c not in denied]] for rf in access['row_filters']: if rf and rf.get('column'): df=_apply_one(df,rf) # Campos calculados: expressões pandas limitadas, persistidas pelo Studio. for cf in definition.get('calculated_fields',[]): name=cf.get('name'); expr=cf.get('expression') if name and expr: try: df[name]=df.eval(expr,engine='python') except Exception as exc: raise HTTPException(400,f'Campo calculado inválido: {name}') from exc for rule in definition.get('filters',[]): df=_apply_one(df,rule) dimensions=[c for c in definition.get('dimensions',[]) if c in df.columns] measures=definition.get('measures',[]) if measures: named={} for m in measures: col=m.get('column'); agg=m.get('aggregation','sum'); alias=m.get('alias') or f'{agg}_{col}' if agg not in ALLOWED_AGGS or (col not in df.columns and agg!='count'): raise HTTPException(400,'Métrica inválida') named[alias]=(col, 'nunique' if agg=='count_distinct' else ('mean' if agg=='avg' else agg)) if dimensions: result=df.groupby(dimensions,dropna=False).agg(**named).reset_index() else: result=pd.DataFrame([{alias:(len(df) if spec[1]=='count' else getattr(df[spec[0]],spec[1])()) for alias,spec in named.items()}]) else: cols=[c for c in definition.get('columns',[]) if c in df.columns] or list(df.columns) result=df[cols] for sort in definition.get('sort',[]): col=sort.get('column') if col in result.columns: result=result.sort_values(col,ascending=sort.get('direction','asc')!='desc') result=mask_frame(result,dataset).head(min(max_rows,50000)) return {'columns':list(result.columns),'rows':result.where(pd.notnull(result),None).to_dict(orient='records'),'row_count':len(result)}