feat: 完成账号目录slug工具

- 新增 app/config.py,生成稳定账号 slug 并创建 user-data-dir

- db 账号创建复用统一 slug 规则

- 新增 config 单元测试覆盖 slug 和目录创建

- 更新任务看板、API 合约、当前状态和进度记录
This commit is contained in:
chengma
2026-06-27 09:21:33 +08:00
parent aba272dd21
commit d3afe69c10
7 changed files with 123 additions and 18 deletions
+48
View File
@@ -0,0 +1,48 @@
"""Account path helpers for Chrome user-data directories."""
from __future__ import annotations
import hashlib
import os
import re
from . import appconfig
SLUG_PATTERN = re.compile(r"^[a-z0-9_]+$")
class ConfigPathError(RuntimeError):
"""Raised when account path configuration is invalid."""
def make_slug(alias) -> str:
"""Return a stable unique slug for an account alias."""
text = str(alias or "").strip()
if not text:
raise ConfigPathError("别名不能为空")
base = re.sub(r"[^a-z0-9_]+", "_", text.lower()).strip("_")
suffix = hashlib.sha1(text.encode("utf-8")).hexdigest()[:8]
return f"{base}_{suffix}" if base else f"account_{suffix}"
def _validate_slug(slug) -> str:
text = str(slug or "").strip()
if not text:
raise ConfigPathError("slug 不能为空")
if not SLUG_PATTERN.fullmatch(text):
raise ConfigPathError("slug 只能包含小写字母、数字和下划线")
return text
def ensure_user_data_dir(slug, root=None, config=None) -> str:
"""Create and return the absolute user-data-dir path for a slug."""
safe_slug = _validate_slug(slug)
user_data_root = root if root is not None else appconfig.user_data_root(config)
if not str(user_data_root).strip():
raise ConfigPathError("user_data_root 不能为空")
path = os.path.abspath(os.path.join(str(user_data_root), safe_slug))
os.makedirs(path, exist_ok=True)
return path
+2 -9
View File
@@ -2,10 +2,8 @@
from __future__ import annotations
import hashlib
import json
import os
import re
import sqlite3
import uuid
from contextlib import contextmanager
@@ -14,6 +12,7 @@ from datetime import datetime
from typing import Iterable, Optional
from . import appconfig
from .config import make_slug
DEFAULT_BUSY_TIMEOUT_MS = 5000
@@ -220,12 +219,6 @@ def _validate_fields(fields, allowed):
raise DbError(f"不支持更新字段: {', '.join(unknown)}")
def _slug_from_alias(alias: str) -> str:
base = re.sub(r"[^a-z0-9_]+", "_", alias.lower()).strip("_")
suffix = hashlib.sha1(alias.encode("utf-8")).hexdigest()[:8]
return f"{base}_{suffix}" if base else f"account_{suffix}"
def _row_key(batch_id, source_file_abs, source_sheet, source_row) -> str:
return f"{batch_id}:{source_file_abs}:{source_sheet}:{source_row}"
@@ -330,7 +323,7 @@ def add_account(
path=None,
conn=None,
):
slug = slug or _slug_from_alias(alias)
slug = slug or make_slug(alias)
user_data_dir = user_data_dir or os.path.join("chrome_user_data_dir", slug)
now = _now()
with _connection(conn, path) as database: