| 473 | @router.post("/uploadDsSchema/{id}", response_model=None, summary=f"{PLACEHOLDER_PREFIX}ds_upload_ds_schema") |
| 474 | @require_permissions(permission=SqlbotPermission(role=['ws_admin'], type='ds', keyExpression="id")) |
| 475 | async def upload_ds_schema(session: SessionDep, id: int = Path(..., description=f"{PLACEHOLDER_PREFIX}ds_id"), |
| 476 | file: UploadFile = File(...)): |
| 477 | ALLOWED_EXTENSIONS = {"xlsx", "xls"} |
| 478 | if not file.filename.lower().endswith(tuple(ALLOWED_EXTENSIONS)): |
| 479 | raise HTTPException(400, "Only support .xlsx/.xls") |
| 480 | |
| 481 | try: |
| 482 | contents = await file.read() |
| 483 | excel_file = io.BytesIO(contents) |
| 484 | |
| 485 | sheet_names = pd.ExcelFile(excel_file, engine="openpyxl").sheet_names |
| 486 | |
| 487 | excel_file.seek(0) |
| 488 | |
| 489 | field_sheets = [] |
| 490 | table_sheet = None # [] |
| 491 | for sheet in sheet_names: |
| 492 | df = pd.read_excel(excel_file, sheet_name=sheet, engine="openpyxl").fillna('') |
| 493 | if sheet == t_sheet: |
| 494 | table_sheet = df.where(pd.notnull(df), None).to_dict(orient="records") |
| 495 | else: |
| 496 | field_sheets.append( |
| 497 | {'sheet_name': sheet, 'data': df.where(pd.notnull(df), None).to_dict(orient="records")}) |
| 498 | |
| 499 | # print(field_sheets) |
| 500 | |
| 501 | # sheet table mapping |
| 502 | sheet_table_map = {} |
| 503 | |
| 504 | # get data and update |
| 505 | # update table comment |
| 506 | if table_sheet and len(table_sheet) > 0: |
| 507 | for table in table_sheet: |
| 508 | sheet_table_map[table[t_s_col]] = table[t_n_col] |
| 509 | session.query(CoreTable).filter( |
| 510 | and_(CoreTable.ds_id == id, CoreTable.table_name == table[t_n_col])).update( |
| 511 | {'custom_comment': table[t_c_col]}) |
| 512 | |
| 513 | # update field comment |
| 514 | if field_sheets and len(field_sheets) > 0: |
| 515 | for fields in field_sheets: |
| 516 | if len(fields['data']) > 0: |
| 517 | # get table id |
| 518 | table_name = sheet_table_map.get(fields['sheet_name']) |
| 519 | table = session.query(CoreTable).filter( |
| 520 | and_(CoreTable.ds_id == id, CoreTable.table_name == table_name)).first() |
| 521 | if table: |
| 522 | for field in fields['data']: |
| 523 | session.query(CoreField).filter( |
| 524 | and_(CoreField.ds_id == id, |
| 525 | CoreField.table_id == table.id, |
| 526 | CoreField.field_name == field[f_n_col])).update( |
| 527 | {'custom_comment': field[f_c_col]}) |
| 528 | session.commit() |
| 529 | |
| 530 | return True |
| 531 | except Exception as e: |
| 532 | raise HTTPException(status_code=500, detail=f"Parse Excel Failed: {str(e)}") |