Floke
57206d0916
- Implement relational data structure in Notion as per the plan. - Add scripts for initial data import (import_product.py) and distribution to related databases (distribute_product_data.py). - Create helper scripts for reading Notion content. - Update Notion_Dashboard.md and GEMINI.md with the latest implementation status, database IDs, and key lessons learned from the MVP phase, including API constraints and schema-first principles.
264 lines
11 KiB
Python
264 lines
11 KiB
Python
import requests
|
|
import json
|
|
import re
|
|
import os
|
|
import time
|
|
|
|
# --- Configuration ---
|
|
# NOTION_TOKEN wird jetzt aus der Datei gelesen
|
|
try:
|
|
with open("notion_token.txt", "r") as f:
|
|
NOTION_TOKEN = f.read().strip()
|
|
except FileNotFoundError:
|
|
print("Error: notion_token.txt not found.")
|
|
print("Please create the notion_token.txt file with your Notion integration token.")
|
|
exit(1)
|
|
|
|
NOTION_VERSION = "2022-06-28"
|
|
NOTION_API_BASE_URL = "https://api.notion.com/v1"
|
|
|
|
HEADERS = {
|
|
"Authorization": f"Bearer {NOTION_TOKEN}",
|
|
"Notion-Version": NOTION_VERSION,
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
# --- Database IDs (from Notion_Dashboard.md) ---
|
|
DB_IDS = {
|
|
"Product Master": "2e288f42-8544-81d8-96f5-c231f84f719a",
|
|
"Sector & Persona Master": "2e288f42-8544-8113-b878-ec99c8a02a6b",
|
|
"Messaging Matrix": "2e288f42-8544-81b0-83d4-c16623cc32d1",
|
|
}
|
|
|
|
# --- Helper Functions ---
|
|
|
|
def clean_json_response(text):
|
|
if text.startswith("```json") and text.endswith("```"):
|
|
return text[7:-3].strip()
|
|
return text
|
|
|
|
def create_notion_page(database_id, properties):
|
|
"""Creates a new page in a Notion database."""
|
|
url = f"{NOTION_API_BASE_URL}/pages"
|
|
payload = {
|
|
"parent": {"database_id": database_id},
|
|
"properties": properties,
|
|
}
|
|
try:
|
|
response = requests.post(url, headers=HEADERS, json=payload)
|
|
response.raise_for_status()
|
|
print(f"Successfully created page in DB {database_id}.")
|
|
return response.json()
|
|
except requests.exceptions.HTTPError as e:
|
|
print(f"HTTP Error creating page in DB {database_id}: {e}")
|
|
print(f"Response content: {response.text}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"An unexpected error occurred while creating a page: {e}")
|
|
return None
|
|
|
|
def update_notion_page(page_id, properties):
|
|
"""Updates an existing page in Notion."""
|
|
url = f"{NOTION_API_BASE_URL}/pages/{page_id}"
|
|
payload = {
|
|
"properties": properties
|
|
}
|
|
try:
|
|
response = requests.patch(url, headers=HEADERS, json=payload)
|
|
response.raise_for_status()
|
|
print(f"Successfully updated page {page_id}.")
|
|
return response.json()
|
|
except requests.exceptions.HTTPError as e:
|
|
print(f"HTTP Error updating page {page_id}: {e}")
|
|
print(f"Response content: {response.text}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"An unexpected error occurred while updating a page: {e}")
|
|
return None
|
|
|
|
def find_notion_page_by_title(database_id, title):
|
|
"""Searches for a page in a Notion database by its title property."""
|
|
url = f"{NOTION_API_BASE_URL}/databases/{database_id}/query"
|
|
filter_payload = {
|
|
"filter": {
|
|
"property": "Name",
|
|
"title": {"equals": title}
|
|
}
|
|
}
|
|
try:
|
|
response = requests.post(url, headers=HEADERS, json=filter_payload)
|
|
response.raise_for_status()
|
|
results = response.json().get("results")
|
|
if results:
|
|
return results[0] # Return the first matching page
|
|
return None
|
|
except requests.exceptions.HTTPError as e:
|
|
print(f"HTTP Error searching page in DB {database_id}: {e}")
|
|
print(f"Response content: {response.text}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"An unexpected error occurred while searching for a page: {e}")
|
|
return None
|
|
|
|
def get_database_properties(database_id):
|
|
"""Retrieves the properties (schema) of a Notion database."""
|
|
url = f"{NOTION_API_BASE_URL}/databases/{database_id}"
|
|
try:
|
|
response = requests.get(url, headers=HEADERS)
|
|
response.raise_for_status()
|
|
return response.json().get("properties")
|
|
except requests.exceptions.HTTPError as e:
|
|
print(f"HTTP Error retrieving database properties for DB {database_id}: {e}")
|
|
print(f"Response content: {response.text}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"An unexpected error occurred while retrieving database properties: {e}")
|
|
return None
|
|
|
|
def format_rich_text(text):
|
|
"""Formats a string into Notion's rich text structure."""
|
|
return {"rich_text": [{"type": "text", "text": {"content": text}}]}
|
|
|
|
def format_title(text):
|
|
"""Formats a string into Notion's title structure."""
|
|
return {"title": [{"type": "text", "text": {"content": text}}]}
|
|
|
|
def format_relation(page_ids):
|
|
"""Formats a list of page IDs into Notion's relation structure."""
|
|
if not isinstance(page_ids, list):
|
|
page_ids = [page_ids] # Ensure it's a list
|
|
return {"relation": [{"id": page_id} for page_id in page_ids]}
|
|
|
|
def extract_section(content, title):
|
|
"""Extracts a section from markdown content based on a ## title."""
|
|
pattern = re.compile(rf"## {re.escape(title)}\n(.*?)(?=\n## |\Z)", re.S)
|
|
match = pattern.search(content)
|
|
return match.group(1).strip() if match else ""
|
|
|
|
# --- Main Import Logic ---
|
|
|
|
def main():
|
|
if NOTION_TOKEN == "YOUR_NOTION_TOKEN":
|
|
print("ERROR: Please replace 'YOUR_NOTION_TOKEN' in the script with your actual Notion token.")
|
|
return
|
|
|
|
# 1. Read the markdown file
|
|
try:
|
|
with open("Puma_m20_2026-01-08.md", "r", encoding="utf-8") as f:
|
|
md_content = f.read()
|
|
except FileNotFoundError:
|
|
print("ERROR: 'Puma_m20_2026-01-08.md' not found. Please make sure the file is in the same directory.")
|
|
return
|
|
|
|
# Define the product name
|
|
PRODUCT_NAME = "Puma M20" # This will be replaced by the user's actual product name.
|
|
|
|
# --- Phase 1: Prepare Product Data ---
|
|
print(f"--- Phase 1: Preparing Product Data for {PRODUCT_NAME} ---")
|
|
product_analysis = extract_section(md_content, "2. Product Analysis")
|
|
key_features = re.search(r"\*\*Key Features:\*\*(.*?)\*\*Constraints:\*\*", product_analysis, re.S).group(1).strip()
|
|
constraints = re.search(r"\*\*Constraints:\*\*(.*)", product_analysis, re.S).group(1).strip()
|
|
target_audience = extract_section(md_content, "3. Target Audience")
|
|
strategy_matrix = extract_section(md_content, "5. Strategy Matrix")
|
|
if len(strategy_matrix) > 2000:
|
|
strategy_matrix = strategy_matrix[:2000] # Truncate to 2000 characters
|
|
print("Warning: 'Strategy Matrix' content truncated to 2000 characters due to Notion API limit.")
|
|
feature_translator = extract_section(md_content, "FEATURE-TO-VALUE TRANSLATOR (PHASE 9)")
|
|
|
|
product_properties = {
|
|
"Name": format_title(PRODUCT_NAME),
|
|
"Beschreibung": format_rich_text("Ein geländegängiger, wetterfester Roboter, der für anspruchsvolle Umgebungen konzipiert wurde."),
|
|
"Key Features": format_rich_text(key_features),
|
|
"Constraints": format_rich_text(constraints),
|
|
"Target Audience": format_rich_text(target_audience),
|
|
"Strategy Matrix": format_rich_text(strategy_matrix),
|
|
"Feature-to-Value Translator": format_rich_text(feature_translator),
|
|
"Layer": {"multi_select": [{"name": "Security"}, {"name": "Service"}]}
|
|
}
|
|
|
|
# Check if product already exists
|
|
existing_product_page = find_notion_page_by_title(DB_IDS["Product Master"], PRODUCT_NAME)
|
|
product_page_id = None
|
|
if existing_product_page:
|
|
product_page_id = existing_product_page["id"]
|
|
print(f"Product '{PRODUCT_NAME}' already exists with ID: {product_page_id}. Updating...")
|
|
updated_page = update_notion_page(product_page_id, product_properties)
|
|
if not updated_page:
|
|
print("Failed to update product page. Aborting.")
|
|
return
|
|
else:
|
|
print(f"Product '{PRODUCT_NAME}' not found. Creating new page...")
|
|
new_product_page = create_notion_page(DB_IDS["Product Master"], product_properties)
|
|
if not new_product_page:
|
|
print("Failed to create product page. Aborting.")
|
|
return
|
|
product_page_id = new_product_page["id"]
|
|
print(f"Created Product '{PRODUCT_NAME}' with ID: {product_page_id}")
|
|
|
|
|
|
# --- Phase 2: Create Sectors in Sector & Persona Master ---
|
|
print("\n--- Phase 2: Creating Sectors ---")
|
|
sector_pages = {}
|
|
sectors = {
|
|
"Chemieparks/Petrochemische Anlagen": {
|
|
"definition": "Anlagen dieser Art haben ausgedehnte Gelände, komplexe Infrastruktur und hohe Sicherheitsanforderungen...",
|
|
"pains": "Umfangreiche Gelände erfordern ständige Sicherheits- und Inspektionsrundgänge, oft unter gefährlichen Bedingungen. Personalmangel und hohe Kosten für manuelle Inspektionen.",
|
|
"personas": ["Head of Security", "Werkschutzleiter", "Geschäftsführer/Vorstand", "Leiter Instandhaltung / Betriebsleiter"]
|
|
},
|
|
"Energieversorgungsunternehmen (z.B. Windparks, Solarparks)": {
|
|
"definition": "Diese Anlagen erstrecken sich oft über große, schwer zugängliche Gebiete...",
|
|
"pains": "Weitläufige Anlagen in oft unwegsamem Gelände. Schwierige und teure Inspektion von Solarmodulen oder Windkraftanlagen. Anfälligkeit für Vandalismus und Diebstahl.",
|
|
"personas": ["Head of Security", "Geschäftsführer/Vorstand", "Leiter Instandhaltung / Betriebsleiter"]
|
|
},
|
|
"Logistikzentren/Großflächenlager": {
|
|
"definition": "Große Lagerflächen und komplexe Logistikprozesse erfordern eine ständige Überwachung und Inspektion.",
|
|
"pains": "Hohe Anforderungen an Sicherheit und Ordnung in großen Lagerhallen... Ineffiziente manuelle Reinigung großer Flächen. Gefahr von Unfällen...",
|
|
"personas": ["Leiter Instandhaltung / Betriebsleiter", "Geschäftsführer/Vorstand", "Head of Security"]
|
|
}
|
|
|
|
}
|
|
|
|
for name, data in sectors.items():
|
|
sector_properties = {
|
|
"Name": format_title(name),
|
|
"RoboPlanet-Definition": format_rich_text(data["definition"]),
|
|
"Pains": format_rich_text(data["pains"]),
|
|
"Personas": {"multi_select": [{"name": p} for p in data["personas"]]}
|
|
}
|
|
sector_page = create_notion_page(DB_IDS["Sector & Persona Master"], sector_properties)
|
|
if sector_page:
|
|
sector_pages[name] = sector_page["id"]
|
|
print(f"Created Sector '{name}' with ID: {sector_page['id']}")
|
|
else:
|
|
print(f"Failed to create sector '{name}'.")
|
|
|
|
|
|
# --- Phase 3: Create Messaging Elements ---
|
|
print("\n--- Phase 3: Creating Messaging Elements (Battlecards) ---")
|
|
battlecards_content = extract_section(md_content, "Kill-Critique Battlecards")
|
|
battlecards = re.findall(r"### Persona: (.*?)\n> \*\*Objection:\*\* \"(.*?)\"\n\n\*\*Response:\*\* (.*?)(?=\n\n---|\Z)", battlecards_content, re.S)
|
|
|
|
for persona, objection, response in battlecards:
|
|
# Determine which sector this battlecard applies to
|
|
current_sector_id = None
|
|
if "Chemiepark" in response or "Wackler Security" in response:
|
|
current_sector_id = sector_pages.get("Chemieparks/Petrochemische Anlagen")
|
|
if "Logistik" in response or "Reinigung" in response:
|
|
current_sector_id = sector_pages.get("Logistikzentren/Großflächenlager")
|
|
|
|
message_properties = {
|
|
"Name": format_title(f"Objection: {objection}"),
|
|
"Satz 1": format_rich_text(f"Persona: {persona.strip()}\nObjection: {objection}"),
|
|
"Satz 2": format_rich_text(response.strip()),
|
|
"Product Master": format_relation(product_page_id),
|
|
}
|
|
if current_sector_id:
|
|
message_properties["Sector Master"] = format_relation(current_sector_id)
|
|
|
|
create_notion_page(DB_IDS["Messaging Matrix"], message_properties)
|
|
|
|
print("\nImport process complete.")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|