calendar-widget/app.py

"""
Flask application providing task management and Google Calendar scheduling.
-------------------------------------------------------
Prerequisites:
• Python 3.9+
• pip install flask peewee google-api-python-client google-auth google-auth-httplib2 google-auth-oauthlib

Configuration:
  – Set environment variable API_KEY with the shared secret that clients must send in the
    HTTP header `X-API-Key`.
  – Place your Google Service‑Account JSON credentials next to this file and set
    GOOGLE_CREDENTIALS_FILE (default 'credentials.json').
  – Set CALENDAR_ID to your target calendar (e.g. 'primary' or a specific ID).

Run with:
    python app.py
"""

import os
import datetime as dt
from functools import wraps
from zoneinfo import ZoneInfo  # Python 3.9+ standard timezone database

from flask import Flask, jsonify, request, abort
from peewee import (
    Model, SqliteDatabase, AutoField, CharField, IntegerField, DateTimeField, 
)

from google.oauth2 import service_account
from googleapiclient.discovery import build

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
DATABASE_FILE = os.getenv("TASK_DB", "tasks.db")
API_KEY = os.getenv("API_KEY", "change-me")
CREDENTIALS_FILE = os.getenv("GOOGLE_CREDENTIALS_FILE", "credentials.json")
CALENDAR_ID = os.getenv("CALENDAR_ID", "primary")

SCOPES = ["https://www.googleapis.com/auth/calendar"]
TZ = ZoneInfo("Asia/Shanghai")  # 使用北京时间


# ---------------------------------------------------------------------------
# Database setup
# ---------------------------------------------------------------------------
db = SqliteDatabase(DATABASE_FILE, check_same_thread=False)

class BaseModel(Model):
    class Meta:
        database = db

class Task(BaseModel):
    id = AutoField()
    name = CharField()
    min_interval_days = IntegerField()          # "最少 x 天执行一次"
    last_schedule_id = CharField(null=True)     # 上次日程 ID
    last_execution_time = DateTimeField(null=True)
    priority = IntegerField(default=0)          # 优先级 (越大越高)

# Create table if it doesn't exist
with db:
    db.create_tables([Task])

# ---------------------------------------------------------------------------
# Google Calendar client
# ---------------------------------------------------------------------------
creds = service_account.Credentials.from_service_account_file(
    CREDENTIALS_FILE, scopes=SCOPES
)
calendar_service = build("calendar", "v3", credentials=creds, cache_discovery=False)

# ---------------------------------------------------------------------------
# Flask app + helpers
# ---------------------------------------------------------------------------
app = Flask(__name__)

def require_api_key(func):
    """Simple header‑based authentication decorator."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        if request.headers.get("X-API-Key") != API_KEY:
            abort(401, description="Invalid API key")
        return func(*args, **kwargs)
    return wrapper

# ---------------------------------------------------------------------------
# Utilities
# ---------------------------------------------------------------------------

def _to_datetime(value):
    """Ensure value is a timezone‑aware datetime."""
    if not value:
        return None
    if isinstance(value, dt.datetime):
        return value if value.tzinfo else value.replace(tzinfo=TZ)
    # Peewee on SQLite may return str – parse ISO 8601
    parsed = dt.datetime.fromisoformat(value)
    return parsed if parsed.tzinfo else parsed.replace(tzinfo=TZ)


def compute_color(task: Task, now: dt.datetime) -> str:
    """Compute traffic‑light color for a task based on last_execution_time."""
    last_dt = _to_datetime(task.last_execution_time)
    if last_dt is None:
        return "green"

    days_passed = (now - last_dt).total_seconds() / 86400
    threshold = task.min_interval_days or 0

    if days_passed <= (2 * threshold) / 3:
        return "green"
    elif days_passed <= threshold:
        return "yellow"
    return "red"

# ---------------------------------------------------------------------------
# Routes
# ---------------------------------------------------------------------------
@app.route("/tasks", methods=["GET"])
@require_api_key
def list_tasks():
    now = dt.datetime.now(TZ)
    tasks = (
        Task.select().order_by(Task.priority.desc())
    )
    response = [
        {
            "id": t.id,
            "name": t.name,
            "color": compute_color(t, now),
        }
        for t in tasks
    ]
    return jsonify(response)

@app.route("/tasks/<int:task_id>/schedule", methods=["POST"])
@require_api_key
def schedule_task(task_id: int):
    task = Task.get_or_none(Task.id == task_id)
    if not task:
        abort(404, description="Task not found")

    # Event time is now (Asia/Shanghai)
    start_dt = dt.datetime.now(TZ).replace(microsecond=0)
    start_time = start_dt.isoformat()

    event_body = {
        "summary": task.name,
        "description": task.name,
        "start": {"dateTime": start_time, "timeZone": "Asia/Shanghai"},
        "end": {"dateTime": start_time, "timeZone": "Asia/Shanghai"},
    }

    event = (
        calendar_service.events()
        .insert(calendarId=CALENDAR_ID, body=event_body)
        .execute()
    )

    # Update task with new schedule info
    task.last_schedule_id = event["id"]
    task.last_execution_time = start_dt
    task.save()

    return jsonify({
        "success": True,
        "event_id": event["id"],
        "task_id": task.id,
    }), 201

# ---------------------------------------------------------------------------
# Application entry point
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=int(os.getenv("PORT", 5000)), debug=True)