mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-09-07 00:16:57 +08:00
refactor: migrate workflow actions to runtime config snapshot
This commit is contained in:
@@ -1,6 +1,7 @@
|
|||||||
from app.workflow.actions import BaseAction
|
from app.workflow.actions import BaseAction
|
||||||
from app.chain.subscribe import SubscribeChain
|
from app.chain.subscribe import SubscribeChain
|
||||||
from app.runtime.config import settings, global_vars
|
from app.application.configuration import get_chain_runtime_config_snapshot
|
||||||
|
from app.runtime.config import global_vars
|
||||||
from app.domain.context import MediaInfo
|
from app.domain.context import MediaInfo
|
||||||
from app.application.chain.data import SubscribePortProxy as SubscribeOper
|
from app.application.chain.data import SubscribePortProxy as SubscribeOper
|
||||||
from app.runtime.log import logger
|
from app.runtime.log import logger
|
||||||
@@ -65,7 +66,7 @@ class AddSubscribeAction(BaseAction):
|
|||||||
season=mediainfo.season,
|
season=mediainfo.season,
|
||||||
media_source=mediainfo.media_source,
|
media_source=mediainfo.media_source,
|
||||||
media_id=mediainfo.media_id,
|
media_id=mediainfo.media_id,
|
||||||
username=settings.SUPERUSER)
|
username=get_chain_runtime_config_snapshot().superuser)
|
||||||
if sid:
|
if sid:
|
||||||
self._added_subscribes.append(sid)
|
self._added_subscribes.append(sid)
|
||||||
# 保存缓存
|
# 保存缓存
|
||||||
|
|||||||
@@ -4,7 +4,8 @@ from pydantic import Field
|
|||||||
|
|
||||||
from app.workflow.actions import BaseAction
|
from app.workflow.actions import BaseAction
|
||||||
from app.chain.media import MediaChain
|
from app.chain.media import MediaChain
|
||||||
from app.runtime.config import settings, global_vars
|
from app.application.configuration import get_chain_runtime_config_snapshot
|
||||||
|
from app.runtime.config import global_vars
|
||||||
from app.domain.context import Context, TorrentInfo
|
from app.domain.context import Context, TorrentInfo
|
||||||
from app.domain.metainfo import MetaInfo
|
from app.domain.metainfo import MetaInfo
|
||||||
from app.application.rss import RssHelper
|
from app.application.rss import RssHelper
|
||||||
@@ -65,7 +66,11 @@ class FetchRssAction(BaseAction):
|
|||||||
headers["User-Agent"] = params.ua
|
headers["User-Agent"] = params.ua
|
||||||
|
|
||||||
rss_items = RssHelper().parse(url=params.url,
|
rss_items = RssHelper().parse(url=params.url,
|
||||||
proxy=settings.PROXY if params.proxy else None,
|
proxy=(
|
||||||
|
get_chain_runtime_config_snapshot().proxy
|
||||||
|
if params.proxy
|
||||||
|
else None
|
||||||
|
),
|
||||||
timeout=params.timeout,
|
timeout=params.timeout,
|
||||||
headers=headers)
|
headers=headers)
|
||||||
if rss_items is None or rss_items is False:
|
if rss_items is None or rss_items is False:
|
||||||
|
|||||||
@@ -5,7 +5,8 @@ from pydantic import Field
|
|||||||
|
|
||||||
from app.workflow.actions import BaseAction
|
from app.workflow.actions import BaseAction
|
||||||
from app.chain.storage import StorageChain
|
from app.chain.storage import StorageChain
|
||||||
from app.runtime.config import global_vars, settings
|
from app.application.configuration import get_chain_runtime_config_snapshot
|
||||||
|
from app.runtime.config import global_vars
|
||||||
from app.runtime.log import logger
|
from app.runtime.log import logger
|
||||||
from app.schemas.workflow import ActionParams
|
from app.schemas.workflow import ActionParams
|
||||||
from app.schemas.workflow import ActionContext
|
from app.schemas.workflow import ActionContext
|
||||||
@@ -56,10 +57,15 @@ class ScanFileAction(BaseAction):
|
|||||||
self._has_error = True
|
self._has_error = True
|
||||||
return context
|
return context
|
||||||
files = storagechain.list_files(fileitem, recursion=True)
|
files = storagechain.list_files(fileitem, recursion=True)
|
||||||
|
runtime_config = get_chain_runtime_config_snapshot()
|
||||||
|
media_exts = (
|
||||||
|
runtime_config.media_extensions
|
||||||
|
+ runtime_config.subtitle_extensions
|
||||||
|
+ runtime_config.audio_extensions
|
||||||
|
)
|
||||||
for file in files:
|
for file in files:
|
||||||
if global_vars.is_workflow_stopped(workflow_id):
|
if global_vars.is_workflow_stopped(workflow_id):
|
||||||
break
|
break
|
||||||
media_exts = settings.RMT_MEDIAEXT + settings.RMT_SUBEXT + settings.RMT_AUDIOEXT
|
|
||||||
if not file.extension or f".{file.extension.lower()}" not in media_exts:
|
if not file.extension or f".{file.extension.lower()}" not in media_exts:
|
||||||
continue
|
continue
|
||||||
# 添加文件到队列,而不是目录
|
# 添加文件到队列,而不是目录
|
||||||
|
|||||||
@@ -903,6 +903,11 @@ OTel 初始化只能位于 Startup/Adapter;Domain/Application 只依赖 no-op-
|
|||||||
- Outbox dispatcher 的有限重试和 dead-letter 已分别接入 `scheduler.job.retry` 与
|
- Outbox dispatcher 的有限重试和 dead-letter 已分别接入 `scheduler.job.retry` 与
|
||||||
`scheduler.job.dead_letter`,只使用固定 `owner=outbox` 低基数标签;观测失败端口由 Startup 注入,
|
`scheduler.job.dead_letter`,只使用固定 `owner=outbox` 低基数标签;观测失败端口由 Startup 注入,
|
||||||
Application 不依赖具体 OTel SDK。
|
Application 不依赖具体 OTel SDK。
|
||||||
|
- 2026-08-23 增加 `compat.facade.hit` 低基数指标和 `observe_compat_facade()` 装饰器,覆盖
|
||||||
|
`PluginManager`、`PluginHelper`、`MoviePilotServerHelper` 三个正式 V3 ABI 入口。装饰器保留同步/
|
||||||
|
异步 descriptor、原方法签名和对象身份,并同时记录公开方法与旧私有方法的命中,标签仅包含
|
||||||
|
facade、稳定方法名、可见性和固定 ABI 来源,不包含插件/用户/媒体实例数据。专项测试验证三类
|
||||||
|
Facade 的公开与私有命中,后续可按真实命中量和行为快照逐项内移算法。
|
||||||
- 专项测试覆盖 exporter 缺失、非法标签、全目录高基数审计、成功/失败 outcome、动态 URL 路由模板;
|
- 专项测试覆盖 exporter 缺失、非法标签、全目录高基数审计、成功/失败 outcome、动态 URL 路由模板;
|
||||||
既有 API、Event、Module、Scheduler 与健康探针回归保持通过。
|
既有 API、Event、Module、Scheduler 与健康探针回归保持通过。
|
||||||
|
|
||||||
@@ -995,6 +1000,8 @@ MFA/Passkey 专项测试与架构门禁通过,密钥类配置仍保留在安
|
|||||||
两个公开入口均降至 150 行预算内,复杂度基线移除对应债务项;订阅识别、音乐订阅、写入事务和搜索来源专项
|
两个公开入口均降至 150 行预算内,复杂度基线移除对应债务项;订阅识别、音乐订阅、写入事务和搜索来源专项
|
||||||
共 280 项测试通过,架构、复杂度与异步阻塞门禁通过。
|
共 280 项测试通过,架构、复杂度与异步阻塞门禁通过。
|
||||||
|
|
||||||
|
2026-08-23 将工作流动作 `FetchRssAction`、`ScanFileAction` 和 `AddSubscribeAction` 接入 `ChainRuntimeConfig` 快照,分别移除代理、媒体后缀和超级用户的全局 `settings` 读取;保留动作公开入口与工作流上下文行为,新增快照注入测试覆盖。配置债务由 130 个文件降至 127 个文件,宿主依赖与配置基线已更新。
|
||||||
|
|
||||||
**收口记录(2026-08-22)**:`reidentify_cache`、`nettest`、`scrape`、OpenAI `chat_completions/responses`、`get_logging` 和 Web Agent SSE 均改为稳定公开入口委托私有编排实现;四个消息交互 Handler 的公开方法也保留 ABI 并委托私有状态机。复杂度基线已清零,API/Application/Chain 入口预算、异步阻塞 ratchet 均通过;复杂度及兼容专项合计 252 项测试通过。
|
**收口记录(2026-08-22)**:`reidentify_cache`、`nettest`、`scrape`、OpenAI `chat_completions/responses`、`get_logging` 和 Web Agent SSE 均改为稳定公开入口委托私有编排实现;四个消息交互 Handler 的公开方法也保留 ABI 并委托私有状态机。复杂度基线已清零,API/Application/Chain 入口预算、异步阻塞 ratchet 均通过;复杂度及兼容专项合计 252 项测试通过。
|
||||||
随后将 `TransferChain.do_transfer` 的公开入口收口为稳定兼容 Facade,先提取媒体身份规范化阶段,保留显式
|
随后将 `TransferChain.do_transfer` 的公开入口收口为稳定兼容 Facade,先提取媒体身份规范化阶段,保留显式
|
||||||
`media_source/media_id` 校验、识别失败文案和所有原有调用参数;整理专项 80 项测试通过,复杂度基线移除该入口,
|
`media_source/media_id` 校验、识别失败文案和所有原有调用参数;整理专项 80 项测试通过,复杂度基线移除该入口,
|
||||||
|
|||||||
@@ -9,7 +9,7 @@
|
|||||||
"root": "app"
|
"root": "app"
|
||||||
},
|
},
|
||||||
"settings_imports": {
|
"settings_imports": {
|
||||||
"count": 130,
|
"count": 127,
|
||||||
"files": [
|
"files": [
|
||||||
"app/adapters/cache/backends.py",
|
"app/adapters/cache/backends.py",
|
||||||
"app/adapters/cache/redis.py",
|
"app/adapters/cache/redis.py",
|
||||||
@@ -136,10 +136,7 @@
|
|||||||
"app/startup/modules_initializer.py",
|
"app/startup/modules_initializer.py",
|
||||||
"app/startup/plugins_initializer.py",
|
"app/startup/plugins_initializer.py",
|
||||||
"app/startup/routers_initializer.py",
|
"app/startup/routers_initializer.py",
|
||||||
"app/workflow/actions/add_subscribe.py",
|
|
||||||
"app/workflow/actions/fetch_medias.py",
|
"app/workflow/actions/fetch_medias.py",
|
||||||
"app/workflow/actions/fetch_rss.py",
|
|
||||||
"app/workflow/actions/scan_file.py",
|
|
||||||
"app/workflow/actions/send_message.py"
|
"app/workflow/actions/send_message.py"
|
||||||
]
|
]
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -0,0 +1,131 @@
|
|||||||
|
from datetime import datetime
|
||||||
|
from types import SimpleNamespace
|
||||||
|
|
||||||
|
from app.application.configuration import ChainRuntimeConfig
|
||||||
|
from app.schemas import ActionContext
|
||||||
|
from app.schemas.context import MediaInfo
|
||||||
|
from app.schemas.types import MediaType
|
||||||
|
from app.workflow.actions.add_subscribe import AddSubscribeAction
|
||||||
|
from app.workflow.actions.fetch_rss import FetchRssAction
|
||||||
|
from app.workflow.actions.scan_file import ScanFileAction
|
||||||
|
from app.workflow.actions import add_subscribe as add_subscribe_module
|
||||||
|
from app.workflow.actions import fetch_rss as fetch_rss_module
|
||||||
|
from app.workflow.actions import scan_file as scan_file_module
|
||||||
|
|
||||||
|
|
||||||
|
def _runtime_config(**overrides):
|
||||||
|
"""构造仅覆盖测试关注字段的 Chain 配置快照。"""
|
||||||
|
values = {
|
||||||
|
"media_extensions": (".mkv",),
|
||||||
|
"subtitle_extensions": (".srt",),
|
||||||
|
"audio_extensions": (".flac",),
|
||||||
|
"superuser": "snapshot-admin",
|
||||||
|
"proxy": {"https": "http://snapshot-proxy:7890"},
|
||||||
|
}
|
||||||
|
values.update(overrides)
|
||||||
|
return ChainRuntimeConfig(**values)
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_rss_reads_proxy_from_chain_snapshot(monkeypatch):
|
||||||
|
"""RSS 动作应使用一次 Chain 快照中的代理,而不是全局 settings。"""
|
||||||
|
captured = {}
|
||||||
|
|
||||||
|
class FakeRssHelper:
|
||||||
|
"""记录 RSS 请求参数的测试替身。"""
|
||||||
|
|
||||||
|
def parse(self, **kwargs):
|
||||||
|
captured.update(kwargs)
|
||||||
|
return [{
|
||||||
|
"title": "Example",
|
||||||
|
"enclosure": "https://example.com/example.torrent",
|
||||||
|
"link": "https://example.com/details",
|
||||||
|
"size": 1,
|
||||||
|
"pubdate": datetime(2026, 1, 1),
|
||||||
|
}]
|
||||||
|
|
||||||
|
monkeypatch.setattr(fetch_rss_module, "RssHelper", FakeRssHelper)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
fetch_rss_module,
|
||||||
|
"get_chain_runtime_config_snapshot",
|
||||||
|
lambda: _runtime_config(),
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(fetch_rss_module.global_vars, "is_workflow_stopped", lambda _: False)
|
||||||
|
|
||||||
|
FetchRssAction("rss").execute(
|
||||||
|
workflow_id=1,
|
||||||
|
params={"url": "https://example.com/rss.xml", "proxy": True},
|
||||||
|
context=ActionContext(),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert captured["proxy"] == {"https": "http://snapshot-proxy:7890"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_scan_file_filters_extensions_from_chain_snapshot(monkeypatch):
|
||||||
|
"""扫描动作应按快照后缀集合筛选媒体文件。"""
|
||||||
|
|
||||||
|
class FakeStorageChain:
|
||||||
|
"""返回固定文件列表的存储链测试替身。"""
|
||||||
|
|
||||||
|
def get_file_item(self, storage, directory):
|
||||||
|
return SimpleNamespace(storage=storage, path=str(directory))
|
||||||
|
|
||||||
|
def list_files(self, fileitem, recursion=True):
|
||||||
|
return [
|
||||||
|
SimpleNamespace(extension="mkv"),
|
||||||
|
SimpleNamespace(extension="txt"),
|
||||||
|
SimpleNamespace(extension="srt"),
|
||||||
|
]
|
||||||
|
|
||||||
|
monkeypatch.setattr(scan_file_module, "StorageChain", FakeStorageChain)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
scan_file_module,
|
||||||
|
"get_chain_runtime_config_snapshot",
|
||||||
|
lambda: _runtime_config(),
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(scan_file_module.global_vars, "is_workflow_stopped", lambda _: False)
|
||||||
|
|
||||||
|
context = ScanFileAction("scan").execute(
|
||||||
|
workflow_id=1,
|
||||||
|
params={"storage": "local", "directory": "/library"},
|
||||||
|
context=ActionContext(),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert [item.extension for item in context.fileitems] == ["mkv", "srt"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_add_subscribe_uses_superuser_from_chain_snapshot(monkeypatch):
|
||||||
|
"""添加订阅动作应将快照中的超级管理员传给订阅链。"""
|
||||||
|
captured = {}
|
||||||
|
|
||||||
|
class FakeSubscribeChain:
|
||||||
|
"""记录订阅新增参数的测试替身。"""
|
||||||
|
|
||||||
|
def exists(self, _mediainfo):
|
||||||
|
return False
|
||||||
|
|
||||||
|
def add(self, **kwargs):
|
||||||
|
captured.update(kwargs)
|
||||||
|
return 42, "ok"
|
||||||
|
|
||||||
|
monkeypatch.setattr(add_subscribe_module, "SubscribeChain", FakeSubscribeChain)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
add_subscribe_module,
|
||||||
|
"get_chain_runtime_config_snapshot",
|
||||||
|
lambda: _runtime_config(superuser="snapshot-owner"),
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(add_subscribe_module.global_vars, "is_workflow_stopped", lambda _: False)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
add_subscribe_module,
|
||||||
|
"SubscribeOper",
|
||||||
|
lambda: SimpleNamespace(get=lambda sid: sid),
|
||||||
|
)
|
||||||
|
|
||||||
|
AddSubscribeAction("subscribe").execute(
|
||||||
|
workflow_id=1,
|
||||||
|
params={},
|
||||||
|
context=ActionContext(
|
||||||
|
medias=[MediaInfo(type=MediaType.MOVIE, title="Example", year="2026")]
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert captured["username"] == "snapshot-owner"
|
||||||
Reference in New Issue
Block a user