238 lines
7.9 KiB
Python
238 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""通用「数据采集 + 模板注入」引擎。
|
|
|
|
读一份配置（YAML/JSON），按声明采集多路数据源，注入 Jinja2 模板，产出 markdown（或任意文本）。
|
|
|
|
用法：
|
|
python3 build.py --config config/digest.yaml [--env K=V ...] [--out PATH] [--dry-run]
|
|
|
|
配置结构（示例见 config/morning-digest.yaml）：
|
|
meta: { name, date_format }
|
|
sources: { <section>: { type, ...adapter 参数, map } }
|
|
render: { template, output, [env] }
|
|
(可选) sections: 控制模板中板块渲染顺序/说明
|
|
|
|
Source 类型：
|
|
http_json GET 一个 JSON 接口，按 items_path 取列表。支持 {param} 循环、| 字段回退、
|
|
@index 序号、{loop.key} 循环值。
|
|
command 跑一条 shell 命令；命令里 {out} 替换为输出 JSON 文件路径（写文件）或直接 stdout JSON。
|
|
static_json 读一个本地 JSON 文件（供 LLM/外部注入）。
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from datetime import datetime
|
|
|
|
import requests
|
|
|
|
try:
|
|
import yaml
|
|
except ImportError: # pragma: no cover
|
|
yaml = None
|
|
|
|
from jinja2 import Environment, FileSystemLoader
|
|
|
|
ENV_RE = re.compile(r"\$\{(\w+)\}")
|
|
|
|
|
|
def env_sub(value, env):
|
|
"""把字符串里的 ${VAR} 替换成 env 值，未定义则保留原样。"""
|
|
if not isinstance(value, str):
|
|
return value
|
|
def _repl(m):
|
|
return env.get(m.group(1), m.group(0))
|
|
return ENV_RE.sub(_repl, value)
|
|
|
|
|
|
def deep_env_sub(obj, env):
|
|
if isinstance(obj, dict):
|
|
return {k: deep_env_sub(v, env) for k, v in obj.items()}
|
|
if isinstance(obj, list):
|
|
return [deep_env_sub(v, env) for v in obj]
|
|
return env_sub(obj, env)
|
|
|
|
|
|
def load_config(path):
|
|
with open(path, "rb") as f:
|
|
raw = f.read()
|
|
if path.endswith((".yaml", ".yml")):
|
|
if yaml is None:
|
|
raise RuntimeError("需要 pyyaml 才能读 YAML 配置")
|
|
return yaml.safe_load(raw) or {}
|
|
return json.loads(raw)
|
|
|
|
|
|
def get_path(obj, dotted):
|
|
"""按点分路径取值，支持 dict key 和 list index。"""
|
|
if not dotted:
|
|
return obj
|
|
for part in str(dotted).split("."):
|
|
if isinstance(obj, dict):
|
|
obj = obj.get(part)
|
|
elif isinstance(obj, list) and part.isdigit():
|
|
obj = obj[int(part)]
|
|
else:
|
|
return None
|
|
return obj
|
|
|
|
|
|
def resolve_spec(spec, item, ctx):
|
|
"""解析 map 取值表达式：a|b|c 依次回退；{loop.key} 取循环值；@index 由调用方特殊处理。"""
|
|
for token in str(spec).split("|"):
|
|
token = token.strip()
|
|
if not token:
|
|
continue
|
|
if token.startswith("{loop.") and token.endswith("}"):
|
|
v = ctx.get(token[6:-1].strip())
|
|
if v not in (None, ""):
|
|
return v
|
|
continue
|
|
v = get_path(item, token)
|
|
if v not in (None, ""):
|
|
return v
|
|
return ""
|
|
|
|
|
|
def map_item(item, mapping, ctx, index):
|
|
out = {}
|
|
for field, spec in (mapping or {}).items():
|
|
if str(spec) == "@index":
|
|
out[field] = index
|
|
else:
|
|
out[field] = resolve_spec(spec, item, ctx)
|
|
return out
|
|
|
|
|
|
# ----------------------------- source adapters ----------------------------- #
|
|
|
|
def collect_http_json(src, env):
|
|
items = []
|
|
loop = src.get("param_loop")
|
|
batches = [{}] if not loop else [{loop["key"]: v} for v in loop.get("values", [])]
|
|
for ctx in batches:
|
|
url = env_sub(src["url"], env)
|
|
for k, v in ctx.items():
|
|
url = url.replace("{%s}" % k, str(v))
|
|
resp = requests.get(url, timeout=src.get("timeout", 20)).json()
|
|
raw = get_path(resp, src.get("items_path", "")) or []
|
|
raw = raw[: src.get("limit", 20)]
|
|
for i, it in enumerate(raw, 1):
|
|
items.append(map_item(it, src.get("map"), ctx, i))
|
|
return items
|
|
|
|
|
|
def collect_command(src, env):
|
|
cmd = env_sub(src["cmd"], env)
|
|
out_file = None
|
|
if "{out}" in cmd:
|
|
out_file = tempfile.mktemp(suffix=".json")
|
|
cmd = cmd.replace("{out}", out_file)
|
|
r = subprocess.run(["sh", "-lc", cmd], capture_output=True, text=True,
|
|
timeout=src.get("timeout", 180))
|
|
if r.returncode != 0:
|
|
raise RuntimeError((r.stderr or r.stdout).strip()[:800])
|
|
file_ref = src.get("items_from_file")
|
|
if file_ref:
|
|
data_path = env_sub(file_ref, env).replace("{out}", out_file or "")
|
|
with open(data_path, encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
else:
|
|
data = json.loads(r.stdout)
|
|
raw = get_path(data, src.get("items_path", "")) or data
|
|
raw = raw[: src.get("limit", 20)]
|
|
return [map_item(it, src.get("map"), {}, i + 1) for i, it in enumerate(raw)]
|
|
|
|
|
|
def collect_static(src, env):
|
|
path = env_sub(src.get("file", ""), env)
|
|
if not path or not os.path.exists(path):
|
|
if src.get("optional"):
|
|
return []
|
|
raise RuntimeError(f"static 文件不存在: {path}")
|
|
with open(path, encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
raw = get_path(data, src.get("items_path", "")) or data
|
|
raw = raw[: src.get("limit", 20)]
|
|
return [map_item(it, src.get("map"), {}, i + 1) for i, it in enumerate(raw)]
|
|
|
|
|
|
ADAPTERS = {"http_json": collect_http_json, "command": collect_command,
|
|
"static_json": collect_static}
|
|
|
|
|
|
def render_template(template_file, data, out_path):
|
|
# 模板路径解析：绝对路径 > 配置目录 > 配置父目录 > 当前目录
|
|
candidates = []
|
|
if not os.path.isabs(template_file):
|
|
base = os.path.dirname(os.path.abspath(config_path if globals().get("config_path") else "."))
|
|
candidates = [os.path.join(base, template_file),
|
|
os.path.join(os.path.dirname(base), template_file)]
|
|
candidates.append(template_file)
|
|
tpl_path = next((p for p in candidates if os.path.exists(p)), candidates[0])
|
|
|
|
env = Environment(loader=FileSystemLoader(os.path.dirname(tpl_path)),
|
|
trim_blocks=True, lstrip_blocks=True)
|
|
tpl = env.get_template(os.path.basename(tpl_path))
|
|
md = tpl.render(**data)
|
|
with open(out_path, "w", encoding="utf-8") as f:
|
|
f.write(md)
|
|
return out_path
|
|
|
|
|
|
def main():
|
|
global config_path
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--config", required=True, help="配置文件路径 (YAML/JSON)")
|
|
ap.add_argument("--env", action="append", default=[], help="覆盖环境变量 K=V（可多次）")
|
|
ap.add_argument("--out", default=None, help="覆盖输出路径")
|
|
ap.add_argument("--dry-run", action="store_true", help="只采集不写文件")
|
|
args = ap.parse_args()
|
|
|
|
config_path = os.path.abspath(args.config)
|
|
env = dict(os.environ)
|
|
for kv in args.env:
|
|
k, _, v = kv.partition("=")
|
|
env[k] = v
|
|
|
|
cfg = deep_env_sub(load_config(config_path), env)
|
|
date = datetime.now()
|
|
|
|
data = {"date": cfg.get("meta", {}).get("date_format", "%Y-%m-%d") and
|
|
date.strftime(cfg.get("meta", {}).get("date_format", "%Y-%m-%d"))}
|
|
|
|
errors = {}
|
|
for name, src in (cfg.get("sources") or {}).items():
|
|
adapter = ADAPTERS.get(src.get("type"))
|
|
if not adapter:
|
|
errors[name] = f"未知 source 类型: {src.get('type')}"
|
|
data[name] = []
|
|
continue
|
|
try:
|
|
data[name] = adapter(src, env)
|
|
except Exception as e:
|
|
errors[name] = str(e)[:300]
|
|
data[name] = []
|
|
|
|
if errors:
|
|
data["_errors"] = errors
|
|
|
|
if args.dry_run:
|
|
print(json.dumps({k: v for k, v in data.items() if k != "_errors"},
|
|
ensure_ascii=False, indent=2))
|
|
return
|
|
|
|
out_path = args.out or env_sub(cfg.get("render", {}).get("output", "/tmp/digest.md"), env)
|
|
render_template(cfg.get("render", {}).get("template", "templates/digest.md"), data, out_path)
|
|
print(f"RENDER_OK: {out_path}")
|
|
if errors:
|
|
print(f"_errors={json.dumps(errors, ensure_ascii=False)}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|