import os import requests import json import re from typing import List, Dict, Optional from getpass import getpass # --- API Helper Functions --- def find_database_by_title(token: str, title: str) -> Optional[str]: """Sucht nach einer Datenbank mit einem bestimmten Titel und gibt deren ID zurück.""" print(f"Suche nach Datenbank '{title}' in Notion...") url = "https://api.notion.com/v1/search" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Notion-Version": "2022-06-28" } payload = { "query": title, "filter": {"value": "database", "property": "object"} } try: response = requests.post(url, headers=headers, json=payload) response.raise_for_status() results = response.json().get("results", []) for db in results: db_title_parts = db.get("title", []) if db_title_parts: db_title = db_title_parts[0].get("plain_text", "") if db_title.lower() == title.lower(): db_id = db["id"] print(f"Datenbank '{title}' gefunden mit ID: {db_id}") return db_id print(f"Fehler: Keine Datenbank mit dem exakten Titel '{title}' gefunden.") return None except requests.exceptions.RequestException as e: print(f"Fehler bei der Suche nach der Notion-Datenbank '{title}': {e}") try: print(f"Antwort des Servers: {json.dumps(e.response.json(), indent=2)}") except: print(f"Antwort des Servers: {e.response.text}") return None def query_notion_database(token: str, database_id: str, filter_payload: Dict = None) -> List[Dict]: """Fragt eine Notion-Datenbank ab und gibt eine Liste der Seiten zurück.""" url = f"https://api.notion.com/v1/databases/{database_id}/query" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Notion-Version": "2022-06-28" } payload = {} if filter_payload: payload["filter"] = filter_payload try: response = requests.post(url, headers=headers, json=payload) response.raise_for_status() return response.json().get("results", []) except requests.exceptions.RequestException as e: print(f"Fehler bei der Abfrage der Notion-Datenbank {database_id}: {e}") try: print(f"Antwort des Servers: {json.dumps(e.response.json(), indent=2)}") except: print(f"Antwort des Servers: {e.response.text}") return [] def get_page_title(page: Dict) -> str: """Extrahiert den Titel einer Notion-Seite.""" for prop_value in page.get("properties", {}).values(): if prop_value.get("type") == "title": title_parts = prop_value.get("title", []) if title_parts: return title_parts[0].get("plain_text", "Unbenannt") return "Unbenannt" def get_database_status_options(token: str, db_id: str) -> List[str]: """Ruft die verfügbaren Status-Optionen für eine Datenbank-Eigenschaft ab.""" url = f"https://api.notion.com/v1/databases/{db_id}" headers = { "Authorization": f"Bearer {token}", "Notion-Version": "2022-06-28" } try: response = requests.get(url, headers=headers) response.raise_for_status() properties = response.json().get("properties", {}) status_property = properties.get("Status") if status_property and status_property["type"] == "status": return [option["name"] for option in status_property["status"]["options"]] except requests.exceptions.RequestException as e: print(f"Fehler beim Abrufen der Datenbank-Eigenschaften: {e}") return [] def update_notion_task_status(token: str, task_id: str, status_value: str = "Doing") -> bool: """Aktualisiert den Status eines Notion-Tasks.""" print(f"\n--- Aktualisiere Status von Task '{task_id}' auf '{status_value}'... ---") url = f"https://api.notion.com/v1/pages/{task_id}" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Notion-Version": "2022-06-28" } payload = { "properties": { "Status": { "status": { "name": status_value } } } } try: response = requests.patch(url, headers=headers, json=payload) response.raise_for_status() print(f"✅ Task-Status erfolgreich auf '{status_value}' aktualisiert.") return True except requests.exceptions.RequestException as e: print(f"❌ FEHLER beim Aktualisieren des Task-Status: {e}") try: print(f"Antwort des Servers: {json.dumps(e.response.json(), indent=2)}") except: print(f"Antwort des Servers: {e.response.text}") return False def create_new_notion_task(token: str, project_id: str, tasks_db_id: str) -> Optional[Dict]: """Erstellt einen neuen Task in Notion und verknüpft ihn mit dem Projekt.""" print("\n--- Neuen Task erstellen ---") task_title = input("Bitte gib den Titel für den neuen Task ein: ") if not task_title: print("Kein Titel angegeben. Abbruch.") return None status_options = get_database_status_options(token, tasks_db_id) if not status_options: print("Fehler: Konnte keine Status-Optionen für die Task-Datenbank finden.") return None initial_status = status_options[0] # Nimm den ersten verfügbaren Status url = "https://api.notion.com/v1/pages" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Notion-Version": "2022-06-28" } payload = { "parent": {"database_id": tasks_db_id}, "properties": { "Name": { "title": [{"text": {"content": task_title}}] }, "Project": { "relation": [{"id": project_id}] }, "Status": { "status": {"name": initial_status} } } } try: response = requests.post(url, headers=headers, json=payload) response.raise_for_status() new_task = response.json() print(f"✅ Task '{task_title}' erfolgreich erstellt.") return new_task except requests.exceptions.RequestException as e: print(f"❌ FEHLER beim Erstellen des Tasks: {e}") try: print(f"Antwort des Servers: {json.dumps(e.response.json(), indent=2)}") except: print(f"Antwort des Servers: {e.response.text}") return None # --- UI Functions --- def select_project(token: str) -> Optional[Dict]: """Zeigt eine Liste der Projekte an und lässt den Benutzer eines auswählen.""" print("--- Lade Projekte aus Notion... ---") projects_db_id = find_database_by_title(token, "Projects [UT]") if not projects_db_id: return None projects = query_notion_database(token, projects_db_id) if not projects: print("Keine Projekte in der Datenbank gefunden.") return None print("\nAn welchem Projekt möchtest du arbeiten?") for i, project in enumerate(projects): print(f"[{i+1}] {get_page_title(project)}") while True: try: choice = int(input("Bitte wähle eine Nummer: ")) if 1 <= choice <= len(projects): return projects[choice - 1] else: print("Ungültige Auswahl.") except ValueError: print("Ungültige Eingabe. Bitte eine Zahl eingeben.") def select_task(token: str, project_id: str, tasks_db_id: str) -> Optional[Dict]: """Zeigt eine Liste der Tasks für ein Projekt an und lässt den Benutzer auswählen.""" print("\n--- Lade Tasks für das ausgewählte Projekt... ---") filter_payload = { "property": "Project", "relation": {"contains": project_id} } tasks = query_notion_database(token, tasks_db_id, filter_payload=filter_payload) if not tasks: print("Keine offenen Tasks für dieses Projekt gefunden.") else: print("\nWelchen Task möchtest du bearbeiten?") for i, task in enumerate(tasks): print(f"[{i+1}] {get_page_title(task)}") print(f"[{len(tasks)+1}] Neuen Task für dieses Projekt erstellen") while True: try: choice = int(input("Bitte wähle eine Nummer: ")) if 1 <= choice <= len(tasks): return tasks[choice - 1] elif choice == len(tasks) + 1: return {"id": "new_task"} # Signal else: print("Ungültige Auswahl.") except ValueError: print("Ungültige Eingabe. Bitte eine Zahl eingeben.") # --- Context Generation --- def generate_cli_context(project_title: str, task_title: str, task_id: str, suggested_branch_name: str) -> str: """Erstellt einen formatierten Kontext-String für die Gemini CLI.""" project_to_path_map = { "Company Explorer": "company-explorer/backend/", "B2B Assistant": "b2b-marketing-assistant/", "GTM Tool": "gtm-architect/", "Market Intelligence": "general-market-intelligence/", "Competitor Analysis": "competitor-analysis-app/", "Meeting Assistant": "transcription-tool/", "Sync Engine": "" # Root-Level } base_path = "pfad/zum/modul/" for key, path in project_to_path_map.items(): if key in project_title: base_path = path break context = ( "\n------------------------------------------------------------------" "\n✅ Setup abgeschlossen!" f"\n\nDu solltest jetzt den Git-Branch manuell erstellen und auschecken:" f"\ngit checkout -b {suggested_branch_name}" f"\n\nDer Notion-Task '{task_title}' wurde auf 'Doing' gesetzt." "\n---" "\n\nKontext für die Gemini CLI (bitte kopieren und einfügen):" f"\n\nIch arbeite jetzt am Projekt '{project_title}'. Der Fokus liegt auf dem Task '{task_title}'." "\n\nDie relevanten Dateien für dieses Projekt sind wahrscheinlich:" "\n- Die primäre Projektdokumentation: @readme.md" f"\n- Die spezifische Dokumentation für dieses Modul: @{base_path}README.md (falls vorhanden)" f"\n- Der Haupt-Code befindet sich wahrscheinlich in: @{base_path}**" f"\n\nMein Ziel ist es, den Task '{task_title}' umzusetzen. Alle Commits für diesen Task sollen die Kennung `[{task_id.split('-')[0]}]` enthalten." "\n------------------------------------------------------------------" ) return context # --- Main Execution --- def main(): """Hauptfunktion des Skripts.""" print("Starte interaktive Entwicklungs-Session...") token = os.environ.get('NOTION_API_KEY') if not token: token = getpass("Bitte gib deinen Notion API Key ein (Eingabe wird nicht angezeigt): ") if not token: print("Kein Token angegeben. Abbruch.") return selected_project = select_project(token) if not selected_project: return project_title = get_page_title(selected_project) print(f"\nProjekt '{project_title}' ausgewählt.") tasks_db_id = find_database_by_title(token, "Tasks [UT]") if not tasks_db_id: return user_choice = select_task(token, selected_project["id"], tasks_db_id) if not user_choice: print("Kein Task ausgewählt. Abbruch.") return selected_task = None if user_choice.get("id") == "new_task": selected_task = create_new_notion_task(token, selected_project["id"], tasks_db_id) if not selected_task: return else: selected_task = user_choice task_title = get_page_title(selected_task) task_id = selected_task["id"] print(f"\nTask '{task_title}' ausgewählt.") title_slug = re.sub(r'[^a-z0-9\s-]', '', task_title.lower()) title_slug = re.sub(r'\s+', '-', title_slug) title_slug = re.sub(r'-+', '-', title_slug).strip('-') suggested_branch_name = f"feature/task-{task_id.split('-')[0]}-{title_slug}" status_updated = update_notion_task_status(token, task_id, "Doing") if not status_updated: print("Warnung: Notion-Task-Status konnte nicht aktualisiert werden.") cli_context = generate_cli_context(project_title, task_title, task_id, suggested_branch_name) print(cli_context) if __name__ == "__main__": main()