#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Personal Life OS - 本地服务 - 托管 life-dashboard.html 等静态文件 - 提供 /sync (POST) 接收来自微信读书的书签按钮同步数据 - 提供 /sync.json (GET) 供工作台读取最新同步结果 """ import http.server import socketserver import html import json 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, "读书收获") HOT_VIDEOS_PATH = os.path.join(WORKSPACE, "hot-videos.json") HOT_VIDEO_AUTO_SCRIPT = os.path.join(WORKSPACE, "hot_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") PORT = 8765 HOT_VIDEO_AUTO_HOUR = 7 READING_AUTO_HOUR = 8 # 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() 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 []) < 10: return "" return updated_at[:10] except Exception: return "" 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_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() 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): 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") 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/hot-videos/run"): self._run_hot_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 _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 _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 ["AI宠物带货"], "author": owner.get("name") or "", "duration": info.get("duration"), "description": info.get("desc") or "", "analysisBasis": "B站公开接口：标题、作者、播放、点赞、粉丝、标签", } 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 = [] if platform == "B站": try: video = {**video, **self._fetch_bilibili_video_meta(url)} 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"] = False 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 _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 def close_ol(): nonlocal in_ol if in_ol: out.append("") in_ol = False for raw in text.splitlines(): line = raw.strip() if not line: close_ol() continue safe = html.escape(line) safe = re.sub(r"\*\*(.+?)\*\*", r"\1", safe) safe = re.sub(r"\*(.+?)\*", r"\1", safe) if safe.startswith("# "): close_ol() out.append(f"

{safe[2:]}

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

{safe[3:]}

