feat: 新增企微 Delta 模式相关配置读取函数
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -3,13 +3,11 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import os
|
import os
|
||||||
import sys
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
from dotenv import load_dotenv
|
from dotenv import load_dotenv
|
||||||
|
|
||||||
ROOT = Path(__file__).resolve().parent.parent
|
ROOT = Path(__file__).resolve().parent.parent
|
||||||
BOT_DIR = ROOT / "bot"
|
|
||||||
OUTPUT_DIR = ROOT / "output"
|
OUTPUT_DIR = ROOT / "output"
|
||||||
LOG_DIR = ROOT / "logs"
|
LOG_DIR = ROOT / "logs"
|
||||||
CACHE_DIR = ROOT / ".cache"
|
CACHE_DIR = ROOT / ".cache"
|
||||||
@@ -45,16 +43,11 @@ def wecom_max_bytes() -> int:
|
|||||||
"""兼容旧配置名。"""
|
"""兼容旧配置名。"""
|
||||||
return wecom_chunk_bytes()
|
return wecom_chunk_bytes()
|
||||||
|
|
||||||
|
|
||||||
load_dotenv(ROOT / ".env")
|
load_dotenv(ROOT / ".env")
|
||||||
load_dotenv(ROOT / ".env.local", override=True)
|
load_dotenv(ROOT / ".env.local", override=True)
|
||||||
|
|
||||||
|
|
||||||
def ensure_bot_on_path() -> None:
|
|
||||||
bot = str(BOT_DIR)
|
|
||||||
if bot not in sys.path:
|
|
||||||
sys.path.insert(0, bot)
|
|
||||||
|
|
||||||
|
|
||||||
def _clean_env_value(raw: str | None) -> str | None:
|
def _clean_env_value(raw: str | None) -> str | None:
|
||||||
if raw is None:
|
if raw is None:
|
||||||
return None
|
return None
|
||||||
@@ -79,3 +72,32 @@ def env_int(key: str, default: int) -> int:
|
|||||||
return int(raw)
|
return int(raw)
|
||||||
except ValueError:
|
except ValueError:
|
||||||
return default
|
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 force_push() -> bool:
|
||||||
|
return env_bool("DAILY_FORCE_PUSH", False)
|
||||||
|
|||||||
42
tests/test_wecom_delta.py
Normal file
42
tests/test_wecom_delta.py
Normal file
@@ -0,0 +1,42 @@
|
|||||||
|
"""Tests for WeCom delta mode."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
|
import unittest
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
from daily.config import (
|
||||||
|
delta_baseline_fallback,
|
||||||
|
env_bool,
|
||||||
|
force_push,
|
||||||
|
news_dedup_days,
|
||||||
|
skip_push_when_silent,
|
||||||
|
wecom_mode,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class ConfigHelpersTests(unittest.TestCase):
|
||||||
|
def test_wecom_mode_defaults_delta(self):
|
||||||
|
with patch.dict(os.environ, {}, clear=True):
|
||||||
|
self.assertEqual(wecom_mode(), "delta")
|
||||||
|
|
||||||
|
def test_wecom_mode_full(self):
|
||||||
|
with patch.dict(os.environ, {"DAILY_WECOM_MODE": "full"}, clear=True):
|
||||||
|
self.assertEqual(wecom_mode(), "full")
|
||||||
|
|
||||||
|
def test_env_bool_truthy(self):
|
||||||
|
with patch.dict(os.environ, {"DAILY_FORCE_PUSH": "1"}, clear=True):
|
||||||
|
self.assertTrue(env_bool("DAILY_FORCE_PUSH", False))
|
||||||
|
|
||||||
|
def test_skip_push_when_silent_default(self):
|
||||||
|
with patch.dict(os.environ, {}, clear=True):
|
||||||
|
self.assertTrue(skip_push_when_silent())
|
||||||
|
|
||||||
|
def test_news_dedup_days_default(self):
|
||||||
|
with patch.dict(os.environ, {}, clear=True):
|
||||||
|
self.assertEqual(news_dedup_days(), 7)
|
||||||
|
|
||||||
|
def test_delta_baseline_fallback_default(self):
|
||||||
|
with patch.dict(os.environ, {}, clear=True):
|
||||||
|
self.assertEqual(delta_baseline_fallback(), "full")
|
||||||
Reference in New Issue
Block a user