armistace/blog_creator
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects 1 Releases Wiki Activity

env vars and starting work on repo_manager

Browse Source
This commit is contained in:
armistace
2025-05-23 15:47:25 +10:00
parent 8a64d9c959
commit c606f72d90
9 changed files with 195 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ai_generators/ollama_md_generator.py
+33 -20
View File
@@ -1,11 +1,11 @@
import os, re
import os, re, json, random, time
from ollama import Client
import chromadb, time
import chromadb
from langchain_ollama import ChatOllama
class OllamaGenerator:
def __init__(self, title: str, content: str, model: str, inner_title: str):
def __init__(self, title: str, content: str, inner_title: str):
self.title = title
self.inner_title = inner_title
self.content = content
@@ -13,15 +13,15 @@ class OllamaGenerator:
self.chroma = chromadb.HttpClient(host="172.18.0.2", port=8000)
ollama_url = f"{os.environ["OLLAMA_PROTOCOL"]}://{os.environ["OLLAMA_HOST"]}:{os.environ["OLLAMA_PORT"]}"
self.ollama_client = Client(host=ollama_url)
self.ollama_model = model
self.embed_model = "snowflake-arctic-embed2:latest"
self.agent_models = ["openthinker:7b", "deepseek-r1:7b", "qwen2.5:7b", "gemma3:latest"]
self.llm = ChatOllama(model=self.ollama_model, temperature=0.7)
self.ollama_model = os.environ["EDITOR_MODEL"]
self.embed_model = os.environ["EMBEDDING_MODEL"]
self.agent_models = json.loads(os.environ["CONTENT_CREATOR_MODELS"])
self.llm = ChatOllama(model=self.ollama_model, temperature=0.6, top_p=0.5) #This is the level head in the room
self.prompt_inject = f"""
You are a journalist, Software Developer and DevOps expert
writing a 1000 word draft blog for other tech enthusiasts.
You like to use almost no code examples and prefer to talk
in a light comedic tone. You are also Australian
in a light comedic tone. You are also Australian
As this person write this blog as a markdown document.
The title for the blog is {self.inner_title}.
Do not output the title in the markdown.
@@ -50,16 +50,24 @@ class OllamaGenerator:
chunks.append(' '.join(current_chunk))
return chunks
def generate_draft(self, model) -> str:
'''Generate a draft blog post using the specified model'''
try:
agent_llm = ChatOllama(model=model, temperature=0.8)
# the idea behind this is to make the "creativity" random amongst the content creators
# contorlling temperature will allow cause the output to allow more "random" connections in sentences
# Controlling top_p will tighten or loosen the embedding connections made
# The result should be varied levels of "creativity" in the writing of the drafts
# for more see https://python.langchain.com/v0.2/api_reference/ollama/chat_models/langchain_ollama.chat_models.ChatOllama.html
temp = random.uniform(0.5, 1.0)
top_p = random.uniform(0.4, 0.8)
top_k = int(random.uniform(30, 80))
agent_llm = ChatOllama(model=model, temperature=temp, top_p=top_p, top_k=top_k)
messages = [
("system", self.prompt_inject),
("system", self.prompt_inject),
("human", "make the blog post in a format to be edited easily" )
]
self.response = agent_llm.invoke(messages)
response = agent_llm.invoke(messages)
# self.response = self.ollama_client.chat(model=model,
# messages=[
# {
@@ -67,11 +75,13 @@ class OllamaGenerator:
# 'content': f'{self.prompt_inject}',
# },
# ])
return self.response.text()#['message']['content']
#print ("draft")
#print (response)
return response.text()#['message']['content']
except Exception as e:
raise Exception(f"Failed to generate blog draft: {e}")
def get_draft_embeddings(self, draft_chunks):
'''Get embeddings for the draft chunks'''
embeds = self.ollama_client.embed(model=self.embed_model, input=draft_chunks)
@@ -96,16 +106,16 @@ class OllamaGenerator:
collection.add(documents=draft_chunks, embeddings=embeds, ids=ids, metadatas=metadata)
return collection
def generate_markdown(self) -> str:
prompt_system = f"""
You are an editor taking information from {len(self.agent_models)} Software
Developers and Data experts
You are an editor taking information from {len(self.agent_models)} Software
Developers and Data experts
writing a 3000 word blog for other tech enthusiasts.
You like when they use almost no code examples and the
voice is in a light comedic tone. You are also Australian
You like when they use almost no code examples and the
voice is in a light comedic tone. You are also Australian
As this person produce and an amalgamtion of this blog as a markdown document.
The title for the blog is {self.inner_title}.
Do not output the title in the markdown. Avoid repeated sentences
@@ -118,6 +128,7 @@ class OllamaGenerator:
collection_query = collection.query(query_embeddings=query_embed, n_results=100)
print("Showing pertinent info from drafts used in final edited edition")
pertinent_draft_info = '\n\n'.join(collection.query(query_embeddings=query_embed, n_results=100)['documents'][0])
#print(pertinent_draft_info)
prompt_human = f"Generate the final document using this information from the drafts: {pertinent_draft_info} - ONLY OUTPUT THE MARKDOWN"
print("Generating final document")
messages = [("system", prompt_system), ("human", prompt_human),]
@@ -129,6 +140,8 @@ class OllamaGenerator:
# 'content': f'{prompt_enhanced}',
# },
# ])
#print ("Markdown Generated")
#print (self.response)
return self.response#['message']['content']
except Exception as e:
src/main.py
+1 -2
View File
@@ -18,10 +18,9 @@ for note in tril_notes:
print(tril_notes[note]['title'])
# print(tril_notes[note]['content'])
print("Generating Document")
os_friendly_title = convert_to_lowercase_with_underscores(tril_notes[note]['title'])
ai_gen = omg.OllamaGenerator(os_friendly_title,
tril_notes[note]['content'],
"gemma3:latest",
tril_notes[note]['title'])
ai_gen.save_to_file(f"/blog_creator/generated_files/{os_friendly_title}.md")
src/repo_management/push_markdown.py
-48
View File
@@ -1,48 +0,0 @@
import os
import sys
from git import Repo
# Set these variables accordingly
REPO_OWNER = "your_repo_owner"
REPO_NAME = "your_repo_name"
def clone_repo(repo_url, branch="main"):
Repo.clone_from(repo_url, ".", branch=branch)
def create_markdown_file(file_name, content):
with open(f"{file_name}.md", "w") as f:
f.write(content)
def commit_and_push(file_name, message):
repo = Repo(".")
repo.index.add([f"{file_name}.md"])
repo.index.commit(message)
repo.remote().push()
def create_new_branch(branch_name):
repo = Repo(".")
repo.create_head(branch_name).checkout()
repo.head.reference.set_tracking_url(f"https://your_git_server/{REPO_OWNER}/{REPO_NAME}.git/{branch_name}")
repo.remote().push()
if __name__ == "__main__":
if len(sys.argv) < 3:
print("Usage: python push_markdown.py <repo_url> <markdown_file_name>")
sys.exit(1)
repo_url = sys.argv[1]
file_name = sys.argv[2]
# Clone the repository
clone_repo(repo_url)
# Create a new Markdown file with content
create_markdown_file(file_name, "Hello, World!\n")
# Commit and push changes to the main branch
commit_and_push(file_name, f"Add {file_name}.md")
# Create a new branch named after the Markdown file
create_new_branch(file_name)
print(f"Successfully created '{file_name}' branch with '{file_name}.md'.")
src/repo_management/repo_manager.py
+45 -32
View File
@@ -1,39 +1,52 @@
import os
from git import Git
from git.repo import BaseRepository
from git.exc import InvalidGitRepositoryError
from git.remote import RemoteAction
import os, shutil
from git import Repo
from git.exc import GitCommandError
class GitRepository:
# This is designed to be transitory it will desctruvtively create the repo at repo_path
# if you have uncommited changes you can kiss them goodbye!
# Don't use the repo created by this function for dev -> its a tool!
# It is expected that when used you will add, commit, push, delete
def __init__(self, repo_path, username=None, password=None):
git_protocol = os.environ["GIT_PROTOCOL"]
git_remote = os.environ["GIT_REMOTE"]
remote = f"{git_protocol}://{username}:{password}@{git_remote}"
def try_something(test):
if os.path.exists(repo_path):
shutil.rmtree(repo_path)
# Set the path to your blog repo here
blog_repo = "/path/to/your/blog/repo"
Repo.clone_from(remote, repo_path)
self.repo = Repo(repo_path)
self.username = username
self.password = password
def clone(self, remote_url, destination_path):
"""Clone a Git repository with authentication"""
try:
self.repo.clone(remote_url, destination_path)
return True
except GitCommandError as e:
print(f"Cloning failed: {e}")
return False
# Checkout a new branch and create a new file for our blog post
branch_name = "new-post"
try:
repo = Git(blog_repo)
repo.checkout("-b", branch_name, "origin/main")
with open("my-blog-post.md", "w") as f:
f.write(content)
except InvalidGitRepositoryError:
# Handle repository errors gracefully
pass
def fetch(self, remote_name='origin', ref_name='main'):
"""Fetch updates from a remote repository with authentication"""
try:
self.repo.remotes[remote_name].fetch(ref_name=ref_name)
return True
except GitCommandError as e:
print(f"Fetching failed: {e}")
return False
# Add and commit the changes to Git
repo.add("my-blog-post.md")
repo.commit("-m", "Added new blog post about DevOps best practices.")
# Push the changes to Git and create a PR
repo.remote().push("refs/heads/{0}:refs/for/main".format(branch_name), "--set-upstream")
base_branch = "origin/main"
target_branch = "main"
pr_title = "DevOps best practices"
try:
repo.create_head("{0}-{1}", base=base_branch, message="{}".format(pr_title))
except RemoteAction.GitExitStatus as e:
# Handle Git exit status errors gracefully
pass
def pull(self, remote_name='origin', ref_name='main'):
"""Pull updates from a remote repository with authentication"""
try:
self.repo.remotes[remote_name].pull(ref_name=ref_name)
return True
except GitCommandError as e:
print(f"Pulling failed: {e}")
return False
def get_branches(self):
"""List all branches in the repository"""
return [branch.name for branch in self.repo.branches]
src/trilium/notes.py
+8 -7
View File
@@ -11,16 +11,16 @@ class TrilumNotes:
self.token = os.environ.get('TRILIUM_TOKEN')
if not all([self.protocol, self.host, self.port, self.tril_pass]):
print("One or more required environment variables not found. Have you set a .env?")
self.server_url = f'{self.protocol}://{self.host}:{self.port}'
if not self.token:
print("Please run get_token and set your token")
else:
self.ea = ETAPI(self.server_url, self.token)
self.new_notes = None
self.note_content = None
self.new_notes = None
self.note_content = None
def get_token(self):
ea = ETAPI(self.server_url)
if self.tril_pass == None:
@@ -44,10 +44,11 @@ class TrilumNotes:
def get_notes_content(self):
content_dict = {}
if self.new_notes is None:
raise ValueError("How did you do this? new_notes is None!")
for note in self.new_notes['results']:
content_dict[note['noteId']] = {"title" : f"{note['title']}",
content_dict[note['noteId']] = {"title" : f"{note['title']}",
"content" : f"{self._get_content(note['noteId'])}"
}
self.note_content = content_dict
return content_dict