二次开发指南:新增监控任务与定时任务¶
本文档说明如何在不改动项目核心逻辑的前提下,用最少改动接入新的监控任务或定时任务,并支持配置热重载与统一推送。
下文以项目内已实现的虎牙监控(监控任务)和 iKuuu 签到 / Demo 任务(定时任务)为例,按步骤对照真实代码说明。
零、开发环境与代码规范¶
代码检测¶
项目使用 black 和 ruff 进行代码格式化和检查。
安装开发依赖¶
uv sync --extra dev
代码格式化¶
使用 black 格式化代码:
# 格式化所有代码
uv run black .
# 检查代码格式(不修改文件)
uv run black --check .
代码检查¶
使用 ruff 检查代码:
# 检查代码并自动修复
uv run ruff check --fix .
# 仅检查代码(不修复)
uv run ruff check .
运行测试¶
uv run pytest
一、架构简述¶
| 类型 | 触发方式 | 配置来源示例 | 项目内示例 |
|---|---|---|---|
| 监控任务 | 固定间隔轮询 | huya.monitor_interval_seconds、weibo.monitor_interval_seconds |
虎牙监控、微博监控 |
| 定时任务 | Cron 每日定点 | checkin.time、tieba.time、plugins.xxx.time |
iKuuu 签到、贴吧签到、Demo 任务 |
新增任务时只需:
- 新增配置:在
config.yml中增加节点;若用顶层配置,还需在src/config.py中补充字段与解析。 - 实现任务逻辑:一个无参的 async 入口函数(内部
get_config(reload=True)、业务逻辑、可选推送)。 - 注册:在任务模块末尾调用
register_monitor或register_task,并在src/job_registry.py的MONITOR_MODULES/TASK_MODULES中追加模块路径。
主入口 main.py 通过 job_registry.discover_and_import() 加载所有列出的模块并注册到调度器,无需再改 main.py。
二、定时任务示例一:iKuuu 签到(顶层配置)¶
iKuuu 签到使用顶层配置(与贴吧签到一致):在 config.yml 中有独立节点 checkin,在 AppConfig 中有对应扁平字段,适合需要强类型、与现有风格统一的场景。
域名自动发现:iKuuu 的可用域名会自动从
ikuuu.club提取,无需在配置中手动填写 URL。系统在每次签到时会访问ikuuu.club,通过多种正则匹配和 HTTP 探测从其混淆 JS 中提取可用域名(如ikuuu.nl、ikuuu.fyi等),并随机选择一个使用。
2.1 配置:config.yml¶
在 config.yml 中增加与 tieba 同级的 checkin 节点(参见 config.yml.sample)。
单账号示例:
checkin:
enable: false
email: your@email.com
password: your_password
time: "08:00" # 每日执行时间 HH:MM
多账号示例(accounts 非空时优先于单账号 email/password):
checkin:
enable: true
time: "08:00"
accounts:
- email: user1@example.com
password: pass1
- email: user2@example.com
password: pass2
2.2 配置:src/config.py¶
在 AppConfig 中增加扁平字段(与 YAML 的 checkin 一一对应):
# 每日签到配置(域名自动从 ikuuu.club 发现,无需手动配置 URL)
checkin_enable: bool = False
checkin_email: str = ""
checkin_password: str = ""
checkin_time: str = "08:00"
在 load_config_from_yml() 中从 yml_config["checkin"] 读到上述字段并写入 config_dict(如 config_dict["checkin_enable"] = checkin["enable"] 等)。
项目内实现见 src/config.py 中 config_mappings 的 checkin 段及多账号 accounts 的特殊处理。
2.3 任务实现:tasks/ikuuu_checkin.py¶
① 配置校验与入口
- 使用 dataclass 从
AppConfig转成任务用配置,并做validate()(未启用或缺少必填项则直接 return): - 域名通过
_extract_ikuuu_domain()自动从ikuuu.club提取,URL 由域名自动构建(@property):
@dataclass
class CheckinConfig:
enable: bool
domain: str # 自动发现的域名,如 ikuuu.nl
email: str
password: str
time: str
@property
def login_url(self) -> str:
return f"https://{self.domain}/auth/login"
@property
def checkin_url(self) -> str:
return f"https://{self.domain}/user/checkin"
@property
def user_page_url(self) -> str:
return f"https://{self.domain}/user"
@classmethod
def from_app_config(cls, config: AppConfig, domain: str) -> CheckinConfig:
return cls(
enable=config.checkin_enable,
domain=domain,
# ...
time=config.checkin_time.strip() or "08:00",
)
async def run_checkin_once() -> None:
app_config = get_config(reload=True)
if not app_config.checkin_enable:
return
# 自动发现 ikuuu 可用域名
domain = await _extract_ikuuu_domain()
if not domain:
logger.error("ikuuu签到:无法自动发现可用域名,跳过本次执行")
return
cfg = CheckinConfig.from_app_config(app_config, domain=domain)
if not cfg.validate():
return
# 业务逻辑:登录 → 签到 → 获取流量信息
async with aiohttp.ClientSession(...) as session:
push_manager = await build_push_manager(
app_config.push_channel_list, session, logger, init_fail_prefix="ikuuu签到:",
channel_names=cfg.push_channels if cfg.push_channels else None, # 指定使用的通道
)
cookie = await _login_and_get_cookie(session, cfg)
if not cookie:
await _send_checkin_push(push_manager, title="ikuuu签到失败:登录失败", ...)
return
ok = await _checkin(session, cfg, cookie)
traffic_info = await _get_user_traffic(session, cfg, cookie)
await _send_checkin_push(push_manager, title=..., msg=..., success=ok, traffic_info=traffic_info)
if push_manager:
await push_manager.close()
② 推送逻辑
- 推送前用
is_in_quiet_hours(app_cfg)判断免打扰,在免打扰时段内只打日志不推送:
async def _send_checkin_push(push_manager, title, msg, success, cfg, traffic_info=None):
if push_manager is None:
return
app_cfg = get_config()
if is_in_quiet_hours(app_cfg):
logger.debug("ikuuu签到:免打扰时段,不发送推送")
return
await push_manager.send_news(
title=f"{title}({masked_email})",
description=...,
to_url=cfg.user_page_url,
picurl="...",
btntxt="查看账户",
)
③ 注册:Cron 触发参数 + register_task
- 执行时间由
checkin.time决定,使用公共方法parse_checkin_time得到 cron 的hour、minute,并在模块末尾注册:
from src.config import AppConfig, get_config, is_in_quiet_hours, parse_checkin_time
from src.job_registry import register_task
def _get_checkin_trigger_kwargs(config: AppConfig) -> dict:
hour, minute = parse_checkin_time(config.checkin_time)
return {"minute": minute, "hour": hour}
register_task("ikuuu_checkin", run_checkin_once, _get_checkin_trigger_kwargs)
④ 当天已运行则跳过(默认行为)
register_task 默认启用 skip_if_run_today=True,任务在执行前会检查当天是否已经运行过:
- 如果已运行:输出日志 {job_id}: 当天已经运行过了,跳过该任务,然后跳过执行
- 如果未运行:正常执行任务,成功后记录运行日期
- 如果任务执行失败:不记录运行日期,允许后续重试
若某个任务需要每次触发都执行(不检查当天是否已运行),可在注册时禁用:
register_task("always_run_task", run_task, _get_trigger_kwargs, skip_if_run_today=False)
⑤ 手动触发执行
通过 Web 管理界面的「任务管理」页面手动触发任务时,会使用 JobDescriptor.original_run_func(原始执行函数),绕过"当天已运行则跳过"检查,确保任务被强制执行。这对于调试或需要立即重新执行的场景非常有用。
2.4 注册表:src/job_registry.py¶
在 TASK_MODULES 中已包含该模块,主程序启动时会导入并执行上述 register_task:
TASK_MODULES: list[str] = [
"tasks.log_cleanup",
"tasks.ikuuu_checkin", # iKuuu 签到
"tasks.tieba_checkin",
"tasks.weibo_chaohua_checkin", # 微博超话签到
# ... 雨云、贴吧、阿里云盘等更多任务见 job_registry.py
"tasks.demo_task", # 二次开发示例,不需要可移除此行
]
小结:顶层定时任务 = config.yml 节点 → AppConfig + load_config_from_yml → 任务模块(run_xxx_once + 推送 + _get_xxx_trigger_kwargs)→ register_task → TASK_MODULES 一行。
三、定时任务示例二:Demo 任务(plugins 配置)¶
Demo 任务使用 plugins 配置:无需改 AppConfig 和 load_config_from_yml(),只需在 config.yml 的 plugins 下增加一个 key,适合快速扩展、字段灵活的场景。
3.1 配置:config.yml¶
plugins:
demo_task:
enable: false
time: "08:30"
message: "Demo 定时任务执行完成"
3.2 任务实现:tasks/demo_task.py¶
- 从
config.plugins.get("demo_task", {})读配置;未启用则直接 return。 - 使用
parse_checkin_time(plug.get("time", "08:00"))得到 cron 的 hour/minute。 - 推送前用
is_in_quiet_hours(config)判断免打扰。
核心片段:
PLUGIN_KEY = "demo_task"
def _get_plugin_config(config: AppConfig) -> dict:
return config.plugins.get(PLUGIN_KEY) or {}
async def run_demo_task_once() -> None:
config = get_config(reload=True)
plug = _get_plugin_config(config)
if not plug.get("enable", False):
return
async with aiohttp.ClientSession(...) as session:
push_manager = await build_push_manager(...)
message = plug.get("message", "Demo 定时任务执行完成。")
if push_manager and not is_in_quiet_hours(config):
await push_manager.send_news(title="Demo 任务执行完成", description=message, ...)
if push_manager:
await push_manager.close()
def _get_demo_task_trigger_kwargs(config: AppConfig) -> dict:
plug = _get_plugin_config(config)
hour, minute = parse_checkin_time((plug.get("time") or "08:00").strip())
return {"minute": minute, "hour": hour}
register_task("demo_task", run_demo_task_once, _get_demo_task_trigger_kwargs)
在 job_registry.TASK_MODULES 中需包含 "tasks.demo_task"(当前已包含)。
完整代码见 tasks/demo_task.py。
四、监控任务示例:虎牙直播监控¶
虎牙监控按固定间隔轮询房间状态,使用顶层配置 + 继承 BaseMonitor,是典型的监控任务写法。
4.1 配置:config.yml¶
huya:
enable: true # 是否启用该监控,默认 true;设为 false 时任务暂停
rooms: 991108,333003,518518 # 逗号分隔的房间号
concurrency: 5
monitor_interval_seconds: 65 # 轮询间隔(秒)
4.2 配置:src/config.py¶
- AppConfig 中增加扁平字段:
huya_enable、huya_rooms、huya_concurrency、huya_monitor_interval_seconds。 - load_config_from_yml:从
yml_config["huya"]读到上述字段写入config_dict。 - 提供 get_huya_config() 返回结构化配置(列表 + 并发数),供监控类使用:
class HuyaConfig(BaseModel):
rooms: list[str]
concurrency: int = 7
def get_huya_config(self) -> HuyaConfig:
rooms = [r.strip() for r in self.huya_rooms.split(",") if r.strip()]
return HuyaConfig(rooms=rooms, concurrency=self.huya_concurrency)
见 src/config.py 中 HuyaConfig、AppConfig.get_huya_config 及 load_config_from_yml 的 huya 段落。
4.3 监控实现:monitors/huya_monitor.py¶
① 继承 BaseMonitor
BaseMonitor负责:config、session、db、push、initialize()(数据库 + 推送)、close()。子类只需实现run()和monitor_name,以及可选的_get_session重写(如固定 User-Agent/Cookie)。
from src.monitor import BaseMonitor
class HuyaMonitor(BaseMonitor):
def __init__(self, config: AppConfig, session=None):
super().__init__(config, session)
self.huya_config = config.get_huya_config()
self.old_data_dict = {}
self._is_first_time = False
async def initialize(self):
await super().initialize()
await self.load_old_info() # 从 DB 加载旧状态
async def run(self):
new_config = get_config(reload=False)
self.config = new_config
self.huya_config = new_config.get_huya_config()
# 并发轮询房间,比对 old_data_dict,有变化则更新 DB 并 push_notification
semaphore = asyncio.Semaphore(self.huya_config.concurrency)
tasks = [process_with_semaphore(rid) for rid in self.huya_config.rooms]
await asyncio.gather(*tasks, return_exceptions=True)
@property
def monitor_name(self) -> str:
return "虎牙直播监控🐯 🐯 🐯"
② 推送
- 在业务逻辑里调用
self.push.send_news(...);推送前用is_in_quiet_hours(self.config)判断免打扰,若在免打扰时段则只打日志不推送。见huya_monitor.py中push_notification。
③ 对外入口与注册
- 对外暴露一个无参的 async 函数,内部
get_config(reload=True)后async with HuyaMonitor(config) as monitor: await monitor.run()。 - 提供
_get_huya_trigger_kwargs(config)返回{"seconds": config.huya_monitor_interval_seconds},并在模块末尾register_monitor:
async def run_huya_monitor() -> None:
config = get_config(reload=True)
async with HuyaMonitor(config) as monitor:
await monitor.run()
def _get_huya_trigger_kwargs(config: AppConfig) -> dict:
return {"seconds": config.huya_monitor_interval_seconds}
from src.job_registry import register_monitor
register_monitor("huya_monitor", run_huya_monitor, _get_huya_trigger_kwargs)
见 monitors/huya_monitor.py 末尾。
4.4 注册表:src/job_registry.py¶
在 MONITOR_MODULES 中已包含虎牙模块:
MONITOR_MODULES: list[str] = [
"monitors.huya_monitor",
"monitors.weibo_monitor",
]
4.5 配置热重载(可选)¶
项目已对现有监控与定时任务的 enable、间隔/执行时间(monitor_interval_seconds、time)、免打扰、推送通道等字段做完整覆盖,修改后约 5 秒内热重载生效。
新增监控或定时任务时,需在 src/config_watcher.py 的 _config_changed() 里增加对新字段的比较(如 my_monitor_rooms、my_monitor_enable、my_task_time 等),否则相关配置变更不会触发调度器更新。
小结:监控任务 = config.yml(业务节点含 enable + scheduler 间隔)→ AppConfig + get_xxx_config + load_config_from_yml → 继承 BaseMonitor 实现 run + 推送 → run_xxx_monitor + _get_xxx_trigger_kwargs → register_monitor → MONITOR_MODULES 一行。enable: false 时任务会被暂停,热重载生效。
五、监控任务需要数据库时该怎么办¶
很多监控任务需要持久化上一次状态(例如上次是否在播、上次微博内容),以便本次轮询时对比、仅在变化时推送。本项目的做法是:继承 BaseMonitor 即自带数据库与推送,数据库使用项目内统一的 SQLite(data/data.db),由 AsyncDatabase 封装。
5.1 继承 BaseMonitor 即获得 self.db¶
在 src/monitor.py 中,BaseMonitor.initialize() 会:
- 创建
self.db = AsyncDatabase()并await self.db.initialize(); - 创建
self.push(统一推送); - 可选
self.session(HTTP)。
因此你的监控类只需继承 BaseMonitor,在 initialize() 里可先 await super().initialize(),再加载本监控需要的“旧数据”;在 run() 里用 self.db 做查询/更新/插入即可。无需自己 new AsyncDatabase 或管理连接。
5.2 AsyncDatabase 常用 API(src/database.py)¶
| 方法 | 说明 | 返回值 |
|---|---|---|
execute_query(sql, params=None) |
查询,占位符用 %(key)s,params 为 dict |
list[tuple],每行一个元组 |
execute_update(sql, params=None) |
执行 UPDATE/INSERT/DELETE | bool(是否成功) |
execute_insert(sql, params=None) |
同 execute_update,语义上用于插入 | bool |
is_table_empty(table_name) |
判断表是否为空(可用于“首次运行”逻辑) | bool |
- SQL 占位符:写
%(name)s、%(room)s等,params 传字典如{"name": "xx", "room": "123"}。模块内部会转换为 SQLite 的:name格式,无需改 SQL。 - 连接:默认使用全局共享连接(单进程内复用),重试与重连已在
execute_*内处理。
5.3 新监控需要新表时:在 database.py 中加表¶
当前所有表结构都在 src/database.py 的 _init_tables() 里统一创建(CREATE TABLE IF NOT EXISTS ...)。
若你的监控需要自己的表(例如 my_monitor),在 src/database.py 的 _init_tables() 中增加一段即可,与现有 weibo、huya 表并列,例如:
# 在 _init_tables(self, conn) 末尾、await conn.commit() 前增加:
# 创建 my_monitor 表(示例)
await conn.execute(
"""
CREATE TABLE IF NOT EXISTS my_monitor (
id TEXT PRIMARY KEY,
name TEXT,
status TEXT,
updated_at TEXT
)
"""
)
await conn.commit()
表名、字段名按你的业务设计即可;主键建议能唯一标识一条监控对象(如房间号、用户 ID)。
5.4 虎牙监控中的用法示例(对照代码)¶
- 加载旧数据(在
initialize()里调用,或run()开头):
用execute_query把上一轮存的状态读进内存,供本轮对比:
async def load_old_info(self):
sql = "SELECT room, name, is_live FROM huya"
results = await self.db.execute_query(sql)
self.old_data_dict = {row[0]: row for row in results}
self._is_first_time = len(self.old_data_dict) == 0
- 有变化时更新:
用execute_update,占位符与字典一一对应:
sql = "UPDATE huya SET name=%(name)s, is_live=%(is_live)s WHERE room=%(room)s"
await self.db.execute_update(sql, data)
- 新对象首次写入:
用execute_insert:
sql = "INSERT INTO huya (room, name, is_live) VALUES (%(room)s, %(name)s, %(is_live)s)"
await self.db.execute_insert(sql, data)
- 首次建表/首次运行:
虎牙用_is_first_time标记“表里之前没有数据”。首次跑满一轮时只写入 DB、不推送,避免历史数据被当成“新变化”刷屏;从第二轮开始才按变化推送。你可按同样思路处理。
完整实现见 monitors/huya_monitor.py(load_old_info、process_room 中的 SQL 与 self.db 调用)。
5.5 小结:监控 + 数据库的步骤¶
- 继承 BaseMonitor,在
initialize()里await super().initialize()后加载旧数据到内存(如load_old_info)。 - 在
src/database.py的_init_tables()里为你的监控增加 CREATE TABLE IF NOT EXISTS(若需要新表)。 - 在
run()里:拉取当前数据 → 与旧数据对比 → 有变化则execute_update/execute_insert更新 DB,并调用self.push.send_news(...);无变化则只打日志。 - SQL 使用
%(key)s+ dict 参数,通过self.db.execute_query/execute_update/execute_insert访问;连接与重试由 AsyncDatabase 统一处理。
六、推送逻辑(统一说明)¶
- 推送通道统一来自
config.push_channel_list(即config.yml的push_channel),无需在任务里新增通道类型。 - 通道选择机制:每个任务可以在配置中通过
push_channels字段指定使用哪些推送通道(按名称匹配)。为空时使用全部已配置的通道。 - 在任务/监控内:
- 使用
await build_push_manager(config.push_channel_list, session, logger, init_fail_prefix="任务名:", channel_names=["通道1", "通道2"])得到UnifiedPushManager。channel_names参数可选,用于指定仅初始化哪些通道(按name字段匹配),为空或 None 时使用全部通道。 - 需要推送时调用
await push_manager.send_news(title=..., description=..., to_url=..., picurl=..., btntxt=...)。 - 遵守免打扰:推送前
if is_in_quiet_hours(config): return(或只打日志),再调用send_news。 - 使用完毕后
await push_manager.close()。
虎牙在类内使用 self.push(BaseMonitor 在 initialize 里已创建,会自动读取任务配置的 push_channels);iKuuu/Demo 在 async 函数内自己创建 push_manager 并在同一 session 生命周期内 close。
推送失败建议用 logger.error(..., exc_info=True) 记录,不中断主流程。
任务专属日志:新增任务无需额外处理,系统会在执行时自动将输出写入 task_{job_id}_YYYYMMDD.log。Handler 挂载在 root logger,可捕获任务内所有 logger(模块、类、推送通道等)的输出。
七、示例文件与代码位置一览¶
| 类型 | 示例 | 配置文件 | 配置解析 | 任务/监控实现 | 注册 |
|---|---|---|---|---|---|
| 定时任务 | iKuuu 签到 | config.yml → checkin |
AppConfig + load_config_from_yml(checkin 段) |
tasks/ikuuu_checkin.py(run_checkin_once、_send_checkin_push、_get_checkin_trigger_kwargs) |
register_task("ikuuu_checkin", ...),TASK_MODULES 含 tasks.ikuuu_checkin |
| 定时任务 | Demo 任务 | config.yml → plugins.demo_task |
无需改 config.py,用 config.plugins.get("demo_task") |
tasks/demo_task.py |
register_task("demo_task", ...),TASK_MODULES 含 tasks.demo_task |
| 定时任务 | Freenom 续期 | config.yml → freenom |
AppConfig + load_config_from_yml(freenom 段,多账号 accounts) |
tasks/freenom_checkin.py(run_freenom_checkin_once、_get_freenom_trigger_kwargs) |
register_task("freenom_checkin", ...),TASK_MODULES 含 tasks.freenom_checkin |
| 定时任务 | 天气推送 | config.yml → weather |
AppConfig + load_config_from_yml(weather 段) |
tasks/weather_push.py(run_weather_push_once、_get_weather_trigger_kwargs) |
register_task("weather_push", ...),TASK_MODULES 含 tasks.weather_push |
| 监控任务 | 虎牙监控 | config.yml → huya(含 monitor_interval_seconds) |
AppConfig、HuyaConfig、get_huya_config、load_config_from_yml(huya 段) |
monitors/huya_monitor.py(HuyaMonitor、run_huya_monitor、_get_huya_trigger_kwargs) |
register_monitor("huya_monitor", ...),MONITOR_MODULES 含 monitors.huya_monitor |
- parse_checkin_time:
src/config.py,将"HH:MM"解析为(hour, minute)字符串元组,供 Cron 使用。 - BaseMonitor:
src/monitor.py,提供config、db、push、initialize、close,子类实现run、monitor_name。
八、检查清单:新增定时任务¶
- [ ] 在
config.yml中增加配置(顶层节点或plugins.xxx)。 - [ ] 若用顶层配置:在
AppConfig与load_config_from_yml()中补充字段;若用plugins,无需改 config.py。 - [ ] 新建
tasks/xxx.py,实现run_xxx_once()(内部get_config(reload=True)、校验、业务、推送)、_get_xxx_trigger_kwargs(config)(返回{"minute": m, "hour": h},可用parse_checkin_time)。 - [ ] 在模块末尾调用
register_task("job_id", run_xxx_once, _get_xxx_trigger_kwargs)。 - 默认启用
skip_if_run_today=True,当天已运行则跳过 - 若需每次触发都执行,设置
skip_if_run_today=False - [ ] 在
src/job_registry.TASK_MODULES中追加"tasks.xxx"。 - [ ] 若使用新的顶层配置项,需在
config_watcher._config_changed中增加对应比较(plugins 已支持)。
九、检查清单:新增监控任务¶
- [ ] 在
config.yml中增加业务节点(如my_monitor),并在该节点下增加monitor_interval_seconds字段(例如my_monitor.monitor_interval_seconds)。 - [ ] 在
AppConfig中增加扁平字段,在load_config_from_yml()中解析;可选:提供get_my_monitor_config()返回结构化配置。 - [ ] 在
config_watcher._config_changed中增加对新字段的比较(以便热重载)。 - [ ] 新建
monitors/xxx.py,继承BaseMonitor实现run()、monitor_name,以及run_xxx_monitor()、_get_xxx_trigger_kwargs(config)(返回{"seconds": config.xxx_interval_seconds})。 - [ ] 若监控需要数据库:在
src/database.py的_init_tables()中增加CREATE TABLE IF NOT EXISTS your_table (...);在监控类initialize()里加载旧数据,在run()里用self.db.execute_query/execute_update/execute_insert读写(参见 五、监控任务需要数据库时该怎么办)。 - [ ] 在模块末尾调用
register_monitor("job_id", run_xxx_monitor, _get_xxx_trigger_kwargs)。 - [ ] 在
src/job_registry.MONITOR_MODULES中追加"monitors.xxx"。
完成以上步骤后,新任务会被主程序自动加载、按配置调度,并在配置变更时通过 ConfigWatcher 热重载。
十、Web 前端对新增配置的响应¶
- 文本视图:直接读写整份
config.yml,新增的任意 key(如plugins、freenom、weather等)都会完整显示、可编辑,保存后整份写回,会正确响应。 - 表格视图:当前展示微博、虎牙、各类签到任务(iKuuu、贴吧、雨云、恩山、阿里云盘、什么值得买、Freenom、夸克、科技玩家、帆软、999、zgfc、双色球等)、调度器、免打扰、推送通道以及插件配置等固定区块。
- 在表格视图中修改并保存时,后端会合并写回,因此文件中已有的
plugins或其他顶层节点不会丢失。 - 插件配置可在配置页底部的「插件/扩展配置」中以 JSON 形式编辑
config.plugins;尚未在表格中单独列出的顶层 key 需使用文本视图编辑。
若你新增了与现有区块同级的配置(例如新的顶层节点),并希望在表格中编辑,需在 Web 前端增加对应卡片及 loadSectionConfig / collectSectionConfig / collectConfig 的处理(可参考本文档中已集成的 freenom/weather/kuake/kjwj 等实现方式)。
十一、青龙面板单任务脚本(ql/*.py)¶
青龙环境下,主程序不运行,而是由青龙按 Cron 调用 ql/*.py 单任务脚本。这些脚本:
- 通过
ql/_runner.py作为统一入口,根据脚本名或命令行参数调用对应任务逻辑 - 配置来自环境变量(
WEBMONITER_*前缀),由src/ql_compat.py的load_config_from_env()解析 - 推送通过 qlapi 通道,调用青龙内置的
QLAPI.systemNotify - 与
tasks/*、monitors/*主流程解耦,共用同一套业务逻辑(如签到、监控 API 调用)
新增青龙脚本:复制 ql/ikuuu_checkin.py 等示例,按需修改任务名、环境变量名,并在 ql/_runner.py 中注册。详见 青龙面板兼容指南。