Floke/Brancheneinstufung2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[2ea88f42] schaut gut aus

Browse Source 
schaut gut aus
This commit is contained in:
Floke
2026-01-29 10:55:26 +00:00
parent 136ed96d26
commit 6c3f033ebb
3 changed files with 339 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.dev_session/SESSION_INFO
View File

@@ -1 +1 @@
{"task_id": "2ea88f42-8544-806f-8946-e61ada8cc059", "token": "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", "session_start_time": "2026-01-29T08:21:44.176590"}
{"task_id": "2ea88f42-8544-806f-8946-e61ada8cc059", "token": "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", "session_start_time": "2026-01-29T10:55:09.010737"}

18
competitor-analysis-app/README.md
View File

@@ -18,3 +18,21 @@ View your app in AI Studio: https://ai.studio/apps/drive/1vJMxbT1hW3SiMDUeEd8cXG
2. Set the `GEMINI_API_KEY` in [.env.local](.env.local) to your Gemini API key
3. Run the app:
`npm run dev`
## Importing Single Competitors
To import a specific competitor from a JSON analysis file into Notion, use the `import_single_competitor.py` script. This script allows you to selectively import data for a single company without affecting other entries.
**Usage:**
```bash
python3 import_single_competitor.py --file <path_to_json_file> --name "<Exact Competitor Name>"
```
**Example:**
To import "FENKA Robotics GmbH" from `Roboplanet_Competitive_Analysis_01_2026.json`:
```bash
python3 import_single_competitor.py --file Roboplanet_Competitive_Analysis_01_2026.json --name "FENKA Robotics GmbH"
```

320
import_single_competitor.py Normal file
View File

