{
  "schema_version": "1.0",
  "query": {
    "title": "Example Feature",
    "original_title": "Example Feature",
    "year": 2026,
    "media_type": "movie",
    "requested_editions": ["director-cut", "extended"]
  },
  "edition_facts": [
    {
      "edition": "director-cut",
      "duration_minutes": 132,
      "source_url": "https://example.com/example-feature/director-cut",
      "confidence": "primary"
    },
    {
      "edition": "theatrical",
      "duration_minutes": 118,
      "source_url": "https://example.com/example-feature/theatrical",
      "confidence": "primary"
    }
  ],
  "candidates": [
    {
      "id": "compact-4k",
      "name": "Example.Feature.2026.Directors.Cut.2160p.HEVC.HDR.CHS.ENG",
      "uri": "magnet:?xt=urn:btih:1111111111111111111111111111111111111111",
      "info_hash": "1111111111111111111111111111111111111111",
      "source": "example-index-a",
      "source_url": "https://example.com/result/compact-4k",
      "size_bytes": 12884901888,
      "seeders": 48,
      "leechers": 6,
      "resolution": "2160p",
      "video_codec": "hevc",
      "hdr": "hdr10",
      "edition": "director-cut",
      "duration_minutes": 132,
      "audio_languages": ["en", "zh"],
      "subtitle_languages": ["zh-Hans", "en"],
      "subtitle_verified": true,
      "metadata_confidence": "verified",
      "trusted_source": true
    },
    {
      "id": "large-remux",
      "name": "Example.Feature.2026.Directors.Cut.2160p.REMUX.HEVC",
      "uri": "magnet:?xt=urn:btih:2222222222222222222222222222222222222222",
      "info_hash": "2222222222222222222222222222222222222222",
      "source": "example-index-b",
      "source_url": "https://example.com/result/large-remux",
      "size_bytes": 53687091200,
      "seeders": 16,
      "leechers": 2,
      "resolution": "2160p",
      "video_codec": "hevc",
      "edition": "director-cut",
      "duration_minutes": 132,
      "audio_languages": ["en"],
      "subtitle_languages": ["en"],
      "subtitle_verified": true,
      "metadata_confidence": "release-record",
      "trusted_source": true
    },
    {
      "id": "healthy-1080p",
      "name": "Example.Feature.2026.Theatrical.1080p.HEVC.CHS.ENG",
      "uri": "magnet:?xt=urn:btih:3333333333333333333333333333333333333333",
      "info_hash": "3333333333333333333333333333333333333333",
      "source": "example-index-c",
      "source_url": "https://example.com/result/healthy-1080p",
      "size_bytes": 7516192768,
      "seeders": 92,
      "leechers": 8,
      "resolution": "1080p",
      "video_codec": "hevc",
      "edition": "theatrical",
      "duration_minutes": 118,
      "audio_languages": ["en"],
      "subtitle_languages": ["zh-Hans", "en"],
      "subtitle_verified": true,
      "metadata_confidence": "verified",
      "trusted_source": true
    }
  ]
}

{
  "status": "Stopped",
  "total": 2,
  "results": [
    {
      "fileName": "Example.Feature.2026.Directors.Cut.2160p.HEVC.CHS.ENG",
      "fileSize": 12884901888,
      "fileUrl": "magnet:?xt=urn:btih:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
      "nbSeeders": 48,
      "nbLeechers": 6,
      "siteUrl": "example-index-a",
      "descrLink": "https://example.com/result/a"
    },
    {
      "fileName": "Example.Feature.2026.1080p.HEVC.CHS.ENG",
      "fileSize": 7516192768,
      "fileUrl": "magnet:?xt=urn:btih:BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB",
      "nbSeeders": 92,
      "nbLeechers": 8,
      "siteUrl": "example-index-b",
      "descrLink": "https://example.com/result/b"
    }
  ]
}

[
  {
    "title": "指环王三部曲：qBittorrent 慢源切换到 aria2 续传",
    "description": "用户只给出片名并偏好加长版、原声、合理体积和简中字幕；任务通过候选检索、版本排序、容量预检、qBittorrent 首轮测速、aria2 同输入续传和最终 ffprobe 验收完成。",
    "input": {
      "request": "指环王三部曲",
      "preferences": {
        "edition": "extended",
        "quality": "balanced",
        "audio": "original",
        "subtitles": ["zh-Hans", "en"]
      },
      "starting_state": "qBittorrent search plugins returned several candidates; the strongest candidate had a large advertised swarm but a weak sustained first probe."
    },
    "prompt": "$lov-media-fetch 指环王三部曲",
    "output": {
      "download_status": "complete",
      "release": "The Lord of the Rings Trilogy 2002 EXTENDED REMASTERED 1080p BluRay HEVC x265 5.1 BONE",
      "info_hash": "AB25E785ABEA95343249A300A516455D14E54144",
      "transport_trace": ["qBittorrent discovery", "qBittorrent first probe", "aria2 resumed same input"],
      "files": 3,
      "size_bytes": 13028499982,
      "verification_status": "passed_with_warnings",
      "media_observation": "three readable 1920x804 HEVC files with original English AAC 6-channel audio and embedded English SubRip subtitles",
      "subtitle_status": "missing",
      "open_item": "zh-Hans was absent from the verified embedded streams; the exact-release subtitle handoff remains a separate follow-up.",
      "evidence": [
        "cases/evidence/lotr-download.json",
        "cases/evidence/lotr-verification.json"
      ]
    }
  }
]

# Acquisition, Monitoring, and Recovery Policy

## Capacity calculation

Before transfer, require:

```text
selected payload
+ incomplete-file overhead (10%)
+ parallel probe count × probe budget
+ fallback continuation metadata and temporary control files
+ configured free-space reserve
```

Use the filesystem containing the closest existing parent of the destination. Report
bytes in both GiB and human-readable form. If the destination does not exist, capacity
belongs to its nearest existing parent.

## Probe policy

- Default concurrency: 3.
- Default duration: 180 seconds after a 60-second warm-up.
- Default temporary budget: 512 MiB per candidate.
- Default slow threshold: sustained speed below 1 MiB/s.
- Deduplicate by info hash before adding.
- Use a unique tag `media-fetch-<job-id>` and exact per-candidate directories.
- Existing hashes are read-only observations; never alter or clean them.
- Prefer aria2 for direct URLs, Metalinks, Magnets, and Torrent inputs. Its `.aria2`
  control file is evidence of continuation state, not a completed payload.
- Treat qBittorrent as optional. Enable it for reviewed search plugins, queue UI,
  deeper swarm inspection, or long-term seeding; do not make WebUI login a prerequisite
  for an aria2-capable run.
- Allocate distinct listen and RPC ports for concurrent aria2 probes.
- Configure DHT, PeX, LSD, and a bounded tracker set. Direct connections are preferred
  when proxy environment variables produce a slow or incomplete swarm. Record the
  selected backend and connection mode in the report.

A probe winner should combine sustained speed, availability, peers, progress, and ETA.
Prefer stability over a single peak sample.

## Active monitoring

- Poll the client every 5–15 seconds inside the worker.
- Surface a user update at least every 60 seconds while the agent is running tools.
- Track progress delta and received bytes, not only reported state.
- Treat metadata retrieval separately from payload speed.
- A candidate is stalled when both progress delta and download traffic remain below
  thresholds for the configured `stall_seconds` after metadata is available.

## Switching

1. Pause the stalled winner.
2. Resume the next candidate with a successful probe.
3. Recalculate storage if the next candidate is larger than the planned payload.
4. When the ranked list is exhausted, keep the best partial task paused, return to
   discovery, add a new wave, and probe again.
5. Record every switch and reason in the acquisition report.
6. When switching transports for the same input, preserve the isolated job directory,
   reuse partial data only when the backend can validate it, and never report two
   parallel full copies as one completed artifact.

## Cleanup

Cleanup starts only after the winner reports complete and the final file path exists.

- Resolve exact task hashes created in this job.
- Delete losing qBittorrent tasks using exact hashes only when that backend was used.
- Delete files only inside the resolved job probe directory.
- Preserve the winner, final destination, report JSON, and every pre-existing task.
- If path resolution falls outside the job directory, skip file cleanup and report it.

## Terminal states

- `complete`: client complete and local files present; ready for media verification.
- `needs_more_sources`: all tested candidates are below thresholds or unavailable.
- `capacity_shortfall`: no payload transfer started.
- `client_error`: connection or task API failure with copyable diagnostic details.
- `cancelled`: user-requested stop; created tasks are paused and exact state is reported.

## Backend routing

Run `scripts/aria2_acquire.py` first for supported inputs. Use a stable `job_id`,
`--watch` for bounded restarts, `--output-name` for opaque direct URLs, and a report
containing `backend_started`, `backend_restarted`, `last_snapshot`, and
`final_snapshot` events. Switch to qBittorrent only after an enabled probe supplies
measured evidence or the requested queue/seeding behavior requires it. The final report
must separate `download_status` from the later media and subtitle verdicts.

# Candidate and Decision Schema

All discovery adapters normalize into one UTF-8 JSON document. Unknown fields should
be omitted or set to `null`; do not invent zero values.

## Candidate manifest

```json
{
  "schema_version": "1.0",
  "query": {
    "title": "Example Title",
    "original_title": "Example Title",
    "year": 2026,
    "media_type": "movie",
    "season": null,
    "episodes": [],
      "requested_editions": ["director-cut", "extended"],
      "title_ambiguous": false,
      "edition_choice_required": false
  },
  "edition_facts": [
    {
      "edition": "director-cut",
      "duration_minutes": 132,
      "source_url": "https://example.com/release-record",
      "confidence": "primary"
    }
  ],
  "candidates": [
    {
      "id": "candidate-1",
      "name": "Example.Title.2026.Directors.Cut.2160p.HEVC",
      "uri": "magnet:?xt=urn:btih:0123456789ABCDEF0123456789ABCDEF01234567",
      "info_hash": "0123456789ABCDEF0123456789ABCDEF01234567",
      "transport_inputs": ["magnet"],
      "source": "adapter-name",
      "source_url": "https://example.com/result",
      "observed_at": "2026-08-08T00:00:00Z",
      "size_bytes": 12884901888,
      "seeders": 42,
      "leechers": 5,
      "resolution": "2160p",
      "video_codec": "hevc",
      "hdr": "hdr10",
      "edition": "director-cut",
      "duration_minutes": 132,
      "audio_languages": ["en", "zh"],
      "subtitle_languages": ["zh-Hans", "en"],
      "subtitle_verified": false,
      "source_health": {
        "advertised_seeders": 42,
        "observed_peers": null,
        "observed_at": "2026-08-08T00:00:00Z",
        "sustained_speed_bytes_per_second": null,
        "metadata_ready": false
      },
      "metadata_confidence": "filename",
      "trusted_source": false
    }
  ]
}
```

`source_url` may point to a description page while `uri` is the actual transfer input.
Never print private tracker query tokens in the user-facing result.

- `transport_inputs` may contain `magnet`, `torrent-file`, or `torrent-url`. Keep a
  Torrent URL or local path when `info_hash` is not available yet; fill the hash after
  metadata resolution instead of discarding the candidate.
- `source_health.advertised_seeders` comes from discovery. `observed_peers`,
  `metadata_ready`, and `sustained_speed_bytes_per_second` come from a live probe and
  must remain distinct.

## Normalization

- `edition`: `director-cut`, `extended`, `uncut`, `complete`, `theatrical`,
  `restored`, `regional`, or `unknown`.
- `resolution`: normalized vertical resolution such as `2160p`, `1080p`, `720p`.
- `video_codec`: `av1`, `hevc`, `h264`, `vp9`, or `unknown`.
- Language tags: prefer BCP 47 where available; normalize common `chs`/`zh-cn` to
  `zh-Hans`, `cht`/`zh-tw` to `zh-Hant`, and preserve `en`.
- `metadata_confidence`: `verified`, `release-record`, `filename`, or `unknown`.
- `info_hash`: uppercase hexadecimal when known. Deduplicate case-insensitively.
- `transport_inputs`: the inputs accepted by the acquisition layer; aria2 is the
  default backend, while optional qBittorrent may consume the same Torrent identity.

## Decision document

`scripts/rank_candidates.py` returns:

```json
{
  "schema_version": "1.0",
  "selected_id": "candidate-1",
  "choice_required": false,
  "reasons": [],
  "ranked": [
    {
      "candidate": {},
      "score": 118.4,
      "strengths": ["verified edition runtime", "efficient 2160p"],
      "warnings": []
    }
  ]
}
```

Acquisition must consume the complete `ranked` list so it retains tested fallbacks.

# Quality and Edition Policy

The default `balanced` mode maximizes useful viewing quality under a size guardrail.

## Decision order

1. Correct title, year, season, and episode.
2. Edition identity and runtime evidence.
3. Source/master credibility and absence of obvious corruption markers.
4. Useful picture quality and efficient codec.
5. Original audio plus requested additional tracks.
6. Verified Simplified Chinese and English subtitles.
7. Current source health and sustained acquisition prospects.
8. Size and storage fit.
9. Evidence quality: advertised health is provisional; observed speed and final
   stream inspection decide whether the release is actually ready.

## Edition rules

- A director's cut is a creative variant, not automatically a superset.
- Extended, uncut, complete, theatrical, restored, and regional cuts may differ in
  scenes, pacing, grading, dubbing, or censorship.
