你发送的字
{{ formatInt(sentChars) }}
@@ -321,40 +321,6 @@ const getLabelStyle = (code) => {
@apply w-8 h-8 rounded-lg border border-[#00000010] flex items-center justify-center flex-shrink-0;
}
-/* 气泡 - 左侧 */
-.bubble-left {
- @apply relative max-w-[85%] bg-white shadow-sm rounded-xl px-3 py-2;
-}
-.bubble-left::before {
- content: '';
- position: absolute;
- left: -6px;
- bottom: 8px;
- width: 0;
- height: 0;
- border-top: 6px solid transparent;
- border-bottom: 6px solid transparent;
- border-right: 6px solid #fff;
- filter: drop-shadow(-1px 0 0 rgba(0,0,0,0.05));
-}
-
-/* 气泡 - 右侧 */
-.bubble-right {
- @apply relative max-w-[85%] bg-[#95EC69] shadow-sm rounded-xl px-3 py-2;
-}
-.bubble-right::after {
- content: '';
- position: absolute;
- right: -6px;
- bottom: 8px;
- width: 0;
- height: 0;
- border-top: 6px solid transparent;
- border-bottom: 6px solid transparent;
- border-left: 6px solid #95EC69;
- filter: drop-shadow(1px 0 0 rgba(0,0,0,0.05));
-}
-
/* 键盘外框 */
.keyboard-outer {
@apply mt-3 rounded-2xl p-1;
diff --git a/src/wechat_decrypt_tool/wrapped/cards/card_00_global_overview.py b/src/wechat_decrypt_tool/wrapped/cards/card_00_global_overview.py
index 04c31a0..c671dfe 100644
--- a/src/wechat_decrypt_tool/wrapped/cards/card_00_global_overview.py
+++ b/src/wechat_decrypt_tool/wrapped/cards/card_00_global_overview.py
@@ -6,7 +6,7 @@ import sqlite3
import time
from collections import Counter
from dataclasses import dataclass
-from datetime import datetime
+from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Optional
@@ -28,6 +28,7 @@ logger = get_logger(__name__)
_MD5_HEX_RE = re.compile(r"(?i)[0-9a-f]{32}")
+_EMOJI_CHAR_RE = re.compile(r"[\U0001F300-\U0001FAFF\u2600-\u27BF]")
# Best-effort heuristics for "new friends added" detection: WeChat system messages vary by version.
_ADDED_FRIEND_PATTERNS: tuple[str, ...] = (
"你已添加了",
@@ -60,6 +61,13 @@ def _year_range_epoch_seconds(year: int) -> tuple[int, int]:
return start, end
+def _days_in_year(year: int) -> int:
+ try:
+ return int((datetime(int(year) + 1, 1, 1) - datetime(int(year), 1, 1)).days)
+ except Exception:
+ return 365
+
+
def _list_message_tables(conn: sqlite3.Connection) -> list[str]:
try:
rows = conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()
@@ -76,6 +84,494 @@ def _list_message_tables(conn: sqlite3.Connection) -> list[str]:
return names
+def _accumulate_db_daily_counts(
+ *,
+ db_path: Path,
+ start_ts: int,
+ end_ts: int,
+ counts: list[int],
+ sender_username: str | None = None,
+) -> int:
+ """Accumulate per-day message counts from one message shard DB into counts list.
+
+ Returns the number of messages counted.
+ """
+
+ if not db_path.exists():
+ return 0
+
+ conn: sqlite3.Connection | None = None
+ try:
+ conn = sqlite3.connect(str(db_path))
+
+ tables = _list_message_tables(conn)
+ if not tables:
+ return 0
+
+ # Convert millisecond timestamps defensively.
+ # The expression yields epoch seconds as INTEGER.
+ ts_expr = (
+ "CASE WHEN create_time > 1000000000000 THEN CAST(create_time/1000 AS INTEGER) ELSE create_time END"
+ )
+
+ # Optional sender filter (best-effort). When provided, we only count
+ # messages whose `real_sender_id` maps to `sender_username`.
+ sender_rowid: int | None = None
+ if sender_username and str(sender_username).strip():
+ try:
+ r = conn.execute(
+ "SELECT rowid FROM Name2Id WHERE user_name = ? LIMIT 1",
+ (str(sender_username).strip(),),
+ ).fetchone()
+ if r is not None and r[0] is not None:
+ sender_rowid = int(r[0])
+ except Exception:
+ sender_rowid = None
+
+ counted = 0
+ for table_name in tables:
+ qt = _quote_ident(table_name)
+ sender_where = ""
+ params: tuple[Any, ...]
+ if sender_rowid is not None:
+ sender_where = " AND real_sender_id = ?"
+ params = (start_ts, end_ts, sender_rowid)
+ else:
+ params = (start_ts, end_ts)
+
+ sql = (
+ "SELECT CAST(strftime('%j', datetime(ts, 'unixepoch', 'localtime')) AS INTEGER) - 1 AS doy, "
+ "COUNT(1) AS cnt "
+ "FROM ("
+ f" SELECT {ts_expr} AS ts"
+ f" FROM {qt}"
+ f" WHERE {ts_expr} >= ? AND {ts_expr} < ?{sender_where}"
+ ") sub "
+ "GROUP BY doy"
+ )
+
+ try:
+ rows = conn.execute(sql, params).fetchall()
+ except Exception:
+ continue
+
+ for doy, cnt in rows:
+ try:
+ d = int(doy if doy is not None else -1)
+ c = int(cnt or 0)
+ except Exception:
+ continue
+ if c <= 0 or d < 0 or d >= len(counts):
+ continue
+ counts[d] += c
+ counted += c
+
+ return counted
+ finally:
+ try:
+ if conn is not None:
+ conn.close()
+ except Exception:
+ pass
+
+
+def compute_annual_daily_counts(*, account_dir: Path, year: int, sender_username: str | None = None) -> list[int]:
+ """Compute per-day message counts for the given year.
+
+ The output is a 0-indexed day-of-year list (length 365/366). Counts default to
+ "messages sent by me" when sender_username is provided.
+ """
+
+ start_ts, end_ts = _year_range_epoch_seconds(year)
+ days = _days_in_year(year)
+ counts: list[int] = [0 for _ in range(days)]
+
+ sender = str(sender_username or "").strip()
+
+ # Prefer using our unified search index if available; it's much faster than scanning all msg tables.
+ index_path = get_chat_search_index_db_path(account_dir)
+ if index_path.exists():
+ conn = sqlite3.connect(str(index_path))
+ try:
+ has_fts = (
+ conn.execute(
+ "SELECT 1 FROM sqlite_master WHERE type='table' AND name='message_fts' LIMIT 1"
+ ).fetchone()
+ is not None
+ )
+ if has_fts:
+ # Convert millisecond timestamps defensively (some datasets store ms).
+ ts_expr = (
+ "CASE "
+ "WHEN CAST(create_time AS INTEGER) > 1000000000000 "
+ "THEN CAST(CAST(create_time AS INTEGER)/1000 AS INTEGER) "
+ "ELSE CAST(create_time AS INTEGER) "
+ "END"
+ )
+ sender_clause = ""
+ if sender:
+ sender_clause = " AND sender_username = ?"
+
+ sql = (
+ "SELECT "
+ "CAST(strftime('%j', datetime(ts, 'unixepoch', 'localtime')) AS INTEGER) - 1 AS doy, "
+ "COUNT(1) AS cnt "
+ "FROM ("
+ f" SELECT {ts_expr} AS ts"
+ " FROM message_fts"
+ f" WHERE {ts_expr} >= ? AND {ts_expr} < ?"
+ " AND db_stem NOT LIKE 'biz_message%'"
+ f"{sender_clause}"
+ ") sub "
+ "GROUP BY doy"
+ )
+
+ t0 = time.time()
+ try:
+ params: tuple[Any, ...] = (start_ts, end_ts)
+ if sender:
+ params = (start_ts, end_ts, sender)
+ rows = conn.execute(sql, params).fetchall()
+ except Exception:
+ rows = []
+
+ total = 0
+ for r in rows:
+ if not r:
+ continue
+ try:
+ doy = int(r[0] if r[0] is not None else -1)
+ cnt = int(r[1] or 0)
+ except Exception:
+ continue
+ if cnt <= 0 or doy < 0 or doy >= days:
+ continue
+ counts[doy] += cnt
+ total += cnt
+
+ logger.info(
+ "Wrapped annual heatmap computed (search index): account=%s year=%s total=%s sender=%s db=%s elapsed=%.2fs",
+ str(account_dir.name or "").strip(),
+ year,
+ total,
+ sender or "*",
+ str(index_path.name),
+ time.time() - t0,
+ )
+
+ return counts
+ finally:
+ try:
+ conn.close()
+ except Exception:
+ pass
+
+ db_paths = _iter_message_db_paths(account_dir)
+ # Default: exclude official/biz shards (biz_message*.db) to reduce noise.
+ db_paths = [p for p in db_paths if not p.name.lower().startswith("biz_message")]
+ my_wxid = str(account_dir.name or "").strip()
+ t0 = time.time()
+ total = 0
+ for db_path in db_paths:
+ total += _accumulate_db_daily_counts(
+ db_path=db_path,
+ start_ts=start_ts,
+ end_ts=end_ts,
+ counts=counts,
+ sender_username=sender or None,
+ )
+
+ logger.info(
+ "Wrapped annual heatmap computed: account=%s year=%s total=%s sender=%s dbs=%s elapsed=%.2fs",
+ my_wxid,
+ year,
+ total,
+ sender or "*",
+ len(db_paths),
+ time.time() - t0,
+ )
+
+ return counts
+
+
+def _ymd_from_doy(*, year: int, doy: int) -> str:
+ try:
+ dt = datetime(int(year), 1, 1) + timedelta(days=int(doy))
+ except Exception:
+ return ""
+ return dt.strftime("%Y-%m-%d")
+
+
+def _best_doy_by_max(values: list[int]) -> tuple[int, int] | tuple[None, int]:
+ best_doy: int | None = None
+ best = 0
+ for i, v in enumerate(values):
+ try:
+ n = int(v or 0)
+ except Exception:
+ n = 0
+ if n > best or (n == best and best_doy is not None and i < best_doy):
+ best = n
+ best_doy = i
+ return best_doy, best
+
+
+def compute_annual_heatmap_highlights(
+ *,
+ account_dir: Path,
+ year: int,
+ sender_username: str,
+ sent_daily_counts: list[int],
+) -> list[dict[str, Any]]:
+ """Compute special day highlights for the annual calendar heatmap (best-effort).
+
+ We prefer the unified search index when available; fallback mode returns the subset
+ that can be derived from `sent_daily_counts` without scanning all message content.
+ """
+
+ days = int(len(sent_daily_counts) or _days_in_year(year))
+ out: list[dict[str, Any]] = []
+
+ def add_highlight(
+ *,
+ key: str,
+ label: str,
+ doy: int,
+ value: int | None = None,
+ value_label: str = "",
+ date: str = "",
+ ) -> None:
+ try:
+ d = int(doy)
+ except Exception:
+ return
+ if d < 0 or d >= days:
+ return
+ ymd = str(date or "") or _ymd_from_doy(year=int(year), doy=d)
+ if not ymd:
+ return
+ obj: dict[str, Any] = {
+ "key": str(key),
+ "label": str(label),
+ "doy": int(d),
+ "date": ymd,
+ }
+ if value is not None:
+ try:
+ obj["value"] = int(value)
+ except Exception:
+ pass
+ if value_label:
+ obj["valueLabel"] = str(value_label)
+ out.append(obj)
+
+ # 3. Sent messages max day (always available from sent daily counts)
+ best_doy, best_val = _best_doy_by_max(sent_daily_counts or [])
+ if best_doy is not None and best_val > 0:
+ add_highlight(key="sent_messages_max", label="发送消息条数最多的一天", doy=int(best_doy), value=int(best_val), value_label=f"{int(best_val)} 条")
+
+ sender = str(sender_username or "").strip()
+ if not sender:
+ return out
+
+ index_path = get_chat_search_index_db_path(account_dir)
+ if not index_path.exists():
+ return out
+
+ start_ts, end_ts = _year_range_epoch_seconds(year)
+
+ conn = sqlite3.connect(str(index_path))
+ try:
+ has_fts = (
+ conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='message_fts' LIMIT 1").fetchone()
+ is not None
+ )
+ if not has_fts:
+ return out
+
+ # Convert millisecond timestamps defensively (some datasets store ms).
+ ts_expr = (
+ "CASE "
+ "WHEN CAST(create_time AS INTEGER) > 1000000000000 "
+ "THEN CAST(CAST(create_time AS INTEGER)/1000 AS INTEGER) "
+ "ELSE CAST(create_time AS INTEGER) "
+ "END"
+ )
+
+ base_where = f"{ts_expr} >= ? AND {ts_expr} < ? AND db_stem NOT LIKE 'biz_message%'"
+ base_params: tuple[Any, ...] = (start_ts, end_ts)
+
+ def fetch_best_doy_value(sql: str, params: tuple[Any, ...]) -> tuple[int, int] | None:
+ try:
+ r = conn.execute(sql, params).fetchone()
+ except Exception:
+ r = None
+ if not r:
+ return None
+ try:
+ doy = int(r[0] if r[0] is not None else -1)
+ val = int(r[1] or 0)
+ except Exception:
+ return None
+ if val <= 0 or doy < 0 or doy >= days:
+ return None
+ return doy, val
+
+ # 1. Sent chars max day (text messages only, non-whitespace approximation)
+ char_expr = (
+ "length(replace(replace(replace(replace(coalesce(text,''),' ',''), char(10), ''), char(13), ''), char(9), ''))"
+ )
+ sql_sent_chars = (
+ "SELECT doy, chars FROM ("
+ " SELECT CAST(strftime('%j', datetime(ts, 'unixepoch', 'localtime')) AS INTEGER) - 1 AS doy, "
+ f" SUM({char_expr}) AS chars "
+ " FROM ("
+ f" SELECT {ts_expr} AS ts, text "
+ " FROM message_fts "
+ f" WHERE {base_where} AND sender_username = ? AND CAST(local_type AS INTEGER) = 1"
+ " ) sub "
+ " GROUP BY doy"
+ ") t ORDER BY chars DESC, doy ASC LIMIT 1"
+ )
+ r_sent_chars = fetch_best_doy_value(sql_sent_chars, base_params + (sender,))
+ if r_sent_chars is not None:
+ doy, v = r_sent_chars
+ add_highlight(key="sent_chars_max", label="发送字最多的一天", doy=doy, value=v, value_label=f"{v} 字")
+
+ # 2. Received chars max day (text messages only)
+ sql_recv_chars = (
+ "SELECT doy, chars FROM ("
+ " SELECT CAST(strftime('%j', datetime(ts, 'unixepoch', 'localtime')) AS INTEGER) - 1 AS doy, "
+ f" SUM({char_expr}) AS chars "
+ " FROM ("
+ f" SELECT {ts_expr} AS ts, text "
+ " FROM message_fts "
+ f" WHERE {base_where} AND COALESCE(sender_username,'') != ? AND CAST(local_type AS INTEGER) = 1"
+ " ) sub "
+ " GROUP BY doy"
+ ") t ORDER BY chars DESC, doy ASC LIMIT 1"
+ )
+ r_recv_chars = fetch_best_doy_value(sql_recv_chars, base_params + (sender,))
+ if r_recv_chars is not None:
+ doy, v = r_recv_chars
+ add_highlight(key="received_chars_max", label="接收字最多的一天", doy=doy, value=v, value_label=f"{v} 字")
+
+ # 4. Received message count max day (exclude system messages)
+ sql_recv_msgs = (
+ "SELECT doy, cnt FROM ("
+ " SELECT CAST(strftime('%j', datetime(ts, 'unixepoch', 'localtime')) AS INTEGER) - 1 AS doy, "
+ " COUNT(1) AS cnt "
+ " FROM ("
+ f" SELECT {ts_expr} AS ts "
+ " FROM message_fts "
+ f" WHERE {base_where} AND COALESCE(sender_username,'') != ? AND CAST(local_type AS INTEGER) != 10000"
+ " ) sub "
+ " GROUP BY doy"
+ ") t ORDER BY cnt DESC, doy ASC LIMIT 1"
+ )
+ r_recv_msgs = fetch_best_doy_value(sql_recv_msgs, base_params + (sender,))
+ if r_recv_msgs is not None:
+ doy, v = r_recv_msgs
+ add_highlight(key="received_messages_max", label="接收消息条数最多的一天", doy=doy, value=v, value_label=f"{v} 条")
+
+ # 5. Added friends max day (best-effort via system message patterns, exclude official/chatrooms)
+ added_like_patterns = [f"%{p}%" for p in _ADDED_FRIEND_PATTERNS if str(p or "").strip()]
+ if added_like_patterns:
+ cond_added = " OR ".join(["text LIKE ?"] * len(added_like_patterns))
+ sql_added = (
+ "SELECT doy, cnt FROM ("
+ " SELECT CAST(strftime('%j', datetime(ts, 'unixepoch', 'localtime')) AS INTEGER) - 1 AS doy, "
+ " COUNT(DISTINCT username) AS cnt "
+ " FROM ("
+ f" SELECT {ts_expr} AS ts, username, text "
+ " FROM message_fts "
+ f" WHERE {base_where} "
+ " AND CAST(local_type AS INTEGER) = 10000 "
+ " AND COALESCE(is_official, 0) = 0 "
+ " AND username NOT LIKE '%@chatroom' "
+ f" AND ({cond_added})"
+ " ) sub "
+ " GROUP BY doy"
+ ") t ORDER BY cnt DESC, doy ASC LIMIT 1"
+ )
+ params_added: tuple[Any, ...] = base_params + tuple(added_like_patterns)
+ r_added = fetch_best_doy_value(sql_added, params_added)
+ if r_added is not None:
+ doy, v = r_added
+ add_highlight(key="added_friends_max", label="加好友最多的一天", doy=doy, value=v, value_label=f"{v} 位")
+
+ # 6. Sticker/emoji messages max day (WeChat local_type=47)
+ sql_emoji_msgs = (
+ "SELECT doy, cnt FROM ("
+ " SELECT CAST(strftime('%j', datetime(ts, 'unixepoch', 'localtime')) AS INTEGER) - 1 AS doy, "
+ " COUNT(1) AS cnt "
+ " FROM ("
+ f" SELECT {ts_expr} AS ts "
+ " FROM message_fts "
+ f" WHERE {base_where} AND sender_username = ? AND CAST(local_type AS INTEGER) = 47"
+ " ) sub "
+ " GROUP BY doy"
+ ") t ORDER BY cnt DESC, doy ASC LIMIT 1"
+ )
+ r_emoji_msgs = fetch_best_doy_value(sql_emoji_msgs, base_params + (sender,))
+ if r_emoji_msgs is not None:
+ doy, v = r_emoji_msgs
+ add_highlight(key="sticker_messages_max", label="发表情包最多的一天", doy=doy, value=v, value_label=f"{v} 条")
+
+ # 7. Unicode emoji chars max day (best-effort; count emoji codepoints in sent text)
+ emoji_counts: list[int] = [0 for _ in range(days)]
+ sql_emoji_text = (
+ "SELECT doy, text FROM ("
+ " SELECT "
+ " CAST(strftime('%j', datetime(ts, 'unixepoch', 'localtime')) AS INTEGER) - 1 AS doy, "
+ " text "
+ " FROM ("
+ f" SELECT {ts_expr} AS ts, text "
+ " FROM message_fts "
+ f" WHERE {base_where} AND sender_username = ? AND CAST(local_type AS INTEGER) = 1"
+ " ) sub2"
+ ") sub "
+ "WHERE text IS NOT NULL AND text != ''"
+ )
+ try:
+ cur = conn.execute(sql_emoji_text, base_params + (sender,))
+ except Exception:
+ cur = None
+ if cur is not None:
+ for r in cur:
+ try:
+ doy = int(r[0] if r and r[0] is not None else -1)
+ except Exception:
+ continue
+ if doy < 0 or doy >= days:
+ continue
+ try:
+ txt = str(r[1] or "")
+ except Exception:
+ txt = ""
+ if not txt:
+ continue
+ emoji_counts[doy] += len(_EMOJI_CHAR_RE.findall(txt))
+
+ best_emoji_doy, best_emoji = _best_doy_by_max(emoji_counts)
+ if best_emoji_doy is not None and best_emoji > 0:
+ add_highlight(
+ key="emoji_chars_max",
+ label="发emoji最多的一天",
+ doy=int(best_emoji_doy),
+ value=int(best_emoji),
+ value_label=f"{int(best_emoji)} 个",
+ )
+
+ finally:
+ try:
+ conn.close()
+ except Exception:
+ pass
+
+ return out
+
+
def _list_session_usernames(session_db_path: Path) -> list[str]:
if not session_db_path.exists():
return []
@@ -769,6 +1265,22 @@ def build_card_00_global_overview(
"action": "你还在微信里发送消息",
}
+ daily_counts = compute_annual_daily_counts(account_dir=account_dir, year=year, sender_username=sender)
+ annual_highlights = compute_annual_heatmap_highlights(
+ account_dir=account_dir,
+ year=year,
+ sender_username=sender,
+ sent_daily_counts=daily_counts,
+ )
+ annual_heatmap = {
+ "year": int(year),
+ "startDate": f"{int(year)}-01-01",
+ "endDate": f"{int(year)}-12-31",
+ "days": int(len(daily_counts)),
+ "dailyCounts": daily_counts,
+ "highlights": annual_highlights,
+ }
+
lines: list[str] = []
if heatmap.total_messages > 0:
lines.append(f"今年以来，你在微信里发送了 {heatmap.total_messages:,} 条消息，平均每天 {messages_per_day:.1f} 条。")
@@ -816,6 +1328,8 @@ def build_card_00_global_overview(
"totalMessages": int(heatmap.total_messages),
"activeDays": int(stats.active_days),
"addedFriends": int(stats.added_friends),
+ "sentMediaCount": int(stats.kind_counts.get("image", 0) + stats.kind_counts.get("video", 0)),
+ "sentStickerCount": int(stats.kind_counts.get("emoji", 0)),
"messagesPerDay": messages_per_day,
"mostActiveHour": most_active_hour,
"mostActiveWeekday": most_active_weekday,
@@ -823,6 +1337,7 @@ def build_card_00_global_overview(
"topContact": top_contact_obj,
"topGroup": top_group_obj,
"topKind": top_kind,
+ "annualHeatmap": annual_heatmap,
"topPhrase": {"phrase": stats.top_phrase[0], "count": int(stats.top_phrase[1])} if stats.top_phrase else None,
"topEmoji": {"emoji": stats.top_emoji[0], "count": int(stats.top_emoji[1])} if stats.top_emoji else None,
"highlight": highlight,