Compare commits
2 Commits
52e74f7709
...
9561e6e196
| Author | SHA1 | Date | |
|---|---|---|---|
| 9561e6e196 | |||
| 9d5e1ff58a |
@@ -0,0 +1,2 @@
|
|||||||
|
-- AlterTable
|
||||||
|
ALTER TABLE "Job" ADD COLUMN "downloadProgress" DOUBLE PRECISION NOT NULL DEFAULT 0;
|
||||||
@@ -53,6 +53,7 @@ model Job {
|
|||||||
claimedAt DateTime?
|
claimedAt DateTime?
|
||||||
createdAt DateTime @default(now())
|
createdAt DateTime @default(now())
|
||||||
updatedAt DateTime @updatedAt
|
updatedAt DateTime @updatedAt
|
||||||
|
downloadProgress Float @default(0)
|
||||||
candidates Candidate[]
|
candidates Candidate[]
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,6 +1,5 @@
|
|||||||
import type { ReactNode } from "react";
|
import type { ReactNode } from "react";
|
||||||
import type { Kind } from "./status";
|
import type { Kind } from "./status";
|
||||||
import { ProgressBar } from "./progress-bar";
|
|
||||||
import { StatusChip } from "./status-chip";
|
import { StatusChip } from "./status-chip";
|
||||||
|
|
||||||
// Map the severity kind to the row's visual class (stripe/bar color).
|
// Map the severity kind to the row's visual class (stripe/bar color).
|
||||||
@@ -15,23 +14,17 @@ export function JobRow({
|
|||||||
album,
|
album,
|
||||||
kind,
|
kind,
|
||||||
label,
|
label,
|
||||||
step,
|
|
||||||
total,
|
|
||||||
meta,
|
meta,
|
||||||
note,
|
note,
|
||||||
indeterminate,
|
bar,
|
||||||
stageLabel,
|
|
||||||
}: {
|
}: {
|
||||||
artist: string;
|
artist: string;
|
||||||
album: string;
|
album: string;
|
||||||
kind: Kind;
|
kind: Kind;
|
||||||
label: string;
|
label: string;
|
||||||
step: number;
|
|
||||||
total: number;
|
|
||||||
meta?: ReactNode;
|
meta?: ReactNode;
|
||||||
note?: ReactNode;
|
note?: ReactNode;
|
||||||
indeterminate?: boolean;
|
bar?: ReactNode;
|
||||||
stageLabel?: string;
|
|
||||||
}) {
|
}) {
|
||||||
return (
|
return (
|
||||||
<article className={`job ${rowClass(kind)}`}>
|
<article className={`job ${rowClass(kind)}`}>
|
||||||
@@ -41,11 +34,7 @@ export function JobRow({
|
|||||||
{album} <span className="artist">· {artist}</span>
|
{album} <span className="artist">· {artist}</span>
|
||||||
</h3>
|
</h3>
|
||||||
{meta ? <div className="meta">{meta}</div> : null}
|
{meta ? <div className="meta">{meta}</div> : null}
|
||||||
{note ? (
|
{note ? <div className="note">{note}</div> : kind !== "attention" ? bar : null}
|
||||||
<div className="note">{note}</div>
|
|
||||||
) : kind !== "attention" ? (
|
|
||||||
<ProgressBar kind={kind} step={step} total={total} indeterminate={indeterminate} stageLabel={stageLabel} />
|
|
||||||
) : null}
|
|
||||||
</div>
|
</div>
|
||||||
<StatusChip kind={kind} label={label} />
|
<StatusChip kind={kind} label={label} />
|
||||||
</article>
|
</article>
|
||||||
|
|||||||
@@ -1,19 +1,17 @@
|
|||||||
import type { Kind } from "./status";
|
import type { Kind } from "./status";
|
||||||
|
|
||||||
/** A thin ruled progress bar. `indeterminate` for open-ended stages (searching);
|
/** A thin ruled progress bar. `indeterminate` for open-ended phases (searching,
|
||||||
* otherwise a discrete step/total fill with a step readout. */
|
* finishing); otherwise a determinate fill to `percent` with a live readout. */
|
||||||
export function ProgressBar({
|
export function ProgressBar({
|
||||||
kind,
|
kind,
|
||||||
step,
|
|
||||||
total,
|
|
||||||
indeterminate,
|
indeterminate,
|
||||||
stageLabel,
|
percent,
|
||||||
|
caption,
|
||||||
}: {
|
}: {
|
||||||
kind: Kind;
|
kind: Kind;
|
||||||
step: number;
|
|
||||||
total: number;
|
|
||||||
indeterminate?: boolean;
|
indeterminate?: boolean;
|
||||||
stageLabel?: string;
|
percent?: number;
|
||||||
|
caption?: string;
|
||||||
}) {
|
}) {
|
||||||
if (indeterminate) {
|
if (indeterminate) {
|
||||||
return (
|
return (
|
||||||
@@ -21,17 +19,17 @@ export function ProgressBar({
|
|||||||
<div className="bar indet">
|
<div className="bar indet">
|
||||||
<span />
|
<span />
|
||||||
</div>
|
</div>
|
||||||
<div className="pct muted">searching</div>
|
<div className="pct muted">{caption ?? "…"}</div>
|
||||||
</div>
|
</div>
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
const pct = total > 0 ? Math.round((step / total) * 100) : 0;
|
const pct = Math.min(100, Math.max(0, percent ?? 0));
|
||||||
return (
|
return (
|
||||||
<div className="prog">
|
<div className="prog">
|
||||||
<div className="bar">
|
<div className="bar">
|
||||||
<span style={{ width: `${pct}%` }} />
|
<span style={{ width: `${pct}%` }} />
|
||||||
</div>
|
</div>
|
||||||
<div className="pct">{stageLabel ? `${stageLabel} · step ${step} of ${total}` : `${step}/${total}`}</div>
|
<div className="pct">{caption ?? `${Math.round(percent ?? 0)}%`}</div>
|
||||||
</div>
|
</div>
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,21 +5,16 @@ describe("describeJob", () => {
|
|||||||
it("maps pipeline states to literal labels + severity kind", () => {
|
it("maps pipeline states to literal labels + severity kind", () => {
|
||||||
expect(describeJob("requested", "intake")).toMatchObject({ label: "Queued", kind: "idle" });
|
expect(describeJob("requested", "intake")).toMatchObject({ label: "Queued", kind: "idle" });
|
||||||
expect(describeJob("matching", "match")).toMatchObject({ label: "Searching", kind: "working" });
|
expect(describeJob("matching", "match")).toMatchObject({ label: "Searching", kind: "working" });
|
||||||
|
expect(describeJob("matched", "match")).toMatchObject({ label: "Searching", kind: "working" });
|
||||||
expect(describeJob("downloading", "download")).toMatchObject({ label: "Downloading", kind: "working" });
|
expect(describeJob("downloading", "download")).toMatchObject({ label: "Downloading", kind: "working" });
|
||||||
expect(describeJob("tagging", "tag")).toMatchObject({ label: "Tagging", kind: "verify" });
|
expect(describeJob("tagging", "tag")).toMatchObject({ label: "Finishing", kind: "verify" });
|
||||||
expect(describeJob("imported", "import")).toMatchObject({ label: "Pressed", kind: "done" });
|
expect(describeJob("imported", "import")).toMatchObject({ label: "Pressed", kind: "done" });
|
||||||
expect(describeJob("needs_attention", "download")).toMatchObject({ label: "Needs attention", kind: "attention" });
|
expect(describeJob("needs_attention", "download")).toMatchObject({ label: "Needs attention", kind: "attention" });
|
||||||
});
|
});
|
||||||
|
|
||||||
it("derives a stepped progress from the pipeline stage", () => {
|
it("falls back gracefully for unknown states", () => {
|
||||||
expect(describeJob("downloading", "download")).toMatchObject({ step: 4, totalSteps: 6 });
|
|
||||||
expect(describeJob("matching", "intake")).toMatchObject({ step: 1, totalSteps: 6 });
|
|
||||||
});
|
|
||||||
|
|
||||||
it("falls back gracefully for unknown states/stages", () => {
|
|
||||||
expect(describeJob("weird", "intake").kind).toBe("idle");
|
expect(describeJob("weird", "intake").kind).toBe("idle");
|
||||||
expect(describeJob("weird", "intake").label).toBe("Weird");
|
expect(describeJob("weird", "intake").label).toBe("Weird");
|
||||||
expect(describeJob("downloading", "???").step).toBe(0);
|
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
+13
-19
@@ -1,16 +1,14 @@
|
|||||||
export type Kind = "idle" | "working" | "verify" | "done" | "attention";
|
export type Kind = "idle" | "working" | "verify" | "done" | "attention";
|
||||||
|
|
||||||
// Pipeline stage order (worker Job.currentStage), used to derive a stepped progress.
|
// Worker Job.state -> a literal, legible label + severity kind, collapsed to three
|
||||||
const STAGES = ["intake", "match", "rank", "download", "tag", "import"];
|
// honest phases: Searching -> Downloading -> Finishing. The record-label metaphor
|
||||||
|
// stays in the chrome; these labels never obscure what's happening.
|
||||||
// Worker Job.state -> a literal, legible label + severity kind. The record-label
|
|
||||||
// metaphor stays in the chrome; these labels never obscure what's happening.
|
|
||||||
const STATE_MAP: Record<string, { label: string; kind: Kind }> = {
|
const STATE_MAP: Record<string, { label: string; kind: Kind }> = {
|
||||||
requested: { label: "Queued", kind: "idle" },
|
requested: { label: "Queued", kind: "idle" },
|
||||||
matching: { label: "Searching", kind: "working" },
|
matching: { label: "Searching", kind: "working" },
|
||||||
matched: { label: "Matched", kind: "working" },
|
matched: { label: "Searching", kind: "working" },
|
||||||
downloading: { label: "Downloading", kind: "working" },
|
downloading: { label: "Downloading", kind: "working" },
|
||||||
tagging: { label: "Tagging", kind: "verify" },
|
tagging: { label: "Finishing", kind: "verify" },
|
||||||
imported: { label: "Pressed", kind: "done" },
|
imported: { label: "Pressed", kind: "done" },
|
||||||
needs_attention: { label: "Needs attention", kind: "attention" },
|
needs_attention: { label: "Needs attention", kind: "attention" },
|
||||||
};
|
};
|
||||||
@@ -29,16 +27,12 @@ export function timeAgo(value: string | Date | null | undefined, now: number): s
|
|||||||
return `${Math.floor(h / 24)}d`;
|
return `${Math.floor(h / 24)}d`;
|
||||||
}
|
}
|
||||||
|
|
||||||
export function describeJob(
|
export function describeJob(state: string, stage: string): { label: string; kind: Kind } {
|
||||||
state: string,
|
void stage; // stage no longer drives the label; kept for call-site compatibility
|
||||||
stage: string,
|
return (
|
||||||
): { label: string; kind: Kind; step: number; totalSteps: number } {
|
STATE_MAP[state] ?? {
|
||||||
const base =
|
label: state ? state[0].toUpperCase() + state.slice(1) : "Unknown",
|
||||||
STATE_MAP[state] ??
|
kind: "idle",
|
||||||
({ label: state ? state[0].toUpperCase() + state.slice(1) : "Unknown", kind: "idle" } as {
|
}
|
||||||
label: string;
|
);
|
||||||
kind: Kind;
|
|
||||||
});
|
|
||||||
const idx = STAGES.indexOf(stage);
|
|
||||||
return { ...base, step: idx >= 0 ? idx + 1 : 0, totalSteps: STAGES.length };
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -62,7 +62,15 @@ describe("requests API", () => {
|
|||||||
data: {
|
data: {
|
||||||
artist: "Chosen Artist",
|
artist: "Chosen Artist",
|
||||||
album: "Chosen Album",
|
album: "Chosen Album",
|
||||||
job: { create: { state: "downloading", currentStage: "download", attempts: 2, error: "boom" } },
|
job: {
|
||||||
|
create: {
|
||||||
|
state: "downloading",
|
||||||
|
currentStage: "download",
|
||||||
|
attempts: 2,
|
||||||
|
error: "boom",
|
||||||
|
downloadProgress: 0.42,
|
||||||
|
},
|
||||||
|
},
|
||||||
},
|
},
|
||||||
include: { job: true },
|
include: { job: true },
|
||||||
});
|
});
|
||||||
@@ -99,6 +107,7 @@ describe("requests API", () => {
|
|||||||
expect(found.job.chosen).toEqual({ source: "qobuz", format: "flac", trackCount: 12 });
|
expect(found.job.chosen).toEqual({ source: "qobuz", format: "flac", trackCount: 12 });
|
||||||
expect(found.job.error).toBe("boom");
|
expect(found.job.error).toBe("boom");
|
||||||
expect(found.job.attempts).toBe(2);
|
expect(found.job.attempts).toBe(2);
|
||||||
|
expect(found.job.downloadProgress).toBe(0.42);
|
||||||
});
|
});
|
||||||
|
|
||||||
it("reports no chosen candidate and zero count when a job has none", async () => {
|
it("reports no chosen candidate and zero count when a job has none", async () => {
|
||||||
@@ -109,5 +118,6 @@ describe("requests API", () => {
|
|||||||
const found = body.requests.find((r: { id: string }) => r.id === created.id);
|
const found = body.requests.find((r: { id: string }) => r.id === created.id);
|
||||||
expect(found.job.candidateCount).toBe(0);
|
expect(found.job.candidateCount).toBe(0);
|
||||||
expect(found.job.chosen).toBeNull();
|
expect(found.job.chosen).toBeNull();
|
||||||
|
expect(found.job.downloadProgress).toBe(0);
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -63,6 +63,7 @@ export async function GET() {
|
|||||||
error: r.job.error,
|
error: r.job.error,
|
||||||
claimedAt: r.job.claimedAt,
|
claimedAt: r.job.claimedAt,
|
||||||
updatedAt: r.job.updatedAt,
|
updatedAt: r.job.updatedAt,
|
||||||
|
downloadProgress: r.job.downloadProgress,
|
||||||
candidateCount: cands.length,
|
candidateCount: cands.length,
|
||||||
chosen: chosen ? { source: chosen.source, format: chosen.format, trackCount: chosen.trackCount } : null,
|
chosen: chosen ? { source: chosen.source, format: chosen.format, trackCount: chosen.trackCount } : null,
|
||||||
};
|
};
|
||||||
|
|||||||
+24
-9
@@ -5,6 +5,7 @@ import { describeJob, timeAgo } from "./_ui/status";
|
|||||||
import { StatTiles } from "./_ui/stat-tiles";
|
import { StatTiles } from "./_ui/stat-tiles";
|
||||||
import { SectionHeader } from "./_ui/section-header";
|
import { SectionHeader } from "./_ui/section-header";
|
||||||
import { JobRow } from "./_ui/job-row";
|
import { JobRow } from "./_ui/job-row";
|
||||||
|
import { ProgressBar } from "./_ui/progress-bar";
|
||||||
import { PressedList } from "./_ui/pressed-list";
|
import { PressedList } from "./_ui/pressed-list";
|
||||||
import { toast } from "./_ui/toast";
|
import { toast } from "./_ui/toast";
|
||||||
|
|
||||||
@@ -24,6 +25,7 @@ type Row = {
|
|||||||
updatedAt: string | null;
|
updatedAt: string | null;
|
||||||
candidateCount: number;
|
candidateCount: number;
|
||||||
chosen: { source: string; format: string; trackCount: number } | null;
|
chosen: { source: string; format: string; trackCount: number } | null;
|
||||||
|
downloadProgress: number;
|
||||||
} | null;
|
} | null;
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -38,6 +40,7 @@ function jobOf(r: Row) {
|
|||||||
updatedAt: null,
|
updatedAt: null,
|
||||||
candidateCount: 0,
|
candidateCount: 0,
|
||||||
chosen: null,
|
chosen: null,
|
||||||
|
downloadProgress: 0,
|
||||||
}
|
}
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
@@ -139,7 +142,7 @@ export function Queue() {
|
|||||||
let meta: string | undefined;
|
let meta: string | undefined;
|
||||||
if (j.state === "downloading" || j.state === "tagging") {
|
if (j.state === "downloading" || j.state === "tagging") {
|
||||||
const parts: string[] = [];
|
const parts: string[] = [];
|
||||||
if (j.chosen) parts.push(`${j.chosen.source} · ${j.chosen.format} · ${j.chosen.trackCount} tracks`);
|
if (j.chosen) parts.push(`${j.chosen.source} · ${j.chosen.format}`);
|
||||||
const elapsed = timeAgo(j.claimedAt ?? j.updatedAt, now);
|
const elapsed = timeAgo(j.claimedAt ?? j.updatedAt, now);
|
||||||
if (elapsed) parts.push(elapsed);
|
if (elapsed) parts.push(elapsed);
|
||||||
meta = parts.join(" · ") || undefined;
|
meta = parts.join(" · ") || undefined;
|
||||||
@@ -151,8 +154,6 @@ export function Queue() {
|
|||||||
} else if (j.state === "requested") {
|
} else if (j.state === "requested") {
|
||||||
const elapsed = timeAgo(r.createdAt, now);
|
const elapsed = timeAgo(r.createdAt, now);
|
||||||
meta = elapsed ? `In the queue · ${elapsed}` : "In the queue";
|
meta = elapsed ? `In the queue · ${elapsed}` : "In the queue";
|
||||||
} else if (j.state === "needs_attention") {
|
|
||||||
meta = undefined;
|
|
||||||
} else {
|
} else {
|
||||||
meta = undefined;
|
meta = undefined;
|
||||||
}
|
}
|
||||||
@@ -166,8 +167,25 @@ export function Queue() {
|
|||||||
</>
|
</>
|
||||||
) : undefined;
|
) : undefined;
|
||||||
|
|
||||||
const stageLabel =
|
let bar: React.ReactNode;
|
||||||
attention || j.state === "matching" || j.state === "matched" ? undefined : j.currentStage;
|
if (j.state === "matching" || j.state === "matched") {
|
||||||
|
bar = <ProgressBar kind="working" indeterminate caption="searching" />;
|
||||||
|
} else if (j.state === "downloading") {
|
||||||
|
const pct = Math.round((j.downloadProgress || 0) * 100);
|
||||||
|
const total = j.chosen?.trackCount ?? 0;
|
||||||
|
const done = total ? Math.round((pct / 100) * total) : 0;
|
||||||
|
bar = (
|
||||||
|
<ProgressBar
|
||||||
|
kind="working"
|
||||||
|
percent={pct}
|
||||||
|
caption={total ? `${pct}% · ${done}/${total} tracks` : `${pct}%`}
|
||||||
|
/>
|
||||||
|
);
|
||||||
|
} else if (j.state === "tagging") {
|
||||||
|
bar = <ProgressBar kind="verify" indeterminate caption="finishing" />;
|
||||||
|
} else {
|
||||||
|
bar = undefined;
|
||||||
|
}
|
||||||
|
|
||||||
return (
|
return (
|
||||||
<JobRow
|
<JobRow
|
||||||
@@ -176,12 +194,9 @@ export function Queue() {
|
|||||||
album={r.album}
|
album={r.album}
|
||||||
kind={d.kind}
|
kind={d.kind}
|
||||||
label={d.label}
|
label={d.label}
|
||||||
step={d.step}
|
|
||||||
total={d.totalSteps}
|
|
||||||
indeterminate={j.state === "matching" || j.state === "matched"}
|
|
||||||
meta={meta}
|
meta={meta}
|
||||||
note={note}
|
note={note}
|
||||||
stageLabel={stageLabel}
|
bar={bar}
|
||||||
/>
|
/>
|
||||||
);
|
);
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -1,4 +1,6 @@
|
|||||||
|
import os
|
||||||
import shutil
|
import shutil
|
||||||
|
import threading
|
||||||
from dataclasses import replace
|
from dataclasses import replace
|
||||||
from typing import Sequence
|
from typing import Sequence
|
||||||
|
|
||||||
@@ -11,6 +13,39 @@ from lyra_worker.quality import quality_class
|
|||||||
from lyra_worker.ranker import rank_candidates
|
from lyra_worker.ranker import rank_candidates
|
||||||
from lyra_worker.types import Candidate, MBTarget
|
from lyra_worker.types import Candidate, MBTarget
|
||||||
|
|
||||||
|
_AUDIO_EXT = {".flac", ".mp3", ".m4a", ".opus", ".ogg", ".aac", ".wav"}
|
||||||
|
|
||||||
|
|
||||||
|
def _count_staged_audio(staging: str) -> int:
|
||||||
|
"""Count audio files anywhere under the staging dir (streamrip nests them in a subfolder)."""
|
||||||
|
n = 0
|
||||||
|
for _root, _dirs, files in os.walk(staging):
|
||||||
|
for f in files:
|
||||||
|
if os.path.splitext(f)[1].lower() in _AUDIO_EXT:
|
||||||
|
n += 1
|
||||||
|
return n
|
||||||
|
|
||||||
|
|
||||||
|
def _poll_download_progress(job_id: str, staging: str, expected: int, stop: "threading.Event") -> None:
|
||||||
|
"""Until `stop`, periodically write Job.downloadProgress = files-in-staging / expected
|
||||||
|
(capped 0.99). Uses its own short-lived connection (psycopg conns aren't shareable across
|
||||||
|
threads). A missing DSN or expected<=0 just no-ops."""
|
||||||
|
dsn = os.environ.get("DATABASE_URL")
|
||||||
|
if not dsn or expected <= 0:
|
||||||
|
return
|
||||||
|
conn = psycopg.connect(dsn)
|
||||||
|
try:
|
||||||
|
while not stop.is_set():
|
||||||
|
frac = min(_count_staged_audio(staging) / expected, 0.99)
|
||||||
|
with conn.cursor() as cur:
|
||||||
|
cur.execute('UPDATE "Job" SET "downloadProgress" = %s WHERE id = %s', (frac, job_id))
|
||||||
|
conn.commit()
|
||||||
|
stop.wait(1.5)
|
||||||
|
except Exception as e: # a progress poller must never affect the job
|
||||||
|
print(f"pipeline: download progress poller error: {e}", flush=True)
|
||||||
|
finally:
|
||||||
|
conn.close()
|
||||||
|
|
||||||
|
|
||||||
def _set_state(conn: psycopg.Connection, job_id: str, state: str, stage: str) -> None:
|
def _set_state(conn: psycopg.Connection, job_id: str, state: str, stage: str) -> None:
|
||||||
with conn.cursor() as cur:
|
with conn.cursor() as cur:
|
||||||
@@ -21,6 +56,12 @@ def _set_state(conn: psycopg.Connection, job_id: str, state: str, stage: str) ->
|
|||||||
conn.commit()
|
conn.commit()
|
||||||
|
|
||||||
|
|
||||||
|
def _set_download_progress(conn: psycopg.Connection, job_id: str, frac: float) -> None:
|
||||||
|
with conn.cursor() as cur:
|
||||||
|
cur.execute('UPDATE "Job" SET "downloadProgress" = %s WHERE id = %s', (frac, job_id))
|
||||||
|
conn.commit()
|
||||||
|
|
||||||
|
|
||||||
def _request_id(conn: psycopg.Connection, job_id: str) -> str:
|
def _request_id(conn: psycopg.Connection, job_id: str) -> str:
|
||||||
with conn.cursor() as cur:
|
with conn.cursor() as cur:
|
||||||
cur.execute('SELECT "requestId" FROM "Job" WHERE id = %s', (job_id,))
|
cur.execute('SELECT "requestId" FROM "Job" WHERE id = %s', (job_id,))
|
||||||
@@ -201,6 +242,14 @@ def run_pipeline(
|
|||||||
staging = staging_dir(staging_root or f"{dest_root}/.staging", job_id)
|
staging = staging_dir(staging_root or f"{dest_root}/.staging", job_id)
|
||||||
winner = None
|
winner = None
|
||||||
result = None
|
result = None
|
||||||
|
_set_download_progress(conn, job_id, 0.0) # reset for this run
|
||||||
|
expected = target.track_count or (ranked[0].track_count if ranked else 0)
|
||||||
|
_stop = threading.Event()
|
||||||
|
_poller = threading.Thread(
|
||||||
|
target=_poll_download_progress, args=(job_id, staging, expected, _stop), daemon=True
|
||||||
|
)
|
||||||
|
_poller.start()
|
||||||
|
try:
|
||||||
try:
|
try:
|
||||||
for candidate in ranked:
|
for candidate in ranked:
|
||||||
adapter = by_source.get(candidate.source)
|
adapter = by_source.get(candidate.source)
|
||||||
@@ -212,9 +261,13 @@ def run_pipeline(
|
|||||||
if result.ok:
|
if result.ok:
|
||||||
winner = candidate
|
winner = candidate
|
||||||
break
|
break
|
||||||
|
finally:
|
||||||
|
_stop.set()
|
||||||
|
_poller.join(timeout=3)
|
||||||
if winner is None or result is None or not result.ok:
|
if winner is None or result is None or not result.ok:
|
||||||
_fail(conn, job_id, "all downloads failed")
|
_fail(conn, job_id, "all downloads failed")
|
||||||
return
|
return
|
||||||
|
_set_download_progress(conn, job_id, 1.0) # download done — UI shows 100% into Finishing
|
||||||
|
|
||||||
# 5. verify completeness on the staging copy, then promote + tag
|
# 5. verify completeness on the staging copy, then promote + tag
|
||||||
_set_state(conn, job_id, "tagging", "tag")
|
_set_state(conn, job_id, "tagging", "tag")
|
||||||
|
|||||||
@@ -0,0 +1,22 @@
|
|||||||
|
import os
|
||||||
|
|
||||||
|
from lyra_worker.pipeline import _count_staged_audio
|
||||||
|
|
||||||
|
|
||||||
|
def test_count_staged_audio_counts_nested_audio_files_only(tmp_path):
|
||||||
|
nested = tmp_path / "Artist" / "Album"
|
||||||
|
nested.mkdir(parents=True)
|
||||||
|
(nested / "01.flac").write_bytes(b"")
|
||||||
|
(nested / "02.flac").write_bytes(b"")
|
||||||
|
(nested / "03.flac").write_bytes(b"")
|
||||||
|
(nested / "cover.jpg").write_bytes(b"")
|
||||||
|
|
||||||
|
assert _count_staged_audio(str(tmp_path)) == 3
|
||||||
|
|
||||||
|
|
||||||
|
def test_count_staged_audio_empty_dir_is_zero(tmp_path):
|
||||||
|
assert _count_staged_audio(str(tmp_path)) == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_count_staged_audio_missing_dir_is_zero():
|
||||||
|
assert _count_staged_audio("/nonexistent/path/for/sure") == 0
|
||||||
@@ -19,6 +19,12 @@ def _request_status(conn, job_id):
|
|||||||
return cur.fetchone()[0]
|
return cur.fetchone()[0]
|
||||||
|
|
||||||
|
|
||||||
|
def _download_progress(conn, job_id):
|
||||||
|
with conn.cursor() as cur:
|
||||||
|
cur.execute('SELECT "downloadProgress" FROM "Job" WHERE id = %s', (job_id,))
|
||||||
|
return cur.fetchone()[0]
|
||||||
|
|
||||||
|
|
||||||
def test_success_picks_best_source_and_imports(conn):
|
def test_success_picks_best_source_and_imports(conn):
|
||||||
job_id = insert_request(conn, artist="Radiohead", album="In Rainbows")
|
job_id = insert_request(conn, artist="Radiohead", album="In Rainbows")
|
||||||
claim_next(conn)
|
claim_next(conn)
|
||||||
@@ -26,6 +32,7 @@ def test_success_picks_best_source_and_imports(conn):
|
|||||||
|
|
||||||
assert _job_state(conn, job_id) == ("imported", "import")
|
assert _job_state(conn, job_id) == ("imported", "import")
|
||||||
assert _request_status(conn, job_id) == "completed"
|
assert _request_status(conn, job_id) == "completed"
|
||||||
|
assert _download_progress(conn, job_id) == 1.0 # a successful download completes at 100%
|
||||||
with conn.cursor() as cur:
|
with conn.cursor() as cur:
|
||||||
cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,))
|
cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,))
|
||||||
assert cur.fetchone()[0] == "qobuz" # highest quality won
|
assert cur.fetchone()[0] == "qobuz" # highest quality won
|
||||||
@@ -41,6 +48,7 @@ def test_falls_through_when_best_download_fails(conn):
|
|||||||
run_pipeline(conn, job_id, [FailingAdapter(), FakeSoulseek()], dest_root="/tmp/lib")
|
run_pipeline(conn, job_id, [FailingAdapter(), FakeSoulseek()], dest_root="/tmp/lib")
|
||||||
|
|
||||||
assert _job_state(conn, job_id) == ("imported", "import")
|
assert _job_state(conn, job_id) == ("imported", "import")
|
||||||
|
assert _download_progress(conn, job_id) == 1.0 # falls through, but still completes at 100%
|
||||||
with conn.cursor() as cur:
|
with conn.cursor() as cur:
|
||||||
cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,))
|
cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,))
|
||||||
assert cur.fetchone()[0] == "soulseek"
|
assert cur.fetchone()[0] == "soulseek"
|
||||||
@@ -75,6 +83,7 @@ def test_all_downloads_fail_goes_to_needs_attention(conn):
|
|||||||
run_pipeline(conn, job_id, [FailingAdapter()], dest_root="/tmp/lib")
|
run_pipeline(conn, job_id, [FailingAdapter()], dest_root="/tmp/lib")
|
||||||
|
|
||||||
assert _job_state(conn, job_id)[0] == "needs_attention"
|
assert _job_state(conn, job_id)[0] == "needs_attention"
|
||||||
|
assert _download_progress(conn, job_id) == 0.0 # never reached 1.0 — all attempts failed
|
||||||
|
|
||||||
|
|
||||||
def test_incomplete_download_goes_to_needs_attention(conn):
|
def test_incomplete_download_goes_to_needs_attention(conn):
|
||||||
|
|||||||
Reference in New Issue
Block a user