- Prefer the user's named edition. Without a named choice, prefer an evidenced
  director's/extended/uncut/complete release only when it is broadly a fuller version
  and does not replace a materially different creative decision.
- When two cuts are both credible and differ materially, require a user choice.
- Runtime within two minutes of a reliable release record is strong evidence; allow a
  larger tolerance only for frame-rate or logo/credits differences that are explained.

## Picture and size rules

Balanced movie defaults:

- Prefer credible 2160p HEVC/AV1 up to 24 GiB when it materially improves the source.
- Otherwise prefer 1080p HEVC/AV1, commonly 5–14 GiB for feature films.
- H.264 remains viable when source health or compatibility outweighs the size penalty.
- Remux/raw-disc files lose balanced-mode rank unless explicitly requested.
- A very small “4K” file receives a credibility penalty; resolution alone does not
  prove detail, bitrate, dynamic range, or source quality.
- HDR is valuable only when the display path and release metadata support it. Dolby
  Vision without a compatible fallback receives a compatibility warning.

For episodic content, apply the configured per-episode cap and then compute the pack's
aggregate size before storage preflight.

## Audio and subtitle rules

- Prefer original-language audio; additional Chinese or English audio is a bonus.
- Prefer embedded `zh-Hans` and `en` subtitles.
- `中字`, `双语`, `CHS`, or `ENG` in a filename is discovery evidence only.
- External subtitles should match exact release timing or be checked against duration
  and several scene boundaries before final acceptance.
- A missing `zh-Hans` stream is a warning on the media verdict until an exact-release
  SRT is matched and validated. Keep `subtitle_status` separate from `verification_status`.
- Image-based subtitles satisfy language coverage when the target player supports them;
  text subtitles are more portable and searchable.

## Automatic decision boundary

Require user choice when any of these remain after research:

- two plausible works share the title;
- top candidates represent materially different cuts and no preference resolves them;
- the first two candidates are within eight ranking points but trade resolution against
  edition, subtitles, or more than 35% size;
- the preferred edition has no runtime or release evidence;
- the only high-ranked candidate exceeds the explicit cap.

Otherwise auto-select and continue.

# Simplified Chinese Subtitle Handoff

Media Fetch owns the exact-release match and the final media report. The companion
`lov-subtitle-freedom-skill` is a subtitle operation, not a release identity oracle.
Use this boundary when the selected media has English subtitles but no `zh-Hans` track.

## Handoff contract

1. Match the subtitle to the exact release using file identity, runtime, frame rate,
   scene boundaries, or several synchronized cues. A language label alone is weak
   evidence.
2. Keep the original video and embedded tracks untouched. Save an external subtitle
   beside the media with a player-compatible name such as `TITLE.zh-Hans.srt`.
3. Preserve cue order, timestamps, and UTF-8 encoding. Validate that timestamps are
   monotonic, cues have text, and the subtitle duration is compatible with the media.
4. The default output is SRT. Learning glosses, character cards, and ASS styling remain
   opt-in operations from `lov-subtitle-freedom-skill`; they are not silently generated
   as a substitute for Simplified Chinese.
5. Report `subtitle_status` as `embedded`, `external_matched`, `missing`, or
   `generated_pending_review`. Keep it independent of the media's technical verdict.

## Evidence to retain

Record the source label, release match method, cue count, encoding, duration delta, and
validation result in the verification report. If an English track is extracted for a
subtitle operation, use a temporary UTF-8 SRT and never overwrite the source media.

# User Configuration

Media Fetch keeps portable preferences across runs while leaving passwords and private
tokens outside the profile.

## Resolution order

1. Current request or explicit CLI flag.
2. `MEDIA_FETCH_*` and `QBITTORRENT_*` environment variables.
3. `media_fetch` in the shared profile.
4. Defaults below.
5. One focused user question only when a remaining choice changes the output.

Shared profile:

```bash
${SKILL_PROFILE_PATH:-$HOME/.skill-publisher/skills/profile.json}
```

## Defaults

```json
{
  "media_fetch": {
    "config_version": 2,
    "output_dir": "$HOME/Downloads/Media",
    "quality_mode": "balanced",
    "max_movie_size_gib": 24,
    "max_episode_size_gib": 6,
    "reserve_free_gib": 15,
    "parallel_probes": 3,
    "probe_seconds": 180,
    "warmup_seconds": 60,
    "probe_budget_mib": 512,
    "slow_speed_mib_s": 1.0,
    "stall_seconds": 180,
    "max_search_waves": 3,
    "preferred_audio": ["original", "zh", "en"],
    "preferred_subtitles": ["zh-Hans", "en"],
    "preferred_editions": [
      "director-cut",
      "extended",
      "uncut",
      "complete",
      "theatrical"
    ],
    "transport_backends": ["aria2"],
    "aria2_binary": "aria2c",
    "aria2_listen_port": 0,
    "aria2_max_peers": 200,
    "aria2_max_restarts": 2,
    "subtitle_repair": "match-exact-release",
    "qbittorrent_url": "http://127.0.0.1:8080",
    "qbittorrent_username": "admin"
  }
}
```

The default size cap is a guardrail, not a target. Explicit per-request values win.
Series packs use the aggregate requested episode size when that is known.
Profiles without `config_version: 2` migrate the former `qbittorrent,aria2` ordering
to the aria2-only default and replace the fixed aria2 listen port with automatic port
selection. Explicit request and environment overrides still win.

## First-run initialization

1. Run `python3 scripts/media_config.py show`.
2. Show the resolved output directory, caps, language preferences, and probe policy.
3. If the user already requested different values, pass them as flags.
4. Run `python3 scripts/media_config.py init --write` only after the values are visible
   in the conversation.
5. Preserve all unrelated profile keys.

## Environment overrides

```bash
export MEDIA_FETCH_OUTPUT_DIR="$HOME/Downloads/Media"
export MEDIA_FETCH_MAX_MOVIE_SIZE_GIB="24"
export MEDIA_FETCH_MAX_EPISODE_SIZE_GIB="6"
export MEDIA_FETCH_RESERVE_FREE_GIB="15"
export MEDIA_FETCH_PARALLEL_PROBES="3"
export MEDIA_FETCH_SLOW_SPEED_MIB_S="1.0"
export MEDIA_FETCH_TRANSPORT_BACKENDS="aria2"
export MEDIA_FETCH_ARIA2_BIN="aria2c"
export MEDIA_FETCH_ARIA2_LISTEN_PORT="0"
export MEDIA_FETCH_ARIA2_MAX_PEERS="200"
export MEDIA_FETCH_ARIA2_MAX_RESTARTS="2"
export QBITTORRENT_URL="http://127.0.0.1:8080"
export QBITTORRENT_USERNAME="admin"
export QBITTORRENT_PASSWORD="read-from-a-secure-source"
```

Add `qbittorrent` to `MEDIA_FETCH_TRANSPORT_BACKENDS` only when its optional search,
queue, swarm-inspection, or seeding capabilities are wanted for the current run.

Do not place `QBITTORRENT_PASSWORD`, cookies, private tracker keys, or provider tokens
in committed Skill source or the shared profile.

