scan_channels.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7ac9964e-153f-482f-babf-3848f2913789",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import boto3\n",
    "import json\n",
    "from concurrent.futures import ThreadPoolExecutor, as_completed\n",
    "from datetime import datetime, timedelta, timezone\n",
    "\n",
    "# CDK created role which has access to lambda/dynamodb/s3/CloudWatch Logs. Copy paste the role from cdk deploy terminal output\n",
    "role_arn = \"arn:aws:iam::736551082663:role/TranscriptCollectorProjec-SageMakerAccessRole9107AB-ndEtKQgg84zA\"\n",
    "session_name = \"SageMakerSession\"\n",
    "\n",
    "# Create an STS client and assume the role\n",
    "sts_client = boto3.client('sts')\n",
    "assumed_role_object = sts_client.assume_role(\n",
    "    RoleArn=role_arn,\n",
    "    RoleSessionName=session_name\n",
    ")\n",
    "\n",
    "# Get temporary credentials and create a new session using the assumed role credentials\n",
    "credentials = assumed_role_object['Credentials']\n",
    "session = boto3.Session(\n",
    "    aws_access_key_id=credentials['AccessKeyId'],\n",
    "    aws_secret_access_key=credentials['SecretAccessKey'],\n",
    "    aws_session_token=credentials['SessionToken'],\n",
    "    region_name='eu-central-1'\n",
    ")\n",
    "\n",
    "lambda_client = session.client('lambda')\n",
    "logs_client = session.client('logs')\n",
    "stage = \"development\"\n",
    "\n",
    "print(\"Assumed role successfully!\")\n",
    "\n",
    "def invoke_lambda(url):\n",
    "    payload = {\"video_url\": url}\n",
    "    lambda_client.invoke(\n",
    "        FunctionName=f'TranscriptCollectorFunction-{stage}',\n",
    "        InvocationType='Event',  # Asynchronous invocation\n",
    "        Payload=json.dumps(payload)\n",
    "    )\n",
    "\n",
    "def get_log_events(log_group_name, filter_pattern, start_time, end_time):\n",
    "    \"\"\"Fetch log events from CloudWatch Logs based on a filter pattern.\"\"\"\n",
    "    response = logs_client.filter_log_events(\n",
    "        logGroupName=log_group_name,\n",
    "        startTime=int(start_time.timestamp() * 1000),\n",
    "        endTime=int(end_time.timestamp() * 1000),\n",
    "        filterPattern=filter_pattern\n",
    "    )\n",
    "    \n",
    "    return response['events']\n",
    "\n",
    "def format_log_message(event):\n",
    "    \"\"\"Format the log message for readability.\"\"\"\n",
    "    log_event = {\n",
    "        'timestamp': datetime.utcfromtimestamp(event['timestamp'] / 1000).isoformat(),\n",
    "        'logStreamName': event['logStreamName'],\n",
    "        'message': event['message']\n",
    "    }\n",
    "    return log_event\n",
    "\n",
    "def fetch_lambda_logs(log_group_name, log_level, hours_back):\n",
    "    \"\"\"Fetch Lambda logs based on log level and time range.\"\"\"\n",
    "    filter_patterns = {\n",
    "        'error': 'ERROR',\n",
    "        'info': 'INFO',\n",
    "        'warning': 'WARNING'\n",
    "    }\n",
    "    \n",
    "    filter_pattern = filter_patterns.get(log_level.lower())\n",
    "    \n",
    "    if not filter_pattern:\n",
    "        raise ValueError(\"Invalid log level specified. Choose from 'error', 'info', or 'warning'.\")\n",
    "    \n",
    "    end_time = datetime.now(timezone.utc)\n",
    "    start_time = end_time - timedelta(hours=hours_back)\n",
    "    \n",
    "    events = get_log_events(log_group_name, filter_pattern, start_time, end_time)\n",
    "    \n",
    "    return [format_log_message(event) for event in events]\n",
    "\n",
    "def main(urls):\n",
    "    # Invoke Lambda functions asynchronously\n",
    "    max_workers = 100  # Start with a high value, adjust based on your testing\n",
    "    with ThreadPoolExecutor(max_workers=max_workers) as executor:\n",
    "        future_to_url = {executor.submit(invoke_lambda, url): url for url in urls}\n",
    "        for future in as_completed(future_to_url):\n",
    "            future.result()  # To catch any exceptions raised during invocation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6b3b4a73-cdc4-4fbc-bb9b-001465c8c001",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import random\n",
    "from pytubefix import Channel\n",
    "\n",
    "# c = Channel(\"https://www.youtube.com/@baldandbankrupt\")\n",
    "# c = Channel(\"https://www.youtube.com/@SabineHossenfelder\")\n",
    "c = Channel(\"https://www.youtube.com/@primitivetechnology9550\")\n",
    "\n",
    "print(f'Channel name: {c.channel_name}')\n",
    "print(f'Total videos: {len(c.videos)}')\n",
    "\n",
    "def construct_video_url(video_id):\n",
    "    return f'https://youtube.com/watch?v={video_id}'\n",
    "\n",
    "# video_urls = [construct_video_url(video.video_id) for video in c.videos[200:300]]\n",
    "video_urls = [construct_video_url(video.video_id) for video in c.videos]\n",
    "\n",
    "\n",
    "results = main(video_urls)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3b2b99ae-fca7-4cae-b0d7-1c3cf388306a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "video_urls = [construct_video_url(\"YVkUvmDQ3HY\")]\n",
    "\n",
    "results = main(video_urls)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f7c833d0-f8d1-4568-9813-3c0696b7ff09",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Fetch and print logs\n",
    "log_group_name = f'/aws/lambda/TranscriptCollectorFunction-{stage}'\n",
    "log_level = 'error'  # Can be 'error', 'info', or 'warning'\n",
    "hours_back = 0.5  # Number of hours in the past\n",
    "\n",
    "logs = fetch_lambda_logs(log_group_name, log_level, hours_back)\n",
    "\n",
    "for log in logs:\n",
    "    print(f\"Timestamp: {log['timestamp']}\")\n",
    "    print(f\"Log Stream: {log['logStreamName']}\")\n",
    "    print(f\"Message: {log['message']}\")\n",
    "    print(\"=\"*60)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8fe70f6d-359d-4f86-812d-79ff82f63c36",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}