account-manager/scripts/db/accounts_repo.py

"""Accounts table: CRUD and queries (no stdout)."""
import json
import time
from datetime import datetime
from typing import Any, Optional

from db.connection import get_conn, init_db
from util.platforms import PLATFORM_URLS, _PLATFORM_NAME_CN


def _now_unix() -> int:
    return int(time.time())


def _unix_to_iso_local(ts: Optional[int]) -> Optional[str]:
    """对外 JSON：本地时区 ISO8601 字符串，便于人读。"""
    if ts is None:
        return None
    try:
        return datetime.fromtimestamp(int(ts)).isoformat(timespec="seconds")
    except (ValueError, OSError, OverflowError):
        return None


def normalize_account_id(account_id):
    if account_id is None:
        return None
    s = str(account_id).strip()
    if s.isdigit():
        return int(s)
    return s


def default_name_for_platform(platform: str, index: int) -> str:
    cn = _PLATFORM_NAME_CN.get(platform)
    if cn:
        return f"{cn}{index}号"
    return f"{platform}_{index}"


def get_account_by_id(account_id):
    init_db()
    aid = normalize_account_id(account_id)
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute(
            """
            SELECT id, name, platform, phone, profile_dir, url, extra_json,
                   created_at, updated_at
            FROM accounts WHERE id = ?
            """,
            (aid,),
        )
        row = cur.fetchone()
        if not row:
            return None
        acc: dict[str, Any] = {
            "id": row[0],
            "name": row[1],
            "platform": row[2],
            "phone": row[3] or "",
            "profile_dir": (row[4] or "").strip(),
            "url": row[5] or PLATFORM_URLS.get(row[2], ""),
            "created_at": _unix_to_iso_local(row[7]),
            "updated_at": _unix_to_iso_local(row[8]),
        }
        if row[6]:
            try:
                extra = json.loads(row[6])
                if isinstance(extra, dict):
                    acc.update(extra)
            except Exception:
                pass
        return acc
    finally:
        conn.close()


def fetch_list_rows(platform_key: str, limit: int):
    """platform_key 'all' or platform slug; returns list of row tuples (full row)."""
    conn = get_conn()
    try:
        cur = conn.cursor()
        sql = (
            "SELECT id, name, platform, phone, profile_dir, url, "
            "extra_json, created_at, updated_at FROM accounts "
        )
        if platform_key == "all":
            cur.execute(sql + "ORDER BY created_at DESC, id DESC LIMIT ?", (int(limit),))
        else:
            cur.execute(
                sql + "WHERE platform = ? ORDER BY created_at DESC, id DESC LIMIT ?",
                (platform_key, int(limit)),
            )
        return cur.fetchall()
    finally:
        conn.close()


def fetch_ids_for_list_json(platform_key: str, lim: int):
    conn = get_conn()
    try:
        cur = conn.cursor()
        if platform_key == "all":
            cur.execute(
                "SELECT id FROM accounts ORDER BY updated_at DESC, id DESC LIMIT ?",
                (lim,),
            )
        else:
            cur.execute(
                "SELECT id FROM accounts WHERE platform = ? "
                "ORDER BY updated_at DESC, id DESC LIMIT ?",
                (platform_key, lim),
            )
        return [r[0] for r in cur.fetchall()]
    finally:
        conn.close()


def fetch_pick_web_candidate_id(platform_key: str) -> Optional[int]:
    """该平台一条账号：按 updated_at、created_at 倒序，与是否登录无关。"""
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute(
            """
            SELECT id FROM accounts
            WHERE platform = ?
            ORDER BY updated_at DESC, created_at DESC, id DESC
            LIMIT 1
            """,
            (platform_key,),
        )
        row = cur.fetchone()
        return int(row[0]) if row else None
    finally:
        conn.close()


def has_duplicate_phone_conn(conn, platform_key: str, phone_store: str) -> bool:
    cur = conn.cursor()
    cur.execute(
        "SELECT id FROM accounts WHERE platform = ? AND phone = ?",
        (platform_key, phone_store),
    )
    return cur.fetchone() is not None


def count_platform_accounts_conn(conn, platform_key: str) -> int:
    cur = conn.cursor()
    cur.execute("SELECT COUNT(*) FROM accounts WHERE platform = ?", (platform_key,))
    return int(cur.fetchone()[0])


def insert_account_row(
    conn,
    name: str,
    platform_key: str,
    phone_store: str,
    profile_dir: str,
    url: str,
    now: int,
) -> int:
    cur = conn.cursor()
    cur.execute(
        """
        INSERT INTO accounts (name, platform, phone, profile_dir, url, extra_json, created_at, updated_at)
        VALUES (?, ?, ?, ?, ?, NULL, ?, ?)
        """,
        (name, platform_key, phone_store, profile_dir, url, now, now),
    )
    return int(cur.lastrowid)


def fetch_delete_row_by_id_conn(conn, aid: int):
    cur = conn.cursor()
    cur.execute(
        "SELECT id, name, platform, phone, profile_dir FROM accounts WHERE id = ?",
        (aid,),
    )
    return cur.fetchone()


def delete_account_by_id_conn(conn, rid: int) -> None:
    cur = conn.cursor()
    cur.execute("DELETE FROM accounts WHERE id = ?", (rid,))


def fetch_ids_profile_for_platform_conn(conn, platform_key: str):
    cur = conn.cursor()
    cur.execute(
        "SELECT id, profile_dir FROM accounts WHERE platform = ?",
        (platform_key,),
    )
    return cur.fetchall()


def delete_accounts_by_platform_conn(conn, platform_key: str) -> None:
    cur = conn.cursor()
    cur.execute("DELETE FROM accounts WHERE platform = ?", (platform_key,))


def fetch_row_platform_phone_conn(conn, platform_key: str, phone_norm: str):
    cur = conn.cursor()
    cur.execute(
        "SELECT id, name, profile_dir FROM accounts WHERE platform = ? AND phone = ?",
        (platform_key, phone_norm),
    )
    return cur.fetchone()


def delete_account_platform_phone_conn(conn, platform_key: str, phone_norm: str) -> None:
    cur = conn.cursor()
    cur.execute(
        "DELETE FROM accounts WHERE platform = ? AND phone = ?",
        (platform_key, phone_norm),
    )