diff --git a/.gitignore b/.gitignore
index 7cabccb..09439d7 100644
--- a/.gitignore
+++ b/.gitignore
@@ -15,3 +15,6 @@ DouYinSparkFlow/core/protocol_sender_debug.mjs
DouYinSparkFlow/**/__pycache__/
DouYinSparkFlow/**/*.pyc
DouYinSparkFlow/config.json
+
+.migration-backups/
+DouYinSparkFlow/webui_users.json
diff --git a/DouYinSparkFlow/.gitignore b/DouYinSparkFlow/.gitignore
index f2c0f18..f40b9dc 100644
--- a/DouYinSparkFlow/.gitignore
+++ b/DouYinSparkFlow/.gitignore
@@ -7,3 +7,6 @@ logs/
usersData.json
webui_settings.json
config.json
+
+webui_users.json
+.migration-backups/
diff --git a/DouYinSparkFlow/core/friends.py b/DouYinSparkFlow/core/friends.py
index cc9165d..c0a6a54 100644
--- a/DouYinSparkFlow/core/friends.py
+++ b/DouYinSparkFlow/core/friends.py
@@ -141,7 +141,21 @@ async def _wait_for_first_friend_or_empty(page):
return False
+async def _wait_for_chat_or_login(page, timeout_seconds=30):
+ deadline = asyncio.get_running_loop().time() + timeout_seconds
+ while asyncio.get_running_loop().time() < deadline:
+ await _ensure_logged_in(page)
+ try:
+ if await page.locator("#sub-app").count() > 0:
+ return
+ except Exception:
+ pass
+ await asyncio.sleep(0.5)
+ raise RuntimeError("chat page did not load within timeout")
+
+
async def collect_friend_names(page):
+ await _wait_for_chat_or_login(page)
await _click_friends_tab(page)
await asyncio.sleep(1)
@@ -217,12 +231,10 @@ async def fetch_account_friends(account):
context.set_default_timeout(120000)
page = await context.new_page()
- await page.goto("https://creator.douyin.com/", wait_until="domcontentloaded", timeout=60000)
await context.add_cookies(cookies)
- await page.goto(CHAT_PAGE_URL, wait_until="domcontentloaded", timeout=60000)
- await asyncio.sleep(2)
+ await page.goto(CHAT_PAGE_URL, wait_until="commit", timeout=30000)
+ await asyncio.sleep(1)
- await _ensure_logged_in(page)
friends = await collect_friend_names(page)
return friends
except RuntimeError:
diff --git a/DouYinSparkFlow/core/tasks.py b/DouYinSparkFlow/core/tasks.py
index bf19271..c2d83e5 100644
--- a/DouYinSparkFlow/core/tasks.py
+++ b/DouYinSparkFlow/core/tasks.py
@@ -1625,6 +1625,13 @@ def _manual_run_unsent_only():
return _is_manual_run() and os.getenv("SPARKFLOW_MANUAL_UNSENT_ONLY") == "1"
+def _requested_account_refs():
+ raw = os.getenv("SPARKFLOW_ACCOUNT_REFS")
+ if raw is None:
+ return None
+ return {item.strip() for item in raw.split(",") if item.strip()}
+
+
def _unsent_retry_max_attempts():
raw_value = str(os.getenv("SPARKFLOW_UNSENT_RETRY_MAX_ATTEMPTS") or "3").strip()
try:
@@ -2719,6 +2726,9 @@ async def _do_user_task_locked(browser, user, send_strategy, profile_config, fri
async def runTasks():
active_config = get_config(force_reload=True)
all_user_data = get_userData(force_reload=True)
+ requested_refs = _requested_account_refs()
+ if requested_refs is not None:
+ all_user_data = [user for user in all_user_data if user.get("account_ref") in requested_refs]
active_user_data = [user for user in all_user_data if user.get("enabled", True)]
disabled_user_data = [user for user in all_user_data if not user.get("enabled", True)]
diff --git a/DouYinSparkFlow/login_desktop_server.py b/DouYinSparkFlow/login_desktop_server.py
index 1100faf..6cbcac6 100644
--- a/DouYinSparkFlow/login_desktop_server.py
+++ b/DouYinSparkFlow/login_desktop_server.py
@@ -44,6 +44,7 @@ GENERIC_WWW_NAMES = {
class LoginDesktopManager:
def __init__(self):
self._lock = asyncio.Lock()
+ self._page_operation_lock = asyncio.Lock()
self.playwright = None
self.context = None
self.page = None
@@ -242,13 +243,14 @@ class LoginDesktopManager:
if page:
current_url = page.url
- try:
- result = await collect_login_result(page, self.context, timeout_ms=1000)
- logged_in = True
- username = result["username"]
- unique_id = result["unique_id"]
- except Exception:
- pass
+ if not self._page_operation_lock.locked():
+ try:
+ result = await collect_login_result(page, self.context, timeout_ms=1000)
+ logged_in = True
+ username = result["username"]
+ unique_id = result["unique_id"]
+ except Exception:
+ pass
payload = {
"running": True,
@@ -263,30 +265,73 @@ class LoginDesktopManager:
return dict(payload)
async def open_login(self):
- await self.refresh_login_qr()
+ self.mark_activity()
+ try:
+ await asyncio.wait_for(self._page_operation_lock.acquire(), timeout=5)
+ except asyncio.TimeoutError as exc:
+ raise RuntimeError("login page is busy; retry shortly") from exc
+ try:
+ page = await self._get_active_page()
+ if page.url.startswith(REMOTE_LOGIN_URL):
+ return {"ok": True, "url": page.url}
+ refresh_url = f"{REMOTE_LOGIN_URL}?qr_refresh={int(time.time() * 1000)}"
+ try:
+ await page.goto(refresh_url, wait_until="commit", timeout=30000)
+ except Exception:
+ await self.stop(clear_profile=False)
+ await self.start()
+ page = await self._get_active_page()
+ await page.goto(refresh_url, wait_until="commit", timeout=30000)
+ return {"ok": True, "url": page.url}
+ finally:
+ self._page_operation_lock.release()
async def refresh_login_qr(self):
self.mark_activity()
+ try:
+ await asyncio.wait_for(self._page_operation_lock.acquire(), timeout=5)
+ except asyncio.TimeoutError as exc:
+ raise RuntimeError("login page is busy; retry shortly") from exc
+ try:
+ return await self._refresh_login_qr_locked()
+ finally:
+ self._page_operation_lock.release()
+
+ async def _refresh_login_qr_locked(self):
refresh_url = f"{REMOTE_LOGIN_URL}?qr_refresh={int(time.time() * 1000)}"
try:
page = await self._get_active_page()
- await page.goto(refresh_url, wait_until="domcontentloaded", timeout=60000)
+ await page.goto(refresh_url, wait_until="commit", timeout=30000)
except Exception:
await self.reset()
page = await self._get_active_page()
- await page.goto(refresh_url, wait_until="domcontentloaded", timeout=60000)
+ await page.goto(refresh_url, wait_until="commit", timeout=30000)
deadline = asyncio.get_running_loop().time() + 45
logged_in = False
qr_ready = False
while asyncio.get_running_loop().time() < deadline:
- qr = page.locator('img[class*="qrcode"]').first
- try:
- if await qr.count() and await qr.is_visible():
- qr_ready = True
+ for selector in (
+ 'img[class*="qrcode"]',
+ 'img[src^="data:image/png;base64"]',
+ ):
+ candidates = page.locator(selector)
+ for index in range(await candidates.count()):
+ qr = candidates.nth(index)
+ try:
+ if not await qr.is_visible():
+ continue
+ box = await qr.bounding_box()
+ if not box or box["width"] < 120 or box["height"] < 120:
+ continue
+ ratio = box["width"] / max(1, box["height"])
+ if 0.8 <= ratio <= 1.25:
+ qr_ready = True
+ break
+ except Exception:
+ pass
+ if qr_ready:
break
- except Exception:
- pass
if "/creator-micro/" in page.url:
logged_in = True
@@ -452,6 +497,12 @@ async def reset():
return {"ok": True}
+@app.post("/close")
+async def close():
+ await manager.stop(clear_profile=True)
+ return {"ok": True}
+
+
@app.post("/refresh-qr")
async def refresh_qr():
return await manager.refresh_login_qr()
@@ -472,6 +523,8 @@ async def export():
@app.get("/qr")
async def login_qr():
+ if manager._page_operation_lock.locked():
+ raise HTTPException(status_code=503, detail="login page is busy; retry shortly")
page = await manager._get_active_page()
expired = await page.locator('[class*="qrcode_expired"]').count()
if expired and await page.locator('[class*="qrcode_expired"]').first.is_visible():
@@ -499,7 +552,7 @@ async def login_qr():
)
except Exception:
continue
- raise HTTPException(status_code=404, detail="login QR code is not ready")
+ raise HTTPException(status_code=202, detail="login QR code is still starting", headers={"Retry-After": "2"})
@app.get("/debug/screenshot")
diff --git a/DouYinSparkFlow/requirements-web.txt b/DouYinSparkFlow/requirements-web.txt
index 4d00dc7..bfc0d7a 100644
--- a/DouYinSparkFlow/requirements-web.txt
+++ b/DouYinSparkFlow/requirements-web.txt
@@ -1,5 +1,7 @@
-fastapi==0.115.6
+fastapi==0.117.1
itsdangerous==2.2.0
Jinja2==3.1.6
python-multipart==0.0.20
uvicorn==0.34.0
+starlette==0.46.2
+tzdata==2025.2
diff --git a/DouYinSparkFlow/requirements.txt b/DouYinSparkFlow/requirements.txt
index 90a8129..a75687f 100644
--- a/DouYinSparkFlow/requirements.txt
+++ b/DouYinSparkFlow/requirements.txt
@@ -1,7 +1,7 @@
certifi==2025.11.12
charset-normalizer==3.4.4
colorama==0.4.6
-fastapi==0.115.6
+fastapi==0.117.1
greenlet==3.2.4
httpx==0.28.1
idna==3.11
@@ -21,3 +21,5 @@ typing_extensions==4.15.0
urllib3==2.5.0
uvicorn==0.34.0
websockets==15.0.1
+starlette==0.46.2
+tzdata==2025.2
diff --git a/DouYinSparkFlow/scripts/migrate_web_users.py b/DouYinSparkFlow/scripts/migrate_web_users.py
new file mode 100644
index 0000000..ba5ca45
--- /dev/null
+++ b/DouYinSparkFlow/scripts/migrate_web_users.py
@@ -0,0 +1,121 @@
+"""Idempotently initialize SparkFlow Web users and account ownership.
+
+Usage (passwords can also be supplied through environment variables):
+ python scripts/migrate_web_users.py --zxb-password '...' --zcf-password '...'
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import shutil
+import sys
+from datetime import datetime
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+ sys.path.insert(0, str(ROOT))
+
+from utils.config import get_userData, save_userData
+from webui.auth import hash_password
+from webui.users import get_web_users, save_web_users
+
+
+BINDINGS = {
+ "zxb": "头像是本人",
+ "zcf": "你成功捕捉一只野生妖孽",
+}
+
+
+def parse_args():
+ parser = argparse.ArgumentParser(description="Initialize SparkFlow Web users and account ownership")
+ parser.add_argument("--zxb-password", default=os.getenv("SPARKFLOW_ZXB_PASSWORD", ""))
+ parser.add_argument("--zcf-password", default=os.getenv("SPARKFLOW_ZCF_PASSWORD", ""))
+ parser.add_argument("--backup-dir", default="")
+ parser.add_argument("--dry-run", action="store_true")
+ return parser.parse_args()
+
+
+def backup_json_files(directory: Path):
+ directory.mkdir(parents=True, exist_ok=True)
+ stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
+ target = directory / stamp
+ target.mkdir(parents=True, exist_ok=True)
+ for relative in ("usersData.json", "webui_settings.json", "webui_users.json"):
+ source = ROOT / relative
+ if source.exists():
+ shutil.copy2(source, target / source.name)
+ return target
+
+
+def main():
+ args = parse_args()
+ passwords = {"zxb": args.zxb_password, "zcf": args.zcf_password}
+ missing = [username for username, password in passwords.items() if not password]
+ if missing:
+ raise SystemExit("Missing password for: " + ", ".join(missing))
+
+ accounts = get_userData(force_reload=True)
+ changed = False
+ for account in accounts:
+ if not str(account.get("account_ref", "")).strip():
+ import uuid
+ account["account_ref"] = f"acc-{uuid.uuid4().hex}"
+ changed = True
+
+ matched = {}
+ for web_username, douyin_username in BINDINGS.items():
+ candidates = [account for account in accounts if str(account.get("username", "")).strip() == douyin_username]
+ if len(candidates) != 1:
+ raise SystemExit(
+ f"Expected exactly one Douyin account named {douyin_username!r} for {web_username}; found {len(candidates)}"
+ )
+ matched[web_username] = candidates[0]
+
+ users = get_web_users()
+ existing = {str(user.get("username", "")).casefold(): user for user in users}
+ assigned_by_other = {
+ ref: user["username"]
+ for user in users
+ for ref in user.get("account_refs", [])
+ if str(user.get("username", "")).casefold() not in BINDINGS
+ }
+ for web_username, account in matched.items():
+ ref = account["account_ref"]
+ owner = assigned_by_other.get(ref)
+ if owner and owner.casefold() != web_username.casefold():
+ raise SystemExit(f"Account {account['username']!r} is already assigned to {owner}")
+ item = existing.get(web_username.casefold())
+ if item is None:
+ item = {
+ "username": web_username,
+ "role": "user",
+ "password_hash": hash_password(passwords[web_username]),
+ "enabled": True,
+ "account_refs": [ref],
+ }
+ users.append(item)
+ else:
+ refs = list(dict.fromkeys(item.get("account_refs", [])))
+ if ref not in refs:
+ refs.append(ref)
+ item["account_refs"] = [ref]
+ item["enabled"] = True
+ changed = True
+
+ if args.dry_run:
+ print(f"accounts={len(accounts)} users={len(users)} matched=zxb,zcf changed={changed}")
+ return 0
+
+ backup_dir = Path(args.backup_dir) if args.backup_dir else ROOT / ".migration-backups"
+ backup_path = backup_json_files(backup_dir)
+ if changed:
+ save_userData(accounts)
+ save_web_users(users)
+ print(f"migration complete: users={len(users)} backup={backup_path}")
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/DouYinSparkFlow/tests/test_config_contract.py b/DouYinSparkFlow/tests/test_config_contract.py
index 0b2361c..ba080c3 100644
--- a/DouYinSparkFlow/tests/test_config_contract.py
+++ b/DouYinSparkFlow/tests/test_config_contract.py
@@ -44,6 +44,11 @@ class ConfigContractTests(unittest.TestCase):
)
self.assertEqual(0, result.returncode, result.stderr)
+ def test_default_schedule_timezone_resolves_without_fallback(self):
+ with patch.dict(os.environ, {"SPARKFLOW_TIMEZONE": ""}, clear=False):
+ schedule_timezone = tasks._schedule_timezone()
+ self.assertEqual("Asia/Shanghai", getattr(schedule_timezone, "key", None))
+
def test_profile_root_environment_override_wins(self):
with patch.dict(os.environ, {"SPARKFLOW_BROWSER_PROFILE_ROOT": "/tmp/sparkflow-profiles"}):
normalized = tasks._normalize_persistent_profile_config(config_module.DEFAULT_CONFIG)
diff --git a/DouYinSparkFlow/tests/test_multiuser.py b/DouYinSparkFlow/tests/test_multiuser.py
new file mode 100644
index 0000000..8d64d6a
--- /dev/null
+++ b/DouYinSparkFlow/tests/test_multiuser.py
@@ -0,0 +1,86 @@
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from webui import login_lock, users
+from webui.auth import hash_password
+
+
+class MultiUserTests(unittest.TestCase):
+ def setUp(self):
+ self.temp_dir = tempfile.TemporaryDirectory()
+ self.users_path = Path(self.temp_dir.name) / "webui_users.json"
+ self.lock_path = Path(self.temp_dir.name) / "login-workspace.lock.json"
+ self.accounts = [
+ {"account_ref": "acc-1", "username": "头像是本人", "unique_id": "111", "targets": [], "enabled": True},
+ {"account_ref": "acc-2", "username": "你成功捕捉一只野生妖孽", "unique_id": "222", "targets": [], "enabled": True},
+ {"account_ref": "acc-3", "username": "管理员账号", "unique_id": "333", "targets": [], "enabled": True},
+ ]
+ self.user_file_patch = patch.object(users, "USERS_FILE", self.users_path)
+ self.user_file_patch.start()
+ self.ensure_patch = patch.object(users, "get_userData", return_value=self.accounts)
+ self.ensure_patch.start()
+ self.save_accounts_patch = patch.object(users, "save_userData")
+ self.save_accounts_patch.start()
+ self.addCleanup(self.ensure_patch.stop)
+ self.addCleanup(self.save_accounts_patch.stop)
+ self.addCleanup(self.user_file_patch.stop)
+ self.addCleanup(self.temp_dir.cleanup)
+
+ def test_user_creation_auth_and_unique_assignment(self):
+ a, changed = users.ensure_account_refs(self.accounts)
+ self.assertFalse(changed)
+ ref = a[0]["account_ref"]
+ created = users.create_web_user("zxb", "zxb123456", account_refs=[ref])
+ self.assertEqual([ref], created["account_refs"])
+ identity = users.authenticate("zxb", "zxb123456")
+ self.assertEqual("user", identity["role"])
+ self.assertEqual([ref], identity["account_refs"])
+ self.assertIsNone(users.authenticate("zxb", "wrong"))
+ with self.assertRaises(users.UserStoreError):
+ users.create_web_user("zcf", "zcf123456", account_refs=[ref])
+
+ def test_visible_accounts_and_admin_reassignment(self):
+ accounts, _ = users.ensure_account_refs(self.accounts)
+ first_ref = accounts[0]["account_ref"]
+ second_ref = accounts[1]["account_ref"]
+ users.create_web_user("zxb", "secret", account_refs=[first_ref])
+ principal = {"role": "user", "account_refs": [first_ref]}
+ self.assertEqual([first_ref], [a["account_ref"] for a in users.get_visible_accounts(principal, accounts)])
+ users.update_web_user("zxb", account_refs=[second_ref])
+ self.assertEqual([second_ref], users.find_web_user("zxb")["account_refs"])
+ self.assertTrue(users.delete_web_user("zxb"))
+ self.assertEqual([], users.get_web_users())
+
+ def test_fifo_queue_promotes_after_active_release(self):
+ with patch.object(login_lock, "LOCK_PATH", self.lock_path):
+ first = login_lock.request_workspace(username="zxb", session_id="s1", account_ref="a1", mode="relogin")
+ second = login_lock.request_workspace(username="zcf", session_id="s2", account_ref="", mode="add")
+ self.assertEqual("active", first["state"])
+ self.assertEqual("queued", second["state"])
+ self.assertEqual("add", second["request"]["mode"])
+ self.assertEqual(1, second["position"])
+ self.assertEqual("queued", login_lock.workspace_status(username="zcf", session_id="s2")["state"])
+ released = login_lock.begin_release(username="zxb", session_id="s1", ticket=first["request"]["ticket"], account_ref="a1")
+ self.assertIsNotNone(released)
+ promoted = login_lock.finish_transition()
+ self.assertEqual("zcf", promoted["username"])
+ self.assertEqual("active", login_lock.workspace_status(username="zcf", session_id="s2")["state"])
+
+ def test_login_workspace_is_serialized_and_expires(self):
+ with patch.object(login_lock, "LOCK_PATH", self.lock_path), patch.object(login_lock, "LOCK_TTL_SECONDS", 1):
+ ok, lock = login_lock.acquire(username="zxb", session_id="s1", account_ref="a1")
+ self.assertTrue(ok)
+ self.assertTrue(login_lock.owns(lock, username="zxb", session_id="s1", account_ref="a1"))
+ blocked, current = login_lock.acquire(username="zcf", session_id="s2", account_ref="a2")
+ self.assertFalse(blocked)
+ self.assertEqual("zxb", current["username"])
+ self.assertTrue(login_lock.refresh(username="zxb", session_id="s1", account_ref="a1"))
+ self.assertTrue(login_lock.release(username="zxb", session_id="s1"))
+ self.assertIsNone(login_lock.get_lock())
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/DouYinSparkFlow/tests/test_webui_safety.py b/DouYinSparkFlow/tests/test_webui_safety.py
index b846e27..4f22ea7 100644
--- a/DouYinSparkFlow/tests/test_webui_safety.py
+++ b/DouYinSparkFlow/tests/test_webui_safety.py
@@ -11,10 +11,23 @@ from fastapi.testclient import TestClient
from core import tasks
from webui import app as app_module
+from webui import login_lock
from webui import ops
class WebUiSafetyTests(unittest.TestCase):
+ def setUp(self):
+ try:
+ login_lock.LOCK_PATH.unlink()
+ except FileNotFoundError:
+ pass
+
+ def tearDown(self):
+ try:
+ login_lock.LOCK_PATH.unlink()
+ except FileNotFoundError:
+ pass
+
def test_windows_invalid_pid_probe_is_treated_as_dead(self):
error = OSError(errno.EINVAL, "invalid pid")
error.winerror = 87
@@ -23,6 +36,19 @@ class WebUiSafetyTests(unittest.TestCase):
with patch.object(tasks.os, "kill", side_effect=error):
self.assertFalse(tasks._pid_is_alive(999999))
+ def test_missing_optional_runtime_tools_do_not_log_warnings(self):
+ with (
+ patch.object(ops.subprocess, "run", side_effect=FileNotFoundError("missing")),
+ patch.object(ops.logger, "warning") as warning,
+ patch.object(ops.logger, "debug") as debug,
+ ):
+ result = ops.run_command(["docker", "ps"])
+ self.assertEqual(1, result.returncode)
+ ops.read_crontab()
+
+ warning.assert_not_called()
+ self.assertGreaterEqual(debug.call_count, 2)
+
def test_stale_lock_inspection_does_not_delete_file(self):
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
@@ -183,6 +209,8 @@ class WebUiSafetyTests(unittest.TestCase):
with (
patch.object(app_module, "current_user", return_value="admin"),
+ patch.object(app_module, "get_login_lock", return_value={"username": "admin", "session_id": ""}),
+ patch.object(app_module, "owns_login_lock", return_value=True),
patch.object(
app_module,
"fetch_login_desktop_asset",
@@ -205,6 +233,8 @@ class WebUiSafetyTests(unittest.TestCase):
upstream.read.return_value = b"fake-png"
with (
patch.object(app_module, "current_user", return_value="admin"),
+ patch.object(app_module, "get_login_lock", return_value={"username": "admin", "session_id": ""}),
+ patch.object(app_module, "owns_login_lock", return_value=True),
patch.object(app_module.urllib.request, "urlopen", return_value=upstream),
):
response = client.get("/login-desktop/qr")
@@ -225,8 +255,10 @@ class WebUiSafetyTests(unittest.TestCase):
block_start = script.index('document.querySelectorAll(".login-desktop-open")')
block_end = script.index('document.querySelectorAll(".login-desktop-save")', block_start)
block = script[block_start:block_end]
- self.assertLess(block.index("window.open(publicUrl"), block.index('await postForm("/login-desktop/open")'))
- self.assertIn("refreshLoginQr(1800)", block)
+ self.assertLess(block.index('window.open("about:blank"'), block.index('postForm("/login-desktop/open"'))
+ self.assertIn("refreshLoginQr(500)", block)
+ self.assertIn('data.state === "queued"', block)
+ self.assertIn("renderWorkspace(data.workspace)", block)
self.assertIn("retries - 1", script)
self.assertIn('/login-desktop/qr/refresh', script)
diff --git a/DouYinSparkFlow/utils/config.py b/DouYinSparkFlow/utils/config.py
index 6bab7ff..1e7cf57 100644
--- a/DouYinSparkFlow/utils/config.py
+++ b/DouYinSparkFlow/utils/config.py
@@ -3,6 +3,8 @@ import logging
import os
import secrets
import sys
+import tempfile
+import uuid
from copy import deepcopy
from enum import Enum
from pathlib import Path
@@ -176,7 +178,20 @@ def _load_json_file(path, defaults=None):
def _save_json_file(path, data):
- path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
+ path.parent.mkdir(parents=True, exist_ok=True)
+ fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent))
+ try:
+ with os.fdopen(fd, "w", encoding="utf-8") as handle:
+ json.dump(data, handle, ensure_ascii=False, indent=2)
+ handle.write("\n")
+ handle.flush()
+ os.fsync(handle.fileno())
+ os.replace(temp_name, path)
+ finally:
+ try:
+ os.unlink(temp_name)
+ except FileNotFoundError:
+ pass
def get_config(force_reload=False):
@@ -230,6 +245,7 @@ def upsert_user_account(unique_id, username, cookies, targets, extra=None):
unique_id = normalize_unique_id(unique_id)
accounts = get_userData(force_reload=True)
payload = {
+ "account_ref": f"acc-{uuid.uuid4().hex}",
"unique_id": unique_id,
"username": username,
"cookies": cookies,
@@ -240,11 +256,12 @@ def upsert_user_account(unique_id, username, cookies, targets, extra=None):
for account in accounts:
if normalize_unique_id(account.get("unique_id")) == unique_id:
+ payload["account_ref"] = account.get("account_ref") or payload["account_ref"]
if "enabled" not in payload:
payload["enabled"] = account.get("enabled", True)
account.update(payload)
save_userData(accounts)
- return payload
+ return account
if "enabled" not in payload:
payload["enabled"] = True
diff --git a/DouYinSparkFlow/webui/app.py b/DouYinSparkFlow/webui/app.py
index 90608f5..ec160f8 100644
--- a/DouYinSparkFlow/webui/app.py
+++ b/DouYinSparkFlow/webui/app.py
@@ -36,6 +36,7 @@ from webui.auth import (
clear_session,
csrf_token,
current_user,
+ current_principal,
is_bootstrapped,
is_https_request,
issue_session,
@@ -43,6 +44,33 @@ from webui.auth import (
validate_csrf,
verify_password,
)
+from webui.users import (
+ UserStoreError,
+ account_by_ref,
+ account_by_unique_id,
+ all_assigned_refs,
+ can_access_account,
+ create_web_user,
+ delete_web_user,
+ ensure_account_refs,
+ get_visible_accounts,
+ get_web_users,
+ remove_account_refs_from_users,
+ update_web_user,
+)
+from webui.login_lock import (
+ begin_expiration as begin_login_expiration,
+ begin_force_reset as begin_login_force_reset,
+ begin_release as begin_login_release,
+ cancel_request as cancel_login_request,
+ finish_transition as finish_login_transition,
+ get_lock as get_login_lock,
+ get_workspace_state,
+ heartbeat as heartbeat_login,
+ owns as owns_login_lock,
+ request_workspace,
+ workspace_status,
+)
from webui.ops import (
TASK_ALREADY_RUNNING,
get_overview_snapshot,
@@ -283,23 +311,42 @@ async def _run_websocket_relays(*coroutines):
await asyncio.gather(*tasks, return_exceptions=True)
-def save_exported_login_result(login_result: dict, *, relogin_unique_id: str = "", display_name: str = "") -> tuple[dict, str]:
+def _dedupe_account_records(accounts: list[dict], *, unique_id: str, keep_ref: str) -> set[str]:
+ normalized = normalize_unique_id(unique_id)
+ removed_refs = set()
+ remaining = []
+ for account in accounts:
+ if normalize_unique_id(account.get("unique_id")) == normalized and str(account.get("account_ref", "")) != str(keep_ref):
+ ref = str(account.get("account_ref", "")).strip()
+ if ref:
+ removed_refs.add(ref)
+ continue
+ remaining.append(account)
+ if len(remaining) != len(accounts):
+ accounts[:] = remaining
+ if removed_refs:
+ remove_account_refs_from_users(removed_refs)
+ return removed_refs
+
+
+def save_exported_login_result(login_result: dict, *, relogin_unique_id: str = "", relogin_account_ref: str = "", display_name: str = "") -> tuple[dict, str]:
unique_id = normalize_unique_id(login_result.get("unique_id"))
username = str(display_name or login_result.get("username") or "").strip()
cookies = list(login_result.get("cookies") or [])
if not unique_id or not username or not cookies:
raise RuntimeError("Exported login result is incomplete")
- accounts = get_userData(force_reload=True)
+ accounts, _ = ensure_account_refs(get_userData(force_reload=True))
- if relogin_unique_id:
- target = find_account(accounts, relogin_unique_id)
+ if relogin_account_ref or relogin_unique_id:
+ target = account_by_ref(accounts, relogin_account_ref) if relogin_account_ref else find_account(accounts, relogin_unique_id)
if not target:
raise RuntimeError("Target account not found for relogin")
target["unique_id"] = unique_id
target["username"] = username
target["cookies"] = cookies
target.setdefault("enabled", True)
+ _dedupe_account_records(accounts, unique_id=unique_id, keep_ref=target.get("account_ref", ""))
save_userData(accounts)
return target, "updated"
@@ -308,10 +355,14 @@ def save_exported_login_result(login_result: dict, *, relogin_unique_id: str = "
existing["username"] = username
existing["cookies"] = cookies
existing.setdefault("enabled", True)
+ _dedupe_account_records(accounts, unique_id=unique_id, keep_ref=existing.get("account_ref", ""))
save_userData(accounts)
return existing, "updated"
account = upsert_user_account(unique_id, username, cookies, [])
+ accounts, _ = ensure_account_refs(get_userData(force_reload=True))
+ _dedupe_account_records(accounts, unique_id=unique_id, keep_ref=account.get("account_ref", ""))
+ save_userData(accounts)
return account, "created"
@@ -336,11 +387,17 @@ def create_app():
@asynccontextmanager
async def lifespan(_app):
+ # Add stable ownership identifiers without changing existing account data.
+ ensure_account_refs()
result = sync_daily_schedule_from_config()
if result.returncode != 0:
logger.warning("Failed to synchronize the configured daily schedule: %s", result.stderr)
- yield
-
+ watchdog = asyncio.create_task(login_workspace_watchdog())
+ try:
+ yield
+ finally:
+ watchdog.cancel()
+ await asyncio.gather(watchdog, return_exceptions=True)
secure_cookie = str(os.getenv("SPARKFLOW_SESSION_COOKIE_SECURE") or "").strip().lower() in {
"1",
"true",
@@ -375,6 +432,8 @@ def create_app():
"current_user": current_user(request),
"csrf_token": csrf_token(request) if current_user(request) else "",
"is_https": is_https_request(request),
+ "principal": current_principal(request),
+ "is_admin": bool(current_principal(request) and current_principal(request).get("role") == "admin"),
"app_settings": public_app_settings(),
"login_desktop_public_url": login_desktop_public_url(request),
}
@@ -390,11 +449,64 @@ def create_app():
def redirect(path="/", status_code=303):
return RedirectResponse(url=path, status_code=status_code)
+ def principal(request):
+ resolved = current_principal(request)
+ if resolved:
+ return resolved
+ # Keep compatibility with older tests/signed sessions that only expose
+ # the legacy ``user`` value.
+ legacy_user = current_user(request)
+ admin_username = str(get_app_settings().get("admin_username", "admin")).strip() or "admin"
+ if legacy_user and str(legacy_user).casefold() == admin_username.casefold():
+ return {"username": admin_username, "role": "admin", "account_refs": [], "session_id": "", "enabled": True}
+ return None
+
def require_user(request):
- if not current_user(request):
+ if not principal(request):
return redirect("/login")
return None
+ def require_admin(request):
+ maybe_redirect = require_user(request)
+ if maybe_redirect:
+ return maybe_redirect
+ if principal(request).get("role") != "admin":
+ return PlainTextResponse("Forbidden", status_code=403)
+ return None
+
+ def account_for_request(request, unique_id):
+ accounts, _ = ensure_account_refs(get_userData(force_reload=True))
+ account = account_by_unique_id(accounts, unique_id)
+ if not account:
+ return accounts, None, PlainTextResponse("Account not found", status_code=404)
+ if not can_access_account(principal(request), account):
+ return accounts, None, PlainTextResponse("Forbidden", status_code=403)
+ return accounts, account, None
+
+ def principal_account_refs(request):
+ current = principal(request)
+ if not current or current.get("role") == "admin":
+ return None
+ return list(current.get("account_refs", []))
+
+ def scoped_ops_snapshot(request):
+ refs = principal_account_refs(request)
+ snapshot = get_ops_snapshot(account_refs=refs)
+ if refs is not None:
+ # Do not place host/container state or global log tails into a
+ # normal user's rendered context.
+ snapshot["containers"] = []
+ snapshot["task_containers"] = []
+ snapshot["crontab"] = ""
+ snapshot["log_tail"] = []
+ snapshot["compose_root"] = ""
+ snapshot["compose_file"] = ""
+ snapshot["image_present"] = False
+ return snapshot
+
+ def scoped_overview_snapshot(request):
+ return get_overview_snapshot(account_refs=principal_account_refs(request))
+
def flash(request, message, level="info"):
request.session["flash"] = {"message": message, "level": level}
@@ -403,7 +515,7 @@ def create_app():
@app.get("/debug-artifacts/{artifact_path:path}")
async def debug_artifact(request: Request, artifact_path: str):
- maybe_redirect = require_user(request)
+ maybe_redirect = require_admin(request)
if maybe_redirect:
return maybe_redirect
root = DEBUG_ARTIFACTS_DIR.resolve()
@@ -452,12 +564,19 @@ def create_app():
form = await request.form()
username = str(form.get("username", "")).strip()
password = str(form.get("password", ""))
- settings = get_app_settings(force_reload=True)
- if username != settings["admin_username"] or not verify_password(password, settings["admin_password_hash"]):
+ from webui.users import authenticate
+
+ identity = authenticate(username, password)
+ if not identity:
flash(request, "Invalid username or password.", "error")
return redirect("/login")
- issue_session(request, username)
+ issue_session(
+ request,
+ identity["username"],
+ role=identity["role"],
+ account_refs=identity.get("account_refs", []),
+ )
flash(request, "Signed in successfully.", "success")
return redirect("/")
@@ -475,24 +594,116 @@ def create_app():
headers={"Cache-Control": "no-store"},
)
return JSONResponse(
- get_overview_snapshot(),
+ scoped_overview_snapshot(request),
headers={"Cache-Control": "no-store"},
)
+ @app.post("/account/password")
+ async def change_own_password(request: Request):
+ maybe_redirect = require_user(request)
+ if maybe_redirect:
+ return maybe_redirect
+ form = await request.form()
+ if not validate_csrf(request, str(form.get("csrf_token", ""))):
+ return Response("Invalid CSRF token", status_code=403)
+ current = principal(request)
+ if current.get("role") == "admin":
+ flash(request, "请在系统设置中修改管理员密码。", "info")
+ return redirect("/")
+ password = str(form.get("new_password", ""))
+ confirm = str(form.get("confirm_password", ""))
+ if not password or password != confirm:
+ flash(request, "两次密码输入不一致。", "error")
+ return redirect("/")
+ try:
+ update_web_user(current["username"], password=password)
+ flash(request, "密码已修改，请重新登录。", "success")
+ clear_session(request)
+ return redirect("/login")
+ except UserStoreError as exc:
+ flash(request, str(exc), "error")
+ return redirect("/")
+
+ @app.post("/admin/users/create")
+ async def create_admin_user(request: Request):
+ maybe_redirect = require_admin(request)
+ if maybe_redirect:
+ return maybe_redirect
+ form = await request.form()
+ if not validate_csrf(request, str(form.get("csrf_token", ""))):
+ return Response("Invalid CSRF token", status_code=403)
+ refs = [value for value in form.getlist("account_refs")] if hasattr(form, "getlist") else []
+ try:
+ create_web_user(
+ str(form.get("username", "")),
+ str(form.get("password", "")),
+ enabled=str(form.get("enabled", "")) == "on",
+ account_refs=refs,
+ )
+ flash(request, "普通用户已创建。", "success")
+ except UserStoreError as exc:
+ flash(request, str(exc), "error")
+ return redirect("/#user-management")
+
+ @app.post("/admin/users/{username}/update")
+ async def update_admin_user(request: Request, username: str):
+ maybe_redirect = require_admin(request)
+ if maybe_redirect:
+ return maybe_redirect
+ form = await request.form()
+ if not validate_csrf(request, str(form.get("csrf_token", ""))):
+ return Response("Invalid CSRF token", status_code=403)
+ refs = [value for value in form.getlist("account_refs")] if hasattr(form, "getlist") else []
+ try:
+ update_web_user(
+ username,
+ new_username=str(form.get("new_username", "")).strip() or None,
+ password=str(form.get("password", "")) or None,
+ enabled=str(form.get("enabled", "")) == "on",
+ account_refs=refs,
+ )
+ flash(request, "普通用户已更新。", "success")
+ except UserStoreError as exc:
+ flash(request, str(exc), "error")
+ return redirect("/#user-management")
+
+ @app.post("/admin/users/{username}/delete")
+ async def delete_admin_user(request: Request, username: str):
+ maybe_redirect = require_admin(request)
+ if maybe_redirect:
+ return maybe_redirect
+ form = await request.form()
+ if not validate_csrf(request, str(form.get("csrf_token", ""))):
+ return Response("Invalid CSRF token", status_code=403)
+ try:
+ if delete_web_user(username):
+ flash(request, "普通用户已删除，抖音账号数据未删除。", "success")
+ else:
+ flash(request, "普通用户不存在。", "error")
+ except UserStoreError as exc:
+ flash(request, str(exc), "error")
+ return redirect("/#user-management")
+
@app.get("/", response_class=HTMLResponse)
async def dashboard(request: Request):
maybe_redirect = require_user(request)
if maybe_redirect:
return maybe_redirect
+ current = principal(request)
+ accounts = get_visible_accounts(current, get_userData(force_reload=True))
return render_template(
request,
"dashboard.html",
{
"flash": pop_flash(request),
- "accounts": get_userData(force_reload=True),
- "runtime_config": get_config(force_reload=True),
- "ops": get_ops_snapshot(),
+ "accounts": accounts,
+ "runtime_config": get_config(force_reload=True) if current.get("role") == "admin" else {},
+ "ops": scoped_ops_snapshot(request),
+ "principal": current,
+ "is_admin": current.get("role") == "admin",
+ "web_users": get_web_users() if current.get("role") == "admin" else [],
+ "all_accounts": get_userData(force_reload=True) if current.get("role") == "admin" else [],
},
)
@@ -507,7 +718,7 @@ def create_app():
"send_console.html",
{
"flash": pop_flash(request),
- "ops": get_ops_snapshot(),
+ "ops": scoped_ops_snapshot(request),
},
)
@@ -524,8 +735,9 @@ def create_app():
username = str(form.get("username", "")).strip()
targets = extract_targets_from_form(form)
- accounts = get_userData(force_reload=True)
- account = find_account(accounts, unique_id)
+ accounts, account, access_error = account_for_request(request, unique_id)
+ if access_error:
+ return access_error
if account:
account["username"] = username or account.get("username", "")
account["targets"] = targets
@@ -547,11 +759,9 @@ def create_app():
if not validate_csrf(request, str(form.get("csrf_token", ""))):
return Response("Invalid CSRF token", status_code=403)
- accounts = get_userData(force_reload=True)
- account = find_account(accounts, unique_id)
- if not account:
- flash(request, "Account not found.", "error")
- return redirect("/")
+ accounts, account, access_error = account_for_request(request, unique_id)
+ if access_error:
+ return access_error
account["enabled"] = not is_account_enabled(account)
save_userData(accounts)
@@ -572,10 +782,9 @@ def create_app():
if not validate_csrf(request, str(form.get("csrf_token", ""))):
return JSONResponse({"error": "Invalid CSRF token"}, status_code=403)
- accounts = get_userData(force_reload=True)
- account = find_account(accounts, unique_id)
- if not account:
- return JSONResponse({"error": "Account not found."}, status_code=404)
+ accounts, account, access_error = account_for_request(request, unique_id)
+ if access_error:
+ return JSONResponse({"error": "Forbidden" if access_error.status_code == 403 else "Account not found."}, status_code=access_error.status_code)
try:
friends = await fetch_account_friends(account)
@@ -594,7 +803,7 @@ def create_app():
@app.post("/accounts/{unique_id}/delete")
async def delete_account(request: Request, unique_id: str):
- maybe_redirect = require_user(request)
+ maybe_redirect = require_admin(request)
if maybe_redirect:
return maybe_redirect
@@ -602,7 +811,9 @@ def create_app():
if not validate_csrf(request, str(form.get("csrf_token", ""))):
return Response("Invalid CSRF token", status_code=403)
- accounts = get_userData(force_reload=True)
+ accounts, account, access_error = account_for_request(request, unique_id)
+ if access_error:
+ return access_error
updated_accounts = [item for item in accounts if normalize_unique_id(item.get("unique_id")) != normalize_unique_id(unique_id)]
if len(updated_accounts) != len(accounts):
save_userData(updated_accounts)
@@ -626,11 +837,9 @@ def create_app():
flash(request, "Target is required for retry.", "error")
return redirect("/ops/send-console")
- accounts = get_userData(force_reload=True)
- account = find_account(accounts, unique_id)
- if not account:
- flash(request, "Account not found.", "error")
- return redirect("/ops/send-console")
+ accounts, account, access_error = account_for_request(request, unique_id)
+ if access_error:
+ return access_error
lock_status = task_run_lock_status()
if lock_status.get("running"):
@@ -682,11 +891,9 @@ def create_app():
flash(request, "Target is required.", "error")
return redirect("/ops/send-console")
- accounts = get_userData(force_reload=True)
- account = find_account(accounts, unique_id)
- if not account:
- flash(request, "Account not found.", "error")
- return redirect("/ops/send-console")
+ accounts, account, access_error = account_for_request(request, unique_id)
+ if access_error:
+ return access_error
changed = mark_target_unconfirmed(account, target_name)
if changed:
@@ -698,7 +905,7 @@ def create_app():
@app.post("/ops/reset-today-unconfirmed")
async def reset_today_unconfirmed(request: Request):
- maybe_redirect = require_user(request)
+ maybe_redirect = require_admin(request)
if maybe_redirect:
return maybe_redirect
@@ -727,7 +934,7 @@ def create_app():
@app.post("/config")
async def save_runtime_config(request: Request):
- maybe_redirect = require_user(request)
+ maybe_redirect = require_admin(request)
if maybe_redirect:
return maybe_redirect
@@ -793,7 +1000,7 @@ def create_app():
@app.post("/settings")
async def save_panel_settings(request: Request):
- maybe_redirect = require_user(request)
+ maybe_redirect = require_admin(request)
if maybe_redirect:
return maybe_redirect
@@ -832,7 +1039,8 @@ def create_app():
if not validate_csrf(request, str(form.get("csrf_token", ""))):
return Response("Invalid CSRF token", status_code=403)
- pid = run_task_now(force_all=True)
+ refs = principal_account_refs(request)
+ pid = run_task_now(force_all=refs is None, account_refs=refs)
if pid == TASK_ALREADY_RUNNING:
flash(request, "已有发送任务正在运行，本次补发全部对象没有启动。请等当前任务结束后再试。", "warning")
elif pid == -1:
@@ -851,7 +1059,8 @@ def create_app():
if not validate_csrf(request, str(form.get("csrf_token", ""))):
return Response("Invalid CSRF token", status_code=403)
- pid = run_failed_retry_now()
+ refs = principal_account_refs(request)
+ pid = run_failed_retry_now(account_refs=refs)
if pid == TASK_ALREADY_RUNNING:
flash(request, "已有发送任务正在运行，本次补发未成功目标没有启动。请等当前任务结束后再试。", "warning")
elif pid == -1:
@@ -870,7 +1079,8 @@ def create_app():
if not validate_csrf(request, str(form.get("csrf_token", ""))):
return Response("Invalid CSRF token", status_code=403)
- pid = run_unsent_retry_now()
+ refs = principal_account_refs(request)
+ pid = run_unsent_retry_now(account_refs=refs)
if pid == TASK_ALREADY_RUNNING:
flash(request, "A send task is already running; unsent retry was not started.", "warning")
elif pid == -1:
@@ -881,7 +1091,7 @@ def create_app():
@app.post("/ops/proxy/refresh")
async def proxy_refresh(request: Request):
- maybe_redirect = require_user(request)
+ maybe_redirect = require_admin(request)
if maybe_redirect:
return maybe_redirect
@@ -895,7 +1105,7 @@ def create_app():
@app.post("/ops/proxy/restart")
async def proxy_restart(request: Request):
- maybe_redirect = require_user(request)
+ maybe_redirect = require_admin(request)
if maybe_redirect:
return maybe_redirect
@@ -909,7 +1119,7 @@ def create_app():
@app.post("/ops/schedule")
async def save_schedule(request: Request):
- maybe_redirect = require_user(request)
+ maybe_redirect = require_admin(request)
if maybe_redirect:
return maybe_redirect
@@ -927,7 +1137,7 @@ def create_app():
@app.get("/ops/logs", response_class=HTMLResponse)
async def logs_page(request: Request):
- maybe_redirect = require_user(request)
+ maybe_redirect = require_admin(request)
if maybe_redirect:
return maybe_redirect
return render_template(
@@ -939,11 +1149,108 @@ def create_app():
},
)
+ login_transition_lock = asyncio.Lock()
+
+ def _workspace_payload(request):
+ current = principal(request)
+ state = get_workspace_state()
+ mine = workspace_status(
+ username=current.get("username", "") if current else "",
+ session_id=current.get("session_id", "") if current else "",
+ )
+ active = state.get("active") or {}
+ is_owner = bool(current and owns_login_lock(
+ active,
+ username=current.get("username", ""),
+ session_id=current.get("session_id", ""),
+ ))
+ return {
+ "state": mine.get("state", "closed"),
+ "position": mine.get("position", 0),
+ "ticket": mine.get("ticket", ""),
+ "remaining_seconds": mine.get("remaining_seconds", 0),
+ "queue_length": len(state.get("queue") or []),
+ "active": is_owner,
+ "active_username": active.get("username", "") if current and current.get("role") == "admin" else (current.get("username", "") if is_owner and current else ""),
+ }
+
+ async def _reset_and_promote(*, force=False, clear_queue=False):
+ """Reset the shared browser profile, then activate the next queue item."""
+ async with login_transition_lock:
+ state = get_workspace_state()
+ if force:
+ transition = begin_login_force_reset(clear_queue=clear_queue)
+ else:
+ transition = begin_login_expiration()
+ state_after = get_workspace_state()
+ needs_reset = bool(transition or state_after.get("phase") == "resetting")
+ if not needs_reset:
+ return True, None
+ try:
+ try:
+ call_login_desktop("/close", method="POST", payload={}, timeout=60)
+ except RuntimeError:
+ # Older login-desktop images do not have /close; reset is
+ # still safe because it clears the temporary login profile.
+ call_login_desktop("/reset", method="POST", payload={}, timeout=120)
+ except RuntimeError as exc:
+ logger.error("Failed to reset login workspace: %s", exc)
+ return False, None
+ promoted = finish_login_transition()
+ if promoted:
+ try:
+ call_login_desktop("/open-login", method="POST", payload={}, timeout=90)
+ except RuntimeError as exc:
+ logger.error("Failed to open login workspace for queued user: %s", exc)
+ return False, promoted
+ return True, promoted
+
+ async def _expire_login_workspace():
+ return await _reset_and_promote()
+
+ async def login_workspace_watchdog():
+ """Reap abandoned leases even when no browser request arrives."""
+ while True:
+ await asyncio.sleep(10)
+ try:
+ await _expire_login_workspace()
+ except asyncio.CancelledError:
+ raise
+ except Exception:
+ logger.exception("login workspace watchdog failed")
+
+ def login_lock_owner(request):
+ current = principal(request)
+ active = get_login_lock()
+ if not current or not active:
+ return current, active, False
+ return current, active, owns_login_lock(
+ active,
+ username=current["username"],
+ session_id=current.get("session_id", ""),
+ )
+
+ def login_lock_required(request, *, api=False):
+ current, active, allowed = login_lock_owner(request)
+ if allowed:
+ return None
+ if api:
+ return JSONResponse({"ok": False, "error": "登录工作区当前未由本会话占用", "workspace": _workspace_payload(request)}, status_code=423)
+ return HTMLResponse(
+ """等待登录工作区
+ 登录工作区尚未分配
+ 请返回账号管理，点击对应抖音账号的“重新登录”。如果前面有其他用户，页面会自动排队等待。
""",
+ status_code=423,
+ )
+
@app.get("/login-desktop/proxy")
async def login_desktop_proxy_root(request: Request):
maybe_redirect = require_user(request)
if maybe_redirect:
return maybe_redirect
+ lock_error = login_lock_required(request)
+ if lock_error:
+ return lock_error
return RedirectResponse(login_desktop_public_url(request), status_code=307)
@app.get("/login-desktop/proxy/{asset_path:path}")
@@ -951,6 +1258,9 @@ def create_app():
maybe_redirect = require_user(request)
if maybe_redirect:
return maybe_redirect
+ lock_error = login_lock_required(request)
+ if lock_error:
+ return lock_error
try:
status, headers, content = await asyncio.to_thread(
fetch_login_desktop_asset,
@@ -963,9 +1273,18 @@ def create_app():
@app.websocket("/login-desktop/proxy/websockify")
async def login_desktop_proxy_websocket(websocket: WebSocket):
- if not current_user(websocket):
+ current = current_principal(websocket)
+ active = get_login_lock()
+ if not current:
await websocket.close(code=4401)
return
+ if not owns_login_lock(
+ active,
+ username=current.get("username", ""),
+ session_id=current.get("session_id", ""),
+ ):
+ await websocket.close(code=4423)
+ return
requested_protocols = [
item.strip()
@@ -1000,10 +1319,7 @@ def create_app():
else:
await websocket.send_text(message)
- await _run_websocket_relays(
- client_to_upstream(),
- upstream_to_client(),
- )
+ await _run_websocket_relays(client_to_upstream(), upstream_to_client())
except (ConnectionClosed, WebSocketDisconnect):
pass
except Exception as exc:
@@ -1016,20 +1332,42 @@ def create_app():
maybe_redirect = require_user(request)
if maybe_redirect:
return maybe_redirect
+ lock_error = login_lock_required(request)
+ if lock_error:
+ return lock_error
url = f"{login_desktop_api_url()}/qr"
try:
upstream_request = urllib.request.Request(url, method="GET")
- content = await asyncio.to_thread(
- lambda: urllib.request.urlopen(upstream_request, timeout=20).read()
- )
- return Response(
- content=content,
- media_type="image/png",
- headers={"Cache-Control": "no-store, max-age=0"},
- )
+ def read_qr_response():
+ upstream = urllib.request.urlopen(upstream_request, timeout=20)
+ try:
+ raw_headers = getattr(upstream, "headers", {})
+ try:
+ headers = dict(raw_headers)
+ except (TypeError, ValueError):
+ headers = {}
+ return getattr(upstream, "status", 200), headers, upstream.read()
+ finally:
+ close = getattr(upstream, "close", None)
+ if close:
+ close()
+ upstream_status, upstream_headers, content = await asyncio.to_thread(read_qr_response)
+ if upstream_status == 202:
+ retry_after = upstream_headers.get("Retry-After", "2")
+ return JSONResponse(
+ {"ok": False, "state": "starting", "retry_after": int(retry_after or 2)},
+ status_code=202,
+ headers={"Retry-After": str(retry_after), "Cache-Control": "no-store"},
+ )
+ return Response(content=content, media_type="image/png", headers={"Cache-Control": "no-store, max-age=0"})
except urllib.error.HTTPError as exc:
- status = exc.code if exc.code in {404, 409} else 502
- return PlainTextResponse("login QR code is not ready", status_code=status)
+ if exc.code in {404, 409, 202}:
+ return JSONResponse(
+ {"ok": False, "state": "starting", "retry_after": 2},
+ status_code=202,
+ headers={"Retry-After": "2", "Cache-Control": "no-store"},
+ )
+ return PlainTextResponse("login QR service is unavailable", status_code=502)
except (urllib.error.URLError, TimeoutError):
return PlainTextResponse("login QR service is unavailable", status_code=502)
@@ -1038,12 +1376,20 @@ def create_app():
maybe_redirect = require_user(request)
if maybe_redirect:
return JSONResponse({"redirect": "/login"}, status_code=401)
+ lock_error = login_lock_required(request, api=True)
+ if lock_error:
+ return lock_error
form = await request.form()
if not validate_csrf(request, str(form.get("csrf_token", ""))):
return JSONResponse({"ok": False, "error": "Invalid CSRF token"}, status_code=403)
+ heartbeat_login(
+ username=principal(request)["username"],
+ session_id=principal(request).get("session_id", ""),
+ ticket=str(form.get("ticket", "")),
+ )
try:
payload = call_login_desktop("/refresh-qr", method="POST", payload={}, timeout=90)
- return JSONResponse({"ok": True, "result": payload})
+ return JSONResponse({"ok": True, "result": payload, "workspace": _workspace_payload(request)})
except RuntimeError as exc:
return JSONResponse({"ok": False, "error": str(exc)}, status_code=503)
@@ -1052,12 +1398,41 @@ def create_app():
maybe_redirect = require_user(request)
if maybe_redirect:
return JSONResponse({"redirect": "/login"}, status_code=401)
+ await _expire_login_workspace()
try:
payload = call_login_desktop("/status")
payload["public_url"] = login_desktop_public_url(request)
+ payload["workspace"] = _workspace_payload(request)
return JSONResponse(payload)
except RuntimeError as exc:
- return JSONResponse({"ok": False, "error": str(exc), "public_url": login_desktop_public_url(request)}, status_code=503)
+ return JSONResponse({"ok": False, "error": str(exc), "public_url": login_desktop_public_url(request), "workspace": _workspace_payload(request)}, status_code=503)
+
+ @app.get("/login-desktop/workspace-status")
+ async def login_desktop_workspace_status(request: Request):
+ maybe_redirect = require_user(request)
+ if maybe_redirect:
+ return JSONResponse({"redirect": "/login"}, status_code=401)
+ await _expire_login_workspace()
+ return JSONResponse({"ok": True, "workspace": _workspace_payload(request)}, headers={"Cache-Control": "no-store"})
+
+ @app.post("/login-desktop/heartbeat")
+ async def login_desktop_heartbeat(request: Request):
+ maybe_redirect = require_user(request)
+ if maybe_redirect:
+ return JSONResponse({"redirect": "/login"}, status_code=401)
+ form = await request.form()
+ if not validate_csrf(request, str(form.get("csrf_token", ""))):
+ return JSONResponse({"ok": False, "error": "Invalid CSRF token"}, status_code=403)
+ await _expire_login_workspace()
+ current = principal(request)
+ ok = heartbeat_login(
+ username=current["username"],
+ session_id=current.get("session_id", ""),
+ ticket=str(form.get("ticket", "")),
+ )
+ if not ok:
+ return JSONResponse({"ok": False, "error": "登录工作区已释放，请重新申请"}, status_code=423)
+ return JSONResponse({"ok": True, "workspace": _workspace_payload(request)})
@app.post("/login-desktop/open")
async def login_desktop_open(request: Request):
@@ -1067,12 +1442,64 @@ def create_app():
form = await request.form()
if not validate_csrf(request, str(form.get("csrf_token", ""))):
return JSONResponse({"ok": False, "error": "Invalid CSRF token"}, status_code=403)
+ await _expire_login_workspace()
+ current = principal(request)
+ relogin_unique_id = str(form.get("relogin_unique_id", "")).strip()
+ requested_mode = str(form.get("mode", "")).strip().lower()
+ mode = requested_mode if requested_mode in {"add", "relogin"} else ("relogin" if relogin_unique_id else "add")
+ account_ref = ""
+ if relogin_unique_id:
+ _, account, access_error = account_for_request(request, relogin_unique_id)
+ if access_error:
+ return JSONResponse({"ok": False, "error": "无权操作该账号"}, status_code=access_error.status_code)
+ account_ref = account.get("account_ref", "")
+ mode = "relogin"
+ elif mode != "add":
+ return JSONResponse({"ok": False, "error": "重新登录已有账号时必须选择账号"}, status_code=400)
+
+ result = request_workspace(
+ username=current["username"],
+ session_id=current.get("session_id", ""),
+ account_ref=account_ref,
+ mode=mode,
+ )
+ if result["state"] == "full":
+ return JSONResponse({"ok": False, "error": "登录排队人数已满，请稍后重试"}, status_code=429)
+ if result["state"] == "queued":
+ return JSONResponse({"ok": True, "state": "queued", "workspace": _workspace_payload(request)}, status_code=202)
try:
call_login_desktop("/open-login", method="POST", payload={}, timeout=90)
- return JSONResponse({"ok": True, "public_url": login_desktop_public_url(request)})
+ return JSONResponse({"ok": True, "state": "active", "public_url": login_desktop_public_url(request), "workspace": _workspace_payload(request)})
except RuntimeError as exc:
+ begin_login_release(username=current["username"], session_id=current.get("session_id", ""), ticket=result["request"].get("ticket", ""), account_ref=account_ref)
+ await _reset_and_promote()
return JSONResponse({"ok": False, "error": str(exc)}, status_code=503)
+ @app.post("/login-desktop/close")
+ async def login_desktop_close(request: Request):
+ maybe_redirect = require_user(request)
+ if maybe_redirect:
+ return JSONResponse({"redirect": "/login"}, status_code=401)
+ form = await request.form()
+ if not validate_csrf(request, str(form.get("csrf_token", ""))):
+ return JSONResponse({"ok": False, "error": "Invalid CSRF token"}, status_code=403)
+ current = principal(request)
+ if current.get("role") == "admin":
+ await _reset_and_promote(force=True)
+ return JSONResponse({"ok": True, "workspace": _workspace_payload(request)})
+ active = get_login_lock()
+ if owns_login_lock(active, username=current["username"], session_id=current.get("session_id", "")):
+ begin_login_release(
+ username=current["username"],
+ session_id=current.get("session_id", ""),
+ ticket=active.get("ticket", ""),
+ account_ref=active.get("account_ref", ""),
+ )
+ await _reset_and_promote()
+ else:
+ cancel_login_request(username=current["username"], session_id=current.get("session_id", ""))
+ return JSONResponse({"ok": True, "workspace": _workspace_payload(request)})
+
@app.post("/login-desktop/reset")
async def login_desktop_reset(request: Request):
maybe_redirect = require_user(request)
@@ -1081,11 +1508,17 @@ def create_app():
form = await request.form()
if not validate_csrf(request, str(form.get("csrf_token", ""))):
return JSONResponse({"ok": False, "error": "Invalid CSRF token"}, status_code=403)
- try:
- payload = call_login_desktop("/reset", method="POST", payload={}, timeout=120)
- return JSONResponse({"ok": True, "result": payload})
- except RuntimeError as exc:
- return JSONResponse({"ok": False, "error": str(exc)}, status_code=503)
+ current = principal(request)
+ if current.get("role") == "admin":
+ await _reset_and_promote(force=True, clear_queue=str(form.get("clear_queue", "")) == "1")
+ return JSONResponse({"ok": True, "workspace": _workspace_payload(request)})
+ active = get_login_lock()
+ if not owns_login_lock(active, username=current["username"], session_id=current.get("session_id", "")):
+ kind, _ = cancel_login_request(username=current["username"], session_id=current.get("session_id", ""))
+ return JSONResponse({"ok": kind == "queued", "workspace": _workspace_payload(request)})
+ begin_login_release(username=current["username"], session_id=current.get("session_id", ""), ticket=active.get("ticket", ""), account_ref=active.get("account_ref", ""))
+ await _reset_and_promote()
+ return JSONResponse({"ok": True, "workspace": _workspace_payload(request)})
@app.post("/login-desktop/save")
async def login_desktop_save(request: Request):
@@ -1096,25 +1529,55 @@ def create_app():
if not validate_csrf(request, str(form.get("csrf_token", ""))):
return JSONResponse({"ok": False, "error": "Invalid CSRF token"}, status_code=403)
+ current = principal(request)
+ active = get_login_lock()
+ if not owns_login_lock(active, username=current["username"], session_id=current.get("session_id", "")):
+ return JSONResponse({"ok": False, "error": "登录工作区已释放，请重新申请"}, status_code=423)
relogin_unique_id = str(form.get("relogin_unique_id", "")).strip()
display_name = str(form.get("display_name", "")).strip()
+ operation = str(active.get("mode", "relogin"))
+ relogin_account_ref = str(active.get("account_ref", ""))
+ if relogin_account_ref:
+ account = account_by_ref(get_userData(force_reload=True), relogin_account_ref)
+ if not account or not can_access_account(current, account):
+ return JSONResponse({"ok": False, "error": "无权操作该账号"}, status_code=403)
+ relogin_unique_id = account.get("unique_id", "")
+ operation = "relogin"
+ elif operation != "add" and current.get("role") != "admin":
+ return JSONResponse({"ok": False, "error": "普通用户必须选择自己的抖音账号"}, status_code=400)
try:
payload = call_login_desktop("/export", method="POST", payload={}, timeout=30)
if not payload.get("ok"):
raise RuntimeError("login-desktop export did not return ok")
+ exported = payload.get("result", {}) or {}
+ existing = account_by_unique_id(get_userData(force_reload=True), exported.get("unique_id"))
+ if existing and str(existing.get("account_ref", "")) != relogin_account_ref and not can_access_account(current, existing):
+ raise RuntimeError("这个抖音账号已经绑定给其他用户，不能覆盖")
+ if operation == "add" and current.get("role") == "user" and existing:
+ relogin_account_ref = existing.get("account_ref", "")
+ relogin_unique_id = existing.get("unique_id", "")
+ operation = "relogin"
account, action = save_exported_login_result(
- payload.get("result", {}),
+ exported,
relogin_unique_id=relogin_unique_id,
+ relogin_account_ref=relogin_account_ref,
display_name=display_name,
)
+ if operation == "add" and current.get("role") == "user":
+ refs = list(dict.fromkeys(list(current.get("account_refs", [])) + [account.get("account_ref", "")]))
+ update_web_user(current["username"], account_refs=refs)
+ begin_login_release(username=current["username"], session_id=current.get("session_id", ""), ticket=active.get("ticket", ""), account_ref=active.get("account_ref", ""))
+ await _reset_and_promote()
return JSONResponse({
"ok": True,
"action": action,
"account": {
+ "account_ref": account.get("account_ref"),
"unique_id": account.get("unique_id"),
"username": account.get("username"),
"enabled": account.get("enabled", True),
},
+ "workspace": _workspace_payload(request),
})
except RuntimeError as exc:
return JSONResponse({"ok": False, "error": str(exc)}, status_code=400)
diff --git a/DouYinSparkFlow/webui/auth.py b/DouYinSparkFlow/webui/auth.py
index 5488a75..70810e2 100644
--- a/DouYinSparkFlow/webui/auth.py
+++ b/DouYinSparkFlow/webui/auth.py
@@ -14,8 +14,10 @@ def hash_password(password, salt=None):
def verify_password(password, stored_hash):
if not stored_hash or "$" not in stored_hash:
return False
-
- algorithm, salt, digest = stored_hash.split("$", 2)
+ try:
+ algorithm, salt, digest = stored_hash.split("$", 2)
+ except ValueError:
+ return False
if algorithm != "pbkdf2_sha256":
return False
candidate = hash_password(password, salt=salt)
@@ -39,9 +41,12 @@ def update_admin_password(password):
return save_app_settings(settings)
-def issue_session(request, username):
+def issue_session(request, username, *, role="admin", account_refs=None):
request.session.clear()
request.session["user"] = username
+ request.session["role"] = role
+ request.session["account_refs"] = list(account_refs or [])
+ request.session["session_id"] = secrets.token_urlsafe(24)
request.session["csrf_token"] = secrets.token_urlsafe(24)
@@ -53,6 +58,44 @@ def current_user(request):
return request.session.get("user")
+def current_principal(request):
+ """Resolve the current session to live role and account assignments."""
+ username = current_user(request)
+ if not username:
+ return None
+ session_id = request.session.get("session_id", "")
+ role = request.session.get("role")
+ admin_username = str(get_app_settings().get("admin_username", "admin")).strip() or "admin"
+ if role == "admin" or (role is None and username.casefold() == admin_username.casefold()):
+ return {
+ "username": admin_username,
+ "role": "admin",
+ "account_refs": [],
+ "session_id": session_id,
+ "enabled": True,
+ }
+
+ try:
+ from webui.users import find_web_user
+ user = find_web_user(username)
+ except Exception:
+ user = None
+ if user and user.get("enabled", True):
+ return {
+ "username": user["username"],
+ "role": "user",
+ "account_refs": list(user.get("account_refs", [])),
+ "session_id": session_id,
+ "enabled": True,
+ }
+ return None
+
+
+def is_admin(request):
+ principal = current_principal(request)
+ return bool(principal and principal.get("role") == "admin")
+
+
def csrf_token(request):
token = request.session.get("csrf_token")
if not token:
diff --git a/DouYinSparkFlow/webui/login_lock.py b/DouYinSparkFlow/webui/login_lock.py
new file mode 100644
index 0000000..f97c4ed
--- /dev/null
+++ b/DouYinSparkFlow/webui/login_lock.py
@@ -0,0 +1,346 @@
+"""Shared login-desktop workspace lease and FIFO queue."""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import threading
+import time
+import uuid
+from copy import deepcopy
+from pathlib import Path
+
+from utils.config import repo_root
+
+
+def _default_lock_path() -> Path:
+ candidates = [
+ Path("/opt/douyin-sparkflow/state"),
+ repo_root().parent / "state",
+ repo_root() / "state",
+ ]
+ for candidate in candidates:
+ if candidate.exists() or candidate.parent.exists():
+ return candidate / "login-workspace.lock.json"
+ return candidates[0] / "login-workspace.lock.json"
+
+
+LOCK_PATH = _default_lock_path()
+LOCK_TTL_SECONDS = 180
+QUEUE_MAX_SIZE = 32
+_MUTEX = threading.Lock()
+
+
+def _empty_state() -> dict:
+ return {"version": 2, "phase": "idle", "active": None, "queue": []}
+
+
+def _read_raw() -> dict | None:
+ try:
+ data = json.loads(LOCK_PATH.read_text(encoding="utf-8"))
+ return data if isinstance(data, dict) else None
+ except (FileNotFoundError, json.JSONDecodeError, OSError):
+ return None
+
+
+def _normalize_state(data: dict | None) -> dict:
+ if not data:
+ return _empty_state()
+ if "active" in data or "queue" in data:
+ state = _empty_state()
+ state.update({key: data.get(key) for key in ("version", "phase", "active", "queue")})
+ state["version"] = 2
+ state["phase"] = str(state.get("phase") or ("active" if state.get("active") else "idle"))
+ state["queue"] = [item for item in (state.get("queue") or []) if isinstance(item, dict)]
+ return state
+ # Backward compatibility with the original single-lock file.
+ legacy = dict(data)
+ now = time.time()
+ legacy.setdefault("ticket", "legacy-" + uuid.uuid4().hex)
+ legacy.setdefault("requested_at", now)
+ legacy.setdefault("started_at", legacy.get("acquired_at", now))
+ legacy.setdefault("last_heartbeat_at", legacy.get("acquired_at", now))
+ return {"version": 2, "phase": "active", "active": legacy, "queue": []}
+
+
+def _write_state(state: dict) -> None:
+ LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
+ fd, temp_name = tempfile.mkstemp(prefix=".login-workspace.", suffix=".tmp", dir=str(LOCK_PATH.parent))
+ try:
+ with os.fdopen(fd, "w", encoding="utf-8") as handle:
+ json.dump(state, handle, ensure_ascii=False, indent=2)
+ handle.write("\n")
+ handle.flush()
+ os.fsync(handle.fileno())
+ os.replace(temp_name, LOCK_PATH)
+ finally:
+ try:
+ os.unlink(temp_name)
+ except FileNotFoundError:
+ pass
+
+
+def _delete_state() -> None:
+ try:
+ LOCK_PATH.unlink()
+ except FileNotFoundError:
+ pass
+
+
+def _now() -> float:
+ return time.time()
+
+
+def _expired(active: dict | None, now: float | None = None) -> bool:
+ if not active:
+ return False
+ now = _now() if now is None else now
+ try:
+ return now - float(active.get("last_heartbeat_at", active.get("acquired_at", 0))) > LOCK_TTL_SECONDS
+ except (TypeError, ValueError):
+ return True
+
+
+def _public_item(item: dict | None) -> dict | None:
+ if not item:
+ return None
+ return {
+ "ticket": item.get("ticket", ""),
+ "username": item.get("username", ""),
+ "account_ref": item.get("account_ref", ""),
+ "requested_at": item.get("requested_at", 0),
+ "started_at": item.get("started_at", 0),
+ "last_heartbeat_at": item.get("last_heartbeat_at", 0),
+ }
+
+
+def get_workspace_state() -> dict:
+ with _MUTEX:
+ return deepcopy(_normalize_state(_read_raw()))
+
+
+def get_lock() -> dict | None:
+ """Compatibility helper: return the current active lease only."""
+ return get_workspace_state().get("active")
+
+
+def owns(lock: dict | None, *, username: str, session_id: str, account_ref: str | None = None, ticket: str | None = None) -> bool:
+ if not lock:
+ return False
+ if str(lock.get("username")) != str(username) or str(lock.get("session_id")) != str(session_id):
+ return False
+ if account_ref is not None and str(lock.get("account_ref", "")) != str(account_ref):
+ return False
+ if ticket is not None and str(lock.get("ticket", "")) != str(ticket):
+ return False
+ return True
+
+
+def _new_item(username: str, session_id: str, account_ref: str, now: float, mode: str = "relogin") -> dict:
+ return {
+ "ticket": "ticket-" + uuid.uuid4().hex,
+ "mode": mode if mode in {"relogin", "add"} else "relogin",
+ "username": username,
+ "session_id": session_id,
+ "account_ref": account_ref,
+ "requested_at": now,
+ "started_at": now,
+ "last_heartbeat_at": now,
+ }
+
+
+def find_request(state: dict, *, username: str, session_id: str) -> tuple[str, dict | None, int]:
+ active = state.get("active")
+ if active and str(active.get("username")) == str(username) and str(active.get("session_id")) == str(session_id):
+ return "active", active, 0
+ for index, item in enumerate(state.get("queue") or [], start=1):
+ if str(item.get("username")) == str(username) and str(item.get("session_id")) == str(session_id):
+ return "queued", item, index
+ return "none", None, -1
+
+
+def request_workspace(*, username: str, session_id: str, account_ref: str = "", mode: str = "relogin") -> dict:
+ with _MUTEX:
+ state = _normalize_state(_read_raw())
+ now = _now()
+ if state.get("phase") == "resetting":
+ queue = state.setdefault("queue", [])
+ kind, existing, position = find_request(state, username=username, session_id=session_id)
+ if kind == "queued":
+ return {"state": "queued", "position": position, "request": deepcopy(existing), "workspace": state}
+ if len(queue) >= QUEUE_MAX_SIZE:
+ return {"state": "full", "position": -1, "request": None, "workspace": state}
+ item = _new_item(username, session_id, account_ref, now, mode=mode)
+ item["started_at"] = 0
+ item["last_heartbeat_at"] = 0
+ queue.append(item)
+ _write_state(state)
+ return {"state": "queued", "position": len(queue), "request": deepcopy(item), "workspace": state}
+ elif state.get("active") and not _expired(state["active"], now):
+ kind, existing, position = find_request(state, username=username, session_id=session_id)
+ if kind == "active":
+ existing["last_heartbeat_at"] = now
+ _write_state(state)
+ return {"state": "active", "position": 0, "request": deepcopy(existing), "workspace": state}
+ if kind == "queued":
+ return {"state": "queued", "position": position, "request": deepcopy(existing), "workspace": state}
+ queue = state.setdefault("queue", [])
+ if len(queue) >= QUEUE_MAX_SIZE:
+ return {"state": "full", "position": -1, "request": None, "workspace": state}
+ item = _new_item(username, session_id, account_ref, now, mode=mode)
+ item["started_at"] = 0
+ item["last_heartbeat_at"] = 0
+ queue.append(item)
+ _write_state(state)
+ return {"state": "queued", "position": len(queue), "request": deepcopy(item), "workspace": state}
+ else:
+ item = _new_item(username, session_id, account_ref, now, mode=mode)
+ state = {"version": 2, "phase": "active", "active": item, "queue": state.get("queue", [])}
+ _write_state(state)
+ return {"state": "active", "position": 0, "request": deepcopy(item), "workspace": state}
+
+
+def heartbeat(*, username: str, session_id: str, ticket: str = "", account_ref: str = "") -> bool:
+ with _MUTEX:
+ state = _normalize_state(_read_raw())
+ active = state.get("active")
+ if not owns(active, username=username, session_id=session_id, account_ref=account_ref or None, ticket=ticket or None):
+ return False
+ active["last_heartbeat_at"] = _now()
+ _write_state(state)
+ return True
+
+
+def begin_expiration() -> dict | None:
+ with _MUTEX:
+ state = _normalize_state(_read_raw())
+ if state.get("phase") != "active" or not _expired(state.get("active")):
+ return None
+ old = deepcopy(state.get("active"))
+ state["phase"] = "resetting"
+ state["active"] = None
+ state["transition_reason"] = "heartbeat_timeout"
+ state["transition_at"] = _now()
+ _write_state(state)
+ return old
+
+
+def begin_force_reset(*, clear_queue: bool = False) -> dict | None:
+ with _MUTEX:
+ state = _normalize_state(_read_raw())
+ active = deepcopy(state.get("active"))
+ if not active and not state.get("queue"):
+ return None
+ state["phase"] = "resetting"
+ state["active"] = None
+ if clear_queue:
+ state["queue"] = []
+ state["transition_reason"] = "admin_reset"
+ state["transition_at"] = _now()
+ _write_state(state)
+ return active
+
+
+def begin_release(*, username: str, session_id: str, ticket: str = "", account_ref: str = "") -> dict | None:
+ with _MUTEX:
+ state = _normalize_state(_read_raw())
+ active = state.get("active")
+ if not owns(active, username=username, session_id=session_id, account_ref=account_ref or None, ticket=ticket or None):
+ return None
+ old = deepcopy(active)
+ state["phase"] = "resetting"
+ state["active"] = None
+ state["transition_reason"] = "released"
+ state["transition_at"] = _now()
+ _write_state(state)
+ return old
+
+
+def finish_transition() -> dict | None:
+ with _MUTEX:
+ state = _normalize_state(_read_raw())
+ queue = state.get("queue") or []
+ if queue:
+ item = queue.pop(0)
+ now = _now()
+ item["started_at"] = now
+ item["last_heartbeat_at"] = now
+ state["phase"] = "active"
+ state["active"] = item
+ state["queue"] = queue
+ state.pop("transition_reason", None)
+ state.pop("transition_at", None)
+ _write_state(state)
+ return deepcopy(item)
+ _delete_state()
+ return None
+
+
+def cancel_request(*, username: str, session_id: str, ticket: str = "") -> tuple[str, dict | None]:
+ with _MUTEX:
+ state = _normalize_state(_read_raw())
+ active = state.get("active")
+ if owns(active, username=username, session_id=session_id, ticket=ticket or None):
+ state["phase"] = "resetting"
+ state["active"] = None
+ state["transition_reason"] = "cancelled"
+ state["transition_at"] = _now()
+ _write_state(state)
+ return "active", deepcopy(active)
+ queue = state.get("queue") or []
+ for index, item in enumerate(queue):
+ if str(item.get("username")) == str(username) and str(item.get("session_id")) == str(session_id) and (not ticket or str(item.get("ticket")) == str(ticket)):
+ removed = queue.pop(index)
+ state["queue"] = queue
+ if queue or state.get("active"):
+ _write_state(state)
+ else:
+ _delete_state()
+ return "queued", deepcopy(removed)
+ return "none", None
+
+
+def workspace_status(*, username: str, session_id: str) -> dict:
+ state = get_workspace_state()
+ kind, item, position = find_request(state, username=username, session_id=session_id)
+ if kind == "active":
+ remaining = max(0, int(LOCK_TTL_SECONDS - (_now() - float(item.get("last_heartbeat_at", _now())))))
+ return {"state": "active", "position": 0, "ticket": item.get("ticket", ""), "remaining_seconds": remaining, "workspace": state}
+ if kind == "queued":
+ return {"state": "queued", "position": position, "ticket": item.get("ticket", ""), "remaining_seconds": 0, "workspace": state}
+ if state.get("phase") == "resetting":
+ return {"state": "resetting", "position": 0, "ticket": "", "remaining_seconds": 0, "workspace": state}
+ return {"state": "closed", "position": 0, "ticket": "", "remaining_seconds": 0, "workspace": state}
+
+
+# Backward-compatible single-lease helpers used by existing tests and callers.
+def acquire(*, username: str, session_id: str, account_ref: str = "", force: bool = False) -> tuple[bool, dict | None]:
+ with _MUTEX:
+ state = _normalize_state(_read_raw())
+ if force:
+ state = _empty_state()
+ active = state.get("active")
+ if active and not owns(active, username=username, session_id=session_id, account_ref=account_ref):
+ return False, active
+ result = request_workspace(username=username, session_id=session_id, account_ref=account_ref)
+ return result["state"] == "active", result.get("request")
+
+
+def refresh(*, username: str, session_id: str, account_ref: str = "") -> bool:
+ return heartbeat(username=username, session_id=session_id, account_ref=account_ref)
+
+
+def release(*, username: str | None = None, session_id: str | None = None, force: bool = False) -> bool:
+ with _MUTEX:
+ state = _normalize_state(_read_raw())
+ active = state.get("active")
+ if not active:
+ return False
+ if not force:
+ if username is not None and str(active.get("username")) != str(username):
+ return False
+ if session_id is not None and str(active.get("session_id")) != str(session_id):
+ return False
+ _delete_state()
+ return True
diff --git a/DouYinSparkFlow/webui/ops.py b/DouYinSparkFlow/webui/ops.py
index 703df29..4d41c04 100644
--- a/DouYinSparkFlow/webui/ops.py
+++ b/DouYinSparkFlow/webui/ops.py
@@ -259,7 +259,11 @@ def run_command(args, cwd=None, timeout=120, check=False):
timeout=timeout,
)
except FileNotFoundError:
- logger.warning("Command not found: %s", args[0] if args else args)
+ # Docker and cron are optional integration points when the UI is run
+ # directly on a developer workstation (especially on Windows). A
+ # status probe must not turn their absence into a warning on every
+ # dashboard refresh.
+ logger.debug("Optional command not found: %s", args[0] if args else args)
return _empty_result()
except subprocess.TimeoutExpired:
logger.warning("Command timed out: %s", args)
@@ -359,7 +363,7 @@ def get_task_container_rows():
return []
-def run_task_now(*, unsent_only=False, failed_only=False, force_all=False):
+def run_task_now(*, unsent_only=False, failed_only=False, force_all=False, account_refs=None):
try:
lock_status = task_run_lock_status()
if lock_status.get("running"):
@@ -376,6 +380,8 @@ def run_task_now(*, unsent_only=False, failed_only=False, force_all=False):
"SPARKFLOW_MANUAL_RUN": "1",
"PYTHONUNBUFFERED": "1",
}
+ if account_refs is not None:
+ run_env["SPARKFLOW_ACCOUNT_REFS"] = ",".join(sorted({str(ref).strip() for ref in account_refs if str(ref).strip()}))
if force_all:
run_env["SPARKFLOW_MANUAL_FORCE_ALL"] = "1"
elif failed_only:
@@ -402,12 +408,12 @@ def run_task_now(*, unsent_only=False, failed_only=False, force_all=False):
return -1
-def run_failed_retry_now():
- return run_task_now(failed_only=True)
+def run_failed_retry_now(*, account_refs=None):
+ return run_task_now(failed_only=True, account_refs=account_refs)
-def run_unsent_retry_now():
- return run_task_now(unsent_only=True)
+def run_unsent_retry_now(*, account_refs=None):
+ return run_task_now(unsent_only=True, account_refs=account_refs)
def refresh_proxy():
@@ -447,6 +453,12 @@ def read_crontab():
if result.returncode != 0:
return ""
return result.stdout
+ except FileNotFoundError:
+ # Native Windows installs do not provide ``crontab``. The caller can
+ # treat an unavailable scheduler as an empty schedule and still serve
+ # the rest of the dashboard.
+ logger.debug("Optional command not found: crontab")
+ return ""
except Exception as exc:
logger.warning("read_crontab failed: %s", exc)
return ""
@@ -927,8 +939,13 @@ def _orphan_records(account, configured_targets):
return orphan_history, orphan_failure
-def get_send_console_snapshot():
- accounts = [account for account in get_userData(force_reload=True) if account.get("enabled", True)]
+def get_send_console_snapshot(account_refs=None):
+ allowed_refs = None if account_refs is None else {str(ref).strip() for ref in account_refs}
+ accounts = [
+ account
+ for account in get_userData(force_reload=True)
+ if account.get("enabled", True) and (allowed_refs is None or account.get("account_ref") in allowed_refs)
+ ]
send_window = _normalize_send_window()
now = datetime.now(_schedule_timezone())
@@ -1049,6 +1066,7 @@ def get_send_console_snapshot():
account_rows.append(
{
+ "account_ref": str(account.get("account_ref") or ""),
"unique_id": str(account.get("unique_id") or ""),
"username": account.get("username") or "",
"total_targets": len(configured_targets),
@@ -1100,8 +1118,8 @@ def get_send_console_snapshot():
}
-def get_overview_snapshot():
- send_console = get_send_console_snapshot()
+def get_overview_snapshot(account_refs=None):
+ send_console = get_send_console_snapshot(account_refs=account_refs)
summary = dict(send_console["summary"])
accounts = []
for row in send_console["accounts"]:
@@ -1154,13 +1172,13 @@ def _check_image_present():
return False
-def get_ops_snapshot():
+def get_ops_snapshot(account_refs=None):
"""Collect operational metrics for the dashboard.
Every external call is individually guarded so the dashboard always
renders, even when Docker or crontab are not available.
"""
- send_console = get_send_console_snapshot()
+ send_console = get_send_console_snapshot(account_refs=account_refs)
return {
"compose_root": str(compose_root()),
"compose_file": str(compose_file_path() or ""),
diff --git a/DouYinSparkFlow/webui/static/app.css b/DouYinSparkFlow/webui/static/app.css
index ddf0845..fd8203f 100644
--- a/DouYinSparkFlow/webui/static/app.css
+++ b/DouYinSparkFlow/webui/static/app.css
@@ -988,6 +988,26 @@ textarea {
accent-color: var(--primary);
}
+.account-assignment-picker {
+ border: 0;
+ padding: 0;
+ margin: 0;
+}
+
+.account-assignment-picker legend {
+ padding: 0;
+ color: var(--text-muted);
+ font-size: 12px;
+ font-weight: 700;
+}
+
+.account-assignment-picker .check-row {
+ padding: 8px 10px;
+ border: 1px solid var(--border);
+ border-radius: 7px;
+ background: var(--surface-alt);
+}
+
.button-row,
.button-grid,
.form-actions {
diff --git a/DouYinSparkFlow/webui/static/app.js b/DouYinSparkFlow/webui/static/app.js
index 04bd633..5f1b07e 100644
--- a/DouYinSparkFlow/webui/static/app.js
+++ b/DouYinSparkFlow/webui/static/app.js
@@ -339,58 +339,82 @@
const section = document.getElementById("interactive-login-section");
const csrfToken = root.dataset.csrfToken || "";
const publicUrl = root.dataset.publicUrl || "";
- const runtimeState = document.getElementById(
- "login-desktop-runtime-state",
- );
+ const runtimeState = document.getElementById("login-desktop-runtime-state");
const statusText = document.getElementById("login-desktop-status-text");
const frame = document.querySelector("[data-login-frame]");
const qrImage = document.querySelector("[data-login-qr]");
const qrStatus = document.querySelector("[data-login-qr-status]");
let timer = null;
+ let heartbeatTimer = null;
+ let countdownTimer = null;
let qrRefreshTimer = null;
+ let workspace = { state: "closed", active: false, position: 0, ticket: "" };
const setStatus = (text, tone = "") => {
if (statusText) statusText.textContent = text;
if (runtimeState) {
runtimeState.className = `pill${tone ? ` ${tone}` : ""}`;
- runtimeState.textContent =
- tone === "success" ? "已登录" : tone === "danger" ? "异常" : "待登录";
+ runtimeState.textContent = tone === "success" ? "使用中" : tone === "danger" ? "异常" : tone === "warning" ? "排队中" : "已关闭";
}
};
const postForm = async (url, payload = {}) => {
const formData = new FormData();
formData.set("csrf_token", csrfToken);
- Object.entries(payload).forEach(([key, value]) => {
- formData.set(key, String(value ?? ""));
- });
- const response = await fetch(url, {
- method: "POST",
- body: formData,
- credentials: "same-origin",
- });
+ Object.entries(payload).forEach(([key, value]) => formData.set(key, String(value ?? "")));
+ const response = await fetch(url, { method: "POST", body: formData, credentials: "same-origin" });
const data = await response.json().catch(() => ({}));
- if (!response.ok || data.ok === false) {
- throw new Error(data.error || `请求失败：${response.status}`);
- }
+ if (!response.ok || data.ok === false) throw new Error(data.error || `请求失败：${response.status}`);
return data;
};
- const loadFrame = () => {
- if (frame && frame.dataset.loaded !== "1" && frame.dataset.src) {
+ const loadFrame = (force = false) => {
+ if (frame && (force || frame.dataset.loaded !== "1") && frame.dataset.src) {
frame.src = frame.dataset.src;
frame.dataset.loaded = "1";
}
};
- const refreshLoginQr = async (delay = 0, retries = 8) => {
- if (!qrImage) return;
+ const closeFrame = () => {
+ if (!frame) return;
+ frame.removeAttribute("src");
+ frame.dataset.loaded = "0";
+ };
+
+ const renderWorkspace = (next) => {
+ workspace = next || { state: "closed", active: false, position: 0, ticket: "" };
+ if (workspace.state === "queued") {
+ setStatus(`登录工作区排队中，前面还有 ${Math.max(0, Number(workspace.position || 1) - 1)} 人。`, "warning");
+ if (qrStatus) qrStatus.textContent = "排队成功，轮到你后会自动打开登录二维码。";
+ return;
+ }
+ if (workspace.state === "resetting") {
+ setStatus("正在清理上一位用户的登录环境，请稍候。", "warning");
+ return;
+ }
+ if (workspace.state === "active" && workspace.active) {
+ const remaining = Math.max(0, Number(workspace.remaining_seconds || 0));
+ const tone = remaining > 0 && remaining <= 60 ? "warning" : "success";
+ setStatus(`登录工作区已分配给当前会话，剩余 ${remaining} 秒。完成扫码后请保存登录态。`, tone);
+ return;
+ }
+ setStatus("登录工作区当前关闭。请从账号卡片点击“重新登录”。");
+ };
+
+ const refreshLoginQr = async (delay = 0, retries = 40) => {
+ if (!qrImage || workspace.state !== "active" || !workspace.active) return;
window.clearTimeout(qrRefreshTimer);
qrRefreshTimer = window.setTimeout(async () => {
if (qrStatus) qrStatus.textContent = "正在读取登录二维码...";
- const url = `/login-desktop/qr?t=${Date.now()}`;
try {
- const response = await fetch(url, { credentials: "same-origin", cache: "no-store" });
+ const response = await fetch(`/login-desktop/qr?t=${Date.now()}`, { credentials: "same-origin", cache: "no-store" });
+ if (response.status === 202) {
+ if (retries > 1 && workspace.state === "active") {
+ if (qrStatus) qrStatus.textContent = "浏览器正在生成二维码，继续等待...";
+ refreshLoginQr(1400, retries - 1);
+ }
+ return;
+ }
if (!response.ok) throw new Error(String(response.status));
const blob = await response.blob();
const previous = qrImage.dataset.objectUrl || "";
@@ -401,62 +425,73 @@
if (previous) URL.revokeObjectURL(previous);
if (qrStatus) qrStatus.textContent = "二维码已加载。如果过期，点击刷新。";
} catch {
- if (retries > 1) {
+ if (retries > 1 && workspace.state === "active") {
if (qrStatus) qrStatus.textContent = "登录页正在加载，继续等待二维码...";
refreshLoginQr(1400, retries - 1);
} else if (qrStatus) {
- qrStatus.textContent = "二维码还未准备好，请点击刷新重试。";
+ qrStatus.textContent = "二维码还未准备好，请确认自己已经获得登录工作区。";
}
}
}, delay);
};
const pollStatus = async () => {
- if (document.visibilityState !== "visible" || (section && !section.open)) {
- return;
- }
+ if (document.visibilityState !== "visible") return;
try {
- const response = await fetch("/login-desktop/status", {
- credentials: "same-origin",
- cache: "no-store",
- });
+ const statusUrl = workspace.state === "active" ? "/login-desktop/status" : "/login-desktop/workspace-status";
+ const response = await fetch(statusUrl, { credentials: "same-origin", cache: "no-store" });
const data = await response.json();
if (!response.ok || data.ok === false) {
- setStatus(
- data.error || "登录工作区不可用，请检查 login-desktop 服务。",
- "danger",
- );
+ setStatus(data.error || "登录工作区不可用，请检查 login-desktop 服务。", "danger");
return;
}
- if (data.logged_in) {
- setStatus(`当前浏览器已登录：${data.username}`, "success");
+ renderWorkspace(data.workspace);
+ if (workspace.state === "active" && workspace.active) {
+ loadFrame();
+ if (data.logged_in) setStatus(`当前浏览器已登录：${data.username}，请保存登录态。`, "success");
} else {
- setStatus("当前浏览器尚未登录，可打开工作区开始人工登录。");
+ closeFrame();
}
} catch (error) {
setStatus(`状态检查失败：${error.message}`, "danger");
}
};
+ const heartbeat = async () => {
+ if (workspace.state !== "active" || !workspace.active || !workspace.ticket) return;
+ try {
+ const data = await postForm("/login-desktop/heartbeat", { ticket: workspace.ticket });
+ renderWorkspace(data.workspace);
+ } catch (error) {
+ workspace = { state: "closed", active: false, position: 0, ticket: "" };
+ closeFrame();
+ setStatus(`登录工作区已释放：${error.message}`, "danger");
+ }
+ };
+
document.querySelectorAll(".login-desktop-open").forEach((button) => {
button.addEventListener("click", async () => {
- // Mobile browsers block window.open after an awaited request. Open the
- // authenticated same-origin workspace while the click gesture is active.
- const popup = publicUrl
- ? window.open(publicUrl, "_blank", "noopener")
- : null;
+ if (section) section.open = true;
+ const popup = publicUrl ? window.open("about:blank", "_blank", "noopener") : null;
try {
- await postForm("/login-desktop/open");
- loadFrame();
- refreshLoginQr(1800);
- if (!popup && frame) {
- frame.scrollIntoView({ behavior: "smooth", block: "start" });
- setStatus("弹窗被浏览器拦截，已在当前页面加载登录工作区。");
- } else {
- setStatus("请在登录工作区完成登录，然后返回此页保存登录态。");
+ const reloginUniqueId = button.dataset.reloginUniqueId || "";
+ const mode = button.dataset.loginMode || (reloginUniqueId ? "relogin" : "add");
+ const data = await postForm("/login-desktop/open", {
+ mode,
+ ...(reloginUniqueId ? { relogin_unique_id: reloginUniqueId } : {}),
+ });
+ renderWorkspace(data.workspace);
+ if (data.state === "queued") {
+ if (popup && !popup.closed) popup.close();
+ return;
}
+ if (popup && !popup.closed) popup.location.href = publicUrl;
+ loadFrame(true);
+ refreshLoginQr(500);
+ if (!popup && frame) frame.scrollIntoView({ behavior: "smooth", block: "start" });
} catch (error) {
- setStatus(`打开登录工作区失败：${error.message}`, "danger");
+ if (popup && !popup.closed) popup.close();
+ setStatus(`申请登录工作区失败：${error.message}`, "danger");
}
});
});
@@ -464,10 +499,9 @@
document.querySelectorAll("[data-refresh-login-qr]").forEach((button) => {
button.addEventListener("click", async () => {
button.disabled = true;
- if (qrStatus) qrStatus.textContent = "正在让抖音重新生成二维码...";
try {
- await postForm("/login-desktop/qr/refresh");
- refreshLoginQr(900);
+ await postForm("/login-desktop/qr/refresh", { ticket: workspace.ticket });
+ refreshLoginQr(500);
} catch (error) {
if (qrStatus) qrStatus.textContent = `刷新二维码失败：${error.message}`;
} finally {
@@ -479,10 +513,10 @@
document.querySelectorAll(".login-desktop-save").forEach((button) => {
button.addEventListener("click", async () => {
try {
- const data = await postForm("/login-desktop/save", {
- relogin_unique_id: button.dataset.reloginUniqueId || "",
- });
+ const data = await postForm("/login-desktop/save", { relogin_unique_id: button.dataset.reloginUniqueId || "" });
+ renderWorkspace(data.workspace);
setStatus(`已保存登录账号：${data.account?.username || ""}`, "success");
+ closeFrame();
window.setTimeout(() => window.location.reload(), 800);
} catch (error) {
setStatus(`保存登录账号失败：${error.message}`, "danger");
@@ -490,14 +524,28 @@
});
});
+ document.querySelectorAll(".login-desktop-close").forEach((button) => {
+ button.addEventListener("click", async () => {
+ try {
+ const data = await postForm("/login-desktop/close");
+ renderWorkspace(data.workspace);
+ closeFrame();
+ if (qrImage) qrImage.hidden = true;
+ } catch (error) {
+ setStatus(`关闭登录界面失败：${error.message}`, "danger");
+ }
+ });
+ });
+
document.querySelectorAll(".login-desktop-reset").forEach((button) => {
button.addEventListener("click", async () => {
try {
- await postForm("/login-desktop/reset");
- setStatus("登录工作区已重置，正在重新初始化。");
- await pollStatus();
+ const data = await postForm("/login-desktop/reset");
+ renderWorkspace(data.workspace);
+ closeFrame();
+ if (qrImage) qrImage.hidden = true;
} catch (error) {
- setStatus(`重置登录工作区失败：${error.message}`, "danger");
+ setStatus(`结束登录流程失败：${error.message}`, "danger");
}
});
});
@@ -513,16 +561,21 @@
});
});
- if (section) {
- section.addEventListener("toggle", () => {
- if (section.open) {
- loadFrame();
- pollStatus();
- }
- });
- }
+ if (section) section.addEventListener("toggle", () => { if (section.open) pollStatus(); });
+ pollStatus();
timer = window.setInterval(pollStatus, 5000);
- window.addEventListener("pagehide", () => window.clearInterval(timer));
+ heartbeatTimer = window.setInterval(heartbeat, 5000);
+ countdownTimer = window.setInterval(() => {
+ if (workspace.state !== "active" || !workspace.active) return;
+ workspace.remaining_seconds = Math.max(0, Number(workspace.remaining_seconds || 0) - 1);
+ const remaining = workspace.remaining_seconds;
+ setStatus(`登录工作区已分配给当前会话，剩余 ${remaining} 秒。完成扫码后请保存登录态。`, remaining <= 60 ? "warning" : "success");
+ }, 1000);
+ window.addEventListener("pagehide", () => {
+ window.clearInterval(timer);
+ window.clearInterval(heartbeatTimer);
+ window.clearInterval(countdownTimer);
+ });
})();
(() => {
diff --git a/DouYinSparkFlow/webui/templates/base.html b/DouYinSparkFlow/webui/templates/base.html
index d1998b5..3683737 100644
--- a/DouYinSparkFlow/webui/templates/base.html
+++ b/DouYinSparkFlow/webui/templates/base.html
@@ -6,7 +6,7 @@
{% block title %}续火花{% endblock %}
-
+
@@ -49,9 +49,11 @@
发送控制台
+ {% if is_admin %}
发送记录
+ {% endif %}
@@ -64,15 +66,21 @@
登录抖音账号
+{% if is_admin %}
运行配置
+ {% endif %}
+{% if is_admin %}
运维操作
+ {% endif %}
+{% if is_admin %}
系统设置
+ {% endif %}
@@ -82,7 +90,7 @@
{{ (current_user or "A")[:1] }}
{{ current_user or "未登录" }}
- 管理员
+ {% if is_admin %}管理员{% else %}普通用户{% endif %}