How We Grew Trello vs Notion: A Head-to-Head

"""
trello_sync.py — Bidirectional sync between two Trello boards.
Detects conflicts via card.modified timestamps and logs drift.

Requirements: pip install requests python-dotenv
Environment: Python 3.11+, Trello API key + token with read/write scope.
Tested: Trello API v1, 2025-01-15.
"""

import os
import sys
import time
import logging
import requests
from datetime import datetime, timezone
from dotenv import load_dotenv

load_dotenv()

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)

API_KEY = os.getenv("TRELLO_API_KEY")
API_TOKEN = os.getenv("TRELLO_API_TOKEN")
BASE_URL = "https://api.trello.com/1"

if not API_KEY or not API_TOKEN:
    logger.error("TRELLO_API_KEY and TRELLO_API_TOKEN must be set in .env")
    sys.exit(1)


def api_get(endpoint: str, params: dict = None) -> dict | None:
    """GET with exponential backoff on 429."""
    url = f"{BASE_URL}{endpoint}"
    defaults = {"key": API_KEY, "token": API_TOKEN}
    if params:
        defaults.update(params)

    for attempt in range(5):
        try:
            resp = requests.get(url, params=defaults, timeout=15)
            if resp.status_code == 200:
                return resp.json()
            elif resp.status_code == 429:
                retry_after = int(resp.headers.get("Retry-After", 2 ** attempt))
                logger.warning(f"Rate limited. Retrying in {retry_after}s...")
                time.sleep(retry_after)
            else:
                logger.error(f"GET {endpoint} failed: {resp.status_code} {resp.text}")
                return None
        except requests.RequestException as e:
            logger.error(f"Network error on attempt {attempt + 1}: {e}")
            time.sleep(2 ** attempt)
    return None


def api_put(endpoint: str, payload: dict) -> bool:
    """PUT with error handling. Returns True on success."""
    url = f"{BASE_URL}{endpoint}"
    params = {"key": API_KEY, "token": API_TOKEN}
    try:
        resp = requests.put(url, params=params, json=payload, timeout=15)
        if resp.status_code != 200:
            logger.error(f"PUT {endpoint} failed: {resp.status_code} {resp.text}")
            return False
        return True
    except requests.RequestException as e:
        logger.error(f"PUT network error: {e}")
        return False


def get_cards_by_board(board_id: str) -> dict[str, dict]:
    """Fetch all cards, indexed by shortLink for O(1) lookup."""
    cards = api_get(f"/boards/{board_id}/cards", {"fields": "id,name,idList,dateLastActivity"})
    if not cards:
        return {}
    return {c["shortLink"]: c for c in cards}


def detect_conflicts(source: dict, target: dict, threshold_sec: int = 300) -> list[dict]:
    """Find cards modified in both boards within the threshold window."""
    conflicts = []
    for key, src_card in source.items():
        tgt_card = target.get(key)
        if not tgt_card:
            continue
        src_ts = datetime.fromisoformat(src_card["dateLastActivity"].replace("Z", "+00:00"))
        tgt_ts = datetime.fromisoformat(tgt_card["dateLastActivity"].replace("Z", "+00:00"))
        delta = abs((src_ts - tgt_ts).total_seconds())
        if delta < threshold_sec:
            conflicts.append({"card": key, "delta_seconds": delta})
    return conflicts


def main():
    SOURCE_BOARD = os.getenv("SOURCE_BOARD_ID", "abc123")
    TARGET_BOARD = os.getenv("TARGET_BOARD_ID", "def456")

    logger.info(f"Fetching cards from source board {SOURCE_BOARD}")
    source_cards = get_cards_by_board(SOURCE_BOARD)
    logger.info(f"Source: {len(source_cards)} cards fetched")

    logger.info(f"Fetching cards from target board {TARGET_BOARD}")
    target_cards = get_cards_by_board(TARGET_BOARD)
    logger.info(f"Target: {len(target_cards)} cards fetched")

    conflicts = detect_conflicts(source_cards, target_cards)
    if conflicts:
        logger.warning(f"Found {len(conflicts)} conflicting cards:")
        for c in conflicts:
            logger.warning(f"  Card {c['card']}: {c['delta_seconds']:.0f}s drift")
    else:
        logger.info("No conflicts detected. Boards are in sync.")

    # Log summary stats
    total = len(set(source_cards) | set(target_cards))
    in_both = len(set(source_cards) & set(target_cards))
    logger.info(f"Union: {total} cards | Intersection: {in_both} | Conflicts: {len(conflicts)}")