") elif safe.startswith("> "): close_ol() out.append(f"
{safe[5:]}
") elif re.match(r"^\d+\. ", safe): 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 == "---": close_ol() out.append("
    ") else: close_ol() out.append(f"

    {safe}

    ") close_ol() return "\n".join(out) def _serve_reading_note_view(self): qs = parse_qs(urlparse(self.path).query) date_str = (qs.get("date") or [""])[0] note_path = self._reading_note_file_for_date(date_str) if not note_path or not os.path.exists(note_path): self.send_response(404) self._cors() self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write("

    读书笔记不存在

    ".encode("utf-8")) return with open(note_path, "r", encoding="utf-8") as f: body = self._markdown_to_html(f.read()) page = f""" 今日读书收获 {html.escape(date_str)}
    ← 返回工作台 {body}
    """ self.send_response(200) self._cors() self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(page.encode("utf-8")) def _strip_markdown(self, text): text = re.sub(r"^#+\s*", "", text) text = re.sub(r"^>\s*", "", text) text = re.sub(r"^\d+\.\s*", "", text) text = re.sub(r"[*_`#]", "", text) return re.sub(r"\s+", " ", text).strip() def _note_summary(self, lines): sections = ("## 深度解读", "## 核心观点", "## 个人思考", "## 金句摘录") for section in sections: try: start = next(i for i, line in enumerate(lines) if line.strip() == section) except StopIteration: continue parts = [] for line in lines[start + 1:]: stripped = line.strip() if stripped.startswith("## "): break clean = self._strip_markdown(stripped) if clean and clean != "---": parts.append(clean) if len(" ".join(parts)) >= 120: break if parts: return " ".join(parts)[:140] for line in lines: clean = self._strip_markdown(line.strip()) if clean and not clean.startswith("今日读书收获"): return clean[:140] return "暂无摘要" def _note_meta(self, path): name = os.path.basename(path) match = re.search(r"(\d{4}-\d{2}-\d{2})", name) date_str = match.group(1) if match else time.strftime("%Y-%m-%d", time.localtime(os.path.getmtime(path))) with open(path, "r", encoding="utf-8") as f: text = f.read() lines = text.splitlines() heading = next((self._strip_markdown(line) for line in lines if line.startswith("# ")), f"读书收获（{date_str}）") primary_match = re.search(r"(?:今日书籍|书名)：\s*《([^》]+)》", text) if not primary_match: primary_match = re.search(r"——\s*《([^》]+)》", text) if not primary_match: primary_match = re.search(r"《([^》]+)》", text) book_title = primary_match.group(1) if primary_match else "综合读书笔记" return { "date": date_str, "title": heading, "bookTitle": book_title, "summary": self._note_summary(lines), "path": os.path.relpath(path, WORKSPACE), "viewUrl": f"/reading-note-view?date={date_str}", "modifiedAt": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(os.path.getmtime(path))), } def _serve_reading_notes(self): notes = [] if os.path.exists(READING_NOTES_DIR): for root, _, files in os.walk(READING_NOTES_DIR): for name in files: if not name.lower().endswith(".md"): continue path = os.path.join(root, name) try: notes.append(self._note_meta(path)) except Exception as e: notes.append({ "date": "", "title": name, "bookTitle": "读取失败", "summary": str(e), "path": os.path.relpath(path, WORKSPACE), "viewUrl": "", "modifiedAt": "", }) notes.sort(key=lambda item: (item.get("date") or "", item.get("modifiedAt") or ""), reverse=True) self._json_response({"ok": True, "notes": notes}) def _serve_reading_note(self): qs = parse_qs(urlparse(self.path).query) date_str = (qs.get("date") or [""])[0] note_path = self._reading_note_file_for_date(date_str) if not note_path or not os.path.exists(note_path): self._json_response({"ok": False, "error": "读书笔记不存在"}, status=404) return with open(note_path, "r", encoding="utf-8") as f: markdown = f.read() self._json_response({ "ok": True, "date": date_str, "meta": self._note_meta(note_path), "html": self._markdown_to_html(markdown), }) def _serve_hot_videos(self): default_data = { "ok": True, "updatedAt": "", "source": "daily automation", "track": "AI宠物带货赛道", "videos": [], "error": "", } if not os.path.exists(HOT_VIDEOS_PATH): self._json_response(default_data) return try: with open(HOT_VIDEOS_PATH, "r", encoding="utf-8") as f: data = json.load(f) if isinstance(data, list): data = {**default_data, "videos": data} elif isinstance(data, dict): data = {**default_data, **data} else: data = {**default_data, "error": "hot-videos.json format is not supported"} data["ok"] = not bool(data.get("error")) self._json_response(data) except Exception as e: self._json_response({**default_data, "ok": False, "error": str(e)}, status=500) def _serve_life_data(self): if not os.path.exists(LIFE_DATA_PATH): self._json_response({"ok": True, "exists": False, "data": None, "updatedAt": 0}) return try: with open(LIFE_DATA_PATH, "r", encoding="utf-8") as f: data = json.load(f) self._json_response({ "ok": True, "exists": True, "data": data, "updatedAt": data.get("__syncUpdatedAt", 0) if isinstance(data, dict) else 0, }) except Exception as e: self._json_response({"ok": False, "error": str(e)}, status=500) def _life_preset_has_value(self, task): if not isinstance(task, dict): return False return bool( task.get("checked") or task.get("sleepTime") or task.get("statusText") or float(task.get("count") or 0) > 0 or float(task.get("readMinutes") or 0) > 0 ) def _merge_life_preset(self, target, source): if not isinstance(target, dict) or not isinstance(source, dict): return if self._life_preset_has_value(source) and not self._life_preset_has_value(target): target.update(source) return if not self._life_preset_has_value(source): return task_id = source.get("id") if task_id == "sleep": if not target.get("sleepTime") and source.get("sleepTime"): target["sleepTime"] = source.get("sleepTime") target["checked"] = bool(target.get("checked") or source.get("checked")) elif task_id == "read": if float(source.get("readMinutes") or 0) > float(target.get("readMinutes") or 0): target["readMinutes"] = source.get("readMinutes") if source.get("statusText"): target["statusText"] = source.get("statusText") target["checked"] = bool(target.get("checked") or source.get("checked")) for key in ("autoCheckin", "thresholdMinutes"): if source.get(key) and not target.get(key): target[key] = source.get(key) elif task_id == "water": target["count"] = max(float(target.get("count") or 0), float(source.get("count") or 0)) target["checked"] = bool(target.get("checked") or source.get("checked")) else: target["checked"] = bool(target.get("checked") or source.get("checked")) def _merge_life_daily_plan(self, incoming, existing): incoming_daily = incoming.setdefault("dailyPlan", {}) existing_daily = existing.get("dailyPlan") if isinstance(existing, dict) else {} if not isinstance(existing_daily, dict): return for date_key, existing_plan in existing_daily.items(): if not isinstance(existing_plan, dict): continue target_plan = incoming_daily.setdefault(date_key, {"presets": [], "customs": []}) target_plan.setdefault("presets", []) target_plan.setdefault("customs", []) target_presets = target_plan["presets"] if not isinstance(target_presets, list): target_presets = [] target_plan["presets"] = target_presets by_id = {p.get("id"): p for p in target_presets if isinstance(p, dict) and p.get("id")} for source_task in existing_plan.get("presets") or []: if not isinstance(source_task, dict) or not source_task.get("id"): continue target_task = by_id.get(source_task.get("id")) if target_task is None: target_presets.append(dict(source_task)) by_id[source_task.get("id")] = target_presets[-1] else: self._merge_life_preset(target_task, source_task) target_customs = target_plan["customs"] if isinstance(target_plan.get("customs"), list) else [] target_plan["customs"] = target_customs seen = {("id:" + str(c.get("id"))) if isinstance(c, dict) and c.get("id") else ("name:" + str(c.get("name", ""))) for c in target_customs if isinstance(c, dict)} for custom in existing_plan.get("customs") or []: if not isinstance(custom, dict): continue key = ("id:" + str(custom.get("id"))) if custom.get("id") else ("name:" + str(custom.get("name", ""))) if key not in seen: target_customs.append(dict(custom)) seen.add(key) def _exercise_key(self, entry): if not isinstance(entry, dict): return "" if entry.get("id"): return "id:" + str(entry.get("id")) return "|".join(str(entry.get(k, "")) for k in ("type", "duration", "seconds", "time", "timestamp", "source")) def _merge_life_exercises(self, incoming, existing): incoming_exercises = incoming.setdefault("exercises", {}) existing_exercises = existing.get("exercises") if isinstance(existing, dict) else {} if not isinstance(existing_exercises, dict): return for date_key, entries in existing_exercises.items(): if not isinstance(entries, list) or not entries: continue target_entries = incoming_exercises.setdefault(date_key, []) if not isinstance(target_entries, list): target_entries = [] incoming_exercises[date_key] = target_entries seen = {self._exercise_key(e) for e in target_entries if self._exercise_key(e)} for entry in entries: key = self._exercise_key(entry) if key and key not in seen: target_entries.append(dict(entry)) seen.add(key) def _merge_life_english(self, incoming, existing): incoming_english = incoming.setdefault("englishLearn", {}) existing_english = existing.get("englishLearn") if isinstance(existing, dict) else {} if not isinstance(existing_english, dict): return incoming_sessions = incoming_english.setdefault("sessions", []) if isinstance(incoming_sessions, list): seen = { s.get("id") or f"{s.get('date')}|{s.get('skill')}|{s.get('key')}|{s.get('timestamp')}" for s in incoming_sessions if isinstance(s, dict) } for session in existing_english.get("sessions") or []: if not isinstance(session, dict): continue key = session.get("id") or f"{session.get('date')}|{session.get('skill')}|{session.get('key')}|{session.get('timestamp')}" if key not in seen: incoming_sessions.append(dict(session)) seen.add(key) def _merge_life_data_for_save(self, incoming): if not os.path.exists(LIFE_DATA_PATH): return incoming try: with open(LIFE_DATA_PATH, "r", encoding="utf-8") as f: existing = json.load(f) except Exception: return incoming if not isinstance(existing, dict): return incoming self._merge_life_daily_plan(incoming, existing) self._merge_life_exercises(incoming, existing) self._merge_life_english(incoming, existing) incoming["__syncUpdatedAt"] = max( int(incoming.get("__syncUpdatedAt") or 0), int(existing.get("__syncUpdatedAt") or 0), int(time.time() * 1000), ) return incoming def _save_life_data(self): try: payload = self._read_json_body() data = payload.get("data") if isinstance(payload, dict) else None if not isinstance(data, dict): self._json_response({"ok": False, "error": "data must be an object"}, status=400) return data = self._merge_life_data_for_save(data) data["__syncUpdatedAt"] = int(data.get("__syncUpdatedAt") or 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) self._json_response({"ok": True, "updatedAt": data["__syncUpdatedAt"]}) except Exception as e: self._json_response({"ok": False, "error": str(e)}, status=500) def _favorite_default_data(self): return { "ok": True, "updatedAt": "", "sources": ["抖音", "小红书"], "categories": ["知识技能", "生活灵感", "搞笑娱乐", "未分类"], "knowledgeSubcategories": ["编程/AI", "职场沟通", "理财", "其他知识"], "syncStatus": { "douyin": "未同步", "xiaohongshu": "未同步", "message": "抖音/小红书收藏需要登录态或导出的分享链接。当前本地服务已准备好入库、分类、转写和分析流程。", }, "videos": [], } def _load_favorite_data(self): default = self._favorite_default_data() if not os.path.exists(FAVORITE_COLLECTIONS_PATH): return default try: with open(FAVORITE_COLLECTIONS_PATH, "r", encoding="utf-8") as f: data = json.load(f) if not isinstance(data, dict): return {**default, "syncStatus": {**default["syncStatus"], "message": "收藏数据格式不正确，已使用空数据。"}} data = {**default, **data} if not isinstance(data.get("videos"), list): data["videos"] = [] data["ok"] = True return data except Exception as e: return {**default, "ok": False, "syncStatus": {**default["syncStatus"], "message": str(e)}} def _save_favorite_data(self, data): data["updatedAt"] = time.strftime("%Y-%m-%d %H:%M:%S") with open(FAVORITE_COLLECTIONS_PATH, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) def _serve_favorite_collections(self): self._json_response(self._load_favorite_data()) def _read_json_body(self): length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length) if length else b"{}" return json.loads(body.decode("utf-8")) def _detect_video_platform(self, text): lower = (text or "").lower() if "douyin.com" in lower or "iesdouyin" in lower: return "抖音" if "xiaohongshu.com" in lower or "xhslink.com" in lower: return "小红书" if "bilibili.com" in lower or "b23.tv" in lower: return "B站" return "未知平台" def _extract_favorite_tags(self, text): tags = re.findall(r"#([^#\s，。,.!！?？]+)", text or "") keywords = ["AI", "编程", "职场", "沟通", "理财", "基金", "省钱", "家居", "穿搭", "美食", "搞笑", "段子", "宠物"] for keyword in keywords: if keyword in (text or "") and keyword not in tags: tags.append(keyword) return tags[:8] def _clean_favorite_title(self, text, url): title = re.sub(r"https?://\S+", " ", text or "") title = re.sub(r"#[^#\s，。,.!！?？]+", " ", title) title = re.sub(r"^[\d.:\sA-Za-z@/]+", " ", title) title = re.sub(r"\s+", " ", title).strip(" ，。,.!！?？") if not title: title = url or "未命名收藏视频" return title[:90] def _classify_favorite_video(self, title, tags): text = f"{title} {' '.join(tags)}" knowledge = ["AI", "编程", "代码", "Python", "Cursor", "效率", "教程", "学习", "职场", "沟通", "表达", "理财", "基金", "股票", "存钱", "副业"] funny = ["搞笑", "段子", "沙雕", "哈哈", "离谱", "整活", "娱乐"] life = ["生活", "家居", "穿搭", "美食", "旅行", "收纳", "做饭", "灵感", "装修", "护肤"] if any(k in text for k in knowledge): if any(k in text for k in ["AI", "编程", "代码", "Python", "Cursor", "豆包", "Coze"]): return "知识技能", "编程/AI" if any(k in text for k in ["职场", "沟通", "表达", "汇报", "面试"]): return "知识技能", "职场沟通" if any(k in text for k in ["理财", "基金", "股票", "存钱", "省钱", "投资"]): return "知识技能", "理财" return "知识技能", "其他知识" if any(k in text for k in funny): return "搞笑娱乐", "" if any(k in text for k in life): return "生活灵感", "" return "未分类", "" def _favorite_analysis(self, item): title = item.get("title") or "未命名收藏视频" transcript = item.get("transcript") or "" category = item.get("category") or "未分类" subcategory = item.get("subcategory") or "" tags = "、".join(item.get("tags") or []) material = transcript if transcript else f"{title} {tags}" if category == "知识技能": focus = f"这条内容适合沉淀为一张知识卡片，重点关注“问题、方法、例子、行动”四层结构。{('子类：' + subcategory + '。') if subcategory else ''}" action = "提炼 3 个可执行步骤，并把其中最容易当天实践的一步加入待办。" elif category == "生活灵感": focus = "这条内容的价值在于场景启发，适合记录触发灵感的画面、物品、地点或生活方式选择。" action = "拆出可复用元素：场景、预算、准备物、替代方案。" elif category == "搞笑娱乐": focus = "这条内容主要依靠情绪释放和反差制造记忆点，适合分析包袱铺垫、节奏停顿和结尾反转。" action = "记录笑点公式：铺垫对象、预期方向、反差动作、结尾回扣。" else: focus = "当前信息较少，建议先补充标题、标签或逐字稿，再做更精确归类。" action = "先判断它是知识、灵感还是娱乐，再决定是否收藏进长期合集。" summary = self._strip_markdown(material)[:180] or "暂无可分析文本。" return { "summary": summary, "contentInsight": focus, "actionSuggestion": action, "updatedAt": time.strftime("%Y-%m-%d %H:%M:%S"), } def _import_favorite_collections(self): try: payload = self._read_json_body() raw = payload.get("text", "") if isinstance(payload, dict) else "" snippets = re.split(r"\n{2,}|\r?\n(?=.*https?://)", raw.strip()) data = self._load_favorite_data() existing_urls = {item.get("url") for item in data.get("videos", []) if item.get("url")} added = [] for snippet in snippets: snippet = snippet.strip() if not snippet: continue url_match = re.search(r"https?://[^\s，。)）]+", snippet) if not url_match: continue url = url_match.group(0) if url in existing_urls: continue tags = self._extract_favorite_tags(snippet) title = self._clean_favorite_title(snippet, url) category, subcategory = self._classify_favorite_video(title, tags) item = { "id": f"fav_{int(time.time() * 1000)}_{len(data.get('videos', [])) + len(added)}", "platform": self._detect_video_platform(url), "title": title, "url": url, "category": category, "subcategory": subcategory, "tags": tags, "collectedAt": time.strftime("%Y-%m-%d %H:%M:%S"), "transcriptStatus": "待转写", "transcript": "", "analysisStatus": "待分析", "analysis": {}, } item["analysis"] = self._favorite_analysis(item) item["analysisStatus"] = "已生成初步分析" added.append(item) existing_urls.add(url) data["videos"] = added + data.get("videos", []) if added: data["syncStatus"]["message"] = f"已导入 {len(added)} 条收藏链接，并完成自动分类与初步分析。" self._save_favorite_data(data) self._json_response({"ok": True, "added": len(added), "data": data}) except Exception as e: self._json_response({"ok": False, "error": str(e)}, status=500) def _node_command(self): if os.path.exists(NODE_EXE): return NODE_EXE return shutil.which("node") def _node_env(self): env = os.environ.copy() paths = [] if os.path.exists(NODE_MODULES): paths.append(NODE_MODULES) if env.get("NODE_PATH"): paths.append(env["NODE_PATH"]) if paths: env["NODE_PATH"] = os.pathsep.join(paths) return env def _favorite_item_from_sync(self, raw_item, index=0): url = raw_item.get("url", "") title = raw_item.get("title") or self._clean_favorite_title("", url) tags = self._extract_favorite_tags(title) category, subcategory = self._classify_favorite_video(title, tags) item = { "id": f"fav_{int(time.time() * 1000)}_{index}", "platform": raw_item.get("platform") or self._detect_video_platform(url), "title": title, "url": url, "category": category, "subcategory": subcategory, "tags": tags, "collectedAt": time.strftime("%Y-%m-%d %H:%M:%S"), "transcriptStatus": "待转写", "transcript": "", "analysisStatus": "待分析", "analysis": {}, "source": raw_item.get("source") or "login-sync", } item["analysis"] = self._favorite_analysis(item) item["analysisStatus"] = "已生成初步分析" return item def _merge_favorite_videos(self, data, raw_videos): existing_urls = {item.get("url") for item in data.get("videos", []) if item.get("url")} added = [] for raw in raw_videos: url = raw.get("url") if not url or url in existing_urls: continue item = self._favorite_item_from_sync(raw, len(data.get("videos", [])) + len(added)) added.append(item) existing_urls.add(url) if added: data["videos"] = added + data.get("videos", []) return added def _login_favorite_collections(self): try: payload = self._read_json_body() except Exception: payload = {} platform = payload.get("platform", "all") if isinstance(payload, dict) else "all" node = self._node_command() if not node: self._json_response({"ok": False, "error": "未找到 Node.js，无法打开登录窗口。"}, status=500) return if not os.path.exists(FAVORITE_SYNC_SCRIPT): self._json_response({"ok": False, "error": "sync-favorites.js 不存在。"}, status=500) return try: os.makedirs(FAVORITE_LOG_DIR, exist_ok=True) profile_dir = os.path.join(FAVORITE_LOG_DIR, "browser-profile") os.makedirs(profile_dir, exist_ok=True) log_path = os.path.join(FAVORITE_LOG_DIR, "login.log") with open(log_path, "a", encoding="utf-8") as log_file: log_file.write(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] open login chrome {platform}\n") chrome = "C:\\Program Files\\Google\\Chrome\\Application\\chrome.exe" if not os.path.exists(chrome): chrome = shutil.which("chrome") or shutil.which("chrome.exe") if not chrome: self._json_response({"ok": False, "error": "未找到 Chrome，无法打开登录窗口。"}, status=500) return urls = [] if platform in ("all", "douyin"): urls.append("https://www.douyin.com/") if platform in ("all", "xiaohongshu"): urls.append("https://www.xiaohongshu.com/") subprocess.Popen( [chrome, f"--user-data-dir={profile_dir}", "--remote-debugging-port=9223", "--new-window", *urls], cwd=WORKSPACE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if os.name == "nt" else 0, ) data = self._load_favorite_data() data["syncStatus"] = { "douyin": "登录窗口已打开", "xiaohongshu": "登录窗口已打开", "message": "已打开独立 Chrome 登录窗口。请扫码登录抖音和小红书；登录完成后关闭这个登录窗口，再回到工作台点击“同步收藏”。", "updatedAt": time.strftime("%Y-%m-%d %H:%M:%S"), } self._save_favorite_data(data) self._json_response({"ok": True, "data": data, "message": data["syncStatus"]["message"]}) except Exception as e: self._json_response({"ok": False, "error": str(e)}, status=500) def _sync_favorite_collections(self): node = self._node_command() data = self._load_favorite_data() if not node: data["syncStatus"] = { "douyin": "同步失败", "xiaohongshu": "同步失败", "message": "未找到 Node.js，无法启动浏览器同步脚本。", "updatedAt": time.strftime("%Y-%m-%d %H:%M:%S"), } self._save_favorite_data(data) self._json_response({"ok": False, "data": data, "error": data["syncStatus"]["message"]}, status=500) return try: proc = subprocess.run( [node, FAVORITE_SYNC_SCRIPT, "sync", "all"], cwd=WORKSPACE, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=180, env=self._node_env(), ) stdout = (proc.stdout or "").strip() result = json.loads(stdout) if stdout else {"ok": False, "error": proc.stderr or "同步脚本无输出", "statuses": [], "videos": []} added = self._merge_favorite_videos(data, result.get("videos") or []) statuses = result.get("statuses") or [] status_map = {s.get("platform"): s for s in statuses if isinstance(s, dict)} data["syncStatus"] = { "douyin": status_map.get("抖音", {}).get("message", "未返回状态"), "xiaohongshu": status_map.get("小红书", {}).get("message", "未返回状态"), "message": f"同步完成，新增 {len(added)} 条收藏。若新增为 0，请确认登录窗口里已经进入收藏页并能看到收藏内容。", "updatedAt": time.strftime("%Y-%m-%d %H:%M:%S"), } if result.get("error"): data["syncStatus"]["message"] = f"同步脚本异常：{result.get('error')}" self._save_favorite_data(data) self._json_response({"ok": len(added) > 0, "added": len(added), "result": result, "data": data}) except subprocess.TimeoutExpired: data["syncStatus"] = { "douyin": "同步超时", "xiaohongshu": "同步超时", "message": "同步超过 3 分钟未完成。请确认登录窗口没有卡在验证码或权限页。", "updatedAt": time.strftime("%Y-%m-%d %H:%M:%S"), } self._save_favorite_data(data) self._json_response({"ok": False, "data": data, "error": data["syncStatus"]["message"]}, status=504) except Exception as e: data["syncStatus"] = { "douyin": "同步失败", "xiaohongshu": "同步失败", "message": str(e), "updatedAt": time.strftime("%Y-%m-%d %H:%M:%S"), } self._save_favorite_data(data) self._json_response({"ok": False, "data": data, "error": str(e)}, status=500) def _tool_status(self): return { "ytDlp": bool(shutil.which("yt-dlp")), "ffmpeg": bool(shutil.which("ffmpeg")), "whisperCli": bool(shutil.which("whisper")), } def _transcribe_favorite_with_whisper(self, item): os.makedirs(FAVORITE_MEDIA_DIR, exist_ok=True) base = re.sub(r"[^a-zA-Z0-9_-]", "_", item.get("id") or str(int(time.time()))) output_tpl = os.path.join(FAVORITE_MEDIA_DIR, base + ".%(ext)s") subprocess.run( ["yt-dlp", "-x", "--audio-format", "mp3", "-o", output_tpl, item.get("url", "")], cwd=WORKSPACE, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=180, check=True, ) audio_path = os.path.join(FAVORITE_MEDIA_DIR, base + ".mp3") if not os.path.exists(audio_path): matches = [os.path.join(FAVORITE_MEDIA_DIR, name) for name in os.listdir(FAVORITE_MEDIA_DIR) if name.startswith(base + ".")] audio_path = matches[0] if matches else audio_path if not os.path.exists(audio_path): raise RuntimeError("音频提取失败：未找到输出音频文件") subprocess.run( ["whisper", audio_path, "--language", "Chinese", "--model", "base", "--output_format", "txt", "--output_dir", FAVORITE_MEDIA_DIR], cwd=WORKSPACE, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=600, check=True, ) transcript_path = os.path.splitext(audio_path)[0] + ".txt" if not os.path.exists(transcript_path): raise RuntimeError("Whisper 已运行，但未找到逐字稿文件") with open(transcript_path, "r", encoding="utf-8", errors="replace") as f: return f.read().strip(), os.path.relpath(transcript_path, WORKSPACE) def _analyze_favorite_collection_item(self): try: payload = self._read_json_body() item_id = payload.get("id", "") if isinstance(payload, dict) else "" data = self._load_favorite_data() item = next((v for v in data.get("videos", []) if v.get("id") == item_id), None) if not item: self._json_response({"ok": False, "error": "收藏视频不存在"}, status=404) return tools = self._tool_status() missing = [name for name, ok in tools.items() if not ok] if missing and not item.get("transcript"): item["transcriptStatus"] = "缺少转写组件" item["transcriptNote"] = "需要安装 yt-dlp、ffmpeg 和 Whisper 后才能从平台链接提取音频并转逐字稿。当前已先基于标题/标签完成内容分析。" else: try: if not item.get("transcript"): item["transcriptStatus"] = "转写中" transcript, transcript_path = self._transcribe_favorite_with_whisper(item) item["transcript"] = transcript item["transcriptPath"] = transcript_path item["transcriptStatus"] = "已转写" item["transcriptNote"] = "已提取音频并完成 Whisper 逐字稿。" except Exception as e: item["transcriptStatus"] = "转写失败" item["transcriptNote"] = f"音频提取或 Whisper 转写失败：{e}。当前已先基于标题/标签完成内容分析。" item["analysis"] = self._favorite_analysis(item) item["analysisStatus"] = "已分析" item["analysis"]["toolStatus"] = tools item["analysis"]["transcriptBasis"] = "逐字稿" if item.get("transcript") else "标题/标签" self._save_favorite_data(data) self._json_response({"ok": True, "item": item, "data": data}) except Exception as e: self._json_response({"ok": False, "error": str(e)}, status=500) def do_GET(self): req_path = urlparse(self.path).path if req_path == "/api/hot-videos": self._serve_hot_videos() elif req_path == "/api/life-data": self._serve_life_data() elif req_path == "/api/favorite-collections": self._serve_favorite_collections() elif req_path == "/api/reading-note": self._serve_reading_note() elif req_path == "/api/reading-notes": self._serve_reading_notes() elif req_path == "/reading-note-view": self._serve_reading_note_view() elif req_path == "/sync.json": self.send_response(200) self._cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() if os.path.exists(SYNC_PATH): with open(SYNC_PATH, "r", encoding="utf-8") as f: self.wfile.write(f.read().encode("utf-8")) else: self.wfile.write(json.dumps({"books": []}).encode("utf-8")) elif self.path.startswith("/api/weread-readtimes"): global _readtimes_cache try: now = time.time() if _readtimes_cache["data"] is not None and now - _readtimes_cache["ts"] < _READTIMES_TTL: data = _readtimes_cache["data"] else: data = self._gw_post("/readdata/detail", {"mode": "weekly"}, timeout=8) _readtimes_cache = {"ts": now, "data": data} 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, "data": data}, ensure_ascii=False).encode("utf-8")) except Exception as e: 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": False, "error": str(e)}, ensure_ascii=False).encode("utf-8")) elif self.path.startswith("/api/last-reading-note"): self.send_response(200) self._cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() if os.path.exists(LAST_NOTE_PATH): with open(LAST_NOTE_PATH, "r", encoding="utf-8") as f: self.wfile.write(f.read().encode("utf-8")) else: self.wfile.write(json.dumps({"ok": False, "error": "not generated yet"}).encode("utf-8")) else: super().do_GET() def end_headers(self): self.send_header("Cache-Control", "no-store") self._cors() super().end_headers() def log_message(self, fmt, *args): pass # 静默日志 def main(): # 初始化空同步文件 if not os.path.exists(SYNC_PATH): with open(SYNC_PATH, "w", encoding="utf-8") as f: json.dump({"books": []}, f, ensure_ascii=False) if not os.path.exists(FAVORITE_COLLECTIONS_PATH): with open(FAVORITE_COLLECTIONS_PATH, "w", encoding="utf-8") as f: json.dump(Handler._favorite_default_data(Handler), f, ensure_ascii=False, indent=2) os.chdir(WORKSPACE) start_hot_video_auto_scheduler() start_reading_auto_scheduler() socketserver.ThreadingTCPServer.allow_reuse_address = True with socketserver.ThreadingTCPServer(("", PORT), Handler) as httpd: print(f"Personal Life OS server running at http://localhost:{PORT}") httpd.serve_forever() if __name__ == "__main__": main()