537 lines
15 KiB
Python
537 lines
15 KiB
Python
"""
|
||
Calink API - Generate reusable calendar invitation links (.ics)
|
||
|
||
Requirements:
|
||
pip install fastapi uvicorn pydantic aiosqlite
|
||
|
||
Example usage:
|
||
# Run the server
|
||
python main.py
|
||
"""
|
||
|
||
from datetime import datetime, timedelta
|
||
from enum import Enum
|
||
from typing import Optional, List
|
||
import os
|
||
import json
|
||
import uuid
|
||
import sqlite3
|
||
from contextlib import asynccontextmanager
|
||
from fastapi import FastAPI, HTTPException, Query
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from fastapi.responses import Response
|
||
from pydantic import BaseModel, Field, field_validator
|
||
from zoneinfo import ZoneInfo
|
||
|
||
|
||
# Configuration
|
||
DATABASE_PATH = os.getenv("DATABASE_PATH", "./data/app.db")
|
||
|
||
|
||
# Database initialization
|
||
def init_db():
|
||
"""Initialize SQLite database with required tables."""
|
||
os.makedirs(os.path.dirname(DATABASE_PATH) if os.path.dirname(DATABASE_PATH) else "./data", exist_ok=True)
|
||
|
||
conn = sqlite3.connect(DATABASE_PATH)
|
||
cursor = conn.cursor()
|
||
|
||
# Events history table
|
||
cursor.execute("""
|
||
CREATE TABLE IF NOT EXISTS events_history (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
created_at TEXT NOT NULL,
|
||
title TEXT NOT NULL,
|
||
description TEXT,
|
||
location TEXT,
|
||
start_dt TEXT NOT NULL,
|
||
end_dt TEXT,
|
||
timezone TEXT NOT NULL,
|
||
reminders_json TEXT,
|
||
last_downloaded_at TEXT
|
||
)
|
||
""")
|
||
|
||
# Reminder templates table
|
||
cursor.execute("""
|
||
CREATE TABLE IF NOT EXISTS reminder_templates (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
name TEXT UNIQUE NOT NULL,
|
||
reminders_json TEXT NOT NULL,
|
||
created_at TEXT NOT NULL
|
||
)
|
||
""")
|
||
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
"""Application lifespan handler."""
|
||
init_db()
|
||
yield
|
||
|
||
|
||
# FastAPI app
|
||
app = FastAPI(
|
||
title="Calink API",
|
||
description="Calink API – Generate reusable calendar invitation links (.ics)",
|
||
version="1.0.0",
|
||
lifespan=lifespan
|
||
)
|
||
|
||
# Configure CORS
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["http://localhost:5173", "http://127.0.0.1:5173", "*"],
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
)
|
||
|
||
|
||
# Models
|
||
class ReminderUnit(str, Enum):
|
||
"""Supported reminder time units."""
|
||
minutes = "minutes"
|
||
hours = "hours"
|
||
days = "days"
|
||
weeks = "weeks"
|
||
|
||
|
||
class Reminder(BaseModel):
|
||
"""Reminder configuration for an event."""
|
||
amount: int = Field(..., gt=0, description="Amount of time before event")
|
||
unit: ReminderUnit = Field(..., description="Time unit for reminder")
|
||
|
||
|
||
class EventRequest(BaseModel):
|
||
"""Request model for creating an event."""
|
||
title: str = Field(..., min_length=1, description="Event title/summary")
|
||
description: Optional[str] = Field(None, description="Event description")
|
||
location: Optional[str] = Field(None, description="Event location")
|
||
start_dt: str = Field(..., description="Event start datetime (ISO 8601 string)")
|
||
end_dt: Optional[str] = Field(None, description="Event end datetime (ISO 8601 string)")
|
||
timezone: str = Field("UTC", description="Timezone for naive datetimes")
|
||
reminders: Optional[List[Reminder]] = Field(None, max_length=10, description="Event reminders (max 10)")
|
||
|
||
@field_validator('reminders')
|
||
@classmethod
|
||
def validate_reminders(cls, v):
|
||
"""Validate reminders list."""
|
||
if v is not None and len(v) > 10:
|
||
raise ValueError("Maximum 10 reminders allowed")
|
||
return v
|
||
|
||
|
||
class ReminderTemplate(BaseModel):
|
||
"""Reminder template model."""
|
||
name: str = Field(..., min_length=1, max_length=100)
|
||
reminders: List[Reminder] = Field(..., max_length=10)
|
||
|
||
|
||
class HistoryEvent(BaseModel):
|
||
"""History event response model."""
|
||
id: int
|
||
created_at: str
|
||
title: str
|
||
description: Optional[str]
|
||
location: Optional[str]
|
||
start_dt: str
|
||
end_dt: Optional[str]
|
||
timezone: str
|
||
reminders: Optional[List[Reminder]]
|
||
last_downloaded_at: Optional[str]
|
||
|
||
|
||
class TemplateResponse(BaseModel):
|
||
"""Template response model."""
|
||
id: int
|
||
name: str
|
||
reminders: List[Reminder]
|
||
created_at: str
|
||
|
||
|
||
# Helper functions
|
||
def get_db():
|
||
"""Get database connection."""
|
||
conn = sqlite3.connect(DATABASE_PATH)
|
||
conn.row_factory = sqlite3.Row
|
||
return conn
|
||
|
||
|
||
def parse_datetime(dt_str: str, tz: str = "UTC") -> datetime:
|
||
"""Parse ISO datetime string and apply timezone if naive."""
|
||
try:
|
||
dt = datetime.fromisoformat(dt_str.replace('Z', '+00:00'))
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=ZoneInfo(tz))
|
||
return dt
|
||
except Exception as e:
|
||
raise ValueError(f"Invalid datetime format: {dt_str}. Expected ISO 8601 format.")
|
||
|
||
|
||
def format_datetime_for_ics(dt: datetime) -> str:
|
||
"""Format datetime for iCalendar format (UTC with Z suffix)."""
|
||
utc_dt = dt.astimezone(ZoneInfo("UTC"))
|
||
return utc_dt.strftime("%Y%m%dT%H%M%SZ")
|
||
|
||
|
||
def reminder_to_trigger(reminder: Reminder) -> str:
|
||
"""Convert reminder to iCalendar TRIGGER format."""
|
||
amount = reminder.amount
|
||
unit = reminder.unit
|
||
|
||
if unit == ReminderUnit.minutes:
|
||
return f"-PT{amount}M"
|
||
elif unit == ReminderUnit.hours:
|
||
return f"-PT{amount}H"
|
||
elif unit == ReminderUnit.days:
|
||
return f"-P{amount}D"
|
||
elif unit == ReminderUnit.weeks:
|
||
return f"-P{amount}W"
|
||
|
||
return "-PT15M"
|
||
|
||
|
||
def generate_ics(event: EventRequest) -> str:
|
||
"""Generate iCalendar .ics file content from event parameters."""
|
||
# Parse datetimes
|
||
start_dt = parse_datetime(event.start_dt, event.timezone)
|
||
|
||
# Calculate end_dt if not provided (default 60 minutes)
|
||
if event.end_dt:
|
||
end_dt = parse_datetime(event.end_dt, event.timezone)
|
||
else:
|
||
end_dt = start_dt + timedelta(minutes=60)
|
||
|
||
# Validate end_dt is after start_dt
|
||
if end_dt <= start_dt:
|
||
raise ValueError("end_dt must be after start_dt")
|
||
|
||
# Generate unique ID and timestamp
|
||
uid = str(uuid.uuid4())
|
||
dtstamp = datetime.now(ZoneInfo("UTC"))
|
||
|
||
# Start building ICS content
|
||
lines = [
|
||
"BEGIN:VCALENDAR",
|
||
"VERSION:2.0",
|
||
"PRODID:-//Calink//EN",
|
||
"CALSCALE:GREGORIAN",
|
||
"METHOD:PUBLISH",
|
||
"BEGIN:VEVENT",
|
||
f"UID:{uid}",
|
||
f"DTSTAMP:{format_datetime_for_ics(dtstamp)}",
|
||
f"DTSTART:{format_datetime_for_ics(start_dt)}",
|
||
f"DTEND:{format_datetime_for_ics(end_dt)}",
|
||
f"SUMMARY:{event.title}",
|
||
]
|
||
|
||
# Add optional DESCRIPTION
|
||
if event.description:
|
||
desc = event.description.replace("\\", "\\\\").replace(",", "\\,").replace(";", "\\;").replace("\n", "\\n")
|
||
lines.append(f"DESCRIPTION:{desc}")
|
||
|
||
# Add optional LOCATION
|
||
if event.location:
|
||
loc = event.location.replace("\\", "\\\\").replace(",", "\\,").replace(";", "\\;")
|
||
lines.append(f"LOCATION:{loc}")
|
||
|
||
# Add VALARM blocks for reminders
|
||
if event.reminders:
|
||
for reminder in event.reminders:
|
||
trigger = reminder_to_trigger(reminder)
|
||
lines.extend([
|
||
"BEGIN:VALARM",
|
||
"ACTION:DISPLAY",
|
||
"DESCRIPTION:Reminder",
|
||
f"TRIGGER:{trigger}",
|
||
"END:VALARM"
|
||
])
|
||
|
||
# Close VEVENT and VCALENDAR
|
||
lines.extend([
|
||
"END:VEVENT",
|
||
"END:VCALENDAR"
|
||
])
|
||
|
||
# Join with CRLF and ensure trailing CRLF
|
||
ics_content = "\r\n".join(lines) + "\r\n"
|
||
|
||
return ics_content
|
||
|
||
|
||
def save_to_history(event: EventRequest, is_download: bool = False):
|
||
"""Save event to history."""
|
||
conn = get_db()
|
||
cursor = conn.cursor()
|
||
|
||
reminders_json = json.dumps([r.dict() for r in event.reminders]) if event.reminders else None
|
||
now = datetime.now(ZoneInfo("UTC")).isoformat()
|
||
|
||
cursor.execute("""
|
||
INSERT INTO events_history
|
||
(created_at, title, description, location, start_dt, end_dt, timezone, reminders_json, last_downloaded_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""", (
|
||
now,
|
||
event.title,
|
||
event.description,
|
||
event.location,
|
||
event.start_dt,
|
||
event.end_dt,
|
||
event.timezone,
|
||
reminders_json,
|
||
now if is_download else None
|
||
))
|
||
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
# API Endpoints
|
||
@app.get("/health")
|
||
async def health():
|
||
"""Health check endpoint."""
|
||
return {"ok": True}
|
||
|
||
|
||
@app.get("/")
|
||
async def root():
|
||
"""Root endpoint with API information."""
|
||
return {
|
||
"app": "Calink",
|
||
"description": "Generate reusable calendar invitation links (.ics)",
|
||
"docs": "/docs",
|
||
"health": "/health"
|
||
}
|
||
|
||
|
||
@app.post("/api/ics")
|
||
async def create_ics(event: EventRequest):
|
||
"""Generate and download an iCalendar .ics file."""
|
||
try:
|
||
ics_content = generate_ics(event)
|
||
save_to_history(event, is_download=True)
|
||
|
||
return Response(
|
||
content=ics_content,
|
||
media_type="text/calendar; charset=utf-8",
|
||
headers={
|
||
"Content-Disposition": 'attachment; filename="event.ics"'
|
||
}
|
||
)
|
||
except ValueError as e:
|
||
raise HTTPException(status_code=400, detail=str(e))
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"Error generating ICS file: {str(e)}")
|
||
|
||
|
||
@app.post("/api/preview")
|
||
async def preview_ics(event: EventRequest):
|
||
"""Generate and return ICS content as plain text for preview."""
|
||
try:
|
||
ics_content = generate_ics(event)
|
||
save_to_history(event, is_download=False)
|
||
|
||
return Response(
|
||
content=ics_content,
|
||
media_type="text/plain; charset=utf-8"
|
||
)
|
||
except ValueError as e:
|
||
raise HTTPException(status_code=400, detail=str(e))
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"Error generating ICS file: {str(e)}")
|
||
|
||
|
||
@app.get("/api/history")
|
||
async def get_history(limit: int = Query(20, ge=1, le=100)):
|
||
"""Get events history."""
|
||
conn = get_db()
|
||
cursor = conn.cursor()
|
||
|
||
cursor.execute("""
|
||
SELECT * FROM events_history
|
||
ORDER BY created_at DESC
|
||
LIMIT ?
|
||
""", (limit,))
|
||
|
||
rows = cursor.fetchall()
|
||
conn.close()
|
||
|
||
events = []
|
||
for row in rows:
|
||
reminders = None
|
||
if row['reminders_json']:
|
||
reminders_data = json.loads(row['reminders_json'])
|
||
reminders = [Reminder(**r) for r in reminders_data]
|
||
|
||
events.append(HistoryEvent(
|
||
id=row['id'],
|
||
created_at=row['created_at'],
|
||
title=row['title'],
|
||
description=row['description'],
|
||
location=row['location'],
|
||
start_dt=row['start_dt'],
|
||
end_dt=row['end_dt'],
|
||
timezone=row['timezone'],
|
||
reminders=reminders,
|
||
last_downloaded_at=row['last_downloaded_at']
|
||
))
|
||
|
||
return events
|
||
|
||
|
||
@app.get("/api/history/{event_id}")
|
||
async def get_history_event(event_id: int):
|
||
"""Get a specific event from history."""
|
||
conn = get_db()
|
||
cursor = conn.cursor()
|
||
|
||
cursor.execute("SELECT * FROM events_history WHERE id = ?", (event_id,))
|
||
row = cursor.fetchone()
|
||
conn.close()
|
||
|
||
if not row:
|
||
raise HTTPException(status_code=404, detail="Event not found")
|
||
|
||
reminders = None
|
||
if row['reminders_json']:
|
||
reminders_data = json.loads(row['reminders_json'])
|
||
reminders = [Reminder(**r) for r in reminders_data]
|
||
|
||
return HistoryEvent(
|
||
id=row['id'],
|
||
created_at=row['created_at'],
|
||
title=row['title'],
|
||
description=row['description'],
|
||
location=row['location'],
|
||
start_dt=row['start_dt'],
|
||
end_dt=row['end_dt'],
|
||
timezone=row['timezone'],
|
||
reminders=reminders,
|
||
last_downloaded_at=row['last_downloaded_at']
|
||
)
|
||
|
||
|
||
@app.delete("/api/history/{event_id}")
|
||
async def delete_history_event(event_id: int):
|
||
"""Delete an event from history."""
|
||
conn = get_db()
|
||
cursor = conn.cursor()
|
||
|
||
cursor.execute("DELETE FROM events_history WHERE id = ?", (event_id,))
|
||
|
||
if cursor.rowcount == 0:
|
||
conn.close()
|
||
raise HTTPException(status_code=404, detail="Event not found")
|
||
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
return {"message": "Event deleted successfully"}
|
||
|
||
|
||
@app.post("/api/templates")
|
||
async def create_template(template: ReminderTemplate):
|
||
"""Create a new reminder template."""
|
||
conn = get_db()
|
||
cursor = conn.cursor()
|
||
|
||
reminders_json = json.dumps([r.dict() for r in template.reminders])
|
||
now = datetime.now(ZoneInfo("UTC")).isoformat()
|
||
|
||
try:
|
||
cursor.execute("""
|
||
INSERT INTO reminder_templates (name, reminders_json, created_at)
|
||
VALUES (?, ?, ?)
|
||
""", (template.name, reminders_json, now))
|
||
|
||
template_id = cursor.lastrowid
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
return {
|
||
"id": template_id,
|
||
"name": template.name,
|
||
"reminders": template.reminders,
|
||
"created_at": now
|
||
}
|
||
except sqlite3.IntegrityError:
|
||
conn.close()
|
||
raise HTTPException(status_code=400, detail="Template with this name already exists")
|
||
|
||
|
||
@app.get("/api/templates")
|
||
async def get_templates():
|
||
"""Get all reminder templates."""
|
||
conn = get_db()
|
||
cursor = conn.cursor()
|
||
|
||
cursor.execute("SELECT * FROM reminder_templates ORDER BY created_at DESC")
|
||
rows = cursor.fetchall()
|
||
conn.close()
|
||
|
||
templates = []
|
||
for row in rows:
|
||
reminders_data = json.loads(row['reminders_json'])
|
||
reminders = [Reminder(**r) for r in reminders_data]
|
||
|
||
templates.append(TemplateResponse(
|
||
id=row['id'],
|
||
name=row['name'],
|
||
reminders=reminders,
|
||
created_at=row['created_at']
|
||
))
|
||
|
||
return templates
|
||
|
||
|
||
@app.get("/api/templates/{template_id}")
|
||
async def get_template(template_id: int):
|
||
"""Get a specific template."""
|
||
conn = get_db()
|
||
cursor = conn.cursor()
|
||
|
||
cursor.execute("SELECT * FROM reminder_templates WHERE id = ?", (template_id,))
|
||
row = cursor.fetchone()
|
||
conn.close()
|
||
|
||
if not row:
|
||
raise HTTPException(status_code=404, detail="Template not found")
|
||
|
||
reminders_data = json.loads(row['reminders_json'])
|
||
reminders = [Reminder(**r) for r in reminders_data]
|
||
|
||
return TemplateResponse(
|
||
id=row['id'],
|
||
name=row['name'],
|
||
reminders=reminders,
|
||
created_at=row['created_at']
|
||
)
|
||
|
||
|
||
@app.delete("/api/templates/{template_id}")
|
||
async def delete_template(template_id: int):
|
||
"""Delete a template."""
|
||
conn = get_db()
|
||
cursor = conn.cursor()
|
||
|
||
cursor.execute("DELETE FROM reminder_templates WHERE id = ?", (template_id,))
|
||
|
||
if cursor.rowcount == 0:
|
||
conn.close()
|
||
raise HTTPException(status_code=404, detail="Template not found")
|
||
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
return {"message": "Template deleted successfully"}
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import uvicorn
|
||
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)
|