UHD_Media_Tool/apps/server/test/e2e/e7-tv-series.test.ts

import { existsSync } from "node:fs";
import { copyFile, mkdir } from "node:fs/promises";
import { dirname, join } from "node:path";
import { createMockRssServer, type MockFeedItem, type MockRssServer } from "@umt/testkit";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { TORRENT_SIZE, torrentFor } from "./fixtures/modules.ts";
import { completeDownload, createE2E, type E2E, pgAvailable, seedBase, waitFor } from "./harness.ts";

// E7 常规剧集纵切(验收 S5,REQ §7.1.1 / §8.4):全部经真实任务引擎 + outbox 中继。
// mock RSS 依次出 S01E01、S01E02 单集,最后出一个整季包;已发布收录策略(剧集 1080p Slot),订阅改为持续追踪。
// 断言:
//   · 两集分别入库到同一个已绑定的季目录(季绑定只建一次,之后固定);
//   · content_unit 覆盖按实际文件计算(E01→E01、E02→E02);
//   · 整季包"声称"整季,但实际文件只有 E01/E02(与库内相同)→ 不新增覆盖,Slot 不因包名判满足;
//   · 第三集从未出现:Slot 仍未 satisfied,订阅保持 active/continuous(总集数未知不结束)。

const hasPg = await pgAvailable();
const BASE = "https://audiences.example";
const E01 = "The.Bear.S01E01.1080p.WEB-DL.x264-GRP";
const E02 = "The.Bear.S01E02.1080p.WEB-DL.x264-GRP";
const PACK = "The.Bear.S01.1080p.WEB-DL.x264-GRP";

const item = (id: number, title: string): MockFeedItem => ({
  guid: `aud-${id}`,
  title,
  link: `${BASE}/details.php?id=${id}&hit=1`,
  enclosureUrl: `${BASE}/download.php?id=${id}&passkey=E7PASSKEY`,
  size: TORRENT_SIZE,
  pubDate: new Date(),
  categories: ["剧集"],
});