#!/usr/bin/env python3
"""Run a resumable aria2 transfer for a direct URL, Magnet, or Torrent input."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import shutil
import signal
import socket
import subprocess
import sys
import time
import urllib.request
from urllib.parse import urlparse
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


VIDEO_SUFFIXES = {".mkv", ".mp4", ".m4v", ".mov", ".ts", ".m2ts", ".webm", ".avi"}
DEFAULT_TRACKERS = (
    "udp://tracker.opentrackr.org:1337/announce",
    "udp://open.stealth.si:80/announce",
    "udp://tracker.torrent.eu.org:451/announce",
    "udp://exodus.desync.com:6969/announce",
    "udp://tracker.cyberia.is:6969/announce",
    "udp://tracker.moeking.me:6969/announce",
    "udp://tracker1.bt.moack.co.kr:80/announce",
    "udp://tracker.tiny-vps.com:6969/announce",
    "http://tracker.openbittorrent.com:80/announce",
    "udp://tracker.openbittorrent.com:6969/announce",
    "udp://tracker.publictracker.xyz:6969/announce",
    "udp://tracker.dler.org:6969/announce",
)


def now() -> str:
    return datetime.now(timezone.utc).isoformat()


def media_files(root: Path) -> list[Path]:
    files: list[Path] = []
    for item in root.rglob("*"):
        if not item.is_file() or item.suffix.lower() not in VIDEO_SUFFIXES:
            continue
        try:
            item.stat()
        except FileNotFoundError:
            continue
        files.append(item)
    return sorted(files, key=lambda item: item.stat().st_size, reverse=True)


def control_files(root: Path) -> list[Path]:
    return sorted(item for item in root.rglob("*.aria2") if item.is_file())


def without_proxy(environment: dict[str, str]) -> dict[str, str]:
    return {key: value for key, value in environment.items() if "proxy" not in key.lower()}


def load_trackers(args: argparse.Namespace) -> list[str]:
    values = list(DEFAULT_TRACKERS)
    for tracker_file in args.tracker_file:
        path = Path(tracker_file).expanduser()
        values.extend(
            line.strip()
            for line in path.read_text(encoding="utf-8").splitlines()
            if line.strip() and not line.lstrip().startswith("#")
        )
    values.extend(args.tracker)
    return list(dict.fromkeys(values))


def resolve_binary(value: str | None) -> str:
    candidate = value or os.environ.get("MEDIA_FETCH_ARIA2_BIN") or "aria2c"
    expanded = str(Path(candidate).expanduser())
    resolved = shutil.which(expanded) or (expanded if Path(expanded).is_file() else None)
    if not resolved:
        raise SystemExit("ERROR: aria2c is required for the primary transfer backend")
    return resolved


def classify_input(value: str) -> str:
    lowered = value.lower()
    if lowered.startswith("magnet:"):
        return "bittorrent"
    parsed = urlparse(value)
    path = parsed.path if parsed.scheme else value
    if path.lower().endswith(".torrent"):
        return "bittorrent"
    return "direct"


def validate_output_name(value: str | None) -> str | None:
    if value is None:
        return None
    name = value.strip()
    if not name or Path(name).name != name or name in {".", ".."}:
        raise SystemExit("ERROR: --output-name must be a plain file name")
    return name


def port_is_available(port: int, *, udp: bool) -> bool:
    socket_type = socket.SOCK_DGRAM if udp else socket.SOCK_STREAM
    with socket.socket(socket.AF_INET, socket_type) as handle:
        try:
            handle.bind(("127.0.0.1", port))
        except OSError:
            return False
    return True


def find_port(seed: int, *, require_udp: bool, excluded: set[int]) -> int:
    for offset in range(20000):
        port = 40000 + ((seed - 40000 + offset) % 20000)
        if port in excluded:
            continue
        if port_is_available(port, udp=False) and (
            not require_udp or port_is_available(port, udp=True)
        ):
            return port
    raise SystemExit("ERROR: no available local port found for aria2")


def resolve_ports(
    job_id: str,
    input_kind: str,
    requested_listen: int,
    requested_rpc: int | None,
) -> tuple[int | None, int]:
    seed = 40000 + int(hashlib.sha1(job_id.encode("utf-8")).hexdigest()[:8], 16) % 20000
    listen_port: int | None = None
    if input_kind == "bittorrent":
        if requested_listen > 0:
            listen_port = requested_listen
        else:
            listen_port = find_port(seed, require_udp=True, excluded=set())
    if requested_rpc and requested_rpc > 0:
        rpc_port = requested_rpc
    else:
        rpc_port = find_port(
            seed + 1,
            require_udp=False,
            excluded={listen_port} if listen_port else set(),
        )
    return listen_port, rpc_port


def build_command(
    args: argparse.Namespace,
    probe_root: Path,
    binary: str,
    trackers: list[str],
    input_kind: str,
    listen_port: int | None,
    rpc_port: int,
) -> list[str]:
    command = [
        binary,
        f"--dir={probe_root}",
        "--file-allocation=none",
        "--continue=true",
        "--enable-rpc=true",
        "--rpc-listen-all=false",
        f"--rpc-listen-port={rpc_port}",
        "--max-connection-per-server=16",
        "--split=16",
        "--min-split-size=1M",
        "--max-tries=0",
        "--retry-wait=2",
        "--connect-timeout=15",
        "--timeout=60",
        f"--summary-interval={args.summary_interval}",
        "--console-log-level=notice",
        "--auto-file-renaming=false",
    ]
    if args.output_name:
        command.append(f"--out={args.output_name}")
    if input_kind == "bittorrent":
        if listen_port is None:
            raise ValueError("BitTorrent input requires a listen port")
        command.extend(
            [
                "--seed-time=0",
                "--enable-dht=true",
                "--bt-enable-lpd=true",
                "--enable-peer-exchange=true",
                f"--dht-listen-port={listen_port}",
                f"--listen-port={listen_port}",
                "--bt-tracker-connect-timeout=15",
                "--bt-tracker-interval=30",
                f"--bt-max-peers={args.max_peers}",
                "--bt-request-peer-speed-limit=1M",
                f"--bt-tracker={','.join(trackers)}",
            ]
        )
    command.append(args.input)
    return command


def rpc_request(rpc_port: int, method: str, params: list[Any]) -> Any:
    payload = json.dumps(
        {
            "jsonrpc": "2.0",
            "id": "media-fetch",
            "method": method,
            "params": params,
        }
    ).encode("utf-8")
    request = urllib.request.Request(
        f"http://127.0.0.1:{rpc_port}/jsonrpc",
        data=payload,
        headers={"Content-Type": "application/json"},
    )
    try:
        with urllib.request.urlopen(request, timeout=5) as response:
            document = json.loads(response.read().decode("utf-8"))
    except (OSError, ValueError, TypeError, json.JSONDecodeError):
        return None
    return document.get("result") if isinstance(document, dict) else None


def rpc_transfer(rpc_port: int) -> dict[str, Any] | None:
    keys = ["status", "completedLength", "totalLength", "downloadSpeed"]
    tasks = rpc_request(rpc_port, "aria2.tellActive", [keys])
    source = "active"
    if not isinstance(tasks, list) or not tasks:
        tasks = rpc_request(rpc_port, "aria2.tellStopped", [0, 10, keys])
        source = "stopped"
    if not isinstance(tasks, list) or not tasks:
        return None
    statuses = [str(item.get("status") or "") for item in tasks]
    return {
        "completed_bytes": sum(int(item.get("completedLength") or 0) for item in tasks),
        "total_bytes": sum(int(item.get("totalLength") or 0) for item in tasks),
        "download_speed_bytes_per_second": sum(
            int(item.get("downloadSpeed") or 0) for item in tasks
        ),
        "rpc_source": source,
        "rpc_statuses": statuses,
        "rpc_complete": source == "stopped" and all(status == "complete" for status in statuses),
    }


def snapshot(probe_root: Path) -> dict[str, Any]:
    files = media_files(probe_root)
    stats = [item.stat() for item in files]
    return {
        "media_files": len(files),
        "media_bytes_logical": sum(info.st_size for info in stats),
        "media_bytes_allocated": sum(
            info.st_size
            if getattr(info, "st_blocks", None) is None
            else int(info.st_blocks) * 512
            for info in stats
        ),
        "control_files": [str(item.relative_to(probe_root)) for item in control_files(probe_root)],
    }


def write_result(path: Path, result: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--input",
        required=True,
        help="HTTP(S) URL, Magnet URI, local .torrent, or .torrent URL",
    )
    parser.add_argument(
        "--output-name",
        help="Plain output file name for a direct URL whose path has no useful media suffix",
    )
    parser.add_argument("--output-dir", type=Path, default=Path.home() / "Downloads/Media")
    parser.add_argument("--result", required=True, type=Path)
    parser.add_argument("--job-id", required=True)
    parser.add_argument("--aria2-bin")
    parser.add_argument(
        "--listen-port",
        type=int,
        default=0,
        help="BitTorrent listen port; 0 selects a job-specific available port",
    )
    parser.add_argument("--rpc-listen-port", type=int)
    parser.add_argument("--max-peers", type=int, default=200)
    parser.add_argument("--summary-interval", type=int, default=15)
    parser.add_argument("--poll-seconds", type=int, default=15)
    parser.add_argument("--stall-seconds", type=int, default=180)
    parser.add_argument("--slow-speed-mib-s", type=float, default=1.0)
    parser.add_argument("--max-runtime-hours", type=float, default=48.0)
    parser.add_argument("--max-restarts", type=int, default=2)
    parser.add_argument("--restart-delay-seconds", type=int, default=15)
    parser.add_argument("--tracker", action="append", default=[])
    parser.add_argument("--tracker-file", action="append", default=[])
    parser.add_argument("--no-proxy", action="store_true")
    parser.add_argument("--watch", action="store_true", help="Restart a stalled aria2 process while its .aria2 state remains")
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()
    args.output_name = validate_output_name(args.output_name)

    output_dir = args.output_dir.expanduser().resolve(strict=False)
    job_id = "".join(
        character if character.isalnum() or character in "-_." else "_"
        for character in args.job_id
    ).strip("._") or "media-job"
    probe_root = output_dir / ".media-fetch-probes" / job_id
    probe_root.mkdir(parents=True, exist_ok=True)
    log_path = probe_root / "aria2.log"
    binary = resolve_binary(args.aria2_bin)
    trackers = load_trackers(args)
    input_kind = classify_input(args.input)
    listen_port, rpc_port = resolve_ports(
        job_id,
        input_kind,
        args.listen_port,
        args.rpc_listen_port,
    )
    command = build_command(
        args,
        probe_root,
        binary,
        trackers,
        input_kind,
        listen_port,
        rpc_port,
    )
    result: dict[str, Any] = {
        "schema_version": "1.0",
        "job_id": job_id,
        "backend": "aria2",
        "input_kind": input_kind,
        "input": args.input,
        "destination": str(output_dir),
        "probe_root": str(probe_root),
        "log_path": str(log_path),
        "started_at": now(),
        "status": "planned" if args.dry_run else "running",
        "trackers_count": len(trackers) if input_kind == "bittorrent" else 0,
        "listen_port": listen_port,
        "rpc_port": rpc_port,
        "events": [],
    }
    if args.dry_run:
        result["command"] = command
        result["environment_mode"] = "direct" if args.no_proxy else "inherited"
        result["download_status"] = result["status"]
        write_result(args.result, result)
        print(json.dumps(result, ensure_ascii=False, indent=2))
        return 0

    environment = without_proxy(dict(os.environ)) if args.no_proxy else dict(os.environ)
    deadline = time.monotonic() + max(0.1, args.max_runtime_hours) * 3600
    restarts = 0
    process: subprocess.Popen[Any] | None = None

    try:
        while time.monotonic() < deadline:
            stalled = False
            stall_reason = ""
            completed_via_rpc = False
            started_monotonic = time.monotonic()
            last_progress_monotonic = started_monotonic
            last_completed_bytes = 0
            with log_path.open("a", encoding="utf-8") as log:
                log.write(f"\n[{now()}] starting aria2: {' '.join(command)}\n")
                log.flush()
                process = subprocess.Popen(
                    command,
                    stdin=subprocess.DEVNULL,
                    stdout=log,
                    stderr=subprocess.STDOUT,
                    env=environment,
                )
                result["events"].append({"at": now(), "type": "backend_started", "pid": process.pid})
                write_result(args.result, result)
                while process.poll() is None and time.monotonic() < deadline:
                    state = snapshot(probe_root)
                    sample_monotonic = time.monotonic()
                    transfer = rpc_transfer(rpc_port) or {
                        "completed_bytes": last_completed_bytes,
                        "total_bytes": 0,
                        "download_speed_bytes_per_second": 0,
                        "rpc_source": "unavailable",
                        "rpc_statuses": [],
                        "rpc_complete": False,
                    }
                    completed_bytes = int(transfer["completed_bytes"])
                    delta_bytes = max(0, completed_bytes - last_completed_bytes)
                    observed_speed = int(transfer["download_speed_bytes_per_second"])
                    if delta_bytes > 0:
                        last_progress_monotonic = sample_monotonic
                        last_completed_bytes = completed_bytes
                    print(
                        json.dumps(
                            {
                                "event": "aria2_progress",
                                "elapsed_seconds": int(max(0, sample_monotonic - started_monotonic)),
                                "completed_bytes": completed_bytes,
                                "total_bytes": int(transfer["total_bytes"]),
                                "observed_speed_bytes_per_second": observed_speed,
                                "rpc_source": transfer["rpc_source"],
                                "rpc_statuses": transfer["rpc_statuses"],
                                **state,
                            },
                        ensure_ascii=False,
                        ),
                        flush=True,
                    )
                    has_payload = state["media_files"] > 0
                    if (
                        bool(transfer["rpc_complete"])
                        and has_payload
                        and not state["control_files"]
                    ):
                        completed_via_rpc = True
                        result["events"].append(
                            {
                                "at": now(),
                                "type": "backend_completed",
                                "completed_bytes": completed_bytes,
                            }
                        )
                        process.send_signal(signal.SIGTERM)
                        process.wait(timeout=30)
                        break
                    stalled_for = sample_monotonic - last_progress_monotonic
                    if (
                        args.watch
                        and has_payload
                        and stalled_for >= max(1, args.stall_seconds)
                        and observed_speed < max(0.0, args.slow_speed_mib_s) * 1024 * 1024
                    ):
                        stalled = True
                        stall_reason = (
                            f"no payload growth for {int(stalled_for)}s; "
                            f"observed speed {observed_speed} B/s"
                        )
                        result["events"].append(
                            {"at": now(), "type": "backend_stalled", "reason": stall_reason}
                        )
                        write_result(args.result, result)
                        process.send_signal(signal.SIGTERM)
                        process.wait(timeout=30)
                        break
                    time.sleep(max(1, args.poll_seconds))
            if process.poll() is None:
                process.send_signal(signal.SIGTERM)
                process.wait(timeout=30)
                result["status"] = "runtime_limit"
                break
            if completed_via_rpc:
                result["status"] = "complete"
                result["completed_at"] = now()
                result["last_snapshot"] = snapshot(probe_root)
                break
            if stalled:
                state = snapshot(probe_root)
                if restarts >= max(0, args.max_restarts):
                    result["status"] = "needs_more_sources"
                    result["last_snapshot"] = state
                    break
                restarts += 1
                result["events"].append(
                    {
                        "at": now(),
                        "type": "backend_restarted",
                        "restart": restarts,
                        "reason": stall_reason,
                    }
                )
                write_result(args.result, result)
                time.sleep(max(1, args.restart_delay_seconds))
                continue
            return_code = int(process.returncode or 0)
            state = snapshot(probe_root)
            result["last_exit_code"] = return_code
            result["last_snapshot"] = state
            if return_code == 0 and state["media_files"] > 0 and not state["control_files"]:
                result["status"] = "complete"
                result["completed_at"] = now()
                break
            if not args.watch or restarts >= max(0, args.max_restarts):
                result["status"] = "needs_more_sources" if state["control_files"] else "client_error"
                break
            restarts += 1
            result["events"].append({"at": now(), "type": "backend_restarted", "restart": restarts, "exit_code": return_code})
            write_result(args.result, result)
            time.sleep(max(1, args.restart_delay_seconds))
    except KeyboardInterrupt:
        if process and process.poll() is None:
            process.send_signal(signal.SIGTERM)
            process.wait(timeout=30)
        result["status"] = "cancelled"
    finally:
        result["finished_at"] = now()
        result["restarts"] = restarts
        result["final_snapshot"] = snapshot(probe_root)
        result["download_status"] = result["status"]
        write_result(args.result, result)

    print(json.dumps({"status": result["status"], "result": str(args.result)}, ensure_ascii=False, indent=2))
    return 0 if result["status"] == "complete" else 3


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, RuntimeError, ValueError, TypeError, json.JSONDecodeError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc

#!/usr/bin/env python3
"""Resolve or initialize portable Media Fetch preferences."""

from __future__ import annotations

import argparse
import json
import os
import sys
from copy import deepcopy
from pathlib import Path
from typing import Any


DEFAULTS: dict[str, Any] = {
    "config_version": 2,
    "output_dir": "$HOME/Downloads/Media",
    "quality_mode": "balanced",
    "max_movie_size_gib": 24.0,
    "max_episode_size_gib": 6.0,
    "reserve_free_gib": 15.0,
    "parallel_probes": 3,
    "probe_seconds": 180,
    "warmup_seconds": 60,
    "probe_budget_mib": 512,
    "slow_speed_mib_s": 1.0,
    "stall_seconds": 180,
    "max_search_waves": 3,
    "preferred_audio": ["original", "zh", "en"],
    "preferred_subtitles": ["zh-Hans", "en"],
    "preferred_editions": [
        "director-cut",
        "extended",
        "uncut",
        "complete",
        "theatrical",
    ],
    "transport_backends": ["aria2"],
    "aria2_binary": "aria2c",
    "aria2_listen_port": 0,
    "aria2_max_peers": 200,
    "aria2_max_restarts": 2,
    "subtitle_repair": "match-exact-release",
    "qbittorrent_url": "http://127.0.0.1:8080",
    "qbittorrent_username": "admin",
}

ENV_MAP = {
    "output_dir": "MEDIA_FETCH_OUTPUT_DIR",
    "max_movie_size_gib": "MEDIA_FETCH_MAX_MOVIE_SIZE_GIB",
    "max_episode_size_gib": "MEDIA_FETCH_MAX_EPISODE_SIZE_GIB",
    "reserve_free_gib": "MEDIA_FETCH_RESERVE_FREE_GIB",
    "parallel_probes": "MEDIA_FETCH_PARALLEL_PROBES",
    "slow_speed_mib_s": "MEDIA_FETCH_SLOW_SPEED_MIB_S",
    "transport_backends": "MEDIA_FETCH_TRANSPORT_BACKENDS",
    "aria2_binary": "MEDIA_FETCH_ARIA2_BIN",
    "aria2_listen_port": "MEDIA_FETCH_ARIA2_LISTEN_PORT",
    "aria2_max_peers": "MEDIA_FETCH_ARIA2_MAX_PEERS",
    "aria2_max_restarts": "MEDIA_FETCH_ARIA2_MAX_RESTARTS",
    "qbittorrent_url": "QBITTORRENT_URL",
    "qbittorrent_username": "QBITTORRENT_USERNAME",
}


def profile_path() -> Path:
    value = os.environ.get(
        "SKILL_PROFILE_PATH", str(Path.home() / ".skill-publisher/skills/profile.json")
    )
    return Path(os.path.expandvars(value)).expanduser()


def load_profile(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise SystemExit(f"ERROR: invalid JSON in {path}: {exc}") from exc
    if not isinstance(data, dict):
        raise SystemExit(f"ERROR: profile root must be an object: {path}")
    return data


def coerce(value: str, template: Any) -> Any:
    if isinstance(template, bool):
        return value.lower() in {"1", "true", "yes", "on"}
    if isinstance(template, int):
        return int(value)
    if isinstance(template, float):
        return float(value)
    if isinstance(template, list):
        return [item.strip() for item in value.split(",") if item.strip()]
    return value


def expand_value(key: str, value: Any) -> Any:
    if key.endswith("_dir") and isinstance(value, str):
        return str(Path(os.path.expandvars(value)).expanduser())
    return value


def resolve(profile: dict[str, Any], overrides: dict[str, Any]) -> dict[str, Any]:
    result = deepcopy(DEFAULTS)
    saved = profile.get("media_fetch")
    if isinstance(saved, dict):
        saved_values = {key: value for key, value in saved.items() if key in DEFAULTS}
        if int(saved_values.get("config_version") or 1) < 2:
            if saved_values.get("transport_backends") == ["qbittorrent", "aria2"]:
                saved_values["transport_backends"] = ["aria2"]
            if saved_values.get("aria2_listen_port") == 53555:
                saved_values["aria2_listen_port"] = 0
            saved_values["config_version"] = 2
        result.update(saved_values)
    for key, env_name in ENV_MAP.items():
        if env_name in os.environ:
            result[key] = coerce(os.environ[env_name], DEFAULTS[key])
    for key, value in overrides.items():
        if value is None:
            continue
        if key in DEFAULTS:
            value = coerce(value, DEFAULTS[key]) if isinstance(value, str) else value
        result[key] = value
    return {key: expand_value(key, value) for key, value in result.items()}


def parser() -> argparse.ArgumentParser:
    root = argparse.ArgumentParser(description=__doc__)
    sub = root.add_subparsers(dest="command", required=True)
    for name in ("show", "init"):
        cmd = sub.add_parser(name)
        cmd.add_argument("--output-dir")
        cmd.add_argument("--max-movie-size-gib", type=float)
        cmd.add_argument("--max-episode-size-gib", type=float)
        cmd.add_argument("--reserve-free-gib", type=float)
        cmd.add_argument("--parallel-probes", type=int)
        cmd.add_argument("--slow-speed-mib-s", type=float)
        cmd.add_argument("--transport-backends")
        cmd.add_argument("--aria2-binary")
        cmd.add_argument("--aria2-listen-port", type=int)
        cmd.add_argument("--aria2-max-peers", type=int)
        cmd.add_argument("--aria2-max-restarts", type=int)
        cmd.add_argument("--json", action="store_true")
        if name == "init":
            cmd.add_argument("--write", action="store_true")
    return root


def main() -> int:
    args = parser().parse_args()
    path = profile_path()
    profile = load_profile(path)
    overrides = {
        "output_dir": args.output_dir,
        "max_movie_size_gib": args.max_movie_size_gib,
        "max_episode_size_gib": args.max_episode_size_gib,
        "reserve_free_gib": args.reserve_free_gib,
        "parallel_probes": args.parallel_probes,
        "slow_speed_mib_s": args.slow_speed_mib_s,
        "transport_backends": args.transport_backends,
        "aria2_binary": args.aria2_binary,
        "aria2_listen_port": args.aria2_listen_port,
        "aria2_max_peers": args.aria2_max_peers,
        "aria2_max_restarts": args.aria2_max_restarts,
    }
    resolved = resolve(profile, overrides)

    wrote = False
    if args.command == "init" and args.write:
        path.parent.mkdir(parents=True, exist_ok=True)
        profile["media_fetch"] = resolved
        path.write_text(
            json.dumps(profile, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
        )
        wrote = True

    payload = {"profile_path": str(path), "resolved": resolved, "written": wrote}
    if args.json:
        print(json.dumps(payload, ensure_ascii=False, indent=2))
    else:
        print(f"profile={path}")
        print(f"written={'yes' if wrote else 'no'}")
        for key, value in resolved.items():
            print(f"{key}={json.dumps(value, ensure_ascii=False)}")
        if args.command == "init" and not args.write:
            print("next=review these values, then rerun with --write")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, ValueError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc

#!/usr/bin/env python3
"""Probe ranked Magnet candidates, monitor the winner, and recover from stalls."""

from __future__ import annotations

import argparse
import base64
import http.cookiejar
import json
import os
import re
import statistics
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


MIB = 1024**2
COMPLETE_STATES = {"uploading", "stalledup", "pausedup", "stoppedup", "forcedup"}


class QBitClient:
    def __init__(self, base_url: str, username: str, password: str) -> None:
        self.base_url = base_url.rstrip("/")
        jar = http.cookiejar.CookieJar()
        self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
        response = self.request(
            "/api/v2/auth/login", {"username": username, "password": password}, raw=True
        )
        if response.strip() not in {"", "Ok."}:
            raise RuntimeError(f"qBittorrent login failed: {response[:120]}")

    def request(
        self, path: str, data: dict[str, Any] | None = None, raw: bool = False
    ) -> Any:
        encoded = urllib.parse.urlencode(data).encode("utf-8") if data is not None else None
        request = urllib.request.Request(self.base_url + path, data=encoded)
        request.add_header("Referer", self.base_url)
        try:
            with self.opener.open(request, timeout=30) as response:
                text = response.read().decode("utf-8", errors="replace")
        except urllib.error.HTTPError as exc:
            detail = exc.read().decode("utf-8", errors="replace")
            raise RuntimeError(f"qBittorrent HTTP {exc.code} for {path}: {detail[:300]}") from exc
        if raw:
            return text
        return json.loads(text) if text else {}

    def torrents(self) -> list[dict[str, Any]]:
        payload = self.request("/api/v2/torrents/info")
        return payload if isinstance(payload, list) else []

    def add(self, uri: str, save_path: Path, tag: str) -> None:
        self.request(
            "/api/v2/torrents/add",
            {
                "urls": uri,
                "savepath": str(save_path),
                "tags": tag,
                "stopped": "false",
                "paused": "false",
            },
            raw=True,
        )

    def action(self, action: str, hashes: list[str]) -> None:
        if not hashes:
            return
        joined = "|".join(hashes)
        paths = [f"/api/v2/torrents/{action}"]
        if action == "start":
            paths.append("/api/v2/torrents/resume")
        elif action == "stop":
            paths.append("/api/v2/torrents/pause")
        last_error: RuntimeError | None = None
        for path in paths:
            try:
                self.request(path, {"hashes": joined}, raw=True)
                return
            except RuntimeError as exc:
                last_error = exc
        if last_error:
            raise last_error

    def set_location(self, hash_value: str, location: Path) -> None:
        self.request(
            "/api/v2/torrents/setLocation",
            {"hashes": hash_value, "location": str(location)},
            raw=True,
        )

    def delete(self, hash_value: str, delete_files: bool) -> None:
        self.request(
            "/api/v2/torrents/delete",
            {"hashes": hash_value, "deleteFiles": str(delete_files).lower()},
            raw=True,
        )


def now() -> str:
    return datetime.now(timezone.utc).isoformat()


def parse_info_hash(uri: str) -> str | None:
    match = re.search(r"(?:[?&]xt=urn:btih:)([A-Za-z0-9]+)", uri, re.I)
    if not match:
        return None
    value = match.group(1)
    if re.fullmatch(r"[0-9A-Fa-f]{40}", value):
        return value.lower()
    if re.fullmatch(r"[A-Z2-7]{32}", value.upper()):
        return base64.b32decode(value.upper()).hex().lower()
    return None


def load_decision(path: Path) -> dict[str, Any]:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise SystemExit(f"ERROR: cannot read decision JSON {path}: {exc}") from exc
    if not isinstance(data, dict) or not isinstance(data.get("ranked"), list):
        raise SystemExit("ERROR: decision document must contain a ranked array")
    return data


def controlled_candidates(
    decision: dict[str, Any], selected_id: str | None, limit: int
) -> list[dict[str, Any]]:
    selected = selected_id or decision.get("selected_id")
    if decision.get("choice_required") and not selected_id:
        reasons = "; ".join(str(value) for value in decision.get("reasons") or [])
        raise SystemExit(f"ERROR: choice_required; provide --selected-id ({reasons})")
    candidates: list[dict[str, Any]] = []
    for row in decision["ranked"]:
        candidate = row.get("candidate") if isinstance(row, dict) else None
        if not isinstance(candidate, dict):
            continue
        uri = str(candidate.get("uri") or "")
        hash_value = str(candidate.get("info_hash") or "").lower() or parse_info_hash(uri)
        if not uri.startswith("magnet:") or not hash_value:
            continue
        item = dict(candidate)
        item["info_hash"] = hash_value
        cap_gib = float(decision.get("max_size_gib") or 0)
        size_bytes = int(item.get("size_bytes") or 0)
        if cap_gib and size_bytes > cap_gib * 1024**3 and item.get("id") != selected:
            continue
        candidates.append(item)
    candidates.sort(key=lambda item: item.get("id") != selected)
    return candidates[: max(1, limit)]


def is_within(path_value: str, parent: Path) -> bool:
    try:
        Path(path_value).expanduser().resolve(strict=False).relative_to(parent.resolve(strict=False))
        return True
    except ValueError:
        return False


def task_map(client: QBitClient, hashes: set[str]) -> dict[str, dict[str, Any]]:
    return {
        str(task.get("hash", "")).lower(): task
        for task in client.torrents()
        if str(task.get("hash", "")).lower() in hashes
    }


def write_result(path: Path, result: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--decision", required=True, type=Path)
    parser.add_argument("--selected-id")
    parser.add_argument("--output-dir", type=Path, default=Path.home() / "Downloads/Media")
    parser.add_argument("--result", required=True, type=Path)
    parser.add_argument("--job-id")
    parser.add_argument("--parallel", type=int, default=3)
    parser.add_argument("--probe-seconds", type=int, default=180)
    parser.add_argument("--warmup-seconds", type=int, default=60)
    parser.add_argument("--probe-budget-mib", type=int, default=512)
    parser.add_argument("--poll-seconds", type=int, default=10)
    parser.add_argument("--slow-speed-mib-s", type=float, default=1.0)
    parser.add_argument("--stall-seconds", type=int, default=180)
    parser.add_argument("--wait-complete", action="store_true")
    parser.add_argument("--max-runtime-hours", type=float, default=48.0)
    parser.add_argument("--cleanup-losers", action="store_true")
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()

    decision = load_decision(args.decision)
    candidates = controlled_candidates(decision, args.selected_id, args.parallel)
    if not candidates:
        raise SystemExit("ERROR: no probeable Magnet candidates")
    output_dir = args.output_dir.expanduser().resolve(strict=False)
    job_id = args.job_id or uuid.uuid4().hex[:12]
    tag = f"media-fetch-{job_id}"
    probe_root = output_dir / ".media-fetch-probes" / job_id
    result: dict[str, Any] = {
        "schema_version": "1.0",
        "job_id": job_id,
        "tag": tag,
        "started_at": now(),
        "status": "planned" if args.dry_run else "probing",
        "destination": str(output_dir),
        "probe_root": str(probe_root),
        "candidate_ids": [item.get("id") for item in candidates],
        "events": [],
    }
    if args.dry_run:
        result["selected_id"] = candidates[0].get("id")
        result["planned_hashes"] = [item["info_hash"] for item in candidates]
        write_result(args.result, result)
        print(json.dumps(result, ensure_ascii=False, indent=2))
        return 0

    password = os.environ.get("QBITTORRENT_PASSWORD", "")
    if not password:
        raise SystemExit("ERROR: QBITTORRENT_PASSWORD is required")
    client = QBitClient(
        os.environ.get("QBITTORRENT_URL", "http://127.0.0.1:8080"),
        os.environ.get("QBITTORRENT_USERNAME", "admin"),
        password,
    )
    output_dir.mkdir(parents=True, exist_ok=True)
    probe_root.mkdir(parents=True, exist_ok=False)
    hashes = {item["info_hash"] for item in candidates}
    preexisting = set(task_map(client, hashes))
    created: list[dict[str, Any]] = []
    for candidate in candidates:
        hash_value = candidate["info_hash"]
        if hash_value in preexisting:
            result["events"].append(
                {"at": now(), "type": "preexisting_skipped", "hash": hash_value, "candidate_id": candidate.get("id")}
            )
            continue
        candidate_dir = probe_root / str(candidate.get("id"))
        candidate_dir.mkdir(parents=True, exist_ok=False)
        client.add(str(candidate["uri"]), candidate_dir, tag)
        created.append(candidate)
        result["events"].append(
            {"at": now(), "type": "added", "hash": hash_value, "candidate_id": candidate.get("id"), "path": str(candidate_dir)}
        )
    if not created:
        result["status"] = "client_error"
        result["error"] = "all candidate hashes already exist; existing tasks were left unchanged"
        write_result(args.result, result)
        print(json.dumps(result, ensure_ascii=False, indent=2))
        return 4

    created_hashes = {item["info_hash"] for item in created}
    ready_deadline = time.monotonic() + 60
    while time.monotonic() < ready_deadline:
        if created_hashes.issubset(task_map(client, created_hashes)):
            break
        time.sleep(2)
    metrics: dict[str, dict[str, Any]] = {
        item["info_hash"]: {
            "speeds": [],
            "warm_speeds": [],
            "peak_speed": 0,
            "probe_capped": False,
            "candidate": item,
        }
        for item in created
    }
    probe_started = time.monotonic()
    probe_end = probe_started + max(args.probe_seconds, args.warmup_seconds + 1)
    while time.monotonic() < probe_end:
        current = task_map(client, created_hashes)
        elapsed = time.monotonic() - probe_started
        for hash_value, metric in metrics.items():
            task = current.get(hash_value, {})
            speed = int(task.get("dlspeed") or 0)
            metric["peak_speed"] = max(metric["peak_speed"], speed)
            metric["availability"] = float(task.get("availability") or 0)
            metric["num_seeds"] = int(task.get("num_seeds") or 0)
            metric["progress"] = float(task.get("progress") or 0)
            metric["state"] = str(task.get("state") or "missing")
            downloaded = int(task.get("downloaded") or task.get("total_downloaded") or 0)
            metric["downloaded_bytes"] = downloaded
            if not metric["probe_capped"] and elapsed < args.warmup_seconds:
                metric["warm_speeds"].append(speed)
            elif not metric["probe_capped"]:
                metric["speeds"].append(speed)
            if (
                not metric["probe_capped"]
                and downloaded >= max(1, args.probe_budget_mib) * MIB
            ):
                client.action("stop", [hash_value])
                metric["probe_capped"] = True
        if int(elapsed) % 30 < max(1, args.poll_seconds):
            print(json.dumps({"event": "probe_progress", "elapsed_seconds": int(elapsed), "tasks": {key: {"speed_mib_s": round((value["speeds"][-1] if value["speeds"] else 0) / MIB, 2), "progress": round(float(value.get("progress", 0)) * 100, 2)} for key, value in metrics.items()}}, ensure_ascii=False), flush=True)
        time.sleep(max(1, args.poll_seconds))

    ranked_probe: list[dict[str, Any]] = []
    for hash_value, metric in metrics.items():
        speeds = metric.pop("speeds") or metric.pop("warm_speeds")
        metric.pop("warm_speeds", None)
        metric["average_speed"] = int(statistics.fmean(speeds)) if speeds else 0
        metric["median_speed"] = int(statistics.median(speeds)) if speeds else 0
        metric["hash"] = hash_value
        metric["candidate_id"] = metric["candidate"].get("id")
        ranked_probe.append(metric)
    ranked_probe.sort(
        key=lambda item: (
            item["median_speed"],
            item["average_speed"],
            float(item.get("availability", 0)),
            int(item.get("num_seeds", 0)),
        ),
        reverse=True,
    )
    result["probe_results"] = [
        {key: value for key, value in item.items() if key != "candidate"} for item in ranked_probe
    ]
    threshold = int(max(0.0, args.slow_speed_mib_s) * MIB)
    if not ranked_probe or ranked_probe[0]["median_speed"] < threshold:
        client.action("stop", list(created_hashes))
        result["status"] = "needs_more_sources"
        result["finished_at"] = now()
        write_result(args.result, result)
        print(json.dumps({"status": result["status"], "result": str(args.result)}, ensure_ascii=False, indent=2))
        return 3

    order = [item["hash"] for item in ranked_probe]
    client.action("stop", order)
    active_index = 0
    active_hash = order[active_index]
    client.action("start", [active_hash])
    result["selected_hash"] = active_hash
    result["selected_id"] = metrics[active_hash]["candidate"].get("id")
    result["status"] = "downloading"
    result["events"].append({"at": now(), "type": "winner_started", "hash": active_hash})
    write_result(args.result, result)
    if not args.wait_complete:
        print(json.dumps({"status": "downloading", "selected_id": result["selected_id"], "result": str(args.result)}, ensure_ascii=False, indent=2))
        return 0

    runtime_deadline = time.monotonic() + max(0.1, args.max_runtime_hours) * 3600
    last_progress = 0.0
    last_activity = time.monotonic()
    last_report = 0.0
    winner_task: dict[str, Any] = {}
    while time.monotonic() < runtime_deadline:
        current = task_map(client, created_hashes)
        task = current.get(active_hash, {})
        winner_task = task
        progress = float(task.get("progress") or 0)
        speed = int(task.get("dlspeed") or 0)
        state = str(task.get("state") or "").lower()
        if progress > last_progress + 0.000001 or speed >= threshold:
            last_activity = time.monotonic()
            last_progress = max(last_progress, progress)
        if progress >= 0.999999 or state in COMPLETE_STATES:
            result["status"] = "complete"
            break
        if time.monotonic() - last_activity >= max(30, args.stall_seconds):
            if active_index + 1 >= len(order):
                result["status"] = "needs_more_sources"
                result["events"].append({"at": now(), "type": "all_candidates_stalled", "hash": active_hash})
                client.action("stop", [active_hash])
                break
            client.action("stop", [active_hash])
            previous = active_hash
            active_index += 1
            active_hash = order[active_index]
            client.action("start", [active_hash])
            last_progress = 0.0
            last_activity = time.monotonic()
            result["events"].append({"at": now(), "type": "switched_after_stall", "from": previous, "to": active_hash})
            result["selected_hash"] = active_hash
            result["selected_id"] = metrics[active_hash]["candidate"].get("id")
            write_result(args.result, result)
        if time.monotonic() - last_report >= 60:
            print(json.dumps({"event": "download_progress", "candidate_id": result["selected_id"], "progress_percent": round(progress * 100, 2), "speed_mib_s": round(speed / MIB, 2), "state": state, "eta_seconds": int(task.get("eta") or 0)}, ensure_ascii=False), flush=True)
            last_report = time.monotonic()
        time.sleep(max(1, args.poll_seconds))
    else:
        result["status"] = "runtime_limit"

    if result["status"] == "complete":
        client.set_location(active_hash, output_dir)
        move_deadline = time.monotonic() + 120
        final_path = ""
        while time.monotonic() < move_deadline:
            moved = task_map(client, {active_hash}).get(active_hash, {})
            candidate_path = str(moved.get("content_path") or "")
            if candidate_path and is_within(candidate_path, output_dir) and Path(candidate_path).exists():
                final_path = candidate_path
                break
            time.sleep(2)
        if not final_path:
            result["status"] = "client_error"
            result["error"] = "download completed but final relocation was not confirmed within 120 seconds"
        else:
            result["final_path"] = final_path
            result["completed_at"] = now()
        if args.cleanup_losers:
            for hash_value in order:
                if hash_value == active_hash:
                    continue
                task = task_map(client, {hash_value}).get(hash_value, {})
                safe = is_within(str(task.get("save_path") or ""), probe_root)
                client.delete(hash_value, delete_files=safe)
                result["events"].append({"at": now(), "type": "loser_removed", "hash": hash_value, "files_deleted": safe})
    result["finished_at"] = now()
    write_result(args.result, result)
    print(json.dumps({"status": result["status"], "selected_id": result.get("selected_id"), "result": str(args.result)}, ensure_ascii=False, indent=2))
    return 0 if result["status"] == "complete" else 3


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, RuntimeError, ValueError, TypeError, json.JSONDecodeError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc

#!/usr/bin/env python3
"""Search enabled qBittorrent plugins and emit normalized Media Fetch candidates."""

from __future__ import annotations

import argparse
import base64
import http.cookiejar
import json
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


class QBitClient:
    def __init__(self, base_url: str, username: str, password: str) -> None:
        self.base_url = base_url.rstrip("/")
        jar = http.cookiejar.CookieJar()
        self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
        payload = self.request(
            "/api/v2/auth/login", {"username": username, "password": password}, raw=True
        )
        if payload.strip() not in {"", "Ok."}:
            raise RuntimeError(f"qBittorrent login failed: {payload[:120]}")

    def request(
        self, path: str, data: dict[str, Any] | None = None, raw: bool = False
    ) -> Any:
        encoded = None
        if data is not None:
            encoded = urllib.parse.urlencode(data).encode("utf-8")
        request = urllib.request.Request(self.base_url + path, data=encoded)
        request.add_header("Referer", self.base_url)
        try:
            with self.opener.open(request, timeout=20) as response:
                text = response.read().decode("utf-8", errors="replace")
        except urllib.error.HTTPError as exc:
            detail = exc.read().decode("utf-8", errors="replace")
            raise RuntimeError(f"qBittorrent HTTP {exc.code} for {path}: {detail[:300]}") from exc
        if raw:
            return text
        return json.loads(text) if text else {}


def info_hash(uri: str) -> str | None:
    match = re.search(r"(?:[?&]xt=urn:btih:)([A-Za-z0-9]+)", uri, re.I)
    if not match:
        return None
    value = match.group(1)
    if re.fullmatch(r"[0-9A-Fa-f]{40}", value):
        return value.upper()
    if re.fullmatch(r"[A-Z2-7]{32}", value.upper()):
        return base64.b32decode(value.upper()).hex().upper()
    return value.upper()


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--query", required=True)
    parser.add_argument("--output", required=True, type=Path)
    parser.add_argument("--plugins", default="all")
    parser.add_argument("--category", default="all")
    parser.add_argument("--timeout", type=int, default=90)
    parser.add_argument("--limit", type=int, default=200)
    parser.add_argument("--title")
    parser.add_argument("--year", type=int)
    parser.add_argument("--media-type", default="movie")
    parser.add_argument(
        "--results-json",
        type=Path,
        help="Normalize a saved qBittorrent search response without connecting",
    )
    args = parser.parse_args()

    if args.results_json:
        payload = json.loads(args.results_json.read_text(encoding="utf-8"))
        search_id = None
        status = "Fixture"
    else:
        base_url = os.environ.get("QBITTORRENT_URL", "http://127.0.0.1:8080")
        username = os.environ.get("QBITTORRENT_USERNAME", "admin")
        password = os.environ.get("QBITTORRENT_PASSWORD", "")
        if not password:
            raise SystemExit("ERROR: QBITTORRENT_PASSWORD is required")
        client = QBitClient(base_url, username, password)
        started = client.request(
            "/api/v2/search/start",
            {"pattern": args.query, "plugins": args.plugins, "category": args.category},
        )
        search_id = int(started.get("id"))
        deadline = time.monotonic() + max(5, args.timeout)
        status = "Running"
        try:
            while time.monotonic() < deadline:
                statuses = client.request(f"/api/v2/search/status?id={search_id}")
                row = statuses[0] if isinstance(statuses, list) and statuses else {}
                status = str(row.get("status") or "")
                if status.lower() not in {"running", "queued"}:
                    break
                time.sleep(2)
            client.request("/api/v2/search/stop", {"id": search_id}, raw=True)
            payload = client.request(
                f"/api/v2/search/results?id={search_id}&limit={max(1, args.limit)}&offset=0"
            )
        finally:
            try:
                client.request("/api/v2/search/delete", {"id": search_id}, raw=True)
            except RuntimeError:
                pass

    observed = datetime.now(timezone.utc).isoformat()
    candidates: list[dict[str, Any]] = []
    seen: set[str] = set()
    for index, row in enumerate(payload.get("results") or []):
        if not isinstance(row, dict):
            continue
        uri = str(row.get("fileUrl") or row.get("file_url") or "")
        if not uri:
            continue
        hash_value = info_hash(uri)
        key = hash_value or uri
        if key in seen:
            continue
        seen.add(key)
        uri_without_query = uri.split("?", 1)[0].lower()
        if uri.lower().startswith("magnet:"):
            transport_inputs = ["magnet"]
        elif uri_without_query.endswith(".torrent") and uri.lower().startswith(("http://", "https://")):
            transport_inputs = ["torrent-url"]
        elif uri_without_query.endswith(".torrent"):
            transport_inputs = ["torrent-file"]
        else:
            transport_inputs = ["url"]
        seeders = int(row.get("nbSeeders") or row.get("seeders") or 0)
        leechers = int(row.get("nbLeechers") or row.get("leechers") or 0)
        candidates.append(
            {
                "id": f"qbt-{index + 1}",
                "name": str(row.get("fileName") or row.get("file_name") or key),
                "uri": uri,
                "info_hash": hash_value,
                "transport_inputs": transport_inputs,
                "source": str(row.get("siteUrl") or row.get("site_url") or "qBittorrent plugin"),
                "source_url": str(row.get("descrLink") or row.get("descr_link") or ""),
                "observed_at": observed,
                "size_bytes": int(row.get("fileSize") or row.get("file_size") or 0),
                "seeders": seeders,
                "leechers": leechers,
                "source_health": {
                    "advertised_seeders": seeders,
                    "advertised_leechers": leechers,
                    "observed_peers": None,
                    "observed_at": observed,
                    "sustained_speed_bytes_per_second": None,
                    "metadata_ready": False,
                },
                "metadata_confidence": "filename",
                "trusted_source": False,
            }
        )
    document = {
        "schema_version": "1.0",
        "query": {
            "title": args.title or args.query,
            "year": args.year,
            "media_type": args.media_type,
            "requested_editions": [],
        },
        "search": {"pattern": args.query, "plugins": args.plugins, "status": status},
        "edition_facts": [],
        "candidates": candidates,
    }
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(document, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print(json.dumps({"search_id": search_id, "status": status, "candidates": len(candidates), "output": str(args.output)}, ensure_ascii=False, indent=2))
    return 0 if candidates else 3


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, RuntimeError, ValueError, TypeError, json.JSONDecodeError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc

#!/usr/bin/env python3
"""Rank normalized Media Fetch candidates and identify genuine user choices."""

from __future__ import annotations

import argparse
import json
import math
import re
import sys
from pathlib import Path
from typing import Any


GIB = 1024**3
EDITION_SCORES = {
    "director-cut": 30,
    "extended": 26,
    "uncut": 24,
    "complete": 22,
    "restored": 18,
    "theatrical": 12,
    "original": 12,
    "regional": 5,
    "unknown": 0,
}
RESOLUTION_SCORES = {"4320p": 18, "2160p": 36, "1080p": 31, "720p": 13}
CODEC_SCORES = {"av1": 16, "hevc": 15, "h265": 15, "h264": 8, "avc": 8, "vp9": 9}


def load_json(path: Path) -> dict[str, Any]:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise SystemExit(f"ERROR: cannot read candidate JSON {path}: {exc}") from exc
    if not isinstance(data, dict) or not isinstance(data.get("candidates"), list):
        raise SystemExit("ERROR: candidate document must contain a candidates array")
    return data


def infer(candidate: dict[str, Any]) -> dict[str, Any]:
    item = dict(candidate)
    name = str(item.get("name", ""))
    lower = name.lower().replace("_", ".").replace(" ", ".")
    if not item.get("resolution"):
        match = re.search(r"(?<!\d)(4320|2160|1080|720)p(?!\d)", lower)
        item["resolution"] = f"{match.group(1)}p" if match else "unknown"
    if not item.get("video_codec"):
        if re.search(r"\b(av1)\b", lower):
            item["video_codec"] = "av1"
        elif re.search(r"\b(hevc|h\.?265|x265)\b", lower):
            item["video_codec"] = "hevc"
        elif re.search(r"\b(avc|h\.?264|x264)\b", lower):
            item["video_codec"] = "h264"
        else:
            item["video_codec"] = "unknown"
    if not item.get("edition"):
        if re.search(r"director.?s?.?cut|directors.?cut", lower):
            item["edition"] = "director-cut"
        elif "extended" in lower:
            item["edition"] = "extended"
        elif re.search(r"uncut|unrated", lower):
            item["edition"] = "uncut"
        elif re.search(r"complete|完整版", lower):
            item["edition"] = "complete"
        elif re.search(r"theatrical|影院版", lower):
            item["edition"] = "theatrical"
        else:
            item["edition"] = "unknown"
    languages = set(item.get("subtitle_languages") or [])
    if re.search(r"\b(chs|zh-cn|zh-hans|中字|简中|双语)\b", lower):
        languages.add("zh-Hans")
    if re.search(r"\b(eng|english|双语)\b", lower):
        languages.add("en")
    item["subtitle_languages"] = sorted(languages)
    return item


def score_candidate(
    candidate: dict[str, Any], max_size_gib: float, requested_editions: list[str]
) -> dict[str, Any]:
    item = infer(candidate)
    score = 0.0
    strengths: list[str] = []
    warnings: list[str] = []

    edition = str(item.get("edition") or "unknown")
    edition_score = EDITION_SCORES.get(edition, 0)
    if requested_editions and edition in requested_editions:
        edition_score += max(0, 12 - requested_editions.index(edition) * 2)
        strengths.append(f"matches requested edition: {edition}")
    score += edition_score

    confidence = str(item.get("metadata_confidence") or "unknown")
    if confidence == "verified":
        score += 18
        strengths.append("verified release metadata")
    elif confidence == "release-record":
        score += 12
        strengths.append("release-record evidence")
    elif confidence == "filename":
        warnings.append("edition and stream claims rely on filename")
    else:
        score -= 4
        warnings.append("weak metadata evidence")

    resolution = str(item.get("resolution") or "unknown").lower()
    codec = str(item.get("video_codec") or "unknown").lower()
    score += RESOLUTION_SCORES.get(resolution, 0)
    score += CODEC_SCORES.get(codec, 0)
    if resolution == "2160p" and codec in {"hevc", "h265", "av1"}:
        strengths.append("efficient 2160p")
    elif resolution == "1080p" and codec in {"hevc", "h265", "av1"}:
        strengths.append("compact high-quality 1080p")

    size_bytes = int(item.get("size_bytes") or 0)
    size_gib = size_bytes / GIB if size_bytes else 0.0
    if size_bytes:
        if size_gib > max_size_gib:
            over = size_gib / max_size_gib
            score -= 55 + min(35, (over - 1) * 30)
            warnings.append(f"exceeds size cap ({size_gib:.1f} GiB > {max_size_gib:.1f} GiB)")
        elif resolution == "2160p" and size_gib < 2.5:
            score -= 20
            warnings.append("implausibly small for a typical 2160p feature")
        elif resolution == "1080p" and size_gib < 1.2:
            score -= 12
            warnings.append("very small for a typical 1080p feature")
        else:
            score += 10
            strengths.append(f"inside size cap at {size_gib:.1f} GiB")
        if "remux" in str(item.get("name", "")).lower() and size_gib > max_size_gib * 0.8:
            score -= 12
            warnings.append("large remux in balanced mode")
    else:
        score -= 10
        warnings.append("size unknown")

    subtitles = {str(value).lower() for value in item.get("subtitle_languages") or []}
    has_zh = bool(subtitles & {"zh-hans", "zh", "chs", "zh-cn"})
    has_en = bool(subtitles & {"en", "eng", "english"})
    if has_zh and has_en:
        score += 20
        strengths.append("Chinese and English subtitles advertised")
    elif has_zh or has_en:
        score += 7
        warnings.append("only one preferred subtitle language advertised")
    else:
        score -= 8
        warnings.append("preferred subtitles absent or unknown")
    if bool(item.get("subtitle_verified")):
        score += 8
        strengths.append("subtitle streams verified")

    if bool(item.get("trusted_source")):
        score += 8
        strengths.append("trusted source")
    seeders = max(0, int(item.get("seeders") or 0))
    score += min(18, math.log2(seeders + 1) * 2.4)
    if seeders >= 20:
        strengths.append(f"healthy seeder count ({seeders})")
    elif seeders == 0:
        warnings.append("no current seeders reported")

    return {
        "candidate": item,
        "score": round(score, 2),
        "strengths": strengths,
        "warnings": warnings,
    }


def choice_reasons(
    ranked: list[dict[str, Any]], cap: float, query: dict[str, Any]
) -> list[str]:
    if not ranked:
        return ["no viable candidates"]
    reasons: list[str] = []
    if bool(query.get("title_ambiguous")):
        reasons.append("multiple plausible works share the requested identity")
    if bool(query.get("edition_choice_required")):
        reasons.append("credible editions contain materially different content")
    first = ranked[0]
    candidate = first["candidate"]
    if str(candidate.get("edition") or "unknown") != "theatrical" and str(
        candidate.get("metadata_confidence") or "unknown"
    ) not in {"verified", "release-record"}:
        reasons.append("preferred edition lacks independent release evidence")
    size_gib = int(candidate.get("size_bytes") or 0) / GIB
    if size_gib > cap:
        reasons.append("top candidate exceeds the explicit size cap")
    if len(ranked) >= 2 and first["score"] - ranked[1]["score"] <= 8:
        second = ranked[1]["candidate"]
        edition_diff = candidate.get("edition") != second.get("edition")
        size_a = int(candidate.get("size_bytes") or 0)
        size_b = int(second.get("size_bytes") or 0)
        size_diff = max(size_a, size_b) / max(1, min(size_a, size_b)) if size_a and size_b else 1
        resolution_diff = candidate.get("resolution") != second.get("resolution")
        if edition_diff or resolution_diff or size_diff > 1.35:
            reasons.append("top candidates are close but trade edition, picture, or size")
    return reasons


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--input", required=True, type=Path)
    parser.add_argument("--output", required=True, type=Path)
    parser.add_argument("--max-size-gib", type=float)
    args = parser.parse_args()

    document = load_json(args.input)
    query = document.get("query") if isinstance(document.get("query"), dict) else {}
    media_type = str(query.get("media_type") or "movie")
    default_cap = 6.0 if media_type == "episode" else 24.0
    max_size_gib = args.max_size_gib or float(query.get("max_size_gib") or default_cap)
    requested = [str(value) for value in query.get("requested_editions") or []]
    seen: set[str] = set()
    ranked: list[dict[str, Any]] = []
    for index, raw in enumerate(document["candidates"]):
        if not isinstance(raw, dict):
            continue
        item = dict(raw)
        item.setdefault("id", f"candidate-{index + 1}")
        key = str(item.get("info_hash") or item.get("uri") or item["id"]).upper()
        if key in seen:
            continue
        seen.add(key)
        ranked.append(score_candidate(item, max_size_gib, requested))
    ranked.sort(key=lambda item: item["score"], reverse=True)
    reasons = choice_reasons(ranked, max_size_gib, query)
    result = {
        "schema_version": "1.0",
        "query": query,
        "max_size_gib": max_size_gib,
        "selected_id": ranked[0]["candidate"]["id"] if ranked else None,
        "choice_required": bool(reasons),
        "reasons": reasons,
        "ranked": ranked,
    }
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print(json.dumps({
        "selected_id": result["selected_id"],
        "choice_required": result["choice_required"],
        "reasons": reasons,
        "candidates": len(ranked),
        "output": str(args.output),
    }, ensure_ascii=False, indent=2))
    return 0 if ranked else 2


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, ValueError, TypeError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc

#!/usr/bin/env python3
"""Check destination capacity before Media Fetch starts payload transfer."""

from __future__ import annotations

import argparse
import json
import shutil
import sys
from pathlib import Path
from typing import Any


GIB = 1024**3
MIB = 1024**2


def read_decision(path: Path) -> dict[str, Any]:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise SystemExit(f"ERROR: cannot read decision JSON {path}: {exc}") from exc
    if not isinstance(data, dict) or not isinstance(data.get("ranked"), list):
        raise SystemExit("ERROR: decision document must contain a ranked array")
    return data


def candidate_for(decision: dict[str, Any], selected_id: str | None) -> dict[str, Any]:
    target = selected_id or decision.get("selected_id")
    for row in decision["ranked"]:
        if not isinstance(row, dict) or not isinstance(row.get("candidate"), dict):
            continue
        candidate = row["candidate"]
        if candidate.get("id") == target:
            return candidate
    raise SystemExit(f"ERROR: selected candidate not found: {target}")


def existing_parent(path: Path) -> Path:
    current = path.expanduser().resolve(strict=False)
    while not current.exists() and current != current.parent:
        current = current.parent
    if not current.exists():
        raise SystemExit(f"ERROR: no existing parent for destination: {path}")
    return current


def human(value: int) -> str:
    return f"{value / GIB:.2f} GiB"


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--decision", required=True, type=Path)
    parser.add_argument("--selected-id")
    parser.add_argument("--output-dir", type=Path, default=Path.home() / "Downloads/Media")
    parser.add_argument("--reserve-free-gib", type=float, default=15.0)
    parser.add_argument("--probe-count", type=int, default=3)
    parser.add_argument("--probe-budget-mib", type=int, default=512)
    parser.add_argument("--overhead-ratio", type=float, default=0.10)
    parser.add_argument("--json-output", type=Path)
    args = parser.parse_args()

    decision = read_decision(args.decision)
    candidate = candidate_for(decision, args.selected_id)
    payload_bytes = int(candidate.get("size_bytes") or 0)
    destination = args.output_dir.expanduser().resolve(strict=False)
    filesystem_path = existing_parent(destination)
    usage = shutil.disk_usage(filesystem_path)

    if payload_bytes <= 0:
        result = {
            "ok": False,
            "status": "size_unknown",
            "candidate_id": candidate.get("id"),
            "destination": str(destination),
            "filesystem_path": str(filesystem_path),
            "available_bytes": usage.free,
            "message": "candidate size is required before payload transfer",
        }
        exit_code = 2
    else:
        overhead = int(payload_bytes * max(0.0, args.overhead_ratio))
        probe = max(0, args.probe_count) * max(0, args.probe_budget_mib) * MIB
        reserve = int(max(0.0, args.reserve_free_gib) * GIB)
        required = payload_bytes + overhead + probe + reserve
        shortfall = max(0, required - usage.free)
        result = {
            "ok": shortfall == 0,
            "status": "ready" if shortfall == 0 else "capacity_shortfall",
            "candidate_id": candidate.get("id"),
            "destination": str(destination),
            "filesystem_path": str(filesystem_path),
            "payload_bytes": payload_bytes,
            "overhead_bytes": overhead,
            "probe_bytes": probe,
            "reserve_bytes": reserve,
            "required_bytes": required,
            "available_bytes": usage.free,
            "shortfall_bytes": shortfall,
            "required_human": human(required),
            "available_human": human(usage.free),
            "shortfall_human": human(shortfall),
        }
        exit_code = 0 if shortfall == 0 else 3

    text = json.dumps(result, ensure_ascii=False, indent=2)
    print(text)
    if args.json_output:
        args.json_output.parent.mkdir(parents=True, exist_ok=True)
        args.json_output.write_text(text + "\n", encoding="utf-8")
    return exit_code


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, ValueError, TypeError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc

#!/usr/bin/env python3
"""Regression tests for aria2-first command routing."""

from __future__ import annotations

import argparse
import unittest
from pathlib import Path
from unittest.mock import patch

import aria2_acquire


def command_args(input_value: str, output_name: str | None = None) -> argparse.Namespace:
    return argparse.Namespace(
        input=input_value,
        output_name=output_name,
        listen_port=0,
        max_peers=200,
        summary_interval=15,
    )


class Aria2RoutingTests(unittest.TestCase):
    def test_classifies_direct_and_bittorrent_inputs(self) -> None:
        self.assertEqual(aria2_acquire.classify_input("https://example.test/movie.mp4"), "direct")
        self.assertEqual(aria2_acquire.classify_input("magnet:?xt=urn:btih:abc"), "bittorrent")
        self.assertEqual(aria2_acquire.classify_input("https://example.test/file.torrent"), "bittorrent")

    def test_direct_command_uses_http_options_without_bt_listeners(self) -> None:
        args = command_args("https://example.test/download?id=1", "Movie.mp4")
        command = aria2_acquire.build_command(
            args,
            Path("/tmp/media-fetch-test"),
            "/usr/bin/aria2c",
            [],
            "direct",
            None,
            41001,
        )
        self.assertIn("--split=16", command)
        self.assertIn("--out=Movie.mp4", command)
        self.assertFalse(any(item.startswith("--listen-port=") for item in command))
        self.assertFalse(any(item.startswith("--bt-tracker=") for item in command))

    def test_bittorrent_command_enables_swarm_features(self) -> None:
        args = command_args("magnet:?xt=urn:btih:abc")
        command = aria2_acquire.build_command(
            args,
            Path("/tmp/media-fetch-test"),
            "/usr/bin/aria2c",
            ["udp://tracker.example:80/announce"],
            "bittorrent",
            41002,
            41003,
        )
        self.assertIn("--enable-dht=true", command)
        self.assertIn("--enable-peer-exchange=true", command)
        self.assertIn("--listen-port=41002", command)
        self.assertIn("--rpc-listen-port=41003", command)

    def test_auto_ports_are_separate(self) -> None:
        listen_port, rpc_port = aria2_acquire.resolve_ports("job-a", "bittorrent", 0, None)
        self.assertIsNotNone(listen_port)
        self.assertNotEqual(listen_port, rpc_port)

    def test_output_name_rejects_paths(self) -> None:
        with self.assertRaises(SystemExit):
            aria2_acquire.validate_output_name("nested/Movie.mp4")

    def test_rpc_uses_stopped_task_to_detect_completion(self) -> None:
        stopped = [
            {
                "status": "complete",
                "completedLength": "2048",
                "totalLength": "2048",
                "downloadSpeed": "0",
            }
        ]
        with patch.object(aria2_acquire, "rpc_request", side_effect=[[], stopped]):
            transfer = aria2_acquire.rpc_transfer(41003)
        self.assertIsNotNone(transfer)
        assert transfer is not None
        self.assertTrue(transfer["rpc_complete"])
        self.assertEqual(transfer["rpc_source"], "stopped")
        self.assertEqual(transfer["completed_bytes"], 2048)


if __name__ == "__main__":
    unittest.main()

#!/usr/bin/env python3
"""Validate a portable local Skill Publisher Skill source directory."""

from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path
from typing import Any, Iterable

try:
    import yaml
except ImportError:
    print(
        "ERROR: PyYAML is required. Install it with: python3 -m pip install PyYAML",
        file=sys.stderr,
    )
    raise SystemExit(2)


FRONTMATTER_KEYS = {
    "name", "description", "license", "compatibility", "allowed-tools", "metadata"
}
TEXT_SUFFIXES = {".md", ".json", ".yaml", ".yml", ".txt", ".svg", ".py"}
JUNK_NAMES = {"__pycache__", ".DS_Store"}
JUNK_SUFFIXES = {".pyc", ".pyo"}
SKIP_DIRS = {".git", "dist", ".venv", "venv", "node_modules"}
SEMVER_RE = re.compile(r"^\d+\.\d+\.\d+(?:-[0-9A-Za-z.-]+)?$")
NAME_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
MARKDOWN_LINK_RE = re.compile(r"!?\[[^\]]*]\(([^)]+)\)")
SKILL_PATH_RE = re.compile(r"\$(SKILL_DIR|KIT_DIR)/([A-Za-z0-9_./-]+)")
CARD_STANDARD = "lovstudio/skill-card/v1"
PRICING_CARD_SCHEMA = "lovstudio/pricing-card/v1"


class ValidationFailure(Exception):
    """Raised when source metadata cannot be parsed."""


def read_text(path: Path) -> str:
    return path.read_text(encoding="utf-8")


def compact_text(value: Any) -> str:
    return re.sub(r"\s+", " ", value).strip() if isinstance(value, str) else ""


def split_frontmatter(path: Path) -> tuple[dict[str, Any], str]:
    text = read_text(path)
    if not text.startswith("---\n"):
        raise ValidationFailure(f"{path}: missing YAML frontmatter")
    marker = text.find("\n---\n", 4)
    if marker < 0:
        raise ValidationFailure(f"{path}: frontmatter is not closed")
    try:
        data = yaml.safe_load(text[4:marker])
    except yaml.YAMLError as exc:
        raise ValidationFailure(
            f"{path}: standard YAML parser rejected frontmatter: {exc}"
        ) from exc
    if not isinstance(data, dict):
        raise ValidationFailure(f"{path}: frontmatter must be a mapping")
    return data, text[marker + 5 :]


def iter_files(root: Path) -> Iterable[Path]:
    for path in root.rglob("*"):
        if any(part in SKIP_DIRS for part in path.relative_to(root).parts):
            continue
        if path.is_file():
            yield path


def is_relative_to(path: Path, parent: Path) -> bool:
    try:
        path.relative_to(parent)
        return True
    except ValueError:
        return False


def validate_skill_file(path: Path, errors: list[str]) -> dict[str, Any] | None:
    try:
        data, body = split_frontmatter(path)
    except ValidationFailure as exc:
        errors.append(str(exc))
        return None

    unexpected = sorted(set(data) - FRONTMATTER_KEYS)
    if unexpected:
        errors.append(f"{path}: unsupported frontmatter keys: {', '.join(unexpected)}")

    name = compact_text(data.get("name"))
    if not NAME_RE.fullmatch(name) or len(name) > 64:
        errors.append(f"{path}: name must be kebab-case and at most 64 characters")

    description = compact_text(data.get("description"))
    if not 50 <= len(description) <= 200:
        errors.append(
            f"{path}: description must contain 50-200 characters "
            f"(found {len(description)})"
        )

    metadata = data.get("metadata")
    if not isinstance(metadata, dict):
        errors.append(f"{path}: metadata must be a mapping")
    else:
        if not compact_text(metadata.get("author")):
            errors.append(f"{path}: metadata.author is required")
        if not SEMVER_RE.fullmatch(compact_text(metadata.get("version"))):
            errors.append(f"{path}: metadata.version must use SemVer")
        tags = metadata.get("tags")
        if not isinstance(tags, list) or not tags or not all(
            isinstance(tag, str) and tag.strip() for tag in tags
        ):
            errors.append(f"{path}: metadata.tags must be a non-empty list")
        dependencies = metadata.get("dependencies", [])
        if not isinstance(dependencies, list):
            errors.append(f"{path}: metadata.dependencies must be a list")

    trigger_block = re.search(
        r"(?ms)^##\s+Triggers\s*$([\s\S]*?)(?=^##\s+|\Z)", body
    )
    if not trigger_block:
        errors.append(f"{path}: add an explicit '## Triggers' section")
    else:
        block = trigger_block.group(1)
        if len(re.findall(r"(?m)^\s*-\s+\S", block)) < 3:
            errors.append(f"{path}: add two activation examples and one non-trigger")
        if not re.search(r"[\u3400-\u9fff]", block):
            errors.append(f"{path}: add a concrete Chinese trigger phrase")
        if not re.search(r"(?i)\b(?:the|a|an|create|build|help|publish|review|use)\b", block):
            errors.append(f"{path}: add a concrete English trigger phrase")
    if not re.search(
        r"(?mi)^###\s+(?:Do not activate when|Non-triggers?|不应触发|不要触发)\s*$",
        body,
    ):
        errors.append(f"{path}: add explicit non-trigger conditions")
    if len(read_text(path).splitlines()) >= 500:
        errors.append(f"{path}: keep SKILL.md below 500 lines")
    if not body.strip():
        errors.append(f"{path}: body is empty")
    return data


def load_yaml(path: Path, errors: list[str]) -> dict[str, Any] | None:
    try:
        data = yaml.safe_load(read_text(path))
    except yaml.YAMLError as exc:
        errors.append(f"{path}: standard YAML parser rejected file: {exc}")
        return None
    if not isinstance(data, dict):
        errors.append(f"{path}: expected a YAML mapping")
        return None
    return data


def has_content(value: Any) -> bool:
    if isinstance(value, str):
        return bool(value.strip())
    if isinstance(value, list):
        return any(has_content(item) for item in value)
    if isinstance(value, dict):
        return any(has_content(item) for item in value.values())
    return value is not None


def contains_placeholder(value: Any) -> bool:
    if isinstance(value, str):
        return bool(re.search(r"\bTODO\b|\{[^}]+\}", value, re.I))
    if isinstance(value, list):
        return any(contains_placeholder(item) for item in value)
    if isinstance(value, dict):
        return any(contains_placeholder(item) for item in value.values())
    return False


def validate_card_bundle(skill_root: Path, errors: list[str]) -> None:
    card_path = skill_root / "skill-card.yaml"
    card_doc_path = skill_root / "skill-card.md"
    cases_path = skill_root / "cases" / "cases.json"
    pricing_path = skill_root / "pricing-card.yaml"
    for path in (card_path, card_doc_path, cases_path, pricing_path):
        if not path.is_file():
            errors.append(f"{path}: required Skill trust-bundle file is missing")

    card = load_yaml(card_path, errors) if card_path.is_file() else None
    if card is not None:
        if card.get("schema") != CARD_STANDARD:
            errors.append(f"{card_path}: schema must be {CARD_STANDARD}")
        required = (
            "description", "owner", "license", "use_case", "deployment",
            "requirements", "risks", "references", "output", "version",
            "ethical_considerations", "dimensions", "pricing", "distribution",
        )
        for key in required:
            if key not in card or not has_content(card.get(key)):
                errors.append(f"{card_path}: required field '{key}' is missing or empty")
        dimensions = card.get("dimensions")
        if not isinstance(dimensions, list) or len(dimensions) < 3:
            errors.append(f"{card_path}: dimensions need at least three entries")
        else:
            ids: set[str] = set()
            for index, dimension in enumerate(dimensions):
                label = f"{card_path}: dimensions[{index}]"
                if not isinstance(dimension, dict):
                    errors.append(f"{label}: expected a mapping")
                    continue
                dimension_id = compact_text(dimension.get("id"))
                if not dimension_id or dimension_id in ids:
                    errors.append(f"{label}: id is required and unique")
                ids.add(dimension_id)
                for key in ("label", "description", "evidence"):
                    if not compact_text(dimension.get(key)):
                        errors.append(f"{label}: '{key}' is required")
        risks = card.get("risks")
        if not isinstance(risks, list) or not risks:
            errors.append(f"{card_path}: risks need risk and mitigation entries")
        else:
            for index, risk in enumerate(risks):
                if not isinstance(risk, dict) or not compact_text(risk.get("risk")) or not compact_text(risk.get("mitigation")):
                    errors.append(f"{card_path}: risks[{index}] needs risk and mitigation")
        distribution = card.get("distribution")
        if not isinstance(distribution, dict) or not isinstance(distribution.get("paid"), list) or not isinstance(distribution.get("free"), list):
            errors.append(f"{card_path}: distribution needs paid and free lists")
        if contains_placeholder(card):
            errors.append(f"{card_path}: unresolved placeholder")

    if card_doc_path.is_file():
        card_doc = read_text(card_doc_path)
        headings = (
            "Description", "Owner", "License", "Use Case", "Deployment Geography",
            "Requirements", "Known Risks", "References", "Skill Output",
            "Skill Version", "Ethical Considerations", "User Cases",
            "Dimension Map", "Pricing Basis", "Distribution",
        )
        for heading in headings:
            if not re.search(rf"(?mi)^#+\s+{re.escape(heading)}", card_doc):
                errors.append(f"{card_doc_path}: missing '{heading}' section")
        if re.search(r"\bTODO\b|\{[^}]+\}", card_doc, re.I):
            errors.append(f"{card_doc_path}: unresolved placeholder")

    if cases_path.is_file():
        try:
            cases = json.loads(read_text(cases_path))
        except json.JSONDecodeError as exc:
            errors.append(f"{cases_path}: invalid JSON: {exc}")
            cases = []
        if not isinstance(cases, list) or not cases:
            errors.append(f"{cases_path}: at least one user case is required")
        else:
            for index, case in enumerate(cases):
                label = f"{cases_path}: cases[{index}]"
                if not isinstance(case, dict):
                    errors.append(f"{label}: expected a mapping")
                    continue
                for key in ("title", "description", "input", "prompt", "output"):
                    if not has_content(case.get(key)):
                        errors.append(f"{label}: '{key}' is required")
                if contains_placeholder(case):
                    errors.append(f"{label}: unresolved placeholder")

    pricing = load_yaml(pricing_path, errors) if pricing_path.is_file() else None
    if pricing is not None:
        if pricing.get("schema") != PRICING_CARD_SCHEMA:
            errors.append(f"{pricing_path}: schema must be {PRICING_CARD_SCHEMA}")
        for key in ("model", "currency", "list_price_cny", "basis", "boundary", "review_trigger", "confidence"):
            if key not in pricing or (key != "list_price_cny" and not has_content(pricing.get(key))):
                errors.append(f"{pricing_path}: required field '{key}' is missing or empty")
        if contains_placeholder(pricing):
            errors.append(f"{pricing_path}: unresolved placeholder")


def validate_kit(root: Path, skill_names: set[str], errors: list[str]) -> None:
    manifest = root / "kit.yaml"
    if not manifest.exists():
        return
    data = load_yaml(manifest, errors)
    if data is None:
        return
    modules = data.get("modules")
    if not isinstance(modules, list) or not modules:
        errors.append(f"{manifest}: modules must be a non-empty list")
        return
    module_ids: set[str] = set()
    for index, module in enumerate(modules):
        label = f"{manifest}: modules[{index}]"
        if not isinstance(module, dict):
            errors.append(f"{label}: expected a mapping")
            continue
        module_id = compact_text(module.get("id"))
        skill_name = compact_text(module.get("skill"))
        relative = compact_text(module.get("path"))
        if not module_id or module_id in module_ids:
            errors.append(f"{label}: id is required and must be unique")
        module_ids.add(module_id)
        module_path = (root / relative).resolve()
        if (
            not relative
            or not is_relative_to(module_path, root.resolve())
            or not (module_path / "SKILL.md").is_file()
        ):
            errors.append(f"{label}: missing module at '{relative}/SKILL.md'")
        if skill_name not in skill_names:
            errors.append(f"{label}: unresolved child skill '{skill_name}'")
    pipelines = data.get("pipelines")
    if not isinstance(pipelines, dict) or not pipelines:
        errors.append(f"{manifest}: pipelines must be a non-empty mapping")
        return
    for pipeline, sequence in pipelines.items():
        if not isinstance(sequence, list) or not sequence:
            errors.append(f"{manifest}: pipeline '{pipeline}' must be a non-empty list")
            continue
        missing = [str(item) for item in sequence if item not in module_ids]
        if missing:
            errors.append(
                f"{manifest}: pipeline '{pipeline}' has unknown modules: "
                + ", ".join(missing)
            )


def validate_local_references(root: Path, errors: list[str]) -> None:
    for path in iter_files(root):
        if path.suffix.lower() != ".md":
            continue
        text = read_text(path)
        for raw in MARKDOWN_LINK_RE.findall(text):
            target = raw.strip().split(maxsplit=1)[0].strip("<>").split("#", 1)[0]
            if (
                not target
                or re.match(r"^[a-z][a-z0-9+.-]*:", target, re.I)
                or any(token in target for token in ("TODO", "{", "}"))
            ):
                continue
            if not (path.parent / target).resolve().exists():
                errors.append(f"{path}: broken local link '{target}'")
        skill_root = path.parent if path.name == "SKILL.md" else root
        for variable, target in SKILL_PATH_RE.findall(text):
            if "TODO" in target:
                continue
            base = skill_root if variable == "SKILL_DIR" else root
            resolved = (base / target.rstrip(".,;:)")).resolve()
            if not is_relative_to(resolved, root.resolve()) or not resolved.exists():
                errors.append(f"{path}: missing required resource '${variable}/{target}'")


def validate_hygiene(root: Path, errors: list[str]) -> None:
    private_path = re.compile(r"(?:/Users/[^/\s]+/|[A-Za-z]:\\\\Users\\\\[^\\\s]+\\\\)")
    for path in root.rglob("*"):
        if any(part in SKIP_DIRS for part in path.relative_to(root).parts):
            continue
        if path.name in JUNK_NAMES or path.suffix.lower() in JUNK_SUFFIXES:
            errors.append(f"{path}: generated/cache artifact must not ship")
    for path in iter_files(root):
        if path.suffix.lower() not in TEXT_SUFFIXES or path.name == "validate_skill.py":
            continue
        text = read_text(path)
        if private_path.search(text):
            errors.append(f"{path}: contains a private absolute user path")
        if path.name != "init_skill.py" and re.search(r"\bTODO\s*[:：]", text):
            errors.append(f"{path}: unresolved TODO placeholder")
    for relative in ("workbuddy", "scripts/build_workbuddy.py"):
        if (root / relative).exists():
            errors.append(
                f"{root / relative}: platform distribution artifacts belong to skill-publish"
            )


def validate_source(root: Path, errors: list[str]) -> None:
    root_skill = root / "SKILL.md"
    skill_files = [root_skill, *sorted((root / "skills").glob("*/SKILL.md"))]
    if not root_skill.is_file():
        errors.append(f"{root_skill}: file is required")
        return
    parsed: list[tuple[Path, dict[str, Any]]] = []
    for path in skill_files:
        data = validate_skill_file(path, errors)
        if data:
            parsed.append((path, data))
    names = {compact_text(data.get("name")) for _, data in parsed}
    if len(names) != len(parsed):
        errors.append(f"{root}: every embedded Skill must have a unique name")
    for path, data in parsed:
        metadata = data.get("metadata")
        if isinstance(metadata, dict) and metadata.get("card_standard") == CARD_STANDARD:
            validate_card_bundle(path.parent, errors)
    validate_kit(root, names, errors)

    readme = root / "README.md"
    if not readme.is_file():
        errors.append(f"{readme}: file is required")
    elif parsed:
        metadata = parsed[0][1].get("metadata")
        version = compact_text(metadata.get("version")) if isinstance(metadata, dict) else ""
        if version and f"version-{version}-" not in read_text(readme):
            errors.append(f"{readme}: version badge must match {version}")

    validate_hygiene(root, errors)
    validate_local_references(root, errors)


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("path", type=Path, help="Local Skill source directory")
    args = parser.parse_args()
    root = args.path.expanduser().resolve()
    if not root.is_dir():
        print(f"ERROR: directory does not exist: {root}", file=sys.stderr)
        return 2
    errors: list[str] = []
    validate_source(root, errors)
    if errors:
        print(f"FAILED: {len(errors)} issue(s)")
        for error in errors:
            print(f"- {error}")
        return 1
    print(f"PASSED: source validation ({root})")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Inspect completed media with ffprobe and emit a structured verification report."""

