Files
Brancheneinstufung2/add_product_to_notion.py
Floke a1e1c777be feat(notion): Initial PoC for Notion Integration
- Added documentation for Notion setup and resources (notion_integration.md).

- Added scripts for authentication test, database creation, and product insertion.

- Successfully tested connection and data mapping for 'RoboPlanet Product Master'.
2026-01-06 20:33:18 +00:00

92 lines
2.7 KiB
Python

import requests
import json
import os
TOKEN_FILE = 'notion_api_key.txt'
DATABASE_ID = "2e088f42-8544-815e-a3f9-e226f817bded"
# Data from the VIGGO S100-N analysis
PRODUCT_DATA = {
"specs": {
"metadata": {
"brand": "VIGGO",
"model_name": "S100-N",
"category": "cleaning",
"manufacturer_url": None
},
"core_specs": {
"battery_runtime_min": 360,
"charge_time_min": 270,
"weight_kg": 395.0,
"max_slope_deg": 10.0
},
"layers": {
"cleaning": {
"fresh_water_l": 60.0,
"area_performance_sqm_h": 3000.0
}
}
}
}
def add_to_notion(token):
url = "https://api.notion.com/v1/pages"
headers = {
"Authorization": f"Bearer {token}",
"Notion-Version": "2022-06-28",
"Content-Type": "application/json"
}
specs = PRODUCT_DATA["specs"]
meta = specs["metadata"]
core = specs["core_specs"]
cleaning = specs["layers"].get("cleaning", {})
properties = {
"Model Name": {"title": [{"text": {"content": meta["model_name"]}}]},
"Brand": {"select": {"name": meta["brand"]}},
"Category": {"select": {"name": meta["category"]}},
"Battery Runtime (min)": {"number": core.get("battery_runtime_min")},
"Charge Time (min)": {"number": core.get("charge_time_min")},
"Weight (kg)": {"number": core.get("weight_kg")},
"Max Slope (deg)": {"number": core.get("max_slope_deg")},
"Fresh Water (l)": {"number": cleaning.get("fresh_water_l")},
"Area Performance (m2/h)": {"number": cleaning.get("area_performance_sqm_h")}
}
# Add URL if present
if meta.get("manufacturer_url"):
properties["Manufacturer URL"] = {"url": meta["manufacturer_url"]}
payload = {
"parent": {"database_id": DATABASE_ID},
"properties": properties
}
print(f"Adding {meta['brand']} {meta['model_name']} to Notion database...")
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
data = response.json()
print("\n=== SUCCESS ===")
print(f"Product added to database!")
print(f"Page URL: {data.get('url')}")
except requests.exceptions.HTTPError as e:
print(f"\n=== ERROR ===")
print(f"HTTP Error: {e}")
print(f"Response: {response.text}")
def main():
try:
with open(TOKEN_FILE, 'r') as f:
token = f.read().strip()
except FileNotFoundError:
print(f"Error: Could not find '{TOKEN_FILE}'")
return
add_to_notion(token)
if __name__ == "__main__":
main()