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, render_template
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
# ---------------------------------------------------------------------------
try:
    creds = service_account.Credentials.from_service_account_file(
        CREDENTIALS_FILE, scopes=SCOPES
    )
    calendar_service = build("calendar", "v3", credentials=creds, cache_discovery=False)
except FileNotFoundError:
    print(f"Warning: Google credentials file '{CREDENTIALS_FILE}' not found. Calendar integration disabled.")
    calendar_service = None

# ---------------------------------------------------------------------------
# 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("/")
def index():
    return render_template("index.html", api_key=API_KEY)
@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),
            "min_interval_days": t.min_interval_days,
            "last_execution_time": _to_datetime(t.last_execution_time).isoformat() if t.last_execution_time else None,
            "priority": t.priority,
        }
        for t in tasks
    ]
    return jsonify(response)

@app.route("/tasks", methods=["POST"])
@require_api_key
def create_task():
    data = request.json
    if not data or "name" not in data:
        abort(400, description="Task name is required")
    
    task = Task.create(
        name=data["name"],
        min_interval_days=data.get("min_interval_days", 7),
        priority=data.get("priority", 0)
    )
    
    return jsonify({
        "id": task.id,
        "name": task.name,
        "min_interval_days": task.min_interval_days,
        "priority": task.priority,
    }), 201

@app.route("/tasks/<int:task_id>", methods=["PUT"])
@require_api_key
def update_task(task_id: int):
    task = Task.get_or_none(Task.id == task_id)
    if not task:
        abort(404, description="Task not found")
    
    data = request.json
    if "name" in data:
        task.name = data["name"]
    if "min_interval_days" in data:
        task.min_interval_days = data["min_interval_days"]
    if "priority" in data:
        task.priority = data["priority"]
    
    task.save()
    
    return jsonify({
        "id": task.id,
        "name": task.name,
        "min_interval_days": task.min_interval_days,
        "priority": task.priority,
    })

@app.route("/tasks/<int:task_id>", methods=["DELETE"])
@require_api_key
def delete_task(task_id: int):
    task = Task.get_or_none(Task.id == task_id)
    if not task:
        abort(404, description="Task not found")
    
    task.delete_instance()
    return jsonify({"success": True})

@app.route("/tasks/reorder", methods=["POST"])
@require_api_key
def reorder_tasks():
    data = request.json
    if not data or "task_ids" not in data:
        abort(400, description="task_ids array is required")
    
    task_ids = data["task_ids"]
    priority = len(task_ids)
    
    for task_id in task_ids:
        task = Task.get_or_none(Task.id == task_id)
        if task:
            task.priority = priority
            task.save()
            priority -= 1
    
    return jsonify({"success": True})

@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")
    
    if calendar_service is None:
        abort(503, description="Calendar service not available")

    # 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)