from __future__ import annotations

import argparse
import json
import re
import shutil
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


VIDEO_SUFFIXES = {".mkv", ".mp4", ".m4v", ".mov", ".ts", ".m2ts", ".webm", ".avi"}
PARTIAL_SUFFIXES = {".part", ".partial", ".crdownload", ".tmp", ".!qB".lower()}


def media_files(path: Path) -> list[Path]:
    if path.is_file():
        return [path] if path.suffix.lower() in VIDEO_SUFFIXES else []
    return sorted(
        (item for item in path.rglob("*") if item.is_file() and item.suffix.lower() in VIDEO_SUFFIXES),
        key=lambda item: item.stat().st_size,
        reverse=True,
    )


def partial_files(path: Path) -> list[str]:
    root = path if path.is_dir() else path.parent
    return [
        str(item)
        for item in root.rglob("*")
        if item.is_file() and item.suffix.lower() in PARTIAL_SUFFIXES
    ]


def ffprobe(path: Path) -> dict[str, Any]:
    command = [
        "ffprobe",
        "-v",
        "error",
        "-show_format",
        "-show_streams",
        "-of",
        "json",
        str(path),
    ]
    completed = subprocess.run(command, check=False, capture_output=True, text=True)
    if completed.returncode != 0:
        raise RuntimeError(f"ffprobe failed for {path}: {completed.stderr.strip()[:500]}")
    return json.loads(completed.stdout)


def language(stream: dict[str, Any]) -> str:
    tags = stream.get("tags") if isinstance(stream.get("tags"), dict) else {}
    return str(tags.get("language") or "und")


def inspect(path: Path) -> dict[str, Any]:
    probe = ffprobe(path)
    streams = probe.get("streams") if isinstance(probe.get("streams"), list) else []
    videos = [item for item in streams if item.get("codec_type") == "video"]
    audios = [item for item in streams if item.get("codec_type") == "audio"]
    subtitles = [item for item in streams if item.get("codec_type") == "subtitle"]
    format_data = probe.get("format") if isinstance(probe.get("format"), dict) else {}
    duration = float(format_data.get("duration") or 0)
    primary = videos[0] if videos else {}
    tags = primary.get("tags") if isinstance(primary.get("tags"), dict) else {}
    transfer = str(primary.get("color_transfer") or "").lower()
    hdr = "hdr10" if transfer in {"smpte2084", "pq"} else "hlg" if transfer == "arib-std-b67" else None
    return {
        "path": str(path.resolve()),
        "size_bytes": path.stat().st_size,
        "container": str(format_data.get("format_name") or "unknown"),
        "duration_seconds": round(duration, 3),
        "duration_minutes": round(duration / 60, 3),
        "video": {
            "streams": len(videos),
            "width": int(primary.get("width") or 0),
            "height": int(primary.get("height") or 0),
            "codec": str(primary.get("codec_name") or "unknown"),
            "pixel_format": str(primary.get("pix_fmt") or "unknown"),
            "hdr": hdr,
            "title": tags.get("title"),
        },
        "audio": [
            {
                "codec": str(item.get("codec_name") or "unknown"),
                "language": language(item),
                "channels": int(item.get("channels") or 0),
            }
            for item in audios
        ],
        "subtitles": [
            {"codec": str(item.get("codec_name") or "unknown"), "language": language(item)}
            for item in subtitles
        ],
    }


