جمع بيانات YouTube لـ RAG: دليل باستخدام بايثون والوكيل

```html

تتطلب أنظمة RAG (توليد معزز بالاسترجاع) بيانات عالية الجودة للتدريب. YouTube هو مصدر هائل للمحتوى المنظم: مقاطع الفيديو مع الترجمة، البيانات الوصفية، التعليقات. في هذه المقالة، سنستعرض كيفية جمع بيانات YouTube بشكل فعال لـ RAG، مع تجنب الحظر والامتثال لحدود API.

ما هو RAG ولماذا تعتبر بيانات YouTube ضرورية

RAG (توليد معزز بالاسترجاع) هو نهج لبناء أنظمة الذكاء الاصطناعي حيث يتم تعزيز نموذج اللغة بقاعدة معرفية. بدلاً من الاعتماد فقط على البيانات التي تم تدريب النموذج عليها، يقوم RAG باستخراج المعلومات ذات الصلة من مصدر خارجي ويستخدمها لتوليد الإجابات.

يحتوي YouTube على ملايين الساعات من المحتوى مع ترجمات بلغات مختلفة. وهذا يجعل المنصة مصدراً قيماً للبيانات لأنظمة RAG في مجالات متعددة:

أنظمة تعليمية — محاضرات، دروس، دورات مع توقيتات
وثائق تقنية — أدلة فيديو للبرمجة، DevOps، إعداد البرمجيات
قواعد بيانات طبية — محاضرات الأطباء، تحليل حالات سريرية
تحليلات الأعمال — مقابلات مع خبراء، دراسات حالة، مراجعات السوق
دعم المنتجات — مراجعات المنتجات، الأسئلة الشائعة في صيغة فيديو

ميزة بيانات YouTube هي وجود هيكل: ترجمات مع توقيتات، بيانات وصفية (فئات، علامات)، سياق اجتماعي (تعليقات، إعجابات). كل هذا يساعد نظام RAG على فهم المحتوى والسياق للمعلومات.

ما هي بيانات YouTube المفيدة لأنظمة RAG

لجعل نظام RAG يعمل بشكل فعال، يجب جمع عدة أنواع من البيانات. كل نوع يحل مشاكله الخاصة في عملية الاستخراج والتوليد للمعلومات.

الترجمات (Transcripts)

المصدر الرئيسي للبيانات النصية. يوفر YouTube نوعين من الترجمات:

تلقائية — يتم إنشاؤها بواسطة خوارزميات التعرف على الصوت من Google. متاحة لمعظم مقاطع الفيديو باللغة الإنجليزية وغيرها من اللغات الشائعة. الدقة تتراوح بين 85-95% حسب جودة الصوت.
يدوية — يتم تحميلها من قبل المؤلفين أو المجتمع. أكثر دقة، وغالبًا ما تحتوي على تنسيق وسياق إضافي.

تتضمن الترجمات توقيتات، مما يسمح بربط النص بلحظات معينة من الفيديو. هذا أمر حاسم لإنشاء روابط دقيقة للمصادر في إجابات RAG.

البيانات الوصفية للفيديو

تساعد البيانات الوصفية نظام RAG على فهم السياق وملاءمة المعلومات:

نوع البيانات	الاستخدام في RAG
العنوان والوصف	بحث دلالي، تحديد الموضوع
العلامات والفئات	تصنيف المحتوى، تصفية
تاريخ النشر	ملاءمة المعلومات (مهم للمواضيع التقنية)
المدة	تقييم عمق الموضوع
الإحصائيات (المشاهدات، الإعجابات)	تقييم جودة وشعبية المصدر
معلومات عن القناة	تحديد موثوقية المصدر

العمل مع YouTube Data API v3: الإعداد والحدود

YouTube Data API v3 هو الطريقة الرسمية للحصول على البيانات. يوفر الوصول إلى البيانات الوصفية، الإحصائيات، التعليقات. يتم الحصول على الترجمات من خلال طرق منفصلة.

الحصول على مفتاح API

للعمل مع API، تحتاج إلى مفتاح من Google Cloud Console:

انتقل إلى console.cloud.google.com
أنشئ مشروعًا جديدًا أو اختر مشروعًا موجودًا
قم بتمكين YouTube Data API v3 في قسم "APIs & Services"
أنشئ بيانات الاعتماد (Credentials) → مفتاح API
انسخ المفتاح — ستحتاجه لجميع الطلبات

الحدود والحصص

يستخدم YouTube API نظام الحصص. كل طلب "يكلف" عددًا معينًا من الوحدات:

العملية	التكلفة في الحصص
بحث الفيديو (search.list)	100 وحدة
الحصول على بيانات الفيديو (videos.list)	1 وحدة
الحصول على التعليقات (commentThreads.list)	1 وحدة

الحد اليومي الافتراضي هو 10,000 وحدة. هذا يعادل حوالي 100 طلب بحث أو 10,000 طلب بيانات وصفية. لزيادة الحصة، يجب تقديم طلب إلى Google.

مثال أساسي للعمل مع API

import requests

API_KEY = 'مفتاح_api_الخاص_بك'
BASE_URL = 'https://www.googleapis.com/youtube/v3'

# بحث الفيديو حسب الاستعلام
def search_videos(query, max_results=10):
    url = f'{BASE_URL}/search'
    params = {
        'part': 'snippet',
        'q': query,
        'type': 'video',
        'maxResults': max_results,
        'key': API_KEY
    }
    
    response = requests.get(url, params=params)
    return response.json()

# الحصول على بيانات الفيديو
def get_video_details(video_id):
    url = f'{BASE_URL}/videos'
    params = {
        'part': 'snippet,contentDetails,statistics',
        'id': video_id,
        'key': API_KEY
    }
    
    response = requests.get(url, params=params)
    return response.json()

# مثال للاستخدام
results = search_videos('دورة تعلم الآلة', max_results=5)
for item in results.get('items', []):
    video_id = item['id']['videoId']
    title = item['snippet']['title']
    print(f'ID: {video_id}, العنوان: {title}')
    
    # نحصل على معلومات تفصيلية
    details = get_video_details(video_id)
    stats = details['items'][0]['statistics']
    print(f"المشاهدات: {stats.get('viewCount')}, الإعجابات: {stats.get('likeCount')}")

استخراج ترجمة الفيديو: تلقائي ويدوي

لا يوفر YouTube Data API v3 وصولاً مباشراً إلى الترجمات. يتم استخدام طرق بديلة للحصول عليها.

استخدام مكتبة youtube-transcript-api

أسهل طريقة هي مكتبة youtube-transcript-api لـ Python. تقوم باستخراج الترجمات مباشرة، بدون مفتاح API:

from youtube_transcript_api import YouTubeTranscriptApi

# الحصول على الترجمات
video_id = 'dQw4w9WgXcQ'

try:
    # محاولة الحصول على الترجمات الروسية، إذا لم تكن موجودة — الإنجليزية
    transcript = YouTubeTranscriptApi.get_transcript(video_id, languages=['ru', 'en'])
    
    # عرض الترجمات مع التوقيتات
    for entry in transcript:
        start_time = entry['start']
        duration = entry['duration']
        text = entry['text']
        print(f"[{start_time:.2f}s] {text}")
        
except Exception as e:
    print(f"خطأ في الحصول على الترجمات: {e}")

# الحصول على قائمة اللغات المتاحة
transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)
for transcript in transcript_list:
    print(f"اللغة: {transcript.language}, تلقائية: {transcript.is_generated}")

تحدد المكتبة تلقائيًا الترجمات المتاحة ويمكنها ترجمتها إلى لغات أخرى (إذا كانت YouTube توفر هذه الإمكانية).

معالجة التوقيتات لـ RAG

بالنسبة لأنظمة RAG، من المهم الحفاظ على ارتباط النص بالتوقيتات. هذا يسمح بإنشاء روابط دقيقة للمصادر:

def format_timestamp(seconds):
    """تحويل الثواني إلى تنسيق MM:SS"""
    minutes = int(seconds // 60)
    secs = int(seconds % 60)
    return f"{minutes:02d}:{secs:02d}"

def create_chunks_with_timestamps(transcript, chunk_size=500):
    """تقسيم الترجمات إلى قطع مع الحفاظ على التوقيتات"""
    chunks = []
    current_chunk = ""
    chunk_start_time = 0
    
    for i, entry in enumerate(transcript):
        if len(current_chunk) == 0:
            chunk_start_time = entry['start']
        
        current_chunk += entry['text'] + " "
        
        # إذا وصلنا إلى الحجم المطلوب أو نهاية الترجمات
        if len(current_chunk) >= chunk_size or i == len(transcript) - 1:
            chunks.append({
                'text': current_chunk.strip(),
                'start_time': chunk_start_time,
                'timestamp': format_timestamp(chunk_start_time),
                'video_id': video_id
            })
            current_chunk = ""
    
    return chunks

# الاستخدام
transcript = YouTubeTranscriptApi.get_transcript(video_id)
chunks = create_chunks_with_timestamps(transcript)

for chunk in chunks[:3]:  # أول 3 قطع
    print(f"[{chunk['timestamp']}] {chunk['text'][:100]}...")

جمع البيانات الوصفية: العناوين، الأوصاف، العلامات

تعزز البيانات الوصفية السياق لنظام RAG. إليك مثال كامل لجمع جميع البيانات اللازمة:

import requests
from datetime import datetime

def collect_video_metadata(video_id, api_key):
    """جمع البيانات الوصفية الكاملة للفيديو"""
    url = f'https://www.googleapis.com/youtube/v3/videos'
    params = {
        'part': 'snippet,contentDetails,statistics,topicDetails',
        'id': video_id,
        'key': api_key
    }
    
    response = requests.get(url, params=params)
    data = response.json()
    
    if 'items' not in data or len(data['items']) == 0:
        return None
    
    item = data['items'][0]
    snippet = item['snippet']
    stats = item.get('statistics', {})
    content = item.get('contentDetails', {})
    
    metadata = {
        'video_id': video_id,
        'title': snippet['title'],
        'description': snippet['description'],
        'channel_title': snippet['channelTitle'],
        'channel_id': snippet['channelId'],
        'published_at': snippet['publishedAt'],
        'tags': snippet.get('tags', []),
        'category_id': snippet.get('categoryId'),
        'duration': content.get('duration'),
        'view_count': int(stats.get('viewCount', 0)),
        'like_count': int(stats.get('likeCount', 0)),
        'comment_count': int(stats.get('commentCount', 0)),
        'topics': item.get('topicDetails', {}).get('topicCategories', [])
    }
    
    return metadata

# مثال للاستخدام
metadata = collect_video_metadata('dQw4w9WgXcQ', API_KEY)
print(f"العنوان: {metadata['title']}")
print(f"القناة: {metadata['channel_title']}")
print(f"المشاهدات: {metadata['view_count']:,}")
print(f"العلامات: {', '.join(metadata['tags'][:5])}")

تحديد ملاءمة المحتوى

بالنسبة للمواضيع التقنية، فإن حداثة المعلومات مهمة. دعونا نضيف وظيفة لتقييم الملاءمة:

from datetime import datetime, timedelta

def calculate_content_freshness(published_date_str):
    """تقييم ملاءمة المحتوى"""
    published_date = datetime.fromisoformat(published_date_str.replace('Z', '+00:00'))
    age_days = (datetime.now(published_date.tzinfo) - published_date).days
    
    if age_days < 30:
        return 'جديد جدًا'
    elif age_days < 180:
        return 'جديد'
    elif age_days < 365:
        return 'معتدل'
    else:
        return 'قديم'

def calculate_quality_score(metadata):
    """حساب درجة جودة المصدر"""
    score = 0
    
    # الشعبية
    views = metadata['view_count']
    if views > 100000:
        score += 3
    elif views > 10000:
        score += 2
    elif views > 1000:
        score += 1
    
    # التفاعل (الإعجابات بالنسبة للمشاهدات)
    if views > 0:
        like_ratio = metadata['like_count'] / views
        if like_ratio > 0.05:
            score += 2
        elif like_ratio > 0.02:
            score += 1
    
    # الملاءمة
    freshness = calculate_content_freshness(metadata['published_at'])
    if freshness == 'جديد جدًا':
        score += 2
    elif freshness == 'جديد':
        score += 1
    
    return score

# الاستخدام
metadata = collect_video_metadata('dQw4w9WgXcQ', API_KEY)
quality = calculate_quality_score(metadata)
freshness = calculate_content_freshness(metadata['published_at'])
print(f"درجة الجودة: {quality}/7")
print(f"الملاءمة: {freshness}")

استخراج التعليقات للتحليل السياقي

يمكن أن تحتوي التعليقات على معلومات قيمة: تصحيحات للأخطاء في الفيديو، موارد إضافية، أسئلة شائعة. بالنسبة لأنظمة RAG، هذا سياق إضافي.

def get_video_comments(video_id, api_key, max_results=100):
    """الحصول على التعليقات على الفيديو"""
    url = 'https://www.googleapis.com/youtube/v3/commentThreads'
    comments = []
    next_page_token = None
    
    while len(comments) < max_results:
        params = {
            'part': 'snippet',
            'videoId': video_id,
            'maxResults': min(100, max_results - len(comments)),
            'order': 'relevance',  # الترتيب حسب الملاءمة
            'key': api_key
        }
        
        if next_page_token:
            params['pageToken'] = next_page_token
        
        response = requests.get(url, params=params)
        data = response.json()
        
        if 'items' not in data:
            break
        
        for item in data['items']:
            top_comment = item['snippet']['topLevelComment']['snippet']
            comments.append({
                'author': top_comment['authorDisplayName'],
                'text': top_comment['textDisplay'],
                'like_count': top_comment['likeCount'],
                'published_at': top_comment['publishedAt'],
                'reply_count': item['snippet']['totalReplyCount']
            })
        
        next_page_token = data.get('nextPageToken')
        if not next_page_token:
            break
    
    return comments

def filter_valuable_comments(comments, min_likes=5):
    """تصفية التعليقات القيمة"""
    valuable = []
    
    for comment in comments:
        # معايير القيمة:
        # 1. الكثير من الإعجابات (الشعبية)
        # 2. وجود ردود (أثار نقاش)
        # 3. نص طويل (تعليق مفصل)
        
        if (comment['like_count'] >= min_likes or 
            comment['reply_count'] > 0 or 
            len(comment['text']) > 200):
            valuable.append(comment)
    
    return valuable

# الاستخدام
comments = get_video_comments('dQw4w9WgXcQ', API_KEY, max_results=50)
valuable_comments = filter_valuable_comments(comments)

print(f"إجمالي التعليقات: {len(comments)}")
print(f"التعليقات القيمة: {len(valuable_comments)}")

for comment in valuable_comments[:3]:
    print(f"\n[{comment['like_count']} إعجابات] {comment['author']}:")
    print(comment['text'][:200])

استخدام البروكسي للتوسع في جمع البيانات

عند جمع البيانات على نطاق واسع، تظهر مشكلتان: حدود YouTube API (10,000 حصة في اليوم) والحظر عند استخراج الترجمات. تساعد البروكسي في حل كلا المهمتين.

متى تحتاج إلى بروكسي لاستخراج YouTube

تجاوز حصص API — باستخدام عدة مفاتيح API عبر عناوين IP مختلفة، يمكن زيادة الحد اليومي
استخراج الترجمات بدون API — مكتبة youtube-transcript-api تقوم بإجراء طلبات مباشرة، والتي قد يتم حظرها عند تكرارها بشكل كبير
جمع البيانات من مناطق مختلفة — بعض مقاطع الفيديو متاحة فقط في دول معينة
جمع متوازي — توزيع الحمل على عدة IP لتسريع العملية

اختيار نوع البروكسي

نوع البروكسي	المزايا	متى تستخدمه
مركز البيانات	سرعة عالية، سعر منخفض	العمل مع API، أحجام صغيرة
سكنية	خطر منخفض للحظر، عناوين IP حقيقية	استخراج جماعي للترجمات، تجاوز القيود
محمول	أقصى ثقة، حظر نادر	جمع البيانات من التطبيقات المحمولة، مهام حرجة

لمعظم مهام أنظمة RAG، تناسب البروكسي السكنية — حيث توفر توازنًا بين التكلفة والموثوقية عند استخراج البيانات على نطاق واسع.

إعداد البروكسي في الكود

import requests
from youtube_transcript_api import YouTubeTranscriptApi
from youtube_transcript_api._api import TranscriptListFetcher

# إعداد البروكسي
PROXY = {
    'http': 'http://username:password@proxy-server:port',
    'https': 'http://username:password@proxy-server:port'
}

# للعمل مع API عبر البروكسي
def get_video_details_with_proxy(video_id, api_key, proxy):
    url = f'https://www.googleapis.com/youtube/v3/videos'
    params = {
        'part': 'snippet,statistics',
        'id': video_id,
        'key': api_key
    }
    
    response = requests.get(url, params=params, proxies=proxy, timeout=10)
    return response.json()

# لاستخراج الترجمات عبر البروكسي
class ProxiedTranscriptApi:
    def __init__(self, proxy):
        self.proxy = proxy
    
    def get_transcript(self, video_id, languages=['en']):
        # إنشاء جلسة مخصصة مع البروكسي
        session = requests.Session()
        session.proxies = self.proxy
        
        # استخدام الجلسة للطلبات
        fetcher = TranscriptListFetcher(session)
        transcript_list = fetcher.fetch(video_id)
        
        # الحصول على اللغة المطلوبة
        for lang in languages:
            try:
                transcript = transcript_list.find_transcript([lang])
                return transcript.fetch()
            except:
                continue
        
        raise Exception(f"لم يتم العثور على الترجمات للغات: {languages}")

# الاستخدام
api = ProxiedTranscriptApi(PROXY)
transcript = api.get_transcript('dQw4w9WgXcQ', languages=['ru', 'en'])
print(f"تم الحصول على {len(transcript)} مقاطع من الترجمات")

تدوير البروكسي للتوسع

عند جمع البيانات من آلاف الفيديوهات، من المهم توزيع الحمل بين عدة بروكسي:

import random
import time

class ProxyRotator:
    def __init__(self, proxy_list):
        self.proxies = proxy_list
        self.current_index = 0
    
    def get_next_proxy(self):
        """تدوير متسلسل"""
        proxy = self.proxies[self.current_index]
        self.current_index = (self.current_index + 1) % len(self.proxies)
        return proxy
    
    def get_random_proxy(self):
        """تدوير عشوائي"""
        return random.choice(self.proxies)

# قائمة البروكسي
PROXY_LIST = [
    {'http': 'http://user:pass@proxy1:port', 'https': 'http://user:pass@proxy1:port'},
    {'http': 'http://user:pass@proxy2:port', 'https': 'http://user:pass@proxy2:port'},
    {'http': 'http://user:pass@proxy3:port', 'https': 'http://user:pass@proxy3:port'},
]

rotator = ProxyRotator(PROXY_LIST)

def collect_data_with_rotation(video_ids):
    results = []
    
    for video_id in video_ids:
        proxy = rotator.get_next_proxy()
        
        try:
            # الحصول على البيانات الوصفية
            metadata = get_video_details_with_proxy(video_id, API_KEY, proxy)
            
            # الحصول على الترجمات
            api = ProxiedTranscriptApi(proxy)
            transcript = api.get_transcript(video_id)
            
            results.append({
                'video_id': video_id,
                'metadata': metadata,
                'transcript': transcript
            })
            
            # تأخير بين الطلبات
            time.sleep(1)
            
        except Exception as e:
            print(f"خطأ لـ {video_id}: {e}")
            continue
    
    return results

# الاستخدام
video_ids = ['video1', 'video2', 'video3', 'video4', 'video5']
data = collect_data_with_rotation(video_ids)
print(f"تم جمع البيانات لـ {len(data)} فيديو")

معالجة البيانات وتحضيرها لـ RAG

بعد جمع البيانات، يجب معالجتها وهيكلتها للعمل بشكل فعال مع أنظمة RAG.

إنشاء تمثيلات متجهة

تستخدم أنظمة RAG البحث المتجهي للعثور على مقاطع ذات صلة. يجب تحويل النص إلى تمثيلات متجهة:

from sentence_transformers import SentenceTransformer
import numpy as np

# تحميل النموذج لإنشاء التمثيلات المتجهة
model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

def create_embeddings_from_transcript(transcript_chunks):
    """إنشاء تمثيلات متجهة لقطع الترجمات"""
    embeddings = []
    
    for chunk in transcript_chunks:
        # دمج النص مع البيانات الوصفية للحصول على سياق أفضل
        text_with_context = f"{chunk['title']} | {chunk['text']}"
        
        # إنشاء تمثيل متجه
        embedding = model.encode(text_with_context)
        
        embeddings.append({
            'video_id': chunk['video_id'],
            'timestamp': chunk['timestamp'],
            'text': chunk['text'],
            'embedding': embedding.tolist(),
            'metadata': {
                'title': chunk['title'],
                'channel': chunk['channel'],
                'views': chunk['views']
            }
        })
    
    return embeddings

# تحضير البيانات
def prepare_rag_data(video_data):
    """تحضير جميع البيانات لـ RAG"""
    all_chunks = []
    
    for video in video_data:
        metadata = video['metadata']
        transcript = video['transcript']
        
        # تقسيم الترجمات إلى قطع
        chunks = create_chunks_with_timestamps(transcript)
        
        # إضافة البيانات الوصفية لكل قطعة
        for chunk in chunks:
            chunk['title'] = metadata['title']
            chunk['channel'] = metadata['channel_title']
            chunk['views'] = metadata['view_count']
            all_chunks.append(chunk)
    
    # إنشاء التمثيلات المتجهة
    embeddings = create_embeddings_from_transcript(all_chunks)
    
    return embeddings

# الاستخدام
rag_data = prepare_rag_data(collected_videos)
print(f"تم تحضير {len(rag_data)} مقاطع لـ RAG")

حفظ في قاعدة بيانات متجهة

لضمان البحث الفعال، يتم حفظ التمثيلات المتجهة في قواعد بيانات متخصصة. الخيارات الشائعة: Pinecone، Weaviate، Qdrant، ChromaDB.

import chromadb
from chromadb.config import Settings

# تهيئة ChromaDB (قاعدة بيانات متجهة محلية)
client = chromadb.Client(Settings(
    chroma_db_impl="duckdb+parquet",
    persist_directory="./youtube_rag_db"
))

# إنشاء مجموعة
collection = client.create_collection(
    name="youtube_transcripts",
    metadata={"description": "نصوص فيديوهات YouTube لـ RAG"}
)

def store_in_vector_db(embeddings_data, collection):
    """حفظ التمثيلات المتجهة في قاعدة بيانات متجهة"""
    
    ids = []
    embeddings = []
    documents = []
    metadatas = []
    
    for i, item in enumerate(embeddings_data):
        ids.append(f"{item['video_id']}_{i}")
        embeddings.append(item['embedding'])
        documents.append(item['text'])
        metadatas.append({
            'video_id': item['video_id'],
            'timestamp': item['timestamp'],
            'title': item['metadata']['title'],
            'channel': item['metadata']['channel'],
            'views': str(item['metadata']['views']),
            'youtube_url': f"https://youtube.com/watch?v={item['video_id']}&t={int(float(item['timestamp'].split(':')[0])*60 + float(item['timestamp'].split(':')[1]))}s"
        })
    
    # إضافة إلى المجموعة
    collection.add(
        ids=ids,
        embeddings=embeddings,
        documents=documents,
        metadatas=metadatas
    )
    
    print(f"تم حفظ {len(ids)} تمثيلات في قاعدة بيانات متجهة")

# الاستخدام
store_in_vector_db(rag_data, collection)

البحث وتوليد الإجابات

الخطوة النهائية هي تنفيذ البحث RAG وتوليد الإجابات:

def search_youtube_knowledge(query, collection, model, top_k=3):
    """البحث عن مقاطع ذات صلة من YouTube"""
    
    # إنشاء تمثيل متجه للاستعلام
    query_embedding = model.encode(query).tolist()
    
    # البحث في قاعدة البيانات المتجهة
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k
    )
    
    # تنسيق النتائج
    sources = []
    for i in range(len(results['ids'][0])):
        sources.append({
            'text': results['documents'][0][i],
            'metadata': results['metadatas'][0][i],
            'distance': results['distances'][0][i] if 'distances' in results else None
        })
    
    return sources

def generate_rag_answer(query, sources, llm_api_key):
    """توليد إجابة بناءً على المصادر التي تم العثور عليها"""
    
    # تشكيل السياق من المصادر التي تم العثور عليها
    context = "\n\n".join([
        f"المصدر: {s['metadata']['title']} ({s['metadata']['timestamp']})\n{s['text']}"
        for s in sources
    ])
    
    # المطالبة لـ LLM
    prompt = f"""استنادًا إلى المقاطع التالية من فيديوهات YouTube، أجب على سؤال المستخدم.
تأكد من ذكر المصادر مع التوقيتات.

السياق:
{context}

السؤال: {query}

الإجابة:"""
    
    # هنا استدعاء LLM الخاص بك (OpenAI، Claude، نموذج محلي)
    # مثال مع OpenAI:
    # response = openai.ChatCompletion.create(
    #     model="gpt-4",
    #     messages=[{"role": "user", "content": prompt}]
    # )
    # answer = response.choices[0].message.content
    
    # كمثال، نعيد المطالبة
    return {
        'answer': 'ستكون الإجابة هنا من LLM',
        'sources': sources
    }

# الاستخدام
query = "كيف أعد البروكسي في Python؟"
sources = search_youtube_knowledge(query, collection, model, top_k=3)

print("المصادر التي تم العثور عليها:")
for source in sources:
    print(f"\n{source['metadata']['title']}")
    print(f"الوقت: {source['metadata']['timestamp']}")
    print(f"الرابط: {source['metadata']['youtube_url']}")
    print(f"النص: {source['text'][:200]}...")

تحسين جودة RAG

بعض النصائح لتحسين جودة نظام RAG على بيانات YouTube:

قم بتصفية المحتوى منخفض الجودة — استخدم مقاييس المشاهدات، الإعجابات، حداثة النشر
احتفظ بالسياق — أضف عنوان الفيديو والقناة إلى كل قطعة نصية
قم بتحسين حجم القطع — للمحتوى التقني، 300-500 كلمة هو الأمثل
استخدم البيانات الوصفية للتصنيف — قد يحصل المحتوى الأكثر حداثة وشعبية على أولوية
أضف التعليقات — غالبًا ما تحتوي على توضيحات مهمة وأسئلة شائعة
تحقق من توفر الفيديوهات — قد يتم حذف بعض الفيديوهات أو تصبح خاصة

نصيحة: لجمع البيانات على نطاق واسع من YouTube، يُوصى باستخدام مزيج من API الرسمي (للحصول على البيانات الوصفية) واستخراج البيانات عبر البروكسي (للحصول على الترجمات). هذا يسمح بتجاوز الحدود والحصول على أقصى قدر من المعلومات.

الخاتمة

جمع بيانات YouTube لأنظمة RAG هو عملية متعددة الخطوات، تشمل العمل مع API، استخراج الترجمات، معالجة البيانات الوصفية وإنشاء التمثيلات المتجهة. النقاط الرئيسية:

يوفر YouTube Data API v3 البيانات الوصفية، الإحصائيات والتعليقات مع حد 10,000 حصة في اليوم
تستخرج الترجمات عبر مكتبة youtube-transcript-api أو الطلبات المباشرة
التوقيتات ضرورية لإنشاء روابط دقيقة للمصادر
تساعد البيانات الوصفية (المشاهدات، الإعجابات، التاريخ) في تقييم جودة وملاءمة المحتوى
تضيف التعليقات سياقًا وغالبًا ما تحتوي على أسئلة شائعة
تعتبر البروكسي ضرورية للتوسع وتجاوز الحدود
تضمن التمثيلات المتجهة وقواعد البيانات المتخصصة بحثًا سريعًا دلاليًا

مع الإعداد الصحيح للعملية، يمكن جمع عشرات الآلاف من مقاطع المحتوى عالية الجودة يوميًا، مما يخلق قاعدة معرفية قوية لأنظمة RAG في أي مجال موضوعي.

إذا كنت تخطط لجمع بيانات YouTube على نطاق واسع مع تجاوز الحدود والحظر، نوصي باستخدام البروكسي السكنية — حيث توفر الاستقرار عند استخراج آلاف الفيديوهات وتقلل من خطر الحظر من YouTube.

```

كيفية جمع بيانات يوتيوب لأنظمة RAG: استخراج الترجمة والبيانات الوصفية باستخدام بايثون