Skip to content

M
macos-runtime
Commit faf1233d authored by 谢宇轩's avatar 谢宇轩
Browse files
Options

feat: unify harvest results and scheduling

parent 216de72b
Show whitespace changes
Inline Side-by-side
Showing with 1546 additions and 157 deletions
bin/add-source.sh
View file @ faf1233d
......@@ -5,11 +5,13 @@
# sudo ./bin/add-source.sh <source-name> [选项]
#
# 选项:
# --time HH:MM 运行时间,如 09:30(不指定则自动错开:现有最晚时间 +15 分钟)
# --time HH:MM[,HH:MM] 运行时间,如 09:30(不指定则自动错开:现有最晚时间 +15 分钟)
# --time 可重复传入，多个时间排序去重；修改时替换全部运行时间。
# --count N 抓取条数(写入 <source>.conf)
# --mail-to ADDR 收件人(写入 <source>.conf)
# --mail-on-success 0|1 成功是否发邮件(写入 <source>.conf)
# --provider NAME Claude provider 名称(写入 <source>.conf 的 PROVIDER=)
# --fallback-provider NAME 备用 provider 名称(写入 FALLBACK_PROVIDER=)
# --yes, -y 跳过确认
# --no-enable 只创建文件,不 enable
# --force 同名 timer 已存在时覆盖
......@@ -25,6 +27,9 @@
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
SCHEDULE_HELPER="$SCRIPT_DIR/schedule.py"
SYSTEMD_DIR="${SYSTEMD_DIR:-/etc/systemd/system}"
CONF_DIR="${CONF_DIR:-/etc/news-harvester}"
......@@ -34,6 +39,7 @@ COUNT=""
MAIL_TO=""
MAIL_ON_SUCCESS=""
PROVIDER=""
FALLBACK_PROVIDER=""
ASSUME_NO=0
NO_ENABLE=0
FORCE=0
......@@ -45,11 +51,12 @@ usage() {
# --- 参数解析 ---
while [ $# -gt 0 ]; do
case "$1" in
--time) TIME="${2:?--time 需要值}"; shift 2;;
--time) TIME="${TIME:+$TIME,}${2:?--time 需要值}"; shift 2;;
--count) COUNT="${2:?--count 需要值}"; shift 2;;
--mail-to) MAIL_TO="${2:?--mail-to 需要值}"; shift 2;;
--mail-on-success) MAIL_ON_SUCCESS="${2:?--mail-on-success 需要值}"; shift 2;;
--provider) PROVIDER="${2:?--provider 需要值}"; shift 2;;
--fallback-provider) FALLBACK_PROVIDER="${2:?--fallback-provider 需要值}"; shift 2;;
--yes|-y) ASSUME_NO=1; shift;;
--no-enable) NO_ENABLE=1; shift;;
--force) FORCE=1; shift;;
......@@ -88,9 +95,6 @@ confirm() {
[ -f "/home/yichu/Documents/News-Harvester/bin/harvest.sh" ] || warn "harvest.sh 不在默认位置,确认已安装(可忽略若路径不同)"
# 校验时间格式
validate_time() {
[[ "$1" =~ ^([01][0-9]|2[0-3]):[0-5][0-9]$ ]] || err "时间格式应为 HH:MM(24小时制),如 09:30,得到: $1"
}
validate_count() {
[[ "$1" =~ ^[0-9]+$ ]] && [ "$1" -gt 0 ] || err "count 应为正整数: $1"
}
......@@ -99,6 +103,9 @@ validate_mos() {
}
[ -n "$COUNT" ] && validate_count "$COUNT"
[ -n "$MAIL_ON_SUCCESS" ] && validate_mos "$MAIL_ON_SUCCESS"
# 兼容旧版 Web：default 表示不创建栏目级覆盖。
[ "$PROVIDER" = "default" ] && PROVIDER=""
[ "$FALLBACK_PROVIDER" = "default" ] && FALLBACK_PROVIDER=""
# --- 检查是否已存在 ---
TIMER_PATH="$SYSTEMD_DIR/news-harvester@${SOURCE}.timer"
......@@ -108,42 +115,12 @@ if [ -f "$TIMER_PATH" ] && [ "$FORCE" = "0" ]; then
fi
# --- 自动推算时间 ---
suggest_time() {
# 扫描现有所有 news-harvester@*.timer 的 OnCalendar,取最晚时间 +15 分钟
local max_epoch=0
shopt -s nullglob
for t in "$SYSTEMD_DIR"/news-harvester@*.timer; do
[ -f "$t" ] || continue
local line tc h m epoch
line=$(grep '^OnCalendar=' "$t" 2>/dev/null | head -1)
[ -z "$line" ] && continue
# 形如 OnCalendar=*-*-* 09:00:00,提取 HH:MM:SS
tc=$(printf '%s' "$line" | grep -oE '[0-9]{2}:[0-9]{2}:[0-9]{2}' | head -1)
[ -z "$tc" ] && continue
h=${tc:0:2}; m=${tc:3:2}
epoch=$((10#$h * 60 + 10#$m))
[ "$epoch" -gt "$max_epoch" ] && max_epoch=$epoch
done
shopt -u nullglob
if [ "$max_epoch" -eq 0 ]; then
echo "09:00" # 没有任何现有 timer,默认 09:00
return
fi
local new_epoch=$((max_epoch + 15))
if [ "$new_epoch" -ge 1440 ]; then
# 超过 23:59,wrap 到次日 00:00 起算
new_epoch=$((new_epoch - 1440))
warn "最晚栏目已接近午夜,新栏目自动 wrap 到次日 $new_epoch"
fi
printf '%02d:%02d\n' $((new_epoch / 60)) $((new_epoch % 60))
}
if [ -z "$TIME" ]; then
TIME=$(suggest_time)
TIME=$(python3 "$SCHEDULE_HELPER" suggest "$SYSTEMD_DIR")
AUTO_TIME=1
else
validate_time "$TIME"
TIME=$(python3 "$SCHEDULE_HELPER" normalize "$TIME") || exit 1
AUTO_TIME=0
fi
......@@ -152,12 +129,13 @@ echo "=== 即将创建栏目: $SOURCE ==="
echo " 运行时间: $TIME"
[ "$AUTO_TIME" = "1" ] && echo " (自动推算:现有最晚栏目 +15 分钟)"
echo " timer 文件: $TIMER_PATH"
if [ -n "$COUNT" ] || [ -n "$MAIL_TO" ] || [ -n "$MAIL_ON_SUCCESS" ] || [ -n "$PROVIDER" ]; then
if [ -n "$COUNT" ] || [ -n "$MAIL_TO" ] || [ -n "$MAIL_ON_SUCCESS" ] || [ -n "$PROVIDER" ] || [ -n "$FALLBACK_PROVIDER" ]; then
echo " conf 文件: $CONF_PATH"
[ -n "$COUNT" ] && echo " COUNT=$COUNT"
[ -n "$MAIL_TO" ] && echo " MAIL_TO=$MAIL_TO"
[ -n "$MAIL_ON_SUCCESS" ] && echo " MAIL_ON_SUCCESS=$MAIL_ON_SUCCESS"
[ -n "$PROVIDER" ] && echo " PROVIDER=$PROVIDER"
[ -n "$FALLBACK_PROVIDER" ] && echo " FALLBACK_PROVIDER=$FALLBACK_PROVIDER"
else
echo " conf 文件: (不创建,继承 default.conf)"
fi
......@@ -172,7 +150,7 @@ cat > "$TIMER_PATH" <<EOF
Description=Run News Harvester ($SOURCE) daily at $TIME
[Timer]
OnCalendar=*-*-* ${TIME}:00
$(python3 "$SCHEDULE_HELPER" calendar "$TIME")
Persistent=true
AccuracySec=1min
Unit=news-harvester@${SOURCE}.service
......@@ -180,10 +158,11 @@ Unit=news-harvester@${SOURCE}.service
[Install]
WantedBy=timers.target
EOF
python3 "$SCHEDULE_HELPER" update "$TIMER_PATH" "$TIME"
ok "已创建 $TIMER_PATH"
# --- 写 conf(仅当有覆盖项) ---
if [ -n "$COUNT" ] || [ -n "$MAIL_TO" ] || [ -n "$MAIL_ON_SUCCESS" ] || [ -n "$PROVIDER" ]; then
if [ -n "$COUNT" ] || [ -n "$MAIL_TO" ] || [ -n "$MAIL_ON_SUCCESS" ] || [ -n "$PROVIDER" ] || [ -n "$FALLBACK_PROVIDER" ]; then
mkdir -p "$CONF_DIR"
{
echo "# /etc/news-harvester/${SOURCE}.conf — 由 add-source.sh 生成 $(date '+%Y-%m-%d %H:%M:%S')"
......@@ -193,6 +172,8 @@ if [ -n "$COUNT" ] || [ -n "$MAIL_TO" ] || [ -n "$MAIL_ON_SUCCESS" ] || [ -n "$P
[ -n "$MAIL_TO" ] && echo "MAIL_TO=$MAIL_TO"
[ -n "$MAIL_ON_SUCCESS" ] && echo "MAIL_ON_SUCCESS=$MAIL_ON_SUCCESS"
[ -n "$PROVIDER" ] && echo "PROVIDER=$PROVIDER"
[ -n "$FALLBACK_PROVIDER" ] && echo "FALLBACK_PROVIDER=$FALLBACK_PROVIDER"
true
} > "$CONF_PATH"
ok "已创建 $CONF_PATH"
fi
......
bin/continue.sh
View file @ faf1233d
......@@ -6,7 +6,7 @@
#
# 选项:
# --mode resume|repair resume=中断续采(默认),repair=清理重采
# --budget N 单次预算上限美元(默认 15)
# --budget N 临时覆盖配置中的 MAX_BUDGET_USD
# --provider NAME 指定 claude provider(优先级最高;默认用 FALLBACK_PROVIDER)
# --yes, -y 跳过确认
# --debug 详细诊断输出
......@@ -22,6 +22,8 @@
# MAIL_TO 收件人;留空则不发邮件
# MAIL_ON_SUCCESS 1=成功也发,0=仅失败发(默认 1)
# ALERT_COST_OVER 单次成本(美元)超过此值则在邮件标题标记 [ALERT]
# MAX_BUDGET_USD Claude 单次运行预算上限美元(默认 15)
# PROVIDER_BUDGET_OVERRIDES 按 provider 覆盖预算，如 glm=25,easyrouter=30
# WORK_DIR 项目工作目录(默认 /home/yichu/Documents/News-Harvester)
# LOG_DIR 日志目录(默认 <WORK_DIR>/logs)
......@@ -33,7 +35,7 @@ export PATH="/home/yichu/.local/bin:/home/yichu/.nvm/versions/node/v24.14.0/bin:
SOURCE=""
MODE="resume"
BUDGET=15
CLI_BUDGET=""
ASSUME_YES=0
DEBUG=0
CLI_PROVIDER=""
......@@ -46,7 +48,7 @@ usage() {
while [ $# -gt 0 ]; do
case "$1" in
--mode) MODE="${2:?--mode 需要值}"; shift 2;;
--budget) BUDGET="${2:?--budget 需要值}"; shift 2;;
--budget) CLI_BUDGET="${2:?--budget 需要值}"; shift 2;;
--provider) CLI_PROVIDER="${2:?--provider 需要值}"; shift 2;;
--yes|-y) ASSUME_YES=1; shift;;
--debug) DEBUG=1; shift;;
......@@ -80,8 +82,6 @@ case "$MODE" in
resume|repair) ;;
*) err "--mode 只能是 resume 或 repair: $MODE";;
esac
[[ "$BUDGET" =~ ^[0-9]+\.?[0-9]*$ ]] || err "--budget 应为正数: $BUDGET"
# --- 加载配置(同 harvest.sh 的 EnvironmentFile 层级) ---
# Web 面板直接调用时不走 systemd,EnvironmentFile 不会自动加载,需手动 source
CONF_DIR="${CONF_DIR:-/etc/news-harvester}"
......@@ -93,6 +93,7 @@ done
MAIL_TO="${MAIL_TO:-}"
MAIL_ON_SUCCESS="${MAIL_ON_SUCCESS:-1}"
ALERT_COST_OVER="${ALERT_COST_OVER:-}"
MAX_BUDGET_USD="${MAX_BUDGET_USD:-15}"
PROVIDER="${PROVIDER:-}"
FALLBACK_PROVIDER="${FALLBACK_PROVIDER:-}"
WORK_DIR="${WORK_DIR:-/home/yichu/Documents/News-Harvester}"
......@@ -104,6 +105,8 @@ SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
# --provider 参数优先级最高
source "$SCRIPT_DIR/provider.sh"
resolve_continue_provider "$CLI_PROVIDER"
resolve_max_budget_usd "$CLI_BUDGET" || err "无法确定预算上限"
BUDGET="$RESOLVED_MAX_BUDGET_USD"
mkdir -p "$LOG_DIR"
......@@ -128,6 +131,9 @@ if [ -z "$SESSION_ID" ] || [ "$SESSION_ID" = "null" ]; then
err "JSON 日志中没有 session_id,无法恢复对话: $LATEST_JSON\n 可能上次运行 claude 启动失败,未产生有效 session"
fi
PREVIOUS_COUNT=$(jq -r 'if (.result | type) == "object" then .result.requested_count // empty else empty end' "$LATEST_JSON")
COUNT="${PREVIOUS_COUNT:-${COUNT:-20}}"
MODE_TAG=""
case "$MODE" in
resume) MODE_TAG="续采";;
......@@ -191,6 +197,11 @@ HUMAN_LOG="$LOG_DIR/${SOURCE}.last.log"
cd "$WORK_DIR"
source "$SCRIPT_DIR/result.sh"
prepare_result_contract || exit 1
PROMPT+=$'\n\n'"$RESULT_REQUIREMENTS"
START_EPOCH=$(date +%s)
# --- 运行 claude 无头模式,--resume 恢复对话 ---
......@@ -206,6 +217,7 @@ claude -p \
--permission-mode bypassPermissions \
--allowed-tools Bash Write \
--output-format json \
--json-schema "$RESULT_SCHEMA" \
--max-budget-usd "$BUDGET" \
"$PROMPT" \
> "$JSON_OUT" 2> "$RAW_OUT"
......@@ -245,6 +257,9 @@ fi
END_EPOCH=$(date +%s)
WALL_DURATION=$((END_EPOCH - START_EPOCH))
# Strict business validation also turns quality failures into a nonzero service exit.
normalize_result || { echo "结果校验程序失败，请查看原始输出" >&2; exit 1; }
# --- 解析 JSON 元数据(同 harvest.sh) ---
HAVE_JQ=0
command -v jq >/dev/null 2>&1 && [ -s "$JSON_OUT" ] && HAVE_JQ=1
......@@ -256,7 +271,7 @@ if [ "$HAVE_JQ" = "1" ]; then
fi
if [ "$HAVE_JQ" = "1" ]; then
IS_ERROR=$(jq -r '.is_error // "unknown"' "$JSON_OUT")
IS_ERROR=$(jq -r '.is_error' "$JSON_OUT")
SUBTYPE=$(jq -r '.subtype // "unknown"' "$JSON_OUT")
COST=$(jq -r '.total_cost_usd // 0' "$JSON_OUT")
NUM_TURNS=$(jq -r '.num_turns // 0' "$JSON_OUT")
......@@ -268,7 +283,7 @@ if [ "$HAVE_JQ" = "1" ]; then
CACHE_CREATE=$(jq -r '.usage.cache_creation_input_tokens // 0' "$JSON_OUT")
DENIALS=$(jq -r '.permission_denials | length' "$JSON_OUT")
TERMINAL=$(jq -r '.terminal_reason // "n/a"' "$JSON_OUT")
RESULT_TEXT=$(jq -r '.result // "(无输出)"' "$JSON_OUT")
RESULT_TEXT=$(python3 "$SCRIPT_DIR/harvest-result.py" summary "$JSON_OUT")
else
IS_ERROR="unknown"; SUBTYPE="unknown"; COST=0; NUM_TURNS=0; API_MS=0
SESSION_ID="n/a"; IN_TOK=0; OUT_TOK=0; CACHE_READ=0; CACHE_CREATE=0
......@@ -277,7 +292,7 @@ else
fi
# --- 判定状态 ---
if [ "$RC" -eq 0 ] && [ "$IS_ERROR" != "true" ]; then
if [ "$RC" -eq 0 ] && [ "$IS_ERROR" = "false" ]; then
STATUS_EMOJI="✅"
STATUS_TEXT="成功"
STATUS_TAG="SUCCESS"
......@@ -339,8 +354,6 @@ if [ "$SHOULD_NOTIFY" = "1" ]; then
fi
SUBJECT="[News Harvester] [CONTINUE] ${ALERT_PREFIX}${STATUS_TAG}: ${SOURCE} (${MODE_TAG})"
RESULT_TAIL=$(printf '%s\n' "$RESULT_TEXT" | tail -n 60)
BODY="$(cat <<EOF
${STATUS_EMOJI} News Harvester ${MODE_TAG}任务通知
......@@ -365,7 +378,7 @@ EOF
BODY+=$'\n成本阈值: \$'"${ALERT_COST_OVER} (本次正常)"
fi
fi
BODY+=$'\n\n----- result 末尾 60 行 -----\n'"${RESULT_TAIL}"
BODY+=$'\n\n----- 执行结果 -----\n'"${RESULT_TEXT}"
printf '%s' "$BODY" | "$SCRIPT_DIR/notify-mail.sh" "$MAIL_TO" "$SUBJECT" \
2>> "$RAW_OUT" || echo "[notify] 邮件发送失败(已忽略)" >> "$HUMAN_LOG"
......
bin/daily-digest.sh
View file @ faf1233d
......@@ -86,7 +86,7 @@ build_summary() {
base="$(basename "$f")"
src="${base%%.*}"
rest="${base#*.}"
ts="${rest#*.}"; ts="${ts%%.*}"
ts="${base%.json}"; ts="${ts##*.}"
if ! jq empty "$f" 2>/dev/null; then
[ "$first" = "1" ] || echo -n ","; first=0
......@@ -98,7 +98,8 @@ build_summary() {
jq -c --arg src "$src" --arg ts "$ts" '{
source:$src,time:$ts,
is_error:(.is_error//false),
status:(if(.is_error//false) then "FAILED" else "SUCCESS" end),
status:(if (.result | type) == "object" then .result.status // "FAILED"
elif (.is_error//false) then "FAILED" else "SUCCESS" end),
subtype:(.subtype//"unknown"),
cost:(.total_cost_usd//0),
turns:(.num_turns//0),
......@@ -107,7 +108,9 @@ build_summary() {
output_tokens:(.usage.output_tokens//0),
denials:((.permission_denials//[])|length),
terminal:(.terminal_reason//"n/a"),
result:(.result//"")
result:(if (.result | type) == "object" then .result | .articles |= map(
del(.body) | if (.archive | type) == "object" then .archive |= del(.body) else . end
) else .result // "" end)
}' "$f"
done
echo -n "]"
......@@ -165,8 +168,8 @@ claude -p \
重要:概况数字和每源表格由脚本自动生成并展示在邮件中,你**不需要**重复这些。你只负责:
1. anomalies:分析需要关注的异常模式(不是简单重复表格):
- result 文本是否暗示未完成(如只说"正在后台抓取"但无归档结果)
- 同一源当天多次运行(可能首次失败后重跑,产生重复成本)
- 旧版 result 文本是否暗示未完成，新版检查 result.failures、warnings 及各文章 status（正文已省略）(如只说"正在后台抓取"但无归档结果)
- 同一源重复失败或修复失败（正常定时多次运行不算异常）
- 成本/耗时/token 异常偏高的源
- 权限拒绝、解析失败等
2. recommendations:可执行建议(如有)
......@@ -179,7 +182,7 @@ claude -p \
- 单源成本超 \$$COST_ALERT_THRESHOLD → 标注
- 权限拒绝 > 0 → 标注
- result 提到"失败""错误""error""未完成""后台"等 → 关注
- 同源当日多次运行 → 说明原因
- 同源当日多次运行属于正常支持的多时间调度，不因次数本身告警
汇总数据:
$SUMMARY_CONTENT" \
......@@ -212,6 +215,14 @@ else
AI_NOTE=""
fi
# Business failures always alert, even if the model overlooks them.
if [ "$FAILED_COUNT" -gt 0 ]; then
ALERT="true"
RULE_FAILED=$(jq -c '[.[] | select(.status != "SUCCESS") | .source] | unique' "$SUMMARY_JSON")
FAILED=$(printf '%s' "$RULE_FAILED" | jq -r 'join(",")')
ALERT_REASONS=$(printf '%s' "$ALERT_REASONS" | jq --arg failed "$FAILED" '. + ["采集结果校验失败: " + $failed] | unique')
fi
# ============================================================
# 6. 写 digest.md(纯文本存档)
# ============================================================
......
bin/deploy.sh
View file @ faf1233d
......@@ -90,9 +90,9 @@ rsync -avz --delete \
ok "项目目录已同步"
# --- 2. 复制 bin + web 到工作目录 ---
info "2/6 复制 bin/*.sh + web/ → ~/Documents/News-Harvester/"
info "2/6 复制运行脚本、Schema + web/ → ~/Documents/News-Harvester/"
# deploy.sh / migrate.sh 是仓库级工具,不复制到运行目录
ssh "$HOST" 'mkdir -p ~/Documents/News-Harvester/bin ~/Documents/News-Harvester/web/templates && for f in ~/news-harvester-pro/bin/*.sh; do case "$(basename "$f")" in deploy.sh|migrate.sh) ;; *) cp "$f" ~/Documents/News-Harvester/bin/ ;; esac; done && chmod +x ~/Documents/News-Harvester/bin/*.sh && cp ~/news-harvester-pro/web/app.py ~/Documents/News-Harvester/web/ && cp ~/news-harvester-pro/web/templates/*.html ~/Documents/News-Harvester/web/templates/'
ssh "$HOST" 'mkdir -p ~/Documents/News-Harvester/bin ~/Documents/News-Harvester/web/templates && for f in ~/news-harvester-pro/bin/*.sh ~/news-harvester-pro/bin/*.py ~/news-harvester-pro/bin/*.schema.json; do case "$(basename "$f")" in deploy.sh|migrate.sh) ;; *) cp "$f" ~/Documents/News-Harvester/bin/ ;; esac; done && chmod +x ~/Documents/News-Harvester/bin/*.sh && cp ~/news-harvester-pro/web/app.py ~/Documents/News-Harvester/web/ && cp ~/news-harvester-pro/web/templates/*.html ~/Documents/News-Harvester/web/templates/'
ok "bin 脚本 + web 应用已就位"
# --- 3. 同步 systemd unit ---
......
bin/harvest-result.py 0 → 100644
View file @ faf1233d
#!/usr/bin/env python3
"""Validate structured Claude output and determine business success (stdlib only)."""
import argparse
import copy
import html
import json
import re
from pathlib import Path
SCHEMA_PATH = Path(__file__).with_name("harvest-result.schema.json")
SHORT_BODY_LIMIT = 500
def validate(value, schema, path="result"):
"""Validate the deliberately small JSON Schema vocabulary used by our contract."""
types = {"object": dict, "array": list, "string": str, "integer": int,
"boolean": bool, "null": type(None)}
expected = schema["type"]
expected = expected if isinstance(expected, list) else [expected]
if not any(type(value) is types[k] for k in expected):
raise ValueError(f"{path}: 类型应为 {expected}")
if "enum" in schema and value not in schema["enum"]:
raise ValueError(f"{path}: 非法枚举值")
if type(value) is int and value < schema.get("minimum", value):
raise ValueError(f"{path}: 不得小于 {schema['minimum']}")
if isinstance(value, dict):
if set(schema["required"]) - value.keys():
raise ValueError(f"{path}: 缺少必要字段")
if schema.get("additionalProperties") is False and value.keys() - schema["properties"].keys():
raise ValueError(f"{path}: 存在未约定字段")
for key, item in value.items():
validate(item, schema["properties"][key], f"{path}.{key}")
if isinstance(value, list):
for index, item in enumerate(value):
validate(item, schema["items"], f"{path}[{index}]")
def body_length(body):
return len(re.sub(r"\s", "", body))
def parse_scalar(value):
value = value.strip()
if not value:
return ""
if value.startswith('"') and value.endswith('"'):
try:
return json.loads(value)
except ValueError:
return value[1:-1]
if value.startswith("'") and value.endswith("'"):
return value[1:-1].replace("''", "'")
if value in ("null", "Null", "NULL", "~"):
return None
if value.lower() in ("true", "false"):
return value.lower() == "true"
return value
def parse_frontmatter(text):
"""Parse the flat properties emitted by the news-harvester skill."""
match = re.match(r"\A(?:\ufeff)?[ \t\r\n]*---\s*\r?\n(.*?)\r?\n---\s*(?:\r?\n|\Z)",
text, re.DOTALL)
if not match:
raise ValueError("归档文件缺少 YAML 属性区")
properties = {}
current_list = None
for raw_line in match.group(1).splitlines():
list_item = re.match(r"^\s+-\s+(.*)$", raw_line)
if list_item and current_list:
properties[current_list].append(parse_scalar(list_item.group(1)))
continue
field = re.match(r"^([A-Za-z_][A-Za-z0-9_-]*):(?:\s*(.*))?$", raw_line)
if not field:
if raw_line.strip() and not raw_line.lstrip().startswith("#"):
raise ValueError(f"无法解析归档属性：{raw_line.strip()}")
continue
key, raw_value = field.groups()
if raw_value:
properties[key] = parse_scalar(raw_value)
current_list = None
else:
properties[key] = []
current_list = key
return properties
def parse_original_archive(text):
properties = parse_frontmatter(text)
if properties.get("type") != "原文":
raise ValueError("archive_path 不是原文归档")
heading = re.search(r"(?m)^##\s+正文\s*$", text)
if not heading:
raise ValueError("原文归档缺少“## 正文”区段")
content = text[heading.end():].lstrip("\r\n")
separator = re.search(r"(?m)^---\s*$", content)
if separator:
content = content[:separator.start()]
return properties, content.strip()
def clean_archive_path(archive_path):
"""Undo common shell/Markdown formatting accidentally emitted by the model."""
value = html.unescape(archive_path).strip()
if len(value) >= 2 and value[0] == value[-1] and value[0] in "'\"`":
value = value[1:-1]
value = re.sub(r"\\\r?\n", "", value)
for _ in range(2):
value = re.sub(r"\\(.)", r"\1", value)
value = re.sub(r"\[([^\]\r\n]+原文\.md)\]\([^)\r\n]+\)", r"\1", value)
return value.strip()
def locate_archive(path, expected_url=""):
if path.is_file():
return path
cursor = path.parent
while cursor != cursor.parent and not cursor.exists():
cursor = cursor.parent
if not cursor.is_dir():
raise FileNotFoundError(path)
try:
missing_parts = path.relative_to(cursor).parts
except ValueError:
missing_parts = ()
if missing_parts:
joined = cursor / "-".join(missing_parts)
if joined.is_file():
return joined
if expected_url:
for candidate in cursor.rglob("*原文.md"):
try:
properties = parse_frontmatter(candidate.read_text(encoding="utf-8"))
except (OSError, UnicodeError, ValueError):
continue
if properties.get("source") == expected_url:
return candidate
raise FileNotFoundError(path)
def read_archive(archive_path, work_dir=None, vault_root=None, expected_url=""):
path = Path(clean_archive_path(archive_path)).expanduser()
if not path.is_absolute():
path = Path(work_dir or Path.cwd()) / path
path = path.resolve(strict=False)
if vault_root:
root = Path(vault_root).expanduser().resolve(strict=True)
try:
path.relative_to(root)
except ValueError as exc:
raise ValueError("archive_path 超出配置的 Obsidian Vault") from exc
path = locate_archive(path, expected_url).resolve(strict=True)
if vault_root:
try:
path.relative_to(root)
except ValueError as exc:
raise ValueError("实际归档文件超出配置的 Obsidian Vault") from exc
if path.suffix.lower() != ".md" or not path.name.endswith("原文.md"):
raise ValueError("archive_path 必须指向“原文.md”文件")
properties, body = parse_original_archive(path.read_text(encoding="utf-8"))
return {
"loaded": True,
"path": str(path),
"properties": properties,
"body": body,
"body_length": body_length(body),
"error": "",
}
def empty_result(source, count):
return dict(schema_version=1, source=source, requested_count=count,
attempted_count=0, succeeded_count=0, failed_count=0, status="FAILED",
source_exhausted=False, exhaustion_reason="", warnings=[], failures=[], articles=[])
def one_line(value):
return re.sub(r"\s+", " ", str(value or "")).strip()
def related_title(article):
related = article.get("archive", {}).get("properties", {}).get("related", "")
if isinstance(related, list):
related = related[0] if related else ""
match = re.fullmatch(r"\[\[([^\]]+)\]\]", one_line(related))
return match.group(1).split("|", 1)[0] if match else one_line(related)
def infer_push_status(messages):
text = " ".join(one_line(item) for item in messages).lower()
for status in ("updated", "created", "synced", "skipped", "failed"):
if re.search(rf"\b{status}\b", text):
return status
if "推送失败" in text or "同步失败" in text:
return "failed"
if "推送成功" in text or "同步成功" in text:
return "synced"
return "—"
def format_summary(result):
state = "成功" if result["status"] == "SUCCESS" else "失败"
icon = "✓" if result["status"] == "SUCCESS" else "✗"
lines = [
f"{icon} 执行{state} | 成功 {result['succeeded_count']} | 失败 {result['failed_count']} | 目标 {result['requested_count']}",
"",
"文章：",
]
global_push = infer_push_status(result.get("warnings", []))
article_notices = set()
article_urls = set()
for article in result["articles"]:
title = one_line(article.get("title")) or one_line(article.get("url"))
translated = related_title(article)
names = f"{title} | {translated}" if translated and translated != title else title
archive = article.get("archive", {})
archive_status = "✓" if archive.get("loaded") else "✗"
push_status = infer_push_status(article.get("quality_evidence", []))
if push_status == "—":
push_status = global_push
detail = f"归档 {archive_status} | 推送 {push_status}"
if article.get("status") == "FAILED":
detail += f" | {article.get('failure_type') or 'UNKNOWN'}：{one_line(article.get('failure_reason'))}"
lines.append(f"- {names} | {detail}")
article_urls.add(article.get("url", ""))
article_notices.add(
f"{article.get('title') or article.get('url')}：{article.get('failure_type')} — {article.get('failure_reason')}"
)
if not result["articles"]:
lines.append("- 无")
task_failures = [item for item in result.get("failures", [])
if not item.get("url") or item.get("url") not in article_urls]
if task_failures:
lines.extend(["", "任务异常："])
for failure in task_failures:
lines.append(f"- [{failure['type']}] {one_line(failure['message'])}")
notes = []
for item in result.get("warnings", []):
if item in article_notices:
continue
note = re.sub(r"^(?:提示|警告)[:：]\s*", "", one_line(item))
if note and note not in notes:
notes.append(note)
if result.get("source_exhausted"):
notes.append(f"数据已取尽：{one_line(result.get('exhaustion_reason'))}")
if notes:
lines.extend(["", "说明："])
lines.extend(f"- {note}" for note in notes)
return "\n".join(lines)
def normalize(envelope, source, count, process_rc, work_dir=None, vault_root=None):
schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8"))
result = empty_result(source, count)
task_failed = False
def fail(kind, message, url="", fatal=True):
nonlocal task_failed
failure = dict(type=kind, message=message, url=url)
if failure not in result["failures"]:
result["failures"].append(failure)
if fatal:
task_failed = True
try:
candidate = envelope.get("structured_output")
validate(candidate, schema)
if candidate["source"] != source or candidate["requested_count"] != count:
raise ValueError("source/requested_count 与本次调用不一致")
result = copy.deepcopy(candidate)
for failure in result["failures"]:
if (not failure["url"] or
failure["type"] in {"RUNTIME_ERROR", "PERMISSION_DENIED", "INVALID_OUTPUT"}):
task_failed = True
except ValueError as exc:
fail("INVALID_OUTPUT", str(exc))
if process_rc != 0 or envelope.get("is_error") is True or envelope.get("subtype", "success") != "success":
fail("RUNTIME_ERROR", f"Claude 运行异常：exit={process_rc}, subtype={envelope.get('subtype', 'unknown')}")
if envelope.get("permission_denials"):
fail("PERMISSION_DENIED", "Claude 存在工具权限拒绝，请检查原始输出")
seen = set()
for article in result["articles"]:
url = article["url"]
if not re.match(r"^https?://[^\s/]+", url) or url in seen:
fail("INVALID_OUTPUT", "文章 URL 无效或重复", url)
article.update(status="FAILED", failure_type="INVALID_OUTPUT", failure_reason="文章 URL 无效或重复")
seen.add(url)
# Claude commonly emits null for the optional meaning of this required field.
# The schema accepts it, while persisted results keep one stable string type.
article["failure_reason"] = article["failure_reason"] or ""
if article["status"] == "SUCCESS":
if article["failure_type"] or article["failure_reason"]:
fail("INVALID_OUTPUT", "文章成功状态与失败原因矛盾", url)
article.update(status="FAILED", failure_type=article["failure_type"] or "INVALID_OUTPUT",
failure_reason=article["failure_reason"] or "文章成功状态与失败原因矛盾")
elif not all(article[k].strip() for k in ("title", "archive_path")):
article.update(status="FAILED", failure_type="INVALID_OUTPUT", failure_reason="成功文章缺少标题或归档路径")
archive = {"loaded": False, "properties": {}, "body": "", "body_length": None, "error": ""}
if article["archive_path"]:
try:
archive = read_archive(article["archive_path"], work_dir, vault_root, url)
article["archive_path"] = archive["path"]
except (OSError, UnicodeError, ValueError) as exc:
archive["error"] = str(exc)
if article["status"] == "SUCCESS":
article.update(status="FAILED", failure_type="ARCHIVE_FAILED",
failure_reason=f"无法读取原文归档：{exc}")
elif article["status"] == "SUCCESS":
article.update(status="FAILED", failure_type="ARCHIVE_FAILED",
failure_reason="成功文章没有原文归档路径")
article["archive"] = archive
if article["status"] == "FAILED":
article["failure_type"] = article["failure_type"] or "UNKNOWN"
article["failure_reason"] = article["failure_reason"] or "采集未完成，未提供具体原因"
fail(article["failure_type"], article["failure_reason"], url, fatal=False)
notice = f"{article['title'] or url}：{article['failure_type']} — {article['failure_reason']}"
if notice not in result["warnings"]:
result["warnings"].append(notice)
result["attempted_count"] = len(result["articles"])
result["succeeded_count"] = sum(a["status"] == "SUCCESS" for a in result["articles"])
result["failed_count"] = result["attempted_count"] - result["succeeded_count"]
if (result["attempted_count"] and
result["failed_count"] * 2 >= result["attempted_count"]):
fail("FETCH_FAILED",
f"{result['attempted_count']} 篇文章中有 {result['failed_count']} 篇失败，未取得多数成功")
readable = [a for a in result["articles"] if a["archive"]["loaded"]]
short_count = sum(a["archive"]["body_length"] < SHORT_BODY_LIMIT for a in readable)
if readable and short_count * 2 >= len(readable):
ratio = short_count * 100 / len(readable)
message = (f"{short_count}/{len(readable)} 篇可读取的归档原文正文少于 "
f"{SHORT_BODY_LIMIT} 字（{ratio:.0f}%），疑似触发付费墙")
fail("PAYWALL", message)
if message not in result["warnings"]:
result["warnings"].append(message)
if result["source_exhausted"] and not result["exhaustion_reason"].strip():
fail("INVALID_OUTPUT", "声明已无更多新文章，但未说明检查依据")
if result["attempted_count"] < count and not result["source_exhausted"]:
fail("FETCH_FAILED", f"要求 {count} 篇，仅报告 {result['attempted_count']} 篇，且未确认数据已取尽")
result["status"] = "FAILED" if task_failed else "SUCCESS"
output = dict(envelope)
# Preserve the provider's text separately; downstream uses only the verified result.
output["result_text"] = envelope.get("result", "")
output.pop("structured_output", None)
output["result"] = result
output["claude_is_error"] = envelope.get("is_error")
output["is_error"] = result["status"] == "FAILED"
output["process_exit_code"] = process_rc
output["exit_code"] = process_rc if process_rc else (1 if output["is_error"] else 0)
return output
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("command", choices=["normalize", "summary"])
parser.add_argument("path", type=Path)
parser.add_argument("--source")
parser.add_argument("--count", type=int, default=20)
parser.add_argument("--exit-code", type=int, default=0)
parser.add_argument("--work-dir", type=Path)
parser.add_argument("--vault-root", type=Path)
args = parser.parse_args()
if args.count < 1:
parser.error("采集数量必须为正整数")
if args.command == "summary":
result = json.loads(args.path.read_text(encoding="utf-8"))["result"]
print(format_summary(result))
return 0
if not args.source:
parser.error("normalize 需要 --source")
try:
envelope = json.loads(args.path.read_text(encoding="utf-8"))
if not isinstance(envelope, dict):
envelope = {}
except (ValueError, OSError):
envelope = {}
output = normalize(envelope, args.source, args.count, args.exit_code,
args.work_dir, args.vault_root)
temporary = args.path.with_suffix(".tmp")
temporary.write_text(json.dumps(output, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
temporary.replace(args.path)
return 0 # Business exit code is in the output, not the normalizer's exit code.
if __name__ == "__main__":
raise SystemExit(main())
bin/harvest-result.schema.json 0 → 100644
View file @ faf1233d
{
"type": "object",
"properties": {
"schema_version": {
"type": "integer",
"enum": [
1
]
},
"source": {
"type": "string"
},
"requested_count": {
"type": "integer",
"minimum": 0
},
"attempted_count": {
"type": "integer",
"minimum": 0
},
"succeeded_count": {
"type": "integer",
"minimum": 0
},
"failed_count": {
"type": "integer",
"minimum": 0
},
"status": {
"type": "string",
"enum": [
"SUCCESS",
"FAILED"
]
},
"source_exhausted": {
"type": "boolean",
"description": "True only after verifying no more new unarchived articles; access failure is not exhaustion."
},
"exhaustion_reason": {
"type": "string"
},
"warnings": {
"type": "array",
"items": {
"type": "string"
}
},
"failures": {
"type": "array",
"items": {
"type": "object",
"properties": {
"type": {
"type": "string",
"enum": [
"PAYWALL",
"LOGIN_REQUIRED",
"CAPTCHA",
"EMPTY_BODY",
"BODY_TOO_SHORT",
"CONTENT_INCOMPLETE",
"FETCH_FAILED",
"ARCHIVE_FAILED",
"PUSH_FAILED",
"RUNTIME_ERROR",
"PERMISSION_DENIED",
"INVALID_OUTPUT",
"UNKNOWN"
]
},
"message": {
"type": "string"
},
"url": {
"type": "string"
}
},
"required": [
"type",
"message",
"url"
],
"additionalProperties": false
}
},
"articles": {
"type": "array",
"items": {
"type": "object",
"properties": {
"url": {
"type": "string"
},
"title": {
"type": "string"
},
"published_at": {
"type": "string"
},
"archive_path": {
"type": "string",
"description": "Absolute path of the archived original Markdown file; empty when not archived."
},
"status": {
"type": "string",
"enum": [
"SUCCESS",
"FAILED"
]
},
"failure_type": {
"type": [
"string",
"null"
],
"enum": [
null,
"PAYWALL",
"LOGIN_REQUIRED",
"CAPTCHA",
"EMPTY_BODY",
"BODY_TOO_SHORT",
"CONTENT_INCOMPLETE",
"FETCH_FAILED",
"ARCHIVE_FAILED",
"PUSH_FAILED",
"RUNTIME_ERROR",
"PERMISSION_DENIED",
"INVALID_OUTPUT",
"UNKNOWN"
]
},
"failure_reason": {
"type": [
"string",
"null"
],
"description": "Failure detail; null or empty for successful articles."
},
"quality_evidence": {
"type": "array",
"items": {
"type": "string"
},
"description": "Observed page access or truncation evidence; no invented explanations."
}
},
"required": [
"url",
"title",
"published_at",
"archive_path",
"status",
"failure_type",
"failure_reason",
"quality_evidence"
],
"additionalProperties": false
}
}
},
"required": [
"schema_version",
"source",
"requested_count",
"attempted_count",
"succeeded_count",
"failed_count",
"status",
"source_exhausted",
"exhaustion_reason",
"warnings",
"failures",
"articles"
],
"additionalProperties": false,
"description": "News Harvester v1 business result. Article content and properties are loaded locally from archive_path; counts and final status are recomputed locally."
}
bin/harvest.sh
View file @ faf1233d
......@@ -15,6 +15,8 @@
# MAIL_TO 收件人;留空则不发邮件
# MAIL_ON_SUCCESS 1=成功也发,0=仅失败发(默认 1)
# ALERT_COST_OVER 单次成本(美元)超过此值则在邮件标题标记 [ALERT],留空不判断
# MAX_BUDGET_USD Claude 单次运行预算上限美元(默认 15)
# PROVIDER_BUDGET_OVERRIDES 按 provider 覆盖预算，如 glm=25,easyrouter=30
# WORK_DIR 项目工作目录(默认 /home/yichu/Documents/News-Harvester)
# LOG_DIR 日志目录(默认 <WORK_DIR>/logs)
......@@ -25,10 +27,12 @@ export NVM_DIR="/home/yichu/.nvm"
export PATH="/home/yichu/.local/bin:/home/yichu/.nvm/versions/node/v24.14.0/bin:$PATH"
SOURCE="${1:?用法: harvest.sh <source-name> [count]}"
COUNT="${2:-20}"
[[ "$SOURCE" =~ ^[a-zA-Z0-9_-]+$ ]] || { echo "栏目名格式错误" >&2; exit 1; }
COUNT="${2:-${COUNT:-20}}"
MAIL_TO="${MAIL_TO:-}"
MAIL_ON_SUCCESS="${MAIL_ON_SUCCESS:-1}"
ALERT_COST_OVER="${ALERT_COST_OVER:-}"
MAX_BUDGET_USD="${MAX_BUDGET_USD:-15}"
PROVIDER="${PROVIDER:-}"
WORK_DIR="${WORK_DIR:-/home/yichu/Documents/News-Harvester}"
LOG_DIR="${LOG_DIR:-$WORK_DIR/logs}"
......@@ -39,6 +43,7 @@ SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
CONF_DIR="${CONF_DIR:-/etc/news-harvester}"
source "$SCRIPT_DIR/provider.sh"
build_claude_settings_args "$PROVIDER"
resolve_max_budget_usd || exit 1
mkdir -p "$LOG_DIR"
DATE_TAG="$(date +%Y%m%d)"
......@@ -49,6 +54,9 @@ HUMAN_LOG="$LOG_DIR/${SOURCE}.last.log"
cd "$WORK_DIR"
source "$SCRIPT_DIR/result.sh"
prepare_result_contract || exit 1
START_EPOCH=$(date +%s)
# --- 运行 claude 无头模式,输出 JSON ---
......@@ -59,7 +67,8 @@ claude -p \
--permission-mode bypassPermissions \
--allowed-tools Bash Write \
--output-format json \
--max-budget-usd 15 \
--json-schema "$RESULT_SCHEMA" \
--max-budget-usd "$RESOLVED_MAX_BUDGET_USD" \
"使用news-harvester获取${SOURCE}的最新${COUNT}条数据，并保存到obsidian中。
执行要求：
......@@ -67,7 +76,7 @@ claude -p \
- 不要使用后台 shell、nohup、& 或任何形式的异步执行。
- 所有文章抓取完毕后再依次生成摘要并写入 Obsidian，全部完成后再返回结果。
- 如果某篇文章抓取失败，跳过它继续下一篇，不要因此中断整个任务。
- 最终结果中简要说明：成功归档几篇、失败几篇、各失败原因。" \
${RESULT_REQUIREMENTS}" \
> "$JSON_OUT" 2> "$RAW_OUT"
RC=$?
set -e
......@@ -93,6 +102,9 @@ fi
END_EPOCH=$(date +%s)
WALL_DURATION=$((END_EPOCH - START_EPOCH))
# Strict business validation also turns quality failures into a nonzero service exit.
normalize_result || { echo "结果校验程序失败，请查看原始输出" >&2; exit 1; }
# --- 解析 JSON 元数据 ---
HAVE_JQ=0
command -v jq >/dev/null 2>&1 && [ -s "$JSON_OUT" ] && HAVE_JQ=1
......@@ -105,7 +117,7 @@ if [ "$HAVE_JQ" = "1" ]; then
fi
if [ "$HAVE_JQ" = "1" ]; then
IS_ERROR=$(jq -r '.is_error // "unknown"' "$JSON_OUT")
IS_ERROR=$(jq -r '.is_error' "$JSON_OUT")
SUBTYPE=$(jq -r '.subtype // "unknown"' "$JSON_OUT")
COST=$(jq -r '.total_cost_usd // 0' "$JSON_OUT")
NUM_TURNS=$(jq -r '.num_turns // 0' "$JSON_OUT")
......@@ -117,7 +129,7 @@ if [ "$HAVE_JQ" = "1" ]; then
CACHE_CREATE=$(jq -r '.usage.cache_creation_input_tokens // 0' "$JSON_OUT")
DENIALS=$(jq -r '.permission_denials | length' "$JSON_OUT")
TERMINAL=$(jq -r '.terminal_reason // "n/a"' "$JSON_OUT")
RESULT_TEXT=$(jq -r '.result // "(无输出)"' "$JSON_OUT")
RESULT_TEXT=$(python3 "$SCRIPT_DIR/harvest-result.py" summary "$JSON_OUT")
else
IS_ERROR="unknown"; SUBTYPE="unknown"; COST=0; NUM_TURNS=0; API_MS=0
SESSION_ID="n/a"; IN_TOK=0; OUT_TOK=0; CACHE_READ=0; CACHE_CREATE=0
......@@ -126,8 +138,8 @@ else
fi
# --- 判定状态 ---
# 成功条件:进程退出码 0 且 is_error 非 true
if [ "$RC" -eq 0 ] && [ "$IS_ERROR" != "true" ]; then
# 成功条件:Claude 运行与正文质量、业务结果校验全部通过
if [ "$RC" -eq 0 ] && [ "$IS_ERROR" = "false" ]; then
STATUS_EMOJI="✅"
STATUS_TEXT="成功"
STATUS_TAG="SUCCESS"
......@@ -191,9 +203,6 @@ if [ "$SHOULD_NOTIFY" = "1" ]; then
fi
SUBJECT="[News Harvester] ${ALERT_PREFIX}${STATUS_TAG}: ${SOURCE}"
# 取 result 末尾 60 行,避免邮件过长
RESULT_TAIL=$(printf '%s\n' "$RESULT_TEXT" | tail -n 60)
BODY="$(cat <<EOF
${STATUS_EMOJI} News Harvester 任务通知
......@@ -217,7 +226,7 @@ EOF
BODY+=$'\n成本阈值: \$'"${ALERT_COST_OVER} (本次正常)"
fi
fi
BODY+=$'\n\n----- result 末尾 60 行 -----\n'"${RESULT_TAIL}"
BODY+=$'\n\n----- 执行结果 -----\n'"${RESULT_TEXT}"
# 通知失败不影响任务退出码
printf '%s' "$BODY" | "$SCRIPT_DIR/notify-mail.sh" "$MAIL_TO" "$SUBJECT" \
......
bin/migrate.sh
View file @ faf1233d
......@@ -104,8 +104,9 @@ fi
# === 正常安装/迁移 ===
need_root "$@"
# 前置检查:jq / msmtp / claude
# 前置检查:python3 / jq / msmtp / claude
echo "=== 前置依赖检查 ==="
command -v python3 >/dev/null || err "需要 Python 3: sudo pacman -S python"
command -v jq >/dev/null && ok "jq 已安装" || warn "jq 未安装: sudo pacman -S jq"
command -v msmtp >/dev/null && ok "msmtp 已安装" || warn "msmtp 未安装: sudo pacman -S msmtp"
command -v claude >/dev/null && ok "claude 已安装" || warn "claude 未在 root PATH 中(可能正常,脚本以 $USER_NAME 身份运行时会从用户 profile 加载)"
......@@ -166,7 +167,7 @@ fi
echo ""
echo "=== 步骤 3/5: 安装新版脚本 ==="
mkdir -p "$WORK_DIR/bin"
for f in "$SCRIPT_DIR"/*.sh; do
for f in "$SCRIPT_DIR"/*.sh "$SCRIPT_DIR"/*.py "$SCRIPT_DIR"/*.schema.json; do
base="$(basename "$f")"
# deploy.sh / migrate.sh 是仓库级工具,不装到运行目录
case "$base" in deploy.sh|migrate.sh) continue ;; esac
......
bin/modify-source.sh
View file @ faf1233d
......@@ -5,7 +5,8 @@
# sudo ./bin/modify-source.sh <source-name> [选项]
#
# 选项:
# --time HH:MM 新的运行时间,如 10:30(必填)
# --time HH:MM[,HH:MM] 新的运行时间,如 10:30(必填)
# --time 可重复传入，多个时间排序去重；修改时替换全部运行时间。
# --count N 更新抓取条数(写入/覆盖 <source>.conf)
# --mail-to ADDR 更新收件人
# --mail-on-success 0|1 更新成功是否发邮件
......@@ -24,6 +25,9 @@
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
SCHEDULE_HELPER="$SCRIPT_DIR/schedule.py"
SYSTEMD_DIR="${SYSTEMD_DIR:-/etc/systemd/system}"
CONF_DIR="${CONF_DIR:-/etc/news-harvester}"
......@@ -34,6 +38,8 @@ MAIL_TO=""
MAIL_ON_SUCCESS=""
PROVIDER=""
FALLBACK_PROVIDER=""
PROVIDER_SET=0
FALLBACK_PROVIDER_SET=0
ASSUME_NO=0
usage() {
......@@ -43,12 +49,12 @@ usage() {
# --- 参数解析 ---
while [ $# -gt 0 ]; do
case "$1" in
--time) TIME="${2:?--time 需要值}"; shift 2;;
--time) TIME="${TIME:+$TIME,}${2:?--time 需要值}"; shift 2;;
--count) COUNT="${2:?--count 需要值}"; shift 2;;
--mail-to) MAIL_TO="${2:?--mail-to 需要值}"; shift 2;;
--mail-on-success) MAIL_ON_SUCCESS="${2:?--mail-on-success 需要值}"; shift 2;;
--provider) PROVIDER="${2:?--provider 需要值}"; shift 2;;
--fallback-provider) FALLBACK_PROVIDER="${2:?--fallback-provider 需要值}"; shift 2;;
--provider) PROVIDER="${2:?--provider 需要值}"; PROVIDER_SET=1; shift 2;;
--fallback-provider) FALLBACK_PROVIDER="${2:?--fallback-provider 需要值}"; FALLBACK_PROVIDER_SET=1; shift 2;;
--yes|-y) ASSUME_NO=1; shift;;
--help|-h) usage; exit 0;;
--) shift; break;;
......@@ -79,18 +85,17 @@ confirm() {
[ -n "$TIME" ] || err "必须指定 --time HH:MM"
[ "$(id -u)" -eq 0 ] || err "请用 sudo 运行"
validate_time() {
[[ "$1" =~ ^([01][0-9]|2[0-3]):[0-5][0-9]$ ]] || err "时间格式应为 HH:MM(24小时制),如 10:30,得到: $1"
}
validate_count() {
[[ "$1" =~ ^[0-9]+$ ]] && [ "$1" -gt 0 ] || err "count 应为正整数: $1"
}
validate_mos() {
case "$1" in 0|1) ;; *) err "--mail-on-success 只能是 0 或 1: $1";; esac
}
validate_time "$TIME"
TIME=$(python3 "$SCHEDULE_HELPER" normalize "$TIME") || exit 1
[ -n "$COUNT" ] && validate_count "$COUNT"
[ -n "$MAIL_ON_SUCCESS" ] && validate_mos "$MAIL_ON_SUCCESS"
[ "$PROVIDER" = "default" ] && PROVIDER=""
[ "$FALLBACK_PROVIDER" = "default" ] && FALLBACK_PROVIDER=""
# --- 检查 timer 是否存在 ---
TIMER_PATH="$SYSTEMD_DIR/news-harvester@${SOURCE}.timer"
......@@ -108,10 +113,10 @@ local_cal=$(echo "$local_show" | grep '^OnCalendar=' | cut -d= -f2- | tr '\n' '
if [ -z "${local_cal// /}" ]; then
local_cal=$(echo "$local_show" | grep -oE 'OnCalendar=[^;}]*' | cut -d= -f2- | tr '\n' ' ' || true)
fi
CURRENT_TIME=$(echo "$local_cal" | grep -oE '[0-9]{2}:[0-9]{2}' | head -1 || true)
CURRENT_TIME=$(echo "$local_cal" | grep -oE '[0-9]{2}:[0-9]{2}' | sort -u | paste -sd, - || true)
# 方法2:回退到直接读 timer 文件
if [ -z "$CURRENT_TIME" ]; then
CURRENT_TIME=$(grep '^OnCalendar=' "$TIMER_PATH" 2>/dev/null | head -1 | grep -oE '[0-9]{2}:[0-9]{2}' | head -1 || true)
CURRENT_TIME=$(python3 "$SCHEDULE_HELPER" read "$TIMER_PATH")
fi
[ -z "$CURRENT_TIME" ] && CURRENT_TIME="(未知)"
......@@ -139,27 +144,19 @@ echo " 定时时间: $CURRENT_TIME → $TIME"
[ -n "$COUNT" ] && { echo " 抓取条数: ${CUR_COUNT:-(继承default)} → $COUNT"; }
[ -n "$MAIL_TO" ] && { echo " 收件人: ${CUR_MAIL_TO:-(继承default)} → $MAIL_TO"; }
[ -n "$MAIL_ON_SUCCESS" ] && { echo " 成功发信: ${CUR_MOS:-(继承default)} → $MAIL_ON_SUCCESS"; }
[ -n "$PROVIDER" ] && { echo " Provider: ${CUR_PROVIDER:-(继承default)} → $PROVIDER"; }
[ -n "$FALLBACK_PROVIDER" ] && { echo " Fallback: ${CUR_FALLBACK:-(未设置)} → $FALLBACK_PROVIDER"; }
[ "$PROVIDER_SET" = "1" ] && { echo " Provider: ${CUR_PROVIDER:-(继承default)} → ${PROVIDER:-(继承default)}"; }
[ "$FALLBACK_PROVIDER_SET" = "1" ] && { echo " Fallback: ${CUR_FALLBACK:-(继承default)} → ${FALLBACK_PROVIDER:-(继承default)}"; }
echo " timer 文件: $TIMER_PATH"
if [ "$HAS_DROPIN" = "1" ]; then
warn "检测到 drop-in 覆盖目录: $DROPIN_DIR"
echo " 这会覆盖主文件的 OnCalendar,将被清除以确保新时间生效。"
echo " 将仅清除其中的 OnCalendar 行，保留其他设置，确保新时间生效。"
fi
echo ""
confirm "确认修改?" || { echo "已取消。"; exit 0; }
# --- 改 timer 文件的 OnCalendar + Description ---
# 先清除 drop-in 覆盖目录(若有),否则会覆盖主文件的新 OnCalendar
if [ "$HAS_DROPIN" = "1" ]; then
rm -rf "$DROPIN_DIR"
warn "已清除 drop-in 目录: $DROPIN_DIR"
fi
# 用 sed 原地替换 OnCalendar 行
sed -i -E "s|^OnCalendar=.*|OnCalendar=*-*-* ${TIME}:00|" "$TIMER_PATH"
# 同步更新 Description 里的时间(若有)
sed -i -E "s|^(Description=.* at ).*|\1${TIME}|" "$TIMER_PATH"
ok "已更新 $TIMER_PATH (OnCalendar → ${TIME}:00)"
# --- 替换全部运行时间，并保留无关 drop-in 设置 ---
python3 "$SCHEDULE_HELPER" update "$TIMER_PATH" "$TIME"
ok "已更新 $TIMER_PATH (运行时间 → $TIME)"
# --- 更新 conf(若有 conf 选项) ---
update_conf() {
......@@ -172,7 +169,12 @@ update_conf() {
echo "${key}=${val}" >> "$CONF_PATH"
fi
}
if [ -n "$COUNT" ] || [ -n "$MAIL_TO" ] || [ -n "$MAIL_ON_SUCCESS" ] || [ -n "$PROVIDER" ] || [ -n "$FALLBACK_PROVIDER" ]; then
remove_conf() {
local key="$1"
[ -f "$CONF_PATH" ] || return 0
sed -i -E "/^${key}=/d" "$CONF_PATH"
}
if [ -n "$COUNT" ] || [ -n "$MAIL_TO" ] || [ -n "$MAIL_ON_SUCCESS" ] || [ "$PROVIDER_SET" = "1" ] || [ "$FALLBACK_PROVIDER_SET" = "1" ]; then
mkdir -p "$CONF_DIR"
if [ ! -f "$CONF_PATH" ]; then
{
......@@ -184,8 +186,12 @@ if [ -n "$COUNT" ] || [ -n "$MAIL_TO" ] || [ -n "$MAIL_ON_SUCCESS" ] || [ -n "$P
[ -n "$COUNT" ] && { update_conf COUNT "$COUNT"; }
[ -n "$MAIL_TO" ] && { update_conf MAIL_TO "$MAIL_TO"; }
[ -n "$MAIL_ON_SUCCESS" ] && { update_conf MAIL_ON_SUCCESS "$MAIL_ON_SUCCESS"; }
[ -n "$PROVIDER" ] && { update_conf PROVIDER "$PROVIDER"; }
[ -n "$FALLBACK_PROVIDER" ] && { update_conf FALLBACK_PROVIDER "$FALLBACK_PROVIDER"; }
if [ "$PROVIDER_SET" = "1" ]; then
if [ -n "$PROVIDER" ]; then update_conf PROVIDER "$PROVIDER"; else remove_conf PROVIDER; fi
fi
if [ "$FALLBACK_PROVIDER_SET" = "1" ]; then
if [ -n "$FALLBACK_PROVIDER" ]; then update_conf FALLBACK_PROVIDER "$FALLBACK_PROVIDER"; else remove_conf FALLBACK_PROVIDER; fi
fi
ok "已更新 $CONF_PATH"
fi
......@@ -194,7 +200,7 @@ fi
systemctl stop "$TIMER_NAME" 2>/dev/null || true
systemctl daemon-reload
ok "systemd 已重新加载"
systemctl start "$TIMER_NAME"
systemctl enable --now "$TIMER_NAME"
ok "已重启 $TIMER_NAME(新时间生效)"
echo ""
......
bin/provider.sh
View file @ faf1233d
......@@ -11,7 +11,7 @@
# PROVIDER 为空或 "default" 时不加 --settings,使用系统全局配置。
# 依赖外部变量: CONF_DIR (默认 /etc/news-harvester)
# 设置外部变量: CLAUDE_SETTINGS_ARGS (数组,可能为空)
# 设置外部变量: CLAUDE_SETTINGS_ARGS (数组,可能为空)、ACTIVE_PROVIDER、USED_PROVIDER
PROVIDER_DIR="${PROVIDER_DIR:-${CONF_DIR:-/etc/news-harvester}/providers}"
......@@ -23,6 +23,7 @@ PROVIDER_DIR="${PROVIDER_DIR:-${CONF_DIR:-/etc/news-harvester}/providers}"
build_claude_settings_args() {
local name="${1:-${PROVIDER:-}}"
CLAUDE_SETTINGS_ARGS=()
ACTIVE_PROVIDER="default"
USED_PROVIDER="${name:-default}"
# default 或空 → 不加 --settings
......@@ -63,6 +64,41 @@ build_claude_settings_args() {
local settings_json
settings_json=$(echo "$json_content" | jq -c '{env: .}')
CLAUDE_SETTINGS_ARGS=(--settings "$settings_json")
ACTIVE_PROVIDER="$name"
}
# resolve_max_budget_usd [cli_override]
# 优先级: CLI 临时值 > 当前 provider 独立预算 > MAX_BUDGET_USD > 15。
# PROVIDER_BUDGET_OVERRIDES 格式: provider=金额,provider=金额。
# 结果写入 RESOLVED_MAX_BUDGET_USD，harvest/continue 共用同一规则。
resolve_max_budget_usd() {
local cli_override="${1:-}"
local resolved="${MAX_BUDGET_USD:-15}"
local active="${ACTIVE_PROVIDER:-default}"
local entry key value
if [ -n "${PROVIDER_BUDGET_OVERRIDES:-}" ]; then
local entries=()
IFS=',' read -r -a entries <<< "$PROVIDER_BUDGET_OVERRIDES"
for entry in "${entries[@]}"; do
key="${entry%%=*}"
value="${entry#*=}"
if [ "$entry" = "$key" ] || [ -z "$key" ] || [ -z "$value" ]; then
echo "[provider] Provider 预算格式无效: $entry" >&2
return 1
fi
if [ "$key" = "$active" ]; then
resolved="$value"
fi
done
fi
[ -n "$cli_override" ] && resolved="$cli_override"
if ! awk -v value="$resolved" 'BEGIN { exit !(value ~ /^[0-9]+([.][0-9]+)?$/ && value + 0 > 0) }'; then
echo "[provider] max-budget-usd 应为正数: $resolved" >&2
return 1
fi
RESOLVED_MAX_BUDGET_USD="$resolved"
}
# build_fallback_claude_settings_args
......
bin/result.sh 0 → 100644
View file @ faf1233d
#!/bin/bash
# Shared contract and normalization for harvest.sh / continue.sh.
prepare_result_contract() {
command -v python3 >/dev/null 2>&1 || { echo "需要 python3 来校验采集结果" >&2; return 1; }
command -v jq >/dev/null 2>&1 || { echo "需要 jq 来读取采集结果" >&2; return 1; }
[[ "$COUNT" =~ ^[0-9]+$ ]] && [ "$COUNT" -gt 0 ] || { echo "COUNT 必须是正整数" >&2; return 1; }
RESULT_SCHEMA=$(cat "$SCRIPT_DIR/harvest-result.schema.json") || return 1
RESULT_REQUIREMENTS="
结果输出约定：
- 按所给 JSON Schema 返回完整结果，source=${SOURCE}，requested_count=${COUNT}，schema_version=1。
- articles 只填写 URL、标题、发布时间、原文归档文件的绝对路径、状态、失败类型、失败原因和已有的质量证据。
- archive_path 复制归档时实际写入的路径，不要按标题推测或重新拼接，也不要使用 shell 转义或 Markdown 链接格式。
- 不要在结构化输出中重复正文、摘要、分类或标签；外层会从 archive_path 指向的 Obsidian 原文文件读取这些数据并计算正文长度。
- 最终通过结构化输出返回 JSON。"
}
normalize_result() {
# Keep exact stdout even when it is malformed or empty; do not match *.json scans.
cp "$JSON_OUT" "${JSON_OUT%.json}.claude.raw" || return 1
local normalize_args=(
normalize "$JSON_OUT"
--source "$SOURCE"
--count "$COUNT"
--exit-code "$RC"
--work-dir "$WORK_DIR"
)
if [ -n "${OBSIDIAN_VAULT_PATH:-}" ]; then
normalize_args+=(--vault-root "$OBSIDIAN_VAULT_PATH")
fi
python3 "$SCRIPT_DIR/harvest-result.py" "${normalize_args[@]}" || return 1
RC=$(jq -r '.exit_code' "$JSON_OUT")
}
bin/schedule.py 0 → 100644
View file @ faf1233d
#!/usr/bin/env python3
"""Daily schedules shared by management scripts and the web panel."""
import argparse
import re
from pathlib import Path
def parse_times(value):
parts = [part.strip() for part in value.split(",")]
if not parts or any(not re.fullmatch(r"(?:[01][0-9]|2[0-3]):[0-5][0-9]", p) for p in parts):
raise ValueError("时间格式应为 HH:MM，多个时间用逗号分隔，如 09:00,14:30,20:00")
return sorted(set(parts))
def calendar_times(specs):
return sorted(set(time for spec in specs for time in
re.findall(r"(?<![0-9])(?:[01][0-9]|2[0-3]):[0-5][0-9](?![0-9])", spec)))
def calendar_specs(text):
"""Read only [Timer] entries and honor an empty OnCalendar reset."""
specs = []
section = ""
for line in text.splitlines():
stripped = line.strip()
if stripped.startswith("["):
section = stripped
if section == "[Timer]" and re.match(r"OnCalendar\s*=", stripped):
value = stripped.split("=", 1)[1].strip()
if value:
specs.append(value)
else:
specs.clear()
return specs
def read_timer_times(path):
# Include per-instance drop-ins for offline fallback and auto scheduling.
text = path.read_text(encoding="utf-8")
dropins = path.with_name(path.name + ".d")
for conf in sorted(dropins.glob("*.conf")):
text += "\n" + conf.read_text(encoding="utf-8")
return calendar_times(calendar_specs(text))
def replace_calendar(text, times=None):
lines = []
section = ""
inserted = False
for line in text.splitlines():
stripped = line.strip()
if stripped.startswith("["):
section = stripped
if section == "[Timer]" and re.match(r"OnCalendar\s*=", stripped):
continue
lines.append(line)
if stripped == "[Timer]" and times is not None and not inserted:
lines.extend(f"OnCalendar=*-*-* {time}:00" for time in times)
inserted = True
if times is not None and not inserted:
lines.extend(["", "[Timer]"] + [f"OnCalendar=*-*-* {t}:00" for t in times])
return "\n".join(lines) + "\n"
def update_timer(path, times):
text = replace_calendar(path.read_text(encoding="utf-8"), times)
text = re.sub(r"^(Description=.* at ).*$", lambda m: m[1] + ",".join(times), text, flags=re.MULTILINE)
# Remove only calendar overrides, preserving unrelated timer settings.
for conf in sorted(path.with_name(path.name + ".d").glob("*.conf")):
old = conf.read_text(encoding="utf-8")
new = replace_calendar(old)
if new != old:
conf.write_text(new, encoding="utf-8")
path.write_text(text, encoding="utf-8")
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("command", choices=["normalize", "calendar", "read", "suggest", "update"])
parser.add_argument("value")
parser.add_argument("times", nargs="?")
args = parser.parse_args()
try:
if args.command in ("normalize", "calendar"):
times = parse_times(args.value)
print(",".join(times) if args.command == "normalize" else
"\n".join(f"OnCalendar=*-*-* {t}:00" for t in times))
elif args.command == "read":
print(",".join(read_timer_times(Path(args.value))))
elif args.command == "update":
update_timer(Path(args.value), parse_times(args.times or ""))
else:
times = [t for path in Path(args.value).glob("news-harvester@*.timer") for t in read_timer_times(path)]
if times:
hour, minute = map(int, max(times).split(":"))
minute = (hour * 60 + minute + 15) % 1440
print(f"{minute // 60:02d}:{minute % 60:02d}")
else:
print("09:00")
except (ValueError, OSError) as exc:
parser.error(str(exc))
if __name__ == "__main__":
main()
bin/status.sh
View file @ faf1233d
......@@ -153,14 +153,14 @@ show_overview() {
# NextElapse 格式: Wed 2026-08-27 09:00:00 CST
local short sort_key run_fmt
# 先从 OnCalendar 提取(直接取 HH:MM,兼容带秒/不带秒两种写法)
short=$(echo "$on_cal" | grep -oE '[0-9]{2}:[0-9]{2}' | head -1)
short=$(echo "$on_cal" | grep -oE '[0-9]{2}:[0-9]{2}' | sort -u | paste -sd, -)
# 如果 OnCalendar 没有,从 NextElapse 提取(去掉星期和日期,只留时间)
if [ -z "$short" ] && [ -n "$next_el" ] && [ "$next_el" != "n/a" ]; then
short=$(echo "$next_el" | grep -oE '[0-9]{2}:[0-9]{2}' | head -1)
fi
if [ -n "$short" ]; then
sort_key="$short"
sort_key="${short%%,*}"
run_fmt="$short"
else
sort_key="99:99"
......
config/default.conf.example
View file @ faf1233d
......@@ -16,6 +16,13 @@ MAIL_TO=you@example.com
MAIL_ON_SUCCESS=1
# === 成本告警 ===
# Claude 正常采集和续采/修复的单次预算上限(美元)
# 单栏目可在 <source>.conf 中覆盖；continue.sh 的 --budget 可临时覆盖。
MAX_BUDGET_USD=15
# 某些 provider 单价较高或任务需要更宽松预算时，可按实际使用的 provider 单独覆盖。
# 格式为 provider=美元，多个用逗号分隔，不要加入空格；未列出的 provider 使用上面的默认值。
# PROVIDER_BUDGET_OVERRIDES=easyrouter-claude=25,glm=30
# 单次抓取成本(美元)超过此值则在邮件标题标记 [ALERT],留空不判断
ALERT_COST_OVER=1
......@@ -47,3 +54,5 @@ DIGEST_BUDGET_USD=2
# === 路径(通常不用改) ===
# WORK_DIR=/home/yichu/Documents/News-Harvester
# LOG_DIR=/home/yichu/Documents/News-Harvester/logs
# Obsidian Vault 根目录；设置后会拒绝读取该目录之外的 archive_path
# OBSIDIAN_VAULT_PATH=/home/yichu/Documents/Obsidian/MyVault
systemd/news-harvester@.service
View file @ faf1233d
......@@ -14,7 +14,7 @@ Group=yichu
WorkingDirectory=/home/yichu/Documents/News-Harvester
# 配置加载顺序:default.conf 提供全局默认,%i.conf 覆盖/补充单栏目设置。
# 同名变量后者覆盖前者(systemd 规则)。文件不存在不报错(前导 -)。
# 可用变量:COUNT、MAIL_TO、MAIL_ON_SUCCESS、ALERT_COST_OVER、WORK_DIR、LOG_DIR
# 可用变量:COUNT、MAIL_TO、MAIL_ON_SUCCESS、MAX_BUDGET_USD、PROVIDER_BUDGET_OVERRIDES、ALERT_COST_OVER、WORK_DIR、LOG_DIR
EnvironmentFile=-/etc/news-harvester/default.conf
EnvironmentFile=-/etc/news-harvester/%i.conf
# %i=栏目名, ${COUNT}=来自 conf 的条数(未设则为空,脚本兜底 20)
......
tests/test_current_version.py 0 → 100644
View file @ faf1233d
"""Offline regression tests: no Claude API, systemd changes, mail or real articles."""
import copy
import importlib.util
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import time
import unittest
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'bin'))
import schedule
def load_module(name, path):
spec = importlib.util.spec_from_file_location(name, path)
module = importlib.util.module_from_spec(spec)
sys.modules[name] = module
spec.loader.exec_module(module)
return module
result_module = load_module('harvest_result', ROOT / 'bin/harvest-result.py')
def article(**changes):
value = dict(url='https://example.com/news/1', title='测试新闻', published_at='',
archive_path='', status='SUCCESS',
failure_type=None, failure_reason='', quality_evidence=[])
value.update(changes)
return value
def envelope(articles=None, count=1):
result = result_module.empty_result('sample', count)
result.update(status='SUCCESS', articles=articles if articles is not None else [])
return dict(result='原始文字结果', structured_output=result, is_error=False,
subtype='success', session_id='fixture-session', total_cost_usd=0.01,
num_turns=1, duration_api_ms=100, permission_denials=[], usage={})
class ArchiveFixtureMixin:
def write_archive(self, body='这是测试文章的完整正文。' * 50, name=None,
title='测试新闻', url='https://example.com/news/1'):
self.archive_index = getattr(self, 'archive_index', 0) + 1
path = self.base / (name or f'测试新闻{self.archive_index}原文.md')
long_abstract = 'This is an English abstract that must not count as article body. ' * 20
path.write_text(f'''---
title: "{title}"
date: 2026-09-08T04:00:17.866Z
tags:
- 芯片
- 人工智能
category: 科技
source: "{url}"
publisher: 测试媒体
authors: "测试作者"
type: 原文
related: "[[测试新闻中文]]"
saved: 2026-09-08
---
# {title}
> {long_abstract}
---
## 正文
{body}
---
[[测试新闻]]
''', encoding='utf-8')
return path
def make_article(self, body='这是测试文章的完整正文。' * 50, **changes):
url = changes.get('url', 'https://example.com/news/1')
title = changes.get('title', '测试新闻')
archive_path = changes.pop('archive_path', None)
if archive_path is None:
archive_path = str(self.write_archive(body, title=title, url=url))
return article(archive_path=archive_path, **changes)
class ResultTests(ArchiveFixtureMixin, unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory(prefix='nh-result-test-')
self.addCleanup(self.tmp.cleanup)
self.base = Path(self.tmp.name)
def normalize(self, value, count=1, rc=0):
return result_module.normalize(value, 'sample', count, rc, self.base, self.base)
def test_success_recomputes_counts_length_preserves_metadata(self):
item = self.make_article()
output = self.normalize(envelope([item]))
result = output['result']
self.assertEqual(result['status'], 'SUCCESS')
self.assertEqual(result['succeeded_count'], 1)
self.assertEqual(result['articles'][0]['archive']['body_length'],
result_module.body_length('这是测试文章的完整正文。' * 50))
self.assertEqual(result['articles'][0]['archive']['properties']['category'], '科技')
self.assertEqual(result['articles'][0]['archive']['properties']['tags'], ['芯片', '人工智能'])
self.assertEqual(output['exit_code'], 0)
self.assertFalse(output['is_error'])
self.assertEqual(output['session_id'], 'fixture-session')
self.assertEqual(output['result_text'], '原始文字结果')
result_module.validate(envelope([item])['structured_output'],
json.loads(result_module.SCHEMA_PATH.read_text()))
def test_null_failure_reason_is_accepted_and_normalized(self):
output = self.normalize(envelope([self.make_article(failure_reason=None)]))
self.assertEqual(output['result']['status'], 'SUCCESS')
self.assertEqual(output['result']['articles'][0]['failure_reason'], '')
def test_half_short_articles_trigger_task_paywall(self):
items = [self.make_article('短文' * 100),
self.make_article(url='https://example.com/news/2')]
output = self.normalize(envelope(items, count=2), count=2)
result = output['result']
self.assertEqual(result['status'], 'FAILED')
self.assertIn('PAYWALL', [failure['type'] for failure in result['failures']])
self.assertIn('1/2 篇可读取的归档原文正文少于 500 字（50%）', result['warnings'][-1])
self.assertEqual([item['status'] for item in result['articles']], ['SUCCESS', 'SUCCESS'])
self.assertEqual(result['succeeded_count'], 2)
def test_fewer_than_half_short_articles_do_not_trigger(self):
items = [self.make_article('短文' * 100),
self.make_article(url='https://example.com/news/2'),
self.make_article(url='https://example.com/news/3')]
output = self.normalize(envelope(items, count=3), count=3)
self.assertEqual(output['result']['status'], 'SUCCESS')
self.assertNotIn('PAYWALL', [failure['type'] for failure in output['result']['failures']])
def test_outer_layer_does_not_match_paywall_keywords(self):
body = 'Subscribe to continue reading. ' + '这是完整正文。' * 100
output = self.normalize(envelope([self.make_article(body)]))
self.assertEqual(output['result']['status'], 'SUCCESS')
def test_body_length_excludes_whitespace(self):
self.assertEqual(result_module.body_length('中 文\n a'), 3)
def test_explicit_incomplete_failure_preserved(self):
item = article(status='FAILED', failure_type='CONTENT_INCOMPLETE', failure_reason='仅取得前半段')
output = self.normalize(envelope([item]))
self.assertEqual(output['result']['failures'][0]['type'], 'CONTENT_INCOMPLETE')
def test_partial_failure_keeps_successful_articles(self):
failed = article(url='https://example.com/2', status='FAILED',
failure_type='FETCH_FAILED', failure_reason='正文抓取失败')
output = self.normalize(envelope([self.make_article(), failed], count=2), count=2)
self.assertEqual(output['result']['succeeded_count'], 1)
self.assertEqual(output['result']['failed_count'], 1)
self.assertEqual(output['result']['status'], 'FAILED')
def test_one_failure_does_not_fail_a_majority_successful_run(self):
items = [self.make_article(url=f'https://example.com/news/{index}')
for index in range(1, 10)]
items.append(article(url='https://example.com/news/10', title='Subscribe to read',
status='FAILED', failure_type='PAYWALL',
failure_reason='页面仅提供订阅引导'))
value = envelope(items, count=10)
value['structured_output']['status'] = 'FAILED'
value['structured_output']['failures'].append({
'type': 'PAYWALL', 'message': '页面仅提供订阅引导',
'url': 'https://example.com/news/10',
})
output = self.normalize(value, count=10)
result = output['result']
self.assertEqual(result['status'], 'SUCCESS')
self.assertEqual(result['succeeded_count'], 9)
self.assertEqual(result['failed_count'], 1)
self.assertEqual(output['exit_code'], 0)
self.assertIn('PAYWALL', [failure['type'] for failure in result['failures']])
def test_invalid_contracts_never_succeed(self):
for mutate in [lambda e: e.pop('structured_output'),
lambda e: e['structured_output'].update(articles='bad'),
lambda e: e['structured_output']['articles'][0].update(failure_type='NOT_AN_ENUM'),
lambda e: e['structured_output'].update(source='other'),
lambda e: e['structured_output']['articles'][0].update(body='模型重复返回的正文')]:
value = envelope([self.make_article()])
mutate(value)
output = self.normalize(value)
self.assertEqual(output['exit_code'], 1)
self.assertIn('INVALID_OUTPUT', [f['type'] for f in output['result']['failures']])
def test_runtime_and_permissions_override_success(self):
value = envelope([self.make_article()])
for candidate, rc in [(value, 42), ({**value, 'is_error': True}, 0),
({**value, 'permission_denials': [{}]}, 0)]:
self.assertEqual(self.normalize(candidate, rc=rc)['result']['status'], 'FAILED')
def test_exhaustion_requires_explicit_evidence(self):
value = envelope([], count=20)
self.assertEqual(self.normalize(value, count=20)['exit_code'], 1)
value['structured_output'].update(source_exhausted=True)
self.assertEqual(self.normalize(value, count=20)['exit_code'], 1)
value['structured_output'].update(exhaustion_reason='已检查列表，无新增未归档文章')
self.assertEqual(self.normalize(value, count=20)['exit_code'], 0)
def test_duplicate_urls_fail(self):
output = self.normalize(envelope([self.make_article(), self.make_article()], 2), 2)
self.assertEqual(output['exit_code'], 1)
self.assertEqual(output['result']['failed_count'], 1)
def test_contradictory_article_status_is_counted_as_failure(self):
output = self.normalize(envelope([self.make_article(failure_type='CONTENT_INCOMPLETE', failure_reason='正文截断')]))
self.assertEqual(output['result']['articles'][0]['status'], 'FAILED')
self.assertEqual(output['result']['failed_count'], 1)
def test_archive_must_be_original_and_inside_vault(self):
summary = self.write_archive('摘要内容', name='测试新闻摘要.md')
output = self.normalize(envelope([article(archive_path=str(summary))]))
self.assertIn('ARCHIVE_FAILED', [failure['type'] for failure in output['result']['failures']])
outside = Path(self.tmp.name).parent / '越界测试原文.md'
outside.write_text('---\ntype: 原文\n---\n## 正文\n正文', encoding='utf-8')
self.addCleanup(outside.unlink, missing_ok=True)
output = self.normalize(envelope([article(archive_path=str(outside))]))
self.assertIn('超出配置的 Obsidian Vault', output['result']['articles'][0]['failure_reason'])
def test_archive_path_repairs_shell_escapes_and_sanitized_slash(self):
name = "South Korea's UBATT targets 500 Wh-kg lithium-metal batteries - 电子时报原文.md"
actual = self.write_archive(name=name, url='https://example.com/ubatt')
emitted = str(actual).replace(' ', r'\ ').replace("'", r"\\'").replace('Wh-kg', 'Wh/kg')
output = self.normalize(envelope([
article(url='https://example.com/ubatt', archive_path=emitted)
]))
item = output['result']['articles'][0]
self.assertEqual(item['status'], 'SUCCESS')
self.assertEqual(item['archive_path'], str(actual.resolve()))
def test_archive_path_repairs_markdown_wrapped_filename(self):
name = 'Nepal rescue focused on workers trapped - 路透社世界原文.md'
actual = self.write_archive(name=name, url='https://example.com/nepal')
emitted = str(actual.parent / 'Nepal rescue focused on workers trapped - ') + \
'[路透社世界原文.md](http://xn--example.md/)'
output = self.normalize(envelope([
article(url='https://example.com/nepal', archive_path=emitted)
]))
item = output['result']['articles'][0]
self.assertEqual(item['status'], 'SUCCESS')
self.assertEqual(item['archive_path'], str(actual.resolve()))
def test_summary_is_article_list_without_repeated_tip_prefix(self):
item = self.make_article(quality_evidence=['归档完成', 'push updated'])
result = self.normalize(envelope([item]))['result']
result['warnings'] = ['提示：本批处理完成']
summary = result_module.format_summary(result)
self.assertIn('- 测试新闻 | 测试新闻中文 | 归档 ✓ | 推送 updated', summary)
self.assertIn('说明：\n- 本批处理完成', summary)
self.assertNotIn('提示：', summary)
class TemporaryEnvironment(ArchiveFixtureMixin, unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory(prefix='nh-test-')
self.addCleanup(self.tmp.cleanup)
self.base = Path(self.tmp.name)
self.bin = self.base / 'bin'
self.bin.mkdir()
for path in (ROOT / 'bin').iterdir():
if path.is_file():
shutil.copy2(path, self.bin / path.name)
self.mock_bin = self.base / 'mock-bin'
self.mock_bin.mkdir()
self.logs = self.base / 'logs'
self.logs.mkdir()
self.units = self.base / 'units'
self.units.mkdir()
self.conf = self.base / 'conf'
self.conf.mkdir()
self.fixture = self.base / 'fixture.json'
self.env = {**os.environ, 'PATH': f'{self.mock_bin}:{os.environ["PATH"]}',
'WORK_DIR': str(self.base), 'LOG_DIR': str(self.logs),
'OBSIDIAN_VAULT_PATH': str(self.base),
'CONF_DIR': str(self.conf), 'SYSTEMD_DIR': str(self.units),
'PROVIDER': 'default', 'FALLBACK_PROVIDER': '', 'MAIL_TO': '',
'NH_FIXTURE': str(self.fixture), 'NH_CALLS': str(self.base / 'calls'),
'DIGEST_RETAIN_DAYS': '0'}
self.executable('claude', '''#!/usr/bin/env python3
import json, os, sys
from pathlib import Path
args = sys.argv[1:]
assert '--json-schema' in args
json.loads(args[args.index('--json-schema') + 1])
Path(os.environ['NH_CALLS']).write_text(json.dumps(args))
print(Path(os.environ['NH_FIXTURE']).read_text())
sys.exit(int(os.environ.get('NH_RC', '0')))
''')
self.executable('id', '#!/bin/sh\necho 0\n')
self.executable('systemctl', '#!/bin/sh\nprintf "%s\\n" "$*" >> "$NH_CALLS.systemctl"\n')
mail = self.bin / 'notify-mail.sh'
mail.write_text('#!/bin/sh\ncat > "$NH_CALLS.mail"\n')
def executable(self, name, text):
path = self.mock_bin / name
path.write_text(text)
path.chmod(0o755)
def run_script(self, name, *args):
return subprocess.run(['bash', str(self.bin / name), *args], env=self.env,
text=True, capture_output=True, timeout=15)
def save_fixture(self, value):
self.fixture.write_text(json.dumps(value, ensure_ascii=False))
class ScriptTests(TemporaryEnvironment):
def test_harvest_failure_log_mail_and_raw(self):
self.save_fixture(envelope([self.make_article('Subscribe to continue reading')]))
self.env.update(MAIL_TO='test@example.invalid', MAIL_ON_SUCCESS='0')
process = self.run_script('harvest.sh', 'sample', '1')
self.assertEqual(process.returncode, 1, process.stderr)
result = json.loads(next(self.logs.glob('*.json')).read_text())
self.assertEqual(result['result']['failed_count'], 0)
self.assertEqual(result['result']['status'], 'FAILED')
self.assertIn('PAYWALL', [failure['type'] for failure in result['result']['failures']])
self.assertIn('失败', (self.logs / 'sample.last.log').read_text())
self.assertIn('PAYWALL', (self.base / 'calls.mail').read_text())
raw = next(self.logs.glob('*.claude.raw'))
self.assertEqual(json.loads(raw.read_text()), json.loads(self.fixture.read_text()))
def test_harvest_success_and_environment_count(self):
self.save_fixture(envelope([self.make_article(quality_evidence=['push updated'])]))
self.env['COUNT'] = '1'
self.env['MAX_BUDGET_USD'] = '7.5'
self.env['PROVIDER_BUDGET_OVERRIDES'] = 'default=9.25,costly=30'
process = self.run_script('harvest.sh', 'sample')
self.assertEqual(process.returncode, 0, process.stderr)
args = json.loads((self.base / 'calls').read_text())
self.assertEqual(args[args.index('--max-budget-usd') + 1], '9.25')
self.assertIn('成功', (self.logs / 'sample.last.log').read_text())
self.assertIn('测试新闻 | 测试新闻中文 | 归档 ✓ | 推送 updated',
(self.logs / 'sample.last.log').read_text())
def test_malformed_or_empty_stdout_fail(self):
for raw in ['', 'not JSON', '[]']:
self.fixture.write_text(raw)
process = self.run_script('harvest.sh', 'sample', '1')
self.assertEqual(process.returncode, 1, process.stderr)
result = json.loads(next(self.logs.glob('*.json')).read_text())
self.assertEqual(result['result']['failures'][0]['type'], 'INVALID_OUTPUT')
def test_continue_and_repair_share_contract(self):
initial = result_module.normalize(envelope([self.make_article()]), 'sample', 1, 0,
self.base, self.base)
(self.logs / 'sample.20260101.090000.json').write_text(json.dumps(initial))
self.save_fixture(envelope([self.make_article()]))
provider_dir = self.conf / 'providers'
provider_dir.mkdir()
(provider_dir / 'backup.json').write_text('{}')
(self.conf / 'default.conf').write_text(
'MAX_BUDGET_USD=6.25\n'
'PROVIDER_BUDGET_OVERRIDES=backup=11.5\n'
'FALLBACK_PROVIDER=backup\n'
)
for mode in ['resume', 'repair']:
process = self.run_script('continue.sh', 'sample', '--mode', mode, '--yes')
self.assertEqual(process.returncode, 0, process.stderr)
args = json.loads((self.base / 'calls').read_text())
self.assertEqual(args[args.index('--resume') + 1], 'fixture-session')
self.assertEqual(args[args.index('--max-budget-usd') + 1], '11.5')
self.assertIn('requested_count=1', args[-1])
process = self.run_script('continue.sh', 'sample', '--mode', 'resume',
'--budget', '8', '--yes')
self.assertEqual(process.returncode, 0, process.stderr)
args = json.loads((self.base / 'calls').read_text())
self.assertEqual(args[args.index('--max-budget-usd') + 1], '8')
def test_harvest_and_continue_reject_invalid_budget(self):
self.save_fixture(envelope([self.make_article()]))
self.env['MAX_BUDGET_USD'] = '0'
process = self.run_script('harvest.sh', 'sample', '1')
self.assertNotEqual(process.returncode, 0)
self.assertIn('max-budget-usd 应为正数', process.stderr)
def test_digest_retains_failures_without_sending_full_bodies_to_model(self):
today = time.strftime('%Y%m%d')
output = result_module.normalize(envelope([self.make_article('短文' * 100)]),
'sample', 1, 0, self.base, self.base)
(self.logs / f'sample.continue.{today}.143000.json').write_text(json.dumps(output))
(self.logs / f'sample.{today}.090000.json').write_text(json.dumps(
envelope([self.make_article()]) | {'structured_output': None}))
self.save_fixture({'structured_output': {'alert': False, 'alert_reasons': [], 'anomalies': [], 'recommendations': [], 'failed_sources': []}})
process = self.run_script('daily-digest.sh')
self.assertEqual(process.returncode, 0, process.stderr)
summary = json.loads((self.logs / f'.digest-summary.{today}.json').read_text())
failed = next(row for row in summary if row['time'] == '143000')
self.assertEqual(failed['status'], 'FAILED')
self.assertNotIn('body', failed['result']['articles'][0])
self.assertNotIn('body', failed['result']['articles'][0]['archive'])
self.assertIn('采集结果校验失败', (self.logs / f'digest.{today}.md').read_text())
class ScheduleTests(TemporaryEnvironment):
def test_parse_times_and_reject_injection(self):
self.assertEqual(schedule.parse_times('20:00,09:00, 20:00'), ['09:00', '20:00'])
for value in ['24:00', '09:60', '9:00', '09:00,', '09:00\nOnCalendar=daily', '']:
with self.assertRaises(ValueError):
schedule.parse_times(value)
def test_add_modify_preserve_overrides_and_all_times(self):
(self.units / 'news-harvester@.service').write_text('[Service]\nExecStart=/bin/true\n')
process = self.run_script('add-source.sh', 'sample', '--time', '20:00,09:00', '--time', '14:30,09:00', '--count', '3', '--yes')
self.assertEqual(process.returncode, 0, process.stderr + process.stdout)
timer = self.units / 'news-harvester@sample.timer'
conf = self.conf / 'sample.conf'
self.assertIn('COUNT=3', conf.read_text())
self.assertNotIn('PROVIDER=', conf.read_text())
self.assertNotIn('FALLBACK_PROVIDER=', conf.read_text())
self.assertEqual(schedule.read_timer_times(timer), ['09:00', '14:30', '20:00'])
dropin = timer.with_name(timer.name + '.d')
dropin.mkdir()
overrides = dropin / 'override.conf'
overrides.write_text('[Timer]\nOnCalendar=\nOnCalendar=*-*-* 22:00:00\nRandomizedDelaySec=30\n')
process = self.run_script('modify-source.sh', 'sample', '--time', '08:15,18:45', '--yes')
self.assertEqual(process.returncode, 0, process.stderr + process.stdout)
self.assertEqual(schedule.read_timer_times(timer), ['08:15', '18:45'])
self.assertIn('RandomizedDelaySec=30', overrides.read_text())
self.assertNotIn('OnCalendar=', overrides.read_text())
process = self.run_script('modify-source.sh', 'sample', '--time', '07:00', '--yes')
self.assertEqual(process.returncode, 0, process.stderr)
self.assertEqual(schedule.read_timer_times(timer), ['07:00'])
self.assertIn('enable --now', (self.base / 'calls.systemctl').read_text())
conf.write_text(conf.read_text() + 'PROVIDER=default\nFALLBACK_PROVIDER=default\n')
process = self.run_script('modify-source.sh', 'sample', '--time', '07:00',
'--provider', 'default', '--fallback-provider', 'default', '--yes')
self.assertEqual(process.returncode, 0, process.stderr)
self.assertNotIn('PROVIDER=', conf.read_text())
self.assertNotIn('FALLBACK_PROVIDER=', conf.read_text())
def test_add_ignores_legacy_default_provider_arguments(self):
(self.units / 'news-harvester@.service').write_text('[Service]\nExecStart=/bin/true\n')
process = self.run_script('add-source.sh', 'legacy', '--time', '10:00', '--count', '2',
'--provider', 'default', '--fallback-provider', 'default', '--yes')
self.assertEqual(process.returncode, 0, process.stderr + process.stdout)
conf = (self.conf / 'legacy.conf').read_text()
self.assertEqual([line for line in conf.splitlines() if line and not line.startswith('#')],
['COUNT=2'])
def test_suggest_uses_latest_of_all_times_and_wraps_midnight(self):
timer = self.units / 'news-harvester@other.timer'
timer.write_text('[Timer]\nOnCalendar=*-*-* 09:00:00\nOnCalendar=*-*-* 23:50:00\n')
process = subprocess.run([sys.executable, str(self.bin / 'schedule.py'), 'suggest', str(self.units)], capture_output=True, text=True)
self.assertEqual(process.stdout.strip(), '00:05')
try:
import flask
except ImportError:
flask = None
@unittest.skipUnless(flask, 'Install Flask to run route tests')
class WebTests(TemporaryEnvironment):
def setUp(self):
super().setUp()
self.web = load_module('web_test_app', ROOT / 'web/app.py')
self.web.SYSTEMD_DIR = self.units
self.web.LOG_DIR = self.logs
self.web.CONF_DIR = self.conf
self.web.WEB_PASS = ''
self.web.app.config['TESTING'] = True
self.client = self.web.app.test_client()
self.timer = self.units / 'news-harvester@sample.timer'
self.timer.write_text('[Timer]\nOnCalendar=*-*-* 09:00:00\nOnCalendar=*-*-* 20:00:00\n')
patcher = patch.object(self.web.subprocess, 'run', return_value=subprocess.CompletedProcess([], 0, '', ''))
patcher.start()
self.addCleanup(patcher.stop)
def test_structured_preview_escapes_body(self):
item = self.make_article('<script>alert(1)</script>' + '正文' * 200)
value = result_module.normalize(envelope([item]), 'sample', 1, 0,
self.base, self.base)
(self.logs / 'sample.20260907.090000.json').write_text(json.dumps(value))
response = self.client.get('/source/sample')
self.assertEqual(response.status_code, 200)
html = response.get_data(as_text=True)
self.assertIn('09:00,20:00', html)
self.assertIn('预览归档原文与属性', html)
self.assertIn('&lt;script&gt;', html)
self.assertNotIn('<script>alert(1)</script>', html)
def test_legacy_and_latest_log_by_mtime(self):
old = self.logs / 'sample.continue.20260901.120000.json'
new = self.logs / 'sample.20260907.090000.json'
old.write_text(json.dumps({'result': '旧的续采记录'}))
new.write_text(json.dumps({'result': '新的采集记录'}))
os.utime(old, (1, 1))
self.assertEqual(self.web.get_latest_json('sample')['path'], str(new))
html = self.client.get('/source/sample').get_data(as_text=True)
self.assertIn('新的采集记录', html)
self.assertIn('未经过新版正文质量核验', html)
def test_forms_accept_many_times_reject_invalid(self):
with patch.object(self.web, 'run_script', return_value=(0, '', '')) as run:
self.assertEqual(self.client.post('/add', data={'source': 'sample', 'time': '09:00,20:00'}).status_code, 302)
self.assertIn('09:00,20:00', run.call_args.args[0])
self.assertNotIn('--provider', run.call_args.args[0])
self.assertNotIn('--fallback-provider', run.call_args.args[0])
self.assertEqual(self.client.post('/edit/sample', data={
'time': '08:00,19:00', 'provider': '', 'fallback_provider': '',
}).status_code, 302)
self.assertEqual(run.call_args.args[0][-4:],
['--provider', 'default', '--fallback-provider', 'default'])
run.reset_mock()
self.client.post('/edit/sample', data={'time': '09:00,25:00'})
run.assert_not_called()
self.assertIn('09:00,20:00', self.client.get('/edit/sample').get_data(as_text=True))
def test_effective_timer_multiple_expressions(self):
show = 'TimersCalendar={ OnCalendar=*-*-* 20:00:00 ; next_elapse=Fri 2026-09-11 20:00:00 CST } { OnCalendar=*-*-* 09:00:00 ; next_elapse=Fri 2026-09-11 09:00:00 CST }'
with patch.object(self.web.subprocess, 'run', return_value=subprocess.CompletedProcess([], 0, show, '')):
self.assertEqual(self.web.read_oncalendar_from_systemd('sample'), '09:00,20:00')
def test_disable_clears_all_times_and_overrides(self):
dropin = self.timer.with_name(self.timer.name + '.d')
dropin.mkdir()
(dropin / 'override.conf').write_text('[Timer]\nOnCalendar=*-*-* 22:00:00\nAccuracySec=10\n')
response = self.client.post('/disable/sample')
self.assertEqual(response.status_code, 302)
self.assertEqual(schedule.read_timer_times(self.timer), [])
self.assertIn('AccuracySec=10', (dropin / 'override.conf').read_text())
if __name__ == '__main__':
unittest.main()
web/app.py
View file @ faf1233d
......@@ -19,6 +19,7 @@ import json
import os
import re
import subprocess
import sys
import threading
from datetime import datetime
from pathlib import Path
......@@ -34,6 +35,8 @@ PROVIDER_DIR = CONF_DIR / "providers"
WORK_DIR = Path(os.environ.get("WORK_DIR", "/home/yichu/Documents/News-Harvester"))
LOG_DIR = Path(os.environ.get("LOG_DIR", str(WORK_DIR / "logs")))
PROJECT_DIR = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_DIR / "bin"))
from schedule import calendar_times, parse_times, read_timer_times, update_timer
# 可被 deploy.sh 部署的 bin 脚本路径(服务器上)
ADD_SOURCE = str(WORK_DIR / "bin" / "add-source.sh")
......@@ -148,17 +151,19 @@ def get_timer_files():
def parse_timer_file(timer_path):
"""从 timer 文件解析 OnCalendar 时间"""
"""从 timer 文件及实例 drop-in 读取全部日运行时间。"""
try:
text = timer_path.read_text(encoding="utf-8", errors="replace")
return ",".join(read_timer_times(timer_path)) or "-"
except OSError:
return "-"
m = re.search(r"^OnCalendar=.*?\d{2}:\d{2}(?::\d{2})?", text, re.MULTILINE)
if m:
time_match = re.search(r"(\d{2}):(\d{2})(?::\d{2})?", m.group(0))
if time_match:
return f"{time_match.group(1)}:{time_match.group(2)}"
return "-"
def valid_run_times(value):
try:
parse_times(value)
return True
except ValueError:
return False
def read_oncalendar_from_systemd(timer_name):
......@@ -166,7 +171,7 @@ def read_oncalendar_from_systemd(timer_name):
先读直接属性 OnCalendar;多数版本该属性输出为空,再从
TimersCalendar={ OnCalendar=*-*-* 22:00:00 ; next_elapse=... CST } 中提取。
秒可以省略(OnCalendar=*-*-* 22:00 合法),多个表达式时取最后一个。"""
秒可以省略，多个表达式按时间排序去重。"""
try:
out = subprocess.run(
["systemctl", "show", timer_name, "-p", "OnCalendar", "-p", "TimersCalendar"],
......@@ -175,10 +180,11 @@ def read_oncalendar_from_systemd(timer_name):
specs = [s for s in re.findall(r"^OnCalendar=(.*)$", out, re.MULTILINE) if s.strip()]
if not specs:
specs = re.findall(r"OnCalendar=([^;}\n]*)", out)
for spec in reversed(specs):
m = re.search(r"(\d{2}):(\d{2})(?::\d{2})?", spec)
if m:
return f"{m.group(1)}:{m.group(2)}"
times = calendar_times(specs)
if times:
return ",".join(times)
if specs or "TimersCalendar=" in out:
return "-"
except Exception:
pass
return None
......@@ -247,12 +253,14 @@ def parse_last_log(source):
def get_latest_json(source):
"""读取最新的 Claude JSON 输出文件"""
pattern = str(LOG_DIR / f"{source}.*.json")
files = sorted(glob.glob(pattern), reverse=True)
files = sorted(glob.glob(pattern), key=lambda path: Path(path).stat().st_mtime_ns, reverse=True)
if not files:
return None
try:
with open(files[0], encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, dict):
return None
return {"path": files[0], "data": data}
except Exception:
return None
......@@ -438,21 +446,6 @@ def source_detail(name):
json_info = get_latest_json(name)
conf = read_conf(name)
# 解析 JSON 中的元数据
json_meta = None
if json_info and json_info["data"]:
d = json_info["data"]
json_meta = {
"cost": d.get("total_cost_usd", "-"),
"turns": d.get("num_turns", "-"),
"duration_api": d.get("duration_api_ms", "-"),
"session_id": d.get("session_id", "-"),
"is_error": d.get("is_error", False),
"terminal": d.get("terminal_reason", "-"),
"usage": d.get("usage", {}),
"permission_denials": d.get("permission_denials", []),
}
# 检查后台 continue 任务状态
continue_status = _check_continue_status(name)
auto_refresh = False
......@@ -469,9 +462,33 @@ def source_detail(name):
log_info = parse_last_log(name)
json_info = get_latest_json(name)
# 解析 JSON 中的元数据
json_meta = None
if json_info and json_info["data"]:
d = json_info["data"]
json_meta = {
"cost": d.get("total_cost_usd", "-"),
"turns": d.get("num_turns", "-"),
"duration_api": d.get("duration_api_ms", "-"),
"session_id": d.get("session_id", "-"),
"is_error": d.get("is_error", False),
"terminal": d.get("terminal_reason", "-"),
"usage": d.get("usage", {}),
"permission_denials": d.get("permission_denials", []),
}
business_result = None
legacy_result = None
if json_info:
value = json_info["data"].get("result")
if isinstance(value, dict) and value.get("schema_version") == 1:
business_result = value
elif isinstance(value, str):
legacy_result = value
return render_template("detail.html", name=name, run_time=run_time,
props=props, log_info=log_info, json_info=json_info,
json_meta=json_meta, conf=conf,
json_meta=json_meta, conf=conf, business_result=business_result, legacy_result=legacy_result,
continue_status=continue_status, auto_refresh=auto_refresh,
providers=get_providers())
......@@ -489,6 +506,7 @@ def add_source():
mail_to = request.form.get("mail_to", "").strip()
mail_on_success = request.form.get("mail_on_success", "").strip()
provider = request.form.get("provider", "").strip()
fallback_provider = request.form.get("fallback_provider", "").strip()
if not source or not re.match(r"^[a-zA-Z0-9_-]+$", source):
flash("栏目名只能含字母、数字、下划线、连字符", "error")
......@@ -496,8 +514,8 @@ def add_source():
cmd = [ADD_SOURCE, source, "--yes"]
if time:
if not re.match(r"^([01]\d|2[0-3]):[0-5]\d$", time):
flash("时间格式应为 HH:MM(24小时制)", "error")
if not valid_run_times(time):
flash("时间格式应为 HH:MM，多个时间用逗号分隔", "error")
return render_template("add.html", form=request.form, providers=get_providers())
cmd += ["--time", time]
if count:
......@@ -508,6 +526,8 @@ def add_source():
cmd += ["--mail-on-success", mail_on_success]
if provider and provider != "default":
cmd += ["--provider", provider]
if fallback_provider and fallback_provider != "default":
cmd += ["--fallback-provider", fallback_provider]
rc, stdout, stderr = run_script(cmd, timeout=30)
if rc == 0:
......@@ -542,8 +562,8 @@ def edit_source(name):
provider = request.form.get("provider", "").strip()
fallback_provider = request.form.get("fallback_provider", "").strip()
if not time or not re.match(r"^([01]\d|2[0-3]):[0-5]\d$", time):
flash("时间格式应为 HH:MM(24小时制)", "error")
if not valid_run_times(time):
flash("时间格式应为 HH:MM，多个时间用逗号分隔", "error")
timer_name = f"news-harvester@{name}.timer"
current_time = read_oncalendar_from_systemd(timer_name) or parse_timer_file(timer_path)
conf = read_conf(name)
......@@ -557,10 +577,10 @@ def edit_source(name):
cmd += ["--mail-to", mail_to]
if mail_on_success:
cmd += ["--mail-on-success", mail_on_success]
if provider:
cmd += ["--provider", provider]
if fallback_provider:
cmd += ["--fallback-provider", fallback_provider]
if "provider" in request.form:
cmd += ["--provider", provider or "default"]
if "fallback_provider" in request.form:
cmd += ["--fallback-provider", fallback_provider or "default"]
rc, stdout, stderr = run_script(cmd, timeout=30)
if rc == 0:
......@@ -689,11 +709,8 @@ def disable_source(name):
messages.append(f"已停用 {timer_name}")
# 清空 OnCalendar(改为永不触发),保留文件
if timer_path.exists():
import re as _re
text = timer_path.read_text(encoding="utf-8", errors="replace")
text = _re.sub(r"^OnCalendar=.*", "OnCalendar=", text, flags=_re.MULTILINE)
timer_path.write_text(text, encoding="utf-8")
messages.append(f"已清空运行时间(OnCalendar 置空,不再触发)")
update_timer(timer_path, [])
messages.append(f"已清空全部运行时间（含实例覆盖配置）")
run_script(["systemctl", "daemon-reload"], timeout=10)
flash(f"栏目 {name} 已停用\n" + "\n".join(messages) + "\n\n文件保留,可通过编辑重新设置时间来启用", "success")
return redirect(url_for("source_detail", name=name))
......
web/templates/add.html
View file @ faf1233d
......@@ -15,12 +15,12 @@
</div>
<div class="form-group">
<label for="time">运行时间 (HH:MM)</label>
<label for="time">运行时间（可多个）</label>
<input type="text" id="time" name="time"
pattern="([01]\d|2[0-3]):[0-5]\d"
pattern="([01][0-9]|2[0-3]):[0-5][0-9](,\s*([01][0-9]|2[0-3]):[0-5][0-9])*"
placeholder="留空自动错开(现有最晚 +15 分钟)"
value="{{ form.time if form else '' }}">
<div class="hint">24 小时制,如 09:30。留空则自动推算。</div>
<div class="hint">24 小时制，多个时间用逗号分隔，如 09:00,14:30,20:00。留空则自动推算。</div>
</div>
<div class="form-group">
......@@ -49,12 +49,23 @@
<div class="form-group">
<label for="provider">Claude Provider</label>
<select id="provider" name="provider">
<option value="default">默认 (系统全局配置)</option>
<option value="" {% if not form or not form.get('provider') %}selected{% endif %}>继承 default.conf（不写栏目配置）</option>
{% for p in providers %}
<option value="{{ p.name }}" {% if form and form.get('provider') == p.name %}selected{% endif %}>{{ p.name }}</option>
{% endfor %}
</select>
<div class="hint">选择该栏目使用的 Claude provider。默认 = 不加 --settings,用系统全局配置。</div>
<div class="hint">留空时不写 PROVIDER，由 default.conf 决定。</div>
</div>
<div class="form-group">
<label for="fallback_provider">备用 Provider (FALLBACK)</label>
<select id="fallback_provider" name="fallback_provider">
<option value="" {% if not form or not form.get('fallback_provider') %}selected{% endif %}>继承 default.conf（不写栏目配置）</option>
{% for p in providers %}
<option value="{{ p.name }}" {% if form and form.get('fallback_provider') == p.name %}selected{% endif %}>{{ p.name }}</option>
{% endfor %}
</select>
<div class="hint">留空时不写 FALLBACK_PROVIDER，由 default.conf 决定。</div>
</div>
<div class="actions">
......
web/templates/detail.html
View file @ faf1233d
......@@ -121,6 +121,48 @@
<p style="color:#666">无独立配置,继承 <code>default.conf</code></p>
{% endif %}
{% if business_result %}
<h2>采集结果</h2>
<div class="cards">
<div class="card"><div class="label">结果状态</div><div class="value {{ 'green' if business_result.status == 'SUCCESS' else 'red' }}">{{ '成功' if business_result.status == 'SUCCESS' else '失败' }}</div></div>
<div class="card"><div class="label">成功 / 失败 / 目标</div><div class="value">{{ business_result.succeeded_count }} / {{ business_result.failed_count }} / {{ business_result.requested_count }}</div></div>
</div>
{% for failure in business_result.failures %}
<p style="color:#f44747"><strong>{{ failure.type }}</strong> · {{ failure.message }} {% if failure.url %}（{{ failure.url }}）{% endif %}</p>
{% endfor %}
{% for warning in business_result.warnings %}<p>提示：{{ warning }}</p>{% endfor %}
{% if business_result.source_exhausted %}<p>已无更多新文章：{{ business_result.exhaustion_reason }}</p>{% endif %}
<table>
<thead><tr><th>文章</th><th>状态</th><th>正文长度</th><th>分类 / 标签</th><th>失败原因</th></tr></thead>
<tbody>
{% for article in business_result.articles %}
{% set archive = article.archive if article.archive is defined else {} %}
{% set archive_props = archive.get('properties', {}) %}
<tr>
<td>{% if article.url.startswith('https://') or article.url.startswith('http://') %}<a class="source-link" href="{{ article.url }}" target="_blank" rel="noopener noreferrer">{{ article.title or article.url }}</a>{% else %}{{ article.title }}{% endif %}
<details><summary>预览归档原文与属性</summary>
{% if archive.get('loaded') %}<pre style="max-height:360px;overflow:auto;white-space:pre-wrap">{{ archive.get('body', '') }}</pre>
<details><summary>查看 Obsidian 属性</summary><pre>{{ archive_props | tojson(indent=2) }}</pre></details>
{% elif article.body is defined %}<p>{{ article.summary | default('') }}</p><pre style="max-height:360px;overflow:auto;white-space:pre-wrap">{{ article.body }}</pre>
{% else %}<p>未能读取归档原文{% if archive.get('error') %}：{{ archive.error }}{% endif %}</p>{% endif %}
<p>归档：{{ article.archive_path or '未归档' }}</p>
</details>
</td>
<td><span class="badge {{ 'ok' if article.status == 'SUCCESS' else 'fail' }}">{{ '成功' if article.status == 'SUCCESS' else '失败' }}</span></td>
<td>{% if archive.get('body_length') is not none %}{{ archive.body_length }} 字符{% elif article.body_length is defined %}{{ article.body_length }} 字符{% else %}—{% endif %}</td>
<td>{{ archive_props.get('category', article.category | default('')) }}<br>{{ archive_props.get('tags', article.tags | default([])) | join(' · ') }}</td>
<td>{{ article.failure_type or '—' }}{% if article.failure_reason %}<br>{{ article.failure_reason }}{% endif %}{% for evidence in article.quality_evidence %}{% if evidence != article.failure_reason %}<br>{{ evidence }}{% endif %}{% endfor %}</td>
</tr>
{% endfor %}
</tbody>
</table>
<details><summary>查看统一 JSON</summary><pre>{{ business_result | tojson(indent=2) }}</pre></details>
{% elif legacy_result is not none %}
<h2>历史采集结果</h2>
<p>此记录使用旧版文本格式，未经过新版正文质量核验。</p>
<pre>{{ legacy_result }}</pre>
{% endif %}
{% if log_info %}
<h2>最近运行</h2>
<div class="cards">
......
web/templates/edit.html
View file @ faf1233d
......@@ -6,11 +6,11 @@
<form method="POST" action="{{ url_for('edit_source', name=name) }}">
<div class="form-group">
<label for="time">运行时间 (HH:MM) *</label>
<label for="time">运行时间（可多个） *</label>
<input type="text" id="time" name="time" required
pattern="([01]\d|2[0-3]):[0-5]\d"
pattern="([01][0-9]|2[0-3]):[0-5][0-9](,\s*([01][0-9]|2[0-3]):[0-5][0-9])*"
value="{{ current_time }}">
<div class="hint">24 小时制,如 10:30。当前: {{ current_time }}</div>
<div class="hint">多个时间用逗号分隔，如 09:00,14:30,20:00。保存后替换全部运行时间。当前: {{ current_time }}</div>
</div>
<div class="form-group">
......@@ -42,7 +42,7 @@
<div class="form-group">
<label for="provider">Claude Provider</label>
<select id="provider" name="provider">
<option value="default" {% if not conf.get('PROVIDER') or conf.get('PROVIDER') == 'default' %}selected{% endif %}>默认 (系统全局配置)</option>
<option value="" {% if not conf.get('PROVIDER') or conf.get('PROVIDER') == 'default' %}selected{% endif %}>继承 default.conf（删除栏目覆盖）</option>
{% for p in providers %}
<option value="{{ p.name }}" {% if conf.get('PROVIDER') == p.name %}selected{% endif %}>{{ p.name }}</option>
{% endfor %}
......@@ -53,7 +53,7 @@
<div class="form-group">
<label for="fallback_provider">备用 Provider (FALLBACK)</label>
<select id="fallback_provider" name="fallback_provider">
<option value="default" {% if not conf.get('FALLBACK_PROVIDER') or conf.get('FALLBACK_PROVIDER') == 'default' %}selected{% endif %}>无 (不使用备用)</option>
<option value="" {% if not conf.get('FALLBACK_PROVIDER') or conf.get('FALLBACK_PROVIDER') == 'default' %}selected{% endif %}>继承 default.conf（删除栏目覆盖）</option>
{% for p in providers %}
<option value="{{ p.name }}" {% if conf.get('FALLBACK_PROVIDER') == p.name %}selected{% endif %}>{{ p.name }}</option>
{% endfor %}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment