mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-06-01 13:40:54 +08:00
495 lines
18 KiB
Python
495 lines
18 KiB
Python
"""feedback-issue skill 脚本共享逻辑。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import re
|
||
import sys
|
||
import time
|
||
import uuid
|
||
from pathlib import Path
|
||
from typing import Any, Optional
|
||
from urllib.parse import quote
|
||
|
||
|
||
def _find_repo_root() -> Path:
|
||
"""从当前工作目录和脚本路径向上查找 MoviePilot 仓库根目录。"""
|
||
script_path = Path(__file__).resolve()
|
||
candidates = [Path.cwd().resolve(), *Path.cwd().resolve().parents]
|
||
candidates.extend([script_path.parent, *script_path.parents])
|
||
for candidate in candidates:
|
||
if (candidate / "app" / "core" / "config.py").is_file():
|
||
return candidate
|
||
return script_path.parents[3]
|
||
|
||
|
||
REPO_ROOT = _find_repo_root()
|
||
if str(REPO_ROOT) not in sys.path:
|
||
sys.path.insert(0, str(REPO_ROOT))
|
||
|
||
from app.core.config import settings # noqa: E402
|
||
|
||
|
||
FEEDBACK_REPO_OWNER = "jxxghp"
|
||
FEEDBACK_REPO_NAME = "MoviePilot"
|
||
FEEDBACK_REPO = f"{FEEDBACK_REPO_OWNER}/{FEEDBACK_REPO_NAME}"
|
||
FEEDBACK_ISSUE_API = f"https://api.github.com/repos/{FEEDBACK_REPO}/issues"
|
||
FEEDBACK_ISSUE_NEW_URL = f"https://github.com/{FEEDBACK_REPO}/issues/new"
|
||
FEEDBACK_ISSUE_TEMPLATE = "bug_report.yml"
|
||
FEEDBACK_REQUEST_TIMEOUT = 15
|
||
|
||
ALLOWED_ENVIRONMENTS = ("Docker", "Windows")
|
||
ALLOWED_ISSUE_TYPES = ("主程序运行问题", "插件问题", "其他问题")
|
||
|
||
MAX_TITLE_CHARS = 256
|
||
MAX_BODY_CHARS = 60 * 1024
|
||
MAX_LOGS_CHARS = 8 * 1024
|
||
MAX_URL_LOGS_CHARS = 3 * 1024
|
||
MAX_PREVIEW_LOGS_CHARS = 3 * 1024
|
||
|
||
DEDUP_TTL_SECONDS = 60
|
||
USER_COOLDOWN_SECONDS = 30 * 60
|
||
USER_DAILY_QUOTA = 10
|
||
USER_DAILY_WINDOW_SECONDS = 24 * 60 * 60
|
||
MAX_USER_SUBMISSIONS_BUCKETS = 200
|
||
|
||
MIN_TITLE_BODY_CHARS = 8
|
||
MIN_DESCRIPTION_CHARS = 50
|
||
TITLE_PREFIX = "[错误报告]:"
|
||
|
||
_QUALITY_BLOCKLIST = (
|
||
"测试issue", "测试 issue", "test issue",
|
||
"test123", "testtest", "测试测试",
|
||
"测试一下", "测试提交", "测试请求", "测试反馈",
|
||
"看能否跑通", "能否跑通", "跑通流程", "链路测试",
|
||
"模拟问题", "模拟问题描述", "模拟描述", "模拟 bug", "模拟bug",
|
||
"编造", "虚假 bug", "虚假bug",
|
||
"asdf", "asdfasdf", "qwer", "qwerty", "qweqwe",
|
||
"占位", "占个坑", "随便", "随便写",
|
||
"abcabc", "xxxxxx", "xxx xxx",
|
||
"hello world", "你好世界",
|
||
"lorem ipsum", "dolor sit amet",
|
||
)
|
||
|
||
_FABRICATED_LOG_PHRASES = (
|
||
"无相关日志", "没有相关日志", "未捕获到相关日志",
|
||
"这是模拟", "模拟问题", "模拟描述", "用户反馈",
|
||
)
|
||
|
||
_DESCRIPTION_REQUIRED_SIGNALS = (
|
||
("现象", ("现象", "报错", "错误", "无法", "失败", "异常")),
|
||
("复现步骤", ("复现", "步骤", "触发", "操作", "调用", "点击")),
|
||
("期望行为", ("期望", "应该", "预期", "正常")),
|
||
)
|
||
|
||
_REPEAT_GIBBERISH = re.compile(r"([^\s=\-_*#~`./\\+|])\1{7,}", re.UNICODE)
|
||
|
||
_REDACTED = "<REDACTED>"
|
||
_REDACTED_PATH = "/<USER>/"
|
||
_REDACTED_EMAIL = "<EMAIL>"
|
||
_REDACTED_IP = "<IP>"
|
||
|
||
_SENSITIVE_PATTERNS: tuple[tuple[re.Pattern, str], ...] = (
|
||
(re.compile(r"(?i)(Cookie\s*:\s*)[^\r\n]+"), rf"\1{_REDACTED}"),
|
||
(re.compile(r"(?i)(Set-Cookie\s*:\s*)[^\r\n]+"), rf"\1{_REDACTED}"),
|
||
(
|
||
re.compile(r"(?i)(Authorization\s*:\s*)(Bearer|Basic|Token)\s+\S+"),
|
||
rf"\1\2 {_REDACTED}",
|
||
),
|
||
(re.compile(r"(?i)(X-(?:Api-Key|Auth-Token|Access-Token)\s*:\s*)\S+"), rf"\1{_REDACTED}"),
|
||
(re.compile(r"\bghp_[A-Za-z0-9]{20,}\b"), _REDACTED),
|
||
(re.compile(r"\bgho_[A-Za-z0-9]{20,}\b"), _REDACTED),
|
||
(re.compile(r"\bgithub_pat_[A-Za-z0-9_]{20,}\b"), _REDACTED),
|
||
(re.compile(r"\b(sk|xoxb|xoxp|xoxa)-[A-Za-z0-9-]{12,}\b"), _REDACTED),
|
||
(re.compile(r"\buser_\d{4,}_\d+\b"), _REDACTED),
|
||
(re.compile(r"(?i)\b(passkey|rsskey|authkey|access_key)=[A-Za-z0-9]{8,}"), rf"\1={_REDACTED}"),
|
||
(
|
||
re.compile(
|
||
r"https?://(qyapi\.weixin\.qq\.com|oapi\.dingtalk\.com|open\.feishu\.cn|"
|
||
r"hooks\.slack\.com|discord(?:app)?\.com/api/webhooks)/\S+"
|
||
),
|
||
rf"\1/{_REDACTED}",
|
||
),
|
||
(
|
||
re.compile(
|
||
r"(?i)\b("
|
||
r"api[_-]?key|apikey|access[_-]?token|refresh[_-]?token|id[_-]?token|"
|
||
r"client[_-]?secret|client[_-]?id|app[_-]?secret|app[_-]?key|"
|
||
r"corp[_-]?secret|corp[_-]?id|agent[_-]?id|"
|
||
r"password|secret|token|auth|credential|"
|
||
r"chat[_-]?id|webhook|api[_-]?token|bot[_-]?token|"
|
||
r"user[_-]?id|userid|username|user[_-]?name|"
|
||
r"session[_-]?id|sessionid|"
|
||
r"open[_-]?id|openid|union[_-]?id|unionid"
|
||
r")(\s*[:=]\s*)['\"]?[^\s'\"&\r\n]{2,}"
|
||
),
|
||
rf"\1\2{_REDACTED}",
|
||
),
|
||
(
|
||
re.compile(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}"),
|
||
_REDACTED_EMAIL,
|
||
),
|
||
(
|
||
re.compile(
|
||
r"\b(?!(?:127|10)\.)"
|
||
r"(?!172\.(?:1[6-9]|2\d|3[01])\.)"
|
||
r"(?!192\.168\.)"
|
||
r"(?:\d{1,3}\.){3}\d{1,3}\b"
|
||
),
|
||
_REDACTED_IP,
|
||
),
|
||
(re.compile(r"/Users/[^/\s]+/"), _REDACTED_PATH),
|
||
(re.compile(r"/home/[^/\s]+/"), _REDACTED_PATH),
|
||
(re.compile(r"C:\\Users\\[^\\\s]+\\", re.IGNORECASE), r"C:\\Users\\<USER>\\"),
|
||
)
|
||
|
||
|
||
def feedback_runtime_dir() -> Path:
|
||
"""返回 feedback-issue 脚本使用的运行时目录并确保存在。"""
|
||
runtime_dir = settings.TEMP_PATH / "feedback-issue"
|
||
runtime_dir.mkdir(parents=True, exist_ok=True)
|
||
return runtime_dir
|
||
|
||
|
||
def runtime_file(prefix: str, suffix: str) -> Path:
|
||
"""在运行时目录下生成一个随机文件路径。"""
|
||
safe_prefix = re.sub(r"[^a-zA-Z0-9_-]+", "-", prefix).strip("-") or "feedback"
|
||
return feedback_runtime_dir() / f"{safe_prefix}-{uuid.uuid4().hex[:12]}{suffix}"
|
||
|
||
|
||
def ensure_runtime_file(path: str | Path) -> Path:
|
||
"""校验脚本间传递的文件必须位于 feedback-issue 运行时目录内。"""
|
||
candidate = Path(path).expanduser().resolve()
|
||
runtime_dir = feedback_runtime_dir().resolve()
|
||
if not candidate.is_relative_to(runtime_dir):
|
||
raise ValueError(f"只允许读取 feedback-issue 运行时目录内的文件: {candidate}")
|
||
return candidate
|
||
|
||
|
||
def read_json_file(path: str | Path) -> dict[str, Any]:
|
||
"""读取 JSON 文件并确保顶层对象是 dict。"""
|
||
json_path = Path(path).expanduser()
|
||
data = json.loads(json_path.read_text(encoding="utf-8"))
|
||
if not isinstance(data, dict):
|
||
raise ValueError(f"JSON 顶层必须是对象: {json_path}")
|
||
return data
|
||
|
||
|
||
def write_json_file(path: str | Path, payload: dict[str, Any]) -> Path:
|
||
"""把 JSON 对象写入文件并返回实际路径。"""
|
||
json_path = Path(path)
|
||
json_path.parent.mkdir(parents=True, exist_ok=True)
|
||
json_path.write_text(
|
||
json.dumps(payload, ensure_ascii=False, indent=2),
|
||
encoding="utf-8",
|
||
)
|
||
return json_path
|
||
|
||
|
||
def validate_enum(value: str, allowed: tuple[str, ...], field_name: str) -> Optional[str]:
|
||
"""校验枚举字段，返回错误信息；通过时返回 None。"""
|
||
if value not in allowed:
|
||
return (
|
||
f"{field_name} 必须是以下之一：{', '.join(allowed)}；"
|
||
f"当前传入：{value!r}"
|
||
)
|
||
return None
|
||
|
||
|
||
def redact_logs(raw: str) -> str:
|
||
"""对日志文本做统一脱敏，覆盖常见 token、Cookie、PII 和本机路径。"""
|
||
out = raw
|
||
for pattern, replacement in _SENSITIVE_PATTERNS:
|
||
out = pattern.sub(replacement, out)
|
||
return out
|
||
|
||
|
||
def truncate(text: str, limit: int, marker: str = "\n...（已截断）") -> str:
|
||
"""按字符数截断文本并附加截断标记。"""
|
||
if not text or len(text) <= limit:
|
||
return text
|
||
return text[: max(0, limit - len(marker))] + marker
|
||
|
||
|
||
def sanitize_logs(logs: Optional[str], limit: int) -> str:
|
||
"""清洗日志：去空白、脱敏并按指定长度截断。"""
|
||
if not logs or not logs.strip():
|
||
return ""
|
||
return truncate(redact_logs(logs.strip()), limit)
|
||
|
||
|
||
def build_issue_body(
|
||
*,
|
||
version: str,
|
||
environment: str,
|
||
issue_type: str,
|
||
description: str,
|
||
logs: Optional[str],
|
||
) -> str:
|
||
"""构造与上游 bug_report.yml 表单渲染接近的 Issue Markdown 正文。"""
|
||
log_block = sanitize_logs(logs, MAX_LOGS_CHARS) or "会话中未捕获到相关后端日志。"
|
||
body = (
|
||
"### 确认\n\n"
|
||
"- [x] 我的版本是最新版本，我的版本号与 "
|
||
"[version](https://github.com/jxxghp/MoviePilot/releases/latest) 相同。\n"
|
||
"- [x] 我已经 [issue](https://github.com/jxxghp/MoviePilot/issues) "
|
||
"中搜索过，确认我的问题没有被提出过。\n"
|
||
"- [x] 我已经 [Telegram频道](https://t.me/moviepilot_channel) "
|
||
"中搜索过，确认我的问题没有被提出过。\n"
|
||
"- [x] 我已经修改标题，将标题中的 描述 替换为我遇到的问题。\n\n"
|
||
f"### 当前程序版本\n\n{version}\n\n"
|
||
f"### 运行环境\n\n{environment}\n\n"
|
||
f"### 问题类型\n\n{issue_type}\n\n"
|
||
f"### 问题描述\n\n{description.strip()}\n\n"
|
||
"### 发生问题时系统日志和配置文件\n\n"
|
||
f"```bash\n{log_block}\n```\n"
|
||
"\n---\n"
|
||
"_本 Issue 由 MoviePilot Agent 协助用户提交。_"
|
||
)
|
||
return truncate(body, MAX_BODY_CHARS)
|
||
|
||
|
||
def build_prefill_url(
|
||
*,
|
||
title: str,
|
||
version: str,
|
||
environment: str,
|
||
issue_type: str,
|
||
description: str,
|
||
logs: Optional[str],
|
||
) -> str:
|
||
"""生成 GitHub Issue Forms 预填 URL，供无 token 或 API 失败时手动提交。"""
|
||
params = {
|
||
"template": FEEDBACK_ISSUE_TEMPLATE,
|
||
"title": title,
|
||
"version": version,
|
||
"environment": environment,
|
||
"type": issue_type,
|
||
"what-happened": description,
|
||
"logs": sanitize_logs(logs, MAX_URL_LOGS_CHARS),
|
||
}
|
||
encoded = "&".join(
|
||
f"{quote(k, safe='')}={quote(v, safe='')}" for k, v in params.items()
|
||
)
|
||
return f"{FEEDBACK_ISSUE_NEW_URL}?{encoded}"
|
||
|
||
|
||
def classify_failure(status_code: Optional[int], headers: Optional[dict] = None) -> str:
|
||
"""把 GitHub API HTTP 状态码映射成脚本输出的稳定失败原因。"""
|
||
headers = headers or {}
|
||
if status_code == 401:
|
||
return "no_permission"
|
||
if status_code == 403:
|
||
remaining = headers.get("X-RateLimit-Remaining") or headers.get(
|
||
"x-ratelimit-remaining"
|
||
)
|
||
if remaining == "0":
|
||
return "rate_limited"
|
||
return "no_permission"
|
||
if status_code == 404:
|
||
return "no_permission"
|
||
if status_code == 422:
|
||
return "invalid_payload"
|
||
if status_code is not None and status_code >= 500:
|
||
return "github_unavailable"
|
||
return "api_error"
|
||
|
||
|
||
def safe_response_dict(response: Any) -> dict[str, Any]:
|
||
"""安全解析 HTTP 响应 JSON，非 dict 或解析失败时返回空 dict。"""
|
||
try:
|
||
data = response.json()
|
||
except Exception:
|
||
return {}
|
||
if isinstance(data, dict):
|
||
return data
|
||
return {}
|
||
|
||
|
||
def check_content_quality(
|
||
*,
|
||
title: str,
|
||
description: str,
|
||
original_user_request: str,
|
||
logs: Optional[str] = None,
|
||
) -> Optional[str]:
|
||
"""检查 Issue 内容质量，拦截测试、占位、乱码和结构缺失的提交。"""
|
||
original_stripped = (original_user_request or "").strip()
|
||
if not original_stripped:
|
||
return (
|
||
"缺少原始用户请求，无法判断本次提交是否来自真实故障。"
|
||
"请传入触发反馈的用户原话，不能只传改写后的 Issue 草稿。"
|
||
)
|
||
|
||
title_body = title.strip()
|
||
if title_body.startswith(TITLE_PREFIX):
|
||
title_body = title_body[len(TITLE_PREFIX):].strip()
|
||
if len(title_body) < MIN_TITLE_BODY_CHARS:
|
||
return (
|
||
f"标题正文太短（剔除 {TITLE_PREFIX!r} 前缀后只有 {len(title_body)} 字，"
|
||
f"至少 {MIN_TITLE_BODY_CHARS} 字）。请用一句完整的话概括症状。"
|
||
)
|
||
|
||
desc_stripped = description.strip()
|
||
if len(desc_stripped) < MIN_DESCRIPTION_CHARS:
|
||
return (
|
||
f"问题描述太短（{len(desc_stripped)} 字，至少 {MIN_DESCRIPTION_CHARS} 字）。"
|
||
"请补充：现象 / 复现步骤 / 期望行为。"
|
||
)
|
||
|
||
missing_signals = [
|
||
label
|
||
for label, choices in _DESCRIPTION_REQUIRED_SIGNALS
|
||
if not any(choice in desc_stripped for choice in choices)
|
||
]
|
||
if missing_signals:
|
||
return (
|
||
"问题描述缺少可复现 bug 所需的结构信息："
|
||
f"{' / '.join(missing_signals)}。请补充真实现象、触发步骤和期望行为。"
|
||
)
|
||
|
||
haystack = "\n".join(
|
||
part for part in (title, description, original_stripped) if part
|
||
).lower()
|
||
for phrase in _QUALITY_BLOCKLIST:
|
||
if phrase.lower() in haystack:
|
||
return (
|
||
f"原始请求、标题或描述命中明显占位/测试关键词「{phrase}」，"
|
||
"已拒绝提交。如果是真实问题，请用正常的中文描述具体现象。"
|
||
)
|
||
|
||
match = (
|
||
_REPEAT_GIBBERISH.search(title)
|
||
or _REPEAT_GIBBERISH.search(description)
|
||
or _REPEAT_GIBBERISH.search(original_stripped)
|
||
)
|
||
if match:
|
||
return (
|
||
f"标题或描述里出现疑似乱码片段「{match.group(0)[:12]}...」，"
|
||
"请用正常文字描述问题。"
|
||
)
|
||
|
||
log_text = (logs or "").strip()
|
||
if log_text:
|
||
lowered_logs = log_text.lower()
|
||
for phrase in _FABRICATED_LOG_PHRASES:
|
||
if phrase.lower() in lowered_logs and len(log_text) < 200:
|
||
return (
|
||
f"日志字段疑似填入了叙述性占位内容「{phrase}」，"
|
||
"请只提交真实日志；没有日志时留空。"
|
||
)
|
||
return None
|
||
|
||
|
||
def normalize_username(username: Optional[str]) -> str:
|
||
"""归一化用户名，作为脚本级提交频率限制的桶 key。"""
|
||
return (username or "").strip().lower()
|
||
|
||
|
||
def load_submission_state() -> dict[str, Any]:
|
||
"""读取脚本持久化的短期提交状态。"""
|
||
state_file = feedback_runtime_dir() / "submission-state.json"
|
||
if not state_file.exists():
|
||
return {"recent_submissions": {}, "user_submissions": {}}
|
||
try:
|
||
state = read_json_file(state_file)
|
||
except Exception:
|
||
return {"recent_submissions": {}, "user_submissions": {}}
|
||
state.setdefault("recent_submissions", {})
|
||
state.setdefault("user_submissions", {})
|
||
return state
|
||
|
||
|
||
def save_submission_state(state: dict[str, Any]) -> None:
|
||
"""写回脚本持久化的短期提交状态。"""
|
||
write_json_file(feedback_runtime_dir() / "submission-state.json", state)
|
||
|
||
|
||
def check_recent_duplicate(title: str, body: str, state: dict[str, Any]) -> Optional[str]:
|
||
"""检查 60 秒内是否提交过同 title + body 的内容。"""
|
||
now = time.time()
|
||
recent = state.setdefault("recent_submissions", {})
|
||
for key, ts in list(recent.items()):
|
||
if now - float(ts or 0) > DEDUP_TTL_SECONDS:
|
||
recent.pop(key, None)
|
||
key = hashlib.sha256(f"{title}\x00{body}".encode("utf-8", errors="replace")).hexdigest()
|
||
if key in recent:
|
||
return key
|
||
return None
|
||
|
||
|
||
def record_submission(title: str, body: str, state: dict[str, Any]) -> None:
|
||
"""记录一次提交内容摘要，供短时间去重使用。"""
|
||
key = hashlib.sha256(f"{title}\x00{body}".encode("utf-8", errors="replace")).hexdigest()
|
||
state.setdefault("recent_submissions", {})[key] = time.time()
|
||
|
||
|
||
def evict_user_submissions_if_needed(state: dict[str, Any]) -> None:
|
||
"""限制用户提交状态桶数量，避免运行时文件无限增长。"""
|
||
buckets = state.setdefault("user_submissions", {})
|
||
if len(buckets) <= MAX_USER_SUBMISSIONS_BUCKETS:
|
||
return
|
||
excess = len(buckets) - MAX_USER_SUBMISSIONS_BUCKETS
|
||
oldest_keys = sorted(
|
||
buckets.items(),
|
||
key=lambda kv: kv[1][-1] if kv[1] else 0,
|
||
)[:excess]
|
||
for key, _ in oldest_keys:
|
||
buckets.pop(key, None)
|
||
|
||
|
||
def check_user_rate_limit(username: str, state: dict[str, Any]) -> Optional[str]:
|
||
"""检查单用户 30 分钟冷却和 24 小时提交配额。"""
|
||
key = normalize_username(username)
|
||
if not key:
|
||
return "无法识别调用用户身份，rate limit 拒绝以防误用。"
|
||
now = time.time()
|
||
buckets = state.setdefault("user_submissions", {})
|
||
timestamps = buckets.get(key, [])
|
||
active = [float(ts) for ts in timestamps if now - float(ts or 0) < USER_DAILY_WINDOW_SECONDS]
|
||
if active:
|
||
buckets[key] = active
|
||
else:
|
||
buckets.pop(key, None)
|
||
if active:
|
||
since_last = now - active[-1]
|
||
if since_last < USER_COOLDOWN_SECONDS:
|
||
remaining = int(USER_COOLDOWN_SECONDS - since_last)
|
||
minutes, seconds = divmod(remaining, 60)
|
||
return (
|
||
f"为避免给上游刷屏，同一管理员两次提交之间至少间隔 "
|
||
f"{USER_COOLDOWN_SECONDS // 60} 分钟。请等 {minutes} 分 {seconds} 秒后再试。"
|
||
)
|
||
if len(active) >= USER_DAILY_QUOTA:
|
||
recover_in = int(USER_DAILY_WINDOW_SECONDS - (now - active[0]))
|
||
hours, remainder = divmod(recover_in, 3600)
|
||
minutes = remainder // 60
|
||
return (
|
||
f"你今日已提交 {USER_DAILY_QUOTA} 个 Issue，已达 24 小时配额上限。"
|
||
f"最早一条将在 {hours} 小时 {minutes} 分钟后过期，请到时再提。"
|
||
)
|
||
return None
|
||
|
||
|
||
def record_user_submission(username: str, state: dict[str, Any]) -> None:
|
||
"""记录一次用户级提交时间戳，供冷却和配额检查使用。"""
|
||
key = normalize_username(username)
|
||
if not key:
|
||
return
|
||
state.setdefault("user_submissions", {}).setdefault(key, []).append(time.time())
|
||
evict_user_submissions_if_needed(state)
|
||
|
||
|
||
def load_diagnostics_logs(diagnostics_file: str | Path) -> tuple[str, dict[str, Any]]:
|
||
"""读取诊断文件中的日志正文并返回日志与诊断元数据。"""
|
||
path = ensure_runtime_file(diagnostics_file)
|
||
data = read_json_file(path)
|
||
logs = sanitize_logs(str(data.get("logs") or ""), MAX_LOGS_CHARS)
|
||
return logs, data
|
||
|
||
|
||
def result_payload(**fields: Any) -> str:
|
||
"""把脚本结果格式化为 Agent 容易解析的 JSON 字符串。"""
|
||
return json.dumps(fields, ensure_ascii=False, indent=2)
|