# -*- coding: utf-8 -*- """ Database schema v2 — Credential & Browser Profile Manager. Time stored as INTEGER Unix seconds (UTC). DDL comments are preserved in sqlite_master, visible in DBeaver / Navicat. Schema version management: - PRAGMA user_version = 2 (set in connection.py after ensure_schema). - _schema_meta schema_version: "2" after migrate_v1_to_v2, "2.1" after migrate_v2_to_v2_1. """ import sqlite3 import sys import textwrap from db.auth_strategy import AUTH_STRATEGY_PASSWORD_AUTO # --------------------------------------------------------------------------- # platforms — Registry of all platforms (logistics, LLM, content, etc.) # --------------------------------------------------------------------------- PLATFORMS_TABLE_SQL = textwrap.dedent("""\ CREATE TABLE IF NOT EXISTS platforms ( platform_key TEXT PRIMARY KEY, -- 平台唯一键，例如 maersk / deepseek / sohu display_name TEXT NOT NULL, -- 展示名称，例如 Maersk / DeepSeek / 搜狐号 domain TEXT NOT NULL, -- 业务域：logistics / llm / content / ecommerce / generic provider_code TEXT, -- 供应商代码，物流 profile 可复用；例如 maersk、cma_cgm default_url TEXT, -- 默认入口 URL aliases_json TEXT NOT NULL DEFAULT '[]',-- 别名 JSON 数组，中英文别名 capabilities_json TEXT NOT NULL DEFAULT '{}', -- 能力声明，例如 {"web":true,"api_key":true,"rpa":true} enabled INTEGER NOT NULL DEFAULT 1,-- 是否启用 (1=yes 0=no) default_auth_strategy TEXT, -- 新账号默认认证策略；可空，手工维护 created_at INTEGER NOT NULL, -- 记录创建时间，Unix 秒 UTC updated_at INTEGER NOT NULL -- 记录最后更新时间，Unix 秒 UTC ); """) PLATFORMS_INDEXES_SQL = textwrap.dedent("""\ CREATE INDEX IF NOT EXISTS idx_platforms_domain ON platforms(domain); CREATE INDEX IF NOT EXISTS idx_platforms_provider_code ON platforms(provider_code); """) # --------------------------------------------------------------------------- # accounts — An identity that can be used (web account, RPA profile, API account) # --------------------------------------------------------------------------- ACCOUNTS_TABLE_SQL = textwrap.dedent("""\ CREATE TABLE IF NOT EXISTS accounts ( id INTEGER PRIMARY KEY AUTOINCREMENT, -- 账号主键（自增） platform_key TEXT NOT NULL, -- 对应 platforms.platform_key account_label TEXT NOT NULL, -- 人可读名称，例如 Maersk simulator booking login_id TEXT, -- 登录标识：邮箱、用户名、手机号、客户代码等 login_id_type TEXT NOT NULL DEFAULT 'unknown', -- email / username / phone / customer_code / api_account / unknown tenant_id TEXT, -- 租户/客户/项目标识，可空 environment TEXT NOT NULL DEFAULT 'production',-- simulator / staging / production / test role TEXT NOT NULL DEFAULT 'default', -- booking / tracking / admin / sales / api / default provider_code TEXT, -- 冗余供应商码，便于业务 skill 查询 profile_dir TEXT, -- Playwright 持久化浏览器用户数据目录；web/rpa 账号使用 url TEXT, -- 账号默认入口 URL；为空时用 platforms.default_url status TEXT NOT NULL DEFAULT 'active', -- active / disabled / needs_review session_status TEXT NOT NULL DEFAULT 'unknown', -- unknown / likely_valid / needs_login / expired last_used_at INTEGER, -- 最近被业务 skill pick/使用时间 last_login_check_at INTEGER, -- 最近一次登录状态检查时间，可空 last_error_code TEXT, -- 最近一次错误代码 last_error_message TEXT, -- 最近一次错误消息 extra_json TEXT NOT NULL DEFAULT '{}', -- 扩展字段 JSON auth_strategy TEXT NOT NULL, -- 认证策略（见 db.auth_strategy） session_persistent INTEGER NOT NULL DEFAULT 1, -- 是否需要 profile 持久化 (0/1) device_fingerprint TEXT, -- device_bound 策略用；可空 created_at INTEGER NOT NULL, -- 记录创建时间，Unix 秒 UTC updated_at INTEGER NOT NULL -- 记录最后更新时间，Unix 秒 UTC ); """) ACCOUNTS_INDEXES_SQL = textwrap.dedent("""\ CREATE INDEX IF NOT EXISTS idx_accounts_platform ON accounts(platform_key); CREATE INDEX IF NOT EXISTS idx_accounts_pick_web ON accounts(platform_key, environment, tenant_id, role, status, updated_at); CREATE INDEX IF NOT EXISTS idx_accounts_provider ON accounts(provider_code); CREATE INDEX IF NOT EXISTS idx_accounts_status ON accounts(status); """) # Note: Unique constraint (platform_key, login_id, tenant_id, environment, role) # is intentionally NOT added here because SQLite allows multiple NULL login_id. # Application-level checks are used when login_id is provided. # --------------------------------------------------------------------------- # credentials — Secret-type credentials (API Key, Token, Password, etc.) # Raw secret is NEVER stored in SQLite. Only metadata, secret_ref, secret_mask. # --------------------------------------------------------------------------- CREDENTIALS_TABLE_SQL = textwrap.dedent("""\ CREATE TABLE IF NOT EXISTS credentials ( id INTEGER PRIMARY KEY AUTOINCREMENT, -- 凭据主键（自增） account_id INTEGER, -- 可空；关联 accounts.id；有些 API Key 可直接挂平台，也可挂具体账号 platform_key TEXT NOT NULL, -- 平台唯一键 credential_label TEXT NOT NULL, -- 人可读名称，例如 DeepSeek main API key credential_type TEXT NOT NULL, -- api_key / password / bearer_token / oauth_client / refresh_token / browser_profile / note secret_storage TEXT NOT NULL, -- none / env / windows_credential / local_encrypted secret_ref TEXT, -- env 时为环境变量名；windows_credential 时为 target name；none/local_encrypted 时为空 secret_mask TEXT, -- 打码展示，例如 sk-****abcd；不能反推出原文 secret_ciphertext TEXT, -- secret_storage='local_encrypted' 时为 Fernet 密文；其它后端为 NULL expires_at INTEGER, -- 过期时间，Unix 秒 UTC；可空 status TEXT NOT NULL DEFAULT 'active', -- active / disabled / expired / needs_rotation last_used_at INTEGER, -- 最近使用时间 extra_json TEXT NOT NULL DEFAULT '{}', -- 扩展字段 JSON created_at INTEGER NOT NULL, -- 记录创建时间，Unix 秒 UTC updated_at INTEGER NOT NULL -- 记录最后更新时间，Unix 秒 UTC ); """) CREDENTIALS_INDEXES_SQL = textwrap.dedent("""\ CREATE INDEX IF NOT EXISTS idx_credentials_platform_type ON credentials(platform_key, credential_type); CREATE INDEX IF NOT EXISTS idx_credentials_status ON credentials(status); """) # --------------------------------------------------------------------------- # account_leases — RPA / profile concurrency locks # Prevents multiple tasks from opening the same profile_dir simultaneously. # --------------------------------------------------------------------------- ACCOUNT_LEASES_TABLE_SQL = textwrap.dedent("""\ CREATE TABLE IF NOT EXISTS account_leases ( id INTEGER PRIMARY KEY AUTOINCREMENT, -- 租约主键（自增） account_id INTEGER NOT NULL, -- 关联 accounts.id lease_token TEXT NOT NULL UNIQUE, -- 随机 token，用于 release / heartbeat holder TEXT NOT NULL, -- 持有者，例如 monitor-competitor-rates 或 run id purpose TEXT, -- 用途，例如 maersk_sim_rpa / booking_collect expires_at INTEGER NOT NULL, -- 租约过期时间，Unix 秒 UTC heartbeat_at INTEGER, -- 最近一次心跳时间 released_at INTEGER, -- 主动释放时间 status TEXT NOT NULL DEFAULT 'active', -- active / released / expired created_at INTEGER NOT NULL, -- 记录创建时间，Unix 秒 UTC updated_at INTEGER NOT NULL -- 记录最后更新时间，Unix 秒 UTC ); """) ACCOUNT_LEASES_INDEXES_SQL = textwrap.dedent("""\ CREATE INDEX IF NOT EXISTS idx_leases_account_status ON account_leases(account_id, status); CREATE INDEX IF NOT EXISTS idx_leases_token ON account_leases(lease_token); CREATE INDEX IF NOT EXISTS idx_leases_expires ON account_leases(expires_at); """) # --------------------------------------------------------------------------- # schema meta — logical schema version (v1→v2 migration marker) # --------------------------------------------------------------------------- SCHEMA_META_TABLE_SQL = textwrap.dedent("""\ CREATE TABLE IF NOT EXISTS _schema_meta ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); """) # --------------------------------------------------------------------------- # Full DDL assembly # --------------------------------------------------------------------------- ALL_DDL = ( PLATFORMS_TABLE_SQL + PLATFORMS_INDEXES_SQL + ACCOUNTS_TABLE_SQL + ACCOUNTS_INDEXES_SQL + CREDENTIALS_TABLE_SQL + CREDENTIALS_INDEXES_SQL + ACCOUNT_LEASES_TABLE_SQL + ACCOUNT_LEASES_INDEXES_SQL + SCHEMA_META_TABLE_SQL ) SCHEMA_VERSION = 2 def column_exists(conn: sqlite3.Connection, table: str, col: str) -> bool: cur = conn.cursor() cur.execute(f"PRAGMA table_info({table})") return any(row[1] == col for row in cur.fetchall()) def _schema_meta_get(conn: sqlite3.Connection, key: str) -> str | None: cur = conn.cursor() cur.execute("SELECT value FROM _schema_meta WHERE key = ?", (key,)) row = cur.fetchone() return str(row[0]) if row and row[0] is not None else None def _schema_meta_upsert(conn: sqlite3.Connection, key: str, value: str) -> None: cur = conn.cursor() cur.execute( """ INSERT INTO _schema_meta (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value """, (key, value), ) def migrate_v1_to_v2(conn: sqlite3.Connection) -> None: """Apply ALTER steps + backfills once; idempotent via _schema_meta.schema_version.""" from db.platform_defaults import PLATFORM_DEFAULT_AUTH_STRATEGY cur = conn.cursor() sv = _schema_meta_get(conn, "schema_version") if sv == "2" or (sv and sv.startswith("2.")): return n_platform_updates = 0 # 1–2: platforms.default_auth_strategy if not column_exists(conn, "platforms", "default_auth_strategy"): cur.execute("ALTER TABLE platforms ADD COLUMN default_auth_strategy TEXT") for pk, strat in PLATFORM_DEFAULT_AUTH_STRATEGY.items(): cur.execute( "UPDATE platforms SET default_auth_strategy = ? WHERE platform_key = ?", (strat, pk), ) n_platform_updates += cur.rowcount # 3: accounts columns (nullable auth_strategy during ALTER for legacy rows) if not column_exists(conn, "accounts", "auth_strategy"): cur.execute("ALTER TABLE accounts ADD COLUMN auth_strategy TEXT") if not column_exists(conn, "accounts", "session_persistent"): cur.execute( "ALTER TABLE accounts ADD COLUMN session_persistent INTEGER NOT NULL DEFAULT 1" ) if not column_exists(conn, "accounts", "device_fingerprint"): cur.execute("ALTER TABLE accounts ADD COLUMN device_fingerprint TEXT") # 4: fill accounts.auth_strategy + session_persistent cur.execute( """ UPDATE accounts SET auth_strategy = COALESCE( (SELECT p.default_auth_strategy FROM platforms p WHERE p.platform_key = accounts.platform_key), ? ), session_persistent = COALESCE(session_persistent, 1) WHERE auth_strategy IS NULL OR TRIM(auth_strategy) = '' """, (AUTH_STRATEGY_PASSWORD_AUTO,), ) m_accounts_updated = cur.rowcount # 5: credentials.account_id (v1 may omit it) if not column_exists(conn, "credentials", "account_id"): cur.execute("ALTER TABLE credentials ADD COLUMN account_id INTEGER") cur.execute( "CREATE INDEX IF NOT EXISTS idx_credentials_account ON credentials(account_id)" ) cur.execute("SELECT COUNT(*) FROM platforms WHERE default_auth_strategy IS NULL") k_null_platforms = int(cur.fetchone()[0]) _schema_meta_upsert(conn, "schema_version", "2") conn.commit() print( f"[migration] account-manager v1→v2: updated {n_platform_updates} platforms, " f"{m_accounts_updated} accounts.", file=sys.stderr, ) print( "[migration] platforms with default_auth_strategy=NULL: " f"{k_null_platforms} (manually fix later)", file=sys.stderr, ) def migrate_v2_to_v2_1(conn: sqlite3.Connection) -> None: """v2 → v2.1: credentials 表加 secret_ciphertext 列。幂等。""" cur = conn.cursor() if _schema_meta_get(conn, "schema_version") == "2.1": return if column_exists(conn, "credentials", "secret_ciphertext"): _schema_meta_upsert(conn, "schema_version", "2.1") conn.commit() return cur.execute("ALTER TABLE credentials ADD COLUMN secret_ciphertext TEXT") _schema_meta_upsert(conn, "schema_version", "2.1") conn.commit() print( "[migration] account-manager v2→v2.1: credentials.secret_ciphertext column added.", file=sys.stderr, ) def ensure_schema(conn: sqlite3.Connection) -> None: """CREATE TABLE/INDEX IF NOT EXISTS, then run migrate_v1_to_v2.""" cur = conn.cursor() cur.executescript(ALL_DDL) conn.commit() migrate_v1_to_v2(conn) migrate_v2_to_v2_1(conn)