From 0d92a613f5ca432be6533ecc4cebd286f152c156 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=83=A1=E7=BB=8D=E5=BA=B7?= <3052147989@qq.com> Date: Mon, 16 Mar 2026 19:20:02 +0800 Subject: [PATCH] feat: implement PRD-00009 MT-1 internal email pool with full P0 contract (claim/release/complete/stats, ownership checks, lease expiry, cooldown recovery, 38 pool tests) --- outlook_web/app.py | 20 +- outlook_web/controllers/pool.py | 133 +++++ outlook_web/db.py | 97 +++- outlook_web/repositories/pool.py | 282 +++++++++ outlook_web/routes/pool.py | 30 + outlook_web/services/pool.py | 225 +++++++ outlook_web/services/scheduler.py | 159 ++++- tests/test_pool.py | 933 ++++++++++++++++++++++++++++++ 8 files changed, 1841 insertions(+), 38 deletions(-) create mode 100644 outlook_web/controllers/pool.py create mode 100644 outlook_web/repositories/pool.py create mode 100644 outlook_web/routes/pool.py create mode 100644 outlook_web/services/pool.py create mode 100644 tests/test_pool.py diff --git a/outlook_web/app.py b/outlook_web/app.py index 1739212..9c491ae 100644 --- a/outlook_web/app.py +++ b/outlook_web/app.py @@ -36,6 +36,7 @@ def create_app(*, autostart_scheduler: Optional[bool] = None): groups, oauth, pages, + pool, scheduler, settings, system, @@ -88,7 +89,9 @@ def create_app(*, autostart_scheduler: Optional[bool] = None): # ProxyFix 中间件（仅在配置启用时应用） # 注意：启用前必须配置 TRUSTED_PROXIES 环境变量 if config.get_proxy_fix_enabled(): - app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1) + app.wsgi_app = ProxyFix( + app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1 + ) # DB teardown（请求结束释放连接） register_db(app) @@ -98,7 +101,11 @@ def create_app(*, autostart_scheduler: Optional[bool] = None): import sys _log_handler = logging.StreamHandler(sys.stderr) - _log_handler.setFormatter(logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s", datefmt="%H:%M:%S")) + _log_handler.setFormatter( + logging.Formatter( + "%(asctime)s %(name)s %(levelname)s %(message)s", datefmt="%H:%M:%S" + ) + ) _ow_logger = logging.getLogger("outlook_web") if not _ow_logger.handlers: _ow_logger.addHandler(_log_handler) @@ -125,6 +132,7 @@ def create_app(*, autostart_scheduler: Optional[bool] = None): app.register_blueprint(scheduler.create_blueprint()) app.register_blueprint(system.create_blueprint()) app.register_blueprint(audit.create_blueprint()) + app.register_blueprint(pool.create_blueprint()) # 打印初始化信息 print("=" * 60) @@ -141,11 +149,15 @@ def create_app(*, autostart_scheduler: Optional[bool] = None): from outlook_web.services import scheduler as scheduler_service if scheduler_service.should_autostart_scheduler(): - scheduler_service.init_scheduler(_APP_INSTANCE, graph_service.test_refresh_token_with_rotation) + scheduler_service.init_scheduler( + _APP_INSTANCE, graph_service.test_refresh_token_with_rotation + ) elif autostart_scheduler: from outlook_web.services import graph as graph_service from outlook_web.services import scheduler as scheduler_service - scheduler_service.init_scheduler(_APP_INSTANCE, graph_service.test_refresh_token_with_rotation) + scheduler_service.init_scheduler( + _APP_INSTANCE, graph_service.test_refresh_token_with_rotation + ) return _APP_INSTANCE diff --git a/outlook_web/controllers/pool.py b/outlook_web/controllers/pool.py new file mode 100644 index 0000000..5b022fa --- /dev/null +++ b/outlook_web/controllers/pool.py @@ -0,0 +1,133 @@ +""" +邮箱池控制器（PRD-00009 MT-1） + +职责： +- 解析 HTTP 请求参数 +- 调用 services/pool.py 的业务方法 +- 将 service 层业务结果映射为 FD 统一响应格式 +- 将 PoolServiceError 映射为统一错误响应 + +响应格式（FD-00009 §6）： +- 成功：{"success": true, "data": {...}} +- 失败：{"success": false, "error": "error_code", "message": "错误说明"} + +HTTP 状态码映射（TD-00009 §6.3）： +- 参数错误 → 400 +- 归属/凭据错误 → 403 +- 状态冲突 → 409 +- 业务失败（如 no_available_account） → 200 + success=false +""" + +from __future__ import annotations + +from flask import jsonify, request + +from outlook_web.services.pool import ( + PoolServiceError, + claim_random, + complete_claim, + get_pool_stats, + release_claim, +) + + +def _success_response(data: dict, http_status: int = 200): + """统一成功响应：{"success": true, "data": {...}}""" + return jsonify({"success": True, "data": data}), http_status + + +def _error_response(error_code: str, message: str, http_status: int = 400): + """统一失败响应：{"success": false, "error": "...", "message": "..."}""" + return jsonify( + {"success": False, "error": error_code, "message": message} + ), http_status + + +def api_claim_random(): + body = request.get_json(silent=True) or {} + caller_id = body.get("caller_id", "") + task_id = body.get("task_id", "") + provider = body.get("provider") + + try: + account = claim_random( + caller_id=caller_id, + task_id=task_id, + provider=provider, + ) + return _success_response( + { + "account_id": account["id"], + "email": account["email"], + "claim_token": account["claim_token"], + "lease_expires_at": account["lease_expires_at"], + } + ) + except PoolServiceError as exc: + return _error_response(exc.error_code, str(exc), exc.http_status) + + +def api_claim_release(): + body = request.get_json(silent=True) or {} + account_id = body.get("account_id") + claim_token = body.get("claim_token", "") + caller_id = body.get("caller_id", "") + task_id = body.get("task_id", "") + reason = body.get("reason") + + if account_id is None: + return _error_response("account_id_missing", "account_id 不能为空", 400) + try: + account_id = int(account_id) + except (TypeError, ValueError): + return _error_response("account_id_invalid", "account_id 必须为整数", 400) + + try: + release_claim( + account_id=account_id, + claim_token=claim_token, + caller_id=caller_id, + task_id=task_id, + reason=reason, + ) + return _success_response({"account_id": account_id, "pool_status": "available"}) + except PoolServiceError as exc: + return _error_response(exc.error_code, str(exc), exc.http_status) + + +def api_claim_complete(): + body = request.get_json(silent=True) or {} + account_id = body.get("account_id") + claim_token = body.get("claim_token", "") + caller_id = body.get("caller_id", "") + task_id = body.get("task_id", "") + result = body.get("result", "") + detail = body.get("detail") + + if account_id is None: + return _error_response("account_id_missing", "account_id 不能为空", 400) + try: + account_id = int(account_id) + except (TypeError, ValueError): + return _error_response("account_id_invalid", "account_id 必须为整数", 400) + + try: + new_status = complete_claim( + account_id=account_id, + claim_token=claim_token, + caller_id=caller_id, + task_id=task_id, + result=result, + detail=detail, + ) + return _success_response({"account_id": account_id, "pool_status": new_status}) + except PoolServiceError as exc: + return _error_response(exc.error_code, str(exc), exc.http_status) + + +def api_pool_stats(): + try: + stats = get_pool_stats() + return _success_response(stats) + except Exception as exc: + return _error_response("internal_error", str(exc), 500) diff --git a/outlook_web/db.py b/outlook_web/db.py index 6392bb8..d9e6aa7 100644 --- a/outlook_web/db.py +++ b/outlook_web/db.py @@ -24,7 +24,8 @@ from outlook_web.security.crypto import ( # v8：PRD-00008 P1 — 上游真实探测结果缓存表 # v9：PRD-00008 P2 — 多 API Key 表 # v10：PRD-00008 P2 — 调用方日级使用统计表 -DB_SCHEMA_VERSION = 10 +# v11：PRD-00009 MT-1 — 邮箱池字段（pool_status/claimed_by/...）+ account_claim_logs 表 + pool settings +DB_SCHEMA_VERSION = 11 DB_SCHEMA_VERSION_KEY = "db_schema_version" DB_SCHEMA_LAST_UPGRADE_TRACE_ID_KEY = "db_schema_last_upgrade_trace_id" DB_SCHEMA_LAST_UPGRADE_ERROR_KEY = "db_schema_last_upgrade_error" @@ -124,7 +125,9 @@ def init_db(database_path: Optional[str] = None): """) # 在锁内读取当前 schema 版本（保证一致性） - row = cursor.execute("SELECT value FROM settings WHERE key = ?", (DB_SCHEMA_VERSION_KEY,)).fetchone() + row = cursor.execute( + "SELECT value FROM settings WHERE key = ?", (DB_SCHEMA_VERSION_KEY,) + ).fetchone() current_version = int(row["value"]) if row and row["value"] is not None else 0 upgrading = current_version < DB_SCHEMA_VERSION @@ -133,7 +136,9 @@ def init_db(database_path: Optional[str] = None): if db_existed: try: print("=" * 60) - print(f"[升级提示] 检测到数据库需要升级：v{current_version} -> v{DB_SCHEMA_VERSION}") + print( + f"[升级提示] 检测到数据库需要升级：v{current_version} -> v{DB_SCHEMA_VERSION}" + ) print(f"[升级提示] 强烈建议先备份数据库文件：{path}") print(f'[升级提示] 示例：cp "{path}" "{path}.backup"') print(f"[升级提示] trace_id={migration_trace_id}") @@ -347,25 +352,39 @@ def init_db(database_path: Optional[str] = None): if "remark" not in columns: cursor.execute("ALTER TABLE accounts ADD COLUMN remark TEXT") if "status" not in columns: - cursor.execute("ALTER TABLE accounts ADD COLUMN status TEXT DEFAULT 'active'") + cursor.execute( + "ALTER TABLE accounts ADD COLUMN status TEXT DEFAULT 'active'" + ) if "updated_at" not in columns: - cursor.execute("ALTER TABLE accounts ADD COLUMN updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP") + cursor.execute( + "ALTER TABLE accounts ADD COLUMN updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP" + ) if "last_refresh_at" not in columns: cursor.execute("ALTER TABLE accounts ADD COLUMN last_refresh_at TIMESTAMP") if "account_type" not in columns: - cursor.execute("ALTER TABLE accounts ADD COLUMN account_type TEXT DEFAULT 'outlook'") + cursor.execute( + "ALTER TABLE accounts ADD COLUMN account_type TEXT DEFAULT 'outlook'" + ) if "provider" not in columns: - cursor.execute("ALTER TABLE accounts ADD COLUMN provider TEXT DEFAULT 'outlook'") + cursor.execute( + "ALTER TABLE accounts ADD COLUMN provider TEXT DEFAULT 'outlook'" + ) if "imap_host" not in columns: cursor.execute("ALTER TABLE accounts ADD COLUMN imap_host TEXT") if "imap_port" not in columns: - cursor.execute("ALTER TABLE accounts ADD COLUMN imap_port INTEGER DEFAULT 993") + cursor.execute( + "ALTER TABLE accounts ADD COLUMN imap_port INTEGER DEFAULT 993" + ) if "imap_password" not in columns: cursor.execute("ALTER TABLE accounts ADD COLUMN imap_password TEXT") if "telegram_push_enabled" not in columns: - cursor.execute("ALTER TABLE accounts ADD COLUMN telegram_push_enabled INTEGER NOT NULL DEFAULT 0") + cursor.execute( + "ALTER TABLE accounts ADD COLUMN telegram_push_enabled INTEGER NOT NULL DEFAULT 0" + ) if "telegram_last_checked_at" not in columns: - cursor.execute("ALTER TABLE accounts ADD COLUMN telegram_last_checked_at TEXT DEFAULT NULL") + cursor.execute( + "ALTER TABLE accounts ADD COLUMN telegram_last_checked_at TEXT DEFAULT NULL" + ) cursor.execute("PRAGMA table_info(groups)") group_columns = [col[1] for col in cursor.fetchall()] @@ -640,6 +659,64 @@ def init_db(database_path: Optional[str] = None): ON external_api_consumer_usage_daily(consumer_key, usage_date) """) + # v11: 邮箱池字段 + account_claim_logs 表（PRD-00009 MT-1） + cursor.execute("PRAGMA table_info(accounts)") + accounts_columns_v11 = [col[1] for col in cursor.fetchall()] + for col_def in [ + ("pool_status", "TEXT DEFAULT NULL"), + ("claimed_by", "TEXT DEFAULT NULL"), + ("claimed_at", "TEXT DEFAULT NULL"), + ("lease_expires_at", "TEXT DEFAULT NULL"), + ("claim_token", "TEXT DEFAULT NULL"), + ("last_claimed_at", "TEXT DEFAULT NULL"), + ("last_result", "TEXT DEFAULT NULL"), + ("last_result_detail", "TEXT DEFAULT NULL"), + ("success_count", "INTEGER DEFAULT 0"), + ("fail_count", "INTEGER DEFAULT 0"), + ]: + if col_def[0] not in accounts_columns_v11: + cursor.execute( + f"ALTER TABLE accounts ADD COLUMN {col_def[0]} {col_def[1]}" + ) + + cursor.execute(""" + CREATE TABLE IF NOT EXISTS account_claim_logs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + account_id INTEGER NOT NULL, + claim_token TEXT NOT NULL, + caller_id TEXT NOT NULL, + task_id TEXT NOT NULL, + action TEXT NOT NULL, + result TEXT DEFAULT NULL, + detail TEXT DEFAULT NULL, + created_at TEXT NOT NULL, + FOREIGN KEY (account_id) REFERENCES accounts(id) + ) + """) + cursor.execute(""" + CREATE INDEX IF NOT EXISTS idx_claim_logs_account_id + ON account_claim_logs(account_id) + """) + cursor.execute(""" + CREATE INDEX IF NOT EXISTS idx_claim_logs_task_id + ON account_claim_logs(task_id) + """) + cursor.execute(""" + CREATE INDEX IF NOT EXISTS idx_claim_logs_claim_token + ON account_claim_logs(claim_token) + """) + cursor.execute(""" + CREATE INDEX IF NOT EXISTS idx_accounts_pool_status + ON accounts(pool_status) + """) + + cursor.execute( + "INSERT OR IGNORE INTO settings (key, value) VALUES ('pool_cooldown_seconds', '86400')" + ) + cursor.execute( + "INSERT OR IGNORE INTO settings (key, value) VALUES ('pool_default_lease_seconds', '600')" + ) + # 迁移现有明文数据为加密数据 migrate_sensitive_data(conn) diff --git a/outlook_web/repositories/pool.py b/outlook_web/repositories/pool.py new file mode 100644 index 0000000..1325c0d --- /dev/null +++ b/outlook_web/repositories/pool.py @@ -0,0 +1,282 @@ +from __future__ import annotations + +import secrets +from datetime import datetime, timedelta, timezone +from typing import Dict, List, Optional + +import sqlite3 + + +RESULT_TO_POOL_STATUS: Dict[str, str] = { + "success": "used", + "verification_timeout": "cooldown", + "provider_blocked": "frozen", + "credential_invalid": "retired", + "network_error": "available", +} + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc).replace(tzinfo=None) + + +def claim_atomic( + conn: sqlite3.Connection, + caller_id: str, + task_id: str, + lease_seconds: int, + provider: Optional[str] = None, + group_id: Optional[int] = None, + tags: Optional[List[str]] = None, + exclude_recent_minutes: Optional[int] = None, +) -> Optional[dict]: + sql = """ + SELECT a.* FROM accounts a + WHERE a.pool_status = 'available' + AND a.status = 'active' + """ + params: list = [] + + if provider: + sql += " AND a.provider = ?" + params.append(provider) + + if group_id is not None: + sql += " AND a.group_id = ?" + params.append(group_id) + + if tags: + for tag_name in tags: + sql += """ + AND EXISTS ( + SELECT 1 FROM account_tags at2 + JOIN tags t2 ON at2.tag_id = t2.id + WHERE at2.account_id = a.id AND t2.name = ? + ) + """ + params.append(tag_name) + + if exclude_recent_minutes and exclude_recent_minutes > 0: + cutoff = ( + _utcnow() - timedelta(minutes=exclude_recent_minutes) + ).isoformat() + "Z" + sql += " AND (a.last_claimed_at IS NULL OR a.last_claimed_at < ?)" + params.append(cutoff) + + sql += " ORDER BY RANDOM() LIMIT 1" + + conn.execute("BEGIN IMMEDIATE") + account = conn.execute(sql, params).fetchone() + + if account is None: + conn.execute("ROLLBACK") + return None + + now_str = _utcnow().isoformat() + "Z" + lease_expires_at_str = ( + _utcnow() + timedelta(seconds=lease_seconds) + ).isoformat() + "Z" + token = "clm_" + secrets.token_urlsafe(9) + + conn.execute( + """ + UPDATE accounts SET + pool_status = 'claimed', + claimed_by = ?, + claimed_at = ?, + lease_expires_at = ?, + claim_token = ?, + last_claimed_at = ?, + updated_at = ? + WHERE id = ? + """, + ( + f"{caller_id}:{task_id}", + now_str, + lease_expires_at_str, + token, + now_str, + now_str, + account["id"], + ), + ) + conn.execute( + """ + INSERT INTO account_claim_logs + (account_id, claim_token, caller_id, task_id, action, result, detail, created_at) + VALUES (?, ?, ?, ?, 'claim', NULL, NULL, ?) + """, + (account["id"], token, caller_id, task_id, now_str), + ) + conn.execute("COMMIT") + return dict(account) | { + "claim_token": token, + "lease_expires_at": lease_expires_at_str, + } + + +def release( + conn: sqlite3.Connection, + account_id: int, + claim_token: str, + caller_id: str, + task_id: str, + reason: Optional[str], +) -> None: + now_str = _utcnow().isoformat() + "Z" + conn.execute("BEGIN IMMEDIATE") + conn.execute( + """ + UPDATE accounts SET + pool_status = 'available', + claimed_by = NULL, + claimed_at = NULL, + lease_expires_at = NULL, + claim_token = NULL, + updated_at = ? + WHERE id = ? + """, + (now_str, account_id), + ) + conn.execute( + """ + INSERT INTO account_claim_logs + (account_id, claim_token, caller_id, task_id, action, result, detail, created_at) + VALUES (?, ?, ?, ?, 'release', 'manual_release', ?, ?) + """, + (account_id, claim_token, caller_id, task_id, reason, now_str), + ) + conn.execute("COMMIT") + + +def complete( + conn: sqlite3.Connection, + account_id: int, + claim_token: str, + caller_id: str, + task_id: str, + result: str, + detail: Optional[str], +) -> str: + new_pool_status = RESULT_TO_POOL_STATUS[result] + is_success = result == "success" + now_str = _utcnow().isoformat() + "Z" + + conn.execute("BEGIN IMMEDIATE") + conn.execute( + """ + UPDATE accounts SET + pool_status = ?, + claimed_by = NULL, + claimed_at = NULL, + lease_expires_at = NULL, + claim_token = NULL, + last_result = ?, + last_result_detail = ?, + success_count = success_count + ?, + fail_count = fail_count + ?, + updated_at = ? + WHERE id = ? + """, + ( + new_pool_status, + result, + detail, + 1 if is_success else 0, + 0 if is_success else 1, + now_str, + account_id, + ), + ) + conn.execute( + """ + INSERT INTO account_claim_logs + (account_id, claim_token, caller_id, task_id, action, result, detail, created_at) + VALUES (?, ?, ?, ?, 'complete', ?, ?, ?) + """, + (account_id, claim_token, caller_id, task_id, result, detail, now_str), + ) + conn.execute("COMMIT") + return new_pool_status + + +def expire_stale_claims(conn: sqlite3.Connection) -> int: + now_str = _utcnow().isoformat() + "Z" + expired = conn.execute( + """ + SELECT id, claim_token, claimed_by FROM accounts + WHERE pool_status = 'claimed' AND lease_expires_at < ? + """, + (now_str,), + ).fetchall() + + for account in expired: + parts = (account["claimed_by"] or ":").split(":", 1) + caller_id = parts[0] + task_id = parts[1] if len(parts) > 1 else "" + + conn.execute("BEGIN IMMEDIATE") + conn.execute( + """ + UPDATE accounts SET + pool_status = 'cooldown', + claimed_by = NULL, + claimed_at = NULL, + lease_expires_at = NULL, + claim_token = NULL, + fail_count = fail_count + 1, + last_result = 'lease_expired', + updated_at = ? + WHERE id = ? + """, + (now_str, account["id"]), + ) + conn.execute( + """ + INSERT INTO account_claim_logs + (account_id, claim_token, caller_id, task_id, action, result, detail, created_at) + VALUES (?, ?, ?, ?, 'expire', 'lease_expired', 'lease timeout, auto moved to cooldown', ?) + """, + (account["id"], account["claim_token"], caller_id, task_id, now_str), + ) + conn.execute("COMMIT") + + return len(expired) + + +def recover_cooldown(conn: sqlite3.Connection, cooldown_seconds: int) -> int: + cutoff_str = (_utcnow() - timedelta(seconds=cooldown_seconds)).isoformat() + "Z" + now_str = _utcnow().isoformat() + "Z" + cursor = conn.execute( + """ + UPDATE accounts SET pool_status = 'available', updated_at = ? + WHERE pool_status = 'cooldown' AND updated_at < ? + """, + (now_str, cutoff_str), + ) + conn.commit() + return cursor.rowcount + + +def get_stats(conn: sqlite3.Connection) -> dict: + rows = conn.execute( + """ + SELECT pool_status, COUNT(*) as cnt FROM accounts + GROUP BY pool_status + """ + ).fetchall() + pool_counts: dict = { + "available": 0, + "claimed": 0, + "used": 0, + "cooldown": 0, + "frozen": 0, + "retired": 0, + "not_in_pool": 0, + } + for row in rows: + key = row["pool_status"] if row["pool_status"] is not None else "not_in_pool" + if key in pool_counts: + pool_counts[key] = row["cnt"] + + return {"pool_counts": pool_counts} diff --git a/outlook_web/routes/pool.py b/outlook_web/routes/pool.py new file mode 100644 index 0000000..61151fc --- /dev/null +++ b/outlook_web/routes/pool.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +from flask import Blueprint + +from outlook_web.controllers import pool as pool_controller + + +def create_blueprint() -> Blueprint: + bp = Blueprint("pool", __name__) + bp.add_url_rule( + "/api/pool/claim-random", + view_func=pool_controller.api_claim_random, + methods=["POST"], + ) + bp.add_url_rule( + "/api/pool/claim-release", + view_func=pool_controller.api_claim_release, + methods=["POST"], + ) + bp.add_url_rule( + "/api/pool/claim-complete", + view_func=pool_controller.api_claim_complete, + methods=["POST"], + ) + bp.add_url_rule( + "/api/pool/stats", + view_func=pool_controller.api_pool_stats, + methods=["GET"], + ) + return bp diff --git a/outlook_web/services/pool.py b/outlook_web/services/pool.py new file mode 100644 index 0000000..c320508 --- /dev/null +++ b/outlook_web/services/pool.py @@ -0,0 +1,225 @@ +""" +邮箱池服务层（PRD-00009 MT-1） + +职责： +- 输入校验（caller_id / task_id / lease_seconds / result / detail 长度） +- 读取 settings（在 Flask app_context 下用 get_db，或直接接受 conn） +- 调用 repositories/pool.py 的原子操作 +- 将 repository 层的异常转换为业务错误码 +""" + +from __future__ import annotations + +from typing import Optional + +from outlook_web.db import create_sqlite_connection +from outlook_web.repositories import pool as pool_repo + +CALLER_ID_MAX_LEN = 64 +TASK_ID_MAX_LEN = 128 +REASON_MAX_LEN = 256 +DETAIL_MAX_LEN = 512 + +VALID_RESULTS = set(pool_repo.RESULT_TO_POOL_STATUS.keys()) + + +class PoolServiceError(Exception): + """业务错误，包含 HTTP 状态码和错误码。""" + + def __init__(self, message: str, error_code: str, http_status: int = 400): + super().__init__(message) + self.error_code = error_code + self.http_status = http_status + + +def _validate_caller_id(caller_id: str) -> None: + if not caller_id or not caller_id.strip(): + raise PoolServiceError("caller_id 不能为空", "caller_id_empty") + if len(caller_id) > CALLER_ID_MAX_LEN: + raise PoolServiceError( + f"caller_id 超过最大长度 {CALLER_ID_MAX_LEN}", "caller_id_too_long" + ) + + +def _validate_task_id(task_id: str) -> None: + if not task_id or not task_id.strip(): + raise PoolServiceError("task_id 不能为空", "task_id_empty") + if len(task_id) > TASK_ID_MAX_LEN: + raise PoolServiceError( + f"task_id 超过最大长度 {TASK_ID_MAX_LEN}", "task_id_too_long" + ) + + +def _validate_lease_seconds(lease_seconds: int, max_lease: int = 3600) -> None: + if lease_seconds <= 0: + raise PoolServiceError("lease_seconds 必须大于 0", "lease_seconds_invalid") + if lease_seconds > max_lease: + raise PoolServiceError( + f"lease_seconds 不能超过 {max_lease} 秒", "lease_seconds_too_large" + ) + + +def _read_settings_via_conn(conn) -> dict: + """在独立连接场景下直接从 settings 表读取池相关配置。""" + rows = conn.execute( + "SELECT key, value FROM settings WHERE key IN (?, ?)", + ("pool_cooldown_seconds", "pool_default_lease_seconds"), + ).fetchall() + result = {"pool_cooldown_seconds": 86400, "pool_default_lease_seconds": 600} + for row in rows: + try: + result[row["key"]] = int(row["value"]) + except (TypeError, ValueError): + pass + return result + + +def claim_random( + *, + caller_id: str, + task_id: str, + provider: Optional[str] = None, +) -> dict: + _validate_caller_id(caller_id) + _validate_task_id(task_id) + + conn = create_sqlite_connection() + try: + settings = _read_settings_via_conn(conn) + default_lease = settings["pool_default_lease_seconds"] + _validate_lease_seconds(default_lease) + + account = pool_repo.claim_atomic( + conn, + caller_id=caller_id, + task_id=task_id, + lease_seconds=default_lease, + provider=provider, + ) + if account is None: + raise PoolServiceError( + "池中没有符合条件的可用邮箱", "no_available_account", http_status=200 + ) + return account + finally: + conn.close() + + +def release_claim( + *, + account_id: int, + claim_token: str, + caller_id: str, + task_id: str, + reason: Optional[str] = None, +) -> None: + """释放已领取的邮箱账号（不计入成功/失败统计，直接回 available）。""" + _validate_caller_id(caller_id) + _validate_task_id(task_id) + if not claim_token or not claim_token.strip(): + raise PoolServiceError("claim_token 不能为空", "claim_token_empty") + if reason and len(reason) > REASON_MAX_LEN: + raise PoolServiceError( + f"reason 超过最大长度 {REASON_MAX_LEN}", "reason_too_long" + ) + + conn = create_sqlite_connection() + try: + row = conn.execute( + "SELECT id, claim_token, claimed_by, pool_status FROM accounts WHERE id = ?", + (account_id,), + ).fetchone() + if row is None: + raise PoolServiceError("账号不存在", "account_not_found", http_status=400) + if row["pool_status"] != "claimed": + raise PoolServiceError( + f"账号当前状态为 '{row['pool_status']}'，无法 release", + "not_claimed", + http_status=409, + ) + if row["claim_token"] != claim_token: + raise PoolServiceError( + "claim_token 不匹配", "token_mismatch", http_status=403 + ) + expected_claimed_by = f"{caller_id}:{task_id}" + if row["claimed_by"] != expected_claimed_by: + raise PoolServiceError( + "caller_id 或 task_id 与领取记录不一致", + "caller_mismatch", + http_status=403, + ) + + pool_repo.release(conn, account_id, claim_token, caller_id, task_id, reason) + finally: + conn.close() + + +def complete_claim( + *, + account_id: int, + claim_token: str, + caller_id: str, + task_id: str, + result: str, + detail: Optional[str] = None, +) -> str: + """ + 标记领取结果并驱动状态机流转。 + + 返回账号的新 pool_status。 + """ + _validate_caller_id(caller_id) + _validate_task_id(task_id) + if not claim_token or not claim_token.strip(): + raise PoolServiceError("claim_token 不能为空", "claim_token_empty") + if result not in VALID_RESULTS: + raise PoolServiceError( + f"result 必须是 {sorted(VALID_RESULTS)} 之一", + "invalid_result", + ) + if detail and len(detail) > DETAIL_MAX_LEN: + raise PoolServiceError( + f"detail 超过最大长度 {DETAIL_MAX_LEN}", "detail_too_long" + ) + + conn = create_sqlite_connection() + try: + row = conn.execute( + "SELECT id, claim_token, claimed_by, pool_status FROM accounts WHERE id = ?", + (account_id,), + ).fetchone() + if row is None: + raise PoolServiceError("账号不存在", "account_not_found", http_status=400) + if row["pool_status"] != "claimed": + raise PoolServiceError( + f"账号当前状态为 '{row['pool_status']}'，无法 complete", + "not_claimed", + http_status=409, + ) + if row["claim_token"] != claim_token: + raise PoolServiceError( + "claim_token 不匹配", "token_mismatch", http_status=403 + ) + expected_claimed_by = f"{caller_id}:{task_id}" + if row["claimed_by"] != expected_claimed_by: + raise PoolServiceError( + "caller_id 或 task_id 与领取记录不一致", + "caller_mismatch", + http_status=403, + ) + + new_status = pool_repo.complete( + conn, account_id, claim_token, caller_id, task_id, result, detail + ) + return new_status + finally: + conn.close() + + +def get_pool_stats() -> dict: + """返回池状态统计（不修改任何数据）。""" + conn = create_sqlite_connection() + try: + return pool_repo.get_stats(conn) + finally: + conn.close() diff --git a/outlook_web/services/scheduler.py b/outlook_web/services/scheduler.py index 6c22937..cfba41f 100644 --- a/outlook_web/services/scheduler.py +++ b/outlook_web/services/scheduler.py @@ -97,7 +97,10 @@ def _configure_telegram_push_job(scheduler, app) -> None: def _configure_probe_poll_job(scheduler, app) -> None: """P2: 注册异步探测轮询 Job。每 5 秒执行一次，处理 pending 探测并清理过期记录。""" - from outlook_web.services.external_api import cleanup_expired_probes, poll_pending_probes + from outlook_web.services.external_api import ( + cleanup_expired_probes, + poll_pending_probes, + ) try: scheduler.remove_job("external_probe_poll") @@ -125,20 +128,94 @@ def _configure_probe_poll_job(scheduler, app) -> None: print("✓ 对外 API 异步探测 Job 已配置（轮询间隔：5 秒）") +def _configure_pool_maintenance_jobs(scheduler) -> None: + try: + scheduler.remove_job("pool_expire_stale_claims") + except Exception: + pass + try: + scheduler.remove_job("pool_recover_cooldown") + except Exception: + pass + + def _expire_task(): + try: + from outlook_web.db import create_sqlite_connection + from outlook_web.repositories import pool as pool_repo + + conn = create_sqlite_connection() + try: + expired = pool_repo.expire_stale_claims(conn) + if expired: + print(f"[pool] 过期 claims 回收：{expired} 条") + finally: + conn.close() + except Exception: + pass + + def _recover_task(): + try: + from outlook_web.db import create_sqlite_connection + from outlook_web.repositories import pool as pool_repo + + conn = create_sqlite_connection() + try: + row = conn.execute( + "SELECT value FROM settings WHERE key = 'pool_cooldown_seconds'" + ).fetchone() + cooldown_seconds = int(row["value"]) if row and row["value"] else 86400 + recovered = pool_repo.recover_cooldown(conn, cooldown_seconds) + if recovered: + print(f"[pool] cooldown 恢复：{recovered} 条") + finally: + conn.close() + except Exception: + pass + + scheduler.add_job( + func=_expire_task, + trigger="interval", + seconds=60, + id="pool_expire_stale_claims", + name="邮箱池过期 Claim 回收", + replace_existing=True, + max_instances=1, + coalesce=True, + misfire_grace_time=60, + ) + scheduler.add_job( + func=_recover_task, + trigger="interval", + seconds=300, + id="pool_recover_cooldown", + name="邮箱池 Cooldown 恢复", + replace_existing=True, + max_instances=1, + coalesce=True, + misfire_grace_time=120, + ) + print("✓ 邮箱池维护 Job 已配置（expire: 60s，recover: 300s）") + + def configure_scheduler_jobs(scheduler, app, test_refresh_token) -> None: - """根据当前 settings 重新配置定时刷新 Job（配置变更即时生效）""" try: from apscheduler.triggers.cron import CronTrigger except Exception: return with app.app_context(): - enable_scheduled = settings_repo.get_setting("enable_scheduled_refresh", "true").lower() == "true" - use_cron = settings_repo.get_setting("use_cron_schedule", "false").lower() == "true" - refresh_interval_days = int(settings_repo.get_setting("refresh_interval_days", "30")) + enable_scheduled = ( + settings_repo.get_setting("enable_scheduled_refresh", "true").lower() + == "true" + ) + use_cron = ( + settings_repo.get_setting("use_cron_schedule", "false").lower() == "true" + ) + refresh_interval_days = int( + settings_repo.get_setting("refresh_interval_days", "30") + ) cron_expr = settings_repo.get_setting("refresh_cron", "0 2 * * *") - # 心跳 Job：始终存在（可服务/可刷新可验证） try: scheduler.remove_job("scheduler_heartbeat") except Exception: @@ -155,11 +232,9 @@ def configure_scheduler_jobs(scheduler, app, test_refresh_token) -> None: misfire_grace_time=60, ) - # Telegram 推送 Job _configure_telegram_push_job(scheduler, app) - - # P2: 异步探测轮询 Job（每 5 秒检查一次 pending 探测） _configure_probe_poll_job(scheduler, app) + _configure_pool_maintenance_jobs(scheduler) # 刷新 Job：根据 enable_scheduled 决定是否启用 try: @@ -184,7 +259,9 @@ def configure_scheduler_jobs(scheduler, app, test_refresh_token) -> None: if len(parts) != 5: raise ValueError("Cron 表达式格式错误，应为 5 段") minute, hour, day, month, day_of_week = parts - trigger = CronTrigger(minute=minute, hour=hour, day=day, month=month, day_of_week=day_of_week) + trigger = CronTrigger( + minute=minute, hour=hour, day=day, month=month, day_of_week=day_of_week + ) scheduler.add_job( func=scheduled_task, trigger=trigger, @@ -210,7 +287,9 @@ def configure_scheduler_jobs(scheduler, app, test_refresh_token) -> None: coalesce=True, misfire_grace_time=600, ) - print(f"✓ 定时任务已配置：每天凌晨 2:00 检查刷新（周期：{refresh_interval_days} 天）") + print( + f"✓ 定时任务已配置：每天凌晨 2:00 检查刷新（周期：{refresh_interval_days} 天）" + ) def init_scheduler(app, test_refresh_token): @@ -253,9 +332,17 @@ def scheduled_refresh_task(app, test_refresh_token): try: with app.app_context(): - enable_scheduled = settings_repo.get_setting("enable_scheduled_refresh", "true").lower() == "true" - use_cron = settings_repo.get_setting("use_cron_schedule", "false").lower() == "true" - refresh_interval_days = int(settings_repo.get_setting("refresh_interval_days", "30")) + enable_scheduled = ( + settings_repo.get_setting("enable_scheduled_refresh", "true").lower() + == "true" + ) + use_cron = ( + settings_repo.get_setting("use_cron_schedule", "false").lower() + == "true" + ) + refresh_interval_days = int( + settings_repo.get_setting("refresh_interval_days", "30") + ) delay_seconds = int(settings_repo.get_setting("refresh_delay_seconds", "5")) run_id = create_refresh_run(conn, "scheduled", trace_id, total=0) @@ -297,13 +384,19 @@ def scheduled_refresh_task(app, test_refresh_token): return # 获取刷新间隔配置 - delay_row = conn.execute("SELECT value FROM settings WHERE key = 'refresh_delay_seconds'").fetchone() + delay_row = conn.execute( + "SELECT value FROM settings WHERE key = 'refresh_delay_seconds'" + ).fetchone() delay_seconds = int(delay_row["value"]) if delay_row else 5 # 清理超过半年的刷新记录 try: - conn.execute("DELETE FROM account_refresh_logs WHERE created_at < datetime('now', '-6 months')") - conn.execute("DELETE FROM distributed_locks WHERE expires_at < ?", (time.time(),)) + conn.execute( + "DELETE FROM account_refresh_logs WHERE created_at < datetime('now', '-6 months')" + ) + conn.execute( + "DELETE FROM distributed_locks WHERE expires_at < ?", (time.time(),) + ) conn.commit() except Exception: pass @@ -326,9 +419,13 @@ def scheduled_refresh_task(app, test_refresh_token): ttl_seconds = max(60 * 60 * 2, estimated) ttl_seconds = min(ttl_seconds, 60 * 60 * 24) - ok, lock_info = acquire_distributed_lock(conn, REFRESH_LOCK_NAME, lock_owner_id, ttl_seconds) + ok, lock_info = acquire_distributed_lock( + conn, REFRESH_LOCK_NAME, lock_owner_id, ttl_seconds + ) if not ok: - finish_refresh_run(conn, run_id, "skipped", total, 0, 0, "刷新任务冲突：已有刷新在执行") + finish_refresh_run( + conn, run_id, "skipped", total, 0, 0, "刷新任务冲突：已有刷新在执行" + ) return lock_acquired = True @@ -343,7 +440,11 @@ def scheduled_refresh_task(app, test_refresh_token): # 解密 refresh_token try: - refresh_token = decrypt_data(encrypted_refresh_token) if encrypted_refresh_token else encrypted_refresh_token + refresh_token = ( + decrypt_data(encrypted_refresh_token) + if encrypted_refresh_token + else encrypted_refresh_token + ) except Exception as e: failed_count += 1 error_msg = f"解密 token 失败: {str(e)}" @@ -372,13 +473,17 @@ def scheduled_refresh_task(app, test_refresh_token): group_id = account["group_id"] if group_id: try: - group_row = conn.execute("SELECT proxy_url FROM groups WHERE id = ?", (group_id,)).fetchone() + group_row = conn.execute( + "SELECT proxy_url FROM groups WHERE id = ?", (group_id,) + ).fetchone() if group_row: proxy_url = group_row["proxy_url"] or "" except Exception: proxy_url = "" - success, error_msg, new_refresh_token = test_refresh_token(client_id, refresh_token, proxy_url) + success, error_msg, new_refresh_token = test_refresh_token( + client_id, refresh_token, proxy_url + ) try: conn.execute( @@ -398,7 +503,11 @@ def scheduled_refresh_task(app, test_refresh_token): if success: # refresh token 可能滚动更新：保存新的 refresh_token（加密存储） - if isinstance(new_refresh_token, str) and new_refresh_token.strip() and new_refresh_token != refresh_token: + if ( + isinstance(new_refresh_token, str) + and new_refresh_token.strip() + and new_refresh_token != refresh_token + ): conn.execute( """ UPDATE accounts @@ -461,7 +570,9 @@ def should_autostart_scheduler() -> bool: return False # Flask CLI (flask run) + reloader：父进程不启动 - if config.env_true("FLASK_RUN_FROM_CLI", False) and not config.env_true("WERKZEUG_RUN_MAIN", False): + if config.env_true("FLASK_RUN_FROM_CLI", False) and not config.env_true( + "WERKZEUG_RUN_MAIN", False + ): return False return True diff --git a/tests/test_pool.py b/tests/test_pool.py new file mode 100644 index 0000000..69e88b9 --- /dev/null +++ b/tests/test_pool.py @@ -0,0 +1,933 @@ +import json +import unittest + +from tests._import_app import import_web_app_module + + +class PoolRepositoryTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.module = import_web_app_module() + from outlook_web.repositories import pool as pool_repo + from outlook_web.db import create_sqlite_connection + + cls.pool_repo = pool_repo + cls.create_conn = staticmethod(lambda: create_sqlite_connection()) + + def _make_account(self, conn, email_suffix="", pool_status="available"): + import secrets + + email = f"pool_test_{secrets.token_hex(4)}{email_suffix}@example.com" + conn.execute( + """ + INSERT INTO accounts (email, client_id, refresh_token, status, pool_status) + VALUES (?, 'test_client', 'test_token', 'active', ?) + """, + (email, pool_status), + ) + conn.commit() + row = conn.execute( + "SELECT id FROM accounts WHERE email = ?", (email,) + ).fetchone() + return row["id"] + + def test_claim_and_complete_success(self): + conn = self.create_conn() + try: + self._make_account(conn) + result = self.pool_repo.claim_atomic( + conn, caller_id="reg_bot", task_id="task_001", lease_seconds=60 + ) + self.assertIsNotNone(result) + self.assertTrue(result["claim_token"].startswith("clm_")) + claimed_id = result["id"] + + row = conn.execute( + "SELECT pool_status, claim_token FROM accounts WHERE id = ?", + (claimed_id,), + ).fetchone() + self.assertEqual(row["pool_status"], "claimed") + + new_status = self.pool_repo.complete( + conn, + account_id=claimed_id, + claim_token=result["claim_token"], + caller_id="reg_bot", + task_id="task_001", + result="success", + detail="注册成功", + ) + self.assertEqual(new_status, "used") + + row2 = conn.execute( + "SELECT pool_status, success_count FROM accounts WHERE id = ?", + (claimed_id,), + ).fetchone() + self.assertEqual(row2["pool_status"], "used") + self.assertEqual(row2["success_count"], 1) + finally: + conn.close() + + def test_claim_and_release(self): + conn = self.create_conn() + try: + account_id = self._make_account(conn) + result = self.pool_repo.claim_atomic( + conn, caller_id="reg_bot", task_id="task_002", lease_seconds=60 + ) + self.assertIsNotNone(result) + + self.pool_repo.release( + conn, + account_id=account_id, + claim_token=result["claim_token"], + caller_id="reg_bot", + task_id="task_002", + reason="任务取消", + ) + + row = conn.execute( + "SELECT pool_status, claim_token FROM accounts WHERE id = ?", + (account_id,), + ).fetchone() + self.assertEqual(row["pool_status"], "available") + self.assertIsNone(row["claim_token"]) + finally: + conn.close() + + def test_claim_result_maps_to_cooldown(self): + conn = self.create_conn() + try: + self._make_account(conn) + result = self.pool_repo.claim_atomic( + conn, caller_id="reg_bot", task_id="task_003", lease_seconds=60 + ) + self.assertIsNotNone(result) + claimed_id = result["id"] + + before = conn.execute( + "SELECT fail_count FROM accounts WHERE id = ?", (claimed_id,) + ).fetchone()["fail_count"] + + new_status = self.pool_repo.complete( + conn, + account_id=claimed_id, + claim_token=result["claim_token"], + caller_id="reg_bot", + task_id="task_003", + result="verification_timeout", + detail=None, + ) + self.assertEqual(new_status, "cooldown") + + row = conn.execute( + "SELECT pool_status, fail_count FROM accounts WHERE id = ?", + (claimed_id,), + ).fetchone() + self.assertEqual(row["pool_status"], "cooldown") + self.assertEqual(row["fail_count"], before + 1) + finally: + conn.close() + + def test_no_available_account_returns_none(self): + conn = self.create_conn() + try: + result = self.pool_repo.claim_atomic( + conn, + caller_id="reg_bot", + task_id="task_none", + lease_seconds=60, + provider="nonexistent_provider_xyz", + ) + self.assertIsNone(result) + finally: + conn.close() + + def test_expire_stale_claims(self): + conn = self.create_conn() + try: + account_id = self._make_account(conn) + conn.execute( + """ + UPDATE accounts SET + pool_status = 'claimed', + claimed_by = 'reg_bot:task_exp', + claim_token = 'clm_expired_test', + lease_expires_at = '2000-01-01T00:00:00Z', + updated_at = datetime('now') + WHERE id = ? + """, + (account_id,), + ) + conn.commit() + + expired_count = self.pool_repo.expire_stale_claims(conn) + self.assertGreaterEqual(expired_count, 1) + + row = conn.execute( + "SELECT pool_status FROM accounts WHERE id = ?", (account_id,) + ).fetchone() + self.assertEqual(row["pool_status"], "cooldown") + finally: + conn.close() + + def test_recover_cooldown(self): + conn = self.create_conn() + try: + account_id = self._make_account(conn, pool_status="cooldown") + conn.execute( + "UPDATE accounts SET updated_at = '2000-01-01T00:00:00' WHERE id = ?", + (account_id,), + ) + conn.commit() + + recovered = self.pool_repo.recover_cooldown(conn, cooldown_seconds=3600) + self.assertGreaterEqual(recovered, 1) + + row = conn.execute( + "SELECT pool_status FROM accounts WHERE id = ?", (account_id,) + ).fetchone() + self.assertEqual(row["pool_status"], "available") + finally: + conn.close() + + def test_get_stats_shape(self): + conn = self.create_conn() + try: + stats = self.pool_repo.get_stats(conn) + self.assertIn("pool_counts", stats) + self.assertNotIn("today", stats) + self.assertNotIn("overall", stats) + self.assertIn("available", stats["pool_counts"]) + self.assertIn("claimed", stats["pool_counts"]) + finally: + conn.close() + + def test_complete_network_error_returns_available_and_increments_fail_count(self): + conn = self.create_conn() + try: + account_id = self._make_account(conn) + result = self.pool_repo.claim_atomic( + conn, caller_id="reg_bot", task_id="task_ne", lease_seconds=60 + ) + self.assertIsNotNone(result) + claimed_id = result["id"] + + before = conn.execute( + "SELECT fail_count FROM accounts WHERE id = ?", (claimed_id,) + ).fetchone()["fail_count"] + + new_status = self.pool_repo.complete( + conn, + account_id=claimed_id, + claim_token=result["claim_token"], + caller_id="reg_bot", + task_id="task_ne", + result="network_error", + detail=None, + ) + self.assertEqual(new_status, "available") + + row = conn.execute( + "SELECT pool_status, fail_count FROM accounts WHERE id = ?", + (claimed_id,), + ).fetchone() + self.assertEqual(row["pool_status"], "available") + self.assertEqual(row["fail_count"], before + 1) + finally: + conn.close() + + def test_complete_provider_blocked_returns_frozen(self): + conn = self.create_conn() + try: + account_id = self._make_account(conn) + result = self.pool_repo.claim_atomic( + conn, caller_id="reg_bot", task_id="task_pb", lease_seconds=60 + ) + self.assertIsNotNone(result) + claimed_id = result["id"] + + before = conn.execute( + "SELECT fail_count FROM accounts WHERE id = ?", (claimed_id,) + ).fetchone()["fail_count"] + + new_status = self.pool_repo.complete( + conn, + account_id=claimed_id, + claim_token=result["claim_token"], + caller_id="reg_bot", + task_id="task_pb", + result="provider_blocked", + detail="IP 被封", + ) + self.assertEqual(new_status, "frozen") + + row = conn.execute( + "SELECT pool_status, fail_count FROM accounts WHERE id = ?", + (claimed_id,), + ).fetchone() + self.assertEqual(row["pool_status"], "frozen") + self.assertEqual(row["fail_count"], before + 1) + finally: + conn.close() + + def test_complete_credential_invalid_returns_retired(self): + conn = self.create_conn() + try: + account_id = self._make_account(conn) + result = self.pool_repo.claim_atomic( + conn, caller_id="reg_bot", task_id="task_ci", lease_seconds=60 + ) + self.assertIsNotNone(result) + claimed_id = result["id"] + + before = conn.execute( + "SELECT fail_count FROM accounts WHERE id = ?", (claimed_id,) + ).fetchone()["fail_count"] + + new_status = self.pool_repo.complete( + conn, + account_id=claimed_id, + claim_token=result["claim_token"], + caller_id="reg_bot", + task_id="task_ci", + result="credential_invalid", + detail=None, + ) + self.assertEqual(new_status, "retired") + + row = conn.execute( + "SELECT pool_status, fail_count FROM accounts WHERE id = ?", + (claimed_id,), + ).fetchone() + self.assertEqual(row["pool_status"], "retired") + self.assertEqual(row["fail_count"], before + 1) + finally: + conn.close() + + def test_claim_log_written_on_claim(self): + conn = self.create_conn() + try: + account_id = self._make_account(conn) + result = self.pool_repo.claim_atomic( + conn, caller_id="log_bot", task_id="log_task_001", lease_seconds=60 + ) + self.assertIsNotNone(result) + claimed_id = result["id"] + log_row = conn.execute( + """ + SELECT * FROM account_claim_logs + WHERE account_id = ? AND action = 'claim' + ORDER BY created_at DESC LIMIT 1 + """, + (claimed_id,), + ).fetchone() + self.assertIsNotNone(log_row) + self.assertEqual(log_row["caller_id"], "log_bot") + self.assertEqual(log_row["task_id"], "log_task_001") + self.assertEqual(log_row["claim_token"], result["claim_token"]) + self.assertIsNone(log_row["result"]) + finally: + conn.close() + + def test_complete_log_written_on_complete(self): + conn = self.create_conn() + try: + account_id = self._make_account(conn) + result = self.pool_repo.claim_atomic( + conn, caller_id="log_bot", task_id="log_task_002", lease_seconds=60 + ) + self.assertIsNotNone(result) + claimed_id = result["id"] + self.pool_repo.complete( + conn, + account_id=claimed_id, + claim_token=result["claim_token"], + caller_id="log_bot", + task_id="log_task_002", + result="success", + detail="ok", + ) + log_row = conn.execute( + """ + SELECT * FROM account_claim_logs + WHERE account_id = ? AND action = 'complete' + ORDER BY created_at DESC LIMIT 1 + """, + (claimed_id,), + ).fetchone() + self.assertIsNotNone(log_row) + self.assertEqual(log_row["result"], "success") + self.assertEqual(log_row["detail"], "ok") + finally: + conn.close() + + def test_expire_stale_claims_increments_fail_count(self): + conn = self.create_conn() + try: + account_id = self._make_account(conn) + conn.execute( + """ + UPDATE accounts SET + pool_status = 'claimed', + claimed_by = 'bot:task_exp2', + claim_token = 'clm_exp2_test', + lease_expires_at = '2000-01-01T00:00:00Z', + updated_at = datetime('now') + WHERE id = ? + """, + (account_id,), + ) + conn.commit() + + self.pool_repo.expire_stale_claims(conn) + + row = conn.execute( + "SELECT pool_status, fail_count FROM accounts WHERE id = ?", + (account_id,), + ).fetchone() + self.assertEqual(row["pool_status"], "cooldown") + self.assertEqual(row["fail_count"], 1) + finally: + conn.close() + + def test_exclude_recent_minutes_skips_recently_claimed_account(self): + import time as _time + + conn = self.create_conn() + try: + account_id = self._make_account(conn) + result1 = self.pool_repo.claim_atomic( + conn, caller_id="bot", task_id="excl_t1", lease_seconds=60 + ) + self.assertIsNotNone(result1) + claimed_id = result1["id"] + self.pool_repo.release( + conn, + account_id=claimed_id, + claim_token=result1["claim_token"], + caller_id="bot", + task_id="excl_t1", + reason="test", + ) + + conn.execute( + "UPDATE accounts SET pool_status = 'used' WHERE pool_status = 'available' AND id != ?", + (claimed_id,), + ) + conn.commit() + + result2 = self.pool_repo.claim_atomic( + conn, + caller_id="bot", + task_id="excl_t2", + lease_seconds=60, + exclude_recent_minutes=60, + ) + self.assertIsNone(result2) + finally: + conn.close() + + def test_multiple_claim_tokens_are_unique(self): + conn = self.create_conn() + try: + tokens = set() + for i in range(5): + account_id = self._make_account(conn) + result = self.pool_repo.claim_atomic( + conn, caller_id="bot", task_id=f"uniq_t{i}", lease_seconds=60 + ) + self.assertIsNotNone(result) + tokens.add(result["claim_token"]) + self.pool_repo.complete( + conn, + account_id=result["id"], + claim_token=result["claim_token"], + caller_id="bot", + task_id=f"uniq_t{i}", + result="success", + detail=None, + ) + self.assertEqual(len(tokens), 5, "5 次 claim 应产生 5 个不同的 token") + finally: + conn.close() + + def test_concurrent_claim_unique(self): + """并发场景：多个 claim 不应领到同一个邮箱（P0-6）""" + conn = self.create_conn() + try: + import secrets + + iso_provider = f"conc_iso_{secrets.token_hex(4)}" + email = f"pool_conc_{secrets.token_hex(4)}@example.com" + conn.execute( + """ + INSERT INTO accounts (email, client_id, refresh_token, status, pool_status, provider) + VALUES (?, 'test_client', 'test_token', 'active', 'available', ?) + """, + (email, iso_provider), + ) + conn.commit() + + r1 = self.pool_repo.claim_atomic( + conn, + caller_id="bot_a", + task_id="conc_1", + lease_seconds=60, + provider=iso_provider, + ) + self.assertIsNotNone(r1) + + r2 = self.pool_repo.claim_atomic( + conn, + caller_id="bot_b", + task_id="conc_2", + lease_seconds=60, + provider=iso_provider, + ) + self.assertIsNone(r2, "只有 1 个可用邮箱时，第二个 claim 应返回 None") + finally: + conn.close() + + def test_release_log_written_on_release(self): + """release 动作应写入 claim log（P0-6）""" + conn = self.create_conn() + try: + account_id = self._make_account(conn) + result = self.pool_repo.claim_atomic( + conn, caller_id="log_bot", task_id="log_rel_001", lease_seconds=60 + ) + self.assertIsNotNone(result) + claimed_id = result["id"] + + self.pool_repo.release( + conn, + account_id=claimed_id, + claim_token=result["claim_token"], + caller_id="log_bot", + task_id="log_rel_001", + reason="任务取消", + ) + + log_row = conn.execute( + """ + SELECT * FROM account_claim_logs + WHERE account_id = ? AND action = 'release' + ORDER BY created_at DESC LIMIT 1 + """, + (claimed_id,), + ).fetchone() + self.assertIsNotNone(log_row, "release 应写入 claim log") + self.assertEqual(log_row["caller_id"], "log_bot") + self.assertEqual(log_row["task_id"], "log_rel_001") + self.assertEqual(log_row["claim_token"], result["claim_token"]) + self.assertEqual(log_row["result"], "manual_release") + self.assertEqual(log_row["detail"], "任务取消") + finally: + conn.close() + + def test_expire_log_written_on_expire(self): + """expire（租约超时）动作应写入 claim log（P0-6）""" + conn = self.create_conn() + try: + account_id = self._make_account(conn) + token = "clm_expire_log_test" + conn.execute( + """ + UPDATE accounts SET + pool_status = 'claimed', + claimed_by = 'exp_bot:exp_task', + claim_token = ?, + lease_expires_at = '2000-01-01T00:00:00Z', + updated_at = datetime('now') + WHERE id = ? + """, + (token, account_id), + ) + conn.commit() + + self.pool_repo.expire_stale_claims(conn) + + log_row = conn.execute( + """ + SELECT * FROM account_claim_logs + WHERE account_id = ? AND action = 'expire' + ORDER BY created_at DESC LIMIT 1 + """, + (account_id,), + ).fetchone() + self.assertIsNotNone(log_row, "expire 应写入 claim log") + self.assertEqual(log_row["caller_id"], "exp_bot") + self.assertEqual(log_row["task_id"], "exp_task") + self.assertEqual(log_row["claim_token"], token) + self.assertEqual(log_row["result"], "lease_expired") + finally: + conn.close() + + def test_expire_affects_stats(self): + """expire 后 stats 中 cooldown 应增加、claimed 应减少（P0-6）""" + conn = self.create_conn() + try: + account_id = self._make_account(conn) + conn.execute( + """ + UPDATE accounts SET + pool_status = 'claimed', + claimed_by = 'stat_bot:stat_task', + claim_token = 'clm_stat_test', + lease_expires_at = '2000-01-01T00:00:00Z', + updated_at = datetime('now') + WHERE id = ? + """, + (account_id,), + ) + conn.commit() + + stats_before = self.pool_repo.get_stats(conn) + claimed_before = stats_before["pool_counts"]["claimed"] + cooldown_before = stats_before["pool_counts"]["cooldown"] + + self.pool_repo.expire_stale_claims(conn) + + stats_after = self.pool_repo.get_stats(conn) + claimed_after = stats_after["pool_counts"]["claimed"] + cooldown_after = stats_after["pool_counts"]["cooldown"] + + self.assertLess( + claimed_after, claimed_before, "expire 后 claimed 数量应减少" + ) + self.assertGreater( + cooldown_after, cooldown_before, "expire 后 cooldown 数量应增加" + ) + finally: + conn.close() + + +class PoolServiceTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.module = import_web_app_module() + from outlook_web.services import pool as pool_service + from outlook_web.db import create_sqlite_connection + + cls.pool_service = pool_service + cls.create_conn = staticmethod(lambda: create_sqlite_connection()) + + def _make_account(self, pool_status="available"): + import secrets + + conn = self.create_conn() + try: + email = f"svc_test_{secrets.token_hex(4)}@example.com" + conn.execute( + """ + INSERT INTO accounts (email, client_id, refresh_token, status, pool_status) + VALUES (?, 'test_client', 'test_token', 'active', ?) + """, + (email, pool_status), + ) + conn.commit() + row = conn.execute( + "SELECT id FROM accounts WHERE email = ?", (email,) + ).fetchone() + return row["id"] + finally: + conn.close() + + def test_claim_random_returns_account(self): + self._make_account() + result = self.pool_service.claim_random(caller_id="bot", task_id="t1") + self.assertIn("claim_token", result) + self.assertIn("lease_expires_at", result) + + def test_claim_random_caller_id_validation(self): + with self.assertRaises(self.pool_service.PoolServiceError) as ctx: + self.pool_service.claim_random(caller_id="", task_id="t1") + self.assertEqual(ctx.exception.error_code, "caller_id_empty") + + def test_claim_random_no_account_200(self): + with self.assertRaises(self.pool_service.PoolServiceError) as ctx: + self.pool_service.claim_random( + caller_id="bot", + task_id="t_no_acct", + provider="provider_that_does_not_exist_xyz", + ) + self.assertEqual(ctx.exception.http_status, 200) + self.assertEqual(ctx.exception.error_code, "no_available_account") + + def test_complete_claim_invalid_result(self): + account_id = self._make_account() + with self.assertRaises(self.pool_service.PoolServiceError) as ctx: + self.pool_service.complete_claim( + account_id=account_id, + claim_token="clm_fake", + caller_id="bot", + task_id="t2", + result="not_a_valid_result", + ) + self.assertEqual(ctx.exception.error_code, "invalid_result") + + def test_release_claim_token_mismatch(self): + self._make_account() + result = self.pool_service.claim_random(caller_id="bot", task_id="t_rel") + claimed_id = result["id"] + with self.assertRaises(self.pool_service.PoolServiceError) as ctx: + self.pool_service.release_claim( + account_id=claimed_id, + claim_token="clm_wrong_token", + caller_id="bot", + task_id="t_rel", + ) + self.assertEqual(ctx.exception.error_code, "token_mismatch") + + def test_release_account_not_found(self): + """release 对不存在的 account_id 应返回 account_not_found (400)""" + with self.assertRaises(self.pool_service.PoolServiceError) as ctx: + self.pool_service.release_claim( + account_id=999999, + claim_token="clm_any", + caller_id="bot", + task_id="t_release_anf", + ) + self.assertEqual(ctx.exception.error_code, "account_not_found") + self.assertEqual(ctx.exception.http_status, 400) + + def test_release_not_claimed(self): + """release 对 pool_status != claimed 的账号应返回 not_claimed (409)""" + account_id = self._make_account(pool_status="available") + with self.assertRaises(self.pool_service.PoolServiceError) as ctx: + self.pool_service.release_claim( + account_id=account_id, + claim_token="clm_any", + caller_id="bot", + task_id="t_release_nc", + ) + self.assertEqual(ctx.exception.error_code, "not_claimed") + self.assertEqual(ctx.exception.http_status, 409) + + def test_release_caller_mismatch(self): + """release 使用正确 token 但不同 caller_id:task_id 应返回 caller_mismatch (403)""" + self._make_account() + result = self.pool_service.claim_random(caller_id="bot_a", task_id="t_cm") + claimed_id = result["id"] + with self.assertRaises(self.pool_service.PoolServiceError) as ctx: + self.pool_service.release_claim( + account_id=claimed_id, + claim_token=result["claim_token"], + caller_id="bot_b", + task_id="t_cm", + ) + self.assertEqual(ctx.exception.error_code, "caller_mismatch") + self.assertEqual(ctx.exception.http_status, 403) + + def test_complete_account_not_found(self): + """complete 对不存在的 account_id 应返回 account_not_found (400)""" + with self.assertRaises(self.pool_service.PoolServiceError) as ctx: + self.pool_service.complete_claim( + account_id=999999, + claim_token="clm_any", + caller_id="bot", + task_id="t_comp_anf", + result="success", + ) + self.assertEqual(ctx.exception.error_code, "account_not_found") + self.assertEqual(ctx.exception.http_status, 400) + + def test_complete_not_claimed(self): + """complete 对 pool_status != claimed 的账号应返回 not_claimed (409)""" + account_id = self._make_account(pool_status="available") + with self.assertRaises(self.pool_service.PoolServiceError) as ctx: + self.pool_service.complete_claim( + account_id=account_id, + claim_token="clm_any", + caller_id="bot", + task_id="t_comp_nc", + result="success", + ) + self.assertEqual(ctx.exception.error_code, "not_claimed") + self.assertEqual(ctx.exception.http_status, 409) + + def test_complete_token_mismatch(self): + """complete 使用错误 token 应返回 token_mismatch (403)""" + self._make_account() + result = self.pool_service.claim_random(caller_id="bot", task_id="t_comp_tm") + claimed_id = result["id"] + with self.assertRaises(self.pool_service.PoolServiceError) as ctx: + self.pool_service.complete_claim( + account_id=claimed_id, + claim_token="clm_wrong_token", + caller_id="bot", + task_id="t_comp_tm", + result="success", + ) + self.assertEqual(ctx.exception.error_code, "token_mismatch") + self.assertEqual(ctx.exception.http_status, 403) + + def test_complete_caller_mismatch(self): + """complete 使用正确 token 但不同 caller_id:task_id 应返回 caller_mismatch (403)""" + self._make_account() + result = self.pool_service.claim_random(caller_id="bot_a", task_id="t_comp_cm") + claimed_id = result["id"] + with self.assertRaises(self.pool_service.PoolServiceError) as ctx: + self.pool_service.complete_claim( + account_id=claimed_id, + claim_token=result["claim_token"], + caller_id="bot_b", + task_id="t_comp_cm", + result="success", + ) + self.assertEqual(ctx.exception.error_code, "caller_mismatch") + self.assertEqual(ctx.exception.http_status, 403) + + def test_ownership_check_order_release(self): + """验证 release 校验顺序：not_claimed 应优先于 token_mismatch（TD §5.4）""" + account_id = self._make_account(pool_status="available") + with self.assertRaises(self.pool_service.PoolServiceError) as ctx: + self.pool_service.release_claim( + account_id=account_id, + claim_token="clm_wrong", + caller_id="bot_wrong", + task_id="t_wrong", + ) + self.assertEqual(ctx.exception.error_code, "not_claimed") + + def test_ownership_check_order_complete(self): + """验证 complete 校验顺序：token_mismatch 应优先于 caller_mismatch（TD §5.4）""" + self._make_account() + result = self.pool_service.claim_random(caller_id="bot_a", task_id="t_order") + claimed_id = result["id"] + with self.assertRaises(self.pool_service.PoolServiceError) as ctx: + self.pool_service.complete_claim( + account_id=claimed_id, + claim_token="clm_wrong", + caller_id="bot_b", + task_id="t_order", + result="success", + ) + self.assertEqual(ctx.exception.error_code, "token_mismatch") + + +class PoolApiTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.module = import_web_app_module() + cls.client = cls.module.app.test_client() + from outlook_web.db import create_sqlite_connection + + cls.create_conn = staticmethod(lambda: create_sqlite_connection()) + + def _make_account(self): + import secrets + + conn = self.create_conn() + try: + email = f"api_test_{secrets.token_hex(4)}@example.com" + conn.execute( + """ + INSERT INTO accounts (email, client_id, refresh_token, status, pool_status) + VALUES (?, 'test_client', 'test_token', 'active', 'available') + """, + (email,), + ) + conn.commit() + row = conn.execute( + "SELECT id FROM accounts WHERE email = ?", (email,) + ).fetchone() + return row["id"] + finally: + conn.close() + + def test_stats_endpoint(self): + resp = self.client.get("/api/pool/stats") + self.assertEqual(resp.status_code, 200) + data = json.loads(resp.data) + self.assertTrue(data["success"]) + self.assertIn("data", data) + self.assertIn("pool_counts", data["data"]) + + def test_claim_random_endpoint(self): + self._make_account() + resp = self.client.post( + "/api/pool/claim-random", + json={ + "caller_id": "test_bot", + "task_id": "api_task_1", + }, + ) + self.assertEqual(resp.status_code, 200) + data = json.loads(resp.data) + self.assertTrue(data["success"]) + self.assertIn("data", data) + self.assertIn("account_id", data["data"]) + self.assertIn("email", data["data"]) + self.assertIn("claim_token", data["data"]) + self.assertIn("lease_expires_at", data["data"]) + + def test_claim_random_missing_caller_id(self): + resp = self.client.post( + "/api/pool/claim-random", + json={"task_id": "api_task_2"}, + ) + self.assertEqual(resp.status_code, 400) + data = json.loads(resp.data) + self.assertFalse(data["success"]) + self.assertIn("error", data) + + def test_claim_complete_flow(self): + self._make_account() + claim_resp = self.client.post( + "/api/pool/claim-random", + json={"caller_id": "test_bot", "task_id": "flow_task"}, + ) + self.assertEqual(claim_resp.status_code, 200) + claim_data = json.loads(claim_resp.data) + account_data = claim_data["data"] + + complete_resp = self.client.post( + "/api/pool/claim-complete", + json={ + "account_id": account_data["account_id"], + "claim_token": account_data["claim_token"], + "caller_id": "test_bot", + "task_id": "flow_task", + "result": "success", + "detail": "test ok", + }, + ) + self.assertEqual(complete_resp.status_code, 200) + complete_data = json.loads(complete_resp.data) + self.assertTrue(complete_data["success"]) + self.assertEqual( + complete_data["data"]["account_id"], account_data["account_id"] + ) + self.assertEqual(complete_data["data"]["pool_status"], "used") + + def test_claim_release_flow(self): + self._make_account() + claim_resp = self.client.post( + "/api/pool/claim-random", + json={"caller_id": "test_bot", "task_id": "rel_task"}, + ) + self.assertEqual(claim_resp.status_code, 200) + claim_data = json.loads(claim_resp.data) + account_data = claim_data["data"] + + release_resp = self.client.post( + "/api/pool/claim-release", + json={ + "account_id": account_data["account_id"], + "claim_token": account_data["claim_token"], + "caller_id": "test_bot", + "task_id": "rel_task", + "reason": "test cancel", + }, + ) + self.assertEqual(release_resp.status_code, 200) + data = json.loads(release_resp.data) + self.assertTrue(data["success"]) + self.assertEqual(data["data"]["account_id"], account_data["account_id"]) + self.assertEqual(data["data"]["pool_status"], "available") + + +if __name__ == "__main__": + unittest.main()