Update main.py

main.py

@@ -1,4 +1,4 @@
-# main.py (Increasing timeout and adding logging in generate_summary)
+# main.py (Implementing Async Apify Flow)
 import os
 import re
 import logging
@@ -7,7 +7,7 @@ import json
 import html
 import contextlib
 import traceback
-from typing import Optional, Dict, Any
+from typing import Optional, Dict, Any, Tuple
 
 # --- Frameworks ---
 from starlette.applications import Starlette
@@ -152,43 +152,110 @@ async def get_transcript_via_supadata(video_id: str, api_key: str) -> Optional[s
             elif response.status_code == 404: logger.warning(f"[Supadata] Not found (404) for {video_id}."); return None
             else: logger.error(f"[Supadata] Unexpected status {response.status_code} for {video_id}. Resp: {response.text[:200]}"); return None
     except httpx.TimeoutException: logger.error(f"[Supadata] Timeout connecting for {video_id}"); return None
-    except httpx.RequestError as e: logger.error(f"[Supadata] Request error for {video_id}: {e}"); return None
+    except httpx.RequestError as e:
+        if "CERTIFICATE_VERIFY_FAILED" in str(e): logger.error(f"[Supadata] SSL Cert Verify Failed for {video_id}: {e}")
+        else: logger.error(f"[Supadata] Request error for {video_id}: {e}")
+        return None
     except Exception as e: logger.error(f"[Supadata] Unexpected error for {video_id}: {e}", exc_info=True); return None
 
 async def get_transcript_via_apify(video_url: str, api_token: str) -> Optional[str]:
+    """Fetches YouTube transcript using Apify REST API (async start + poll + dataset fetch)."""
     global APIFY_ACTOR_ID
-    if not video_url: logger.error("[Apify] No video_url provided"); return None
-    if not api_token: logger.error("[Apify] API token missing."); return None
-    logger.info(f"[Apify] Attempting fetch via REST for URL: {video_url} (Actor: {APIFY_ACTOR_ID})")
-    api_endpoint = f"https://api.apify.com/v2/acts/{APIFY_ACTOR_ID}/run-sync-get-dataset-items"
-    params = {"token": api_token}
-    payload = { "urls": [video_url], "outputFormat": "singleStringText", "maxRetries": 5, "channelHandleBoolean": False, "channelNameBoolean": False, "datePublishedBoolean": False, "relativeDateTextBoolean": False, }
+    if not video_url: logger.error("[Apify Async] No video_url provided"); return None
+    if not api_token: logger.error("[Apify Async] API token missing."); return None
+    logger.info(f"[Apify Async] Attempting fetch for URL: {video_url} (Actor: {APIFY_ACTOR_ID})")
+
+    start_run_endpoint = f"https://api.apify.com/v2/acts/{APIFY_ACTOR_ID}/runs"
+    params_base = {"token": api_token}
+    payload = {
+        "urls": [video_url], "outputFormat": "singleStringText", "maxRetries": 5,
+        "channelHandleBoolean": False, "channelNameBoolean": False,
+        "datePublishedBoolean": False, "relativeDateTextBoolean": False,
+    }
     headers = {"Content-Type": "application/json"}
+
+    run_id: Optional[str] = None
+    dataset_id: Optional[str] = None
+    max_wait_seconds = 120 # Total time to wait for the run to finish
+    poll_interval = 5 # Seconds between status checks
+
     try:
-        async with httpx.AsyncClient(timeout=90.0) as client:
-            logger.debug(f"[Apify] Sending request to run actor {APIFY_ACTOR_ID} synchronously for {video_url}")
-            response = await client.post(api_endpoint, headers=headers, params=params, json=payload)
-            logger.debug(f"[Apify] Received status code {response.status_code} for {video_url}")
-            if response.status_code == 200:
+        async with httpx.AsyncClient(timeout=30.0) as client: # Shorter timeout for individual API calls
+            # 1. Start the run
+            logger.debug(f"[Apify Async] Starting actor run for {video_url}")
+            response_start = await client.post(start_run_endpoint, headers=headers, params=params_base, json=payload)
+            logger.debug(f"[Apify Async] Start run status: {response_start.status_code}")
+
+            if response_start.status_code == 201: # Expected status for starting a run
+                try:
+                    run_data = response_start.json().get("data", {})
+                    run_id = run_data.get("id")
+                    dataset_id = run_data.get("defaultDatasetId")
+                    if not run_id or not dataset_id:
+                        logger.error(f"[Apify Async] Started run but missing runId or datasetId. Data: {run_data}")
+                        return None
+                    logger.info(f"[Apify Async] Run started. Run ID: {run_id}, Dataset ID: {dataset_id}")
+                except Exception as e:
+                    logger.error(f"[Apify Async] Error parsing start run response: {e}. Response: {response_start.text[:200]}", exc_info=True)
+                    return None
+            else:
+                logger.error(f"[Apify Async] Failed to start run. Status: {response_start.status_code}. Response: {response_start.text[:200]}")
+                return None
+
+            # 2. Poll for run completion
+            run_status_endpoint = f"https://api.apify.com/v2/actor-runs/{run_id}"
+            elapsed_time = 0
+            final_status = None
+            while elapsed_time < max_wait_seconds:
+                await asyncio.sleep(poll_interval)
+                elapsed_time += poll_interval
+                logger.debug(f"[Apify Async] Polling status for run {run_id} ({elapsed_time}s elapsed)")
                 try:
-                    results = response.json()
-                    if isinstance(results, list) and len(results) > 0:
-                        item = results[0]
-                        content = item.get("text") or item.get("transcript") or item.get("captions_concatenated")
-                        if not content and item.get("captions") and isinstance(item["captions"], list): logger.info("[Apify] Processing 'captions' format."); content = " ".join(cap.get("text", "") for cap in item["captions"] if cap.get("text"))
-                        if content and isinstance(content, str): logger.info(f"[Apify] Success via REST for {video_url}. Length: {len(content)}"); return content.strip()
-                        else: logger.warning(f"[Apify] Actor success but transcript empty/not found for {video_url}. Item: {item}"); return None
-                    else: logger.warning(f"[Apify] Actor success but dataset empty for {video_url}. Response: {results}"); return None
-                except json.JSONDecodeError: logger.error(f"[Apify] Failed JSON decode for {video_url}. Status:{response.status_code}. Resp:{response.text[:200]}"); return None
-                except Exception as e: logger.error(f"[Apify] Error processing success response for {video_url}: {e}", exc_info=True); return None
-            elif response.status_code == 400: logger.error(f"[Apify] Bad Request (400) for {video_url}. Check payload. Resp:{response.text[:200]}"); return None
-            elif response.status_code == 401: logger.error("[Apify] Auth error (401). Check token."); return None
-            else: logger.error(f"[Apify] Unexpected status {response.status_code} for {video_url}. Resp:{response.text[:200]}"); return None
-    except httpx.TimeoutException: logger.error(f"[Apify] Timeout running actor for {video_url}"); return None
-    except httpx.RequestError as e: logger.error(f"[Apify] Request error running actor for {video_url}: {e}"); return None
-    except Exception as e: logger.error(f"[Apify] Unexpected error during Apify REST call for {video_url}: {e}", exc_info=True); return None
+                    response_status = await client.get(run_status_endpoint, params=params_base)
+                    if response_status.status_code == 200:
+                        status_data = response_status.json().get("data", {})
+                        final_status = status_data.get("status")
+                        logger.debug(f"[Apify Async] Run status: {final_status}")
+                        if final_status in ["SUCCEEDED", "FAILED", "ABORTED", "TIMED-OUT"]:
+                            break # Exit polling loop on terminal status
+                    else:
+                        logger.warning(f"[Apify Async] Non-200 status ({response_status.status_code}) polling run {run_id}.")
+                        await asyncio.sleep(poll_interval * 2) # Wait longer on error
+                except Exception as poll_err:
+                    logger.error(f"[Apify Async] Error polling run status {run_id}: {poll_err}")
+                    await asyncio.sleep(poll_interval * 2) # Wait longer on error
+
+            if final_status != "SUCCEEDED":
+                logger.warning(f"[Apify Async] Run {run_id} did not succeed. Final status: {final_status}")
+                return None
+
+            # 3. Fetch dataset items
+            logger.info(f"[Apify Async] Run {run_id} succeeded. Fetching items from dataset {dataset_id}")
+            dataset_endpoint = f"https://api.apify.com/v2/datasets/{dataset_id}/items"
+            params_dataset = {"token": api_token, "format": "json", "limit": 5}
+            response_dataset = await client.get(dataset_endpoint, params=params_dataset)
+            logger.debug(f"[Apify Async] Dataset fetch status: {response_dataset.status_code}")
+            response_dataset.raise_for_status() # Raise exception for non-2xx status
+
+            results = response_dataset.json()
+            if isinstance(results, list) and len(results) > 0:
+                item = results[0]
+                content = item.get("text") or item.get("transcript") or item.get("captions_concatenated")
+                if not content and item.get("captions") and isinstance(item["captions"], list):
+                    content = " ".join(cap.get("text", "") for cap in item["captions"] if cap.get("text"))
+                if content and isinstance(content, str):
+                    logger.info(f"[Apify Async] Success via ASYNC REST for {video_url}. Length: {len(content)}")
+                    return content.strip()
+                else: logger.warning(f"[Apify Async] Dataset item found but transcript empty/not found for {video_url}. Item: {item}"); return None
+            else: logger.warning(f"[Apify Async] Dataset {dataset_id} was empty for {video_url}. Response: {results}"); return None
+
+    except httpx.TimeoutException as e: logger.error(f"[Apify Async] Timeout during API interaction for {video_url}: {e}"); return None
+    except httpx.HTTPStatusError as e: logger.error(f"[Apify Async] HTTP Status Error during API interaction for {video_url}: {e}"); return None
+    except httpx.RequestError as e: logger.error(f"[Apify Async] Request error during API interaction for {video_url}: {e}"); return None
+    except Exception as e: logger.error(f"[Apify Async] Unexpected error during Apify Async REST call for {video_url}: {e}", exc_info=True); return None
 
 async def get_youtube_transcript(video_id: str, video_url: str) -> Optional[str]:
+    # This function remains the same, calling the updated helper functions
     global SUPADATA_API_KEY, APIFY_API_TOKEN
     if not video_id: logger.error("get_youtube_transcript: No video_id"); return None
     logger.info(f"Fetching transcript for video ID: {video_id} (URL: {video_url})")
@@ -212,16 +279,17 @@ async def get_youtube_transcript(video_id: str, video_url: str) -> Optional[str]
             else: logger.warning(f"[Fallback YT 1] Supadata failed or no content for {video_id}.")
         else: logger.warning("[Fallback YT 1] Supadata API key unavailable. Skipping.")
     if transcript_text is None:
-        logger.info("[Fallback YT 2] Trying Apify REST API...")
+        logger.info("[Fallback YT 2] Trying Apify REST API (Async)...") # Updated log message
         if APIFY_API_TOKEN:
-            transcript_text = await get_transcript_via_apify(video_url, APIFY_API_TOKEN)
-            if transcript_text: logger.info(f"[Fallback YT 2] Success via Apify REST for {video_url}"); return transcript_text
-            else: logger.warning(f"[Fallback YT 2] Apify REST failed or no content for {video_url}.")
+            transcript_text = await get_transcript_via_apify(video_url, APIFY_API_TOKEN) # Calls new async version
+            if transcript_text: logger.info(f"[Fallback YT 2] Success via Apify ASYNC REST for {video_url}"); return transcript_text
+            else: logger.warning(f"[Fallback YT 2] Apify ASYNC REST failed or no content for {video_url}.")
         else: logger.warning("[Fallback YT 2] Apify API token unavailable. Skipping.")
     if transcript_text is None: logger.error(f"All methods failed for YT transcript: {video_id}"); return None
-    return transcript_text
+    return transcript_text # Will be None if all failed
 
 async def get_website_content(url: str) -> Optional[str]:
+    # This function remains the same
     if not url: logger.error("get_website_content: No URL"); return None
     logger.info(f"[Primary Web] Fetching website content for: {url}")
     html_content = await fetch_url_content_for_scrape(url)
@@ -243,6 +311,7 @@ async def get_website_content(url: str) -> Optional[str]:
     except Exception as e: logger.error(f"[Primary Web] Error scraping/parsing {url}: {e}", exc_info=True); return None
 
 async def get_website_content_via_api(url: str, api_key: str) -> Optional[str]:
+    # This function remains the same
     if not url: logger.error("[Fallback Web API] No URL"); return None
     if not api_key: logger.error("[Fallback Web API] urltotext.com API key missing."); return None
     logger.info(f"[Fallback Web API] Attempting fetch for: {url} using urltotext.com API")
@@ -270,6 +339,7 @@ async def get_website_content_via_api(url: str, api_key: str) -> Optional[str]:
     except Exception as e: logger.error(f"[Fallback Web API] Unexpected error during urltotext.com API call for {url}: {e}", exc_info=True); return None
 
 # --- Summarization Function ---
+# (generate_summary remains the same)
 async def generate_summary(text: str, summary_type: str) -> str:
     global OPENROUTER_API_KEY, OPENROUTER_MODEL
     logger.info(f"Generating {summary_type} summary using {OPENROUTER_MODEL}. Input length: {len(text)}")
@@ -280,29 +350,16 @@ async def generate_summary(text: str, summary_type: str) -> str:
     if len(text) > MAX_INPUT_LENGTH: logger.warning(f"Input length ({len(text)}) exceeds limit ({MAX_INPUT_LENGTH}). Truncating."); text = text[:MAX_INPUT_LENGTH] + "... (Content truncated)"
     full_prompt = f"{prompt}\n\n{text}"
     headers = { "Authorization": f"Bearer {OPENROUTER_API_KEY}", "Content-Type": "application/json" }; payload = { "model": OPENROUTER_MODEL, "messages": [{"role": "user", "content": full_prompt}] }; openrouter_api_endpoint = "https://openrouter.ai/api/v1/chat/completions"
-
-    # *** FIX: Increase timeout and add logging ***
-    # Set a longer timeout (e.g., 180 seconds = 3 minutes)
-    api_timeout = 180.0
-    response = None # Initialize response variable
-
+    api_timeout = 180.0; response = None
     try:
         async with httpx.AsyncClient(timeout=api_timeout) as client:
             logger.info(f"Sending request to OpenRouter ({OPENROUTER_MODEL}) with timeout {api_timeout}s...")
             response = await client.post(openrouter_api_endpoint, headers=headers, json=payload)
-            # Check response immediately after await returns
-            if response:
-                 logger.info(f"Received response from OpenRouter. Status code: {response.status_code}")
-            else:
-                 # This case should technically not happen if await returns, but good for debugging
-                 logger.error("No response received from OpenRouter after await completed (unexpected).")
-                 return "Sorry, communication with the AI service failed unexpectedly."
-
-            # Process the response (status code check and JSON parsing)
+            if response: logger.info(f"Received response from OpenRouter. Status code: {response.status_code}")
+            else: logger.error("No response received from OpenRouter after await completed (unexpected)."); return "Sorry, communication with the AI service failed unexpectedly."
             if response.status_code == 200:
                 try:
                     data = response.json()
-                    # ... (rest of the success processing logic remains the same)
                     if data.get("choices") and isinstance(data["choices"], list) and len(data["choices"]) > 0:
                         message = data["choices"][0].get("message")
                         if message and isinstance(message, dict):
@@ -313,38 +370,26 @@ async def generate_summary(text: str, summary_type: str) -> str:
                     else: logger.error(f"Unexpected choices structure in OpenRouter resp: {data.get('choices')}. Full: {data}"); return "Sorry, could not parse AI response (choices)."
                 except json.JSONDecodeError: logger.error(f"Failed JSON decode OpenRouter. Status:{response.status_code}. Resp:{response.text[:500]}"); return "Sorry, failed to understand AI response."
                 except Exception as e: logger.error(f"Error processing OpenRouter success response: {e}", exc_info=True); return "Sorry, error processing AI response."
-            # ... (rest of the error status code handling remains the same)
             elif response.status_code == 401: logger.error("OpenRouter API key invalid (401)."); return "Error: AI model configuration key is invalid."
             elif response.status_code == 402: logger.error("OpenRouter Payment Required (402)."); return "Sorry, AI service limits/payment issue."
             elif response.status_code == 429: logger.warning("OpenRouter Rate Limit Exceeded (429)."); return "Sorry, AI model is busy. Try again."
             elif response.status_code == 500: logger.error(f"OpenRouter Internal Server Error (500). Resp:{response.text[:500]}"); return "Sorry, AI service internal error."
             else: logger.error(f"Unexpected status {response.status_code} from OpenRouter. Resp:{response.text[:500]}"); return f"Sorry, AI service returned unexpected status ({response.status_code})."
-
-    except httpx.TimeoutException:
-        logger.error(f"Timeout error ({api_timeout}s) connecting to OpenRouter API.")
-        return f"Sorry, the request to the AI model timed out after {api_timeout} seconds. The content might be too long or the service busy. Please try again later or with shorter content."
-    except httpx.RequestError as e:
-        logger.error(f"Request error connecting to OpenRouter API: {e}")
-        return "Sorry, there was an error connecting to the AI model service."
+    except httpx.TimeoutException: logger.error(f"Timeout error ({api_timeout}s) connecting to OpenRouter API."); return f"Sorry, the request to the AI model timed out after {api_timeout} seconds. The content might be too long or the service busy. Please try again later or with shorter content."
+    except httpx.RequestError as e: logger.error(f"Request error connecting to OpenRouter API: {e}"); return "Sorry, there was an error connecting to the AI model service."
     except Exception as e:
-        # Catch any other unexpected errors during the request/response cycle
         logger.error(f"Unexpected error in generate_summary (OpenRouter request phase): {e}", exc_info=True)
-        # Log response status if available
         if response: logger.error(f"--> Last response status before error: {response.status_code}")
         return "Sorry, an unexpected error occurred while trying to generate the summary."
 
 # --- Background Task Processing ---
-async def process_summary_task(
-    user_id: int, chat_id: int, message_id_to_edit: Optional[int],
-    url: str, summary_type: str, bot_token: str
-) -> None:
-    task_id = f"{user_id}-{message_id_to_edit or 'new'}"
-    logger.info(f"[Task {task_id}] Starting processing for URL: {url}")
+# (process_summary_task remains the same)
+async def process_summary_task( user_id: int, chat_id: int, message_id_to_edit: Optional[int], url: str, summary_type: str, bot_token: str ) -> None:
+    task_id = f"{user_id}-{message_id_to_edit or 'new'}"; logger.info(f"[Task {task_id}] Starting processing for URL: {url}")
     background_request: Optional[BaseRequest] = None; bot: Optional[Bot] = None
     try: background_request = HTTPXRequest( connect_timeout=15.0, read_timeout=60.0, write_timeout=60.0, pool_timeout=60.0 ); bot = Bot(token=bot_token, request=background_request)
     except Exception as e: logger.critical(f"[Task {task_id}] Failed to create background bot: {e}", exc_info=True); return
-    content = None; user_feedback_message = None; success = False
-    status_message_id = message_id_to_edit; message_to_delete_later_id : Optional[int] = None
+    content = None; user_feedback_message = None; success = False; status_message_id = message_id_to_edit; message_to_delete_later_id : Optional[int] = None
     try:
         processing_message_text = f"Got it! Generating '{summary_type}' summary for:\n`{url}`\n\nThis might take a moment..."
         if status_message_id:
@@ -405,6 +450,7 @@ async def process_summary_task(
         logger.info(f"[Task {task_id}] Task completed. Success: {success}")
 
 # --- Telegram Bot Handlers ---
+# (start, help_command, handle_potential_url, handle_summary_type_callback, error_handler remain the same)
 async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
     user = update.effective_user; mention = user.mention_html()
     if not user or not update.message: return
@@ -468,7 +514,9 @@ async def error_handler(update: object, context: ContextTypes.DEFAULT_TYPE) -> N
     if isinstance(context.error, ignore_errors) and "object has no attribute" in str(context.error): logger.debug(f"Ignoring known/handled error in error_handler: {context.error}"); return
     logger.error("Exception while handling an update:", exc_info=context.error)
 
+
 # --- Bot Setup ---
+# (setup_bot_config remains the same)
 async def setup_bot_config() -> Application:
     logger.info("Configuring Telegram Application..."); global TELEGRAM_TOKEN
     if not TELEGRAM_TOKEN: raise ValueError("TELEGRAM_TOKEN missing.")
@@ -479,6 +527,7 @@ async def setup_bot_config() -> Application:
     application.add_error_handler(error_handler); logger.info("Telegram application handlers configured."); return application
 
 # --- ASGI Lifespan ---
+# (lifespan remains the same)
 @contextlib.asynccontextmanager
 async def lifespan(app: Starlette):
     global ptb_app, WEBHOOK_SECRET, TELEGRAM_TOKEN
@@ -495,7 +544,7 @@ async def lifespan(app: Starlette):
             except Exception as e: logger.warning(f"Could not delete webhook: {e}"); await asyncio.sleep(1)
         space_host = os.environ.get("SPACE_HOST"); webhook_path = "/webhook"; full_webhook_url = None
         if space_host:
-            protocol = "https"; host = space_host.split('://')[-1]; full_webhook_url = f"{protocol}://{host.rstrip('/')}{webhook_path}" # Corrected URL construction
+            protocol = "https"; host = space_host.split('://')[-1]; full_webhook_url = f"{protocol}://{host.rstrip('/')}{webhook_path}"
             if full_webhook_url:
                 logger.info(f"Setting webhook: {full_webhook_url}"); set_webhook_args = { "url": full_webhook_url, "allowed_updates": Update.ALL_TYPES, "drop_pending_updates": True }
                 if WEBHOOK_SECRET: set_webhook_args["secret_token"] = WEBHOOK_SECRET; logger.info("Using webhook secret.")
@@ -524,6 +573,7 @@ async def lifespan(app: Starlette):
         logger.info("ASGI Lifespan: Shutdown complete.")
 
 # --- Starlette Route Handlers ---
+# (health_check and telegram_webhook remain the same)
 async def health_check(request: Request) -> PlainTextResponse:
     global OPENROUTER_MODEL, APIFY_ACTOR_ID, _apify_token_exists; bot_status = "Not Initialized"
     if ptb_app and ptb_app.bot:
@@ -547,10 +597,12 @@ async def telegram_webhook(request: Request) -> Response:
     except Exception as e: logger.error(f"Error processing webhook update: {e}", exc_info=True); return Response(status_code=200) # OK despite error
 
 # --- Create Starlette ASGI Application ---
+# (app definition remains the same)
 app = Starlette( debug=False, lifespan=lifespan, routes=[ Route("/", endpoint=health_check, methods=["GET"]), Route("/webhook", endpoint=telegram_webhook, methods=["POST"]), ] )
 logger.info("Starlette ASGI application created with native routes.")
 
 # --- Development Server Block ---
+# (remains the same)
 if __name__ == '__main__':
     import uvicorn
     logger.warning("Running in development mode using Uvicorn directly")