feat(Notion): Refactor product DBs for GTM strategy
- Implements a 3-tier database architecture (Canonical Products, Portfolio, Companies) to separate product master data from company-specific portfolio information. - Upgrades import_competitive_radar.py to an intelligent "upsert" script that prevents duplicates by checking for existing entries before importing. - This enables detailed GTM strategy tracking for RoboPlanet products while monitoring competitor portfolios. - Updates documentation to reflect the new architecture and import process.
This commit is contained in:
@@ -1,135 +1,179 @@
|
||||
import json
|
||||
import os
|
||||
import requests
|
||||
import sys
|
||||
|
||||
# Configuration
|
||||
# --- CONFIGURATION ---
|
||||
JSON_FILE = 'analysis_robo-planet.de-4.json'
|
||||
TOKEN_FILE = 'notion_token.txt'
|
||||
PARENT_PAGE_ID = "2e088f42-8544-8024-8289-deb383da3818"
|
||||
NOTION_TOKEN = "" # Will be loaded from file
|
||||
HEADERS = {
|
||||
"Authorization": f"Bearer {NOTION_TOKEN}",
|
||||
"Content-Type": "application/json",
|
||||
"Notion-Version": "2022-06-28",
|
||||
}
|
||||
|
||||
# Database Titles
|
||||
DB_TITLE_HUB = "📦 Competitive Radar (Companies) v6"
|
||||
DB_TITLE_LANDMINES = "💣 Competitive Radar (Landmines) v6"
|
||||
DB_TITLE_REFS = "🏆 Competitive Radar (References) v6"
|
||||
DB_TITLE_PRODUCTS = "🤖 Competitive Radar (Products) v6"
|
||||
# --- DATABASE IDs ---
|
||||
COMPANIES_DB_ID = "2e688f42-8544-8158-8673-d8b1e3eca5b5"
|
||||
CANONICAL_PRODUCTS_DB_ID = "2f088f42-8544-81d5-bec7-d9189f3bacd4"
|
||||
PORTFOLIO_DB_ID = "2e688f42-8544-81df-8fcc-f1d7f8745e00"
|
||||
LANDMINES_DB_ID = "" # Optional: Add if you want to re-import landmines
|
||||
REFERENCES_DB_ID = "" # Optional: Add if you want to re-import references
|
||||
|
||||
def load_json_data(filepath):
|
||||
with open(filepath, 'r') as f:
|
||||
return json.load(f)
|
||||
# --- API HELPERS ---
|
||||
def query_db(db_id, filter_payload=None):
|
||||
"""Retrieves all pages from a Notion database, with optional filter."""
|
||||
url = f"https://api.notion.com/v1/databases/{db_id}/query"
|
||||
all_pages = []
|
||||
start_cursor = None
|
||||
|
||||
while True:
|
||||
payload = {}
|
||||
if start_cursor:
|
||||
payload["start_cursor"] = start_cursor
|
||||
if filter_payload:
|
||||
payload["filter"] = filter_payload
|
||||
|
||||
response = requests.post(url, headers=HEADERS, json=payload)
|
||||
|
||||
if response.status_code != 200:
|
||||
print(f"Error querying DB {db_id}: {response.status_code}")
|
||||
print(response.json())
|
||||
return None
|
||||
|
||||
data = response.json()
|
||||
all_pages.extend(data["results"])
|
||||
|
||||
if data.get("has_more"):
|
||||
start_cursor = data["next_cursor"]
|
||||
else:
|
||||
break
|
||||
|
||||
return all_pages
|
||||
|
||||
def load_notion_token(filepath):
|
||||
with open(filepath, 'r') as f:
|
||||
return f.read().strip()
|
||||
|
||||
def create_database(token, parent_id, title, properties):
|
||||
url = "https://api.notion.com/v1/databases"
|
||||
headers = {"Authorization": f"Bearer {token}", "Notion-Version": "2022-06-28", "Content-Type": "application/json"}
|
||||
payload = {"parent": {"type": "page_id", "page_id": parent_id}, "title": [{"type": "text", "text": {"content": title}}], "properties": properties}
|
||||
r = requests.post(url, headers=headers, json=payload)
|
||||
if r.status_code != 200:
|
||||
print(f"Error creating DB '{title}': {r.text}")
|
||||
sys.exit(1)
|
||||
return r.json()['id']
|
||||
|
||||
def create_page(token, db_id, properties):
|
||||
def create_page(db_id, properties):
|
||||
"""Creates a new page in a Notion database."""
|
||||
url = "https://api.notion.com/v1/pages"
|
||||
headers = {"Authorization": f"Bearer {token}", "Notion-Version": "2022-06-28", "Content-Type": "application/json"}
|
||||
payload = {"parent": {"database_id": db_id}, "properties": properties}
|
||||
r = requests.post(url, headers=headers, json=payload)
|
||||
if r.status_code != 200:
|
||||
print(f"Error creating page: {r.text}")
|
||||
return r.json().get('id')
|
||||
|
||||
response = requests.post(url, headers=HEADERS, data=json.dumps(payload))
|
||||
if response.status_code == 200:
|
||||
return response.json()
|
||||
else:
|
||||
print(f"Error creating page in DB {db_id}: {response.status_code}")
|
||||
print(response.json())
|
||||
return None
|
||||
|
||||
# --- STATE AWARENESS HELPERS ---
|
||||
def get_existing_items_map(db_id, name_property="Name"):
|
||||
"""Fetches all items from a DB and returns a map of {name: id}."""
|
||||
print(f"Fetching existing items from DB {db_id} to build cache...")
|
||||
pages = query_db(db_id)
|
||||
if pages is None:
|
||||
sys.exit(f"Could not fetch items from DB {db_id}. Aborting.")
|
||||
|
||||
item_map = {}
|
||||
for page in pages:
|
||||
try:
|
||||
item_name = page["properties"][name_property]["title"][0]["text"]["content"]
|
||||
item_map[item_name] = page["id"]
|
||||
except (KeyError, IndexError):
|
||||
continue
|
||||
print(f" - Found {len(item_map)} existing items.")
|
||||
return item_map
|
||||
|
||||
def get_existing_portfolio_links(db_id):
|
||||
"""Fetches all portfolio links and returns a set of (company_id, product_id) tuples."""
|
||||
print(f"Fetching existing portfolio links from DB {db_id}...")
|
||||
pages = query_db(db_id)
|
||||
if pages is None:
|
||||
sys.exit(f"Could not fetch portfolio links from DB {db_id}. Aborting.")
|
||||
|
||||
link_set = set()
|
||||
for page in pages:
|
||||
try:
|
||||
company_id = page["properties"]["Related Competitor"]["relation"][0]["id"]
|
||||
product_id = page["properties"]["Canonical Product"]["relation"][0]["id"]
|
||||
link_set.add((company_id, product_id))
|
||||
except (KeyError, IndexError):
|
||||
continue
|
||||
print(f" - Found {len(link_set)} existing portfolio links.")
|
||||
return link_set
|
||||
|
||||
# --- MAIN LOGIC ---
|
||||
def main():
|
||||
token = load_notion_token(TOKEN_FILE)
|
||||
data = load_json_data(JSON_FILE)
|
||||
|
||||
print("🚀 Level 5 Import starting (v6 Databases)...")
|
||||
|
||||
# 1. Create Databases
|
||||
hub_id = create_database(token, PARENT_PAGE_ID, DB_TITLE_HUB, {
|
||||
"Name": {"title": {}},
|
||||
"Website": {"url": {}},
|
||||
"Target Industries": {"multi_select": {}}
|
||||
})
|
||||
|
||||
lm_id = create_database(token, PARENT_PAGE_ID, DB_TITLE_LANDMINES, {
|
||||
"Question": {"title": {}},
|
||||
"Topic": {"select": {}},
|
||||
"Related Competitor": {"relation": {"database_id": hub_id, "dual_property": {"synced_property_name": "Landmines"}}}
|
||||
})
|
||||
|
||||
prod_id = create_database(token, PARENT_PAGE_ID, DB_TITLE_PRODUCTS, {
|
||||
"Product": {"title": {}},
|
||||
"Category": {"select": {}},
|
||||
"Purpose": {"rich_text": {}},
|
||||
"Related Competitor": {"relation": {"database_id": hub_id, "dual_property": {"synced_property_name": "Products"}}}
|
||||
})
|
||||
global NOTION_TOKEN, HEADERS
|
||||
try:
|
||||
with open("notion_token.txt", "r") as f:
|
||||
NOTION_TOKEN = f.read().strip()
|
||||
HEADERS["Authorization"] = f"Bearer {NOTION_TOKEN}"
|
||||
except FileNotFoundError:
|
||||
print("Error: `notion_token.txt` not found.")
|
||||
return
|
||||
|
||||
ref_id = create_database(token, PARENT_PAGE_ID, DB_TITLE_REFS, {
|
||||
"Customer": {"title": {}},
|
||||
"Industry": {"select": {}},
|
||||
"Quote": {"rich_text": {}},
|
||||
"Related Competitor": {"relation": {"database_id": hub_id, "dual_property": {"synced_property_name": "References"}}}
|
||||
})
|
||||
# --- Phase 1: State Awareness ---
|
||||
print("\n--- Phase 1: Reading current state from Notion ---")
|
||||
companies_map = get_existing_items_map(COMPANIES_DB_ID)
|
||||
products_map = get_existing_items_map(CANONICAL_PRODUCTS_DB_ID)
|
||||
portfolio_links = get_existing_portfolio_links(PORTFOLIO_DB_ID)
|
||||
|
||||
# --- Phase 2: Processing JSON ---
|
||||
print("\n--- Phase 2: Processing local JSON file ---")
|
||||
try:
|
||||
with open(JSON_FILE, 'r') as f:
|
||||
data = json.load(f)
|
||||
except FileNotFoundError:
|
||||
print(f"Error: `{JSON_FILE}` not found.")
|
||||
return
|
||||
|
||||
# 2. Import Companies & Products
|
||||
comp_map = {}
|
||||
for analysis in data.get('analyses', []):
|
||||
c = analysis['competitor']
|
||||
name = c['name']
|
||||
competitor = analysis['competitor']
|
||||
competitor_name = competitor['name']
|
||||
print(f"\nProcessing competitor: {competitor_name}")
|
||||
|
||||
# --- Phase 3: "Upsert" Company ---
|
||||
if competitor_name not in companies_map:
|
||||
print(f" - Company '{competitor_name}' not found. Creating...")
|
||||
props = {"Name": {"title": [{"text": {"content": competitor_name}}]}}
|
||||
new_company = create_page(COMPANIES_DB_ID, props)
|
||||
if new_company:
|
||||
companies_map[competitor_name] = new_company["id"]
|
||||
else:
|
||||
print(f" - Failed to create company '{competitor_name}'. Skipping.")
|
||||
continue
|
||||
|
||||
# v5: 'target_industries' is at root level of analysis object
|
||||
industries = analysis.get('target_industries', [])
|
||||
|
||||
props = {
|
||||
"Name": {"title": [{"text": {"content": name}}]},
|
||||
"Website": {"url": c['url'] or "https://google.com"},
|
||||
"Target Industries": {"multi_select": [{"name": i[:100].replace(',', '')} for i in industries if i]}
|
||||
}
|
||||
pid = create_page(token, hub_id, props)
|
||||
if pid:
|
||||
comp_map[name] = pid
|
||||
print(f" - Created Company: {name}")
|
||||
company_id = companies_map[competitor_name]
|
||||
|
||||
# --- Phase 4: "Upsert" Products and Portfolio Links ---
|
||||
for product in analysis.get('portfolio', []):
|
||||
product_name = product['product']
|
||||
|
||||
# Upsert Canonical Product
|
||||
if product_name not in products_map:
|
||||
print(f" - Product '{product_name}' not found. Creating canonical product...")
|
||||
props = {"Name": {"title": [{"text": {"content": product_name}}]}}
|
||||
new_product = create_page(CANONICAL_PRODUCTS_DB_ID, props)
|
||||
if new_product:
|
||||
products_map[product_name] = new_product["id"]
|
||||
else:
|
||||
print(f" - Failed to create canonical product '{product_name}'. Skipping.")
|
||||
continue
|
||||
|
||||
for prod in analysis.get('portfolio', []):
|
||||
p_props = {
|
||||
"Product": {"title": [{"text": {"content": prod['product'][:100]}}]},
|
||||
"Category": {"select": {"name": prod.get('category', 'Other')[:100]}},
|
||||
"Purpose": {"rich_text": [{"text": {"content": prod.get('purpose', '')[:2000]}}]},
|
||||
"Related Competitor": {"relation": [{"id": pid}]}
|
||||
product_id = products_map[product_name]
|
||||
|
||||
# Check and create Portfolio Link
|
||||
if (company_id, product_id) not in portfolio_links:
|
||||
print(f" - Portfolio link for '{competitor_name}' -> '{product_name}' not found. Creating...")
|
||||
portfolio_props = {
|
||||
"Product": {"title": [{"text": {"content": f"{competitor_name} - {product_name}"}}]},
|
||||
"Related Competitor": {"relation": [{"id": company_id}]},
|
||||
"Canonical Product": {"relation": [{"id": product_id}]}
|
||||
}
|
||||
create_page(token, prod_id, p_props)
|
||||
new_portfolio_entry = create_page(PORTFOLIO_DB_ID, portfolio_props)
|
||||
if new_portfolio_entry:
|
||||
portfolio_links.add((company_id, product_id)) # Add to cache to prevent re-creation in same run
|
||||
else:
|
||||
print(f" - Portfolio link for '{competitor_name}' -> '{product_name}' already exists. Skipping.")
|
||||
|
||||
# 3. Import Battlecards (Landmines)
|
||||
for card in data.get('battlecards', []):
|
||||
cid = comp_map.get(card['competitor_name'])
|
||||
if not cid: continue
|
||||
for q in card.get('landmine_questions', []):
|
||||
# Handle both string and object formats from LLM
|
||||
text = q['text'] if isinstance(q, dict) else q
|
||||
cat = q.get('category', 'General') if isinstance(q, dict) else 'General'
|
||||
|
||||
create_page(token, lm_id, {
|
||||
"Question": {"title": [{"text": {"content": text[:100]}}]},
|
||||
"Topic": {"select": {"name": cat}},
|
||||
"Related Competitor": {"relation": [{"id": cid}]}
|
||||
})
|
||||
|
||||
# 4. Import References
|
||||
for ref_analysis in data.get('reference_analysis', []):
|
||||
cid = comp_map.get(ref_analysis['competitor_name'])
|
||||
if not cid: continue
|
||||
for ref in ref_analysis.get('references', []):
|
||||
create_page(token, ref_id, {
|
||||
"Customer": {"title": [{"text": {"content": ref['name'][:100]}}]},
|
||||
"Industry": {"select": {"name": ref.get('industry', 'Unknown')[:100].replace(',', '')}},
|
||||
"Quote": {"rich_text": [{"text": {"content": ref.get('testimonial_snippet', '')[:2000]}}]},
|
||||
"Related Competitor": {"relation": [{"id": cid}]}
|
||||
})
|
||||
|
||||
print("✅ DONE")
|
||||
print("\n--- ✅ Import script finished ---")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
main()
|
||||
Reference in New Issue
Block a user