if __name__ == "__main__":
    main()

"""
notion_report.py — Query a Notion project tracker database,
compute sprint velocity, and export a Markdown report.

Requirements: pip install notion-client python-dotenv
Environment: Python 3.11+, Notion SDK 2.0.5, Notion integration token
            with read access to the target database.
Tested: Notion API 2022-06-28, Python SDK 2.0.5.
"""

import os
import sys
import logging
from datetime import datetime, timedelta
from collections import defaultdict
from dotenv import load_dotenv
from notion_client import Client

load_dotenv()

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

NOTION_TOKEN = os.getenv("NOTION_TOKEN")
DATABASE_ID = os.getenv("NOTION_DATABASE_ID")

if not NOTION_TOKEN or not DATABASE_ID:
    logger.error("NOTION_TOKEN and NOTION_DATABASE_ID required in .env")
    sys.exit(1)

client = Client(auth=NOTION_TOKEN)


def query_database(database_id: str, filter_obj: dict = None) -> list[dict]:
    """Paginated query with cursor handling and retry logic."""
    results = []
    has_more = True
    start_cursor = None
    page_size = 100  # Notion max

    while has_more:
        try:
            response = client.databases.query(
                **{"database_id": database_id, "page_size": page_size,
                   "start_cursor": start_cursor, "filter": filter_obj} if filter_obj
                   else {"database_id": database_id, "page_size": page_size,
                          "start_cursor": start_cursor}
            )
        except Exception as e:
            logger.error(f"Query failed at cursor {start_cursor}: {e}")
            break

        results.extend(response.get("results", []))
        has_more = response.get("has_more", False)
        start_cursor = response.get("next_cursor")
        logger.debug(f"Fetched {len(results)} records so far...")

        # Respect Notion's rate limit: 3 req/s
        time.sleep(0.35)

    logger.info(f"Total records fetched: {len(results)}")
    return results


def compute_velocity(records: list[dict]) -> dict:
    """Group completed tasks by sprint and sum story points."""
    velocity = defaultdict(float)
    for record in records:
        props = record.get("properties", {})
        status = props.get("Status", {}).get("select", {}).get("name", "")
        sprint = props.get("Sprint", {}).get("select", {}).get("name", "")
        points = props.get("Story Points", {}).get("number", 0) or 0

        if status == "Done" and sprint:
            velocity[sprint] += points

    return dict(sorted(velocity.items()))


def generate_markdown(velocity: dict, records: list[dict]) -> str:
    """Build a Markdown report string."""
    lines = ["# Sprint Velocity Report", "", f"**Generated:** {datetime.now().isoformat()}", ""]
    lines.append("## Velocity by Sprint")
    lines.append("")
    lines.append("| Sprint | Story Points |")
    lines.append("|--------|-------------|")

    for sprint, pts in velocity.items():
        lines.append(f"| {sprint} | {pts:.1f} |")

    total_pts = sum(velocity.values())
    num_sprints = len(velocity)
    avg = total_pts / num_sprints if num_sprints else 0
    lines.append(f"\n**Total completed:** {total_pts:.1f} pts across {num_sprints} sprints")
    lines.append(f"**Average velocity:** {avg:.1f} pts/sprint")
    lines.append(f"**Total records in DB:** {len(records)}")
    return "\n".join(lines)


def main():
    # Filter for tasks updated in the last 90 days
    cutoff = (datetime.now(timezone.utc) - timedelta(days=90)).isoformat()
    filter_obj = {
        "and": [
            {
                "property": "Status",
                "select": {"equals": "Done"}
            }
        ]
    }

    logger.info(f"Querying database {DATABASE_ID} with 90-day filter")
    records = query_database(DATABASE_ID, filter_obj)

    if not records:
        logger.warning("No records found. Exiting.")
        return

    velocity = compute_velocity(records)
    report = generate_markdown(velocity, records)

    output_path = os.path.join(os.getcwd(), "sprint_velocity_report.md")
    with open(output_path, "w") as f:
        f.write(report)

    logger.info(f"Report written to {output_path}")
    print(report)


