From 58ef5122acbfd966ca4e533e632384deb31c031b Mon Sep 17 00:00:00 2001 From: Zsolt Tasnadi Date: Wed, 4 Mar 2026 22:19:36 +0100 Subject: [PATCH] refact --- generator.py | 369 +++++++++++++++++++++++++++------------------------ 1 file changed, 197 insertions(+), 172 deletions(-) diff --git a/generator.py b/generator.py index fb74377..0f25cbd 100644 --- a/generator.py +++ b/generator.py @@ -25,16 +25,16 @@ import urllib.request import urllib.error # --------------------------------------------------------------------------- -# Config +# Config & Templates # --------------------------------------------------------------------------- -SOURCE_FILE = "SOURCE.md" -BLOGPOST_FILE = "BLOGPOST.md" -TRANSLATED_FILE = "TRANSLATED_BLOGPOST.md" +SOURCE_FILE = "SOURCE.md" +BLOGPOST_FILE = "BLOGPOST.md" +TRANSLATED_FILE = "TRANSLATED_BLOGPOST.md" INSTRUCTIONS_FILE = "INSTRUCTIONS.md" -GEMINI_MODEL = "gemini-flash-latest" -GEMINI_BASE_URL = "https://generativelanguage.googleapis.com/v1beta/models" +GEMINI_MODEL = "gemini-flash-latest" +GEMINI_BASE_URL = "https://generativelanguage.googleapis.com/v1beta/models" WRITE_PROMPT_TEMPLATE = """Please read the following instructions carefully and follow them to write a blog post. @@ -54,6 +54,63 @@ TRANSLATE_PROMPT_TEMPLATE = """Translate the following Markdown blog post into { {blogpost}""" +# --------------------------------------------------------------------------- +# GraphQL Queries +# --------------------------------------------------------------------------- + +QUERY_GET_PAGE = """ +query ($path: String!) { + pages { + singleByPath(path: $path, locale: "en") { + id + title + description + content + } + } +} +""" + +QUERY_FIND_PAGE = """ +query ($path: String!) { + pages { + singleByPath(path: $path, locale: "en") { + id + } + } +} +""" + +MUTATION_UPDATE_PAGE = """ +mutation ($id: Int!, $content: String!) { + pages { + update(id: $id, content: $content, tags: ["blog"]) { + responseResult { succeeded message } + } + } +} +""" + +MUTATION_CREATE_PAGE = """ +mutation ($path: String!, $title: String!, $content: String!) { + pages { + create( + path: $path + title: $title + content: $content + editor: "markdown" + locale: "en" + isPublished: true + isPrivate: false + tags: ["blog"] + description: "" + ) { + responseResult { succeeded message } + page { id } + } + } +} +""" # --------------------------------------------------------------------------- # Helpers @@ -82,30 +139,6 @@ def http_post(url: str, payload: dict, headers: dict) -> dict: sys.exit(1) -def wiki_graphql(base: str, token: str, query: str, variables: dict = None) -> dict: - url = f"{base}/graphql" - payload = {"query": query} - if variables: - payload["variables"] = variables - headers = { - "Authorization": f"Bearer {token}", - "Content-Type": "application/json", - } - return http_post(url, payload, headers) - - -def gemini_generate(api_key: str, prompt: str) -> str: - url = f"{GEMINI_BASE_URL}/{GEMINI_MODEL}:generateContent" - payload = {"contents": [{"parts": [{"text": prompt}]}]} - headers = {"Content-Type": "application/json", "X-goog-api-key": api_key} - resp = http_post(url, payload, headers) - try: - return resp["candidates"][0]["content"]["parts"][0]["text"] - except (KeyError, IndexError) as e: - print(f"ERROR: Unexpected Gemini response structure: {resp}", file=sys.stderr) - sys.exit(1) - - def to_kebab(text: str) -> str: text = text.lower() text = re.sub(r"[^a-z0-9\s-]", "", text) @@ -128,166 +161,153 @@ def write_file(path: str, content: str) -> None: # --------------------------------------------------------------------------- -# Commands +# Classes # --------------------------------------------------------------------------- -def cmd_fetch(args): - """Download a Wiki.js page as Markdown via GraphQL.""" - base = require_env("WIKI_BASE_DOMAIN") - token = require_env("WIKI_TOKEN") +class WikiJS: + def __init__(self, base_domain: str, token: str): + self.base_domain = base_domain.rstrip("/") + self.token = token + self.api_url = f"{self.base_domain}/graphql" - # Strip base domain from URL if full URL was given, then strip leading slash - page_path = args.url.replace(base, "").lstrip("/") - print(f"→ Fetching wiki page: /{page_path}") - - query = """ - query ($path: String!) { - pages { - singleByPath(path: $path, locale: "en") { - id - title - description - content + def graphql(self, query: str, variables: dict = None) -> dict: + payload = {"query": query} + if variables: + payload["variables"] = variables + headers = { + "Authorization": f"Bearer {self.token}", + "Content-Type": "application/json", } - } - } - """ + return http_post(self.api_url, payload, headers) - resp = wiki_graphql(base, token, query, {"path": page_path}) - page = resp.get("data", {}).get("pages", {}).get("singleByPath") + def get_page(self, path: str): + resp = self.graphql(QUERY_GET_PAGE, {"path": path}) + return resp.get("data", {}).get("pages", {}).get("singleByPath"), resp - if not page: - errors = resp.get("errors", resp) - print(f"ERROR: Page not found at '{page_path}': {errors}", file=sys.stderr) - sys.exit(1) + def find_page_id(self, path: str): + resp = self.graphql(QUERY_FIND_PAGE, {"path": path}) + page = resp.get("data", {}).get("pages", {}).get("singleByPath") + return page.get("id") if page else None - write_file(SOURCE_FILE, page["content"]) + def update_page(self, page_id: int, content: str): + variables = {"id": page_id, "content": content} + resp = self.graphql(MUTATION_UPDATE_PAGE, variables) + return resp.get("data", {}).get("pages", {}).get("update", {}).get("responseResult", {}), resp + + def create_page(self, path: str, title: str, content: str): + variables = {"path": path, "title": title, "content": content} + resp = self.graphql(MUTATION_CREATE_PAGE, variables) + return resp.get("data", {}).get("pages", {}).get("create", {}).get("responseResult", {}), resp -def cmd_write(args): - """Generate a blog post from SOURCE.md using Gemini.""" - api_key = require_env("GEMINI_API_KEY") - original_lang = require_env("ORIGINAL_LANG", "Hungarian") +class GoogleGemini: + def __init__(self, api_key: str, model: str = GEMINI_MODEL): + self.api_key = api_key + self.model = model + self.url = f"{GEMINI_BASE_URL}/{self.model}:generateContent" - instructions = read_file(INSTRUCTIONS_FILE) - source = read_file(SOURCE_FILE) - - print(f"→ Generating blog post in {original_lang} with Gemini...") - - prompt = WRITE_PROMPT_TEMPLATE.format( - instructions=instructions, - original_lang=original_lang, - source=source - ) - - result = gemini_generate(api_key, prompt) - write_file(BLOGPOST_FILE, result) + def generate(self, prompt: str) -> str: + payload = {"contents": [{"parts": [{"text": prompt}]}]} + headers = {"Content-Type": "application/json", "X-goog-api-key": self.api_key} + resp = http_post(self.url, payload, headers) + try: + return resp["candidates"][0]["content"]["parts"][0]["text"] + except (KeyError, IndexError): + print(f"ERROR: Unexpected Gemini response structure: {resp}", file=sys.stderr) + sys.exit(1) -def cmd_translate(args): - """Translate BLOGPOST.md to TRANSLATED_BLOGPOST.md using Gemini.""" - api_key = require_env("GEMINI_API_KEY") - translate_lang = require_env("TRANSLATE_LANG", "English") +class BlogWriter: + def __init__(self): + self.wiki = WikiJS( + require_env("WIKI_BASE_DOMAIN"), + require_env("WIKI_TOKEN") + ) + self.gemini = GoogleGemini( + require_env("GEMINI_API_KEY") + ) - blogpost = read_file(BLOGPOST_FILE) + def fetch(self, url: str): + # Strip base domain from URL if full URL was given, then strip leading slash + page_path = url.replace(self.wiki.base_domain, "").lstrip("/") + print(f"→ Fetching wiki page: /{page_path}") - print(f"→ Translating blog post to {translate_lang} with Gemini...") + page, resp = self.wiki.get_page(page_path) - prompt = TRANSLATE_PROMPT_TEMPLATE.format( - translate_lang=translate_lang, - blogpost=blogpost - ) + if not page: + errors = resp.get("errors", resp) + print(f"ERROR: Page not found at '{page_path}': {errors}", file=sys.stderr) + sys.exit(1) - result = gemini_generate(api_key, prompt) - write_file(TRANSLATED_FILE, result) + write_file(SOURCE_FILE, page["content"]) + def write(self): + original_lang = require_env("ORIGINAL_LANG", "Hungarian") + instructions = read_file(INSTRUCTIONS_FILE) + source = read_file(SOURCE_FILE) -def cmd_upload(args): - """Upload TRANSLATED_BLOGPOST.md to Wiki.js under /blog/{kebab-title}.""" - base = require_env("WIKI_BASE_DOMAIN") - token = require_env("WIKI_TOKEN") + print(f"→ Generating blog post in {original_lang} with Gemini...") - content = read_file(TRANSLATED_FILE) + prompt = WRITE_PROMPT_TEMPLATE.format( + instructions=instructions, + original_lang=original_lang, + source=source + ) - # Extract H1 title - match = re.search(r"^#\s+(.+)", content, re.MULTILINE) - if not match: - print(f"ERROR: No H1 heading found in {TRANSLATED_FILE}", file=sys.stderr) - sys.exit(1) + result = self.gemini.generate(prompt) + write_file(BLOGPOST_FILE, result) - title = match.group(1).strip() - content = re.sub(r"^#\s+.+\n?", "", content, count=1, flags=re.MULTILINE).lstrip("\n") - kebab = to_kebab(title) - page_path = f"blog/{kebab}" + def translate(self): + translate_lang = require_env("TRANSLATE_LANG", "English") + blogpost = read_file(BLOGPOST_FILE) - print(f"→ Uploading to Wiki.js") - print(f" Title : {title}") - print(f" Path : /{page_path}") + print(f"→ Translating blog post to {translate_lang} with Gemini...") - # Check if page already exists - find_query = """ - query ($path: String!) { - pages { - singleByPath(path: $path, locale: "en") { - id - } - } - } - """ - find_resp = wiki_graphql(base, token, find_query, {"path": page_path}) - existing = find_resp.get("data", {}).get("pages", {}).get("singleByPath") - existing_id = existing.get("id") if existing else None + prompt = TRANSLATE_PROMPT_TEMPLATE.format( + translate_lang=translate_lang, + blogpost=blogpost + ) - if existing_id: - print(f" Found existing page id={existing_id}, updating...") - mutation = """ - mutation ($id: Int!, $content: String!) { - pages { - update(id: $id, content: $content, tags: ["blog"]) { - responseResult { succeeded message } - } - } - } - """ - variables = {"id": existing_id, "content": content} - resp = wiki_graphql(base, token, mutation, variables) - result = resp.get("data", {}).get("pages", {}).get("update", {}).get("responseResult", {}) - else: - print(" Page not found, creating new...") - mutation = """ - mutation ($path: String!, $title: String!, $content: String!) { - pages { - create( - path: $path - title: $title - content: $content - editor: "markdown" - locale: "en" - isPublished: true - isPrivate: false - tags: ["blog"] - description: "" - ) { - responseResult { succeeded message } - page { id } - } - } - } - """ - variables = {"path": page_path, "title": title, "content": content} - resp = wiki_graphql(base, token, mutation, variables) - result = resp.get("data", {}).get("pages", {}).get("create", {}).get("responseResult", {}) + result = self.gemini.generate(prompt) + write_file(TRANSLATED_FILE, result) - errors = resp.get("errors") - if errors: - print(f"ERROR: {json.dumps(errors, indent=2)}", file=sys.stderr) - sys.exit(1) + def upload(self): + content = read_file(TRANSLATED_FILE) - if not result.get("succeeded"): - print(f"ERROR: Operation failed: {result.get('message')}", file=sys.stderr) - sys.exit(1) + # Extract H1 title + match = re.search(r"^#\s+(.+)", content, re.MULTILINE) + if not match: + print(f"ERROR: No H1 heading found in {TRANSLATED_FILE}", file=sys.stderr) + sys.exit(1) - print(f"✓ Successfully uploaded to {base}/{page_path}") + title = match.group(1).strip() + content = re.sub(r"^#\s+.+\n?", "", content, count=1, flags=re.MULTILINE).lstrip("\n") + kebab = to_kebab(title) + page_path = f"blog/{kebab}" + + print(f"→ Uploading to Wiki.js") + print(f" Title : {title}") + print(f" Path : /{page_path}") + + existing_id = self.wiki.find_page_id(page_path) + + if existing_id: + print(f" Found existing page id={existing_id}, updating...") + result, resp = self.wiki.update_page(existing_id, content) + else: + print(" Page not found, creating new...") + result, resp = self.wiki.create_page(page_path, title, content) + + errors = resp.get("errors") + if errors: + print(f"ERROR: {json.dumps(errors, indent=2)}", file=sys.stderr) + sys.exit(1) + + if not result.get("succeeded"): + print(f"ERROR: Operation failed: {result.get('message')}", file=sys.stderr) + sys.exit(1) + + print(f"✓ Successfully uploaded to {self.wiki.base_domain}/{page_path}") # --------------------------------------------------------------------------- @@ -305,23 +325,28 @@ def main(): # fetch p_fetch = subparsers.add_parser("fetch", help="Download a Wiki.js page as Markdown") p_fetch.add_argument("url", help="Page path or full URL, e.g. /my-page or https://wiki.example.com/my-page") - p_fetch.set_defaults(func=cmd_fetch) # write - p_write = subparsers.add_parser("write", help=f"Generate blog post from {SOURCE_FILE} using Gemini") - p_write.set_defaults(func=cmd_write) + subparsers.add_parser("write", help=f"Generate blog post from {SOURCE_FILE} using Gemini") # translate - p_translate = subparsers.add_parser("translate", help=f"Translate {BLOGPOST_FILE} using Gemini") - p_translate.set_defaults(func=cmd_translate) + subparsers.add_parser("translate", help=f"Translate {BLOGPOST_FILE} using Gemini") # upload - p_upload = subparsers.add_parser("upload", help=f"Upload {TRANSLATED_FILE} to Wiki.js") - p_upload.set_defaults(func=cmd_upload) + subparsers.add_parser("upload", help=f"Upload {TRANSLATED_FILE} to Wiki.js") args = parser.parse_args() - args.func(args) + writer = BlogWriter() + + if args.command == "fetch": + writer.fetch(args.url) + elif args.command == "write": + writer.write() + elif args.command == "translate": + writer.translate() + elif args.command == "upload": + writer.upload() if __name__ == "__main__": - main() \ No newline at end of file + main()