feat: Add api

.gitignore

@@ -168,3 +168,4 @@ cython_debug/
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
 
+tasks.db

app.py

@@ -0,0 +1,176 @@
+"""
+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 value is None:
+        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)