raspiBlog code bibliotheek

Gedetailleerde Uitleg van de Python Library: raspiBlogLib.py

Deze pagina biedt een gedetailleerde uitleg van functies in het Python-script raspiBlogLib.py. Deze library wordt aangeroepen in de verschillende scripts om date te verzamelen en de blog post te maken.

set_run_id(row, run_id)

Stelt de run_id in voor een gegeven artikelrij.

def set_run_id(row, run_id):
    row["run_id"] = run_id

clear_article_row(row)

Maakt alle velden in een artikelrij leeg.

def clear_article_row(row):
    row.update({
        "run_id": None,
        "full_url": None,
        "topic": None,
        "summary": None,
        "text": None,
        "pub_date": None,
        "title": None,
        "supply_channel": None
    })

maak_DALLE3_PROMPT(client, blog_text, model)

Genereert een DALL-E 3 prompt op basis van een blogtekst.

def maak_DALLE3_PROMPT(client, blog_text, model):
    instruction = ("Maak een gedetailleerde DALLE-3 prompt gebaseerd op de volgende tekst..." + blog_text)
    response = client.chat.completions.create(model=model, messages=[{"role": "user", "content": instruction}])
    return response.choices[0].message.content

read_summaries(file_path)

Leest samenvattingen uit een bestand en zet ze in een dictionary.

def read_summaries(file_path):
    summaries = {}
    with open(file_path, 'r', encoding='utf-8') as file:
        lines = file.readlines()
        current_label, url, summary = None, None, None
        for line in lines:
            if line.startswith("Label: "):
                if current_label and url and summary:
                    summaries.setdefault(current_label, []).append((url, summary))
                current_label = line.split("Label: ")[1].strip()
                url, summary = None, None
            elif line.startswith("URL: "):
                url = line.split("URL: ")[1].strip()
            elif line.startswith("Samenvatting: "):
                summary = line.split("Samenvatting: ")[1].strip()
    return summaries

generate_blog_content(client, summaries, model)

Genereert blogcontent op basis van samenvattingen.

def generate_blog_content(client, summaries, model):
    blog_content = "\n".join(
        generate_blog_content_per_topic(topic, articles, client, model)
        for topic, articles in summaries.items()
    )
    return blog_content

update_run_status(run_id, AI_provider, status)

Werkt de status van een run bij in de database.

def update_run_status(run_id, AI_provider, status):
    now = datetime.now()
    with connect(database=database, user=username, password=password, host=host, port=port) as conn:
        with conn.cursor() as cursor:
            cursor.execute("UPDATE rb_runs SET status = %s, ai_provider = %s, end_datetime = %s WHERE id = %s", 
                           (status, AI_provider, now, run_id))
            conn.commit()

generate_summaries(view_name)

Haalt samenvattingen op uit de database en slaat ze op in een bestand.

def generate_summaries(view_name):
    with connect(database=database, user=username, password=password, host=host, port=port) as conn:
        with conn.cursor(cursor_factory=RealDictCursor) as cursor:
            query = f"SELECT id, label, url, summary FROM public.{view_name};"
            cursor.execute(query)
            rows = cursor.fetchall()
            with open(SUMMARIES_FILE, 'w', encoding='utf-8') as file:
                for row in rows:
                    file.write(f"Label: {row['label']}\nURL: {row['url']}\nSamenvatting: {row['summary']}\n\n")

maak_blogPost(summaries)

Genereert een blogpost met AI-ondersteuning en kiest een beschikbare provider.

def maak_blogPost(summaries):
    client_providers = [
        ("DeepSeek", DEEPSEEK_API_KEY, DEEPSEEK_API_URL, DEEPSEEK_MODEL),
        ("Qwen", QWEN_API_KEY, QWEN_API_URL, QWEN_MODEL),
        ("OpenAI", OPENAI_API_KEY, OPENAI_API_URL, OPENAI_MODEL)
    ]
    
    for provider, api_key, base_url, model in client_providers:
        try:
            client = OpenAI(api_key=api_key, base_url=base_url)
            blog_content = generate_blog_content(client, summaries, model)
            with open(DALLE3_PROMPT, 'w', encoding='utf-8') as file:
                file.write(maak_DALLE3_PROMPT(client, blog_content, model))
            return blog_content, provider
        except Exception as e:
            print(f"Fout met {provider}: {e}")
    return None, None

Deze pagina bevat een gedetailleerde uitleg van de belangrijkste functies uit raspiBlogLib.py.