""" 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/", 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/", 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//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)