#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Personal Life OS - 本地服务 - 托管 life-dashboard.html 等静态文件 - 提供 /sync (POST) 接收来自微信读书的书签按钮同步数据 - 提供 /sync.json (GET) 供工作台读取最新同步结果 """ import http.server import socketserver import copy import html import json import math import os import re import ssl import subprocess import sys import time import shutil import datetime import threading import urllib.parse import urllib.request from urllib.parse import parse_qs, urlparse WORKSPACE = os.path.dirname(os.path.abspath(__file__)) SYNC_PATH = os.path.join(WORKSPACE, "sync.json") KEY_PATH = os.path.join(WORKSPACE, ".weread", "apikey.txt") GW_URL = "https://i.weread.qq.com/api/agent/gateway" LAST_NOTE_PATH = os.path.join(WORKSPACE, ".weread", "tmp", "last_reading_note.json") READING_AUTO_SCRIPT = os.path.join(WORKSPACE, ".weread", "tmp", "run_reading_auto.py") READING_NOTES_DIR = os.path.join(WORKSPACE, "读书收获") READING_BOOKS_DIR = os.path.join(READING_NOTES_DIR, "书籍") WEREAD_SHELF_CACHE_PATH = os.path.join(WORKSPACE, ".weread", "tmp", "shelf.json") HOT_VIDEOS_PATH = os.path.join(WORKSPACE, "learning-hot-videos.json") HOT_VIDEO_AUTO_SCRIPT = os.path.join(WORKSPACE, "hot_video_auto.py") AI_LEARNING_VIDEOS_PATH = os.path.join(WORKSPACE, "ai-learning-videos.json") AI_LEARNING_VIDEO_AUTO_SCRIPT = os.path.join(WORKSPACE, "ai_learning_video_auto.py") LIFE_DATA_PATH = os.path.join(WORKSPACE, "life-data.json") FAVORITE_COLLECTIONS_PATH = os.path.join(WORKSPACE, "favorite-collections.json") FAVORITE_MEDIA_DIR = os.path.join(WORKSPACE, ".favorites", "media") FAVORITE_SYNC_SCRIPT = os.path.join(WORKSPACE, "sync-favorites.js") NODE_EXE = os.path.join(os.path.expanduser("~"), ".cache", "codex-runtimes", "codex-primary-runtime", "dependencies", "node", "bin", "node.exe") NODE_MODULES = os.path.join(os.path.expanduser("~"), ".cache", "codex-runtimes", "codex-primary-runtime", "dependencies", "node", "node_modules") FAVORITE_LOG_DIR = os.path.join(WORKSPACE, ".favorites") DAILY_ENGLISH_EVENTS_DB = os.path.join(os.path.expanduser("~"), "AppData", "Roaming", "ting_en", "events.db") DAILY_ENGLISH_CONFIG_JSON = os.path.join(os.path.expanduser("~"), "AppData", "Roaming", "ting_en", "config.json") CLOUD_SERVER_ORIGIN = os.environ.get("WORKBUDDY_CLOUD_ORIGIN", "https://xm.cloudmall.fun").rstrip("/") READING_SYNC_TOKEN = os.environ.get("WORKBUDDY_READING_SYNC_TOKEN", "workbuddy-reading-sync-20260910") PORT = 8765 HOT_VIDEO_AUTO_HOUR = 7 READING_AUTO_HOUR = 8 ALWAYS_DAILY_PRESET_IDS = {"exercise", "read", "english", "ailearn"} SUNDAY_PRESET_ID = "readingNote" DAILY_PRESET_IDS = ALWAYS_DAILY_PRESET_IDS | {SUNDAY_PRESET_ID} PRESET_DISPLAY_NAMES = { "exercise": "💃", "read": "📖", "english": "🎧", "ailearn": "Ai", "readingNote": "📖🪙", } # Simple in-memory cache for readtimes _readtimes_cache = {"ts": 0, "data": None} _READTIMES_TTL = 60 # seconds _reading_auto_lock = threading.Lock() _hot_video_auto_lock = threading.Lock() _ai_learning_video_auto_lock = threading.Lock() _reading_cloud_sync_lock = threading.Lock() _english_cloud_sync_lock = threading.Lock() READING_SYNC_PREFIXES = ( "读书收获/书籍/", "读书收获/reading-insights.json", ".weread/tmp/shelf.json", ".weread/tmp/last_reading_note.json", "reading-note-status.json", ) def _local_today_str(): tz8 = datetime.timezone(datetime.timedelta(hours=8)) return datetime.datetime.now(tz8).strftime("%Y-%m-%d") def _last_reading_note_date(): if not os.path.exists(LAST_NOTE_PATH): return "" try: with open(LAST_NOTE_PATH, "r", encoding="utf-8") as f: data = json.load(f) return str(data.get("date") or "") except Exception: return "" def _last_hot_video_date(): if not os.path.exists(HOT_VIDEOS_PATH): return "" try: with open(HOT_VIDEOS_PATH, "r", encoding="utf-8") as f: data = json.load(f) updated_at = str(data.get("updatedAt") or "") videos = data.get("videos") if isinstance(data, dict) else [] if len(videos or []) < 3: return "" return updated_at[:10] except Exception: return "" def _is_allowed_reading_sync_path(rel_path): rel_path = (rel_path or "").replace("\\", "/").lstrip("/") return any(rel_path == prefix.rstrip("/") or rel_path.startswith(prefix) for prefix in READING_SYNC_PREFIXES) def _collect_reading_sync_files(): files = [] candidates = [ os.path.join(WORKSPACE, "reading-note-status.json"), os.path.join(READING_NOTES_DIR, "reading-insights.json"), os.path.join(WORKSPACE, ".weread", "tmp", "shelf.json"), os.path.join(WORKSPACE, ".weread", "tmp", "last_reading_note.json"), ] if os.path.exists(READING_BOOKS_DIR): for root, _, names in os.walk(READING_BOOKS_DIR): for name in names: if name not in ("书籍档案.md", "metadata.json"): continue candidates.append(os.path.join(root, name)) seen = set() for path in candidates: if not path or not os.path.exists(path) or not os.path.isfile(path): continue rel_path = os.path.relpath(path, WORKSPACE).replace("\\", "/") if rel_path in seen or not _is_allowed_reading_sync_path(rel_path): continue seen.add(rel_path) try: with open(path, "r", encoding="utf-8") as f: content = f.read() files.append({ "path": rel_path, "content": content, "mtime": int(os.path.getmtime(path)), }) except Exception: continue return files def sync_reading_artifacts_to_cloud(timeout=20): if os.name != "nt": return {"ok": True, "skipped": True, "reason": "cloud sync runs from the Windows desktop service"} if not _reading_cloud_sync_lock.acquire(blocking=False): return {"ok": False, "error": "读书档案云同步正在运行"} try: files = _collect_reading_sync_files() if not files: return {"ok": True, "skipped": True, "files": 0} payload = json.dumps({ "token": READING_SYNC_TOKEN, "updatedAt": int(time.time()), "files": files, }, ensure_ascii=False).encode("utf-8") req = urllib.request.Request( f"{CLOUD_SERVER_ORIGIN}/api/reading-sync/import", data=payload, headers={"Content-Type": "application/json; charset=utf-8"}, method="POST", ) with urllib.request.urlopen(req, timeout=timeout) as resp: result = json.loads(resp.read().decode("utf-8", "ignore")) return result except Exception as e: return {"ok": False, "error": str(e)} finally: _reading_cloud_sync_lock.release() def read_daily_english_listening_stats(): if not os.path.exists(DAILY_ENGLISH_EVENTS_DB): fallback = read_daily_english_stats_from_life_data() if fallback.get("ok"): fallback["warning"] = "当前服务无法读取每日英语听力 Windows 客户端日志，已使用云端工作台同步记录。" return fallback return { "ok": False, "error": "未找到每日英语听力 Windows 客户端日志 events.db", "sourcePath": DAILY_ENGLISH_EVENTS_DB, } checkin_info = {} plan_duration = 0 if os.path.exists(DAILY_ENGLISH_CONFIG_JSON): try: with open(DAILY_ENGLISH_CONFIG_JSON, "r", encoding="utf-8") as f: cfg = json.load(f) raw_checkin = cfg.get("Key_User_CheckIn_Info") if isinstance(raw_checkin, str) and raw_checkin.strip(): checkin_info = json.loads(raw_checkin) plan_duration = int(float(cfg.get("planDuration") or 0)) except Exception: checkin_info = {} plan_duration = 0 try: import sqlite3 con = sqlite3.connect(f"file:{DAILY_ENGLISH_EVENTS_DB}?mode=ro", uri=True) con.row_factory = sqlite3.Row cur = con.cursor() rows = cur.execute( """ select epoch, count(*) as count, sum(coalesce(duration, 0)) as listenSeconds, sum(coalesce(read_duration, 0)) as readSeconds, min(timestamp) as firstTimestamp, max(timestamp) as lastTimestamp from Logs where type = 'LISTEN_DURATION' group by epoch order by epoch desc """ ).fetchall() con.close() except Exception as e: return {"ok": False, "error": str(e), "sourcePath": DAILY_ENGLISH_EVENTS_DB} read_times = {} listen_times = {} records = [] for row in rows: raw_epoch = str(row["epoch"] or "") if not re.fullmatch(r"\d{8}", raw_epoch): continue date_key = f"{raw_epoch[:4]}-{raw_epoch[4:6]}-{raw_epoch[6:8]}" listen_seconds = int(row["listenSeconds"] or 0) read_seconds = int(row["readSeconds"] or 0) study_seconds = max(0, listen_seconds + read_seconds) listen_times[date_key] = listen_seconds read_times[date_key] = study_seconds records.append({ "date": date_key, "listenSeconds": listen_seconds, "listenMinutes": int(round(listen_seconds / 60)), "readSeconds": read_seconds, "studySeconds": study_seconds, "studyMinutes": int(round(study_seconds / 60)), "count": int(row["count"] or 0), "firstTimestamp": int(row["firstTimestamp"] or 0), "lastTimestamp": int(row["lastTimestamp"] or 0), }) fallback_date = str(checkin_info.get("checkinDateLocal") or checkin_info.get("checkinDate") or "") fallback_seconds = max(0, plan_duration * 60) if re.fullmatch(r"\d{4}-\d{2}-\d{2}", fallback_date) and fallback_seconds > int(read_times.get(fallback_date) or 0): read_times[fallback_date] = fallback_seconds listen_times[fallback_date] = max(int(listen_times.get(fallback_date) or 0), fallback_seconds) if not any(item.get("date") == fallback_date for item in records): records.insert(0, { "date": fallback_date, "listenSeconds": fallback_seconds, "listenMinutes": int(round(fallback_seconds / 60)), "readSeconds": 0, "studySeconds": fallback_seconds, "studyMinutes": int(round(fallback_seconds / 60)), "count": 0, "firstTimestamp": 0, "lastTimestamp": 0, "sourceDetail": "checkin-planDuration", }) life_fallback = read_daily_english_stats_from_life_data() if life_fallback.get("ok"): existing_records = {item.get("date"): item for item in records if isinstance(item, dict)} for date_key, seconds in (life_fallback.get("readTimes") or {}).items(): try: seconds = int(seconds or 0) except Exception: seconds = 0 if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", str(date_key or "")) or seconds <= int(read_times.get(date_key) or 0): continue minutes = int(round(seconds / 60)) read_times[date_key] = seconds listen_times[date_key] = max(int(listen_times.get(date_key) or 0), seconds) existing = existing_records.get(date_key) if existing: existing.update({ "listenSeconds": max(int(existing.get("listenSeconds") or 0), seconds), "listenMinutes": minutes, "studySeconds": seconds, "studyMinutes": minutes, "sourceDetail": "life-data-englishMinutes-merged", }) else: records.append({ "date": date_key, "listenSeconds": seconds, "listenMinutes": minutes, "readSeconds": 0, "studySeconds": seconds, "studyMinutes": minutes, "count": 1, "firstTimestamp": 0, "lastTimestamp": 0, "sourceDetail": "life-data-englishMinutes-merged", }) records.sort(key=lambda item: item.get("date") or "", reverse=True) return { "ok": True, "source": "每日英语听力 Windows 客户端 events.db", "sourcePath": DAILY_ENGLISH_EVENTS_DB, "configPath": DAILY_ENGLISH_CONFIG_JSON, "checkinInfo": checkin_info, "planDurationMinutes": plan_duration, "readTimes": read_times, "listenTimes": listen_times, "records": records, } def _load_life_data_file(): if not os.path.exists(LIFE_DATA_PATH): return {} try: with open(LIFE_DATA_PATH, "r", encoding="utf-8") as f: data = json.load(f) return data if isinstance(data, dict) else {} except Exception: return {} def read_daily_english_stats_from_life_data(): data = _load_life_data_file() daily = data.get("dailyPlan") if isinstance(data, dict) else {} if not isinstance(daily, dict): daily = {} read_times = {} records = [] for date_key, plan in daily.items(): if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", str(date_key or "")): continue presets = plan.get("presets") if isinstance(plan, dict) else [] if not isinstance(presets, list): continue task = next((item for item in presets if isinstance(item, dict) and item.get("id") == "english"), None) minutes = int(round(float(task.get("englishMinutes") or 0))) if isinstance(task, dict) else 0 if minutes <= 0: continue seconds = minutes * 60 read_times[date_key] = seconds records.append({ "date": date_key, "listenSeconds": seconds, "listenMinutes": minutes, "readSeconds": 0, "studySeconds": seconds, "studyMinutes": minutes, "count": 1, "firstTimestamp": int(task.get("completedAt") or 0) if isinstance(task, dict) else 0, "lastTimestamp": int(task.get("completedAt") or 0) if isinstance(task, dict) else 0, "sourceDetail": "life-data-englishMinutes", }) records.sort(key=lambda item: item.get("date") or "", reverse=True) return { "ok": bool(read_times), "source": "云端工作台 life-data 英语打卡记录", "sourcePath": LIFE_DATA_PATH, "planDurationMinutes": 5, "readTimes": read_times, "listenTimes": read_times, "records": records, } def _ensure_life_plan_for_date(data, date_key): daily = data.setdefault("dailyPlan", {}) plan = daily.setdefault(date_key, {"presets": [], "customs": []}) presets = plan.setdefault("presets", []) plan.setdefault("customs", []) for preset_id in ("exercise", "read", "english", "ailearn"): if not any(isinstance(item, dict) and item.get("id") == preset_id for item in presets): presets.append({ "id": preset_id, "name": PRESET_DISPLAY_NAMES.get(preset_id, preset_id), "checked": False, "type": "preset", }) return plan def _upsert_english_session(data, date_key, minutes, timestamp=0): english = data.setdefault("englishLearn", {}) sessions = english.setdefault("sessions", []) if not isinstance(sessions, list): sessions = [] english["sessions"] = sessions session_id = f"daily_english_listening_{date_key}" existing = next((item for item in sessions if isinstance(item, dict) and item.get("id") == session_id), None) payload = { "id": session_id, "date": date_key, "skill": "每日英语听力", "minutes": int(minutes), "detail": "每日英语听力 Windows 客户端同步", "key": "daily-english-listening", "time": time.strftime("%H:%M", time.localtime((timestamp or time.time() * 1000) / 1000)), "timestamp": int(timestamp or time.time() * 1000), } if existing: if int(minutes) >= int(existing.get("minutes") or 0): existing.update(payload) else: sessions.append(payload) def sync_daily_english_listening_to_life_data(timeout=20): if os.name != "nt": return {"ok": True, "skipped": True, "reason": "每日英语听力同步只在 Windows 桌面服务读取本机客户端日志"} if not _english_cloud_sync_lock.acquire(blocking=False): return {"ok": False, "error": "每日英语听力云同步正在运行"} try: stats = read_daily_english_listening_stats() if not stats.get("ok"): return stats data = _load_life_data_file() changed = False cutoff = datetime.datetime.now().date() - datetime.timedelta(days=30) for record in stats.get("records") or []: date_key = str(record.get("date") or "") if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", date_key): continue try: if datetime.datetime.strptime(date_key, "%Y-%m-%d").date() < cutoff: continue except ValueError: continue minutes = int(round((record.get("studySeconds") or 0) / 60)) if minutes <= 0: continue plan = _ensure_life_plan_for_date(data, date_key) task = next((item for item in plan.get("presets", []) if isinstance(item, dict) and item.get("id") == "english"), None) if not task: continue previous = int(task.get("englishMinutes") or 0) if minutes > previous: task["englishMinutes"] = minutes task["autoCheckin"] = "dailyEnglishListening" task["thresholdMinutes"] = 5 task["checked"] = minutes >= 5 or bool(task.get("checked")) if task["checked"] and not task.get("completedAt"): task["completedAt"] = int(record.get("lastTimestamp") or time.time() * 1000) task["statusText"] = f"每日英语听力 {minutes} 分钟，已自动打卡" if task["checked"] else f"每日英语听力 {minutes} 分钟，满 5 分钟自动打卡" _upsert_english_session(data, date_key, minutes, record.get("lastTimestamp") or 0) changed = True if not changed: return {"ok": True, "changed": False, "stats": stats} data["__syncUpdatedAt"] = int(time.time() * 1000) data["__syncSavedAt"] = time.strftime("%Y-%m-%d %H:%M:%S") with open(LIFE_DATA_PATH, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) payload = json.dumps({"data": data}, ensure_ascii=False).encode("utf-8") req = urllib.request.Request( f"{CLOUD_SERVER_ORIGIN}/api/life-data", data=payload, headers={"Content-Type": "application/json; charset=utf-8"}, method="POST", ) with urllib.request.urlopen(req, timeout=timeout) as resp: cloud = json.loads(resp.read().decode("utf-8", "ignore")) return {"ok": True, "changed": True, "cloud": cloud, "stats": stats} except Exception as e: return {"ok": False, "error": str(e)} finally: _english_cloud_sync_lock.release() def start_english_listening_sync_scheduler(): if os.name != "nt": return def loop(): time.sleep(3) while True: sync_daily_english_listening_to_life_data(timeout=20) time.sleep(90) threading.Thread(target=loop, daemon=True, name="english-listening-cloud-sync").start() def run_hot_video_auto_job(timeout=180): if not os.path.exists(HOT_VIDEO_AUTO_SCRIPT): return {"ok": False, "error": "hot_video_auto.py not found"} if not _hot_video_auto_lock.acquire(blocking=False): return {"ok": False, "error": "热门视频抓取正在运行，请稍后再试。"} try: proc = subprocess.run( [sys.executable, HOT_VIDEO_AUTO_SCRIPT], cwd=WORKSPACE, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout, ) stdout = (proc.stdout or "").strip() stderr = (proc.stderr or "").strip() result = json.loads(stdout.splitlines()[-1]) if stdout else {"ok": proc.returncode == 0} if proc.returncode != 0: result["ok"] = False result["error"] = stderr or stdout or f"exit code {proc.returncode}" result["stdout"] = stdout if stderr: result["stderr"] = stderr return result except subprocess.TimeoutExpired: return {"ok": False, "error": "热门视频抓取超时，请稍后重试。"} except Exception as e: return {"ok": False, "error": str(e)} finally: _hot_video_auto_lock.release() def run_ai_learning_video_job(timeout=180, completed_count=None, refresh_offset=0): if not os.path.exists(AI_LEARNING_VIDEO_AUTO_SCRIPT): return {"ok": False, "error": "ai_learning_video_auto.py not found"} if not _ai_learning_video_auto_lock.acquire(blocking=False): return {"ok": False, "error": "AI学习视频推荐正在生成中，请稍后再试。"} try: cmd = [sys.executable, AI_LEARNING_VIDEO_AUTO_SCRIPT] if completed_count is not None: cmd.append(f"--completed-count={int(completed_count)}") if refresh_offset: cmd.append(f"--refresh-offset={max(0, int(refresh_offset))}") proc = subprocess.run( cmd, cwd=WORKSPACE, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout, ) stdout = (proc.stdout or "").strip() stderr = (proc.stderr or "").strip() result = json.loads(stdout.splitlines()[-1]) if stdout else {"ok": proc.returncode == 0} if proc.returncode != 0: result["ok"] = False result["error"] = stderr or stdout or f"exit code {proc.returncode}" result["stdout"] = stdout if stderr: result["stderr"] = stderr return result except subprocess.TimeoutExpired: return {"ok": False, "error": "AI学习视频推荐生成超时，请稍后重试。"} except Exception as e: return {"ok": False, "error": str(e)} finally: _ai_learning_video_auto_lock.release() def run_reading_auto_job(timeout=180): if not os.path.exists(READING_AUTO_SCRIPT): return { "ok": False, "error": "reading note generator script not found", "script": READING_AUTO_SCRIPT, } if not _reading_auto_lock.acquire(blocking=False): return {"ok": False, "error": "读书收获正在生成中，请稍后查看。"} try: proc = subprocess.run( [sys.executable, READING_AUTO_SCRIPT], cwd=WORKSPACE, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout, ) if proc.returncode != 0: return { "ok": False, "error": "生成脚本执行失败", "stdout": proc.stdout, "stderr": proc.stderr, "returncode": proc.returncode, } global _readtimes_cache _readtimes_cache = {"ts": 0, "data": None} status_data = {"ok": True} if os.path.exists(LAST_NOTE_PATH): try: with open(LAST_NOTE_PATH, "r", encoding="utf-8") as f: status_data = json.load(f) except Exception as e: status_data = {"ok": True, "warning": f"笔记已生成，但状态文件读取失败：{e}"} status_data["ok"] = True status_data["stdout"] = proc.stdout.strip() status_data["cloudSync"] = sync_reading_artifacts_to_cloud(timeout=20) return status_data except subprocess.TimeoutExpired: return {"ok": False, "error": "生成超时，请稍后重试。"} except Exception as e: return {"ok": False, "error": str(e)} finally: _reading_auto_lock.release() def start_reading_auto_scheduler(): def loop(): while True: try: tz8 = datetime.timezone(datetime.timedelta(hours=8)) now = datetime.datetime.now(tz8) if now.hour >= READING_AUTO_HOUR and _last_reading_note_date() != now.strftime("%Y-%m-%d"): result = run_reading_auto_job(timeout=180) log_dir = os.path.join(WORKSPACE, ".weread", "tmp") os.makedirs(log_dir, exist_ok=True) with open(os.path.join(log_dir, "reading-auto.log"), "a", encoding="utf-8") as f: f.write(f"[{now.strftime('%Y-%m-%d %H:%M:%S')}] scheduled 08:00 result={json.dumps(result, ensure_ascii=False)}\n") except Exception as e: try: log_dir = os.path.join(WORKSPACE, ".weread", "tmp") os.makedirs(log_dir, exist_ok=True) with open(os.path.join(log_dir, "reading-auto.log"), "a", encoding="utf-8") as f: f.write(f"[scheduler-error] {e}\n") except Exception: pass time.sleep(60) threading.Thread(target=loop, daemon=True, name="reading-auto-08").start() def start_hot_video_auto_scheduler(): def loop(): while True: try: tz8 = datetime.timezone(datetime.timedelta(hours=8)) now = datetime.datetime.now(tz8) if now.hour >= HOT_VIDEO_AUTO_HOUR and _last_hot_video_date() != now.strftime("%Y-%m-%d"): result = run_hot_video_auto_job(timeout=180) log_dir = os.path.join(WORKSPACE, ".workbuddy") os.makedirs(log_dir, exist_ok=True) with open(os.path.join(log_dir, "hot-video-auto.log"), "a", encoding="utf-8") as f: f.write(f"[{now.strftime('%Y-%m-%d %H:%M:%S')}] scheduled 07:00 result={json.dumps(result, ensure_ascii=False)}\n") except Exception as e: try: log_dir = os.path.join(WORKSPACE, ".workbuddy") os.makedirs(log_dir, exist_ok=True) with open(os.path.join(log_dir, "hot-video-auto.log"), "a", encoding="utf-8") as f: f.write(f"[scheduler-error] {e}\n") except Exception: pass time.sleep(60) threading.Thread(target=loop, daemon=True, name="hot-video-auto-07").start() class Handler(http.server.SimpleHTTPRequestHandler): def guess_type(self, path): content_type = super().guess_type(path) if content_type.startswith("text/") and "charset=" not in content_type: return f"{content_type}; charset=utf-8" return content_type def _cors(self): existing = b"".join(getattr(self, "_headers_buffer", [])).lower() if b"access-control-allow-origin:" not in existing: self.send_header("Access-Control-Allow-Origin", "*") if b"access-control-allow-methods:" not in existing: self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") if b"access-control-allow-headers:" not in existing: self.send_header("Access-Control-Allow-Headers", "Content-Type") def do_OPTIONS(self): self.send_response(200) self._cors() self.end_headers() def _read_key(self): if not os.path.exists(KEY_PATH): return None try: with open(KEY_PATH, "r", encoding="utf-8") as f: return f.read().strip() except Exception: return None def _gw_post(self, api_name, payload, timeout=10): key = self._read_key() if not key: raise RuntimeError("WEREAD_API_KEY not configured") body = json.dumps({"skill_version": "1.0.4", "api_name": api_name, **payload}, ensure_ascii=False).encode("utf-8") req = urllib.request.Request( GW_URL, data=body, headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"}, method="POST", ) ctx = ssl.create_default_context() with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp: return json.loads(resp.read().decode("utf-8")) def do_POST(self): if self.path.startswith("/api/generate-reading-note"): self._generate_reading_note() elif self.path.startswith("/api/delete-reading-note"): self._delete_reading_note() elif self.path.startswith("/api/reading-sync/import"): self._import_reading_sync() elif self.path.startswith("/api/hot-videos/run"): self._run_hot_videos() elif self.path.startswith("/api/ai-learning-videos/run"): self._run_ai_learning_videos() elif self.path.startswith("/api/viral-video/analyze"): self._analyze_viral_video() elif self.path.startswith("/api/life-data"): self._save_life_data() elif self.path.startswith("/api/favorite-collections/import"): self._import_favorite_collections() elif self.path.startswith("/api/favorite-collections/login"): self._login_favorite_collections() elif self.path.startswith("/api/favorite-collections/sync"): self._sync_favorite_collections() elif self.path.startswith("/api/favorite-collections/analyze"): self._analyze_favorite_collection_item() elif self.path.startswith("/sync"): try: length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length) if length else b"{}" data = json.loads(body.decode("utf-8")) except Exception as e: self.send_response(400) self._cors() self.end_headers() self.wfile.write(json.dumps({"ok": False, "error": str(e)}).encode("utf-8")) return books = data.get("books", []) if isinstance(data, dict) else [] snapshot = {"updated": data.get("updated", "") if isinstance(data, dict) else "", "books": books} try: with open(SYNC_PATH, "w", encoding="utf-8") as f: json.dump(snapshot, f, ensure_ascii=False, indent=2) except Exception as e: self.send_response(500) self._cors() self.end_headers() self.wfile.write(json.dumps({"ok": False, "error": str(e)}).encode("utf-8")) return self.send_response(200) self._cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps({"ok": True, "count": len(books)}).encode("utf-8")) else: self.send_response(404) self._cors() self.end_headers() def _generate_reading_note(self): status_data = run_reading_auto_job(timeout=180) self._json_response(status_data, status=200 if status_data.get("ok") else 500) def _delete_reading_note(self): try: payload = self._read_json_body() date_str = str(payload.get("date") or "") except Exception as e: self._json_response({"ok": False, "error": f"请求格式错误：{e}"}, status=400) return note_path = self._reading_note_file_for_date(date_str) if not note_path: self._json_response({"ok": False, "error": "日期格式错误"}, status=400) return deleted = False if os.path.exists(note_path): os.remove(note_path) deleted = True if _last_reading_note_date() == date_str: for status_path in ( LAST_NOTE_PATH, os.path.join(WORKSPACE, "reading-note-status.json"), ): try: if os.path.exists(status_path): os.remove(status_path) except Exception: pass self._json_response({"ok": True, "deleted": deleted, "date": date_str}) def _import_reading_sync(self): try: payload = self._read_json_body() if not isinstance(payload, dict) or payload.get("token") != READING_SYNC_TOKEN: self._json_response({"ok": False, "error": "unauthorized"}, status=403) return files = payload.get("files") if not isinstance(files, list): self._json_response({"ok": False, "error": "files must be a list"}, status=400) return written = 0 skipped = 0 for item in files: if not isinstance(item, dict): skipped += 1 continue rel_path = str(item.get("path") or "").replace("\\", "/").lstrip("/") content = item.get("content") if not rel_path or not _is_allowed_reading_sync_path(rel_path) or not isinstance(content, str): skipped += 1 continue target = os.path.abspath(os.path.join(WORKSPACE, *rel_path.split("/"))) if not target.startswith(os.path.abspath(WORKSPACE) + os.sep): skipped += 1 continue os.makedirs(os.path.dirname(target), exist_ok=True) with open(target, "w", encoding="utf-8") as f: f.write(content) mtime = item.get("mtime") try: if mtime: os.utime(target, (int(mtime), int(mtime))) except Exception: pass written += 1 self._json_response({"ok": True, "written": written, "skipped": skipped}) except Exception as e: self._json_response({"ok": False, "error": str(e)}, status=500) def _run_hot_videos(self): result = run_hot_video_auto_job(timeout=180) self._json_response(result, status=200 if result.get("ok") else 500) def _run_ai_learning_videos(self): completed_count = None refresh_offset = 0 try: payload = self._read_json_body() if isinstance(payload, dict) and payload.get("completedCount") is not None: completed_count = int(payload.get("completedCount")) if isinstance(payload, dict) and payload.get("refreshOffset") is not None: refresh_offset = max(0, int(payload.get("refreshOffset"))) except Exception: completed_count = None refresh_offset = 0 result = run_ai_learning_video_job(timeout=180, completed_count=completed_count, refresh_offset=refresh_offset) self._json_response(result, status=200 if result.get("ok") else 500) def _extract_first_url(self, text): match = re.search(r"https?://[^\s，。)）]+", text or "", re.I) return match.group(0) if match else "" def _load_hot_videos(self): if not os.path.exists(HOT_VIDEOS_PATH): return [] try: with open(HOT_VIDEOS_PATH, "r", encoding="utf-8") as f: data = json.load(f) return data.get("videos", []) if isinstance(data, dict) else data if isinstance(data, list) else [] except Exception: return [] def _normalize_video_url(self, url): url = str(url or "").strip().lower() url = re.sub(r"^https?://", "", url) url = re.sub(r"[?#].*$", "", url) return url.rstrip("/") def _match_hot_video(self, url): target = self._normalize_video_url(url) if not target: return None for video in self._load_hot_videos(): current = self._normalize_video_url(video.get("url") or video.get("link") or "") if current and (current == target or current in target or target in current): return video return None def _fetch_json_url(self, url, referer="https://www.bilibili.com/", timeout=12): req = urllib.request.Request( url, headers={ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36", "Referer": referer, "Accept": "application/json, text/plain, */*", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", }, ) with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read().decode("utf-8", "ignore")) def _fetch_bilibili_video_meta(self, url): text = str(url or "") bvid_match = re.search(r"(BV[0-9A-Za-z]+)", text) aid_match = re.search(r"(?:/video/|av)(\d+)", text) if bvid_match: api = "https://api.bilibili.com/x/web-interface/view?" + urllib.parse.urlencode({"bvid": bvid_match.group(1)}) elif aid_match: api = "https://api.bilibili.com/x/web-interface/view?" + urllib.parse.urlencode({"aid": aid_match.group(1)}) else: raise RuntimeError("没有识别到 B站 BV 号或 av 号") data = self._fetch_json_url(api) if data.get("code") != 0: raise RuntimeError(data.get("message") or "B站接口未返回视频信息") info = data.get("data") or {} stat = info.get("stat") or {} owner = info.get("owner") or {} fans = None if owner.get("mid"): try: rel = self._fetch_json_url( "https://api.bilibili.com/x/relation/stat?" + urllib.parse.urlencode({"vmid": owner.get("mid")}) ) fans = (rel.get("data") or {}).get("follower") except Exception: fans = None tags = [] try: tag_data = self._fetch_json_url( "https://api.bilibili.com/x/tag/archive/tags?" + urllib.parse.urlencode({"bvid": info.get("bvid")}) ) tags = [item.get("tag_name") for item in (tag_data.get("data") or []) if item.get("tag_name")] except Exception: tags = [] plays = stat.get("view") ratio = round(plays / fans, 2) if isinstance(plays, int) and isinstance(fans, int) and fans > 0 else None return { "platform": "B站", "title": info.get("title") or "", "url": f"https://www.bilibili.com/video/{info.get('bvid')}/" if info.get("bvid") else url, "plays": plays, "playCount": plays, "likes": stat.get("like"), "likeCount": stat.get("like"), "fans": fans, "followerCount": fans, "playFanRatio": ratio, "tags": tags[:8] or ["学习视频"], "author": owner.get("name") or "", "duration": info.get("duration"), "description": info.get("desc") or "", "analysisBasis": "B站公开接口：标题、作者、播放、点赞、粉丝、标签", } def _openai_api_key(self): key = os.environ.get("OPENAI_API_KEY", "").strip() if key: return key for path in ( os.path.join(WORKSPACE, ".openai", "api_key.txt"), os.path.join(os.path.expanduser("~"), ".openai", "api_key.txt"), ): try: if os.path.exists(path): with open(path, "r", encoding="utf-8") as f: key = f.read().strip() if key: return key except Exception: pass return "" def _analyze_viral_images(self, images, raw_text="", timeout=45): valid = [] for item in images if isinstance(images, list) else []: if not isinstance(item, dict): continue data_url = str(item.get("dataUrl") or "") if not data_url.startswith("data:image/"): continue if len(data_url) > 4_500_000: continue valid.append({ "name": str(item.get("name") or "image"), "type": str(item.get("type") or "image"), "dataUrl": data_url, }) if len(valid) >= 8: break if not valid: return None key = self._openai_api_key() if not key: return { "ok": False, "error": "缺少视觉模型 API Key：请配置环境变量 OPENAI_API_KEY，或把 Key 写入 .openai/api_key.txt。", "imageCount": len(valid), } prompt = ( "你是短视频爆款拆解助手。请只根据用户上传的截图/图片和输入文本提取可见证据，" "不要编造看不到的画面、字幕、评论或数据。输出中文，结构为：" "1. 画面事实；2. 可见文字/OCR；3. 视频可能类型；4. 可用于拆解的内容结构线索；" "5. 仍缺少的信息。用户输入文本如下：\n" f"{raw_text[:1200]}" ) content = [{"type": "input_text", "text": prompt}] for item in valid: content.append({"type": "input_image", "image_url": item["dataUrl"]}) body = json.dumps({ "model": os.environ.get("OPENAI_VISION_MODEL", "gpt-4o-mini"), "input": [{"role": "user", "content": content}], "max_output_tokens": 900, }, ensure_ascii=False).encode("utf-8") req = urllib.request.Request( "https://api.openai.com/v1/responses", data=body, headers={ "Authorization": f"Bearer {key}", "Content-Type": "application/json", }, method="POST", ) try: with urllib.request.urlopen(req, timeout=timeout) as resp: result = json.loads(resp.read().decode("utf-8", "ignore")) except Exception as e: return {"ok": False, "error": f"视觉识别失败：{e}", "imageCount": len(valid)} text = str(result.get("output_text") or "").strip() if not text: parts = [] for out in result.get("output", []) if isinstance(result.get("output"), list) else []: for part in out.get("content", []) if isinstance(out, dict) and isinstance(out.get("content"), list) else []: if isinstance(part, dict) and part.get("text"): parts.append(str(part.get("text"))) text = "\n".join(parts).strip() return {"ok": True, "imageCount": len(valid), "analysis": text} def _analyze_viral_video(self): try: payload = self._read_json_body() raw = str(payload.get("url") or payload.get("text") or "") if isinstance(payload, dict) else "" except Exception as e: self._json_response({"ok": False, "error": f"请求格式错误：{e}"}, status=400) return url = self._extract_first_url(raw) or raw.strip() platform = self._detect_video_platform(url or raw) matched = self._match_hot_video(url) video = dict(matched or {}) limitations = [] supplement_text = str(payload.get("supplementText") or "") if isinstance(payload, dict) else "" images = payload.get("images") if isinstance(payload, dict) else [] vision = self._analyze_viral_images(images, supplement_text or raw) if vision: if vision.get("ok") and vision.get("analysis"): video["imageAnalysis"] = vision.get("analysis") video["description"] = (str(video.get("description") or "") + "\n\n图片识别结果：\n" + vision.get("analysis")).strip() video["analysisBasis"] = (str(video.get("analysisBasis") or "").strip("；") + "；图片视觉识别").strip("；") video["mediaReadable"] = True else: limitations.append(vision.get("error") or "图片已收到，但视觉识别失败。") if platform == "B站": try: video = {**video, **self._fetch_bilibili_video_meta(url)} if vision and vision.get("ok") and vision.get("analysis"): video["imageAnalysis"] = vision.get("analysis") video["description"] = (str(video.get("description") or "") + "\n\n图片识别结果：\n" + vision.get("analysis")).strip() video["analysisBasis"] = (str(video.get("analysisBasis") or "").strip("；") + "；图片视觉识别").strip("；") video["mediaReadable"] = True except Exception as e: limitations.append(f"B站详情接口读取失败：{e}") elif platform in ("抖音", "小红书"): limitations.append(f"{platform}视频详情、画面、音轨通常需要登录态和平台前端授权；当前本地服务拿不到你的平台登录 Cookie，所以只能使用分享文本和榜单已抓到的公开字段。") else: limitations.append("未识别平台，无法调用平台公开接口。") tools = self._tool_status() missing = [name for name, ok in tools.items() if not ok] if missing: limitations.append("本机缺少 yt-dlp / ffmpeg / Whisper 中的部分组件，无法下载视频、提取音频并转逐字稿。") if not video: video = {"platform": platform, "url": url, "tags": []} video["matchedHotVideo"] = bool(matched) video["mediaReadable"] = bool(video.get("mediaReadable")) self._json_response({"ok": True, "video": video, "toolStatus": tools, "limitations": limitations}) def _json_response(self, payload, status=200): self.send_response(status) self._cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps(payload, ensure_ascii=False).encode("utf-8")) def _daily_english_listening_stats(self): return read_daily_english_listening_stats() def _reading_note_file_for_date(self, date_str): if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", date_str or ""): return None year = date_str[:4] return os.path.join(WORKSPACE, "读书收获", year, f"今日读书收获_{date_str}.md") def _markdown_to_html(self, text): out = [] in_ol = False in_ul = False def close_ol(): nonlocal in_ol if in_ol: out.append("") in_ol = False def close_ul(): nonlocal in_ul if in_ul: out.append("") in_ul = False for raw in text.splitlines(): line = raw.strip() if not line: close_ol() close_ul() continue safe = html.escape(line) safe = re.sub(r"\*\*(.+?)\*\*", r"\1", safe) safe = re.sub(r"\*(.+?)\*", r"\1", safe) if safe.startswith("# "): close_ol() close_ul() heading = safe[2:] if "｜作者：" in heading: book_title, author = heading.split("｜作者：", 1) heading = f'{book_title}作者：{author}' out.append(f"

{heading}

") elif safe.startswith("## "): close_ol() close_ul() out.append(f"

{safe[3:]}

") elif safe.startswith("### "): close_ol() close_ul() out.append(f"

{safe[4:]}

") elif safe.startswith("> "): close_ol() close_ul() out.append(f"
{safe[5:]}
") elif re.match(r"^\d+\. ", safe): close_ul() if not in_ol: out.append("
    ") in_ol = True item_text = re.sub(r"^\d+\. ", "", safe) out.append(f"
  1. {item_text}
    2. ") elif safe.startswith("- "): close_ol() if not in_ul: out.append("