daily-robots/daily/config.py

"""项目路径与环境变量加载。"""

from __future__ import annotations

import os
from pathlib import Path

from dotenv import load_dotenv

ROOT = Path(__file__).resolve().parent.parent
OUTPUT_DIR = ROOT / "output"
LOG_DIR = ROOT / "logs"
CACHE_DIR = ROOT / ".cache"
SNAPSHOT_FILE = CACHE_DIR / "last-report.json"

# 企微 webhook markdown.content 硬上限
WECOM_MARKDOWN_LIMIT = 4096


def wecom_chunk_bytes() -> int:
    """单条企微 markdown 消息字节上限。"""
    chunk = env_int("DAILY_WECOM_CHUNK_BYTES", -1)
    if chunk < 0:
        chunk = env_int("DAILY_WECOM_MAX_BYTES", WECOM_MARKDOWN_LIMIT)
    return min(chunk, WECOM_MARKDOWN_LIMIT)


def wecom_skill_desc_limit() -> int:
    """企微 Skills 榜单条简介建议字数。"""
    return env_int("DAILY_WECOM_SKILL_DESC_LIMIT", 56)


def wecom_news_desc_limit() -> int:
    """企微新闻摘要建议字数；在句读/词边界截断，不加省略号。"""
    return max(24, env_int("DAILY_WECOM_NEWS_DESC_LIMIT", 72))


def wecom_ai_news_tech_limit() -> int:
    """research 模式下技术类时讯条数（叠加在 DAILY_WECOM_AI_NEWS 之上）。"""
    return max(0, env_int("DAILY_WECOM_AI_NEWS_TECH", 5))


def wecom_pad_pool_size(display_limit: int) -> int:
    """Delta 补榜候选池大小（展示条数之上多取，避免去重后凑不满）。"""
    explicit = env_int("DAILY_WECOM_PAD_POOL", -1)
    if explicit > 0:
        return explicit
    return max(display_limit * 5, 50)


def full_desc_limit() -> int:
    """完整版早报摘要长度；0 表示不截断。"""
    return env_int("DAILY_FULL_DESC_LIMIT", 0)


def news_summary_limit() -> int:
    return env_int("DAILY_FULL_NEWS_SUMMARY_LIMIT", 0)


def wecom_max_bytes() -> int:
    """兼容旧配置名。"""
    return wecom_chunk_bytes()


load_dotenv(ROOT / ".env")
load_dotenv(ROOT / ".env.local", override=True)


def _clean_env_value(raw: str | None) -> str | None:
    if raw is None:
        return None
    value = raw.strip()
    if not value:
        return None
    # .env 行内注释（未加引号时 python-dotenv 不会自动去掉）
    if " #" in value:
        value = value.split(" #", 1)[0].rstrip()
    return value or None


def env(key: str, default: str | None = None) -> str | None:
    return _clean_env_value(os.getenv(key, default))


def env_int(key: str, default: int) -> int:
    raw = env(key)
    if not raw:
        return default
    try:
        return int(raw)
    except ValueError:
        return default


def env_bool(key: str, default: bool) -> bool:
    raw = env(key)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


def wecom_mode() -> str:
    raw = (env("DAILY_WECOM_MODE") or "delta").strip().lower()
    return raw if raw in {"full", "delta"} else "delta"


def news_dedup_days() -> int:
    return max(1, env_int("DAILY_NEWS_DEDUP_DAYS", 7))


def skip_push_when_silent() -> bool:
    return env_bool("DAILY_SKIP_PUSH_WHEN_SILENT", True)


def delta_baseline_fallback() -> str:
    raw = (env("DAILY_DELTA_BASELINE_FALLBACK") or "full").strip().lower()
    return raw if raw in {"full", "empty"} else "full"


def wecom_delta_pad() -> bool:
    """Delta 模式下新入榜优先，不足时用当日 Top 榜补满；补榜排除近 N 天 baseline 已出现条目。"""
    return env_bool("DAILY_WECOM_DELTA_PAD", True)


def delta_pad_lookback_days() -> int:
    """补榜时排除近 N 天 baseline 已出现过的条目（默认与异动对比窗口一致）。"""
    fallback = env_int("DAILY_DELTA_LOOKBACK_DAYS", 7)
    return max(1, env_int("DAILY_DELTA_PAD_LOOKBACK_DAYS", fallback))


def force_push() -> bool:
    return env_bool("DAILY_FORCE_PUSH", False)


def schedule_timezone_name() -> str:
    return (env("DAILY_SCHEDULE_TZ") or "Asia/Shanghai").strip() or "Asia/Shanghai"


def schedule_generate_at() -> str:
    return (env("DAILY_SCHEDULE_GENERATE_AT") or "08:50").strip() or "08:50"


def schedule_push_at() -> str:
    return (env("DAILY_SCHEDULE_PUSH_AT") or "09:00").strip() or "09:00"


def board_dedup_days() -> int:
    return max(1, env_int("DAILY_BOARD_DEDUP_DAYS", 7))


def board_pool_size() -> int:
    fallback = env_int("DAILY_WECOM_SKILL_POOL", 400)
    return max(1, env_int("DAILY_BOARD_POOL_SIZE", max(200, fallback)))


def featured_dedup_days() -> int:
    return max(1, env_int("DAILY_FEATURED_DEDUP_DAYS", 30))


def theme_ban_days() -> int:
    return max(1, env_int("DAILY_THEME_BAN_DAYS", 7))


def narrative_axis_days() -> int:
    return max(1, env_int("DAILY_NARRATIVE_AXIS_DAYS", 3))


def news_backfill_enabled() -> bool:
    return env_bool("DAILY_NEWS_BACKFILL", False)


def workday_only() -> bool:
    """仅工作日生成/推送；法定节假日与周末跳过（调休补班日照常）。"""
    return env_bool("DAILY_WORKDAY_ONLY", True)