import json, os, ssl, urllib.request, datetime, re WORKSPACE = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) KEY_PATH = os.path.join(WORKSPACE, '.weread', 'apikey.txt') GW_URL = 'https://i.weread.qq.com/api/agent/gateway' STREAK_PATH = os.path.join(WORKSPACE, 'reading-streak.json') TZ8 = datetime.timezone(datetime.timedelta(hours=8)) TODAY = datetime.datetime.now(TZ8).date() TODAY_STR = TODAY.strftime('%Y-%m-%d') TODAY_MID = int(datetime.datetime(TODAY.year, TODAY.month, TODAY.day, tzinfo=TZ8).timestamp()) YESTERDAY_MID = TODAY_MID - 86400 OUT_DIR = os.path.join(WORKSPACE, '读书收获', str(TODAY.year)) OUT_FILE = os.path.join(OUT_DIR, f'今日读书收获_{TODAY_STR}.md') with open(KEY_PATH, encoding='utf-8') as f: KEY = f.read().strip() CTX = ssl.create_default_context() def gw(api, **kw): body = json.dumps({'skill_version': '1.0.4', 'api_name': api, **kw}, 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') try: with urllib.request.urlopen(req, timeout=30, context=CTX) as r: d = json.loads(r.read().decode('utf-8')) if isinstance(d, dict) and 'upgrade_info' in d: print('UPGRADE_INFO:', d['upgrade_info']) return d except urllib.error.HTTPError as e: print('HTTP error for', api, ':', e.code, e.reason) return {} except Exception as e: print('Error for', api, ':', type(e).__name__, e) return {} # ---------- 1. 拉取真实数据 ---------- shelf = gw('/shelf/sync') books = [b for b in shelf.get('books', []) if b.get('finishReading') != 1] rd = gw('/readdata/detail', mode='weekly') rt = rd.get('readTimes', {}) today_dur = int(rt.get(str(TODAY_MID), 0) or 0) yesterday_dur = int(rt.get(str(YESTERDAY_MID), 0) or 0) # 只在「今天」有阅读时才把今天当汇报日；否则昨天 if today_dur > 0: report_mid = TODAY_MID report_date = TODAY else: report_mid = YESTERDAY_MID report_date = TODAY - datetime.timedelta(days=1) # ---------- 2. 收集今天真实划线（严格过滤质量） ---------- def is_good_quote(text): t = (text or '').strip() if len(t) < 8: return False cn = sum(1 for c in t if '\u4e00' <= c <= '\u9fff') if cn == 0: # 纯英文/数字：要求更长且以完整标点结尾 if len(t) < 20: return False if t[-1] not in '.!?"\'’）：': return False else: # 含中文：至少 10 字，且不以截断特征结尾 if len(t) < 10: return False if t[-1] in ',;: （(': return False return True def quote_quality_score(text): t = re.sub(r'\s+', '', (text or '').strip()) if not is_good_quote(t): return -100 score = 0 if 14 <= len(t) <= 90: score += 8 elif len(t) <= 140: score += 3 else: score -= 8 value_words = '人生 心 自由 自律 勇气 选择 成长 认知 时间 责任 命运 世界 真实 理想 行动 坚持 克制 清醒 边界 关系 知识 思考 判断 改变 长期 复盘 改革 利益 规则 代价'.split() structure_words = '不是 而是 因为 所以 只有 才能 应当 必须 真正 意味着 关键 在于'.split() plot_words = '皇上 皇帝 正月 天子 大臣 衙门 朝廷 奉旨 传杯递盏 风向 官员 宫里 序幕 这一刻'.split() score += sum(3 for w in value_words if w in t) score += sum(2 for w in structure_words if w in t) score -= sum(4 for w in plot_words if w in t) if t.startswith('从这一刻起') or t.startswith('这一刻起') or ('揭开' in t and '序幕' in t): score -= 12 if not any(w in t for w in value_words + structure_words): score -= 6 score -= min(t.count('，') + t.count('、'), 8) if re.search(r'[。！？.!?]$', t): score += 2 if re.search(r'[“”《》]', t): score -= 1 if len(re.findall(r'[，。；：、]', t)) >= 6: score -= 5 return score def best_quote(candidates, min_score=-100): scored = [(quote_quality_score(q[0]), q) for q in candidates] scored = [item for item in scored if item[0] >= min_score] if not scored: return None scored.sort(key=lambda item: item[0], reverse=True) return scored[0][1] def build_deep_insights(quote_text, book_title): t = (quote_text or '').strip() if not t: return [ f"今天没有提取到足够有价值的句子，后续会继续从《{book_title}》中筛选更适合复盘的内容。", "建议阅读时优先划下能解释人生选择、认知变化、情绪管理或长期行动的句子。", "真正值得收录的金句，应该能在明天仍然提醒你做出一个更好的判断。" ] themes = [] if re.search(r'权|势|官|皇|朝|规矩|风向|风险', t): themes.append('复杂系统里的清醒判断') if re.search(r'时间|今日|明日|长期|坚持|自律|行动|完成', t): themes.append('长期主义和行动节奏') if re.search(r'自由|选择|勇气|责任|边界|自己', t): themes.append('个人选择与自我负责') if re.search(r'心|情绪|痛苦|孤独|恐惧|欲望|克制', t): themes.append('情绪觉察与内在秩序') if re.search(r'学|知|认知|思考|真理|世界|判断', t): themes.append('认知升级和判断力') if not themes: themes.append('把阅读内容转化成现实判断') main_theme = themes[0] return [ f"这句话值得保留的核心，不是字面情节，而是它提醒你看到“{main_theme}”：人在具体处境里做判断时，往往不能只看表面信息，还要看背后的关系、风险和代价。", f"它能引发的思考是：我今天遇到的问题，哪些只是表层事件，哪些才是真正决定结果的底层变量？把注意力从情绪反应移到结构判断上，才更容易做出稳的选择。", "对个人成长的价值在于训练复盘能力。每天只抓住一句有分量的话，逼自己把阅读从“看过”推进到“理解、迁移、使用”，这比堆很多摘抄更有效。" ] quotes = [] # (text, label, source) source: 'today'(当日新增个人划线) / 'popular'(书中热门划线) book_progress = [] for b in books: bid = b['bookId'] title = b.get('title', '未知书名') author = b.get('author', '') pg = gw('/book/getprogress', bookId=bid) prog = pg.get('book', pg) if isinstance(pg, dict) else {} progress = prog.get('progress') # 最近阅读的信号：优先用 book.updateTime（最后阅读进度更新时间）， # 其次回退到顶层 timestamp（最后同步时间）。两者都不可用时为 0。 read_ts = (prog.get('updateTime') or pg.get('timestamp') or 0) book_progress.append({'bid': bid, 'title': title, 'author': author, 'progress': progress, 'read_ts': read_ts}) bm = gw('/book/bookmarklist', bookId=bid) if not isinstance(bm, dict): continue chaps = {c.get('chapterUid'): c.get('title', '') for c in bm.get('chapters', [])} for m in bm.get('updated', []): ct = m.get('createTime', 0) txt = (m.get('markText') or '').strip() if not is_good_quote(txt): continue # 仅收集「汇报日当天」新增的个人划线 if report_mid <= ct < report_mid + 86400: ch = chaps.get(m.get('chapterUid'), '') label = f"《{title}》" + (f" · {ch}" if ch else "") quotes.append((txt, label, 'today')) # 最近阅读的书：按 getprogress 的 timestamp 降序，最靠前的一本即「最近阅读的书」 recent_book = max(book_progress, key=lambda x: (x['read_ts'] or 0)) if book_progress else None # 当天没有新增个人划线，或当天划线质量不足 → 从「最近阅读的书」中取 1 条高质量热门划线替代 today_best = best_quote([q for q in quotes if q[2] == 'today'], min_score=4) popular_quotes = [] if (not today_best) and recent_book: bid = recent_book['bid'] title = recent_book['title'] bm = gw('/book/bestbookmarks', bookId=bid, chapterUid=0) if isinstance(bm, dict): chaps = {c.get('chapterUid'): c.get('title', '') for c in bm.get('chapters', [])} for item in bm.get('items', []): txt = (item.get('markText') or '').strip() if is_good_quote(txt): ch = chaps.get(item.get('chapterUid'), '') label = f"《{title}》" + (f" · {ch}" if ch else "") popular_quotes.append((txt, label + '（来自本书热门划线）', 'popular')) popular_best = best_quote(popular_quotes, min_score=4) if popular_best: quotes.append(popular_best) # ---------- 3. 维护打卡天数 ---------- streak_data = {} if os.path.exists(STREAK_PATH): try: with open(STREAK_PATH, encoding='utf-8') as f: streak_data = json.load(f) except Exception: pass if not streak_data: streak_data = {'firstReadDate': '', 'streak': 0, 'lastReadDate': ''} report_str = report_date.strftime('%Y-%m-%d') if streak_data.get('lastReadDate') != report_str: streak_data['streak'] = streak_data.get('streak', 0) + 1 streak_data['lastReadDate'] = report_str if not streak_data.get('firstReadDate'): streak_data['firstReadDate'] = report_str with open(STREAK_PATH, 'w', encoding='utf-8') as f: json.dump(streak_data, f, ensure_ascii=False, indent=2) streak = streak_data.get('streak', 0) # ---------- 4. 生成单篇格式：每天只保留 1 本书 + 1 句金句 ---------- min_total = today_dur // 60 today_quotes = [q for q in quotes if q[2] == 'today'] selected_quote = best_quote(today_quotes, min_score=4) or best_quote([q for q in quotes if q[2] == 'popular'], min_score=4) or best_quote(quotes) if selected_quote: quotes = [selected_quote] quote_text, quote_label, quote_source = selected_quote m = re.search(r'《([^》]+)》', quote_label) selected_book_title = m.group(1) if m else (recent_book['title'] if recent_book else '今日读书') else: quote_text = '' quote_label = '' quote_source = '' selected_book_title = recent_book['title'] if recent_book else '今日读书' selected_book_count = 1 if selected_book_title else 0 selected_quote_count = 1 if selected_quote else 0 if selected_quote: source_note = "今日个人划线" if quote_source == 'today' else "本书热门划线" quotes_section = ( "## 今日金句\n\n" f"> {quote_text} ——{quote_label}\n\n" f"- 来源：{source_note}\n" ) else: quotes_section = ( "## 今日金句\n\n" f"> 今天未获取到《{selected_book_title}》里适合摘录的金句。\n\n" "- 来源：暂无可用划线\n" ) core_points = build_deep_insights(quote_text, selected_book_title) core_view = "## 深度解读\n\n" + "\n".join(f"{i + 1}. {point}" for i, point in enumerate(core_points)) + "\n" think = ( "## 明日行动\n\n" f"继续从《{selected_book_title}》里选 1 句真正有触动的内容；如果当天没有新增高质量划线，" "就从最近阅读的这一本书里选 1 条更适合复盘的金句作为替代。\n" ) content = ( f"{quotes_section}\n" f"{core_view}" ) os.makedirs(OUT_DIR, exist_ok=True) with open(OUT_FILE, 'w', encoding='utf-8') as f: # 整篇覆盖，避免重复追加 f.write(content) # 状态供 UI 读取（quoteCount 只统计今日新增划线，而非历史兜底） status = { 'date': TODAY_STR, 'streak': streak, 'minutes': min_total, 'bookCount': selected_book_count, 'bookTitle': selected_book_title, 'quoteText': quote_text, 'quoteLabel': quote_label, 'quoteSource': '今日个人划线' if quote_source == 'today' else ('本书热门划线' if quote_source == 'popular' else '暂无可用划线'), 'quoteQualityScore': quote_quality_score(quote_text), 'corePoints': core_points, 'scheduledAt': '08:00', 'generatedAt': datetime.datetime.now(TZ8).strftime('%Y-%m-%d %H:%M:%S'), 'path': OUT_FILE, 'quoteCount': selected_quote_count, } with open(os.path.join(WORKSPACE, '.weread', 'tmp', 'last_reading_note.json'), 'w', encoding='utf-8') as f: json.dump(status, f, ensure_ascii=False, indent=2) # 额外输出一份到工作区根目录，便于未走本地服务时前端也能读取状态 with open(os.path.join(WORKSPACE, 'reading-note-status.json'), 'w', encoding='utf-8') as f: json.dump(status, f, ensure_ascii=False, indent=2) print('done. streak=', streak, 'minutes=', min_total, 'books=', len(book_progress), 'selected_book=', selected_book_title, 'quotes=', selected_quote_count)