def episode_key(name: str) -> str | None:
    match = re.search(r"(?i)S(\d{1,2})E(\d{1,3})", name)
    return f"S{int(match.group(1)):02d}E{int(match.group(2)):02d}" if match else None


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--path", required=True, type=Path)
    parser.add_argument("--output", required=True, type=Path)
    parser.add_argument("--expected-duration-minutes", type=float)
    parser.add_argument("--duration-tolerance-minutes", type=float, default=2.0)
    parser.add_argument("--expected-resolution")
    parser.add_argument("--preferred-subtitle", action="append", default=["zh-Hans", "en"])
    parser.add_argument("--expected-episode", action="append", default=[])
    args = parser.parse_args()

    if shutil.which("ffprobe") is None:
        raise SystemExit("ERROR: ffprobe is required")
    target = args.path.expanduser().resolve()
    if not target.exists():
        raise SystemExit(f"ERROR: media path does not exist: {target}")
    files = media_files(target)
    if not files:
        raise SystemExit(f"ERROR: no supported media files under {target}")

    inspected: list[dict[str, Any]] = []
    errors: list[str] = []
    for path in files:
        try:
            inspected.append(inspect(path))
        except (RuntimeError, json.JSONDecodeError, OSError) as exc:
            errors.append(str(exc))
    primary = inspected[0] if inspected else {}
    warnings: list[str] = []
    if not primary or primary.get("video", {}).get("streams", 0) < 1:
        errors.append("no readable video stream")
    if float(primary.get("duration_seconds") or 0) <= 0:
        errors.append("media duration is zero or unavailable")
    if args.expected_duration_minutes and primary:
        delta = abs(float(primary.get("duration_minutes") or 0) - args.expected_duration_minutes)
        if delta > args.duration_tolerance_minutes:
            errors.append(
                f"duration differs from expected edition by {delta:.2f} minutes"
            )
    if args.expected_resolution and primary:
        height = int(primary.get("video", {}).get("height") or 0)
        expected = int(re.sub(r"\D", "", args.expected_resolution) or 0)
        if expected and abs(height - expected) > max(16, expected * 0.08):
            warnings.append(f"observed height {height} differs from {args.expected_resolution}")

    subtitle_languages = {
        str(item.get("language") or "und").lower()
        for media in inspected
        for item in media.get("subtitles") or []
    }
    aliases = {
        "zh-hans": {"zh-hans", "zho", "chi", "chs", "zh", "zh-cn"},
        "en": {"en", "eng"},
    }
    missing: list[str] = []
    for preferred in dict.fromkeys(args.preferred_subtitle):
        accepted = aliases.get(preferred.lower(), {preferred.lower()})
        if not subtitle_languages & accepted:
            missing.append(preferred)
    if missing:
        warnings.append("missing preferred subtitle streams: " + ", ".join(missing))

    observed_episodes = {key for path in files if (key := episode_key(path.name))}
    missing_episodes = sorted(set(args.expected_episode) - observed_episodes)
    if missing_episodes:
        errors.append("missing requested episodes: " + ", ".join(missing_episodes))
    partials = partial_files(target)
    if partials:
        warnings.append(f"partial-suffix files remain: {len(partials)}")

    status = "failed" if errors else "passed_with_warnings" if warnings else "passed"
    report = {
        "schema_version": "1.0",
        "verified_at": datetime.now(timezone.utc).isoformat(),
        "status": status,
        "target": str(target),
        "files": inspected,
        "preferred_subtitles_missing": missing,
        "observed_episodes": sorted(observed_episodes),
        "partial_files": partials,
        "warnings": warnings,
        "errors": errors,
    }
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print(json.dumps({"status": status, "files": len(inspected), "warnings": len(warnings), "errors": len(errors), "output": str(args.output)}, ensure_ascii=False, indent=2))
    return 0 if status != "failed" else 3


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, RuntimeError, ValueError, TypeError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc