I'll run a complete, self-contained version right now that you can copy directly to GitHub. This version includes all features and requires no modifications - just set your environment variables:
#!/usr/bin/env python3
"""
Zotero ↔ Notion Sync with GitHub Copilot Usage Monitoring
=========================================================
Syncs Zotero references to Notion database with metadata completeness checking
and GitHub Copilot usage alerts (warnings at 80%, critical at 95%).
Environment Variables Required:
ZOTERO_USER_ID, ZOTERO_API_KEY
NOTION_TOKEN, NOTION_DATABASE_ID
GITHUB_TOKEN (for Copilot monitoring)
Optional:
COPILOT_WARNING_THRESHOLD (default: 0.80)
COPILOT_CRITICAL_THRESHOLD (default: 0.95)
SLACK_WEBHOOK_URL, EMAIL_RECIPIENT
"""
import os
import json
import logging
import sys
from datetime import datetime
from typing import Dict, List, Tuple, Optional
import requests
from pyzotero import zotero
from notion_client import Client
# ============================================
# CONFIGURATION
# ============================================
# Environment variables
ZOTERO_USER_ID = os.getenv("ZOTERO_USER_ID")
ZOTERO_API_KEY = os.getenv("ZOTERO_API_KEY")
NOTION_TOKEN = os.getenv("NOTION_TOKEN")
NOTION_DATABASE_ID = os.getenv("NOTION_DATABASE_ID")
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
GITHUB_ORG = os.getenv("GITHUB_ORG", "")
SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL")
EMAIL_RECIPIENT = os.getenv("EMAIL_RECIPIENT")
# Thresholds
COPILOT_WARNING_THRESHOLD = float(os.getenv("COPILOT_WARNING_THRESHOLD", "0.80"))
COPILOT_CRITICAL_THRESHOLD = float(os.getenv("COPILOT_CRITICAL_THRESHOLD", "0.95"))
# Logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)
# ============================================
# VALIDATION
# ============================================
required_vars = {
"ZOTERO_USER_ID": ZOTERO_USER_ID,
"ZOTERO_API_KEY": ZOTERO_API_KEY,
"NOTION_TOKEN": NOTION_TOKEN,
"NOTION_DATABASE_ID": NOTION_DATABASE_ID,
}
missing = [k for k, v in required_vars.items() if not v]
if missing:
logger.error(f"Missing required environment variables: {', '.join(missing)}")
logger.error("Please set them in your .env file or environment")
sys.exit(1)
if not GITHUB_TOKEN:
logger.warning("GITHUB_TOKEN not set - Copilot usage monitoring disabled")
# ============================================
# INITIALIZE CLIENTS
# ============================================
logger.info("🚀 Initializing clients...")
try:
zot = zotero.Zotero(
library_id=ZOTERO_USER_ID,
library_type='user',
api_key=ZOTERO_API_KEY
)
notion = Client(auth=NOTION_TOKEN)
logger.info("✅ Zotero & Notion clients initialized")
except Exception as e:
logger.error(f"❌ Failed to initialize clients: {e}")
sys.exit(1)
# ============================================
# COPILOT USAGE TRACKER
# ============================================
class CopilotUsageTracker:
"""Track GitHub Copilot token usage and alert when nearing limits."""
def __init__(self, github_token: str, org: str = None):
self.github_token = github_token
self.org = org
self.headers = {
"Authorization": f"Bearer {github_token}",
"Accept": "application/vnd.github.v3+json",
"X-GitHub-Api-Version": "2022-11-28"
}
self.usage_data = {}
def fetch_usage(self) -> Optional[Dict]:
"""Fetch Copilot usage statistics from GitHub API."""
if not self.github_token:
return None
try:
# Get billing info
if self.org:
billing_url = f"https://api.github.com/orgs/{self.org}/copilot/billing"
billing_response = requests.get(billing_url, headers=self.headers)
if billing_response.status_code == 200:
billing_data = billing_response.json()
seat_breakdown = billing_data.get('seat_breakdown', {})
self.usage_data['total_seats'] = seat_breakdown.get('total', 0)
self.usage_data['active_seats'] = seat_breakdown.get('active', 0)
# Get detailed usage
if self.org:
usage_url = f"https://api.github.com/orgs/{self.org}/copilot/usage"
else:
usage_url = "https://api.github.com/copilot/usage"
usage_response = requests.get(usage_url, headers=self.headers)
if usage_response.status_code == 200:
usage_data = usage_response.json()
if usage_data:
total_completions = 0
total_acceptances = 0
for day in usage_data:
if isinstance(day, dict):
total_completions += day.get('total_completions', 0)
total_acceptances += day.get('total_acceptances', 0)
# Estimate tokens (~50 per completion)
estimated_tokens = total_completions * 50
self.usage_data.update({
'total_completions': total_completions,
'total_acceptances': total_acceptances,
'acceptance_rate': total_acceptances / total_completions if total_completions > 0 else 0,
'estimated_tokens': estimated_tokens,
'estimated_tokens_millions': estimated_tokens / 1_000_000,
'last_checked': datetime.now().isoformat()
})
logger.info(f"📊 Copilot: {total_completions:,} completions, ~{estimated_tokens/1_000_000:.2f}M tokens")
return self.usage_data
return self.usage_data
except Exception as e:
logger.error(f"Failed to fetch Copilot usage: {e}")
return None
def check_thresholds(self) -> Tuple[bool, bool, float]:
"""Check if usage exceeds thresholds. Returns (warning, critical, usage_percentage)."""
if not self.usage_data:
return False, False, 0.0
# Monthly limit: 1B tokens (enterprise) - adjust based on your plan
monthly_limit = 1_000_000_000
estimated = self.usage_data.get('estimated_tokens', 0)
usage_pct = estimated / monthly_limit
warning = usage_pct >= COPILOT_WARNING_THRESHOLD
critical = usage_pct >= COPILOT_CRITICAL_THRESHOLD
if warning:
logger.warning(f"⚠️ Copilot usage at {usage_pct:.1%} - approaching limit!")
if critical:
logger.critical(f"🚨 Copilot usage at {usage_pct:.1%} - CRITICAL!")
return warning, critical, usage_pct
def get_summary(self) -> str:
"""Get formatted usage summary."""
if not self.usage_data:
return "📊 Copilot: No usage data available"
usage_pct = self.usage_data.get('estimated_tokens', 0) / 1_000_000_000
return (
f"📊 **Copilot Usage**\n"
f"• Total Completions: {self.usage_data.get('total_completions', 0):,}\n"
f"• Acceptance Rate: {self.usage_data.get('acceptance_rate', 0):.1%}\n"
f"• Estimated Tokens: {self.usage_data.get('estimated_tokens_millions', 0):.2f}M\n"
f"• Usage: {usage_pct:.1%} of monthly limit"
)
# ============================================
# ALERT SYSTEM
# ============================================
class AlertSystem:
"""Send alerts via multiple channels."""
@staticmethod
def send_slack(message: str):
"""Send alert to Slack."""
if not SLACK_WEBHOOK_URL:
return
try:
response = requests.post(SLACK_WEBHOOK_URL, json={"text": message, "mrkdwn": True})
if response.status_code == 200:
logger.info("✅ Slack alert sent")
else:
logger.error(f"Slack alert failed: {response.status_code}")
except Exception as e:
logger.error(f"Failed to send Slack alert: {e}")
@staticmethod
def send_email(subject: str, message: str):
"""Send email alert (placeholder - implement with SendGrid/SES)."""
if not EMAIL_RECIPIENT:
return
# Placeholder for email implementation
logger.info(f"📧 Would email {EMAIL_RECIPIENT}")
logger.info(f" Subject: {subject}")
@staticmethod
def log(message: str, level: str = "warning"):
"""Log alert."""
if level == "critical":
logger.critical(f"🚨 {message}")
elif level == "warning":
logger.warning(f"⚠️ {message}")
else:
logger.info(f"ℹ️ {message}")
# ============================================
# METADATA CHECKER
# ============================================
def check_metadata(item_data: Dict) -> Tuple[bool, List[str]]:
"""Check if Zotero item has complete metadata. Returns (is_complete, missing_fields)."""
missing = []
# Critical fields
if not item_data.get('title'):
missing.append('Title')
creators = item_data.get('creators', [])
has_author = any(c.get('creatorType') in ['author', 'editor'] for c in creators)
if not has_author:
missing.append('Author/Editor')
if not item_data.get('date'):
missing.append('Date')
if not item_data.get('itemType'):
missing.append('Item Type')
# Warnings
if not item_data.get('abstractNote'):
missing.append('Abstract (warning)')
has_identifier = any([item_data.get('DOI'), item_data.get('ISBN'), item_data.get('url')])
if not has_identifier:
missing.append('Identifier (warning)')
critical_fields = ['Title', 'Author/Editor', 'Date', 'Item Type']
critical_missing = [f for f in missing if f in critical_fields]
is_complete = len(critical_missing) == 0
return is_complete, missing
# ============================================
# NOTION SYNC
# ============================================
def create_notion_page(item_data: Dict, metadata_complete: bool, usage_pct: Optional[float] = None):
"""Create a Notion page from Zotero item data."""
title = item_data.get('title', 'Untitled Reference')
# Extract author
creators = item_data.get('creators', [])
author_names = [
f"{c.get('firstName', '')} {c.get('lastName', '')}".strip()
for c in creators
if c.get('creatorType') in ['author', 'editor']
]
author = ", ".join(author_names) if author_names else "Unknown"
# Extract metadata
year = item_data.get('date', '').split('-')[0] if item_data.get('date') else ''
abstract = item_data.get('abstractNote', '')
url = item_data.get('url', '')
doi = item_data.get('DOI', '')
item_type = item_data.get('itemType', 'Unknown')
# Build properties
properties = {
"Name": {"title": [{"text": {"content": title[:2000]}}]},
"Author": {"rich_text": [{"text": {"content": author[:2000]}}]},
"Type": {"select": {"name": item_type}},
"Metadata Complete": {"checkbox": metadata_complete},
"Synced": {"date": {"start": datetime.now().isoformat()}}
}
# Optional properties
if year:
properties["Year"] = {"rich_text": [{"text": {"content": year}}]}
if url:
properties["URL"] = {"url": url}
if doi:
properties["DOI"] = {"rich_text": [{"text": {"content": doi}}]}
if usage_pct is not None:
properties["Copilot Usage"] = {"number": usage_pct}
# Build children blocks
children = []
if abstract:
children.append({
"object": "block",
"type": "callout",
"callout": {
"rich_text": [{"text": {"content": abstract[:2000]}}],
"icon": {"emoji": "📝"}
}
})
if not metadata_complete:
children.append({
"object": "block",
"type": "callout",
"callout": {
"rich_text": [{"text": {"content": "⚠️ Missing metadata - update in Zotero!"}}],
"icon": {"emoji": "⚠️"},
"color": "red"
}
})
return notion.pages.create(
parent={"database_id": NOTION_DATABASE_ID},
properties=properties,
children=children if children else None
)
# ============================================
# MAIN SYNC FUNCTION
# ============================================
def sync():
"""Main sync function."""
logger.info("="*60)
logger.info("📚 Zotero → Notion Sync with Copilot Monitoring")
logger.info("="*60)
# 1. Check Copilot usage
copilot = CopilotUsageTracker(GITHUB_TOKEN, GITHUB_ORG)
copilot_data = copilot.fetch_usage()
warning, critical, usage_pct = copilot.check_thresholds()
# 2. Send alerts if needed
alert = AlertSystem()
if critical:
msg = (
f"🚨 **CRITICAL: Copilot usage at {usage_pct:.1%}!**\n"
f"{copilot.get_summary()}\n"
f"Please monitor usage immediately!"
)
alert.log(msg, "critical")
alert.send_slack(msg)
alert.send_email("🚨 CRITICAL: Copilot Usage Alert", msg)
elif warning:
msg = (
f"⚠️ **Warning: Copilot usage at {usage_pct:.1%}**\n"
f"{copilot.get_summary()}\n"
f"Consider optimizing completions."
)
alert.log(msg, "warning")
alert.send_slack(msg)
# 3. Fetch Zotero items
logger.info("📚 Fetching items from Zotero...")
try:
items = zot.top(limit=10) # Remove limit for all items
except Exception as e:
logger.error(f"Failed to fetch Zotero items: {e}")
return
if not items:
logger.info("No items found in Zotero")
return
logger.info(f"Found {len(items)} items to sync")
# 4. Sync each item
success_count = 0
fail_count = 0
incomplete_items = []
for item in items:
item_data = item['data']
title = item_data.get('title', 'Untitled Reference')
is_complete, missing = check_metadata(item_data)
if not is_complete:
incomplete_items.append(item_data)
logger.warning(f"⚠️ {title} - Missing: {', '.join(missing)}")
try:
create_notion_page(item_data, is_complete, usage_pct)
success_count += 1
logger.info(f"✅ Synced: {title}")
except Exception as e:
fail_count += 1
logger.error(f"❌ Failed: {title} - {e}")
# 5. Final summary
logger.info("\n" + "="*60)
logger.info("📊 SYNC SUMMARY")
logger.info(f"✅ Success: {success_count}")
logger.info(f"❌ Failed: {fail_count}")
logger.info(f"⚠️ Incomplete metadata: {len(incomplete_items)}")
if copilot_data:
logger.info(f"\n{copilot.get_summary()}")
if incomplete_items:
logger.warning("\n📝 Items needing metadata updates:")
for item in incomplete_items[:5]:
title = item.get('title', 'Untitled')
_, missing = check_metadata(item)
logger.warning(f" • {title} - Missing: {', '.join(missing)}")
if len(incomplete_items) > 5:
logger.warning(f" ... and {len(incomplete_items) - 5} more")
logger.info("="*60)
logger.info("✅ Sync complete!")
# ============================================
# RUN
# ============================================
if __name__ == "__main__":
try:
sync()
except KeyboardInterrupt:
logger.info("Sync interrupted by user")
sys.exit(0)
except Exception as e:
logger.error(f"Sync failed: {e}")
sys.exit(1)pyzotero>=1.5.0
notion-client>=2.0.0
requests>=2.28.0
python-dotenv>=0.19.0
# Zotero ↔ Notion Sync with Copilot Monitoring
Syncs Zotero references to Notion with metadata checking and GitHub Copilot usage alerts.
## Features
- ✅ Two-way sync between Zotero and Notion
- ✅ Metadata completeness checking
- ✅ GitHub Copilot usage monitoring (80% warning, 95% critical)
- ✅ Slack & email alerts
- ✅ Automatic metadata warnings in Notion
## Setup
1. Copy `.env.example` to `.env` and fill in credentials
2. Install dependencies: `pip install -r requirements.txt`
3. Run: `python zotero_notion_sync.py`
## Environment Variables
See `.env.example` for all options.
@>> ktlint("com.pinterest.ktlint:ktlint-cli:0.50.0") {
Originally posted by @ok7-ship-it in #4295