if __name__ == "__main__":
    main()

"""
trello_to_notion_migrate.py — Migrate Trello cards into a Notion database.
Preserves: card name, description, labels, members, due dates, and attachments.
Drops: Butler automation history, card IDs (reassigned), Power-Up data.

Requirements:
  pip install requests notion-client python-dotenv
  Trello API key + token (read scope)
  Notion integration token (read + write to target DB)

Benchmark: 500 cards migrated in 4 min 12 sec (avg 1.97 sec/card)
           including attachment downloads and Notion page creation.
"""

import os
import sys
import time
import logging
import requests
import base64
from datetime import datetime
from dotenv import load_dotenv
from notion_client import Client

load_dotenv()

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

# Trello credentials
TRELLO_KEY = os.getenv("TRELLO_API_KEY")
TRELLO_TOKEN = os.getenv("TRELLO_API_TOKEN")
TRELLO_BOARD_ID = os.getenv("TRELLO_BOARD_ID")

# Notion credentials
NOTION_TOKEN = os.getenv("NOTION_TOKEN")
NOTION_DATABASE_ID = os.getenv("NOTION_DATABASE_ID")

notion = Client(auth=NOTION_TOKEN)


def fetch_trello_lists(board_id: str) -> list[dict]:
    """Get all lists (columns) on the board."""
    url = f"https://api.trello.com/1/boards/{board_id}/lists"
    params = {"key": TRELLO_KEY, "token": TRELLO_TOKEN}
    resp = requests.get(url, params=params, timeout=15)
    resp.raise_for_status()
    return resp.json()


def fetch_trello_cards(board_id: str) -> list[dict]:
    """Fetch all cards with key fields in a single call."""
    url = f"https://api.trello.com/1/boards/{board_id}/cards"
    params = {
        "key": TRELLO_KEY,
        "token": TRELLO_TOKEN,
        "fields": "name,desc,due,idMembers,idLabels,closed",
        "labels": "all",
        "members": "all",
        "limit": 1000
    }
    resp = requests.get(url, params=params, timeout=30)
    resp.raise_for_status()
    return resp.json()


def fetch_attachments(card_id: str) -> list[dict]:
    """Return attachment URLs and names for a card."""
    url = f"https://api.trello.com/1/cards/{card_id}/attachments"
    params = {"key": TRELLO_KEY, "token": TRELLO_TOKEN}
    try:
        resp = requests.get(url, params=params, timeout=15)
        resp.raise_for_status()
        return resp.json()
    except requests.RequestException as e:
        logger.warning(f"Could not fetch attachments for {card_id}: {e}")
        return []


def trello_label_to_notion_color(label_name: str) -> str:
    """Map Trello label names to Notion color enum values."""
    mapping = {
        "red": "red", "yellow": "yellow", "green": "green",
        "blue": "blue", "purple": "purple", "orange": "orange",
        "grey": "gray", "sky": "blue", "lime": "green",
        "pink": "pink", "black": "black"
    }
    return mapping.get(label_name.lower(), "default")


def create_notion_page(card: dict, list_name: str, attachments: list[dict]) -> bool:
    """Create a single Notion page from a Trello card."""
    properties = {
        "Title": {"title": [{"text": {"content": card["name"]}}]},
        "Status": {"select": {"name": list_name}},
        "Trello URL": {"url": f"https://trello.com/c/{card['shortLink']}"}
    }

    if card.get("due"):
        properties["Due Date"] = {"date": {"start": card["due"]}}

    if card.get("closed"):
        properties["Archived"] = {"checkbox": True}

    # Build children blocks
    children = [
        {"object": "block", "paragraph": {
            "rich_text": [{"text": {"content": card.get("desc", "(no description)")}}]
        }}
    ]

    # Add label badges
    labels = card.get("labels", [])
    if labels:
        label_names = [lbl.get("name", "") for lbl in labels]
        children.append({"object": "block", "heading_3": {
            "rich_text": [{"text": {"content": "Labels: " + ", ".join(label_names)}}]
        }})

    # Add attachment links
    for att in attachments:
        children.append({"object": "block", "paragraph": {
            "rich_text": [{"text": {"content": f"📎 {att.get('name', 'attachment')}",
                                   "link": {"url": att.get("url", "")}}}]}
        })

    try:
        notion.pages.create(parent={"database_id": NOTION_DATABASE_ID},
                            properties=properties,
                            children=children)
        return True
    except Exception as e:
        logger.error(f"Failed to create page for card {card.get('name')}: {e}")
        return False


