diff --git a/README.md b/README.md index 82aa0f3..39fd860 100644 --- a/README.md +++ b/README.md @@ -105,7 +105,7 @@ From a folder containing a movie, run: fixsub ``` -The default provider set is ASSRT plus SubHD. A final subtitle is written only when the selected candidate is not poor; otherwise inspect the saved candidates and diagnostics. +The default provider set is ASSRT plus SubHD. When at least one usable Chinese candidate exists, `fixsub` writes the highest-ranked candidate and records its confidence and selection outcome in the console and metadata. ### Safe preview @@ -122,7 +122,7 @@ fixsub --providers subhd fixsub --providers assrt,subhd ``` -Use `fixsub --providers subhd` to avoid ASSRT credentials. Use `fixsub --providers assrt,subhd` to explicitly select both providers (also the default). Provider search or download failures are logged and the run can continue with other available providers, but provider outages can leave no usable candidate. +Use `fixsub --providers subhd` to avoid ASSRT credentials. Use `fixsub --providers assrt,subhd` to explicitly select both providers (also the default). SubHD downloads use its prepared-download flow before fetching the final file. Provider search or download failures are logged and the run can continue with other available providers, but provider outages can leave no usable candidate. ### Audio and synchronization @@ -131,7 +131,9 @@ fixsub --audio a:0 fixsub --no-sync ``` -By default, `fixsub` probes audio with `ffprobe`, selects a reference stream, and tries `ffsubsync` for each candidate. `fixsub --audio a:0` forces the stream passed to `ffsubsync`. `fixsub --no-sync` skips audio synchronization and ranks original candidates only; use it only when necessary because structural timestamp checks cannot establish dialogue or audio alignment. A failed or low-quality synchronization makes a candidate poor and prevents automatic application. +By default, `fixsub` probes audio with `ffprobe`, selects a reference stream, and tries `ffsubsync` for each candidate. `fixsub --audio a:0` forces the stream passed to `ffsubsync`. `fixsub --no-sync` skips audio synchronization and ranks original candidates only; use it only when necessary because structural timestamp checks cannot establish dialogue or audio alignment. + +`fixsub` first asks `ffsubsync` to skip low-quality alignments. If that conservative pass explicitly refuses the alignment, `fixsub` retries once in forced mode. When at least one usable Chinese candidate exists, the best candidate is written even if confidence remains low; the console and metadata identify forced synchronization or original fallback. A hard stop remains when no candidate can be downloaded, extracted, parsed, or accepted as Chinese. ### Candidate and language controls @@ -188,8 +190,8 @@ Before opening an issue, remove `.fixsub/logs/fixsub.log`, `.fixsub/metadata/res - **A `.rar` or `.7z` archive cannot be extracted:** install archive support with `brew install unar` and retry. `.zip` files do not need `unar`. - **ASSRT reports missing credentials:** run `fixsub auth set`, check `fixsub auth status`, or set a temporary `ASSRT_TOKEN`. Use `fixsub --providers subhd` if you do not intend to use ASSRT. - **A provider is unavailable or no candidate is found:** retry later, select another provider, and inspect `.fixsub/logs/fixsub.log`; provider results and downloads can change independently of the tool. -- **Candidates are rejected as low confidence:** inspect `.fixsub/candidates/` and `.fixsub/metadata/results.json`. Low structural timing scores or failed synchronization intentionally prevent automatic output. -- **Synchronization fails:** install `ffsubsync`, confirm `ffprobe` can read the video, choose the correct stream with `fixsub --audio a:0`, and inspect the log. `fixsub --no-sync` is available only as a manual-risk fallback. +- **A low-confidence subtitle is applied:** inspect `.fixsub/candidates/`, `.fixsub/synced/`, and `.fixsub/metadata/results.json`. The console and metadata identify whether it used forced synchronization or the original fallback. +- **Synchronization fails:** install `ffsubsync`, confirm `ffprobe` can read the video, choose the correct stream with `fixsub --audio a:0`, and inspect the log. `fixsub` ranks and applies the original fallback when it remains the best usable candidate. ## Development diff --git a/README.zh-CN.md b/README.zh-CN.md index 016454a..7db2766 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -105,7 +105,7 @@ macOS 钥匙串是首选的持久化存储方式。临时的 `ASSRT_TOKEN` 环 fixsub ``` -默认字幕源是 ASSRT 加 SubHD。仅当选中的候选项不属于低质量结果时才会写入最终字幕;否则请检查已保存的候选项和诊断信息。 +默认字幕源是 ASSRT 加 SubHD。只要至少存在一个可用的中文字幕候选项,`fixsub` 就会写入排名最高的候选项,并在控制台和元数据中记录其置信度和选择结果。 ### 安全预览 @@ -122,7 +122,7 @@ fixsub --providers subhd fixsub --providers assrt,subhd ``` -使用 `fixsub --providers subhd` 可避免使用 ASSRT 凭据。使用 `fixsub --providers assrt,subhd` 可显式选择两个字幕源(也是默认值)。字幕源的搜索或下载失败会被记录,运行可继续使用其他可用字幕源;但字幕源故障可能导致没有可用候选项。 +使用 `fixsub --providers subhd` 可避免使用 ASSRT 凭据。使用 `fixsub --providers assrt,subhd` 可显式选择两个字幕源(也是默认值)。SubHD 下载会先走其准备下载流程,再获取最终文件。字幕源的搜索或下载失败会被记录,运行可继续使用其他可用字幕源;但字幕源故障可能导致没有可用候选项。 ### 音轨与同步 @@ -131,7 +131,9 @@ fixsub --audio a:0 fixsub --no-sync ``` -默认情况下,`fixsub` 会通过 `ffprobe` 探测音轨、选择参考流,并对每个候选项尝试 `ffsubsync`。`fixsub --audio a:0` 会强制传给 `ffsubsync` 的流。`fixsub --no-sync` 会跳过音频同步,仅对原始候选项排序;仅应在必要时使用,因为结构化时间戳检查无法证明对白或音频对齐。同步失败或低质量同步会使候选项被视为低质量,因而不会自动应用。 +默认情况下,`fixsub` 会通过 `ffprobe` 探测音轨、选择参考流,并对每个候选项尝试 `ffsubsync`。`fixsub --audio a:0` 会强制传给 `ffsubsync` 的流。`fixsub --no-sync` 会跳过音频同步,仅对原始候选项排序;仅应在必要时使用,因为结构化时间戳检查无法证明对白或音频对齐。 + +`fixsub` 会先要求 `ffsubsync` 跳过低质量对齐;如果保守同步明确拒绝该对齐,则自动以强制模式重试一次。只要至少存在一个可用的中文字幕候选项,即使置信度仍然较低,也会写入排名最高的候选项;控制台和元数据会明确标记强制同步或原字幕回退。只有在没有候选项能够下载、解压、解析或通过中文内容检查时才会停止且不写入字幕。 ### 候选数量与语言标签 @@ -188,8 +190,8 @@ fixsub --debug - **无法解压 `.rar` 或 `.7z` 压缩包:** 使用 `brew install unar` 安装解压支持后重试。`.zip` 文件不需要 `unar`。 - **ASSRT 提示缺少凭据:** 运行 `fixsub auth set`,检查 `fixsub auth status`,或设置临时 `ASSRT_TOKEN`。若不打算使用 ASSRT,请使用 `fixsub --providers subhd`。 - **字幕源不可用或找不到候选项:** 请稍后重试、选择其他字幕源,并检查 `.fixsub/logs/fixsub.log`;字幕源结果和下载状态会独立于工具发生变化。 -- **候选项因低置信度被拒绝:** 检查 `.fixsub/candidates/` 和 `.fixsub/metadata/results.json`。低结构化时间轴评分或同步失败会有意阻止自动输出。 -- **同步失败:** 安装 `ffsubsync`,确认 `ffprobe` 能读取视频,使用 `fixsub --audio a:0` 选择正确的流,并检查日志。`fixsub --no-sync` 仅作为需要人工承担风险的后备选项。 +- **已应用低置信度字幕:** 检查 `.fixsub/candidates/`、`.fixsub/synced/` 和 `.fixsub/metadata/results.json`。控制台和元数据会标记它使用的是强制同步还是原字幕回退。 +- **同步失败:** 安装 `ffsubsync`,确认 `ffprobe` 能读取视频,使用 `fixsub --audio a:0` 选择正确的流,并检查日志。若原字幕仍是排名最高的可用候选项,`fixsub` 会将其作为回退写入。 ## 开发 diff --git a/docs/superpowers/plans/2026-07-22-subhd-download-and-low-confidence-output.md b/docs/superpowers/plans/2026-07-22-subhd-download-and-low-confidence-output.md new file mode 100644 index 0000000..c006076 --- /dev/null +++ b/docs/superpowers/plans/2026-07-22-subhd-download-and-low-confidence-output.md @@ -0,0 +1,698 @@ +# SubHD Download and Low-Confidence Output Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Restore SubHD downloads through its prepared-page protocol and always deliver the best usable Chinese subtitle, using a forced second `ffsubsync` pass when conservative synchronization explicitly refuses a low-quality alignment. + +**Architecture:** Keep provider protocol handling in `fixsub/providers/subhd.py`, synchronization policy in `fixsub/sync.py`, confidence metadata in `fixsub/models.py` and `fixsub/decision.py`, and final-output policy in `fixsub/cli.py`. Each task follows red-green-refactor and commits an independently testable behavior before the next task begins. + +**Tech Stack:** Python 3.11+, httpx, BeautifulSoup, Typer, pytest, ffsubsync CLI integration. + +## Global Constraints + +- Do not add a runtime dependency. +- A normal run writes a final subtitle whenever at least one downloaded, extracted, normalized, parseable Chinese candidate exists. +- `--dry-run` never writes a final subtitle. +- Conservative `ffsubsync` runs first with `--skip-sync-on-low-quality`; only its explicit low-quality diagnostic triggers one retry without that flag. +- A forced-pass failure falls back to the best original Chinese subtitle. +- Low confidence remains visible in console output and metadata; it is never presented as high confidence. +- High-confidence decisions remain ranked ahead of low-confidence decisions. +- Existing downloads, normalized candidates, synchronized files, logs, metadata, URL validation, and final-subtitle backup behavior remain intact. +- SubHD preparation URLs must use HTTPS, belong to an allowed SubHD domain or configured test host, have the exact path `/down/{sid}`, and contain no credentials, non-default port, query, or fragment. + +--- + +### Task 1: Support SubHD Prepared Downloads + +**Files:** +- Modify: `fixsub/providers/subhd.py:260-303` +- Test: `tests/test_subhd.py:303-680` + +**Interfaces:** +- Consumes: `SubhdClient._request_stage(method, url, error_message, **kwargs) -> httpx.Response`, `_is_allowed_subhd_url(url, base_url) -> bool`, and `SearchResult.result_id`. +- Produces: `SubhdClient._prepare_download_url(result: SearchResult, detail_url: str) -> str`; `SubhdClient.download()` uses the returned prepared page URL before the existing download API. + +- [ ] **Step 1: Write failing preparation and request-order tests** + +Add a focused integration test whose transport handles the current five-stage request sequence: + +```python +def test_subhd_client_prepares_download_page_before_opening_gate(tmp_path: Path) -> None: + requests: list[tuple[str, str, bytes, str | None]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append((request.method, str(request.url), request.content, request.headers.get("Referer"))) + if request.url.path == "/a/kAqdvK": + return httpx.Response(200, text="detail", request=request) + if request.url.path == "/api/sub/prepare-download": + return httpx.Response(200, json={"success": True, "url": "/down/kAqdvK"}, request=request) + if request.url.path == "/down/kAqdvK": + return httpx.Response(200, text="gate", request=request) + if request.url.path == "/api/sub/down": + return httpx.Response( + 200, + json={"success": True, "pass": True, "url": "https://dl.subhd.me/result.zip"}, + request=request, + ) + if str(request.url) == "https://dl.subhd.me/result.zip": + return httpx.Response(200, content=b"PK\x03\x04archive", request=request) + raise AssertionError(f"Unexpected request: {request.method} {request.url}") + + client = SubhdClient(http_client=httpx.Client(transport=httpx.MockTransport(handler))) + result = parse_search_response(SEARCH_HTML, "https://subhd.tv/search/Nell%201994")[0] + + downloaded = client.download(result, tmp_path) + + assert downloaded.path.read_bytes() == b"PK\x03\x04archive" + assert [(method, url) for method, url, _body, _referer in requests] == [ + ("GET", "https://subhd.tv/a/kAqdvK"), + ("POST", "https://subhd.tv/api/sub/prepare-download"), + ("GET", "https://subhd.tv/down/kAqdvK"), + ("POST", "https://subhd.tv/api/sub/down"), + ("GET", "https://dl.subhd.me/result.zip"), + ] + assert requests[1][2] == b'{"sid":"kAqdvK"}' + assert requests[1][3] == "https://subhd.tv/a/kAqdvK" + assert requests[2][3] == "https://subhd.tv/a/kAqdvK" +``` + +Add validation coverage for preparation payloads and URLs: + +```python +@pytest.mark.parametrize( + ("payload", "message"), + [ + (None, "invalid JSON"), + ([], "invalid JSON"), + ({"success": False, "msg": "准备失败"}, "准备失败"), + ({"success": True}, "omitted a prepared download URL"), + ({"success": True, "url": "https://evil.example/down/kAqdvK"}, "prepared download URL is not allowed"), + ({"success": True, "url": "/down/other"}, "prepared download URL does not match subtitle"), + ({"success": True, "url": "/down/kAqdvK?token=x"}, "prepared download URL does not match subtitle"), + ], +) +def test_subhd_prepare_download_rejects_invalid_responses(payload: object, message: str) -> None: + def handler(request: httpx.Request) -> httpx.Response: + if payload is None: + return httpx.Response(200, text="not json", request=request) + return httpx.Response(200, json=payload, request=request) + + client = SubhdClient(http_client=httpx.Client(transport=httpx.MockTransport(handler))) + result = parse_search_response(SEARCH_HTML, "https://subhd.tv/search/Nell%201994")[0] + + with pytest.raises(FixsubError, match=message): + client._prepare_download_url(result, "https://subhd.tv/a/kAqdvK") +``` + +Extend the existing stage-specific network test with: + +```python +("prepare", "SubHD download preparation request failed") +``` + +Every existing `client.download(...)` MockTransport handler must respond to preparation before the gate. Insert this exact branch after its detail-page branch, using that test's configured base host: + +```python +if request.url.path == "/api/sub/prepare-download": + return httpx.Response( + 200, + json={"success": True, "url": "/down/kAqdvK"}, + request=request, + ) +``` + +- [ ] **Step 2: Run the focused tests and verify RED** + +Run: + +```bash +.venv/bin/python -m pytest \ + tests/test_subhd.py::test_subhd_client_prepares_download_page_before_opening_gate \ + tests/test_subhd.py::test_subhd_prepare_download_rejects_invalid_responses \ + -v +``` + +Expected: FAIL because `SubhdClient` has no `_prepare_download_url`, and the integration sequence still goes directly from detail to gate. + +- [ ] **Step 3: Implement preparation parsing and URL validation** + +Add this method beside `_api_download_url`: + +```python +def _prepare_download_url(self, result: SearchResult, detail_url: str) -> str: + prepare_url = f"{self.base_url}/api/sub/prepare-download" + response = self._request_stage( + "POST", + prepare_url, + "SubHD download preparation request failed", + json={"sid": result.result_id}, + headers={"Referer": detail_url}, + ) + try: + payload = response.json() + except ValueError as exc: + raise FixsubError("SubHD download preparation returned invalid JSON") from exc + if not isinstance(payload, dict): + raise FixsubError("SubHD download preparation returned invalid JSON") + if payload.get("success") is not True: + message = _sanitize_rejection_message(payload.get("msg")) + raise FixsubError(f"SubHD download preparation rejected the request: {message}") + prepared_url = payload.get("url") + if not isinstance(prepared_url, str) or not prepared_url.strip(): + raise FixsubError("SubHD download preparation omitted a prepared download URL") + resolved_url = urljoin(prepare_url, prepared_url.strip()) + if not _is_allowed_subhd_url(resolved_url, self.base_url): + raise FixsubError(f"SubHD prepared download URL is not allowed: {resolved_url}") + parsed = urlparse(resolved_url) + expected_path = f"/down/{quote(result.result_id, safe='')}" + if parsed.path != expected_path or parsed.query or parsed.fragment: + raise FixsubError(f"SubHD prepared download URL does not match subtitle: {resolved_url}") + return resolved_url +``` + +Change `download()` so the detail request is followed by preparation and then the prepared gate: + +```python +def download(self, result: SearchResult, target_dir) -> DownloadedFile: + detail_url = result.detail_url or f"{self.base_url}/a/{result.result_id}" + self._request_stage("GET", detail_url, "SubHD detail request failed") + + gate_url = self._prepare_download_url(result, detail_url) + gate_response = self._request_stage( + "GET", + gate_url, + "SubHD download page request failed", + headers={"Referer": detail_url}, + ) + gate_content_type = gate_response.headers.get("Content-Type", "").lower() + if not _looks_like_html(gate_response.content, gate_content_type): + return self._save_download(gate_response, result, target_dir) + + download_url = self._api_download_url(result, gate_url) + download_response = self._request_stage("GET", download_url, "SubHD subtitle file request failed") + return self._save_download(download_response, result, target_dir) +``` + +- [ ] **Step 4: Run all SubHD tests and verify GREEN** + +Run: + +```bash +.venv/bin/python -m pytest tests/test_subhd.py -v +``` + +Expected: all SubHD tests PASS, including preparation-stage errors, URL rejection, legacy direct-archive response after a prepared gate, redirects, and final content validation. + +- [ ] **Step 5: Commit Task 1** + +```bash +git add fixsub/providers/subhd.py tests/test_subhd.py +git commit -m "fix: prepare current SubHD downloads" +``` + +--- + +### Task 2: Retry Explicit Low-Quality Synchronization + +**Files:** +- Modify: `fixsub/models.py:104-112` +- Modify: `fixsub/sync.py:22-80` +- Modify: `fixsub/decision.py:10-52` +- Test: `tests/test_sync.py` +- Test: `tests/test_decision_ranking_output.py:21-160` +- Test: `tests/test_paths_models.py:95-126` + +**Interfaces:** +- Consumes: `run_ffsubsync(video_path, subtitle_path, output_path, audio_stream) -> SyncResult` and the existing ffsubsync diagnostic strings. +- Produces: `SyncResult.forced_low_quality: bool = False`; `run_ffsubsync` performs at most two subprocess calls; `decide_candidate_version` preserves forced synchronization as selected but marks it low-confidence. + +- [ ] **Step 1: Write failing forced-retry tests** + +Replace the existing test that expects a low-quality result to fail with a two-call success test: + +```python +def test_run_ffsubsync_retries_low_quality_without_skip_flag(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + monkeypatch.setattr("fixsub.sync.shutil.which", lambda command: "/usr/bin/ffs") + output = tmp_path / "synced" / "candidate.srt" + calls: list[list[str]] = [] + + def fake_run(command: list[str], capture_output: bool, text: bool) -> subprocess.CompletedProcess[str]: + calls.append(command) + output.write_text("forced subtitle", encoding="utf-8") + diagnostics = "score: 12.0\noffset seconds: 31.730\nframerate scale factor: 1.043\n" + if "--skip-sync-on-low-quality" in command: + diagnostics += "low-quality alignment; leaving subtitles unmodified\n" + return subprocess.CompletedProcess(command, 0, stdout=diagnostics, stderr="") + + monkeypatch.setattr("fixsub.sync.subprocess.run", fake_run) + + result = run_ffsubsync( + video_path=tmp_path / "movie.mkv", + subtitle_path=tmp_path / "candidate.srt", + output_path=output, + audio_stream="a:4", + ) + + assert len(calls) == 2 + assert "--skip-sync-on-low-quality" in calls[0] + assert "--skip-sync-on-low-quality" not in calls[1] + assert result.succeeded is True + assert result.output_path == output + assert result.forced_low_quality is True + assert result.offset_seconds == 31.73 +``` + +Add a forced-pass failure test: + +```python +def test_run_ffsubsync_forced_retry_failure_removes_output(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + monkeypatch.setattr("fixsub.sync.shutil.which", lambda command: "/usr/bin/ffs") + output = tmp_path / "synced" / "candidate.srt" + calls = 0 + + def fake_run(command: list[str], capture_output: bool, text: bool) -> subprocess.CompletedProcess[str]: + nonlocal calls + calls += 1 + output.write_text("temporary", encoding="utf-8") + if calls == 1: + return subprocess.CompletedProcess( + command, + 0, + stdout=( + "score: 12.0\noffset seconds: 3.0\nframerate scale factor: 1.0\n" + "low-quality alignment; leaving subtitles unmodified\n" + ), + stderr="", + ) + return subprocess.CompletedProcess(command, 2, stdout="", stderr="forced failure") + + monkeypatch.setattr("fixsub.sync.subprocess.run", fake_run) + + result = run_ffsubsync(tmp_path / "movie.mkv", tmp_path / "candidate.srt", output, "a:0") + + assert calls == 2 + assert result.succeeded is False + assert result.error == "forced failure" + assert result.forced_low_quality is True + assert not output.exists() +``` + +Add decision coverage: + +```python +def test_decision_selects_forced_sync_and_marks_it_low_confidence(tmp_path: Path) -> None: + candidate = make_candidate(tmp_path) + synced = tmp_path / "candidate.synced.ass" + synced.write_text("[Events]\n", encoding="utf-8") + + decision = decide_candidate_version( + candidate=candidate, + original_score=AlignmentScore(0.90, []), + sync_result=SyncResult( + attempted=True, + succeeded=True, + output_path=synced, + forced_low_quality=True, + ), + synced_score=AlignmentScore(0.90, []), + ) + + assert decision.selected_version == "synced" + assert decision.selected_path == synced + assert decision.is_poor is True + assert decision.decision_reason == "ffsubsync forced a low-quality audio alignment." +``` + +- [ ] **Step 2: Run focused tests and verify RED** + +Run: + +```bash +.venv/bin/python -m pytest \ + tests/test_sync.py::test_run_ffsubsync_retries_low_quality_without_skip_flag \ + tests/test_sync.py::test_run_ffsubsync_forced_retry_failure_removes_output \ + tests/test_decision_ranking_output.py::test_decision_selects_forced_sync_and_marks_it_low_confidence \ + -v +``` + +Expected: FAIL because `SyncResult` has no `forced_low_quality` field and `run_ffsubsync` does not retry. + +- [ ] **Step 3: Extend synchronization metadata without breaking positional callers** + +Append the field at the end of `SyncResult` so existing positional construction retains its meaning: + +```python +@dataclass(frozen=True) +class SyncResult: + attempted: bool + succeeded: bool + output_path: Path | None = None + error: str | None = None + ffsubsync_score: float | None = None + offset_seconds: float | None = None + framerate_scale: float | None = None + forced_low_quality: bool = False +``` + +- [ ] **Step 4: Implement a narrowly triggered two-pass subprocess loop** + +Build the conservative command once, run it, and only remove the skip flag after the existing low-quality diagnostic: + +```python +LOW_QUALITY_MARKERS = ("low-quality alignment", "leaving subtitles unmodified") + + +def _is_low_quality(diagnostics: str) -> bool: + lowered = diagnostics.lower() + return any(marker in lowered for marker in LOW_QUALITY_MARKERS) +``` + +Within `run_ffsubsync`, use this control flow while preserving the existing nonzero-exit, missing-output, metric, and `OSError` result shapes: + +```python +conservative_command = [ + "ffs", + str(video_path), + "--reference-stream", + audio_stream, + "--skip-sync-on-low-quality", + "-i", + str(subtitle_path), + "-o", + str(output_path), +] +commands = [conservative_command] +forced_low_quality = False + +for command in commands: + output_path.unlink(missing_ok=True) + try: + result = subprocess.run(command, capture_output=True, text=True) + except OSError as exc: + return SyncResult( + attempted=True, + succeeded=False, + output_path=None, + error=str(exc), + forced_low_quality=forced_low_quality, + ) + diagnostics = "\n".join(part for part in [result.stdout, result.stderr] if part) + metrics = { + "ffsubsync_score": _metric(diagnostics, "score"), + "offset_seconds": _metric(diagnostics, "offset seconds"), + "framerate_scale": _metric(diagnostics, "framerate scale factor"), + } + if not forced_low_quality and result.returncode == 0 and _is_low_quality(diagnostics): + output_path.unlink(missing_ok=True) + forced_low_quality = True + commands.append([part for part in conservative_command if part != "--skip-sync-on-low-quality"]) + continue + if result.returncode != 0: + output_path.unlink(missing_ok=True) + return SyncResult( + attempted=True, + succeeded=False, + output_path=None, + error=result.stderr.strip() or result.stdout.strip(), + forced_low_quality=forced_low_quality, + **metrics, + ) + if not output_path.exists(): + return SyncResult( + attempted=True, + succeeded=False, + output_path=None, + error="ffsubsync exited successfully without writing an output file", + forced_low_quality=forced_low_quality, + **metrics, + ) + if any(value is None for value in metrics.values()): + output_path.unlink(missing_ok=True) + return SyncResult( + attempted=True, + succeeded=False, + output_path=None, + error="ffsubsync output did not include complete alignment metrics", + forced_low_quality=forced_low_quality, + **metrics, + ) + return SyncResult( + attempted=True, + succeeded=True, + output_path=output_path, + error=None, + forced_low_quality=forced_low_quality, + **metrics, + ) +``` + +- [ ] **Step 5: Preserve forced synchronization as selected but low-confidence** + +In `decide_candidate_version`, use a distinct reason and include the forced flag in confidence classification: + +```python +if synced_is_usable: + selected_version = "synced" + selected_path = sync_result.output_path + selected_score = synced_score.score + if sync_result.forced_low_quality: + reason = "ffsubsync forced a low-quality audio alignment." + else: + reason = "ffsubsync audio alignment succeeded." +``` + +Replace the final confidence expression with: + +```python +is_poor = ( + sync_result.forced_low_quality + or usable_score < POOR_ALIGNMENT + or (sync_result.attempted and not synced_is_usable) +) +``` + +- [ ] **Step 6: Run synchronization, decision, and model tests and verify GREEN** + +Run: + +```bash +.venv/bin/python -m pytest \ + tests/test_sync.py \ + tests/test_decision_ranking_output.py \ + tests/test_paths_models.py \ + -v +``` + +Expected: all selected tests PASS. Normal successful synchronization still performs one subprocess call and has `forced_low_quality=False`; only explicit low-quality diagnostics perform two calls. + +- [ ] **Step 7: Commit Task 2** + +```bash +git add fixsub/models.py fixsub/sync.py fixsub/decision.py tests/test_sync.py tests/test_decision_ranking_output.py tests/test_paths_models.py +git commit -m "fix: force low-confidence subtitle synchronization" +``` + +--- + +### Task 3: Deliver the Best Usable Candidate and Document Warnings + +**Files:** +- Modify: `fixsub/cli.py:236-270` +- Test: `tests/test_cli_pipeline.py` +- Modify: `README.md:70-105,135-165` +- Modify: `README.zh-CN.md:70-105,135-165` + +**Interfaces:** +- Consumes: ranked `CandidateDecision` values with `is_poor`, `selected_version`, `selected_score`, and `sync_result.forced_low_quality` from Task 2. +- Produces: low-confidence normal runs write `final_output`; low-confidence dry runs do not; console and metadata clearly report forced synchronization or original fallback. + +- [ ] **Step 1: Write failing low-confidence output tests** + +Add a pipeline test following the existing fake-client setup that forces a poor synchronized decision and asserts final output is written: + +```python +def test_cli_applies_forced_low_confidence_synced_candidate(tmp_path: Path, monkeypatch) -> None: + video = _write_video(tmp_path) + subtitle = tmp_path / ".fixsub" / "candidates" / "subhd_forced.srt" + subtitle.parent.mkdir(parents=True) + subtitle.write_text("1\n00:00:01,000 --> 00:00:02,000\n字幕\n", encoding="utf-8") + synced = tmp_path / ".fixsub" / "synced" / "subhd_forced.synced.srt" + monkeypatch.chdir(tmp_path) + monkeypatch.delenv("ASSRT_TOKEN", raising=False) + + class FakeSubhdClient: + def search(self, query: str) -> list[SearchResult]: + return [SearchResult(provider="subhd", result_id="forced", title="Movie 1992 中文字幕", language="zh-Hans", format="srt")] + + def download(self, result: SearchResult, target_dir: Path) -> DownloadedFile: + return DownloadedFile("subhd_forced", "subhd", subtitle, result.download_url) + + monkeypatch.setattr("fixsub.providers.registry.SubhdClient", FakeSubhdClient) + monkeypatch.setattr("fixsub.cli.extract_archive", lambda path, out_dir: [subtitle]) + monkeypatch.setattr("fixsub.cli.normalize_to_utf8", lambda source, target: target) + monkeypatch.setattr( + "fixsub.cli.probe_video", + lambda path: ProbeResult(7200, [AudioStream(1, 0, "ac3", "eng", 6, True)], {}), + ) + monkeypatch.setattr("fixsub.cli.score_alignment", lambda path, duration: AlignmentScore(0.40, ["low timeline coverage"])) + + def fake_sync(video_path: Path, subtitle_path: Path, output_path: Path, audio_stream: str) -> SyncResult: + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(subtitle_path.read_text(encoding="utf-8"), encoding="utf-8") + return SyncResult( + attempted=True, + succeeded=True, + output_path=output_path, + ffsubsync_score=12.0, + offset_seconds=8.4, + framerate_scale=1.0, + forced_low_quality=True, + ) + + monkeypatch.setattr("fixsub.cli.run_ffsubsync", fake_sync) + + result = CliRunner().invoke(app, ["--providers", "subhd"]) + + final_path = video.with_name(f"{video.stem}.zh.srt") + assert result.exit_code == 0 + assert final_path.exists() + assert "Applied low-confidence subtitle" in result.output + assert "forced synchronization" in result.output + metadata = _read_metadata(tmp_path) + assert metadata["final_output"] == str(final_path) + assert metadata["decisions"][0]["is_poor"] is True + assert metadata["decisions"][0]["sync_result"]["forced_low_quality"] is True +``` + +Add an original-fallback test using `SyncResult(attempted=True, succeeded=False, error="sync failed")`, asserting that the normalized candidate is written and output contains `original fallback`. + +Add a dry-run variant with the same forced synchronization result: + +```python +result = CliRunner().invoke(app, ["--dry-run", "--providers", "subhd"]) +assert result.exit_code == 0 +assert "Dry run complete" in result.output +assert "low-confidence" in result.output.lower() +assert "forced synchronization" in result.output +assert not video.with_name(f"{video.stem}.zh.srt").exists() +assert _read_metadata(tmp_path)["final_output"] is None +``` + +Keep the existing test for zero decisions unchanged: it must still produce `No downloadable or extractable subtitle candidates.` and no final subtitle. + +- [ ] **Step 2: Run the focused CLI tests and verify RED** + +Run: + +```bash +.venv/bin/python -m pytest \ + tests/test_cli_pipeline.py::test_cli_applies_forced_low_confidence_synced_candidate \ + tests/test_cli_pipeline.py::test_cli_applies_original_when_sync_fails \ + tests/test_cli_pipeline.py::test_cli_dry_run_reports_forced_low_confidence_without_writing \ + -v +``` + +Expected: FAIL because `run_pipeline` still blocks every `best.is_poor` result from `write_final_subtitle`. + +- [ ] **Step 3: Replace the low-confidence veto with warned output** + +Add a small outcome label helper above `run_pipeline`: + +```python +def _selection_label(decision: CandidateDecision) -> str: + if decision.sync_result.forced_low_quality and decision.selected_version == "synced": + return "forced synchronization" + if decision.selected_version == "original" and decision.sync_result.attempted: + return "original fallback" + if decision.selected_version == "synced": + return "synchronization" + return "original subtitle" +``` + +Replace the final output branch with: + +```python +selection_label = _selection_label(best) +final_output = None +if options.dry_run: + confidence = "Low-confidence " if best.is_poor else "" + message = ( + f"Dry run complete. {confidence}best candidate: {best.candidate.candidate_id} " + f"({selection_label}, timeline {best.selected_score:.2f})." + ) +else: + final_output = write_final_subtitle(best.selected_path, video_path, options.lang, workdirs.original) + if best.is_poor: + message = ( + f"Applied low-confidence subtitle ({selection_label}, timeline {best.selected_score:.2f}): " + f"{final_output}" + ) + else: + message = f"Applied subtitle: {final_output}" +``` + +Do not change the preceding `if not decisions` hard stop or ranking order. + +- [ ] **Step 4: Run all CLI pipeline tests and verify GREEN** + +Run: + +```bash +.venv/bin/python -m pytest tests/test_cli_pipeline.py -v +``` + +Expected: all CLI pipeline tests PASS. Update only old assertions that intentionally encoded the removed low-confidence veto; retain every no-candidate and dry-run safety assertion. + +- [ ] **Step 5: Update English and Chinese documentation** + +Replace statements that say poor synchronization prevents output with the new best-effort contract. The English text must state: + +```markdown +`fixsub` first asks `ffsubsync` to skip low-quality alignments. If that conservative pass explicitly refuses the alignment, `fixsub` retries once in forced mode. When at least one usable Chinese candidate exists, the best candidate is written even if confidence remains low; the console and metadata identify forced synchronization or original fallback. A hard stop remains when no candidate can be downloaded, extracted, parsed, or accepted as Chinese. +``` + +The Chinese text must state: + +```markdown +`fixsub` 会先要求 `ffsubsync` 跳过低质量对齐;如果保守同步明确拒绝该对齐,则自动以强制模式重试一次。只要至少存在一个可用的中文字幕候选项,即使置信度仍然较低,也会写入排名最高的候选项;控制台和元数据会明确标记强制同步或原字幕回退。只有在没有候选项能够下载、解压、解析或通过中文内容检查时才会停止且不写入字幕。 +``` + +Document that SubHD now uses its prepared-download flow and retain the artifact paths under `.fixsub/`. + +- [ ] **Step 6: Run the complete test suite and documentation checks** + +Run: + +```bash +.venv/bin/python -m pytest -q +``` + +Expected: all tests PASS with zero failures. + +Run: + +```bash +git diff --check +python3 -m build +``` + +Expected: `git diff --check` exits 0 and both source distribution and wheel build successfully. + +- [ ] **Step 7: Commit Task 3** + +```bash +git add fixsub/cli.py tests/test_cli_pipeline.py README.md README.zh-CN.md +git commit -m "fix: deliver low-confidence subtitle repairs" +``` + +--- + +## Final Verification + +After all three task reviews are clean: + +1. Run `.venv/bin/python -m pytest -q` and confirm zero failures. +2. Run `python3 -m build` and confirm both distribution formats are produced. +3. Run `git diff --check` and confirm no whitespace errors. +4. Perform a live SubHD protocol probe for a current public subtitle result: detail `GET`, preparation `POST`, gate `GET`, download API `POST`, final file `GET`. Save any downloaded file only under `/tmp`. +5. Run a full `fixsub --dry-run --providers subhd` inside an available movie folder if its path is available; otherwise report that the deterministic protocol probe and automated pipeline tests cover the behavior. diff --git a/docs/superpowers/specs/2026-07-22-subhd-download-and-low-confidence-output-design.md b/docs/superpowers/specs/2026-07-22-subhd-download-and-low-confidence-output-design.md new file mode 100644 index 0000000..3908f9a --- /dev/null +++ b/docs/superpowers/specs/2026-07-22-subhd-download-and-low-confidence-output-design.md @@ -0,0 +1,151 @@ +# SubHD Download and Low-Confidence Output Design + +**Date:** 2026-07-22 + +**Status:** Approved for implementation planning + +## Goal + +Restore compatibility with SubHD's current prepared-download flow and make `fixsub` fulfill its best-effort repair contract: when at least one downloadable, extractable, parseable Chinese subtitle exists, select the best candidate, attempt audio synchronization, and write a final subtitle even when synchronization confidence is low. + +## Scope + +This change covers two related pipeline failures: + +1. SubHD search results are found but downloads fail because the client directly opens `/down/{sid}` without first preparing that temporary page. +2. Candidates are retained but no final subtitle is written when `ffsubsync` rejects a low-quality alignment or the selected candidate otherwise remains low-confidence. + +The change does not add providers, alter movie detection, introduce interactive candidate selection, or replace the existing ranking model. + +## Confirmed User Experience + +- `fixsub` continues to prefer high-confidence candidates. +- `ffsubsync` first runs conservatively with `--skip-sync-on-low-quality`. +- When that conservative pass explicitly rejects a low-quality alignment, `fixsub` automatically retries without `--skip-sync-on-low-quality` and accepts the best alignment that `ffsubsync` can produce. +- When synchronization fails for another reason, `fixsub` falls back to the best original Chinese subtitle. +- A normal run writes a final subtitle whenever at least one usable Chinese candidate exists. +- A low-confidence or original-fallback result is clearly identified in the console message and metadata. +- `--dry-run` never writes the final subtitle, but reports which candidate and version would be selected and whether the result is low-confidence. +- A run still fails without output when no candidate can be downloaded, extracted, normalized, parsed, or accepted as Chinese. +- Original downloads and normalized candidates remain under `.fixsub/` for inspection and recovery. + +## Architecture + +The provider and synchronization changes remain inside their existing boundaries: + +- `fixsub/providers/subhd.py` owns SubHD's prepared-download protocol and URL validation. +- `fixsub/sync.py` owns conservative synchronization and the forced low-quality retry. +- `fixsub/models.py` records whether a successful synchronization required the forced retry. +- `fixsub/decision.py` continues to classify confidence and select synced versus original content. +- `fixsub/cli.py` ranks decisions and applies the best usable result, treating low confidence as a warning rather than an output veto. + +No new runtime dependency is required. + +## SubHD Download Flow + +For each selected SubHD search result, the client performs these requests in one persistent `httpx.Client` session: + +1. `GET /a/{sid}` to load the subtitle detail page. +2. `POST /api/sub/prepare-download` with JSON `{"sid": sid}` and the detail page as `Referer`. +3. Validate the JSON response and resolve its `url` value. +4. `GET` the prepared `/down/{sid}` page using the detail page as `Referer`. +5. If that response is already a subtitle or archive, save it through the existing direct-download path. +6. Otherwise, `POST /api/sub/down` with JSON `{"sid": sid}` and the prepared page as `Referer`. +7. Validate and download the returned subtitle or archive URL through the existing bounded redirect and content checks. + +The preparation response must be a JSON object with `success: true` and a non-empty string `url`. The resolved URL must: + +- use HTTPS; +- belong to an allowed SubHD-owned domain or the configured test host; +- have a path exactly equal to `/down/{sid}`; +- contain no username, password, non-default port, query string, or fragment. + +Preparation network failures, invalid JSON, rejection responses, missing URLs, and unsafe URLs become specific `FixsubError` messages. Existing URL validation remains in force for the final downloadable file. + +## Two-Pass Synchronization + +`run_ffsubsync` first executes its existing conservative command with `--skip-sync-on-low-quality`. + +If the command reports a low-quality alignment or that it is leaving subtitles unmodified: + +1. Delete any conservative-pass output. +2. Run `ffsubsync` again with the same video, subtitle, output path, and reference audio stream, but without `--skip-sync-on-low-quality`. +3. Require the forced pass to exit successfully, create the output file, and expose the same complete alignment metrics required today. +4. Return a successful `SyncResult` with `forced_low_quality=True`. + +A normal first-pass success returns `forced_low_quality=False`. A forced-pass process failure, missing output, or incomplete metrics remains a synchronization failure and triggers original-subtitle fallback in candidate selection. + +The retry is narrowly triggered only by the existing low-quality diagnostic phrases. Other failures are not retried because they indicate dependency, input, process, or output problems rather than a confidence policy decision. + +## Selection and Output Behavior + +Candidate confidence remains meaningful: + +- `is_poor=False` identifies a high-confidence selected result. +- `is_poor=True` identifies a low-confidence result, including forced low-quality synchronization and original fallback after synchronization failure. +- `rank_decisions` continues to place high-confidence decisions before low-confidence decisions, then compare selected timeline score, provider pre-score, and subtitle format as it does today. + +The output gate changes: + +- If there are no candidate decisions, raise `NoCandidatesError` as today. +- Otherwise, choose the top-ranked decision even when `is_poor=True`. +- In a normal run, write that decision's selected path through `write_final_subtitle`, preserving the existing backup behavior. +- In a dry run, do not write the final subtitle. + +Console messages distinguish four outcomes: + +- high-confidence normal application; +- low-confidence forced synchronization application; +- low-confidence original-subtitle fallback application; +- the corresponding dry-run selections without writing output. + +Messages identify the candidate, selected version, and timeline score so the user can inspect the decision without opening metadata. + +## Metadata + +`SyncResult` gains a boolean `forced_low_quality` field with a default of `False`. Existing serialization includes it automatically. + +The existing decision record retains: + +- `is_poor` as the confidence warning; +- `selected_version` as `original` or `synced`; +- synchronization metrics and errors; +- `decision_reason` describing normal synchronization, forced synchronization, or original fallback. + +`final_output` is populated for low-confidence normal runs and remains null for every dry run. + +## Error Handling and Safety + +- Provider errors continue to be isolated per candidate and written to `.fixsub/logs/fixsub.log`. +- SubHD preparation and download URLs are validated before requests to prevent redirects or API responses from sending the client to arbitrary hosts. +- Download response content continues to reject empty or HTML payloads and to identify supported archive or subtitle formats by signature and metadata. +- Low confidence is visible and retained in metadata; it is not silently presented as high confidence. +- Existing final subtitles are backed up before replacement. +- Forced synchronization never overwrites normalized candidates; it writes under `.fixsub/synced/` before final application. + +## Testing + +Tests are added test-first for these behaviors: + +1. SubHD performs detail, prepare, gate, download-API, and final-file requests in order. +2. SubHD rejects invalid preparation JSON, rejected preparation responses, missing URLs, and unsafe or mismatched prepared URLs. +3. A conservative low-quality result triggers exactly one forced retry without `--skip-sync-on-low-quality`. +4. A normal conservative success does not retry. +5. A forced retry failure returns a failed synchronization result and leaves no stale output. +6. A forced retry success records `forced_low_quality=True` and supplies complete metrics. +7. A low-confidence synced decision is applied during a normal run with a warning message. +8. A synchronization failure applies the best original Chinese candidate with a warning message. +9. Low-confidence dry runs report the selected candidate without creating a final subtitle. +10. Runs with no usable Chinese candidate still fail without a final subtitle. + +Focused tests run before the complete test suite. The final verification includes the full pytest suite and a live SubHD dry run or equivalent protocol probe against a current public subtitle result. + +## Documentation + +The English and Chinese README files are updated to explain: + +- SubHD's prepared-download support; +- the conservative then forced synchronization behavior; +- the fact that low confidence produces a warned best-effort output instead of blocking the final subtitle; +- the remaining hard-stop condition when no usable Chinese candidate exists; +- where to find original, candidate, synchronized, log, and metadata artifacts. diff --git a/fixsub/alignment.py b/fixsub/alignment.py index 40fca87..32ea7ef 100644 --- a/fixsub/alignment.py +++ b/fixsub/alignment.py @@ -14,7 +14,11 @@ def score_alignment(subtitle_path: Path, duration_seconds: float | None) -> Alig reasons: list[str] = [] intervals = parse_subtitle_intervals(subtitle_path) if not intervals: - return AlignmentScore(score=0.0, reasons=["no parseable subtitle intervals"]) + return AlignmentScore( + score=0.0, + reasons=["no parseable subtitle intervals"], + has_parseable_intervals=False, + ) valid_intervals = [(start, end) for start, end in intervals if start >= 0 and end > start] valid_rate = len(valid_intervals) / len(intervals) if valid_rate < 1.0: diff --git a/fixsub/cli.py b/fixsub/cli.py index 10cc9d6..544804f 100644 --- a/fixsub/cli.py +++ b/fixsub/cli.py @@ -135,6 +135,8 @@ def _decide_candidates( decisions: list[CandidateDecision] = [] for candidate in candidates: original_score = score_alignment(candidate.subtitle_path, duration_seconds) + if not original_score.has_parseable_intervals: + continue sync_result = SyncResult(attempted=False, succeeded=False) synced_score = None if not no_sync: @@ -149,6 +151,32 @@ def _decide_candidates( return decisions +def _selection_label(decision: CandidateDecision) -> str: + if decision.sync_result.forced_low_quality and decision.selected_version == "synced": + return "forced synchronization" + if decision.selected_version == "original" and decision.sync_result.attempted: + return "original fallback" + if decision.selected_version == "synced": + return "synchronization" + return "original subtitle" + + +def _format_selection_outcome( + decision: CandidateDecision, + *, + dry_run: bool, + final_output: Path | None = None, +) -> str: + confidence = "low-confidence" if decision.is_poor else "high-confidence" + details = f"({_selection_label(decision)}, timeline {decision.selected_score:.2f})" + if dry_run: + return ( + f"Dry run complete. {confidence.capitalize()} candidate " + f"{decision.candidate.candidate_id} selected {details}." + ) + return f"Applied {confidence} candidate {decision.candidate.candidate_id} {details}: {final_output}" + + def run_pipeline(base_dir: Path, options: RunOptions) -> dict[str, object]: workdirs = create_workdirs(base_dir) log_path = workdirs.logs / "fixsub.log" @@ -237,16 +265,11 @@ def run_pipeline(base_dir: Path, options: RunOptions) -> dict[str, object]: ranked_decisions = rank_decisions(decisions) best = ranked_decisions[0] final_output = None - if best.is_poor: - message = "No high-confidence subtitle found. Inspect candidates in .fixsub/candidates." - elif options.dry_run: - message = ( - f"Dry run complete. Best candidate: {best.candidate.candidate_id} " - f"({best.selected_version}, timeline {best.selected_score:.2f})." - ) + if options.dry_run: + message = _format_selection_outcome(best, dry_run=True) else: final_output = write_final_subtitle(best.selected_path, video_path, options.lang, workdirs.original) - message = f"Applied subtitle: {final_output}" + message = _format_selection_outcome(best, dry_run=False, final_output=final_output) metadata = _write_pipeline_metadata( metadata_path, diff --git a/fixsub/decision.py b/fixsub/decision.py index b66f040..3ad8418 100644 --- a/fixsub/decision.py +++ b/fixsub/decision.py @@ -12,12 +12,20 @@ def decide_candidate_version( synced_score: AlignmentScore | None, ) -> CandidateDecision: synced_output_exists = sync_result.output_path is not None and sync_result.output_path.exists() - synced_is_usable = sync_result.succeeded and synced_score is not None and synced_output_exists + synced_is_usable = ( + sync_result.succeeded + and synced_score is not None + and synced_score.has_parseable_intervals + and synced_output_exists + ) if synced_is_usable: selected_version = "synced" selected_path = sync_result.output_path selected_score = synced_score.score - reason = "ffsubsync audio alignment succeeded." + if sync_result.forced_low_quality: + reason = "ffsubsync forced a low-quality audio alignment." + else: + reason = "ffsubsync audio alignment succeeded." else: selected_version = "original" selected_path = candidate.subtitle_path @@ -28,12 +36,18 @@ def decide_candidate_version( reason = "Synced output missing; original candidate kept." elif sync_result.attempted and synced_score is None: reason = "Synced score missing; original candidate kept." + elif sync_result.attempted and not synced_score.has_parseable_intervals: + reason = "Synced output has no parseable subtitle intervals; original candidate kept." elif sync_result.attempted: reason = "ffsubsync validation did not produce a usable synced subtitle." else: reason = "Audio validation explicitly skipped; original candidate kept." usable_score = synced_score.score if selected_version == "synced" and synced_score is not None else original_score.score - is_poor = usable_score < POOR_ALIGNMENT or (sync_result.attempted and not synced_is_usable) + is_poor = ( + sync_result.forced_low_quality + or usable_score < POOR_ALIGNMENT + or (sync_result.attempted and not synced_is_usable) + ) return CandidateDecision( candidate=candidate, original_score=original_score, diff --git a/fixsub/models.py b/fixsub/models.py index b6dad26..e443a24 100644 --- a/fixsub/models.py +++ b/fixsub/models.py @@ -151,6 +151,7 @@ def to_json(self) -> dict[str, Any]: class AlignmentScore: score: float reasons: list[str] + has_parseable_intervals: bool = True def to_json(self) -> dict[str, Any]: return asdict(self) @@ -165,6 +166,7 @@ class SyncResult: ffsubsync_score: float | None = None offset_seconds: float | None = None framerate_scale: float | None = None + forced_low_quality: bool = False def to_json(self) -> dict[str, Any]: return _path_to_str(asdict(self)) diff --git a/fixsub/providers/subhd.py b/fixsub/providers/subhd.py index 371d0a1..0668b5f 100644 --- a/fixsub/providers/subhd.py +++ b/fixsub/providers/subhd.py @@ -257,6 +257,36 @@ def _save_download(self, response: httpx.Response, result: SearchResult, target_ source_url=str(response.request.url), ) + def _prepare_download_url(self, result: SearchResult, detail_url: str) -> str: + prepare_url = f"{self.base_url}/api/sub/prepare-download" + response = self._request_stage( + "POST", + prepare_url, + "SubHD download preparation request failed", + json={"sid": result.result_id}, + headers={"Referer": detail_url}, + ) + try: + payload = response.json() + except ValueError as exc: + raise FixsubError("SubHD download preparation returned invalid JSON") from exc + if not isinstance(payload, dict): + raise FixsubError("SubHD download preparation returned invalid JSON") + if payload.get("success") is not True: + message = _sanitize_rejection_message(payload.get("msg")) + raise FixsubError(f"SubHD download preparation rejected the request: {message}") + prepared_url = payload.get("url") + if not isinstance(prepared_url, str) or not prepared_url.strip(): + raise FixsubError("SubHD download preparation omitted a prepared download URL") + resolved_url = urljoin(prepare_url, prepared_url.strip()) + if not _is_allowed_subhd_url(resolved_url, self.base_url): + raise FixsubError(f"SubHD prepared download URL is not allowed: {resolved_url}") + parsed = urlparse(resolved_url) + expected_path = f"/down/{quote(result.result_id, safe='')}" + if parsed.path != expected_path or parsed.query or parsed.fragment: + raise FixsubError(f"SubHD prepared download URL does not match subtitle: {resolved_url}") + return resolved_url + def _api_download_url(self, result: SearchResult, gate_url: str) -> str: api_url = f"{self.base_url}/api/sub/down" response = self._request_stage( @@ -285,9 +315,9 @@ def _api_download_url(self, result: SearchResult, gate_url: str) -> str: def download(self, result: SearchResult, target_dir) -> DownloadedFile: detail_url = result.detail_url or f"{self.base_url}/a/{result.result_id}" - detail_response = self._request_stage("GET", detail_url, "SubHD detail request failed") + self._request_stage("GET", detail_url, "SubHD detail request failed") - gate_url = result.download_url or f"{self.base_url}/down/{result.result_id}" + gate_url = self._prepare_download_url(result, detail_url) gate_response = self._request_stage( "GET", gate_url, diff --git a/fixsub/sync.py b/fixsub/sync.py index 8be0bfc..82bba79 100644 --- a/fixsub/sync.py +++ b/fixsub/sync.py @@ -8,12 +8,19 @@ from fixsub.errors import MissingDependencyError from fixsub.models import SyncResult +LOW_QUALITY_MARKERS = ("low-quality alignment", "leaving subtitles unmodified") + def _metric(output: str, label: str) -> float | None: match = re.search(rf"{re.escape(label)}:\s*([-+]?\d+(?:\.\d+)?)", output, re.IGNORECASE) return float(match.group(1)) if match else None +def _is_low_quality(diagnostics: str) -> bool: + lowered = diagnostics.lower() + return any(marker in lowered for marker in LOW_QUALITY_MARKERS) + + def synced_output_path(candidate_path: Path, synced_dir: Path) -> Path: return synced_dir / f"{candidate_path.stem}.synced{candidate_path.suffix}" @@ -22,8 +29,7 @@ def run_ffsubsync(video_path: Path, subtitle_path: Path, output_path: Path, audi if not shutil.which("ffs"): raise MissingDependencyError("ffs", "python3 -m pip install ffsubsync") output_path.parent.mkdir(parents=True, exist_ok=True) - output_path.unlink(missing_ok=True) - command = [ + conservative_command = [ "ffs", str(video_path), "--reference-stream", @@ -34,48 +40,68 @@ def run_ffsubsync(video_path: Path, subtitle_path: Path, output_path: Path, audi "-o", str(output_path), ] - try: - result = subprocess.run(command, capture_output=True, text=True) - except OSError as exc: - return SyncResult(attempted=True, succeeded=False, output_path=None, error=str(exc)) - diagnostics = "\n".join(part for part in [result.stdout, result.stderr] if part) - metrics = { - "ffsubsync_score": _metric(diagnostics, "score"), - "offset_seconds": _metric(diagnostics, "offset seconds"), - "framerate_scale": _metric(diagnostics, "framerate scale factor"), - } - if result.returncode != 0: - return SyncResult( - attempted=True, - succeeded=False, - output_path=None, - error=result.stderr.strip() or result.stdout.strip(), - **metrics, - ) - if "low-quality alignment" in diagnostics.lower() or "leaving subtitles unmodified" in diagnostics.lower(): - output_path.unlink(missing_ok=True) - return SyncResult( - attempted=True, - succeeded=False, - output_path=None, - error="ffsubsync rejected a low-quality alignment", - **metrics, - ) - if not output_path.exists(): - return SyncResult( - attempted=True, - succeeded=False, - output_path=None, - error="ffsubsync exited successfully without writing an output file", - **metrics, - ) - if any(value is None for value in metrics.values()): + commands = [conservative_command] + forced_low_quality = False + + for command in commands: output_path.unlink(missing_ok=True) + try: + result = subprocess.run(command, capture_output=True, text=True) + except OSError as exc: + return SyncResult( + attempted=True, + succeeded=False, + output_path=None, + error=str(exc), + forced_low_quality=forced_low_quality, + ) + diagnostics = "\n".join(part for part in [result.stdout, result.stderr] if part) + metrics = { + "ffsubsync_score": _metric(diagnostics, "score"), + "offset_seconds": _metric(diagnostics, "offset seconds"), + "framerate_scale": _metric(diagnostics, "framerate scale factor"), + } + if not forced_low_quality and result.returncode == 0 and _is_low_quality(diagnostics): + output_path.unlink(missing_ok=True) + forced_low_quality = True + commands.append([part for part in conservative_command if part != "--skip-sync-on-low-quality"]) + continue + if result.returncode != 0: + output_path.unlink(missing_ok=True) + return SyncResult( + attempted=True, + succeeded=False, + output_path=None, + error=result.stderr.strip() or result.stdout.strip(), + forced_low_quality=forced_low_quality, + **metrics, + ) + if not output_path.exists(): + return SyncResult( + attempted=True, + succeeded=False, + output_path=None, + error="ffsubsync exited successfully without writing an output file", + forced_low_quality=forced_low_quality, + **metrics, + ) + if any(value is None for value in metrics.values()): + output_path.unlink(missing_ok=True) + return SyncResult( + attempted=True, + succeeded=False, + output_path=None, + error="ffsubsync output did not include complete alignment metrics", + forced_low_quality=forced_low_quality, + **metrics, + ) return SyncResult( attempted=True, - succeeded=False, - output_path=None, - error="ffsubsync output did not include complete alignment metrics", + succeeded=True, + output_path=output_path, + error=None, + forced_low_quality=forced_low_quality, **metrics, ) - return SyncResult(attempted=True, succeeded=True, output_path=output_path, error=None, **metrics) + + raise AssertionError("ffsubsync command loop completed without a result") diff --git a/tests/test_cli_pipeline.py b/tests/test_cli_pipeline.py index 5229817..ad79c14 100644 --- a/tests/test_cli_pipeline.py +++ b/tests/test_cli_pipeline.py @@ -236,8 +236,8 @@ def download(self, result: SearchResult, target_dir: Path) -> DownloadedFile: ).read_text(encoding="utf-8") -def test_cli_runs_dry_run_pipeline(tmp_path: Path, monkeypatch) -> None: - _write_video(tmp_path) +def test_cli_reports_high_confidence_selection_for_dry_and_normal_runs(tmp_path: Path, monkeypatch) -> None: + video = _write_video(tmp_path) subtitle = tmp_path / ".fixsub" / "candidates" / "assrt_1001.ass" subtitle.parent.mkdir(parents=True) subtitle.write_text("[Events]\nDialogue: 0,0:02:00.00,0:02:03.00,Default,,0,0,0,,Hi\n", encoding="utf-8") @@ -294,12 +294,295 @@ def fake_sync(video_path: Path, subtitle_path: Path, output_path: Path, audio_st monkeypatch.setattr("fixsub.cli.run_ffsubsync", fake_sync) - result = CliRunner().invoke(app, ["--dry-run", "--providers", "assrt"]) + dry_run_result = CliRunner().invoke(app, ["--dry-run", "--providers", "assrt"]) + normal_result = CliRunner().invoke(app, ["--providers", "assrt"]) + + final_path = video.with_name(f"{video.stem}.zh.ass") + assert dry_run_result.exit_code == 0 + assert "Selected reference audio: a:0" in dry_run_result.output + assert ( + "Dry run complete. High-confidence candidate assrt_1001 selected " + "(synchronization, timeline 1.00)." + ) in dry_run_result.output + assert normal_result.exit_code == 0 + assert ( + f"Applied high-confidence candidate assrt_1001 (synchronization, timeline 1.00): {final_path}" + ) in normal_result.output + assert (tmp_path / ".fixsub" / "metadata" / "results.json").exists() + + +def test_cli_applies_forced_low_confidence_synced_candidate(tmp_path: Path, monkeypatch) -> None: + video = _write_video(tmp_path) + subtitle = tmp_path / ".fixsub" / "candidates" / "subhd_forced.srt" + subtitle.parent.mkdir(parents=True) + subtitle.write_text("1\n00:00:01,000 --> 00:00:02,000\n字幕\n", encoding="utf-8") + synced = tmp_path / ".fixsub" / "synced" / "subhd_forced.synced.srt" + monkeypatch.chdir(tmp_path) + monkeypatch.delenv("ASSRT_TOKEN", raising=False) + class FakeSubhdClient: + def search(self, query: str) -> list[SearchResult]: + return [ + SearchResult( + provider="subhd", + result_id="forced", + title="Movie 1992 中文字幕", + language="zh-Hans", + format="srt", + ) + ] + + def download(self, result: SearchResult, target_dir: Path) -> DownloadedFile: + return DownloadedFile("subhd_forced", "subhd", subtitle, result.download_url) + + monkeypatch.setattr("fixsub.providers.registry.SubhdClient", FakeSubhdClient) + monkeypatch.setattr("fixsub.cli.extract_archive", lambda path, out_dir: [subtitle]) + monkeypatch.setattr("fixsub.cli.normalize_to_utf8", lambda source, target: target) + monkeypatch.setattr( + "fixsub.cli.probe_video", + lambda path: ProbeResult(7200, [AudioStream(1, 0, "ac3", "eng", 6, True)], {}), + ) + monkeypatch.setattr("fixsub.cli.score_alignment", lambda path, duration: AlignmentScore(0.40, ["low timeline coverage"])) + + def fake_sync(video_path: Path, subtitle_path: Path, output_path: Path, audio_stream: str) -> SyncResult: + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(subtitle_path.read_text(encoding="utf-8"), encoding="utf-8") + return SyncResult( + attempted=True, + succeeded=True, + output_path=output_path, + ffsubsync_score=12.0, + offset_seconds=8.4, + framerate_scale=1.0, + forced_low_quality=True, + ) + + monkeypatch.setattr("fixsub.cli.run_ffsubsync", fake_sync) + + result = CliRunner().invoke(app, ["--providers", "subhd"]) + + final_path = video.with_name(f"{video.stem}.zh.srt") assert result.exit_code == 0 - assert "Selected reference audio: a:0" in result.output - assert "Dry run complete. Best candidate: assrt_1001 (synced, timeline 1.00)." in result.output - assert (tmp_path / ".fixsub" / "metadata" / "results.json").exists() + assert final_path.exists() + assert ( + f"Applied low-confidence candidate subhd_forced (forced synchronization, timeline 0.40): {final_path}" + ) in result.output + metadata = _read_metadata(tmp_path) + assert metadata["final_output"] == str(final_path) + assert metadata["decisions"][0]["is_poor"] is True + assert metadata["decisions"][0]["sync_result"]["forced_low_quality"] is True + + +def test_cli_applies_original_when_sync_fails(tmp_path: Path, monkeypatch) -> None: + video = _write_video(tmp_path) + subtitle = tmp_path / ".fixsub" / "candidates" / "subhd_fallback.srt" + subtitle.parent.mkdir(parents=True) + subtitle.write_text("1\n00:00:01,000 --> 00:00:02,000\n字幕\n", encoding="utf-8") + monkeypatch.chdir(tmp_path) + monkeypatch.delenv("ASSRT_TOKEN", raising=False) + + class FakeSubhdClient: + def search(self, query: str) -> list[SearchResult]: + return [ + SearchResult( + provider="subhd", + result_id="fallback", + title="Movie 1992 中文字幕", + language="zh-Hans", + format="srt", + ) + ] + + def download(self, result: SearchResult, target_dir: Path) -> DownloadedFile: + return DownloadedFile("subhd_fallback", "subhd", subtitle, result.download_url) + + monkeypatch.setattr("fixsub.providers.registry.SubhdClient", FakeSubhdClient) + monkeypatch.setattr("fixsub.cli.extract_archive", lambda path, out_dir: [subtitle]) + monkeypatch.setattr("fixsub.cli.normalize_to_utf8", lambda source, target: target) + monkeypatch.setattr( + "fixsub.cli.probe_video", + lambda path: ProbeResult(7200, [AudioStream(1, 0, "ac3", "eng", 6, True)], {}), + ) + monkeypatch.setattr("fixsub.cli.score_alignment", lambda path, duration: AlignmentScore(0.40, ["low timeline coverage"])) + monkeypatch.setattr( + "fixsub.cli.run_ffsubsync", + lambda video_path, subtitle_path, output_path, audio_stream: SyncResult( + attempted=True, + succeeded=False, + error="sync failed", + ), + ) + + result = CliRunner().invoke(app, ["--providers", "subhd"]) + + final_path = video.with_name(f"{video.stem}.zh.srt") + assert result.exit_code == 0 + assert final_path.exists() + assert final_path.read_text(encoding="utf-8") == subtitle.read_text(encoding="utf-8") + assert ( + f"Applied low-confidence candidate subhd_fallback (original fallback, timeline 0.40): {final_path}" + ) in result.output + assert _read_metadata(tmp_path)["final_output"] == str(final_path) + + +def test_cli_dry_run_reports_forced_low_confidence_without_writing(tmp_path: Path, monkeypatch) -> None: + video = _write_video(tmp_path) + subtitle = tmp_path / ".fixsub" / "candidates" / "subhd_forced.srt" + subtitle.parent.mkdir(parents=True) + subtitle.write_text("1\n00:00:01,000 --> 00:00:02,000\n字幕\n", encoding="utf-8") + monkeypatch.chdir(tmp_path) + monkeypatch.delenv("ASSRT_TOKEN", raising=False) + + class FakeSubhdClient: + def search(self, query: str) -> list[SearchResult]: + return [ + SearchResult( + provider="subhd", + result_id="forced", + title="Movie 1992 中文字幕", + language="zh-Hans", + format="srt", + ) + ] + + def download(self, result: SearchResult, target_dir: Path) -> DownloadedFile: + return DownloadedFile("subhd_forced", "subhd", subtitle, result.download_url) + + monkeypatch.setattr("fixsub.providers.registry.SubhdClient", FakeSubhdClient) + monkeypatch.setattr("fixsub.cli.extract_archive", lambda path, out_dir: [subtitle]) + monkeypatch.setattr("fixsub.cli.normalize_to_utf8", lambda source, target: target) + monkeypatch.setattr( + "fixsub.cli.probe_video", + lambda path: ProbeResult(7200, [AudioStream(1, 0, "ac3", "eng", 6, True)], {}), + ) + monkeypatch.setattr("fixsub.cli.score_alignment", lambda path, duration: AlignmentScore(0.40, ["low timeline coverage"])) + + def fake_sync(video_path: Path, subtitle_path: Path, output_path: Path, audio_stream: str) -> SyncResult: + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(subtitle_path.read_text(encoding="utf-8"), encoding="utf-8") + return SyncResult( + attempted=True, + succeeded=True, + output_path=output_path, + ffsubsync_score=12.0, + offset_seconds=8.4, + framerate_scale=1.0, + forced_low_quality=True, + ) + + monkeypatch.setattr("fixsub.cli.run_ffsubsync", fake_sync) + + result = CliRunner().invoke(app, ["--dry-run", "--providers", "subhd"]) + + assert result.exit_code == 0 + assert "Dry run complete" in result.output + assert "low-confidence" in result.output.lower() + assert "forced synchronization" in result.output + assert not video.with_name(f"{video.stem}.zh.srt").exists() + assert _read_metadata(tmp_path)["final_output"] is None + + +def test_cli_rejects_unparseable_download_without_touching_existing_final( + tmp_path: Path, monkeypatch +) -> None: + video = _write_video(tmp_path) + final_path = video.with_name(f"{video.stem}.zh.srt") + final_path.write_text("existing final subtitle\n", encoding="utf-8") + subtitle = tmp_path / ".fixsub" / "candidates" / "subhd_unparseable.srt" + subtitle.parent.mkdir(parents=True) + subtitle.write_text("中文字幕 without timing entries\n", encoding="utf-8") + monkeypatch.chdir(tmp_path) + monkeypatch.delenv("ASSRT_TOKEN", raising=False) + + class FakeSubhdClient: + def search(self, query: str) -> list[SearchResult]: + return [ + SearchResult( + provider="subhd", + result_id="unparseable", + title="Movie 1992 中文字幕", + language="zh-Hans", + format="srt", + ) + ] + + def download(self, result: SearchResult, target_dir: Path) -> DownloadedFile: + return DownloadedFile("subhd_unparseable", "subhd", subtitle, result.download_url) + + monkeypatch.setattr("fixsub.providers.registry.SubhdClient", FakeSubhdClient) + monkeypatch.setattr("fixsub.cli.extract_archive", lambda path, out_dir: [subtitle]) + monkeypatch.setattr("fixsub.cli.normalize_to_utf8", lambda source, target: target) + monkeypatch.setattr( + "fixsub.cli.probe_video", + lambda path: ProbeResult(7200, [AudioStream(1, 0, "ac3", "eng", 6, True)], {}), + ) + + result = CliRunner().invoke(app, ["--no-sync", "--providers", "subhd"]) + + assert result.exit_code != 0 + assert "No downloadable or extractable subtitle candidates." in result.output + assert final_path.read_text(encoding="utf-8") == "existing final subtitle\n" + assert list((tmp_path / ".fixsub" / "original").iterdir()) == [] + metadata = _read_metadata(tmp_path) + assert metadata["decisions"] == [] + assert metadata["final_output"] is None + + +def test_cli_falls_back_when_forced_sync_output_is_unparseable(tmp_path: Path, monkeypatch) -> None: + video = _write_video(tmp_path) + subtitle = tmp_path / ".fixsub" / "candidates" / "subhd_parseable.srt" + subtitle.parent.mkdir(parents=True) + subtitle.write_text("1\n00:00:01,000 --> 00:00:02,000\n字幕\n", encoding="utf-8") + monkeypatch.chdir(tmp_path) + monkeypatch.delenv("ASSRT_TOKEN", raising=False) + + class FakeSubhdClient: + def search(self, query: str) -> list[SearchResult]: + return [ + SearchResult( + provider="subhd", + result_id="parseable", + title="Movie 1992 中文字幕", + language="zh-Hans", + format="srt", + ) + ] + + def download(self, result: SearchResult, target_dir: Path) -> DownloadedFile: + return DownloadedFile("subhd_parseable", "subhd", subtitle, result.download_url) + + monkeypatch.setattr("fixsub.providers.registry.SubhdClient", FakeSubhdClient) + monkeypatch.setattr("fixsub.cli.extract_archive", lambda path, out_dir: [subtitle]) + monkeypatch.setattr("fixsub.cli.normalize_to_utf8", lambda source, target: target) + monkeypatch.setattr( + "fixsub.cli.probe_video", + lambda path: ProbeResult(7200, [AudioStream(1, 0, "ac3", "eng", 6, True)], {}), + ) + + def fake_sync(video_path: Path, subtitle_path: Path, output_path: Path, audio_stream: str) -> SyncResult: + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text("forced sync output without timing entries\n", encoding="utf-8") + return SyncResult( + attempted=True, + succeeded=True, + output_path=output_path, + ffsubsync_score=12.0, + offset_seconds=8.4, + framerate_scale=1.0, + forced_low_quality=True, + ) + + monkeypatch.setattr("fixsub.cli.run_ffsubsync", fake_sync) + + result = CliRunner().invoke(app, ["--providers", "subhd"]) + + final_path = video.with_name(f"{video.stem}.zh.srt") + assert result.exit_code == 0 + assert final_path.read_text(encoding="utf-8") == subtitle.read_text(encoding="utf-8") + assert "original fallback" in result.output + metadata = _read_metadata(tmp_path) + assert metadata["decisions"][0]["selected_version"] == "original" + assert "no parseable subtitle intervals" in metadata["decisions"][0]["decision_reason"] def test_cli_writes_metadata_when_token_is_missing_after_video_detection(tmp_path: Path, monkeypatch) -> None: diff --git a/tests/test_decision_ranking_output.py b/tests/test_decision_ranking_output.py index e4c0d95..0fa959a 100644 --- a/tests/test_decision_ranking_output.py +++ b/tests/test_decision_ranking_output.py @@ -56,6 +56,29 @@ def test_decision_selects_audio_validated_sync_even_when_original_timeline_is_ex assert decision.decision_reason == "ffsubsync audio alignment succeeded." +def test_decision_selects_forced_sync_and_marks_it_low_confidence(tmp_path: Path) -> None: + candidate = make_candidate(tmp_path) + synced = tmp_path / "candidate.synced.ass" + synced.write_text("[Events]\n", encoding="utf-8") + + decision = decide_candidate_version( + candidate=candidate, + original_score=AlignmentScore(0.90, []), + sync_result=SyncResult( + attempted=True, + succeeded=True, + output_path=synced, + forced_low_quality=True, + ), + synced_score=AlignmentScore(0.90, []), + ) + + assert decision.selected_version == "synced" + assert decision.selected_path == synced + assert decision.is_poor is True + assert decision.decision_reason == "ffsubsync forced a low-quality audio alignment." + + def test_decision_selects_synced_at_exact_improvement_threshold(tmp_path: Path) -> None: candidate = make_candidate(tmp_path) synced = tmp_path / "candidate.synced.ass" diff --git a/tests/test_paths_models.py b/tests/test_paths_models.py index 115a966..a518fe7 100644 --- a/tests/test_paths_models.py +++ b/tests/test_paths_models.py @@ -123,6 +123,7 @@ def test_candidate_decision_serialization_converts_nested_paths(tmp_path: Path) json.dumps(data) assert data["candidate"]["subtitle_path"].endswith("movie.ass") assert data["sync_result"]["output_path"].endswith("movie.synced.ass") + assert data["sync_result"]["forced_low_quality"] is False assert data["selected_path"].endswith("movie.synced.ass") diff --git a/tests/test_subhd.py b/tests/test_subhd.py index 1683d03..b485167 100644 --- a/tests/test_subhd.py +++ b/tests/test_subhd.py @@ -300,6 +300,70 @@ def handler(request: httpx.Request) -> httpx.Response: assert [result.result_id for result in results] == ["kAqdvK"] +def test_subhd_client_prepares_download_page_before_opening_gate(tmp_path: Path) -> None: + requests: list[tuple[str, str, bytes, str | None]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append((request.method, str(request.url), request.content, request.headers.get("Referer"))) + if request.url.path == "/a/kAqdvK": + return httpx.Response(200, text="detail", request=request) + if request.url.path == "/api/sub/prepare-download": + return httpx.Response(200, json={"success": True, "url": "/down/kAqdvK"}, request=request) + if request.url.path == "/down/kAqdvK": + return httpx.Response(200, text="gate", request=request) + if request.url.path == "/api/sub/down": + return httpx.Response( + 200, + json={"success": True, "pass": True, "url": "https://dl.subhd.me/result.zip"}, + request=request, + ) + if str(request.url) == "https://dl.subhd.me/result.zip": + return httpx.Response(200, content=b"PK\x03\x04archive", request=request) + raise AssertionError(f"Unexpected request: {request.method} {request.url}") + + client = SubhdClient(http_client=httpx.Client(transport=httpx.MockTransport(handler))) + result = parse_search_response(SEARCH_HTML, "https://subhd.tv/search/Nell%201994")[0] + + downloaded = client.download(result, tmp_path) + + assert downloaded.path.read_bytes() == b"PK\x03\x04archive" + assert [(method, url) for method, url, _body, _referer in requests] == [ + ("GET", "https://subhd.tv/a/kAqdvK"), + ("POST", "https://subhd.tv/api/sub/prepare-download"), + ("GET", "https://subhd.tv/down/kAqdvK"), + ("POST", "https://subhd.tv/api/sub/down"), + ("GET", "https://dl.subhd.me/result.zip"), + ] + assert requests[1][2] == b'{"sid":"kAqdvK"}' + assert requests[1][3] == "https://subhd.tv/a/kAqdvK" + assert requests[2][3] == "https://subhd.tv/a/kAqdvK" + + +@pytest.mark.parametrize( + ("payload", "message"), + [ + (None, "invalid JSON"), + ([], "invalid JSON"), + ({"success": False, "msg": "准备失败"}, "准备失败"), + ({"success": True}, "omitted a prepared download URL"), + ({"success": True, "url": "https://evil.example/down/kAqdvK"}, "prepared download URL is not allowed"), + ({"success": True, "url": "/down/other"}, "prepared download URL does not match subtitle"), + ({"success": True, "url": "/down/kAqdvK?token=x"}, "prepared download URL does not match subtitle"), + ], +) +def test_subhd_prepare_download_rejects_invalid_responses(payload: object, message: str) -> None: + def handler(request: httpx.Request) -> httpx.Response: + if payload is None: + return httpx.Response(200, text="not json", request=request) + return httpx.Response(200, json=payload, request=request) + + client = SubhdClient(http_client=httpx.Client(transport=httpx.MockTransport(handler))) + result = parse_search_response(SEARCH_HTML, "https://subhd.tv/search/Nell%201994")[0] + + with pytest.raises(FixsubError, match=message): + client._prepare_download_url(result, "https://subhd.tv/a/kAqdvK") + + def test_subhd_client_download_uses_session_api_and_saves_archive(tmp_path: Path) -> None: requests: list[tuple[str, str]] = [] @@ -312,6 +376,12 @@ def handler(request: httpx.Request) -> httpx.Response: headers={"Set-Cookie": "tk_download=ready; Path=/; HttpOnly"}, request=request, ) + if str(request.url) == "https://subhd.tv/api/sub/prepare-download": + assert request.method == "POST" + assert "tk_download=ready" in request.headers.get("Cookie", "") + assert request.headers["Referer"] == "https://subhd.tv/a/kAqdvK" + assert request.content == b'{"sid":"kAqdvK"}' + return httpx.Response(200, json={"success": True, "url": "/down/kAqdvK"}, request=request) if str(request.url) == "https://subhd.tv/down/kAqdvK": assert "tk_download=ready" in request.headers.get("Cookie", "") assert request.headers["Referer"] == "https://subhd.tv/a/kAqdvK" @@ -346,6 +416,7 @@ def handler(request: httpx.Request) -> httpx.Response: assert downloaded.source_url == "https://dl.subhd.me/subtitles/kAqdvK.zip" assert requests == [ ("GET", "https://subhd.tv/a/kAqdvK"), + ("POST", "https://subhd.tv/api/sub/prepare-download"), ("GET", "https://subhd.tv/down/kAqdvK"), ("POST", "https://subhd.tv/api/sub/down"), ("GET", "https://dl.subhd.me/subtitles/kAqdvK.zip"), @@ -359,6 +430,8 @@ def handler(request: httpx.Request) -> httpx.Response: requested_urls.append(str(request.url)) if str(request.url) == "https://subhd.tv/a/kAqdvK": return httpx.Response(200, text="detail", request=request) + if request.url.path == "/api/sub/prepare-download": + return httpx.Response(200, json={"success": True, "url": "/down/kAqdvK"}, request=request) if str(request.url) == "https://subhd.tv/down/kAqdvK": return httpx.Response(200, content=b"PK\x03\x04legacy", request=request) raise AssertionError(f"Unexpected request: {request.url}") @@ -371,6 +444,7 @@ def handler(request: httpx.Request) -> httpx.Response: assert downloaded.path.read_bytes() == b"PK\x03\x04legacy" assert requested_urls == [ "https://subhd.tv/a/kAqdvK", + "https://subhd.tv/api/sub/prepare-download", "https://subhd.tv/down/kAqdvK", ] @@ -379,6 +453,8 @@ def test_subhd_client_download_rejects_empty_final_response(tmp_path: Path) -> N def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/a/kAqdvK": return httpx.Response(200, text="detail", request=request) + if request.url.path == "/api/sub/prepare-download": + return httpx.Response(200, json={"success": True, "url": "/down/kAqdvK"}, request=request) if request.url.path == "/down/kAqdvK": return httpx.Response(200, text="download gate", request=request) if request.url.path == "/api/sub/down": @@ -405,6 +481,8 @@ def handler(request: httpx.Request) -> httpx.Response: requested_urls.append(str(request.url)) if request.url.path == "/a/kAqdvK": return httpx.Response(200, text="detail", request=request) + if request.url.path == "/api/sub/prepare-download": + return httpx.Response(200, json={"success": True, "url": "/down/kAqdvK"}, request=request) if request.url.path == "/down/kAqdvK": return httpx.Response(200, text="gate", request=request) if request.url.path == "/api/sub/down": @@ -431,6 +509,8 @@ def handler(request: httpx.Request) -> httpx.Response: requested_urls.append(str(request.url)) if request.url.path == "/a/kAqdvK": return httpx.Response(200, text="detail", request=request) + if request.url.path == "/api/sub/prepare-download": + return httpx.Response(200, json={"success": True, "url": "/down/kAqdvK"}, request=request) if request.url.path == "/down/kAqdvK": return httpx.Response(200, text="gate", request=request) if request.url.path == "/api/sub/down": @@ -456,6 +536,7 @@ def handler(request: httpx.Request) -> httpx.Response: ("failed_stage", "expected_message"), [ ("detail", "SubHD detail request failed"), + ("prepare", "SubHD download preparation request failed"), ("gate", "SubHD download page request failed"), ("api", "SubHD download API request failed"), ("file", "SubHD subtitle file request failed"), @@ -472,6 +553,9 @@ def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/a/kAqdvK": stage = "detail" response = httpx.Response(200, text="detail", request=request) + elif request.url.path == "/api/sub/prepare-download": + stage = "prepare" + response = httpx.Response(200, json={"success": True, "url": "/down/kAqdvK"}, request=request) elif request.url.path == "/down/kAqdvK": stage = "gate" response = httpx.Response(200, text="gate", request=request) @@ -524,6 +608,8 @@ def test_subhd_client_reports_download_api_errors( def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/a/kAqdvK": return httpx.Response(200, text="detail", request=request) + if request.url.path == "/api/sub/prepare-download": + return httpx.Response(200, json={"success": True, "url": "/down/kAqdvK"}, request=request) if request.url.path == "/down/kAqdvK": return httpx.Response(200, text="gate", request=request) if request.url.path == "/api/sub/down": @@ -643,6 +729,8 @@ def test_subhd_client_download_rejects_html_final_response( def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/a/kAqdvK": return httpx.Response(200, text="detail", request=request) + if request.url.path == "/api/sub/prepare-download": + return httpx.Response(200, json={"success": True, "url": "/down/kAqdvK"}, request=request) if request.url.path == "/down/kAqdvK": return httpx.Response(200, text="download gate", request=request) if request.url.path == "/api/sub/down": @@ -666,6 +754,8 @@ def test_subhd_client_download_sniffs_plain_srt_content(tmp_path: Path) -> None: def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/a/kAqdvK": return httpx.Response(200, text="detail", request=request) + if request.url.path == "/api/sub/prepare-download": + return httpx.Response(200, json={"success": True, "url": "/down/kAqdvK"}, request=request) if request.url.path == "/down/kAqdvK": return httpx.Response(200, content=b"1\n00:00:01,000 --> 00:00:02,000\nHi\n", request=request) raise AssertionError(f"Unexpected request: {request.url}") diff --git a/tests/test_sync.py b/tests/test_sync.py index 7a0b0c0..2dbf877 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -115,23 +115,18 @@ def fake_run(command: list[str], capture_output: bool, text: bool) -> subprocess ) -def test_run_ffsubsync_rejects_zero_exit_low_quality_result(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: +def test_run_ffsubsync_retries_low_quality_without_skip_flag(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: monkeypatch.setattr("fixsub.sync.shutil.which", lambda command: "/usr/bin/ffs") output = tmp_path / "synced" / "candidate.srt" + calls: list[list[str]] = [] def fake_run(command: list[str], capture_output: bool, text: bool) -> subprocess.CompletedProcess[str]: - output.write_text("original subtitle", encoding="utf-8") - return subprocess.CompletedProcess( - command, - 0, - stdout=( - "score: 305917.769\n" - "offset seconds: 31.730\n" - "framerate scale factor: 1.043\n" - "low-quality alignment; leaving subtitles unmodified\n" - ), - stderr="", - ) + calls.append(command) + output.write_text("forced subtitle", encoding="utf-8") + diagnostics = "score: 12.0\noffset seconds: 31.730\nframerate scale factor: 1.043\n" + if "--skip-sync-on-low-quality" in command: + diagnostics += "low-quality alignment; leaving subtitles unmodified\n" + return subprocess.CompletedProcess(command, 0, stdout=diagnostics, stderr="") monkeypatch.setattr("fixsub.sync.subprocess.run", fake_run) @@ -142,10 +137,46 @@ def fake_run(command: list[str], capture_output: bool, text: bool) -> subprocess audio_stream="a:4", ) - assert result.succeeded is False - assert result.error == "ffsubsync rejected a low-quality alignment" + assert len(calls) == 2 + assert "--skip-sync-on-low-quality" in calls[0] + assert "--skip-sync-on-low-quality" not in calls[1] + assert result.succeeded is True + assert result.output_path == output + assert result.forced_low_quality is True assert result.offset_seconds == 31.73 - assert result.framerate_scale == 1.043 + + +def test_run_ffsubsync_forced_retry_failure_removes_output( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + monkeypatch.setattr("fixsub.sync.shutil.which", lambda command: "/usr/bin/ffs") + output = tmp_path / "synced" / "candidate.srt" + calls = 0 + + def fake_run(command: list[str], capture_output: bool, text: bool) -> subprocess.CompletedProcess[str]: + nonlocal calls + calls += 1 + output.write_text("temporary", encoding="utf-8") + if calls == 1: + return subprocess.CompletedProcess( + command, + 0, + stdout=( + "score: 12.0\noffset seconds: 3.0\nframerate scale factor: 1.0\n" + "low-quality alignment; leaving subtitles unmodified\n" + ), + stderr="", + ) + return subprocess.CompletedProcess(command, 2, stdout="", stderr="forced failure") + + monkeypatch.setattr("fixsub.sync.subprocess.run", fake_run) + + result = run_ffsubsync(tmp_path / "movie.mkv", tmp_path / "candidate.srt", output, "a:0") + + assert calls == 2 + assert result.succeeded is False + assert result.error == "forced failure" + assert result.forced_low_quality is True assert not output.exists()