ffmpeg task
(file_path: FilePath, title: str, db: Session)
| 38 | |
| 39 | # FastAPI BackgroundTasks |
| 40 | def bg_task(file_path: FilePath, title: str, db: Session): |
| 41 | """ffmpeg task""" |
| 42 | output_segment_path = audio_output_path / title |
| 43 | os.makedirs(output_segment_path, exist_ok=True) |
| 44 | ffmpeg.input(file_path).output(f'{output_segment_path}/%03d.wav', f='segment', segment_time='49', c='copy', ac='2', ar='16000').run() |
| 45 | |
| 46 | # update audio segement database(DATA) by looping folder |
| 47 | db_audio = crud.get_audio_by_title(db=db, title=title) |
| 48 | |
| 49 | for x in output_segment_path.iterdir(): |
| 50 | if x.is_file(): |
| 51 | data = { |
| 52 | "segment_title": x.name |
| 53 | } |
| 54 | crud.create_audio_segment(db = db, data = data, audio_id = db_audio.id) |
| 55 | |
| 56 | # run Celery task |
| 57 | celery_worker.create_task.delay(title = title, audio_id = db_audio.id) |
| 58 | |
| 59 | |
| 60 | @application.post("/") |
nothing calls this directly
no outgoing calls
no test coverage detected