104 lines
3.6 KiB
Python
104 lines
3.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
import os
|
|
import sys
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
# Add the app directory to the path
|
|
sys.path.append(os.path.join(os.path.dirname(__file__), 'app'))
|
|
|
|
# Mock the FastAPI UploadFile
|
|
class MockUploadFile:
|
|
def __init__(self, filename, content):
|
|
self.filename = filename
|
|
self.content = content
|
|
self.size = len(content)
|
|
self._file_pos = 0
|
|
|
|
async def read(self, size=-1):
|
|
if size == -1:
|
|
result = self.content[self._file_pos:]
|
|
self._file_pos = len(self.content)
|
|
else:
|
|
result = self.content[self._file_pos:self._file_pos + size]
|
|
self._file_pos += len(result)
|
|
return result
|
|
|
|
def file(self):
|
|
from io import BytesIO
|
|
return BytesIO(self.content.encode() if isinstance(self.content, str) else self.content)
|
|
|
|
async def test_audio_upload():
|
|
from app.api.endpoints.meetings import AUDIO_DIR
|
|
from app.services.qiniu_service import qiniu_service
|
|
|
|
# Path to the problematic audio file
|
|
audio_file_path = "/Users/jiliu/工作/projects/imeeting/backend/uploads/audio/31ce039a-f619-4869-91c8-eab934bbd1d4.m4a"
|
|
|
|
# Read the content of the audio file
|
|
try:
|
|
with open(audio_file_path, "rb") as f:
|
|
test_content = f.read()
|
|
print(f"Successfully read content from {audio_file_path}")
|
|
except FileNotFoundError:
|
|
print(f"Error: The file was not found at {audio_file_path}")
|
|
return
|
|
|
|
# Create mock UploadFile with the real audio content
|
|
mock_file = MockUploadFile("31ce039a-f619-4869-91c8-eab934bbd1d4.m4a", test_content)
|
|
|
|
# Create temporary file for upload (simulating the API endpoint)
|
|
file_extension = ".m4a"
|
|
from uuid import uuid4
|
|
temp_filename = f"{uuid4()}{file_extension}"
|
|
temp_path = AUDIO_DIR / temp_filename
|
|
|
|
print(f"Creating temporary file at: {temp_path}")
|
|
|
|
# Save file temporarily (simulating the API endpoint)
|
|
try:
|
|
# Simulate shutil.copyfileobj(mock_file.file(), open(temp_path, "wb"))
|
|
with open(temp_path, "wb") as buffer:
|
|
buffer.write(mock_file.content) # content is already bytes
|
|
|
|
print(f"Temporary file created successfully. Exists: {temp_path.exists()}")
|
|
print(f"Temporary file size: {temp_path.stat().st_size}")
|
|
except Exception as e:
|
|
print(f"Failed to save temporary file: {str(e)}")
|
|
return
|
|
|
|
# Test upload to Qiniu (simulating the API endpoint)
|
|
try:
|
|
print(f"Attempting to upload audio to Qiniu...")
|
|
print(f"Temp file path: {temp_path}")
|
|
print(f"Temp file exists: {temp_path.exists()}")
|
|
|
|
success, qiniu_url, error_msg = qiniu_service.upload_audio_file(
|
|
str(temp_path), 11, "test-audio.mp3"
|
|
)
|
|
|
|
print(f"Qiniu upload result - success: {success}")
|
|
print(f"Qiniu upload result - url: {qiniu_url}")
|
|
print(f"Qiniu upload result - error: {error_msg}")
|
|
|
|
# Clean up temporary file
|
|
if temp_path.exists():
|
|
temp_path.unlink()
|
|
print("Temporary file cleaned up")
|
|
|
|
if success:
|
|
print("Upload successful!")
|
|
print(f"URL: {qiniu_url}")
|
|
else:
|
|
print(f"Upload failed: {error_msg}")
|
|
|
|
except Exception as e:
|
|
print(f"Exception in audio upload: {str(e)}")
|
|
import traceback
|
|
print(f"Traceback: {traceback.format_exc()}")
|
|
# Clean up temporary file in case of error
|
|
if temp_path.exists():
|
|
temp_path.unlink()
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(test_audio_upload()) |