@@ -0,0 +1,320 @@
import json
import requests
import sys
import argparse
import re
# --- CONFIGURATION ---
NOTION_TOKEN = "" # Will be loaded from file
HEADERS = {
"Authorization": f"Bearer {NOTION_TOKEN}",
"Content-Type": "application/json",
"Notion-Version": "2022-06-28",
}
# --- 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 = "2e688f42-8544-81aa-94f8-d6242be4d0cd"
REFERENCES_DB_ID = "2e688f42-8544-81df-8d83-f4d7f57d8168"
INDUSTRIES_DB_ID = "2ec88f42-8544-8014-ab38-ea664b4c2b81"
# --- 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 create_page(db_id, properties):
"""Creates a new page in a Notion database."""
url = "https://api.notion.com/v1/pages"
payload = {"parent": {"database_id": db_id}, "properties": properties}
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
def update_page(page_id, properties):
"""Updates properties of an existing page in Notion."""
url = f"https://api.notion.com/v1/pages/{page_id}"
payload = {"properties": properties}
response = requests.patch(url, headers=HEADERS, data=json.dumps(payload))
if response.status_code == 200:
return response.json()
else:
print(f"Error updating page {page_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:
# Handle cases where title might be empty or malformed
title_list = page["properties"][name_property].get("title", [])
if title_list:
item_name = title_list[0].get("text", {}).get("content", "").strip()
if item_name:
item_map[item_name] = page["id"]
except (KeyError, IndexError):
continue
print(f" - Found {len(item_map)} existing items.")
return item_map
def get_existing_relations(db_id, relation_property_name, target_relation_id_prop_name):
"""Fetches all items from a DB and returns a set of (item_name, related_id) tuples."""
print(f"Fetching existing relations from DB {db_id}...")
pages = query_db(db_id)
if pages is None:
sys.exit(f"Could not fetch relations from DB {db_id}. Aborting.")
relation_set = set()
for page in pages:
try:
item_name = page["properties"]["Name"]["title"][0]["text"]["content"]
related_ids = [rel["id"] for rel in page["properties"][relation_property_name].get("relation", [])]
target_related_ids = [rel["id"] for rel in page["properties"][target_relation_id_prop_name].get("relation", [])]
if related_ids and target_related_ids:
relation_set.add((item_name, related_ids[0], target_related_ids[0]))
except (KeyError, IndexError):
continue
print(f" - Found {len(relation_set)} existing relations.")
return relation_set
def inspect_database(db_id):
"""Retrieves and prints the properties of a specific Notion database."""
print(f"🔍 Inspecting properties for database ID: {db_id}")
url = f"https://api.notion.com/v1/databases/{db_id}"
response = requests.get(url, headers=HEADERS)
if response.status_code != 200:
print(f"Error retrieving database properties: {response.status_code}")
print(response.json())
return
data = response.json()
properties = data.get("properties", {})
if not properties:
print("No properties found for this database.")
return
print("\n--- Database Properties ---")
for prop_name, prop_data in properties.items():
print(f"- Property Name: '{prop_name}'")
print(f" Type: {prop_data.get('type')}\n")
print("---------------------------\n")
# --- MAIN LOGIC ---
def main():
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
parser = argparse.ArgumentParser(description="Import a single competitor from a JSON analysis file into Notion.")
parser.add_argument('--file', help="Path to the JSON analysis file.")
parser.add_argument('--name', help="Exact name of the competitor to import.")
parser.add_argument('--inspect', help="Database ID to inspect.")
args = parser.parse_args()
if args.inspect:
inspect_database(args.inspect)
return
if not args.file or not args.name:
parser.error("--file and --name are required.")
return
# --- 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)
industries_map = get_existing_items_map(INDUSTRIES_DB_ID, name_property="Vertical")
# For relations, we create a unique key to check for existence
existing_landmines = {f'{page["properties"]["Question"]["title"][0]["text"]["content"]}_{page["properties"]["Related Competitor"]["relation"][0]["id"]}' for page in query_db(LANDMINES_DB_ID) if "Question" in page["properties"] and page["properties"]["Question"]["title"] and page["properties"]["Related Competitor"]["relation"]}
print(f" - Found {len(existing_landmines)} existing landmines.")
existing_references = {f'{page["properties"]["Customer"]["title"][0]["text"]["content"]}_{page["properties"]["Related Competitor"]["relation"][0]["id"]}' for page in query_db(REFERENCES_DB_ID) if "Customer" in page["properties"] and page["properties"]["Customer"]["title"] and page["properties"]["Related Competitor"]["relation"]}
print(f" - Found {len(existing_references)} existing references.")
json_file_path = args.file
target_competitor_name = args.name
# --- Phase 2: Processing JSON ---
print(f"\n--- Phase 2: Processing local JSON file: {json_file_path} for {target_competitor_name} ---")
try:
with open(json_file_path, 'r', encoding='utf-8') as f:
data = json.load(f)
except FileNotFoundError:
print(f"Error: `{json_file_path}` not found.")
return
except json.JSONDecodeError as e:
print(f"Error decoding JSON from {json_file_path}: {e}")
return
# Find the correct analysis and reference data for the target competitor
target_analysis = None
for analysis in data.get('analyses', []):
if analysis['competitor']['name'] == target_competitor_name:
target_analysis = analysis
break
# Find references from the separate reference_analysis block
target_references_data = None
if 'reference_analysis' in data:
for ref_block in data.get('reference_analysis', []):
if ref_block.get('competitor_name') == target_competitor_name:
target_references_data = ref_block.get('references', [])
break
target_battlecard = None
if 'battlecards' in data:
for bc in data.get('battlecards', []):
if bc['competitor_name'] == target_competitor_name:
target_battlecard = bc
break
if not target_analysis:
print(f"Error: Competitor '{target_competitor_name}' not found in 'analyses' list in {json_file_path}.")
return
print(f"\nProcessing target competitor: {target_competitor_name}")
# --- Phase 3: "Upsert" Company ---
if target_competitor_name not in companies_map:
print(f" - Company '{target_competitor_name}' not found. Creating...")
props = {"Name": {"title": [{"text": {"content": target_competitor_name}}]}}
new_company = create_page(COMPANIES_DB_ID, props)
if new_company:
companies_map[target_competitor_name] = new_company["id"]
else:
print(f" - Failed to create company '{target_competitor_name}'. Halting.")
return
company_id = companies_map[target_competitor_name]
# --- Phase 4: Create and Link Target Industries ---
print("\n--- Processing Target Industries ---")
target_industry_relation_ids = []
if INDUSTRIES_DB_ID:
for industry_name in target_analysis.get('target_industries', []):
if industry_name not in industries_map:
print(f" - Industry '{industry_name}' not found in Notion DB. Creating...")
props = {"Vertical": {"title": [{"text": {"content": industry_name}}]}}
new_industry = create_page(INDUSTRIES_DB_ID, props)
if new_industry:
industries_map[industry_name] = new_industry["id"]
target_industry_relation_ids.append({"id": new_industry["id"]})
else:
target_industry_relation_ids.append({"id": industries_map[industry_name]})
if target_industry_relation_ids:
print(f" - Linking company to {len(target_analysis.get('target_industries', []))} industries...")
# Format for multi-select is a list of objects with names
multi_select_payload = [{"name": name} for name in target_analysis.get('target_industries', [])]
update_props = {
"Target Industries": {"multi_select": multi_select_payload}
}
update_page(company_id, update_props)
else:
print(" - INDUSTRIES_DB_ID not set. Skipping.")
# --- Phase 5: Import Landmines ---
if target_battlecard and LANDMINES_DB_ID:
print("\n--- Processing Landmines ---")
for landmine in target_battlecard.get('landmine_questions', []):
unique_key = f"{landmine}_{company_id}"
if unique_key not in existing_landmines:
print(f" - Landmine '{landmine}' not found. Creating...")
props = {
"Question": {"title": [{"text": {"content": landmine}}]},
"Related Competitor": {"relation": [{"id": company_id}]}
}
new_landmine = create_page(LANDMINES_DB_ID, props)
if new_landmine:
existing_landmines.add(unique_key)
else:
print(f" - Landmine '{landmine}' already exists for this competitor. Skipping.")
# --- Phase 6: Import References ---
if target_references_data and REFERENCES_DB_ID:
print("\n--- Processing References ---")
for ref in target_references_data:
ref_name = ref.get("name", "Unknown Reference")
unique_key = f"{ref_name}_{company_id}"
if unique_key not in existing_references:
print(f" - Reference '{ref_name}' not found. Creating...")
props = {
"Customer": {"title": [{"text": {"content": ref_name}}]},
"Related Competitor": {"relation": [{"id": company_id}]},
"Quote": {"rich_text": [{"text": {"content": ref.get("testimonial_snippet", "")[:2000]}}]}
}
# Handle Industry as a select property
ref_industry_name = ref.get("industry")
if ref_industry_name:
props["Industry"] = {"select": {"name": ref_industry_name}}
new_ref = create_page(REFERENCES_DB_ID, props)
if new_ref:
existing_references.add(unique_key)
else:
print(f" - Reference '{ref_name}' already exists for this competitor. Skipping.")
print("\n--- ✅ Import script finished ---")
if __name__ == "__main__":
main()