84 lines
2.5 KiB
Python
84 lines
2.5 KiB
Python
"""企微展示历史：读写 data.wecom_shown_keys，与 movement_baseline 严格分离。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from daily.config import OUTPUT_DIR, board_dedup_days
|
||
from daily.delta import RECENT_BOARD_KEYS, skill_id
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
BOARD_KEYS = RECENT_BOARD_KEYS
|
||
|
||
|
||
def extract_shown_keys(board: str, items: list[dict[str, Any]]) -> list[str]:
|
||
"""从最终展示 items 抽取稳定 identity key。"""
|
||
keys: list[str] = []
|
||
seen: set[str] = set()
|
||
for item in items:
|
||
if board.startswith("skills_"):
|
||
key = skill_id(item)
|
||
else:
|
||
key = str(item.get("repo") or "")
|
||
if not key or key in seen:
|
||
continue
|
||
seen.add(key)
|
||
keys.append(key)
|
||
return keys
|
||
|
||
|
||
def load_recent_shown_keys(
|
||
date_str: str,
|
||
*,
|
||
lookback_days: int | None = None,
|
||
) -> dict[str, set[str]]:
|
||
"""近 N 日 data.wecom_shown_keys 并集（不含当日）。缺省或读失败视为空集。"""
|
||
empty = {board: set() for board in BOARD_KEYS}
|
||
try:
|
||
dt = datetime.strptime(date_str, "%Y-%m-%d")
|
||
except ValueError:
|
||
return empty
|
||
|
||
days = lookback_days if lookback_days is not None else board_dedup_days()
|
||
out: dict[str, set[str]] = {board: set() for board in BOARD_KEYS}
|
||
|
||
for day_offset in range(1, days + 1):
|
||
prev_date = (dt - timedelta(days=day_offset)).strftime("%Y-%m-%d")
|
||
path = OUTPUT_DIR / f"{prev_date}.data.json"
|
||
if not path.exists():
|
||
continue
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
except (OSError, ValueError) as exc:
|
||
logger.warning("读取 wecom_shown_keys %s 失败：%s", path, exc)
|
||
continue
|
||
data = payload.get("data")
|
||
if not isinstance(data, dict):
|
||
continue
|
||
shown = data.get("wecom_shown_keys")
|
||
if not isinstance(shown, dict):
|
||
continue
|
||
for board in BOARD_KEYS:
|
||
keys = shown.get(board) or []
|
||
if not isinstance(keys, list):
|
||
continue
|
||
out[board].update(str(k) for k in keys if k)
|
||
return out
|
||
|
||
|
||
def merge_wecom_shown_into_data(
|
||
data: dict[str, Any],
|
||
shown: dict[str, list[str]],
|
||
) -> dict[str, Any]:
|
||
"""写入 wecom_shown_keys，不修改 movement_baseline。"""
|
||
merged = dict(data)
|
||
merged["wecom_shown_keys"] = {
|
||
board: list(keys) for board, keys in shown.items()
|
||
}
|
||
return merged
|