Ubuntu
360 lines
11 KiB
Python
360 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
自动定时改写发布调度器
|
|
|
|
由 cron 每分钟触发一次，读取 auto_config.yaml，判断当前时段是否需要发布。
|
|
如果当前时间落入某个 schedule slot 且该 slot 今日尚未执行，则随机从不同二级栏目挑选草稿，
|
|
调用 pipeline.run_pipeline() 改写并发布。
|
|
|
|
用法:
|
|
python3 auto_publish.py # 由 cron 调用
|
|
python3 auto_publish.py --dry-run # 模拟运行，不实际发布
|
|
python3 auto_publish.py --now # 忽略时间判断，立即执行一次
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import json
|
|
import random
|
|
import logging
|
|
from datetime import datetime, date
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
# ---- 路径设置 ----
|
|
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
CONFIG_PATH = os.path.join(SCRIPT_DIR, "auto_config.yaml")
|
|
STATE_PATH = os.path.join(SCRIPT_DIR, "auto_state.json")
|
|
LOG_PATH = os.path.join(SCRIPT_DIR, "logs", "auto_publish.log")
|
|
|
|
sys.path.insert(0, SCRIPT_DIR)
|
|
|
|
from cms_ops import _get_conn, find_draft_by_column
|
|
from pipeline import run_pipeline
|
|
|
|
# ---- 日志 ----
|
|
os.makedirs(os.path.dirname(LOG_PATH), exist_ok=True)
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(message)s",
|
|
handlers=[
|
|
logging.FileHandler(LOG_PATH),
|
|
logging.StreamHandler(sys.stdout),
|
|
],
|
|
)
|
|
log = logging.getLogger("auto_publish")
|
|
|
|
|
|
# ============================================================
|
|
# 配置加载
|
|
# ============================================================
|
|
|
|
def load_config():
|
|
"""加载 YAML 配置"""
|
|
if not os.path.exists(CONFIG_PATH):
|
|
log.error(f"配置文件不存在: {CONFIG_PATH}")
|
|
sys.exit(1)
|
|
with open(CONFIG_PATH, "r") as f:
|
|
cfg = yaml.safe_load(f)
|
|
return cfg
|
|
|
|
|
|
# ============================================================
|
|
# 状态管理
|
|
# ============================================================
|
|
|
|
def load_state():
|
|
"""加载状态文件"""
|
|
if os.path.exists(STATE_PATH):
|
|
with open(STATE_PATH, "r") as f:
|
|
return json.load(f)
|
|
return {}
|
|
|
|
|
|
def save_state(state):
|
|
"""保存状态文件"""
|
|
with open(STATE_PATH, "w") as f:
|
|
json.dump(state, f, indent=2, ensure_ascii=False)
|
|
|
|
|
|
def today_str() -> str:
|
|
"""返回当天日期字符串，基于配置的时区"""
|
|
try:
|
|
cfg = load_config()
|
|
tz_name = cfg.get("timezone", "Asia/Shanghai")
|
|
from zoneinfo import ZoneInfo
|
|
tz = ZoneInfo(tz_name)
|
|
return datetime.now(tz).strftime("%Y-%m-%d")
|
|
except Exception:
|
|
return date.today().isoformat()
|
|
|
|
|
|
# ============================================================
|
|
# 文章选取
|
|
# ============================================================
|
|
|
|
def get_available_columns(cfg) -> list:
|
|
"""
|
|
返回有草稿的栏目列表。
|
|
每项: {"name": "尿检", "column_id": 19, "draft_count": 1703}
|
|
"""
|
|
conn = _get_conn()
|
|
cur = conn.cursor()
|
|
available = []
|
|
for col_name in cfg.get("columns", []):
|
|
cur.execute(
|
|
"""SELECT c.id, COUNT(a.id) as cnt
|
|
FROM columns c
|
|
LEFT JOIN articles a ON a.column_id = c.id AND a.status = 'draft'
|
|
WHERE c.name = ? AND c.parent_id IS NOT NULL
|
|
GROUP BY c.id""",
|
|
(col_name,),
|
|
)
|
|
row = cur.fetchone()
|
|
if row and row[1] > 0:
|
|
available.append({
|
|
"name": col_name,
|
|
"column_id": row[0],
|
|
"draft_count": row[1],
|
|
})
|
|
conn.close()
|
|
return available
|
|
|
|
|
|
def pick_articles(cfg, count: int, exclude_ids: set, used_columns_today: set = None, strategy_hint: str = None) -> list:
|
|
"""
|
|
从不同栏目中随机挑选 count 篇草稿，优先选择当天未用过的栏目。
|
|
|
|
Args:
|
|
cfg: 配置字典
|
|
count: 需要挑选的文章数
|
|
exclude_ids: 已发布过的文章 ID 集合（避免重复）
|
|
used_columns_today: 当天已用过的栏目名集合（避免重复栏目）
|
|
strategy_hint: 当前 strategy 提示（用于日志）
|
|
|
|
Returns:
|
|
[{"id": 123, "title": "...", "column": "尿检"}, ...]
|
|
"""
|
|
strategy = strategy_hint or cfg.get("pick_strategy", "random")
|
|
used_columns_today = used_columns_today or set()
|
|
available_cols = get_available_columns(cfg)
|
|
|
|
if not available_cols:
|
|
log.warning("所有栏目均无草稿可发")
|
|
return []
|
|
|
|
# 把栏目分成"未用过"和"已用过"两组，未用过的优先
|
|
fresh_cols = [c for c in available_cols if c["name"] not in used_columns_today]
|
|
used_cols = [c for c in available_cols if c["name"] in used_columns_today]
|
|
|
|
conn = _get_conn()
|
|
cur = conn.cursor()
|
|
picked = []
|
|
picked_columns = set()
|
|
|
|
# 按策略排列栏目顺序（未用过优先，已用过兜底）
|
|
if strategy == "random":
|
|
random.shuffle(fresh_cols)
|
|
random.shuffle(used_cols)
|
|
else:
|
|
state = load_state()
|
|
last_idx = state.get("last_column_index", 0)
|
|
fresh_cols = fresh_cols[last_idx:] + fresh_cols[:last_idx]
|
|
used_cols = used_cols[last_idx:] + used_cols[:last_idx]
|
|
|
|
shuffled = fresh_cols + used_cols
|
|
|
|
for _ in range(count):
|
|
found = False
|
|
for col in shuffled:
|
|
if col["name"] in picked_columns:
|
|
continue # 本次挑选内不重复栏目
|
|
cur.execute(
|
|
"""SELECT id, title FROM articles
|
|
WHERE column_id = ? AND status = 'draft'
|
|
AND id NOT IN ({})
|
|
ORDER BY created_at DESC LIMIT 1""".format(
|
|
",".join("?" * len(exclude_ids)) if exclude_ids else "0"
|
|
),
|
|
[col["column_id"]] + (list(exclude_ids) if exclude_ids else []),
|
|
)
|
|
row = cur.fetchone()
|
|
if row:
|
|
picked.append({
|
|
"id": row[0],
|
|
"title": row[1],
|
|
"column": col["name"],
|
|
})
|
|
picked_columns.add(col["name"])
|
|
exclude_ids.add(row[0])
|
|
found = True
|
|
break
|
|
|
|
if not found:
|
|
# 所有栏目都试过了，放宽"不同栏目"限制
|
|
for col in shuffled:
|
|
cur.execute(
|
|
"""SELECT id, title FROM articles
|
|
WHERE column_id = ? AND status = 'draft'
|
|
AND id NOT IN ({})
|
|
ORDER BY created_at DESC LIMIT 1""".format(
|
|
",".join("?" * len(exclude_ids)) if exclude_ids else "0"
|
|
),
|
|
[col["column_id"]] + (list(exclude_ids) if exclude_ids else []),
|
|
)
|
|
row = cur.fetchone()
|
|
if row:
|
|
picked.append({
|
|
"id": row[0],
|
|
"title": row[1],
|
|
"column": col["name"],
|
|
})
|
|
exclude_ids.add(row[0])
|
|
found = True
|
|
break
|
|
|
|
if not found:
|
|
log.warning("草稿不足，已无更多可选文章")
|
|
break
|
|
|
|
conn.close()
|
|
|
|
# 更新 round_robin 索引
|
|
if strategy == "round_robin" and picked:
|
|
last_col_name = picked[-1]["column"]
|
|
for i, col in enumerate(available_cols):
|
|
if col["name"] == last_col_name:
|
|
state = load_state()
|
|
state["last_column_index"] = (i + 1) % len(available_cols)
|
|
save_state(state)
|
|
break
|
|
|
|
return picked
|
|
|
|
|
|
# ============================================================
|
|
# 主逻辑
|
|
# ============================================================
|
|
|
|
def should_publish(cfg, current_dt) -> list:
|
|
"""
|
|
检查当前时间是否应该触发发布。
|
|
|
|
Returns:
|
|
应触发的 slot 列表，每项: {"time": "08:00", "count": 1}
|
|
"""
|
|
today = current_dt.strftime("%Y-%m-%d")
|
|
state = load_state()
|
|
|
|
# 不是同一天 → 重置状态
|
|
if state.get("date") != today:
|
|
state = {"date": today, "fulfilled_slots": [], "published_ids": [], "used_columns": []}
|
|
save_state(state)
|
|
|
|
fulfilled = set(state.get("fulfilled_slots", []))
|
|
due_slots = []
|
|
|
|
for slot in cfg.get("schedule", []):
|
|
slot_time = slot["time"] # "08:00"
|
|
if slot_time in fulfilled:
|
|
continue
|
|
# 当前时间 >= 计划时间？
|
|
slot_h, slot_m = map(int, slot_time.split(":"))
|
|
now_minutes = current_dt.hour * 60 + current_dt.minute
|
|
slot_minutes = slot_h * 60 + slot_m
|
|
|
|
if now_minutes >= slot_minutes:
|
|
due_slots.append(slot)
|
|
|
|
return due_slots
|
|
|
|
|
|
def execute_slot(cfg, slot, state, dry_run=False):
|
|
"""
|
|
执行一个时间槽的发布任务。
|
|
|
|
Returns:
|
|
成功发布的文章数
|
|
"""
|
|
count = slot.get("count", 1)
|
|
slot_key = slot["time"]
|
|
|
|
log.info(f"⏰ 时段 {slot_key} 触发，计划发布 {count} 篇")
|
|
|
|
exclude_ids = set(state.get("published_ids", []))
|
|
used_columns_today = set(state.get("used_columns", []))
|
|
articles = pick_articles(cfg, count, exclude_ids, used_columns_today)
|
|
|
|
if not articles:
|
|
log.warning(f"时段 {slot_key}: 无可选草稿，跳过")
|
|
return 0
|
|
|
|
success = 0
|
|
for art in articles:
|
|
log.info(f"📝 开始处理 [{art['column']}] {art['title']} (id={art['id']})")
|
|
|
|
if dry_run:
|
|
log.info(f" [DRY-RUN] 将调用 run_pipeline({art['id']})")
|
|
success += 1
|
|
state.setdefault("published_ids", []).append(art["id"])
|
|
state.setdefault("used_columns", []).append(art["column"])
|
|
else:
|
|
try:
|
|
run_pipeline(art["id"])
|
|
log.info(f" ✅ 发布成功: {art['title']}")
|
|
success += 1
|
|
state.setdefault("published_ids", []).append(art["id"])
|
|
state.setdefault("used_columns", []).append(art["column"])
|
|
except Exception as e:
|
|
log.error(f" ❌ 发布失败: {art['title']} — {e}")
|
|
|
|
# 标记 slot 完成
|
|
if success > 0 or not articles:
|
|
state.setdefault("fulfilled_slots", []).append(slot_key)
|
|
|
|
save_state(state)
|
|
log.info(f"时段 {slot_key} 完成: {success}/{count} 篇成功")
|
|
return success
|
|
|
|
|
|
def main():
|
|
dry_run = "--dry-run" in sys.argv
|
|
force_now = "--now" in sys.argv
|
|
|
|
cfg = load_config()
|
|
|
|
# 获取当前时间（按配置时区）
|
|
tz_name = cfg.get("timezone", "Asia/Shanghai")
|
|
try:
|
|
from zoneinfo import ZoneInfo
|
|
tz = ZoneInfo(tz_name)
|
|
except Exception:
|
|
tz = None
|
|
|
|
current_dt = datetime.now(tz) if tz else datetime.now()
|
|
|
|
if force_now:
|
|
log.info(f"⚡ 手动执行所有未完成 slot [{current_dt.strftime('%Y-%m-%d %H:%M:%S')}]")
|
|
state = load_state()
|
|
today = current_dt.strftime("%Y-%m-%d")
|
|
if state.get("date") != today:
|
|
state = {"date": today, "fulfilled_slots": [], "published_ids": [], "used_columns": []}
|
|
for slot in cfg.get("schedule", []):
|
|
if slot["time"] not in state.get("fulfilled_slots", []):
|
|
execute_slot(cfg, slot, state, dry_run=dry_run)
|
|
return
|
|
|
|
due_slots = should_publish(cfg, current_dt)
|
|
|
|
if not due_slots:
|
|
return # 静默退出，cron 定时触发的正常情况
|
|
|
|
state = load_state()
|
|
for slot in due_slots:
|
|
execute_slot(cfg, slot, state, dry_run=dry_run)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|