samelias1 commited on
Commit
0900ba8
Β·
verified Β·
1 Parent(s): d739d0a

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +698 -704
app.py CHANGED
@@ -1,704 +1,698 @@
1
- import os
2
- import json
3
- import requests
4
- import subprocess
5
- import shutil
6
- import time
7
- import sys
8
- import threading
9
- from typing import Dict, List, Optional, Any
10
- from huggingface_hub import HfApi, hf_hub_url
11
- from fastapi import FastAPI, HTTPException
12
- from fastapi.responses import JSONResponse
13
- import uvicorn
14
-
15
- # Fix Unicode encoding for Windows
16
- if sys.platform == 'win32':
17
- import io
18
- sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
19
-
20
- # Initialize FastAPI app
21
- app = FastAPI(title="Audio Transcriber", description="Audio transcription and upload service")
22
-
23
- # ==== CONFIGURATION ====
24
- HF_TOKEN = os.environ.get("HF_TOKEN", "")
25
- SOURCE_REPO_ID = "Samfredoly/BG_Vid" # Fetch audio files from here
26
- TARGET_REPO_ID = "samfred2/A_Text" # Upload transcriptions here
27
- REFERENCE_REPO_ID = "Fred808/BG3" # Reference repo to match audio filenames
28
-
29
- # Path Configuration
30
- DOWNLOAD_FOLDER = "downloads_audio"
31
- TRANSCRIPTIONS_FOLDER = "transcriptions"
32
- LOCAL_STATE_FOLDER = ".state_audio"
33
-
34
- os.makedirs(DOWNLOAD_FOLDER, exist_ok=True)
35
- os.makedirs(TRANSCRIPTIONS_FOLDER, exist_ok=True)
36
- os.makedirs(LOCAL_STATE_FOLDER, exist_ok=True)
37
-
38
- # State Files
39
- FAILED_FILES_LOG = "failed_audio_files.log"
40
- HF_STATE_FILE = "processing_audio_state.json"
41
-
42
- # Processing Parameters
43
- PROCESSING_DELAY = 2
44
- MAX_RETRIES = 3
45
- MIN_FREE_SPACE_GB = 1
46
- WHISPER_MODEL = "small" # Whisper model size
47
-
48
- # Initialize HF API
49
- hf_api = HfApi(token=HF_TOKEN)
50
-
51
- # Global State
52
- processing_status = {
53
- "is_running": False,
54
- "current_file": None,
55
- "total_files": 0,
56
- "processed_files": 0,
57
- "failed_files": 0,
58
- "transcribed_files": 0,
59
- "last_update": None,
60
- "logs": []
61
- }
62
-
63
- def log_message(message: str, level: str = "INFO"):
64
- """Log messages with timestamp"""
65
- timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
66
- log_entry = f"[{timestamp}] {level}: {message}"
67
- print(log_entry)
68
- processing_status["logs"].append(log_entry)
69
- processing_status["last_update"] = timestamp
70
- if len(processing_status["logs"]) > 100:
71
- processing_status["logs"] = processing_status["logs"][-100:]
72
-
73
- def log_failed_file(filename: str, error: str):
74
- """Log failed files to persistent file"""
75
- with open(FAILED_FILES_LOG, "a") as f:
76
- f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} - {filename}: {error}\n")
77
-
78
- def get_disk_usage(path: str) -> Dict[str, float]:
79
- """Get disk usage statistics in GB"""
80
- statvfs = os.statvfs(path)
81
- total = statvfs.f_frsize * statvfs.f_blocks / (1024**3)
82
- free = statvfs.f_frsize * statvfs.f_bavail / (1024**3)
83
- used = total - free
84
- return {"total": total, "free": free, "used": used}
85
-
86
- def check_disk_space(path: str = ".") -> bool:
87
- """Check if there's enough disk space"""
88
- disk_info = get_disk_usage(path)
89
- if disk_info["free"] < MIN_FREE_SPACE_GB:
90
- log_message(f'⚠️ Low disk space: {disk_info["free"]:.2f}GB free, {disk_info["used"]:.2f}GB used')
91
- return False
92
- return True
93
-
94
- def cleanup_temp_files():
95
- """Clean up temporary files to free space"""
96
- log_message("🧹 Cleaning up temporary files...", "INFO")
97
-
98
- current_file = processing_status.get("current_file")
99
- for file in os.listdir(DOWNLOAD_FOLDER):
100
- if file != current_file and file.endswith((".wav", ".mp3")):
101
- try:
102
- os.remove(os.path.join(DOWNLOAD_FOLDER, file))
103
- log_message(f"πŸ—‘οΈ Removed old download: {file}", "INFO")
104
- except:
105
- pass
106
-
107
- def load_json_state(file_path: str, default_value: Dict[str, Any]) -> Dict[str, Any]:
108
- """Load state from JSON file with migration logic for new structure."""
109
- if os.path.exists(file_path):
110
- try:
111
- with open(file_path, "r") as f:
112
- data = json.load(f)
113
-
114
- if "file_states" not in data or not isinstance(data["file_states"], dict):
115
- log_message("ℹ️ Initializing 'file_states' dictionary.", "INFO")
116
- data["file_states"] = {}
117
-
118
- if "next_download_index" not in data:
119
- data["next_download_index"] = 0
120
-
121
- return data
122
- except json.JSONDecodeError:
123
- log_message(f"⚠️ Corrupted state file: {file_path}", "WARNING")
124
- return default_value
125
-
126
- def save_json_state(file_path: str, data: Dict[str, Any]):
127
- """Save state to JSON file"""
128
- with open(file_path, "w") as f:
129
- json.dump(data, f, indent=2)
130
-
131
- def download_hf_state(repo_id: str, filename: str) -> Dict[str, Any]:
132
- """Downloads the state file from Hugging Face or returns a default state."""
133
- local_path = os.path.join(LOCAL_STATE_FOLDER, filename)
134
- default_state = {"next_download_index": 0, "file_states": {}}
135
-
136
- try:
137
- files = hf_api.list_repo_files(repo_id=repo_id, repo_type="dataset")
138
- if filename not in files:
139
- log_message(f"ℹ️ State file {filename} not found in {repo_id}. Starting from default state.", "INFO")
140
- return default_state
141
-
142
- from huggingface_hub import hf_hub_download
143
- hf_hub_download(
144
- repo_id=repo_id,
145
- filename=filename,
146
- repo_type="dataset",
147
- local_dir=LOCAL_STATE_FOLDER,
148
- local_dir_use_symlinks=False
149
- )
150
-
151
- log_message(f"βœ… Successfully downloaded state file from {repo_id}.", "INFO")
152
- return load_json_state(local_path, default_state)
153
-
154
- except Exception as e:
155
- log_message(f"⚠️ Failed to download state file from Hugging Face: {str(e)}. Starting from default state.", "WARNING")
156
- return default_state
157
-
158
- def upload_hf_state(repo_id: str, filename: str, state: Dict[str, Any]) -> bool:
159
- """Uploads the state file to Hugging Face."""
160
- local_path = os.path.join(LOCAL_STATE_FOLDER, filename)
161
-
162
- try:
163
- save_json_state(local_path, state)
164
-
165
- hf_api.upload_file(
166
- path_or_fileobj=local_path,
167
- path_in_repo=filename,
168
- repo_id=repo_id,
169
- repo_type="dataset",
170
- commit_message=f"Update audio processing state: next_index={state['next_download_index']}"
171
- )
172
- log_message(f"βœ… Successfully uploaded updated state file to {repo_id}", "INFO")
173
- return True
174
- except Exception as e:
175
- log_message(f"❌ Failed to upload state file to Hugging Face: {str(e)}", "ERROR")
176
- return False
177
-
178
- def lock_file_for_processing(wav_filename: str, state: Dict[str, Any]) -> bool:
179
- """Marks a file as 'processing' in the state file and uploads the lock."""
180
- log_message(f"πŸ”’ Attempting to lock file: {wav_filename} (Marking as 'processing')", "INFO")
181
-
182
- state["file_states"][wav_filename] = "processing"
183
-
184
- if upload_hf_state(TARGET_REPO_ID, HF_STATE_FILE, state):
185
- log_message(f"βœ… Successfully locked file: {wav_filename}", "INFO")
186
- return True
187
- else:
188
- log_message(f"❌ Failed to upload lock for file: {wav_filename}. Aborting processing.", "ERROR")
189
- if wav_filename in state["file_states"]:
190
- del state["file_states"][wav_filename]
191
- return False
192
-
193
- def unlock_file_as_processed(wav_filename: str, state: Dict[str, Any], next_index: int) -> bool:
194
- """Marks a file as 'processed', updates the index, and uploads the state."""
195
- log_message(f"πŸ”“ Attempting to unlock file: {wav_filename} (Marking as 'processed')", "INFO")
196
-
197
- state["file_states"][wav_filename] = "processed"
198
- state["next_download_index"] = next_index
199
-
200
- if upload_hf_state(TARGET_REPO_ID, HF_STATE_FILE, state):
201
- log_message(f"βœ… Successfully unlocked and marked as processed: {wav_filename}", "INFO")
202
- return True
203
- else:
204
- log_message(f"❌ Failed to upload final state for file: {wav_filename}.", "ERROR")
205
- return False
206
-
207
- def download_with_retry(url: str, dest_path: str, max_retries: int = 3) -> bool:
208
- """Download file with retry logic and disk space checking"""
209
- if not check_disk_space():
210
- cleanup_temp_files()
211
- if not check_disk_space():
212
- log_message("❌ Insufficient disk space even after cleanup", "ERROR")
213
- return False
214
-
215
- try:
216
- os.makedirs(os.path.dirname(dest_path), exist_ok=True)
217
- except Exception as e:
218
- log_message(f"❌ Failed to create directory for download path {os.path.dirname(dest_path)}: {str(e)}", "ERROR")
219
- return False
220
-
221
- headers = {"Authorization": f"Bearer {HF_TOKEN}"}
222
- for attempt in range(max_retries):
223
- try:
224
- with requests.get(url, headers=headers, stream=True) as r:
225
- r.raise_for_status()
226
-
227
- with open(dest_path, "wb") as f:
228
- for chunk in r.iter_content(chunk_size=8192):
229
- if chunk:
230
- f.write(chunk)
231
-
232
- log_message(f"βœ… Download successful: {dest_path}", "INFO")
233
- return True
234
-
235
- except requests.exceptions.RequestException as e:
236
- log_message(f"❌ Download attempt {attempt + 1} failed for {url}: {str(e)}", "WARNING")
237
- time.sleep(PROCESSING_DELAY)
238
- except Exception as e:
239
- log_message(f"❌ An unexpected error occurred during download: {str(e)}", "ERROR")
240
- return False
241
-
242
- log_message(f"❌ Failed to download {url} after {max_retries} attempts.", "ERROR")
243
- return False
244
-
245
- def fetch_reference_files(repo_id: str) -> Dict[str, str]:
246
- """Fetch all files from Fred808/BG3 repo to match with audio filenames."""
247
- log_message(f"πŸ“‹ Fetching file list from {repo_id}...", "INFO")
248
-
249
- try:
250
- files_list = hf_api.list_repo_files(repo_id=repo_id, repo_type="dataset")
251
-
252
- # Include all file types (zip, rar, wav, mp3, etc.)
253
- all_files = [f for f in files_list]
254
-
255
- # Create a mapping of base filename (without extension) to full path
256
- filename_map = {}
257
- for file_path in all_files:
258
- base_name = os.path.splitext(os.path.basename(file_path))[0]
259
- filename_map[base_name] = file_path
260
-
261
- log_message(f"βœ… Found {len(filename_map)} files in reference repo", "INFO")
262
- return filename_map
263
-
264
- except Exception as e:
265
- log_message(f"❌ Failed to fetch reference files: {str(e)}", "ERROR")
266
- return {}
267
-
268
- def find_matching_filename(transcribed_filename: str, reference_map: Dict[str, str]) -> Optional[str]:
269
- """Find matching filename in reference map from Fred808/BG3."""
270
- base_name = os.path.splitext(transcribed_filename)[0]
271
-
272
- # Exact match first
273
- if base_name in reference_map:
274
- full_path = reference_map[base_name]
275
- print(f"\nβœ… MATCH FOUND:")
276
- print(f" Audio: {transcribed_filename}")
277
- print(f" File: {full_path}")
278
- log_message(f"βœ… Found exact match: {transcribed_filename} -> {full_path}", "INFO")
279
- return full_path
280
-
281
- # Partial/fuzzy match (check if reference contains transcribed as substring)
282
- matches = []
283
- for ref_base, ref_full_path in reference_map.items():
284
- if base_name.lower() in ref_base.lower() or ref_base.lower() in base_name.lower():
285
- matches.append((ref_base, ref_full_path))
286
-
287
- # Return first partial match if found
288
- if matches:
289
- ref_base, ref_full_path = matches[0]
290
- print(f"\nβœ… PARTIAL MATCH FOUND:")
291
- print(f" Audio: {transcribed_filename}")
292
- print(f" File: {ref_full_path}")
293
- log_message(f"βœ… Found partial match: {transcribed_filename} -> {ref_full_path}", "INFO")
294
- return ref_full_path
295
-
296
- print(f"\n❌ NO MATCH FOUND:")
297
- print(f" Audio: {transcribed_filename}")
298
- log_message(f"⚠️ No matching filename found for: {transcribed_filename}", "WARNING")
299
- return None
300
-
301
- def transcribe_audio(wav_path: str) -> Optional[Dict[str, Any]]:
302
- """Transcribe audio file using Whisper from Transformers."""
303
- log_message(f"🎀 Transcribing audio file: {wav_path}", "INFO")
304
-
305
- try:
306
- from transformers import pipeline
307
- import librosa
308
-
309
- # Load audio with librosa
310
- log_message(f"Loading audio file: {wav_path}", "INFO")
311
- audio, sr = librosa.load(wav_path, sr=16000)
312
-
313
- # Initialize Whisper pipeline
314
- log_message(f"Loading Whisper {WHISPER_MODEL} model from Transformers...", "INFO")
315
- pipe = pipeline(
316
- "automatic-speech-recognition",
317
- model=f"openai/whisper-{WHISPER_MODEL}",
318
- device=0 if __import__('torch').cuda.is_available() else -1 # GPU if available, else CPU
319
- )
320
-
321
- # Transcribe
322
- log_message("Transcribing audio...", "INFO")
323
- result = pipe(audio)
324
-
325
- # Format result to match openai-whisper format
326
- formatted_result = {
327
- "text": result["text"],
328
- "segments": [{"text": result["text"]}]
329
- }
330
-
331
- log_message(f"βœ… Successfully transcribed: {wav_path}", "INFO")
332
- return formatted_result
333
-
334
- except ImportError as e:
335
- missing_lib = str(e)
336
- log_message(f"❌ Missing library. Install with: pip install transformers librosa torch torchaudio", "ERROR")
337
- log_message(f" Error: {missing_lib}", "ERROR")
338
- return None
339
- except Exception as e:
340
- log_message(f"❌ Failed to transcribe {wav_path}: {str(e)}", "ERROR")
341
- return None
342
-
343
- def process_audio_file(wav_path: str, reference_map: Dict[str, str], matched_filename: str) -> bool:
344
- """
345
- Main processing logic for a single audio file:
346
- 1. Transcribe using Whisper
347
- 2. Save transcription as JSON
348
- 3. Upload to HF dataset
349
- 4. Clean up local files
350
- """
351
- wav_filename = os.path.basename(wav_path)
352
-
353
- # 1. Transcribe audio
354
- transcription = transcribe_audio(wav_path)
355
- if transcription is None:
356
- log_failed_file(wav_filename, "Transcription failed")
357
- return False
358
-
359
- # 2. Save transcription as JSON
360
- json_filename = os.path.splitext(matched_filename)[0] + "_transcription.json"
361
- json_output_path = os.path.join(TRANSCRIPTIONS_FOLDER, json_filename)
362
-
363
- try:
364
- os.makedirs(os.path.dirname(json_output_path), exist_ok=True)
365
-
366
- with open(json_output_path, "w", encoding="utf-8") as f:
367
- json.dump(transcription, f, indent=2, ensure_ascii=False)
368
-
369
- log_message(f"βœ… Saved transcription: {json_output_path}", "INFO")
370
-
371
- except Exception as e:
372
- log_message(f"❌ Failed to save transcription JSON: {str(e)}", "ERROR")
373
- log_failed_file(wav_filename, f"Failed to save JSON: {str(e)}")
374
- return False
375
-
376
- # 3. Upload to HF dataset
377
- try:
378
- path_in_repo = f"transcriptions/{json_filename}"
379
- commit_message = f"Add transcription for: {matched_filename}"
380
-
381
- hf_api.upload_file(
382
- path_or_fileobj=json_output_path,
383
- path_in_repo=path_in_repo,
384
- repo_id=TARGET_REPO_ID,
385
- repo_type="dataset",
386
- commit_message=commit_message
387
- )
388
- log_message(f"βœ… Successfully uploaded transcription: {json_filename}", "INFO")
389
- processing_status["transcribed_files"] += 1
390
-
391
- except Exception as e:
392
- log_message(f"❌ Failed to upload transcription to HF: {str(e)}", "ERROR")
393
- log_failed_file(wav_filename, f"Failed to upload: {str(e)}")
394
- return False
395
-
396
- # 4. Clean up local files
397
- try:
398
- os.remove(json_output_path)
399
- log_message(f"πŸ—‘οΈ Cleaned up local transcription file: {json_output_path}", "INFO")
400
- except:
401
- pass
402
-
403
- return True
404
-
405
- def get_next_file_to_process(repo_id: str, state: Dict[str, Any]) -> Optional[Dict[str, Any]]:
406
- """
407
- Finds the next audio file to process from the source repo in reverse order (oldest to newest).
408
- Returns: { 'filename': str, 'url': str, 'index': int } or None
409
- """
410
- log_message(f"πŸ” Searching for next audio file to process in {repo_id}", "INFO")
411
-
412
- try:
413
- files_list = hf_api.list_repo_files(repo_id=repo_id, repo_type="dataset")
414
-
415
- # Filter for audio files and sort in reverse order (descending)
416
- audio_files = sorted([f for f in files_list if f.endswith(('.wav', '.mp3'))], reverse=True)
417
-
418
- if not audio_files:
419
- log_message("ℹ️ No audio files found in the source repository.", "INFO")
420
- return None
421
-
422
- processing_status["total_files"] = len(audio_files)
423
-
424
- start_index = state.get("next_download_index", 0)
425
-
426
- for index in range(start_index, len(audio_files)):
427
- filename = audio_files[index]
428
- file_state = state["file_states"].get(filename)
429
-
430
- if file_state is None or file_state == "failed":
431
- url = hf_hub_url(repo_id=repo_id, filename=filename, repo_type="dataset", subfolder=None)
432
-
433
- log_message(f"βœ… Found next audio file: {filename} at index {index}", "INFO")
434
- return {
435
- 'filename': filename,
436
- 'url': url,
437
- 'index': index
438
- }
439
-
440
- elif file_state == "processing":
441
- log_message(f"⚠️ File {filename} is currently marked as 'processing'. Skipping for now.", "WARNING")
442
-
443
- elif file_state == "processed":
444
- log_message(f"ℹ️ File {filename} already processed. Skipping.", "INFO")
445
-
446
- log_message("ℹ️ All files up to the current index have been processed or skipped.", "INFO")
447
-
448
- if start_index >= len(audio_files):
449
- log_message("ℹ️ Reached end of file list. Resetting index to 0 for next loop.", "INFO")
450
- state["next_download_index"] = 0
451
- upload_hf_state(TARGET_REPO_ID, HF_STATE_FILE, state)
452
-
453
- return None
454
-
455
- except Exception as e:
456
- log_message(f"❌ Failed to list files from Hugging Face: {str(e)}", "ERROR")
457
- return None
458
-
459
- def main_processing_loop():
460
- """The main loop that orchestrates the download, transcription, and upload cycle."""
461
-
462
- if processing_status["is_running"]:
463
- log_message("⚠️ Processing loop is already running.", "WARNING")
464
- return
465
-
466
- processing_status["is_running"] = True
467
-
468
- try:
469
- log_message("πŸš€ Starting audio transcription processing loop...", "INFO")
470
-
471
- # Fetch reference files from BG_Vid repo once at the start
472
- reference_map = fetch_reference_files(REFERENCE_REPO_ID)
473
-
474
- if not reference_map:
475
- log_message("❌ No reference files found. Cannot proceed.", "ERROR")
476
- return
477
-
478
- while processing_status["is_running"]:
479
-
480
- current_state = download_hf_state(TARGET_REPO_ID, HF_STATE_FILE)
481
- next_file_info = get_next_file_to_process(SOURCE_REPO_ID, current_state)
482
-
483
- if next_file_info is None:
484
- log_message("πŸ’€ No new audio files to process. Sleeping for a while...", "INFO")
485
- time.sleep(PROCESSING_DELAY * 5)
486
- continue
487
-
488
- target_file = next_file_info['filename']
489
- audio_url = next_file_info['url']
490
- target_index = next_file_info['index']
491
-
492
- processing_status["current_file"] = target_file
493
- success = False
494
- matched_filename = None
495
-
496
- try:
497
- if not lock_file_for_processing(target_file, current_state):
498
- log_message(f"❌ Failed to lock file {target_file}. Skipping.", "ERROR")
499
- time.sleep(PROCESSING_DELAY)
500
- continue
501
-
502
- local_wav_path = os.path.join(DOWNLOAD_FOLDER, os.path.basename(target_file))
503
- log_message(f"⬇️ Downloading audio file: {target_file}", "INFO")
504
-
505
- if download_with_retry(audio_url, local_wav_path):
506
-
507
- # Extract base filename for matching
508
- base_filename = os.path.basename(target_file)
509
- matched_filename = find_matching_filename(base_filename, reference_map)
510
-
511
- if matched_filename:
512
- if process_audio_file(local_wav_path, reference_map, matched_filename):
513
- success = True
514
- log_message(f"βœ… Finished processing: {target_file}", "INFO")
515
- else:
516
- log_message(f"❌ Processing failed for: {target_file}", "ERROR")
517
- else:
518
- log_message(f"❌ No matching filename found for: {base_filename}", "ERROR")
519
- log_failed_file(target_file, "No matching reference filename")
520
- else:
521
- log_message(f"❌ Download failed for: {target_file}", "ERROR")
522
-
523
- except Exception as e:
524
- log_message(f"πŸ”₯ An unhandled error occurred while processing {target_file}: {str(e)}", "ERROR")
525
- log_failed_file(target_file, str(e))
526
-
527
- finally:
528
- next_index_to_save = target_index + 1
529
- current_state = download_hf_state(TARGET_REPO_ID, HF_STATE_FILE)
530
-
531
- if success:
532
- unlock_file_as_processed(target_file, current_state, next_index_to_save)
533
- processing_status["processed_files"] += 1
534
- else:
535
- log_message(f"⚠️ Processing failed for {target_file}. Marking as 'failed' and advancing index.", "WARNING")
536
- current_state["file_states"][target_file] = "failed"
537
- current_state["next_download_index"] = next_index_to_save
538
- upload_hf_state(TARGET_REPO_ID, HF_STATE_FILE, current_state)
539
- processing_status["failed_files"] += 1
540
-
541
- if os.path.exists(local_wav_path):
542
- os.remove(local_wav_path)
543
- log_message(f"πŸ—‘οΈ Cleaned up local file: {local_wav_path}", "INFO")
544
-
545
- time.sleep(PROCESSING_DELAY)
546
-
547
- log_message("πŸŽ‰ Processing complete!", "INFO")
548
- log_message(f"πŸ“Š Final stats: {processing_status['transcribed_files']} audio files transcribed, {processing_status['processed_files']} files processed", "INFO")
549
-
550
- except KeyboardInterrupt:
551
- log_message("⏹️ Processing interrupted by user", "WARNING")
552
- except Exception as e:
553
- log_message(f"❌ Fatal error: {str(e)}", "ERROR")
554
- finally:
555
- processing_status["is_running"] = False
556
- cleanup_temp_files()
557
-
558
- if __name__ == "__main__":
559
- main_processing_loop()
560
-
561
- # ===== FASTAPI ENDPOINTS =====
562
-
563
- @app.get("/")
564
- async def root():
565
- """Root endpoint with service info"""
566
- return {
567
- "service": "Audio Transcriber",
568
- "status": "running",
569
- "version": "1.0.0",
570
- "endpoints": {
571
- "status": "/status",
572
- "start": "/start",
573
- "stop": "/stop",
574
- "process": "/process/{filename}",
575
- "logs": "/logs"
576
- }
577
- }
578
-
579
- @app.get("/status")
580
- async def get_status():
581
- """Get current processing status"""
582
- return {
583
- "is_running": processing_status["is_running"],
584
- "current_file": processing_status["current_file"],
585
- "total_files": processing_status["total_files"],
586
- "processed_files": processing_status["processed_files"],
587
- "transcribed_files": processing_status["transcribed_files"],
588
- "failed_files": processing_status["failed_files"],
589
- "last_update": processing_status["last_update"],
590
- "recent_logs": processing_status["logs"][-10:]
591
- }
592
-
593
- @app.post("/start")
594
- async def start_processing():
595
- """Start the main processing loop"""
596
- if processing_status["is_running"]:
597
- raise HTTPException(status_code=400, detail="Processing already running")
598
-
599
- # Start processing in a separate thread
600
- thread = threading.Thread(target=main_processing_loop, daemon=True)
601
- thread.start()
602
-
603
- return {
604
- "message": "Processing started",
605
- "status": "started"
606
- }
607
-
608
- @app.post("/stop")
609
- async def stop_processing():
610
- """Stop the main processing loop"""
611
- if not processing_status["is_running"]:
612
- raise HTTPException(status_code=400, detail="Processing not running")
613
-
614
- processing_status["is_running"] = False
615
-
616
- return {
617
- "message": "Processing stopped",
618
- "status": "stopped"
619
- }
620
-
621
- @app.get("/logs")
622
- async def get_logs(limit: int = 50):
623
- """Get recent logs"""
624
- logs = processing_status["logs"][-limit:]
625
- return {
626
- "total_logs": len(processing_status["logs"]),
627
- "recent_logs": logs
628
- }
629
-
630
- @app.post("/process/{filename}")
631
- async def process_single_file(filename: str):
632
- """Process a single audio file manually"""
633
- try:
634
- log_message(f"🎯 Manual processing requested for: {filename}", "INFO")
635
-
636
- # Download and process the file
637
- reference_map = fetch_reference_files(REFERENCE_REPO_ID)
638
- if not reference_map:
639
- raise HTTPException(status_code=500, detail="Could not fetch reference files")
640
-
641
- # Get file URL
642
- audio_url = hf_hub_url(repo_id=SOURCE_REPO_ID, filename=filename, repo_type="dataset", subfolder=None)
643
- local_wav_path = os.path.join(DOWNLOAD_FOLDER, os.path.basename(filename))
644
-
645
- # Download
646
- if not download_with_retry(audio_url, local_wav_path):
647
- raise HTTPException(status_code=500, detail="Failed to download file")
648
-
649
- # Find match
650
- base_filename = os.path.basename(filename)
651
- matched_filename = find_matching_filename(base_filename, reference_map)
652
-
653
- if not matched_filename:
654
- os.remove(local_wav_path)
655
- raise HTTPException(status_code=404, detail="No matching filename found")
656
-
657
- # Process
658
- if process_audio_file(local_wav_path, reference_map, matched_filename):
659
- processing_status["transcribed_files"] += 1
660
-
661
- if os.path.exists(local_wav_path):
662
- os.remove(local_wav_path)
663
-
664
- return {
665
- "status": "success",
666
- "file": filename,
667
- "matched": matched_filename,
668
- "message": "Audio transcribed and uploaded successfully"
669
- }
670
- else:
671
- if os.path.exists(local_wav_path):
672
- os.remove(local_wav_path)
673
- raise HTTPException(status_code=500, detail="Processing failed")
674
-
675
- except Exception as e:
676
- log_message(f"❌ Manual processing error: {str(e)}", "ERROR")
677
- raise HTTPException(status_code=500, detail=str(e))
678
-
679
- @app.on_event("startup")
680
- async def startup_event():
681
- """Auto-start processing when server starts"""
682
- log_message("πŸš€ Server startup: Checking dependencies...", "INFO")
683
-
684
- try:
685
- import transformers
686
- log_message("βœ… Transformers found", "INFO")
687
- except ImportError:
688
- log_message("⚠️ WARNING: Transformers not installed!", "WARNING")
689
- log_message(" Install with: pip install transformers librosa torch torchaudio", "WARNING")
690
-
691
- log_message("πŸš€ Server startup: Auto-starting processing loop", "INFO")
692
-
693
- # Start processing in a separate thread
694
- thread = threading.Thread(target=main_processing_loop, daemon=True)
695
- thread.start()
696
-
697
- def run_api(host: str = "0.0.0.0", port: int = 8000):
698
- """Run the FastAPI server"""
699
- log_message(f"πŸš€ Starting FastAPI server on {host}:{port}", "INFO")
700
- uvicorn.run(app, host=host, port=port)
701
-
702
- if __name__ == "__main__":
703
- # Run API server (processing will auto-start via startup event)
704
- run_api()
 
