如何构建您自己的 Google NotebookLM

@app.post('/upload')async def upload_files(    token: annotated[parsedtoken, depends(verify_firebase_token)],    project_name: str,    description: str,    website_link: str,    host_count: int,    files: optional[list[uploadfile]] = file(none)):    # validate token    user_id = token['uid']    # generate unique identifiers    project_id = str(uuid.uuid4())    podcast_id = str(uuid.uuid4())    # process and store files    file_urls = await process_uploads(files, user_id, project_id)    # create firestore document    await create_project_document(user_id, project_id, {        'status': 'pending',        'created_at': datetime.now(),        'project_name': project_name,        'description': description,        'file_urls': file_urls    })    # trigger async processing    await publish_to_pubsub(user_id, project_id, podcast_id, file_urls)    return {'project_id': project_id, 'status': 'processing'}

async def update_status(user_id: str, project_id: str, status: str, metadata: dict = none):    doc_ref = db.collection('projects').document(f'{user_id}/{project_id}')    update_data = {        'status': status,        'updated_at': datetime.now()    }    if metadata:        update_data.update(metadata)    await doc_ref.update(update_data)

{    'user_id': 'uid_123',    'project_id': 'proj_456',    'podcast_id': 'pod_789',    'file_urls': ['gs://bucket/file1.pdf'],    'description': 'technical blog post about cloud architecture',    'host_count': 2,    'action': 'create_project'}

import azure.cognitiveservices.speech as speechsdkfrom pathlib import pathclass speechgenerator:    def __init__(self):        self.speech_config = speechsdk.speechconfig(            subscription=os.getenv("azure_speech_key"),            region=os.getenv("azure_speech_region")        )    async def create_speech_segment(self, text, voice, output_file):        try:            self.speech_config.speech_synthesis_voice_name = voice            synthesizer = speechsdk.speechsynthesizer(                speech_config=self.speech_config,                audio_config=none            )            # generate speech from text            result = synthesizer.speak_text_async(text).get()            if result.reason == speechsdk.resultreason.synthesizingaudiocompleted:                with open(output_file, "wb") as audio_file:                    audio_file.write(result.audio_data)                return true            return false        except exception as e:            logger.error(f"speech synthesis failed: {str(e)}")            return false

async def generate_podcast_script(outline: str, analysis: str, host_count: int):    # system instructions for different podcast formats    system_instructions = two_host_system_prompt if host_count > 1 else one_host_system_prompt    # example of how we structure the ai conversation    if host_count > 1:        script_format = """        **alex**: "hello and welcome to mypodify! i'm your host alex, joined by..."        **jane**: "hi everyone! i'm jane, and today we're diving into {topic}..."        """    else:        script_format = """        **alex**: "welcome to mypodify! today we're exploring {topic}..."        """    # generate the complete script using ai    script = await generate_content_from_openai(        content=f"{outline}content details:{analysis}",        system_instructions=system_instructions,        purpose="podcast script"    )    return script

voice_mapping = {    'alex': 'en-us-andrewmultilingualneural',  # male host    'jane': 'en-us-avamultilingualneural',     # female host    'narrator': 'en-us-brandonmultilingualneural'  # neutral voice}

async def process_document(message_data: dict):    try:        # extract content from documents        content = await extract_document_content(message_data['file_urls'])        # analyze and structure content        document_analysis = await analyze_content(content)        # generate podcast script        script = await generate_script(            document_analysis,            speaker_count=message_data['host_count']        )        # convert to audio        audio_segments = await generate_audio_segments(script)        # post-process audio        final_audio = await post_process_audio(audio_segments)        # upload and update status        audio_url = await upload_to_storage(final_audio)        await update_status(            message_data['user_id'],            message_data['project_id'],            'completed',            {'audio_url': audio_url}        )    except exception as e:        await handle_processing_error(message_data, e)

async def track_metrics(stage: str, duration: float, metadata: dict = None):    metrics = {        'stage': stage,        'duration_ms': duration * 1000,        'timestamp': datetime.now()    }    if metadata:        metrics.update(metadata)    await publish_metrics(metrics)