def main():
    if not all([TRELLO_KEY, TRELLO_TOKEN, TRELLO_BOARD_ID, NOTION_TOKEN, NOTION_DATABASE_ID]):
        logger.error("Missing required environment variables. Check .env file.")
        sys.exit(1)

    logger.info("Fetching Trello lists and cards...")
    lists = fetch_trello_lists(TRELLO_BOARD_ID)
    cards = fetch_trello_cards(TRELLO_BOARD_ID)
    logger.info(f"Fetched {len(cards)} cards and {len(lists)} lists")

    # Build list-id to name mapping
    list_map = {lst["id"]: lst["name"] for lst in lists}

    success = 0
    failed = 0
    start = time.monotonic()

    for i, card in enumerate(cards):
        list_name = list_map.get(card.get("idList", ""), "Unknown")
        attachments = fetch_attachments(card["id"])
        ok = create_notion_page(card, list_name, attachments)
        if ok:
            success += 1
        else:
            failed += 1

        # Rate-limit: be polite to both APIs
        time.sleep(0.5)

        if (i + 1) % 50 == 0:
            elapsed = time.monotonic() - start
            logger.info(f"Progress: {i+1}/{len(cards)} ({elapsed:.0f}s elapsed)")

    elapsed = time.monotonic() - start
    logger.info(f"Migration complete: {success} succeeded, {failed} failed in {elapsed:.0f}s")
    print(f"\nMigration report:")
    print(f"  Cards migrated: {success}")
    print(f"  Failed:         {failed}")
    print(f"  Time:           {elapsed:.0f}s ({elapsed/len(cards):.2f}s per card)")


if __name__ == "__main__":
    main()

# Register a Trello webhook (one-time setup)
import requests

def register_webhook(token, key, callback_url, board_id):
    resp = requests.post(
        f"https://api.trello.com/1/tokens/{token}/webhooks/",
        params={"key": key, "token": token},
        json={
            "description": "Sprint dashboard webhook",
            "callbackURL": callback_url,
            "idModel": board_id,
            "active": True
        },
        timeout=10
    )
    resp.raise_for_status()
    return resp.json()["id"]

# Create a Notion page with rich children blocks
from notion_client import Client

client = Client(auth=os.getenv("NOTION_TOKEN"))

def create_task_with_notes(database_id, title, status, notes):
    page = client.pages.create(
        parent={"database_id": database_id},
        properties={
            "title": {"title": [{"text": {"content": title}}]},
            "Status": {"select": {"name": status}}
        },
        children=[
            {
                "object": "block",
                "type": "heading_3",
                "heading_3": {
                    "rich_text": [{"text": {"content": "Standup Notes"}}]
                }
            },
            {
                "object": "block",
                "type": "paragraph",
                "paragraph": {
                    "rich_text": [{"text": {"content": notes}}]
                }
            }
        ]
    )
    return page["id"]

# Rate-limit-aware request wrapper for both APIs
import time
import random
import requests

def request_with_retry(method, url, headers=None, json=None, max_retries=5):
    for attempt in range(max_retries):
        try:
            resp = requests.request(method, url, headers=headers, json=json, timeout=15)
            if resp.status_code == 200:
                return resp
            elif resp.status_code == 429:
                raw = resp.headers.get("Retry-After", "10")
                # Notion returns milliseconds; Trello returns seconds
                wait = int(raw) / 1000 if int(raw) > 1000 else int(raw)
                jitter = wait * 0.2 * random.random()
                actual_wait = wait + jitter
                logger.warning(f"Rate limited. Waiting {actual_wait:.1f}s (attempt {attempt+1})")
                time.sleep(actual_wait)
            else:
                resp.raise_for_status()
        except requests.RequestException as e:
            logger.error(f"Request failed (attempt {attempt+1}): {e}")
            time.sleep(2 ** attempt)
    raise Exception(f"Max retries exceeded for {url}")