Database

def connect(self) -> sqlite3.Connection:
    con = sqlite3.connect(self.path, timeout=10)
    con.row_factory = sqlite3.Row
    con.execute("PRAGMA foreign_keys=ON")
    return con

def import_registry(self, entries: list[RegistryEntry]) -> None:
    with self._lock, closing(self.connect()) as con, con:
        con.executemany(
            """INSERT OR REPLACE INTO id_registry
            (data_id,name,world,category,source_url,verification_status) VALUES(?,?,?,?,?,?)""",
            [
                (e.data_id, e.name, e.world, e.category, e.source_url, e.verification_status)
                for e in entries
            ],
        )

def lookup(self, data_id: int) -> RegistryEntry | None:
    with closing(self.connect()) as con:
        row = con.execute("SELECT * FROM id_registry WHERE data_id=?", (data_id,)).fetchone()
    return RegistryEntry.model_validate(dict(row)) if row else None

def save_account(self, snapshot: AccountSnapshot) -> None:
    imported = datetime.now(UTC).isoformat()
    with self._lock, closing(self.connect()) as con, con:
        con.execute(
            "INSERT OR REPLACE INTO account_snapshots VALUES(?,?,?)",
            (snapshot.tag, imported, snapshot.raw.model_dump_json()),
        )
        con.execute("DELETE FROM account_entities WHERE tag=?", (snapshot.tag,))
        for entity in snapshot.entities:
            con.execute(
                """INSERT OR REPLACE INTO account_entities
                (tag,section,data_id,level,count,raw_json) VALUES(?,?,?,?,?,?)""",
                (
                    snapshot.tag,
                    entity.section,
                    entity.data_id,
                    entity.level,
                    entity.count,
                    entity.model_dump_json(),
                ),
            )
            if self.lookup(entity.data_id) is None:
                con.execute(
                    """INSERT OR IGNORE INTO unknown_entities
                    (data_id,section,first_seen_at,sample_json,status) VALUES(?,?,?,?,?)""",
                    (
                        entity.data_id,
                        entity.section,
                        imported,
                        entity.model_dump_json(),
                        "UNKNOWN",
                    ),
                )

def account_rows(self, tag: str) -> list[AccountRow]:
    with closing(self.connect()) as con:
        rows = con.execute(
            """SELECT ae.*, ir.name, ir.world, ir.category,
            el.next_level, el.upgrade_cost, el.resource_type, el.upgrade_seconds, el.requirement
            FROM account_entities ae LEFT JOIN id_registry ir ON ir.data_id=ae.data_id
            LEFT JOIN entity_levels el ON el.data_id=ae.data_id AND el.level=ae.level
            WHERE ae.tag=? ORDER BY ae.section, COALESCE(ir.name, ae.data_id)""",
            (tag,),
        ).fetchall()
    return [AccountRow.model_validate(dict(row)) for row in rows]

def add_knowledge(self, emulator_id: str, frame_id: str, statement: str, status: str) -> None:
    with closing(self.connect()) as con, con:
        con.execute(
            "INSERT INTO knowledge(emulator_id,frame_id,statement,status,created_at) VALUES(?,?,?,?,?)",
            (emulator_id, frame_id, statement, status, datetime.now(UTC).isoformat()),
        )

def recent_knowledge(self, limit: int = 50) -> list[KnowledgeItem]:
    with closing(self.connect()) as con:
        rows = con.execute(
            "SELECT * FROM knowledge ORDER BY id DESC LIMIT ?", (limit,)
        ).fetchall()
    return [KnowledgeItem.model_validate(dict(row)) for row in reversed(rows)]

def add_task(self, instruction: str) -> int:
    now = datetime.now(UTC).isoformat()
    with closing(self.connect()) as con, con:
        cursor = con.execute(
            "INSERT INTO tasks(instruction,status,progress,created_at,updated_at) VALUES(?,?,?,?,?)",
            (instruction, "PENDING", "等待執行", now, now),
        )
        if cursor.lastrowid is None:
            raise RuntimeError("Task insert did not return an identifier")
        return int(cursor.lastrowid)

def pending_tasks(self) -> list[TaskRecord]:
    with closing(self.connect()) as con:
        rows = con.execute(
            "SELECT * FROM tasks WHERE status IN ('PENDING','RUNNING') ORDER BY id"
        ).fetchall()
    return [TaskRecord.model_validate(dict(row)) for row in rows]

def update_task(self, task_id: int, status: str, progress: str) -> None:
    now = datetime.now(UTC).isoformat()
    with closing(self.connect()) as con, con:
        con.execute(
            "UPDATE tasks SET status=?,progress=?,updated_at=? WHERE id=?",
            (status, progress, now, task_id),
        )