describe.skipIf(!hasPg)("E7 常规剧集纵切", () => {
  let e: E2E;
  let rss: MockRssServer;
  let base: Awaited<ReturnType<typeof seedBase>>;
  let feedId: string;
  let entityId: string;
  let slotId: string;
  let subId: string;

  // 手动抓取按 10 秒时间槽去重(ingestFeed:manual:<feed>:<槽>):同一槽内的第二次请求会被合并,等到下一个槽再发
  let lastSlot = -1;
  const fetchFeed = async () => {
    const slotOf = () => Math.floor(Date.now() / 10_000);
    if (slotOf() === lastSlot) await new Promise((r) => setTimeout(r, (lastSlot + 1) * 10_000 - Date.now() + 50));
    lastSlot = slotOf();
    const r = await e.api("POST", `/feed-sources/${feedId}/fetch`);
    expect(r.status).toBe(200);
  };
  const importedFor = async (title: string) => {
    const hash = torrentFor(title).hash;
    const dt = await waitFor(
      async () => e.db.selectFrom("download_task").selectAll().where("torrent_hash", "=", hash).executeTakeFirst(),
      60_000,
      `${title} download_task`,
    );
    return waitFor(
      async () =>
        e.db
          .selectFrom("organization_record")
          .selectAll()
          .where("download_task_id", "=", dt.id)
          .where("status", "in", ["succeeded", "failed", "partial"])
          .executeTakeFirst(),
      90_000,
      `${title} 整理记录结束`,
    );
  };
  const slot = () => e.db.selectFrom("acquisition_slot").selectAll().where("id", "=", slotId).executeTakeFirstOrThrow();
  const sub = () => e.db.selectFrom("subscription").selectAll().where("id", "=", subId).executeTakeFirstOrThrow();
  const coverage = async () => {
    const r = await e.db
      .selectFrom("asset_unit_coverage as c")
      .innerJoin("content_unit as u", "u.id", "c.content_unit_id")
      .innerJoin("media_asset as a", "a.id", "c.media_asset_id")
      .select(["a.id as asset", "u.season_number", "u.episode_number"])
      .where("a.media_entity_id", "=", entityId)
      .where("a.status", "=", "active")
      .orderBy("u.episode_number")
      .execute();
    return r.map((x) => ({ asset: x.asset, unit: `S${x.season_number}E${x.episode_number}` }));
  };

  beforeAll(async () => {
    e = await createE2E("e7");
    base = await seedBase(e, { baseUrl: `${BASE}/` });
    rss = await createMockRssServer({ passkey: "E7PASSKEY", items: [item(910001, E01)] });
    const urlRef = await e.booted.ctx.secrets.create({ purpose: "feed_url", plaintext: rss.feedUrl });
    const feed = await e.db
      .insertInto("feed_source")
      .values({
        site_id: base.siteId,
        name: "观众-剧集",
        purpose: "series",
        url_secret_ref: urlRef,
        url_display: rss.feedUrl.replace(/passkey=[^&]+/, "passkey=[REDACTED]"),
        poll_interval_sec: 3600,
      })
      .returning("id")
      .executeTakeFirstOrThrow();
    feedId = feed.id;
  }, 120_000);

  afterAll(async () => {
    await rss?.close();
    await e?.close();
  }, 60_000);

  it("发布剧集 1080p 收录策略 → 持续授权", async () => {
    const created = await e.api("POST", "/policy-revisions", {
      policyKey: "series",
      kind: "acquisition",
      title: "剧集 1080p",
      scope: { mediaKinds: ["series"] },
      rules: {
        newTitle: {
          enabled: true,
          windowDays: 3650,
          mediaKinds: ["series"],
          slots: [{ name: "1080p", hardConditions: { resolutions: ["1080p"] } }],
        },
      },
    });
    expect(created.status).toBe(201);
    const pub = await e.api("POST", `/policy-revisions/${created.body.data.id}/publish`, {
      expectedRevision: created.body.data.revision,
    });
    expect(pub.status).toBe(200);
  });

  it("S01E01:识别为剧集 → 订阅改为持续追踪 → 下载 → 入库到季目录;总集数未知,Slot 不判满足", async () => {
    await fetchFeed();
    const rel = await waitFor(
      async () =>
        e.db
          .selectFrom("source_release")
          .selectAll()
          .where("title", "=", E01)
          .where("resolution_status", "=", "resolved")
          .executeTakeFirst(),
      60_000,
      "S01E01 识别完成",
    );
    entityId = rel.media_entity_id!;
    const entity = await e.db
      .selectFrom("media_entity")
      .selectAll()
      .where("id", "=", entityId)
      .executeTakeFirstOrThrow();
    expect(entity.kind).toBe("series");
    const s = await waitFor(
      async () =>
        e.db.selectFrom("subscription").selectAll().where("media_entity_id", "=", entityId).executeTakeFirst(),
      30_000,
      "新作品订阅",
    );
    subId = s.id;
    const patched = await e.api("PATCH", `/subscriptions/${subId}`, { mode: "continuous" });
    expect(patched.status).toBe(200);
    slotId = (
      await e.db
        .selectFrom("acquisition_slot")
        .select("id")
        .where("media_entity_id", "=", entityId)
        .executeTakeFirstOrThrow()
    ).id;

    await completeDownload(e, E01);
    const rec = await importedFor(E01);
    expect(rec).toMatchObject({ status: "succeeded", action: "copy", trigger: "auto" });
    expect((await coverage()).map((c) => c.unit)).toEqual(["S1E1"]);
    expect((await slot()).status).toBe("empty");
    expect(await sub()).toMatchObject({ status: "active", mode: "continuous" });
  }, 180_000);

  let seasonDir: string;
  it("S01E02:第二集进同一个已绑定的季目录,覆盖逐集计算,Slot 仍未满足", async () => {
    const bindings0 = await e.db
      .selectFrom("path_binding")
      .select(["binding_kind", "season_number", "relative_path"])
      .where("media_entity_id", "=", entityId)
      .orderBy("binding_kind")
      .execute();
    expect(bindings0.map((b) => `${b.binding_kind}:${b.season_number ?? "-"}`)).toEqual(["entity:-", "season:1"]);
    seasonDir = bindings0.find((b) => b.binding_kind === "season")!.relative_path;

    rss.addItem(item(910002, E02));
    await fetchFeed();
    await completeDownload(e, E02);
    const rec = await importedFor(E02);
    expect(rec.status).toBe("succeeded");

    const files = await e.db
      .selectFrom("organization_file_result as f")
      .innerJoin("organization_record as r", "r.id", "f.organization_record_id")
      .select(["f.target_path", "f.status"])
      .where("r.media_entity_id", "=", entityId)
      .where("f.role", "=", "main")
      .orderBy("f.target_path")
      .execute();
    expect(files).toHaveLength(2);
    for (const f of files) {
      expect(f.status).toBe("succeeded");
      expect(dirname(f.target_path!)).toBe(seasonDir);
      expect(existsSync(join(e.lib.root, f.target_path!))).toBe(true);
    }
    expect(files.map((f) => f.target_path!)).toEqual([
      expect.stringMatching(/S01E01\.mkv$/),
      expect.stringMatching(/S01E02\.mkv$/),
    ]);
    // 路径绑定固定:没有因为第二集新建任何绑定
    const bindings1 = await e.db
      .selectFrom("path_binding")
      .select(["binding_kind", "season_number", "relative_path"])
      .where("media_entity_id", "=", entityId)
      .orderBy("binding_kind")
      .execute();
    expect(bindings1).toEqual(bindings0);

    const cov = await coverage();
    expect(cov.map((c) => c.unit)).toEqual(["S1E1", "S1E2"]);
    expect(new Set(cov.map((c) => c.asset)).size).toBe(2);
    expect((await slot()).status).toBe("empty");
    expect(await sub()).toMatchObject({ status: "active", mode: "continuous" });
  }, 180_000);

  it("整季包:实际文件只有 E01/E02(与库内相同)→ 不新增覆盖,不因包名判满足;第三集未出现,订阅继续", async () => {
    const libFiles = await e.db
      .selectFrom("organization_file_result as f")
      .innerJoin("organization_record as r", "r.id", "f.organization_record_id")
      .select(["f.target_path"])
      .where("r.media_entity_id", "=", entityId)
      .where("f.role", "=", "main")
      .where("f.status", "=", "succeeded")
      .orderBy("f.target_path")
      .execute();
    const assetsBefore = await e.db
      .selectFrom("media_asset")
      .select("id")
      .where("media_entity_id", "=", entityId)
      .execute();

    rss.addItem(item(910003, PACK));
    await fetchFeed();
    const packRel = await waitFor(
      async () =>
        e.db
          .selectFrom("source_release")
          .selectAll()
          .where("title", "=", PACK)
          .where("resolution_status", "=", "resolved")
          .executeTakeFirst(),
      60_000,
      "整季包识别完成",
    );
    expect(packRel.media_entity_id).toBe(entityId);
    const hash = torrentFor(PACK).hash;
    await waitFor(async () => e.qb.get(hash), 60_000, "qB 收到整季包");

    // 包内真实文件清单:只有 E01、E02,内容与已入库的相同(常见的"同组合集")
    const pack = [E01, E02].map((t, i) => ({
      path: `${PACK}/${t}.mkv`,
      src: join(e.lib.root, libFiles[i]!.target_path!),
    }));
    for (const p of pack) {
      const dst = join(e.cache.root, p.path);
      await mkdir(dirname(dst), { recursive: true });
      await copyFile(p.src, dst);
    }
    await e.qb.complete(hash, { files: pack.map((p) => ({ path: p.path, size: TORRENT_SIZE })), writeFiles: false });
    const rec = await importedFor(PACK);
    expect(rec.status).toBe("succeeded");
    const packFiles = await e.db
      .selectFrom("organization_file_result")
      .select(["status", "skip_reason", "media_asset_id"])
      .where("organization_record_id", "=", rec.id)
      .where("role", "=", "main")
      .execute();
    expect(packFiles).toHaveLength(2);
    for (const f of packFiles) {
      expect(f).toMatchObject({ status: "skipped", skip_reason: "already_in_library" });
      expect(assetsBefore.map((a) => a.id)).toContain(f.media_asset_id);
    }
    // 没有新资产、没有新覆盖;季单元的"整季"声明不算覆盖,也没有凭空出现第三集
    expect(
      await e.db.selectFrom("media_asset").select("id").where("media_entity_id", "=", entityId).execute(),
    ).toHaveLength(assetsBefore.length);
    expect((await coverage()).map((c) => c.unit)).toEqual(["S1E1", "S1E2"]);
    const eps = await e.db
      .selectFrom("content_unit")
      .select(["episode_number", "status"])
      .where("media_entity_id", "=", entityId)
      .where("kind", "=", "episode")
      .orderBy("episode_number")
      .execute();
    expect(eps).toEqual([
      { episode_number: 1, status: "provisional" },
      { episode_number: 2, status: "provisional" },
    ]);
    // 第三集没出现、总集数未知:Slot 不满足,订阅继续追踪
    const s = await slot();
    expect(s.status).not.toBe("satisfied");
    expect(await sub()).toMatchObject({ status: "active", mode: "continuous" });
    // 三个种子各提交一次
    expect(e.qb.callsTo("torrents/add")).toHaveLength(3);
  }, 180_000);
});