#!/usr/bin/env python3
"""资源保护自检：小分段生效，重任务不会并行抢机器。"""
import os
import sys
import tempfile
import threading
import time

if len(sys.argv) > 1:
    arg = sys.argv[1]
    if arg in ("-h", "--help"):
        print(__doc__.strip())
        sys.exit(0)
    if arg in ("-v", "--version"):
        print("fenxuan resource guard 1.6.0")
        sys.exit(0)
    print(f"unknown arguments: {sys.argv[1:]}", file=sys.stderr)
    sys.exit(2)

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(HERE, "..", "server"))
os.environ["FENXUAN_DATA"] = tempfile.mkdtemp(prefix="fenxuan-guard-")

import app as server  # noqa: E402
import separate  # noqa: E402


def main():
    assert server._track_filename("像风/一样", "钢琴", ext="wav") == \
        "像风_一样_钢琴.wav"
    assert server._passes_quality(-10, 0.5)
    assert not server._passes_quality(-30, 0.5)
    assert not server._passes_quality(-10, 0.01)
    assert server.OUTPUT_FORMATS["both"] == ("mp3", "wav")
    assert set(server.PROFILES) == {"vocal", "stems6"}
    assert separate.STEMS_6S == (
        "vocals", "drums", "bass", "other", "guitar", "piano")

    assert not hasattr(separate, "separate_roformer")

    with tempfile.TemporaryDirectory(prefix="fenxuan-mix-") as folder:
        one = os.path.join(folder, "one.wav")
        two = os.path.join(folder, "two.wav")
        mixed = os.path.join(folder, "mixed.wav")
        server.sf.write(one, server.np.full((100, 2), 0.25, dtype="float32"),
                        44100, subtype="FLOAT")
        server.sf.write(two, server.np.full((100, 2), -0.1, dtype="float32"),
                        44100, subtype="FLOAT")
        server._mix_stems([one, two], mixed)
        audio, _ = server.sf.read(mixed, dtype="float32", always_2d=True)
        assert server.np.allclose(audio, 0.15, atol=1e-6)

    old_cache = os.path.join(server.DATA, "old-cache")
    fresh_cache = os.path.join(server.DATA, "fresh-cache")
    os.makedirs(old_cache)
    os.makedirs(fresh_cache)
    now = time.time()
    os.utime(old_cache, (now - 3 * 3600, now - 3 * 3600))
    server._cleanup_storage(now)
    assert not os.path.exists(old_cache)
    assert os.path.exists(fresh_cache)

    active = 0
    peak = 0
    guard = threading.Lock()

    def work(job_id):
        nonlocal active, peak
        with guard:
            active += 1
            peak = max(peak, active)
        time.sleep(0.08)
        with guard:
            active -= 1

    for job_id in ("a", "b"):
        server.JOBS[job_id] = {"stage": "decode", "progress": 0}
    threads = [threading.Thread(target=server._run_one, args=(work, job_id))
               for job_id in ("a", "b")]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    assert peak == 1
    print("RESOURCE_GUARD: PASS")


if __name__ == "__main__":
    main()