1
+ import os
2
+ import json
3
+ import requests
4
+ import subprocess
5
+ import shutil
6
+ import time
7
+ import sys
8
+ import threading
9
+ from typing import Dict, List, Optional, Any
10
+ from fastapi import FastAPI, HTTPException
11
+ from fastapi.responses import JSONResponse
12
+ import uvicorn
13
+ import torch
14
+ import librosa
15
+ from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor, pipeline
16
+
17
+ # Fix Unicode encoding for Windows
18
+ if sys.platform == 'win32':
19
+ import io
20
+ sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
21
+
22
+ # Initialize FastAPI app
23
+ app = FastAPI(title="Audio Transcriber", description="Audio transcription and upload service")
24
+
25
+ # ==== CONFIGURATION ====
26
+ # The new backend URL for state management and transcription upload
27
+ # It is now read from an environment variable, falling back to the default if not set.
28
+ BACKEND_URL = os.environ.get("BACKEND_URL", "https://samfredoly-acp.hf.space")
29
+ # The original Hugging Face repo IDs are still needed for fetching the audio files
30
+ # and the reference file list, as the backend only handles transcription storage.
31
+ SOURCE_REPO_ID = "Samfredoly/BG_Vid" # Fetch audio files from here
32
+ TARGET_REPO_ID = "samfred2/A_Text" # Target repo ID is now a constant for the backend
33
+ REFERENCE_REPO_ID = "Fred808/BG3" # Reference repo to match audio filenames
34
+
35
+ # Path Configuration
36
+ DOWNLOAD_FOLDER = "downloads_audio"
37
+ TRANSCRIPTIONS_FOLDER = "transcriptions"
38
+ LOCAL_STATE_FOLDER = ".state_audio"
39
+
40
+ os.makedirs(DOWNLOAD_FOLDER, exist_ok=True)
41
+ os.makedirs(TRANSCRIPTIONS_FOLDER, exist_ok=True)
42
+ os.makedirs(LOCAL_STATE_FOLDER, exist_ok=True)
43
+
44
+ # Whisper Model Setup (using transformers)
45
+ DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
46
+ TORCH_DTYPE = torch.float16 if torch.cuda.is_available() else torch.float32
47
+ WHISPER_MODEL_ID = f"openai/whisper-small"
48
+
49
+ # Global model cache
50
+ _whisper_model = None
51
+ _whisper_processor = None
52
+ _whisper_pipeline = None
53
+
54
+ def get_whisper_pipeline():
55
+ """Get or initialize the Whisper pipeline."""
56
+ global _whisper_model, _whisper_processor, _whisper_pipeline
57
+
58
+ if _whisper_pipeline is not None:
59
+ return _whisper_pipeline
60
+
61
+ try:
62
+ log_message(f"Loading Whisper model {WHISPER_MODEL_ID}...", "INFO")
63
+
64
+ model = AutoModelForSpeechSeq2Seq.from_pretrained(
65
+ WHISPER_MODEL_ID,
66
+ torch_dtype=TORCH_DTYPE,
67
+ low_cpu_mem_usage=True,
68
+ use_safetensors=True
69
+ )
70
+ model = model.to(DEVICE)
71
+
72
+ processor = AutoProcessor.from_pretrained(WHISPER_MODEL_ID)
73
+
74
+ _whisper_pipeline = pipeline(
75
+ "automatic-speech-recognition",
76
+ model=model,
77
+ tokenizer=processor.tokenizer,
78
+ feature_extractor=processor.feature_extractor,
79
+ torch_dtype=TORCH_DTYPE,
80
+ device=DEVICE
81
+ )
82
+
83
+ log_message(f"βœ… Whisper model loaded successfully on {DEVICE.upper()}", "INFO")
84
+ return _whisper_pipeline
85
+
86
+ except Exception as e:
87
+ log_message(f"❌ Failed to load Whisper model: {str(e)}", "ERROR")
88
+ raise
89
+
90
+ # State Files
91
+ FAILED_FILES_LOG = "failed_audio_files.log"
92
+ HF_STATE_FILE = "processing_audio_state.json" # This is the filename the backend uses
93
+
94
+ # Processing Parameters
95
+ PROCESSING_DELAY = 2
96
+ MAX_RETRIES = 3
97
+ MIN_FREE_SPACE_GB = 1
98
+ WHISPER_MODEL = "small" # Whisper model size
99
+
100
+ # NOTE: The Hugging Face API is still required for listing files in SOURCE_REPO_ID and REFERENCE_REPO_ID
101
+ from huggingface_hub import HfApi, hf_hub_url
102
+ HF_TOKEN = os.environ.get("HF_TOKEN", "")
103
+ hf_api = HfApi(token=HF_TOKEN)
104
+
105
+ # Global State
106
+ processing_status = {
107
+ "is_running": False,
108
+ "current_file": None,
109
+ "total_files": 0,
110
+ "processed_files": 0,
111
+ "failed_files": 0,
112
+ "transcribed_files": 0,
113
+ "last_update": None,
114
+ "logs": []
115
+ }
116
+
117
+ def log_message(message: str, level: str = "INFO"):
118
+ """Log messages with timestamp"""
119
+ timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
120
+ log_entry = f"[{timestamp}] {level}: {message}"
121
+ print(log_entry)
122
+ processing_status["logs"].append(log_entry)
123
+ processing_status["last_update"] = timestamp
124
+ if len(processing_status["logs"]) > 100:
125
+ processing_status["logs"] = processing_status["logs"][-100:]
126
+
127
+ def log_failed_file(filename: str, error: str):
128
+ """Log failed files to persistent file"""
129
+ with open(FAILED_FILES_LOG, "a") as f:
130
+ f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} - {filename}: {error}\n")
131
+
132
+ def get_disk_usage(path: str) -> Dict[str, float]:
133
+ """Get disk usage statistics in GB"""
134
+ statvfs = os.statvfs(path)
135
+ total = statvfs.f_frsize * statvfs.f_blocks / (1024**3)
136
+ free = statvfs.f_frsize * statvfs.f_bavail / (1024**3)
137
+ used = total - free
138
+ return {"total": total, "free": free, "used": used}
139
+
140
+ def check_disk_space(path: str = ".") -> bool:
141
+ """Check if there's enough disk space"""
142
+ disk_info = get_disk_usage(path)
143
+ if disk_info["free"] < MIN_FREE_SPACE_GB:
144
+ log_message(f'⚠️ Low disk space: {disk_info["free"]:.2f}GB free, {disk_info["used"]:.2f}GB used')
145
+ return False
146
+ return True
147
+
148
+ def cleanup_temp_files():
149
+ """Clean up temporary files to free space"""
150
+ log_message("🧹 Cleaning up temporary files...", "INFO")
151
+
152
+ current_file = processing_status.get("current_file")
153
+ for file in os.listdir(DOWNLOAD_FOLDER):
154
+ if file != current_file and file.endswith((".wav", ".mp3")):
155
+ try:
156
+ os.remove(os.path.join(DOWNLOAD_FOLDER, file))
157
+ log_message(f"πŸ—‘οΈ Removed old download: {file}", "INFO")
158
+ except:
159
+ pass
160
+
161
+ # Helper function to save state locally
162
+ def save_json_state(file_path: str, data: Dict[str, Any]):
163
+ """Save state to JSON file"""
164
+ with open(file_path, "w") as f:
165
+ json.dump(data, f, indent=2)
166
+
167
+ # --- NEW API FUNCTIONS FOR STATE MANAGEMENT AND UPLOAD ---
168
+
169
+ def download_state_from_api() -> Dict[str, Any]:
170
+ """Downloads the state file from the backend API."""
171
+ url = f"{BACKEND_URL}/state/"
172
+ default_state = {"next_download_index": 0, "file_states": {}}
173
+
174
+ try:
175
+ response = requests.get(url, timeout=10)
176
+ response.raise_for_status()
177
+
178
+ # The API returns {"state": {...}}
179
+ state_data = response.json().get("state", default_state)
180
+
181
+ # Ensure the structure is correct (migration logic from original load_json_state)
182
+ if "file_states" not in state_data or not isinstance(state_data["file_states"], dict):
183
+ state_data["file_states"] = {}
184
+ if "next_download_index" not in state_data:
185
+ state_data["next_download_index"] = 0
186
+
187
+ log_message(f"βœ… Successfully downloaded state from API.", "INFO")
188
+ return state_data
189
+
190
+ except requests.exceptions.RequestException as e:
191
+ log_message(f"⚠️ Failed to download state from API ({url}): {str(e)}. Starting from default state.", "WARNING")
192
+ return default_state
193
+
194
+ def upload_state_to_api(state: Dict[str, Any]) -> bool:
195
+ """
196
+ Saves the state locally and uploads it to the backend API's /upload/ endpoint.
197
+ This simulates the original HF state upload for locking/unlocking.
198
+ """
199
+ local_path = os.path.join(LOCAL_STATE_FOLDER, HF_STATE_FILE)
200
+ url = f"{BACKEND_URL}/upload/"
201
+
202
+ try:
203
+ # 1. Save the current state locally
204
+ save_json_state(local_path, state)
205
+
206
+ # 2. Upload the state file to the backend
207
+ with open(local_path, "rb") as f:
208
+ files = {'file': (HF_STATE_FILE, f, 'application/json')}
209
+
210
+ response = requests.post(url, files=files, timeout=30)
211
+ response.raise_for_status()
212
+
213
+ log_message(f"βœ… Successfully uploaded state file to API: {HF_STATE_FILE}", "INFO")
214
+ return True
215
+
216
+ except requests.exceptions.RequestException as e:
217
+ log_message(f"❌ Failed to upload state file to API ({url}): {str(e)}", "ERROR")
218
+ # 409 Conflict means the file already exists, which is fine for state updates
219
+ if response.status_code == 409:
220
+ log_message(f"⚠️ State file already exists on server (409 Conflict). Proceeding.", "WARNING")
221
+ return True
222
+ return False
223
+ except Exception as e:
224
+ log_message(f"❌ An unexpected error occurred during API state upload: {str(e)}", "ERROR")
225
+ return False
226
+
227
+ def upload_transcription_to_api(json_output_path: str, matched_filename: str) -> bool:
228
+ """Uploads the transcription JSON file to the backend API's /upload/ endpoint."""
229
+ url = f"{BACKEND_URL}/upload/"
230
+
231
+ try:
232
+ with open(json_output_path, "rb") as f:
233
+ files = {'file': (os.path.basename(json_output_path), f, 'application/json')}
234
+
235
+ response = requests.post(url, files=files, timeout=30)
236
+ response.raise_for_status()
237
+
238
+ log_message(f"βœ… Successfully uploaded transcription to API: {os.path.basename(json_output_path)}", "INFO")
239
+ return True
240
+
241
+ except requests.exceptions.RequestException as e:
242
+ log_message(f"❌ Failed to upload transcription to API ({url}): {str(e)}", "ERROR")
243
+ if response.status_code == 409:
244
+ log_message(f"⚠️ File already exists on server (409 Conflict).", "WARNING")
245
+ return False
246
+ except Exception as e:
247
+ log_message(f"❌ An unexpected error occurred during API upload: {str(e)}", "ERROR")
248
+ return False
249
+
250
+ def lock_file_for_processing(wav_filename: str, state: Dict[str, Any]) -> bool:
251
+ """Marks a file as 'processing' in the state file and uploads the lock via API."""
252
+ log_message(f"πŸ”’ Attempting to lock file: {wav_filename} (Marking as 'processing')", "INFO")
253
+
254
+ state["file_states"][wav_filename] = "processing"
255
+
256
+ if upload_state_to_api(state):
257
+ log_message(f"βœ… Successfully locked file: {wav_filename} via API state upload", "INFO")
258
+ return True
259
+ else:
260
+ log_message(f"❌ Failed to upload lock for file: {wav_filename}. Aborting processing.", "ERROR")
261
+ # Revert local state change if upload fails
262
+ if wav_filename in state["file_states"]:
263
+ del state["file_states"][wav_filename]
264
+ return False
265
+
266
+ def unlock_file_as_processed(wav_filename: str, state: Dict[str, Any], next_index: int) -> bool:
267
+ """Marks a file as 'processed', updates the index, and uploads the state via API."""
268
+ log_message(f"πŸ”“ Attempting to unlock file: {wav_filename} (Marking as 'processed')", "INFO")
269
+
270
+ state["file_states"][wav_filename] = "processed"
271
+ state["next_download_index"] = next_index
272
+
273
+ if upload_state_to_api(state):
274
+ log_message(f"βœ… Successfully unlocked and marked as processed: {wav_filename} via API state upload", "INFO")
275
+ return True
276
+ else:
277
+ log_message(f"❌ Failed to upload final state for file: {wav_filename}.", "ERROR")
278
+ return False
279
+
280
+ # --- END NEW API FUNCTIONS ---
281
+
282
+ def download_with_retry(url: str, dest_path: str, max_retries: int = 3) -> bool:
283
+ """Download file with retry logic and disk space checking"""
284
+ if not check_disk_space():
285
+ cleanup_temp_files()
286
+ if not check_disk_space():
287
+ log_message("❌ Insufficient disk space even after cleanup", "ERROR")
288
+ return False
289
+
290
+ try:
291
+ os.makedirs(os.path.dirname(dest_path), exist_ok=True)
292
+ except Exception as e:
293
+ log_message(f"❌ Failed to create directory for download path {os.path.dirname(dest_path)}: {str(e)}", "ERROR")
294
+ return False
295
+
296
+ # The original code used HF_TOKEN for authorization headers, which is only needed
297
+ # if the source repo is private. We keep it for compatibility.
298
+ headers = {"Authorization": f"Bearer {HF_TOKEN}"}
299
+ for attempt in range(max_retries):
300
+ try:
301
+ with requests.get(url, headers=headers, stream=True) as r:
302
+ r.raise_for_status()
303
+
304
+ with open(dest_path, "wb") as f:
305
+ for chunk in r.iter_content(chunk_size=8192):
306
+ if chunk:
307
+ f.write(chunk)
308
+
309
+ log_message(f"βœ… Download successful: {os.path.basename(dest_path)}", "INFO")
310
+ return True
311
+ except requests.exceptions.RequestException as e:
312
+ log_message(f"⚠️ Download attempt {attempt + 1}/{max_retries} failed for {url}: {str(e)}", "WARNING")
313
+ if attempt < max_retries - 1:
314
+ time.sleep(2 ** attempt) # Exponential backoff
315
+ else:
316
+ log_message(f"❌ Download failed after {max_retries} attempts for {url}", "ERROR")
317
+ return False
318
+ except Exception as e:
319
+ log_message(f"❌ An unexpected error occurred during download: {str(e)}", "ERROR")
320
+ return False
321
+ return False
322
+
323
+ def get_reference_map(reference_repo_id: str) -> Dict[str, str]:
324
+ """
325
+ Downloads the reference file list from the Hugging Face repo and creates a map
326
+ from audio filename (without extension) to the reference filename.
327
+ """
328
+ log_message(f"Fetching reference file list from {reference_repo_id}...", "INFO")
329
+
330
+ # This is a placeholder for the actual logic to get the file list.
331
+ # Assuming the reference repo contains a list of files that match the audio files.
332
+ # In a real scenario, this would involve listing files in the repo.
333
+ # For now, we'll assume a simple list of files can be retrieved.
334
+
335
+ try:
336
+ # Use HfApi to list files in the reference repo
337
+ repo_files = hf_api.list_repo_files(repo_id=reference_repo_id, repo_type="dataset")
338
+
339
+ reference_map = {}
340
+ for file in repo_files:
341
+ # Assuming the reference files are named like 'audio_file_name.txt'
342
+ # and we want to map the audio file name (e.g., 'audio_file_name.wav') to it.
343
+ base_name, ext = os.path.splitext(file)
344
+ if ext.lower() in ['.txt', '.json']: # Only consider text/json files as reference
345
+ # The key is the audio file name without extension
346
+ reference_map[base_name] = file
347
+
348
+ log_message(f"βœ… Successfully created reference map with {len(reference_map)} entries.", "INFO")
349
+ return reference_map
350
+
351
+ except Exception as e:
352
+ log_message(f"❌ Failed to fetch reference map from Hugging Face: {str(e)}", "ERROR")
353
+ return {}
354
+
355
+ def find_matching_filename(audio_filename: str, reference_map: Dict[str, str]) -> Optional[str]:
356
+ """Finds the matching reference filename for a given audio filename."""
357
+ base_name, _ = os.path.splitext(audio_filename)
358
+ return reference_map.get(base_name)
359
+
360
+ def get_next_file_to_process(source_repo_id: str, state: Dict[str, Any]) -> Optional[Dict[str, Any]]:
361
+ """
362
+ Determines the next file to process based on the current state and the file list
363
+ from the source Hugging Face repository.
364
+ """
365
+ log_message(f"Determining next file to process from {source_repo_id}...", "INFO")
366
+
367
+ try:
368
+ # 1. Get the list of all files in the source repo
369
+ repo_files = hf_api.list_repo_files(repo_id=source_repo_id, repo_type="dataset")
370
+
371
+ # Filter for audio files (e.g., .wav, .mp3)
372
+ audio_files = sorted([f for f in repo_files if f.lower().endswith(('.wav', '.mp3'))])
373
+
374
+ processing_status["total_files"] = len(audio_files)
375
+
376
+ if not audio_files:
377
+ log_message("No audio files found in the source repository.", "INFO")
378
+ return None
379
+
380
+ # 2. Get the next index from the state
381
+ next_index = state.get("next_download_index", 0)
382
+ file_states = state.get("file_states", {})
383
+
384
+ # 3. Find the next file that hasn't been processed or is not currently being processed
385
+ for i in range(next_index, len(audio_files)):
386
+ filename = audio_files[i]
387
+ status = file_states.get(filename, "unprocessed")
388
+
389
+ # Skip files that are already processed or currently being processed
390
+ if status in ["processed", "processing"]:
391
+ continue
392
+
393
+ # Found an unprocessed file
394
+ file_url = hf_hub_url(repo_id=source_repo_id, filename=filename, repo_type="dataset")
395
+
396
+ log_message(f"Found next file at index {i}: {filename}", "INFO")
397
+ return {
398
+ "filename": filename,
399
+ "url": file_url,
400
+ "index": i
401
+ }
402
+
403
+ log_message("All files up to the current index have been processed or are locked.", "INFO")
404
+
405
+ # If we reach the end, check from the beginning for any failed files
406
+ for i in range(0, next_index):
407
+ filename = audio_files[i]
408
+ status = file_states.get(filename, "unprocessed")
409
+
410
+ if status == "failed":
411
+ file_url = hf_hub_url(repo_id=source_repo_id, filename=filename, repo_type="dataset")
412
+ log_message(f"Found failed file for retry at index {i}: {filename}", "INFO")
413
+ return {
414
+ "filename": filename,
415
+ "url": file_url,
416
+ "index": i
417
+ }
418
+
419
+ return None
420
+
421
+ except Exception as e:
422
+ log_message(f"❌ Failed to get next file to process: {str(e)}", "ERROR")
423
+ return None
424
+
425
+ def run_whisper_transcription(audio_path: str, output_dir: str, model: str) -> Optional[str]:
426
+ """
427
+ Runs Whisper transcription using the transformers library.
428
+ Returns the path to the generated JSON file on success.
429
+ No ffmpeg dependency required.
430
+ """
431
+ log_message(f"πŸŽ™οΈ Starting transcription for {os.path.basename(audio_path)} with model {model}...", "INFO")
432
+
433
+ try:
434
+ # Get the Whisper pipeline
435
+ pipe = get_whisper_pipeline()
436
+
437
+ # Load audio using librosa
438
+ log_message(f"Loading audio file: {audio_path}", "INFO")
439
+ audio_data, sample_rate = librosa.load(audio_path, sr=16000)
440
+
441
+ # Run transcription
442
+ log_message(f"Running transcription...", "INFO")
443
+ result = pipe(
444
+ audio_data,
445
+ chunk_length_s=30,
446
+ batch_size=8,
447
+ return_timestamps=True
448
+ )
449
+
450
+ # Extract text and chunks
451
+ transcription_text = result.get("text", "")
452
+ chunks = result.get("chunks", [])
453
+
454
+ log_message(f"βœ… Transcription successful: {len(transcription_text)} characters", "INFO")
455
+
456
+ # Prepare output JSON structure
457
+ output_json = {
458
+ "text": transcription_text,
459
+ "chunks": chunks,
460
+ "language": result.get("language", "en")
461
+ }
462
+
463
+ # Save to JSON file
464
+ base_name, _ = os.path.splitext(os.path.basename(audio_path))
465
+ json_output_path = os.path.join(output_dir, f"{base_name}.json")
466
+
467
+ with open(json_output_path, "w", encoding="utf-8") as f:
468
+ json.dump(output_json, f, indent=2, ensure_ascii=False)
469
+
470
+ log_message(f"βœ… Saved transcription to: {json_output_path}", "INFO")
471
+ return json_output_path
472
+
473
+ except Exception as e:
474
+ log_message(f"❌ An error occurred during transcription: {str(e)}", "ERROR")
475
+ import traceback
476
+ log_message(f"Traceback: {traceback.format_exc()}", "ERROR")
477
+ return None
478
+
479
+ def process_audio_file(audio_path: str, reference_map: Dict[str, str], output_filename: str) -> bool:
480
+ """
481
+ Transcribes the audio file, renames the output JSON to match the reference,
482
+ and uploads the result to the API.
483
+ """
484
+
485
+ # 1. Run transcription
486
+ json_output_path = run_whisper_transcription(audio_path, TRANSCRIPTIONS_FOLDER, WHISPER_MODEL)
487
+
488
+ if not json_output_path:
489
+ return False
490
+
491
+ # 2. Rename the JSON file to the matched filename
492
+ # The output_filename already includes the correct extension (e.g., .txt or .json)
493
+ # We assume the reference map provides the full target filename.
494
+
495
+ # The whisper output is a JSON file named after the audio file.
496
+ # We need to rename it to the target filename (which should be a JSON file for the backend).
497
+
498
+ # The output_filename is the matched filename from the reference map (e.g., 'audio_file_name.txt')
499
+ # The backend expects a JSON file. Let's assume the matched filename should be used as the base
500
+ # but with a .json extension for the upload.
501
+
502
+ # Let's stick to the original logic: the backend expects a JSON file with the name
503
+ # of the audio file (or the matched reference file) with a .json extension.
504
+
505
+ # Since the whisper output is already a JSON file, we just need to rename it
506
+ # to the desired final name.
507
+
508
+ # The output_filename passed here is the base name of the audio file or the matched reference file.
509
+ # If it's a reference file name (e.g., 'file.txt'), we should probably use 'file.json'.
510
+
511
+ # For simplicity and to match the backend's expectation (which handles JSON),
512
+ # we will rename the whisper output JSON to the base name of the audio file
513
+ # and ensure it has a .json extension.
514
+
515
+ base_name, _ = os.path.splitext(output_filename)
516
+ final_json_filename = f"{base_name}.json"
517
+ final_json_path = os.path.join(TRANSCRIPTIONS_FOLDER, final_json_filename)
518
+
519
+ try:
520
+ if json_output_path != final_json_path:
521
+ shutil.move(json_output_path, final_json_path)
522
+ log_message(f"βœ… Renamed transcription to: {final_json_filename}", "INFO")
523
+ except Exception as e:
524
+ log_message(f"❌ Failed to rename transcription file: {str(e)}", "ERROR")
525
+ return False
526
+
527
+ # 3. Upload transcription to API
528
+ if upload_transcription_to_api(final_json_path, final_json_filename):
529
+ processing_status["transcribed_files"] += 1
530
+ # Clean up the local transcription file after successful upload
531
+ try:
532
+ os.remove(final_json_path)
533
+ log_message(f"πŸ—‘οΈ Cleaned up local transcription file: {final_json_path}", "INFO")
534
+ except Exception as e:
535
+ log_message(f"❌ Failed to clean up transcription file: {str(e)}", "ERROR")
536
+ return True
537
+ else:
538
+ log_message(f"❌ Failed to upload transcription to API: {final_json_filename}", "ERROR")
539
+ return False
540
+
541
+ def main_processing_loop():
542
+ """The main loop that continuously checks for and processes new audio files."""
543
+ global processing_status
544
+
545
+ if processing_status["is_running"]:
546
+ log_message("Processing loop is already running.", "WARNING")
547
+ return
548
+
549
+ processing_status["is_running"] = True
550
+ log_message("πŸš€ Audio transcription processing loop started.", "INFO")
551
+
552
+ # 1. Get the reference map once
553
+ reference_map = get_reference_map(REFERENCE_REPO_ID)
554
+ if not reference_map:
555
+ log_message("❌ Could not get reference map. Stopping loop.", "CRITICAL")
556
+ processing_status["is_running"] = False
557
+ return
558
+
559
+ try:
560
+ while processing_status["is_running"]:
561
+ time.sleep(PROCESSING_DELAY)
562
+
563
+ # 1. Download state from the new API
564
+ current_state = download_state_from_api()
565
+ next_file_info = get_next_file_to_process(SOURCE_REPO_ID, current_state)
566
+
567
+ if next_file_info is None:
568
+ log_message("πŸ’€ No new audio files to process. Sleeping for a while...", "INFO")
569
+ time.sleep(PROCESSING_DELAY * 5)
570
+ continue
571
+
572
+ target_file = next_file_info['filename']
573
+ audio_url = next_file_info['url']
574
+ target_index = next_file_info['index']
575
+
576
+ processing_status["current_file"] = target_file
577
+ success = False
578
+ matched_filename = None
579
+
580
+ try:
581
+ # 2. Lock file by updating state on the API
582
+ if not lock_file_for_processing(target_file, current_state):
583
+ log_message(f"❌ Failed to lock file {target_file}. Skipping.", "ERROR")
584
+ time.sleep(PROCESSING_DELAY)
585
+ continue
586
+
587
+ local_wav_path = os.path.join(DOWNLOAD_FOLDER, os.path.basename(target_file))
588
+ log_message(f"⬇️ Downloading audio file: {target_file}", "INFO")
589
+
590
+ if download_with_retry(audio_url, local_wav_path):
591
+
592
+ # Extract base filename for matching
593
+ base_filename = os.path.basename(target_file)
594
+ matched_filename = find_matching_filename(base_filename, reference_map)
595
+
596
+ # Use matched filename if found, otherwise use original filename
597
+ output_filename = matched_filename if matched_filename else base_filename
598
+
599
+ # 3. Process and Upload transcription to API
600
+ if process_audio_file(local_wav_path, reference_map, output_filename):
601
+ success = True
602
+ log_message(f"βœ… Finished processing: {target_file}", "INFO")
603
+ else:
604
+ log_message(f"❌ Processing failed for: {target_file}", "ERROR")
605
+ else:
606
+ log_message(f"❌ Download failed for: {target_file}", "ERROR")
607
+
608
+ except Exception as e:
609
+ log_message(f"πŸ”₯ An unhandled error occurred while processing {target_file}: {str(e)}", "ERROR")
610
+ log_failed_file(target_file, str(e))
611
+
612
+ finally:
613
+ # 4. Unlock/Mark as processed by updating state on the API
614
+ next_index_to_save = target_index + 1
615
+
616
+ # Re-download the state to ensure we have the latest index/state
617
+ # before marking as processed, in case another server updated it.
618
+ current_state = download_state_from_api()
619
+
620
+ if success:
621
+ # Mark as processed and update index, then upload state
622
+ unlock_file_as_processed(target_file, current_state, next_index_to_save)
623
+ processing_status["processed_files"] += 1
624
+ else:
625
+ # Mark as failed and upload state
626
+ log_message(f"⚠️ File {target_file} failed. Marking as 'failed' and updating state.", "WARNING")
627
+ current_state["file_states"][target_file] = "failed"
628
+ upload_state_to_api(current_state)
629
+
630
+ # Clean up the downloaded audio file regardless of success
631
+ try:
632
+ if os.path.exists(local_wav_path):
633
+ os.remove(local_wav_path)
634
+ log_message(f"πŸ—‘οΈ Cleaned up local audio file: {local_wav_path}", "INFO")
635
+ except Exception as e:
636
+ log_message(f"❌ Failed to clean up audio file: {str(e)}", "ERROR")
637
+
638
+ processing_status["current_file"] = None
639
+ time.sleep(PROCESSING_DELAY)
640
+
641
+ except Exception as e:
642
+ log_message(f"πŸ”₯ Critical error in main processing loop: {str(e)}", "CRITICAL")
643
+
644
+ finally:
645
+ processing_status["is_running"] = False
646
+ log_message("πŸ›‘ Audio transcription processing loop stopped.", "INFO")
647
+
648
+ # --- FastAPI Endpoints (Unchanged) ---
649
+
650
+ # Add to configuration section
651
+ AUTO_START_PROCESSING = os.environ.get("AUTO_START_PROCESSING", "true").lower() == "true"
652
+
653
+ @app.on_event("startup")
654
+ async def startup_event():
655
+ """Conditionally start processing based on environment variable."""
656
+ if AUTO_START_PROCESSING:
657
+ log_message("πŸš€ AUTO_START_PROCESSING enabled - Starting processing loop...", "INFO")
658
+ thread = threading.Thread(target=main_processing_loop, daemon=True)
659
+ thread.start()
660
+ log_message("βœ… Background processing thread started", "INFO")
661
+ else:
662
+ log_message("⏸️ AUTO_START_PROCESSING disabled - Use /start endpoint to begin", "INFO")
663
+
664
+ @app.get("/")
665
+ async def root():
666
+ """Root endpoint to check service status."""
667
+ return {"message": "Audio Transcriber Service is running", "status": processing_status}
668
+
669
+ @app.get("/status")
670
+ async def get_status():
671
+ """Get the current processing status."""
672
+ return processing_status
673
+
674
+ @app.post("/start")
675
+ async def start_processing():
676
+ """Start the background processing loop."""
677
+ if processing_status["is_running"]:
678
+ return JSONResponse(status_code=200, content={"message": "Processing already running."})
679
+
680
+ thread = threading.Thread(target=main_processing_loop)
681
+ thread.start()
682
+ return JSONResponse(status_code=200, content={"message": "Processing started in background."})
683
+
684
+ @app.post("/stop")
685
+ async def stop_processing():
686
+ """Stop the background processing loop."""
687
+ if not processing_status["is_running"]:
688
+ return JSONResponse(status_code=200, content={"message": "Processing is not running."})
689
+
690
+ processing_status["is_running"] = False
691
+ return JSONResponse(status_code=200, content={"message": "Processing stop requested. Will stop after current file."})
692
+
693
+ # --- Main Execution ---
694
+
695
+ if __name__ == "__main__":
696
+ # This block is for local testing and won't be used in the final sandbox execution
697
+ # but is good practice for a runnable script.
698
+ uvicorn.run(app, host="0.0.0.0", port=8000)