fix(transfer): expire stale jobs and deduplicate diagnostics

This commit is contained in:
jxxghp
2026-08-07 12:42:53 +08:00
parent 63e492be7c
commit 759b9e47eb
13 changed files with 713 additions and 64 deletions

View File

@@ -6,6 +6,7 @@ import traceback
import uuid
from copy import deepcopy
from pathlib import Path
from time import monotonic
from typing import List, Optional, Tuple, Union, Dict, Callable, Any
from app import schemas
@@ -122,11 +123,17 @@ class JobManager:
_season_episodes: Dict[Tuple, List[int]] = {}
# 记录从 meta 作业迁移到 media 作业的关系,用于清理提前失败后残留的 media 作业
_meta_to_media_ids: Dict[Tuple, set[Tuple]] = {}
# 记录任务最近一次状态心跳,供外部异步接管任务的失活检测使用
_task_state_changed_at: Dict[Tuple[str, str], float] = {}
# 记录仍由主程序整理线程直接执行的任务,避免把阻塞中的本地任务误判为失活
_active_executions: set[Tuple[str, str]] = set()
def __init__(self):
self._job_view = {}
self._season_episodes = {}
self._meta_to_media_ids = {}
self._task_state_changed_at = {}
self._active_executions = set()
@staticmethod
def __get_meta_id(meta: MetaBase = None, season: Optional[int] = None) -> Tuple:
@@ -248,6 +255,7 @@ class JobManager:
state=state,
)
)
self._task_state_changed_at[file_key] = monotonic()
# 添加季集信息
if self._season_episodes.get(__mediaid__):
self._season_episodes[__mediaid__].extend(task.meta.episode_list)
@@ -262,7 +270,9 @@ class JobManager:
"""
将任务从 meta 作业迁移到 media 作业
"""
curr_task, source_job_id = self.__remove_task_with_job_id(task.fileitem)
curr_task, source_job_id = self.__remove_task_with_job_id(
task.fileitem, preserve_execution=True
)
if not self.add_task(task, state=curr_task.state if curr_task else "waiting"):
return False
if curr_task and task.mediainfo:
@@ -290,14 +300,116 @@ class JobManager:
"""
移除指定作业和对应季集缓存
"""
if job_id in self._season_episodes:
self._season_episodes.pop(job_id)
if job_id in self._job_view:
self._job_view.pop(job_id)
job = self._job_view.pop(job_id, None)
self._season_episodes.pop(job_id, None)
if not job:
return
for task in job.tasks:
file_key = self.__get_file_key(task.fileitem)
if file_key:
self._task_state_changed_at.pop(file_key, None)
self._active_executions.discard(file_key)
def __remove_done_job_groups(self, job_ids: set[Tuple]):
"""
清理已进入终态的独立作业或关联作业组。
"""
candidates = set(job_ids)
for metaid, mediaids in list(self._meta_to_media_ids.items()):
related_ids = {metaid, *mediaids}
if not related_ids.intersection(candidates):
continue
if all(self.__is_job_done(job_id) for job_id in related_ids):
for job_id in related_ids:
self.__pop_job(job_id)
self._meta_to_media_ids.pop(metaid, None)
candidates.difference_update(related_ids)
referenced_ids = {
job_id
for metaid, mediaids in self._meta_to_media_ids.items()
for job_id in {metaid, *mediaids}
}
for job_id in candidates - referenced_ids:
if self.__is_job_done(job_id):
self.__pop_job(job_id)
def start_execution(self, task: TransferTask):
"""
标记任务仍由主程序整理线程直接执行。
:param task: 整理任务
"""
if not task or not task.fileitem:
return
file_key = self.__get_file_key(task.fileitem)
if not file_key:
return
with job_lock:
self._active_executions.add(file_key)
def finish_execution(self, task: TransferTask):
"""
结束主程序整理线程对任务的直接执行标记。
:param task: 整理任务
"""
if not task or not task.fileitem:
return
file_key = self.__get_file_key(task.fileitem)
if not file_key:
return
with job_lock:
self._active_executions.discard(file_key)
def expire_stale_running_tasks(
self, timeout_seconds: int
) -> List[Tuple[FileItem, int]]:
"""
将外部接管后长期无心跳的运行中任务标记失败并清理作业视图。
主程序整理线程仍在直接执行的任务不会被清理,以免把阻塞中的真实任务
误报为已终止。外部接管方可重复调用 ``running_task`` 刷新状态心跳。
:param timeout_seconds: 失活超时秒数,小于等于 0 时禁用
:return: 已失活任务及其无心跳秒数
"""
if timeout_seconds <= 0:
return []
current_time = monotonic()
expired: List[Tuple[FileItem, int]] = []
affected_job_ids: set[Tuple] = set()
with job_lock:
for mediaid, job in self._job_view.items():
for task in job.tasks:
file_key = self.__get_file_key(task.fileitem)
if (
not file_key
or task.state != "running"
or file_key in self._active_executions
):
continue
updated_at = self._task_state_changed_at.get(file_key, current_time)
inactive_seconds = current_time - updated_at
if inactive_seconds < timeout_seconds:
continue
task.state = "failed"
self._task_state_changed_at[file_key] = current_time
episodes = getattr(task.meta, "episode_list", None) or []
if mediaid in self._season_episodes:
self._season_episodes[mediaid] = list(
set(self._season_episodes[mediaid]) - set(episodes)
)
expired.append((task.fileitem, int(inactive_seconds)))
affected_job_ids.add(mediaid)
self.__remove_done_job_groups(affected_job_ids)
return expired
def running_task(self, task: TransferTask):
"""
设置任务为运行中
设置任务为运行中,并刷新外部异步任务的状态心跳。
"""
with job_lock:
__mediaid__ = self.__get_id(task)
@@ -307,6 +419,9 @@ class JobManager:
for t in self._job_view[__mediaid__].tasks:
if t.fileitem == task.fileitem:
t.state = "running"
file_key = self.__get_file_key(t.fileitem)
if file_key:
self._task_state_changed_at[file_key] = monotonic()
break
def finish_task(self, task: TransferTask):
@@ -321,6 +436,9 @@ class JobManager:
for t in self._job_view[__mediaid__].tasks:
if t.fileitem == task.fileitem:
t.state = "completed"
file_key = self.__get_file_key(t.fileitem)
if file_key:
self._task_state_changed_at[file_key] = monotonic()
break
def fail_task(self, task: TransferTask):
@@ -335,6 +453,9 @@ class JobManager:
for t in self._job_view[__mediaid__].tasks:
if t.fileitem == task.fileitem:
t.state = "failed"
file_key = self.__get_file_key(t.fileitem)
if file_key:
self._task_state_changed_at[file_key] = monotonic()
break
# 移除剧集信息
if __mediaid__ in self._season_episodes:
@@ -359,6 +480,7 @@ class JobManager:
continue
if job_task.state not in ["completed", "failed"]:
job_task.state = "failed"
self._task_state_changed_at[file_key] = monotonic()
if mediaid in self._season_episodes:
self._season_episodes[mediaid] = list(
set(self._season_episodes[mediaid])
@@ -374,7 +496,9 @@ class JobManager:
return task
def __remove_task_with_job_id(
self, fileitem: FileItem
self,
fileitem: FileItem,
preserve_execution: bool = False,
) -> Tuple[Optional[TransferJobTask], Optional[Tuple]]:
"""
根据文件项移除任务并返回任务所在的作业ID
@@ -388,6 +512,9 @@ class JobManager:
for task in job.tasks:
if self.__get_file_key(task.fileitem) == file_key:
job.tasks.remove(task)
self._task_state_changed_at.pop(file_key, None)
if not preserve_execution:
self._active_executions.discard(file_key)
# 如果没有作业了,则移除作业
if not job.tasks:
self._job_view.pop(mediaid)
@@ -407,10 +534,9 @@ class JobManager:
with job_lock:
__mediaid__ = self.__get_id(task)
if __mediaid__ in self._job_view:
# 移除季集信息
if __mediaid__ in self._season_episodes:
self._season_episodes.pop(__mediaid__)
return self._job_view.pop(__mediaid__)
job = self._job_view[__mediaid__]
self.__pop_job(__mediaid__)
return job
return None
def try_remove_job(self, task: TransferTask):
@@ -1509,6 +1635,33 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
return
self.jobview.remove_task(fileitem)
def __start_job_execution(self, task: TransferTask):
"""在作业视图支持执行租约时标记主程序任务开始执行。"""
marker = getattr(self.jobview, "start_execution", None)
if marker:
marker(task)
def __finish_job_execution(self, task: TransferTask):
"""在作业视图支持执行租约时标记主程序任务结束执行。"""
marker = getattr(self.jobview, "finish_execution", None)
if marker:
marker(task)
def __expire_stale_transfer_tasks(self):
"""清理外部接管后失去状态心跳的运行中整理任务。"""
timeout_minutes = max(int(settings.TRANSFER_TASK_TIMEOUT), 0)
expire_tasks = getattr(self.jobview, "expire_stale_running_tasks", None)
expired_tasks = (
expire_tasks(timeout_seconds=timeout_minutes * 60)
if expire_tasks
else []
)
for fileitem, inactive_seconds in expired_tasks:
logger.error(
f"整理任务 {fileitem.path} 已连续 {inactive_seconds // 60} 分钟无状态心跳,"
"已标记失败并从整理队列视图清理"
)
def __fail_transfer_task(self, task: TransferTask):
"""
标记异常整理任务失败并清理作业视图
@@ -1560,6 +1713,7 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
self._active_tasks += 1
try:
self.__start_job_execution(task)
# 更新进度
__process_msg = f"正在整理 {fileitem.name} ..."
logger.info(__process_msg)
@@ -1598,6 +1752,7 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
self._processed_num += 1
self._fail_num += 1
finally:
self.__finish_job_execution(task)
self._queue.task_done()
with task_lock:
# 减少运行中的任务数
@@ -1618,6 +1773,7 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
except queue.Empty:
# 即使队列空了,如果还有任务在运行,也不应该结束进度
# 这部分逻辑已经在 finally 的 active_tasks == 0 中处理了
self.__expire_stale_transfer_tasks()
continue
except Exception as e:
logger.error(f"整理队列处理出现错误:{e} - {traceback.format_exc()}")
@@ -1914,6 +2070,7 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
"""
获取整理任务列表
"""
self.__expire_stale_transfer_tasks()
return self.jobview.list_jobs()
def recommend_name(self, meta: MetaBase, mediainfo: MediaInfo) -> Optional[str]:
@@ -3446,6 +3603,7 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
},
)
try:
self.__start_job_execution(transfer_task)
state, err_msg = self.__handle_transfer(
task=transfer_task,
callback=_preview_callback if preview else self.__default_callback,
@@ -3458,6 +3616,8 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
if not preview:
self.__fail_transfer_task(transfer_task)
state, err_msg = False, str(e)
finally:
self.__finish_job_execution(transfer_task)
if not state:
all_success = False
logger.warn(f"{transfer_task.fileitem.name} {err_msg}")

View File

@@ -391,6 +391,8 @@ class ConfigModel(BaseModel):
# ==================== 整理配置 ====================
# 文件整理线程数
TRANSFER_THREADS: int = 1
# 外部接管的运行中整理任务无状态心跳超时分钟0 表示禁用
TRANSFER_TASK_TIMEOUT: int = 120
# 电影重命名格式
MOVIE_RENAME_FORMAT: str = (
"{{title}}{% if year %} ({{year}}){% endif %}"

View File

@@ -9,6 +9,7 @@ import socket
import sqlite3
import sys
from collections import deque
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Callable, Optional
from urllib.error import HTTPError, URLError
@@ -48,6 +49,11 @@ LOG_ERROR_PATTERNS = (
LOG_RECORD_PATTERN = re.compile(
r"(?:【(?:DEBUG|INFO|WARNING|ERROR|CRITICAL)】|(?:DEBUG|INFO|WARNING|ERROR|CRITICAL):)"
)
LOG_TIMESTAMP_PATTERN = re.compile(
r"(\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2})"
)
LOG_TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S"
LOG_LOOKBACK_HOURS = 24
CONSOLE_LOGGER_PATTERN = re.compile(r"\[([^\]]+)]")
PLUGIN_ERROR_PATTERNS = (
re.compile(r"(?:^|\s-\s)plugin\.py\s+-\s", re.IGNORECASE),
@@ -295,6 +301,78 @@ def _tail_lines(path: Path, max_lines: int = 120, max_bytes: int = 256 * 1024) -
return list(deque((_mask_text(line) for line in text.splitlines()), maxlen=max_lines))
def _parse_log_timestamp(line: str) -> Optional[datetime]:
"""解析日志行中的时间戳。"""
match = LOG_TIMESTAMP_PATTERN.search(line[:96])
if not match:
return None
try:
return datetime.strptime(match.group(1), LOG_TIMESTAMP_FORMAT)
except ValueError:
return None
def _recent_log_lines(
lines: list[str],
now: Optional[datetime] = None,
) -> list[str]:
"""按日志记录边界保留诊断时间窗内的日志。"""
if not lines:
return []
timestamps = [_parse_log_timestamp(line) for line in lines]
if not any(timestamps):
return lines
cutoff = (now or datetime.now()) - timedelta(hours=LOG_LOOKBACK_HOURS)
recent: list[str] = []
include_record = False
for line, timestamp in zip(lines, timestamps):
if timestamp is not None:
include_record = timestamp >= cutoff
if include_record:
recent.append(line)
return recent
def _error_fingerprint(line: str) -> str:
"""生成跨主日志、控制台镜像和插件独立日志可比较的错误指纹。"""
normalized = LOG_TIMESTAMP_PATTERN.sub("<time>", line.strip())
if " - " in normalized:
normalized = normalized.rsplit(" - ", 1)[-1]
normalized = re.sub(
r"^(?:【(?:ERROR|CRITICAL|WARNING)】|(?:ERROR|CRITICAL|WARNING):)\s*",
"",
normalized,
flags=re.IGNORECASE,
)
return re.sub(r"\s+", " ", normalized).strip().lower()
def _aggregate_log_entries(
entries: list[tuple[Path, str]],
max_matches: int = 12,
) -> tuple[list[str], list[str]]:
"""合并跨日志的重复错误,并返回详情行和来源文件。"""
unique_entries: dict[str, dict[str, Any]] = {}
log_files: list[str] = []
for path, line in entries:
path_text = str(path)
if path_text not in log_files:
log_files.append(path_text)
fingerprint = _error_fingerprint(line)
if fingerprint not in unique_entries:
unique_entries[fingerprint] = {"line": line, "sources": []}
sources = unique_entries[fingerprint]["sources"]
if path.name not in sources:
sources.append(path.name)
details = [
f"[{', '.join(item['sources'])}] {item['line']}"
for item in list(unique_entries.values())[-max_matches:]
]
return details, log_files
def _find_error_lines(lines: list[str], max_matches: int = 12) -> list[str]:
"""从近期日志中提取错误关键词命中的行。"""
matches: list[str] = []
@@ -756,11 +834,15 @@ def _check_logs(runner: DoctorRunnerProtocol) -> None:
log_files.extend(plugin_log_files[:20])
found_any = False
entries: dict[str, list[tuple[Path, str]]] = {
"core": [],
"plugin": [],
}
for path in log_files:
if not path.exists() or not path.is_file():
continue
found_any = True
lines = _tail_lines(path)
lines = _recent_log_lines(_tail_lines(path))
is_plugin_log = plugin_log_dir in path.parents
if is_plugin_log:
scoped_errors = [(True, _find_error_lines(lines))]
@@ -770,35 +852,9 @@ def _check_logs(runner: DoctorRunnerProtocol) -> None:
plugin_logger_names,
)
scoped_errors = [(False, core_errors), (True, plugin_errors)]
if not any(errors for _, errors in scoped_errors):
continue
has_core_errors = bool(scoped_errors[0][1]) if not is_plugin_log else False
for is_plugin_error, errors in scoped_errors:
if not errors:
continue
finding_suffix = (
"plugin_errors"
if is_plugin_error and has_core_errors
else "recent_errors"
)
runner.add(
finding_id=f"logs.{path.stem}.{finding_suffix}",
severity=DoctorSeverity.Warn,
status=DoctorFindingStatus.Degraded,
title="最近日志存在插件异常" if is_plugin_error else "最近日志存在错误线索",
detail="\n".join(errors),
recommendation=(
"可使用安全模式启动后检查插件配置。"
if is_plugin_error
else "结合前后的启动日志定位异常;必要时执行 `moviepilot doctor --json` 交给 Agent 或 Issue 流程。"
),
affects_report_status=not is_plugin_error,
context={
"log_file": str(path),
"matches": len(errors),
"component": "plugin" if is_plugin_error else "core",
},
)
component = "plugin" if is_plugin_error else "core"
entries[component].extend((path, error) for error in errors)
if not found_any:
runner.add(
@@ -811,13 +867,50 @@ def _check_logs(runner: DoctorRunnerProtocol) -> None:
)
return
if not any(finding.id.startswith("logs.") and finding.id.endswith("recent_errors") for finding in runner.report.findings):
core_first_path = entries["core"][0][0] if entries["core"] else None
for component in ("core", "plugin"):
component_entries = entries[component]
if not component_entries:
continue
first_path = component_entries[0][0]
finding_suffix = (
"plugin_errors"
if component == "plugin" and first_path == core_first_path
else "recent_errors"
)
detail_lines, source_files = _aggregate_log_entries(component_entries)
runner.add(
finding_id=f"logs.{first_path.stem}.{finding_suffix}",
severity=DoctorSeverity.Warn,
status=DoctorFindingStatus.Degraded,
title="最近日志存在插件异常" if component == "plugin" else "最近日志存在错误线索",
detail="\n".join(detail_lines),
recommendation=(
"可使用安全模式启动后检查插件配置。"
if component == "plugin"
else "结合前后的启动日志定位异常;必要时执行 `moviepilot doctor --json` 交给 Agent 或 Issue 流程。"
),
affects_report_status=component == "core",
context={
"log_file": str(first_path),
"log_files": source_files,
"matches": len(component_entries),
"unique_matches": len(detail_lines),
"component": component,
"lookback_hours": LOG_LOOKBACK_HOURS,
},
)
if not entries["core"]:
runner.add(
finding_id="logs.recent",
severity=DoctorSeverity.Info,
status=DoctorFindingStatus.Ok,
title="最近日志未发现明显错误关键词",
detail=f"已扫描 {settings.LOG_PATH} 下的主日志、启动日志和插件日志。",
detail=(
f"已扫描 {settings.LOG_PATH} 下最近 {LOG_LOOKBACK_HOURS} 小时的主日志、"
"启动日志和插件日志;插件扩展告警不参与核心健康状态。"
),
recommendation="如果问题仍存在,请结合具体操作时间扩大日志范围排查。",
)

View File

@@ -38,7 +38,9 @@ def format_text_report(report: DoctorReport) -> str:
summary = report.summary
lines.extend([
"",
f"汇总: total={summary['total']} error={summary['error']} warn={summary['warn']} fixed={summary['fixed']}",
f"汇总: total={summary['total']} error={summary['error']} "
f"warn={summary['warn']} advisory={summary['advisory']} "
f"fixed={summary['fixed']}",
])
return "\n".join(lines)

View File

@@ -114,11 +114,14 @@ class DoctorReport:
"warn": 0,
"error": 0,
"fixed": 0,
"advisory": 0,
}
for finding in self.findings:
counts[finding.severity.value] += 1
if finding.fixed:
counts["fixed"] += 1
elif not finding.affects_report_status:
counts["advisory"] += 1
return counts
def exit_code(self) -> int:

View File

@@ -378,6 +378,7 @@ moviepilot version
- 通过系统内置的重启入口触发重启时,本地 CLI 安装模式也会复用同一套前后端进程管理完成重启
- 前端默认监听 `NGINX_PORT`,默认值 `3000`
- 后端默认监听 `PORT`,默认值 `3001`
- `TRANSFER_TASK_TIMEOUT` 控制外部异步接管的运行中整理任务失活超时,单位为分钟,默认 `120`,设为 `0` 可禁用;主程序整理线程仍在直接执行的任务不受此项清理
- 前端通过 `service.js` 代理 `/api``/cookiecloud` 到后端
- 本地前端代理在启动时会先确认后端可用;如果后端长时间不可用,前端也会自动退出,避免只剩半套服务
@@ -396,7 +397,8 @@ moviepilot doctor --deep
- `--json` 输出稳定 JSON可供 Agent、脚本或 Issue 流程收集
- `--fix` 只执行白名单安全修复,例如清理过期 runtime 文件或补齐不合法的 `API_TOKEN`
- `--deep` 执行可能较慢的深度探测,例如 PostgreSQL TCP 连通性检查
- 插件日志异常会保留为诊断告警并标记 `affects_report_status=false`,但不会单独降低系统整体状态;核心错误仍正常参与状态聚合
- Doctor 只分析最近 24 小时日志,并跨主日志、控制台镜像和插件独立日志聚合相同错误
- 插件日志异常会保留为诊断告警并标记 `affects_report_status=false`,但不会单独降低系统整体状态;`summary.advisory` 单独统计这类建议项,核心错误仍正常参与状态聚合
- Docker 环境可使用 `docker exec <container> moviepilot doctor`;如果容器已退出,也可用镜像挂载同一配置目录运行 `python -m app.cli doctor`
日志:

View File

@@ -38,7 +38,7 @@ Doctor 默认执行只读检查:
- 运行路径程序目录、配置目录、日志目录、Python 解释器
- 关键配置:`API_TOKEN``PORT``NGINX_PORT`、代理格式、安全模式
- 进程与端口后端、前端端口监听状态runtime 文件是否过期
- 日志线索:后端日志、启动日志、前端日志和插件日志中的近期错误
- 日志线索:后端日志、启动日志、前端日志和插件日志最近 24 小时内的错误
- 核心依赖FastAPI、Pydantic、SQLAlchemy、Uvicorn、CloakBrowser 等是否可导入
- 数据库SQLite 只读打开和完整性检查PostgreSQL 默认做配置检查
- 前端资源:`version.txt``service.js` 或核心静态文件是否存在
@@ -48,6 +48,8 @@ Doctor 默认执行只读检查:
整体状态只聚合会影响 MoviePilot 核心运行的诊断项。插件独立日志以及主日志中可明确识别的插件子系统异常仍会作为 `warn/degraded` 诊断项保留,但其 `affects_report_status``false`,不会单独把整体状态从 `healthy` 降为 `degraded`;同一日志中若还存在核心错误,核心错误仍会参与状态聚合。
Doctor 会按核心与插件两个组件聚合日志发现并对主日志、控制台镜像和插件独立日志中的相同错误去重。JSON 汇总中的 `warn` 保留全部警告数量,`advisory` 单独统计不影响整体状态的建议项;日志发现的 `context.log_files``matches``unique_matches` 分别说明来源、原始命中数和去重后命中数。
## 自救能力
`moviepilot doctor --fix` 只做白名单安全修复:
@@ -87,4 +89,4 @@ Dockerfile 同时提供 `HEALTHCHECK`,用于标记容器健康状态。是否
## Issue 反馈集成
`feedback-issue` skill 的诊断收集脚本会自动调用 `moviepilot doctor --json`,并把 doctor 摘要写入预览和最终 Issue 正文。完整 doctor JSON 存在运行时 diagnostics 文件中,默认不会直接贴入 Issue避免泄露本机路径和过长输出。
`feedback-issue` skill 的诊断收集脚本会自动调用 `moviepilot doctor --json`,并把 doctor 摘要写入预览和最终 Issue 正文。完整 doctor JSON 存在运行时 diagnostics 文件中,默认不会直接贴入 Issue避免泄露本机路径和过长输出。连续重复的同类日志模板会保留首条、末条和重复次数,避免轮询或等待日志挤掉真正的错误上下文。

View File

@@ -1,6 +1,6 @@
---
name: feedback-issue
version: 7
version: 8
description: >-
Use this skill ONLY when the user EXPLICITLY requests filing an
upstream issue for MoviePilot core, frontend, or an installed plugin,
@@ -92,6 +92,9 @@ Log relevance rules:
then applies a recent time window, removes Agent/tool dispatch noise,
and keeps only timestamped log blocks whose first line contains a
normalized keyword.
- Consecutive log records with the same template are compacted to the
first record, a repetition count, and the last record. Verify the
retained boundary records before treating the excerpt as evidence.
- If no specific keyword survives normalization, the script records the
doctor report and log-selection metadata but does not include recent
log lines. This avoids attaching unrelated noise.

View File

@@ -33,6 +33,10 @@ _LOG_TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S"
_LOG_MODULE_RE = re.compile(
r"^【[^】]+】\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2},\d+\s+-\s+([^\s][^\-]*?)\s+-\s+"
)
_LOG_DYNAMIC_VALUE_RE = re.compile(
r"(?<![A-Za-z])(?:[0-9a-f]{8,}|\d+(?:\.\d+)?)(?![A-Za-z])",
re.IGNORECASE,
)
_META_NOISE_MODULES = frozenset({
"collect_feedback_diagnostics.py",
@@ -238,6 +242,43 @@ def is_meta_noise(line: str) -> bool:
return match.group(1).strip() in _META_NOISE_MODULES
def _repetition_fingerprint(line: str) -> str:
"""生成用于识别连续重复日志模板的指纹。"""
if parse_line_timestamp(line) is None:
return line
normalized = _LOG_TIMESTAMP_RE.sub("<time>", line)
normalized = _LOG_DYNAMIC_VALUE_RE.sub("<value>", normalized)
return re.sub(r"\s+", " ", normalized).strip().lower()
def _compact_repeated_lines(lines: list[str]) -> list[str]:
"""压缩连续重复日志模板,同时保留首条、末条和重复次数。"""
compacted: list[str] = []
group: list[str] = []
fingerprint: Optional[str] = None
def flush_group() -> None:
if len(group) < 4:
compacted.extend(group)
return
compacted.append(group[0])
compacted.append(
f"... 同类日志连续重复 {len(group)} 次,已省略 {len(group) - 2} 行 ..."
)
compacted.append(group[-1])
for line in lines:
current_fingerprint = _repetition_fingerprint(line)
if group and current_fingerprint != fingerprint:
flush_group()
group = []
group.append(line)
fingerprint = current_fingerprint
if group:
flush_group()
return compacted
def filter_lines(
text: str,
keywords: list[str],
@@ -285,7 +326,8 @@ def filter_lines(
elif keep_block:
matched.append(line)
if matched:
return matched[-max_lines:], sorted(matched_keywords)
compacted = _compact_repeated_lines(matched)
return compacted[-max_lines:], sorted(matched_keywords)
return [], []

View File

@@ -371,6 +371,14 @@ def build_prefill_url(
return f"{issue_new_url(repo)}?{encoded}"
def _safe_count(value: Any) -> int:
"""把不可信的诊断计数字段转换为非负整数。"""
try:
return max(int(value or 0), 0)
except (TypeError, ValueError):
return 0
def format_doctor_summary(doctor: Optional[dict[str, Any]]) -> str:
"""把 doctor JSON 报告压缩成适合 Issue 和预览展示的摘要。"""
if not isinstance(doctor, dict):
@@ -390,28 +398,75 @@ def format_doctor_summary(doctor: Optional[dict[str, Any]]) -> str:
runtime = environment.get("runtime")
if runtime:
lines.append(f"运行环境:{runtime}")
findings = report.get("findings") or []
summary = report.get("summary") or {}
if isinstance(summary, dict):
advisory_count = summary.get("advisory")
if advisory_count is None and isinstance(findings, list):
advisory_count = sum(
1
for item in findings
if isinstance(item, dict)
and item.get("affects_report_status") is False
and not item.get("fixed")
)
lines.append(
"汇总:"
f"total={summary.get('total', 0)} "
f"error={summary.get('error', 0)} "
f"warn={summary.get('warn', 0)} "
f"advisory={advisory_count or 0} "
f"fixed={summary.get('fixed', 0)}"
)
findings = report.get("findings") or []
if isinstance(findings, list):
important = [
item for item in findings
if isinstance(item, dict) and item.get("severity") in {"error", "warn"}
][:8]
grouped: dict[tuple[str, str, str, bool], dict[str, Any]] = {}
for item in findings:
if not isinstance(item, dict) or item.get("severity") not in {"error", "warn"}:
continue
title = str(item.get("title") or item.get("id") or "未知诊断项")
recommendation = str(item.get("recommendation") or "").strip()
advisory = item.get("affects_report_status") is False
key = (str(item.get("severity")), title, recommendation, advisory)
group = grouped.setdefault(
key,
{
"count": 0,
"matches": 0,
"unique_matches": 0,
"sources": [],
},
)
group["count"] += 1
context = item.get("context") or {}
if not isinstance(context, dict):
continue
group["matches"] += _safe_count(context.get("matches"))
group["unique_matches"] += _safe_count(
context.get("unique_matches") or context.get("matches") or 0
)
source_files = context.get("log_files") or [context.get("log_file")]
for source_file in source_files:
if not source_file:
continue
source_name = Path(str(source_file)).name
if source_name not in group["sources"]:
group["sources"].append(source_name)
important = list(grouped.items())[:8]
if important:
lines.append("关键发现:")
for item in important:
title = str(item.get("title") or item.get("id") or "未知诊断项")
recommendation = str(item.get("recommendation") or "").strip()
line = f"- [{item.get('severity')}] {title}"
for (severity, title, recommendation, advisory), group in important:
marker = f"{severity}/advisory" if advisory else severity
line = f"- [{marker}] {title}"
if group["count"] > 1:
line = f"{line}(合并 {group['count']} 项)"
if group["sources"]:
line = f"{line};来源:{', '.join(group['sources'])}"
if group["matches"]:
line = f"{line};命中:{group['matches']}"
if group["unique_matches"] < group["matches"]:
line = f"{line},去重后 {group['unique_matches']}"
if recommendation:
line = f"{line};建议:{recommendation}"
lines.append(line)

View File

@@ -1,5 +1,6 @@
from __future__ import annotations
from datetime import datetime, timedelta
from types import SimpleNamespace
from app.core.config import settings
@@ -9,6 +10,11 @@ from app.doctor.models import DoctorFinding, DoctorFindingStatus, DoctorSeverity
from app.doctor.runner import DoctorRunner
def _current_log_timestamp() -> str:
"""返回 Doctor 近期日志测试使用的当前时间戳。"""
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def test_doctor_report_has_stable_json_shape(tmp_path, monkeypatch):
"""doctor JSON 报告应包含稳定状态、环境、汇总和发现列表。"""
monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path))
@@ -95,7 +101,7 @@ def test_doctor_plugin_log_error_does_not_degrade_report(tmp_path, monkeypatch):
plugin_log = settings.LOG_PATH / "plugins" / "demo.log"
plugin_log.parent.mkdir(parents=True, exist_ok=True)
plugin_log.write_text(
"【ERROR】2026-07-20 08:00:00 - demo.py - 插件任务执行异常\n",
f"【ERROR】{_current_log_timestamp()} - demo.py - 插件任务执行异常\n",
encoding="utf-8",
)
@@ -121,7 +127,7 @@ def test_doctor_plugin_load_error_in_main_log_does_not_degrade_report(
app_log = settings.LOG_PATH / "moviepilot.log"
app_log.parent.mkdir(parents=True, exist_ok=True)
app_log.write_text(
"【ERROR】2026-07-20 08:00:00 - plugin.py - 加载插件 Demo 出错boom - Traceback (most recent call last):\n"
f"【ERROR】{_current_log_timestamp()} - plugin.py - 加载插件 Demo 出错boom - Traceback (most recent call last):\n"
"Exception: boom\n",
encoding="utf-8",
)
@@ -146,12 +152,12 @@ def test_doctor_plugin_error_mirrored_to_stdio_does_not_degrade_report(
plugin_log = settings.LOG_PATH / "plugins" / "DemoPlugin.log"
plugin_log.parent.mkdir(parents=True, exist_ok=True)
plugin_log.write_text(
"【INFO】2026-07-20 08:00:00 - demo.py - 插件已启动\n",
f"【INFO】{_current_log_timestamp()} - demo.py - 插件已启动\n",
encoding="utf-8",
)
stdio_log = settings.LOG_PATH / "moviepilot.stdout.log"
stdio_log.write_text(
"ERROR: [demoplugin] 2026-07-20 08:01:00 demo.py - task exception\n",
f"ERROR: [demoplugin] {_current_log_timestamp()} demo.py - task exception\n",
encoding="utf-8",
)
@@ -171,7 +177,7 @@ def test_doctor_core_log_error_still_degrades_report(tmp_path, monkeypatch):
app_log = settings.LOG_PATH / "moviepilot.log"
app_log.parent.mkdir(parents=True, exist_ok=True)
app_log.write_text(
"【ERROR】2026-07-20 08:00:00 - rss.py - 解析 RSS 失败 - Traceback (most recent call last):\n"
f"【ERROR】{_current_log_timestamp()} - rss.py - 解析 RSS 失败 - Traceback (most recent call last):\n"
"RuntimeError: boom\n",
encoding="utf-8",
)
@@ -195,9 +201,9 @@ def test_doctor_mixed_plugin_and_core_log_errors_keep_core_status(
app_log = settings.LOG_PATH / "moviepilot.log"
app_log.parent.mkdir(parents=True, exist_ok=True)
app_log.write_text(
"【ERROR】2026-07-20 08:00:00 - plugin.py - 加载插件 Demo 出错boom - Traceback (most recent call last):\n"
f"【ERROR】{_current_log_timestamp()} - plugin.py - 加载插件 Demo 出错boom - Traceback (most recent call last):\n"
"Exception: plugin boom\n"
"【ERROR】2026-07-20 08:01:00 - rss.py - 解析 RSS 失败 - Traceback (most recent call last):\n"
f"【ERROR】{_current_log_timestamp()} - rss.py - 解析 RSS 失败 - Traceback (most recent call last):\n"
"Exception: core boom\n",
encoding="utf-8",
)
@@ -214,3 +220,55 @@ def test_doctor_mixed_plugin_and_core_log_errors_keep_core_status(
assert "core boom" in core_finding.detail
assert "plugin boom" in plugin_finding.detail
assert runner.report.status.value == "degraded"
def test_doctor_deduplicates_mirrored_plugin_errors(tmp_path, monkeypatch):
"""同一插件错误出现在主日志和插件日志时应只生成一条聚合告警。"""
monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path))
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
app_log = settings.LOG_PATH / "moviepilot.log"
plugin_log = settings.LOG_PATH / "plugins" / "demo.log"
plugin_log.parent.mkdir(parents=True, exist_ok=True)
app_log.write_text(
f"【ERROR】{timestamp} - plugin.py - 插件任务执行异常\n",
encoding="utf-8",
)
plugin_log.write_text(
f"【ERROR】{timestamp} - demo.py - 插件任务执行异常\n",
encoding="utf-8",
)
runner = DoctorRunner()
checks._check_logs(runner)
plugin_findings = [
finding
for finding in runner.report.findings
if finding.title == "最近日志存在插件异常"
]
assert len(plugin_findings) == 1
finding = plugin_findings[0]
assert finding.context["matches"] == 2
assert finding.context["unique_matches"] == 1
assert len(finding.context["log_files"]) == 2
assert runner.report.summary["advisory"] == 1
assert runner.report.find("logs.recent") is not None
def test_doctor_ignores_errors_outside_log_window(tmp_path, monkeypatch):
"""超出日志诊断时间窗的历史错误不应污染当前 Doctor 结果。"""
monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path))
timestamp = (datetime.now() - timedelta(hours=25)).strftime("%Y-%m-%d %H:%M:%S")
app_log = settings.LOG_PATH / "moviepilot.log"
app_log.parent.mkdir(parents=True, exist_ok=True)
app_log.write_text(
f"【ERROR】{timestamp} - rss.py - 历史解析错误\n",
encoding="utf-8",
)
runner = DoctorRunner()
checks._check_logs(runner)
assert runner.report.find("logs.moviepilot.recent_errors") is None
assert runner.report.find("logs.recent") is not None
assert runner.report.status.value == "healthy"

View File

@@ -0,0 +1,111 @@
"""Feedback Issue 日志压缩和 Doctor 摘要聚合测试。"""
import importlib.util
import sys
from datetime import datetime, timedelta
from pathlib import Path
import pytest
SCRIPT_DIR = Path(__file__).parents[1] / "skills" / "feedback-issue" / "scripts"
def _load_module(name: str, path: Path):
"""从脚本路径加载测试模块。"""
spec = importlib.util.spec_from_file_location(name, path)
assert spec and spec.loader
module = importlib.util.module_from_spec(spec)
sys.modules[name] = module
spec.loader.exec_module(module)
return module
@pytest.fixture
def feedback_modules():
"""加载 feedback 脚本,并在测试后恢复进程模块和搜索路径。"""
old_path = list(sys.path)
module_names = ["feedback_issue_common", "feedback_issue_collect_quality_test"]
old_modules = {name: sys.modules.get(name) for name in module_names}
try:
sys.path.insert(0, str(SCRIPT_DIR))
common = _load_module(
"feedback_issue_common",
SCRIPT_DIR / "feedback_issue_common.py",
)
collect = _load_module(
"feedback_issue_collect_quality_test",
SCRIPT_DIR / "collect_feedback_diagnostics.py",
)
yield common, collect
finally:
sys.path[:] = old_path
for name, module in old_modules.items():
if module is None:
sys.modules.pop(name, None)
else:
sys.modules[name] = module
def test_filter_lines_compacts_consecutive_repeated_templates(feedback_modules):
"""关键词命中的连续轮询日志应压缩为首条、计数和末条。"""
_, collect = feedback_modules
now = datetime.now()
lines = [
(
f"【INFO】{(now - timedelta(seconds=20 - index)).strftime('%Y-%m-%d %H:%M:%S')},000 "
f"- transfer.py - 等待转存任务完成:{index}/20"
)
for index in range(1, 11)
]
filtered, matched_keywords = collect.filter_lines(
"\n".join(lines),
keywords=["等待转存"],
max_lines=80,
window_start=now - timedelta(minutes=5),
)
assert len(filtered) == 3
assert "1/20" in filtered[0]
assert "连续重复 10 次" in filtered[1]
assert "10/20" in filtered[2]
assert matched_keywords == ["等待转存"]
def test_doctor_summary_groups_legacy_duplicate_advisories(feedback_modules):
"""旧版 Doctor 的重复插件发现也应在反馈摘要中合并展示。"""
common, _ = feedback_modules
findings = [
{
"severity": "warn",
"title": "最近日志存在插件异常",
"recommendation": "检查插件配置。",
"affects_report_status": False,
"context": {
"log_file": f"/config/logs/plugins/plugin-{index}.log",
"matches": 2,
},
}
for index in range(5)
]
summary = common.format_doctor_summary({
"success": True,
"report": {
"status": "healthy",
"summary": {
"total": 5,
"error": 0,
"warn": 5,
"fixed": 0,
},
"findings": findings,
},
})
assert summary.count("最近日志存在插件异常") == 1
assert "advisory=5" in summary
assert "warn/advisory" in summary
assert "合并 5 项" in summary
assert "plugin-0.log" in summary
assert "命中10 条" in summary

View File

@@ -0,0 +1,116 @@
"""整理任务失活收敛行为测试。"""
from app.chain import transfer
from app.chain.transfer import JobManager
from app.schemas import FileItem, TransferTask
from app.schemas.types import MediaType
class _FakeMeta:
"""提供整理任务分组需要的最小元数据。"""
def __init__(self):
self.name = "Test Show"
self.title = "Test Show S01E01"
self.year = "2026"
self.type = MediaType.TV
self.begin_season = 1
self.end_season = None
self.total_season = 1
self.begin_episode = 1
self.end_episode = None
self.total_episode = 1
self.episode_list = [1]
self.season_episode = "S01E01"
self.part = None
def to_dict(self):
"""返回 TransferJobTask 所需的元数据字典。"""
return {
"title": self.title,
"name": self.name,
"year": self.year,
"type": self.type.value,
"begin_season": self.begin_season,
"end_season": self.end_season,
"total_season": self.total_season,
"begin_episode": self.begin_episode,
"end_episode": self.end_episode,
"total_episode": self.total_episode,
"season_episode": self.season_episode,
"episode_list": self.episode_list,
"part": self.part,
}
def _make_task(name: str = "Test.Show.S01E01.mkv") -> TransferTask:
"""创建失活检测使用的整理任务。"""
return TransferTask(
fileitem=FileItem(
storage="local",
path=f"/downloads/{name}",
type="file",
name=name,
basename=name.removesuffix(".mkv"),
extension="mkv",
size=1024,
),
meta=_FakeMeta(),
)
def test_external_running_task_expires_without_heartbeat(monkeypatch):
"""外部接管的运行中任务超过心跳期限后应被标记失败并清理。"""
clock = [100.0]
monkeypatch.setattr(transfer, "monotonic", lambda: clock[0])
manager = JobManager()
task = _make_task()
assert manager.add_task(task)
manager.running_task(task)
clock[0] = 221.0
expired = manager.expire_stale_running_tasks(timeout_seconds=120)
assert expired == [(task.fileitem, 121)]
assert manager.list_jobs() == []
def test_main_thread_execution_is_not_expired(monkeypatch):
"""主程序整理线程仍在执行的任务不应被失活检测伪清理。"""
clock = [100.0]
monkeypatch.setattr(transfer, "monotonic", lambda: clock[0])
manager = JobManager()
task = _make_task()
assert manager.add_task(task)
manager.start_execution(task)
manager.running_task(task)
clock[0] = 500.0
assert manager.expire_stale_running_tasks(timeout_seconds=120) == []
assert manager.total() == 1
manager.finish_execution(task)
expired = manager.expire_stale_running_tasks(timeout_seconds=120)
assert expired == [(task.fileitem, 400)]
assert manager.list_jobs() == []
def test_waiting_task_and_refreshed_heartbeat_do_not_expire(monkeypatch):
"""等待中任务不受失活期限影响,重复运行状态更新可刷新外部心跳。"""
clock = [100.0]
monkeypatch.setattr(transfer, "monotonic", lambda: clock[0])
manager = JobManager()
waiting_task = _make_task("Test.Show.S01E01.waiting.mkv")
running_task = _make_task("Test.Show.S01E02.running.mkv")
running_task.meta.begin_episode = 2
running_task.meta.episode_list = [2]
assert manager.add_task(waiting_task)
assert manager.add_task(running_task)
manager.running_task(running_task)
clock[0] = 180.0
manager.running_task(running_task)
clock[0] = 250.0
assert manager.expire_stale_running_tasks(timeout_seconds=120) == []
assert manager.total() == 2