Skip to content

Commit

Permalink
new
Browse files Browse the repository at this point in the history
  • Loading branch information
@liyaka
liyaka committed Jan 1, 2025
1 parent 15c8bb1 commit c0b3c3a
Showing 1 changed file with 84 additions and 173 deletions.
257 changes: 84 additions & 173 deletions update-notion-database/action.yml
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
name: 'Notion Database Update'
description: 'Updates a Notion database with provided field values, preventing duplicates'
name: 'Update Notion Database'
description: 'Add a new row to a Notion database with field validation and duplicate checking'

inputs:
notion_token:
Expand All @@ -9,209 +9,120 @@ inputs:
description: 'Notion database ID'
required: true
fields_json:
description: 'JSON string of field names and values'
description: 'JSON string containing field names and values'
required: true
unique_fields:
description: 'Comma-separated list of fields that determine uniqueness'
required: false
default: ''
description: 'Comma-separated list of fields to check for duplicates'
required: true

runs:
using: composite
using: "composite"
steps:
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.x'

- name: Install dependencies
shell: bash
run: |
python -m pip install --upgrade pip
pip install notion-df pandas
- name: Create and run update script
shell: bash
- name: Update Notion Database
shell: python
env:
NOTION_TOKEN: ${{ inputs.notion_token }}
NOTION_DATABASE_ID: ${{ inputs.database_id }}
DATABASE_ID: ${{ inputs.database_id }}
FIELDS_JSON: ${{ inputs.fields_json }}
UNIQUE_FIELDS: ${{ inputs.unique_fields }}
run: |
echo "DEBUG:"
echo "::add-mask::false"
echo "NOTION_TOKEN is ${NOTION_TOKEN}"
echo "NOTION_DATABASE_ID is ${NOTION_DATABASE_ID}"
echo "::add-mask::true"
echo "FIELDS_JSON is ${FIELDS_JSON}"
echo "UNIQUE_FIELDS is ${UNIQUE_FIELDS}"
cat << 'EOF' > update_notion.py
import os
import json
import sys
from notion_df import download, upload
import pandas as pd
from datetime import datetime
class NotionFieldError(Exception):
"""Custom exception for Notion field validation errors"""
pass
def parse_value(value, expected_type=None):
if value is None:
return None
if expected_type == 'date' or expected_type == 'datetime':
def validate_field_type(value, expected_type):
if expected_type == "date":
try:
if 'today' in str(value).lower():
return datetime.now().date()
return pd.to_datetime(value).date()
except:
raise NotionFieldError(f"Could not parse '{value}' as a date")
if expected_type == 'number':
try:
if '.' in str(value):
return float(value)
return int(value)
except:
raise NotionFieldError(f"Could not parse '{value}' as a number")
if expected_type in ['select', 'multi-select']:
if isinstance(value, list) and expected_type != 'multi-select':
raise NotionFieldError(f"Field expects single value but got a list: {value}")
return value
return str(value)
def validate_fields(fields, db_schema):
errors = []
for field, properties in db_schema.items():
if properties.get('required', False) and field not in fields:
errors.append(f"Missing required field: {field}")
for field in fields:
if field not in db_schema:
errors.append(f"Unknown field in database: {field}")
if errors:
raise NotionFieldError("\n".join(errors))
def get_database_schema(df):
schema = {}
for column in df.columns:
column_data = df[column].dropna()
if len(column_data) == 0:
schema[column] = {'type': 'text', 'required': False}
continue
sample_value = column_data.iloc[0]
if isinstance(sample_value, (pd.Timestamp, datetime)):
schema[column] = {'type': 'date', 'required': False}
elif isinstance(sample_value, (int, float)):
schema[column] = {'type': 'number', 'required': False}
elif isinstance(sample_value, list):
schema[column] = {'type': 'multi-select', 'required': False}
else:
schema[column] = {'type': 'text', 'required': False}
return schema
def check_duplicate(df, new_row, unique_fields):
if not unique_fields:
return False
query = ' and '.join([
f'(df["{field}"] == new_row["{field}"].iloc[0] or '
f'(pd.isna(df["{field}"]) and pd.isna(new_row["{field}"].iloc[0])))'
for field in unique_fields
])
if query:
matching_rows = df.query(query) if not df.empty else pd.DataFrame()
return len(matching_rows) > 0
return False
def update_notion_database():
datetime.fromisoformat(value.replace('Z', '+00:00'))
return True
except (ValueError, AttributeError):
return False
elif expected_type == "number":
return isinstance(value, (int, float))
elif expected_type == "string":
return isinstance(value, str)
elif expected_type == "boolean":
return isinstance(value, bool)
return True
def main():
try:
notion_token = os.environ.get("NOTION_TOKEN")
if not notion_token:
raise NotionFieldError("NOTION_TOKEN environment variable is not set")
database_id = os.environ.get("NOTION_DATABASE_ID")
if not database_id:
raise NotionFieldError("NOTION_DATABASE_ID environment variable is not set")
fields_json = os.environ.get("FIELDS_JSON")
if not fields_json:
raise NotionFieldError("FIELDS_JSON environment variable is not set")
unique_fields = os.environ.get("UNIQUE_FIELDS", "").strip()
unique_fields = [f.strip() for f in unique_fields.split(',')] if unique_fields else []
# Get environment variables
notion_token = os.environ['NOTION_TOKEN']
database_id = os.environ['DATABASE_ID']
fields_json = os.environ['FIELDS_JSON']
unique_fields = os.environ['UNIQUE_FIELDS'].split(',')
# Parse fields JSON
try:
fields = json.loads(fields_json)
except json.JSONDecodeError as e:
raise NotionFieldError(f"Invalid JSON format in fields_json: {str(e)}")
except json.JSONDecodeError:
print("Error: Invalid JSON format in fields_json")
sys.exit(1)
try:
df = download(
database_id=database_id,
notion_token=notion_token
)
except Exception as e:
raise NotionFieldError(f"Failed to download Notion database: {str(e)}")
db_schema = get_database_schema(df)
validate_fields(fields, db_schema)
for field in unique_fields:
if field not in db_schema:
raise NotionFieldError(f"Unique field '{field}' not found in database schema")
processed_fields = {}
for field, value in fields.items():
try:
expected_type = db_schema[field]['type']
processed_fields[field] = parse_value(value, expected_type)
except Exception as e:
raise NotionFieldError(f"Error processing field '{field}': {str(e)}")
new_row = pd.DataFrame([processed_fields])
if check_duplicate(df, new_row, unique_fields):
print("Skipping update: Row with the same unique field values already exists")
print("Unique fields checked:", unique_fields)
print("Values:", {field: processed_fields[field] for field in unique_fields})
return
updated_df = pd.concat([df, new_row], ignore_index=True)
try:
upload(
df=updated_df,
database_id=database_id,
notion_token=notion_token,
update_existing=False
)
print("Successfully updated Notion database")
print("Added fields:", json.dumps(processed_fields, default=str, indent=2))
except Exception as e:
raise NotionFieldError(f"Failed to upload to Notion database: {str(e)}")
# Download existing database
df_existing = download(
database_id,
notion_token
)
# Check for required fields
if not fields:
print("Error: No fields provided")
sys.exit(1)
# Create new row as DataFrame
new_row = pd.DataFrame([fields])
# Check for duplicates based on unique fields
duplicate_found = False
for unique_field in unique_fields:
if unique_field not in df_existing.columns:
print(f"Warning: Unique field '{unique_field}' not found in database")
continue
if unique_field in fields:
duplicate_mask = df_existing[unique_field] == fields[unique_field]
if duplicate_mask.any():
duplicate_found = True
print(f"Duplicate entry found for field '{unique_field}'")
break
if duplicate_found:
print("Skipping upload due to duplicate entry")
sys.exit(0)
# Append new row and upload
df_updated = pd.concat([df_existing, new_row], ignore_index=True)
except NotionFieldError as e:
print(f"Validation Error: {str(e)}")
exit(1)
upload(
df_updated,
database_id,
notion_token,
chunk_size=1
)
print("Successfully updated Notion database")
except KeyError as e:
print(f"Error: Missing required environment variable: {str(e)}")
sys.exit(1)
except Exception as e:
print(f"Unexpected error: {str(e)}")
exit(1)
print(f"Error: {str(e)}")
sys.exit(1)
if __name__ == "__main__":
update_notion_database()
EOF
python update_notion.py
main()

0 comments on commit c0b3c3a

Please sign in to comment.