()
| 73 | async def export_excel(session: SessionDep, trans: Trans, current_user: CurrentUser, |
| 74 | question: Optional[str] = Query(None, description="搜索术语(可选)")): |
| 75 | def inner(): |
| 76 | _list = get_all_data_training(session, question, oid=current_user.oid) |
| 77 | |
| 78 | data_list = [] |
| 79 | for obj in _list: |
| 80 | _data = { |
| 81 | "question": obj.question, |
| 82 | "description": obj.description, |
| 83 | "datasource_name": obj.datasource_name, |
| 84 | "advanced_application_name": obj.advanced_application_name, |
| 85 | } |
| 86 | data_list.append(_data) |
| 87 | |
| 88 | fields = [] |
| 89 | fields.append(AxisObj(name=trans('i18n_data_training.problem_description'), value='question')) |
| 90 | fields.append(AxisObj(name=trans('i18n_data_training.sample_sql'), value='description')) |
| 91 | fields.append(AxisObj(name=trans('i18n_data_training.effective_data_sources'), value='datasource_name')) |
| 92 | if current_user.oid == 1: |
| 93 | fields.append( |
| 94 | AxisObj(name=trans('i18n_data_training.advanced_application'), value='advanced_application_name')) |
| 95 | |
| 96 | md_data, _fields_list = DataFormat.convert_object_array_for_pandas(fields, data_list) |
| 97 | |
| 98 | df = pd.DataFrame(md_data, columns=_fields_list) |
| 99 | |
| 100 | buffer = io.BytesIO() |
| 101 | |
| 102 | with pd.ExcelWriter(buffer, engine='xlsxwriter', |
| 103 | engine_kwargs={'options': {'strings_to_numbers': False}}) as writer: |
| 104 | df.to_excel(writer, sheet_name='Sheet1', index=False) |
| 105 | |
| 106 | buffer.seek(0) |
| 107 | return io.BytesIO(buffer.getvalue()) |
| 108 | |
| 109 | result = await asyncio.to_thread(inner) |
| 110 | return StreamingResponse(result, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") |
nothing calls this directly
no test coverage detected