#!/usr/bin/env python3 """ 日程管家 · 本地服务器 ==================== 提供 Dashboard 的完整交互功能：读取、添加、修改任务。 """ import json import sys import os from http.server import HTTPServer, SimpleHTTPRequestHandler from urllib.parse import urlparse, parse_qs from pathlib import Path BASE_DIR = Path(__file__).parent DATA_DIR = BASE_DIR / "data" class ScheduleAPI(SimpleHTTPRequestHandler): """API + 静态文件混合服务器""" def __init__(self, *args, **kwargs): super().__init__(*args, directory=str(BASE_DIR), **kwargs) def do_GET(self): parsed = urlparse(self.path) if parsed.path == "/api/tasks": self._serve_json("tasks.json") return elif parsed.path == "/api/projects": self._serve_json("projects.json") return elif parsed.path == "/api/plan": params = parse_qs(parsed.query) date_str = params.get("date", [None])[0] self._run_scheduler("plan", date_str) return elif parsed.path == "/api/week": self._run_scheduler("week") return elif parsed.path == "/api/conflicts": self._run_scheduler("conflicts") return # Default: serve static files super().do_GET() def do_POST(self): parsed = urlparse(self.path) if parsed.path == "/api/tasks/add": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) task_data = json.loads(body) result = self._add_task(task_data) self._json_response(result, 201) return elif parsed.path == "/api/tasks/update": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body) task_id = data.pop("id", None) result = self._update_task(task_id, data) self._json_response(result) return elif parsed.path == "/api/conflicts/resolve": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) choices = json.loads(body) result = self._resolve_conflicts(choices) self._json_response(result) return self.send_error(404) def _json_response(self, data, status=200): self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(json.dumps(data, ensure_ascii=False).encode("utf-8")) def _serve_json(self, filename): path = DATA_DIR / filename if not path.exists(): self.send_error(404, f"File not found: {filename}") return with open(path, "r", encoding="utf-8") as f: data = json.load(f) self._json_response(data) def _run_scheduler(self, cmd, arg=None): import subprocess python = sys.executable script = BASE_DIR / "scheduler.py" args = [python, str(script), cmd] if arg: args.append(arg) result = subprocess.run(args, capture_output=True, text=True, cwd=str(BASE_DIR)) if result.returncode == 0: self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(result.stdout.encode("utf-8")) else: self.send_error(500, result.stderr) def _add_task(self, task_data): path = DATA_DIR / "tasks.json" with open(path, "r", encoding="utf-8") as f: data = json.load(f) existing_ids = [int(t["id"].replace("t-", "")) for t in data["tasks"] if t["id"].startswith("t-")] new_id = f"t-{max(existing_ids or [0]) + 1:03d}" from datetime import datetime now = datetime.now().isoformat() task = { "id": new_id, "category": task_data.get("category", "other_work"), "project_id": task_data.get("project_id", ""), "title": task_data.get("title", ""), "description": task_data.get("description", ""), "estimated_minutes": task_data.get("estimated_minutes", 60), "priority": task_data.get("priority", 3), "deadline": task_data.get("deadline"), "status": "pending", "scheduled_date": task_data.get("scheduled_date", ""), "scheduled_time": task_data.get("scheduled_time", "09:00"), "tags": task_data.get("tags", []), "source": task_data.get("source", "manual"), "source_raw": task_data.get("source_raw"), "source_task_id": task_data.get("source_task_id"), "created_at": now, "updated_at": now } data["tasks"].append(task) with open(path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) return task def _update_task(self, task_id, updates): path = DATA_DIR / "tasks.json" with open(path, "r", encoding="utf-8") as f: data = json.load(f) from datetime import datetime for t in data["tasks"]: if t["id"] == task_id: t.update(updates) t["updated_at"] = datetime.now().isoformat() with open(path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) return t return {"error": f"Task {task_id} not found"} def _resolve_conflicts(self, choices): path = DATA_DIR / "tasks.json" import subprocess python = sys.executable script = BASE_DIR / "scheduler.py" args = [python, str(script), "resolve", json.dumps(choices)] result = subprocess.run(args, capture_output=True, text=True, cwd=str(BASE_DIR)) if result.returncode == 0: return json.loads(result.stdout) return {"error": result.stderr} def do_OPTIONS(self): self.send_response(204) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() def log_message(self, format, *args): # 简洁日志 sys.stderr.write(f"[日程管家] {args[0]}\n") if __name__ == "__main__": host = os.environ.get("SCHEDULE_HOST", "0.0.0.0") port = int(os.environ.get("SCHEDULE_PORT", "8765")) print(f"日程管家服务器启动: http://{host}:{port}") print(f"Dashboard: http://{host}:{port}/dashboard.html") HTTPServer((host, port), ScheduleAPI).serve_forever()