From 24b1d1025b33684ee2426122eacbd155cdb6b6eb Mon Sep 17 00:00:00 2001 From: stu2not Date: Mon, 20 Apr 2026 20:07:35 +0800 Subject: [PATCH] =?UTF-8?q?=E5=B0=8F=E7=A8=8B=E5=BA=8F=E5=90=8E=E7=AB=AF?= =?UTF-8?q?=E6=8E=A5=E5=85=A5=20talkingq=20=E5=85=B1=E4=BA=AB=E5=BA=93?= =?UTF-8?q?=E5=B9=B6=E6=96=B0=E5=A2=9E=E6=B6=88=E6=81=AF=E4=B8=8E=E5=AE=9A?= =?UTF-8?q?=E4=BD=8D=E8=83=BD=E5=8A=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- database/migrate_to_shared_talkingq.py | 264 ++++++++ database/talkingq_shared_schema.sql | 479 +++++++++++++ mini-program/.env.example | 3 +- mini-program/app/dao/binding.py | 92 ++- mini-program/app/db.py | 65 +- mini-program/app/main.py | 37 +- mini-program/app/middleware/auth.py | 2 + mini-program/app/routers/bindings.py | 30 +- mini-program/app/routers/device_im.py | 305 +++++++++ mini-program/app/routers/device_location.py | 63 ++ mini-program/app/routers/devices.py | 253 +++++++ mini-program/app/routers/im.py | 541 +++++++++++++++ mini-program/app/routers/messages.py | 627 ----------------- mini-program/app/schemas/im.py | 111 +++ mini-program/app/schemas/location.py | 53 ++ mini-program/app/schemas/message.py | 57 -- mini-program/app/service/binding.py | 33 +- mini-program/app/service/im.py | 707 ++++++++++++++++++++ mini-program/app/service/location.py | 327 +++++++++ mini-program/app/settings.py | 11 +- mini-program/database/init.sql | 7 + mini-program/design.md | 164 ++++- 22 files changed, 3469 insertions(+), 762 deletions(-) create mode 100644 database/migrate_to_shared_talkingq.py create mode 100644 database/talkingq_shared_schema.sql create mode 100644 mini-program/app/routers/device_im.py create mode 100644 mini-program/app/routers/device_location.py create mode 100644 mini-program/app/routers/devices.py create mode 100644 mini-program/app/routers/im.py delete mode 100644 mini-program/app/routers/messages.py create mode 100644 mini-program/app/schemas/im.py create mode 100644 mini-program/app/schemas/location.py delete mode 100644 mini-program/app/schemas/message.py create mode 100644 mini-program/app/service/im.py create mode 100644 mini-program/app/service/location.py diff --git a/database/migrate_to_shared_talkingq.py b/database/migrate_to_shared_talkingq.py new file mode 100644 index 0000000..eda5a18 --- /dev/null +++ b/database/migrate_to_shared_talkingq.py @@ -0,0 +1,264 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import argparse +import os +import sys +from pathlib import Path + +import pymysql + + +ROOT_DIR = Path(__file__).resolve().parent +DEFAULT_SCHEMA_PATH = ROOT_DIR / "talkingq_shared_schema.sql" +DEFAULT_ENV_PATH = ROOT_DIR.parent / "mini-program" / ".env" +DEFAULT_TARGET_DB = "talkingq" +DEFAULT_LEGACY_DB = "mini_program" + +TALKINGQ_URL_TABLES = [ + "device_auth", + "device_configs", + "conversation_histories", + "conversation_messages", + "roles", + "role_languages", + "device_firmware_update", + "system_config", +] + +MINI_PROGRAM_TABLES = [ + "parents", + "children", + "parent_child_relations", + "device_bindings", + "device_bind_sessions", + "device_bind_history", + "cards", + "device_settings", + "im_conversations", + "im_messages", + "child_location_current", + "child_location_history", +] + + +def _env_first(*keys: str, default: str) -> str: + for key in keys: + value = os.getenv(key) + if value: + return value + return default + + +def load_env_file(path: Path) -> dict[str, str]: + if not path.exists(): + return {} + + env: dict[str, str] = {} + for raw_line in path.read_text(encoding="utf-8").splitlines(): + line = raw_line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, value = line.split("=", 1) + key = key.strip() + value = value.strip() + if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}: + value = value[1:-1] + env[key] = value + return env + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Apply the shared Banban schema to the single source-of-truth database. " + "Existing talkingq-url data in the target database is preserved; legacy " + "mini_program data is ignored." + ) + ) + parser.add_argument( + "--env-file", + default=str(DEFAULT_ENV_PATH), + help="Optional env file used as fallback for DB_HOST/DB_PORT/DB_USER/DB_PASSWORD.", + ) + parser.add_argument("--host") + parser.add_argument("--port", type=int) + parser.add_argument("--user") + parser.add_argument("--password") + parser.add_argument("--target-db") + parser.add_argument("--legacy-db") + parser.add_argument("--schema", default=str(DEFAULT_SCHEMA_PATH)) + parser.add_argument( + "--drop-legacy-db", + action="store_true", + help="Drop the legacy mini_program database after the shared schema has been applied.", + ) + args = parser.parse_args() + + env_values = load_env_file(Path(args.env_file).expanduser().resolve()) + args.host = args.host or os.getenv("MYSQL_HOST") or os.getenv("DB_HOST") or env_values.get("DB_HOST") or "127.0.0.1" + args.port = ( + args.port + or int(os.getenv("MYSQL_PORT") or os.getenv("DB_PORT") or env_values.get("DB_PORT") or "3306") + ) + args.user = args.user or os.getenv("MYSQL_USER") or os.getenv("DB_USER") or env_values.get("DB_USER") or "root" + args.password = ( + args.password + or os.getenv("MYSQL_PASSWORD") + or os.getenv("DB_PASSWORD") + or env_values.get("DB_PASSWORD") + or "" + ) + args.target_db = args.target_db or DEFAULT_TARGET_DB + args.legacy_db = args.legacy_db or DEFAULT_LEGACY_DB + return args + + +def split_sql_statements(sql: str) -> list[str]: + statements: list[str] = [] + buffer: list[str] = [] + quote_char: str | None = None + escaped = False + + for char in sql: + buffer.append(char) + if quote_char is not None: + if escaped: + escaped = False + elif char == "\\" and quote_char in {"'", '"'}: + escaped = True + elif char == quote_char: + quote_char = None + continue + + if char in {"'", '"', "`"}: + quote_char = char + continue + + if char == ";": + statement = "".join(buffer).strip() + if statement: + statements.append(statement) + buffer = [] + + tail = "".join(buffer).strip() + if tail: + statements.append(tail) + return statements + + +def load_schema_sql(path: Path, *, target_db: str) -> str: + raw_sql = path.read_text(encoding="utf-8-sig") + sql_lines = [] + for line in raw_sql.splitlines(): + stripped = line.lstrip() + if stripped.startswith("--"): + continue + sql_lines.append(line) + sql = "\n".join(sql_lines) + sql = sql.replace("CREATE DATABASE IF NOT EXISTS `talkingq`", f"CREATE DATABASE IF NOT EXISTS `{target_db}`") + sql = sql.replace("USE `talkingq`;", f"USE `{target_db}`;") + return sql + + +def execute_statements(connection: pymysql.Connection, statements: list[str]) -> None: + with connection.cursor() as cursor: + for index, statement in enumerate(statements, start=1): + try: + cursor.execute(statement) + except Exception as exc: # pragma: no cover - operational path + snippet = " ".join(statement.split())[:200] + raise RuntimeError(f"statement #{index} failed: {snippet}") from exc + + +def fetch_table_names(connection: pymysql.Connection, *, database: str) -> set[str]: + with connection.cursor() as cursor: + cursor.execute( + """ + SELECT table_name + FROM information_schema.tables + WHERE table_schema = %s + """, + (database,), + ) + return {row[0] for row in cursor.fetchall()} + + +def database_exists(connection: pymysql.Connection, *, database: str) -> bool: + with connection.cursor() as cursor: + cursor.execute("SHOW DATABASES LIKE %s", (database,)) + return cursor.fetchone() is not None + + +def drop_database(connection: pymysql.Connection, *, database: str) -> None: + with connection.cursor() as cursor: + cursor.execute(f"DROP DATABASE IF EXISTS `{database}`") + + +def print_summary(connection: pymysql.Connection, *, target_db: str) -> None: + tables = fetch_table_names(connection, database=target_db) + expected_tables = set(TALKINGQ_URL_TABLES + MINI_PROGRAM_TABLES) + missing_tables = sorted(expected_tables - tables) + + print(f"target database: {target_db}") + print(f"talkingq-url tables present: {len(set(TALKINGQ_URL_TABLES) & tables)}/{len(TALKINGQ_URL_TABLES)}") + print(f"mini-program tables present: {len(set(MINI_PROGRAM_TABLES) & tables)}/{len(MINI_PROGRAM_TABLES)}") + if missing_tables: + print("missing tables:") + for table_name in missing_tables: + print(f" - {table_name}") + raise SystemExit(1) + + +def main() -> int: + args = parse_args() + schema_path = Path(args.schema).resolve() + if not schema_path.exists(): + print(f"schema file not found: {schema_path}", file=sys.stderr) + return 1 + + if args.drop_legacy_db and args.legacy_db == args.target_db: + print("legacy database must be different from target database", file=sys.stderr) + return 1 + + schema_sql = load_schema_sql(schema_path, target_db=args.target_db) + statements = split_sql_statements(schema_sql) + if not statements: + print("schema file contains no executable statements", file=sys.stderr) + return 1 + + connection = pymysql.connect( + host=args.host, + port=args.port, + user=args.user, + password=args.password, + charset="utf8mb4", + autocommit=True, + ) + try: + print(f"applying shared schema to `{args.target_db}` using `{schema_path}`") + print(f"db connection target: {args.user}@{args.host}:{args.port}") + if args.env_file: + print(f"env fallback file: {Path(args.env_file).expanduser().resolve()}") + print("migration mode: talkingq-url data is preserved; legacy mini_program data is ignored") + execute_statements(connection, statements) + print_summary(connection, target_db=args.target_db) + + if args.drop_legacy_db: + if database_exists(connection, database=args.legacy_db): + drop_database(connection, database=args.legacy_db) + print(f"dropped legacy database: {args.legacy_db}") + else: + print(f"legacy database not found, skip drop: {args.legacy_db}") + else: + print(f"legacy database left untouched: {args.legacy_db}") + finally: + connection.close() + + print("shared database migration completed") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/database/talkingq_shared_schema.sql b/database/talkingq_shared_schema.sql new file mode 100644 index 0000000..d694bf7 --- /dev/null +++ b/database/talkingq_shared_schema.sql @@ -0,0 +1,479 @@ +-- Banban shared schema +-- Sources: +-- 1. talkingq-url/mysql/init/01-init.sql +-- 2. mini-program/database/init.sql +-- Target database: talkingq +-- +-- Notes: +-- 1. talkingq-url base tables are kept as the device-domain foundation. +-- 2. mini-program business tables are merged into the same database. +-- 3. device_auth.device_id is the shared device identity key. +-- 4. Device <-> AI conversation records remain in +-- conversation_histories / conversation_messages. +-- 5. The database owner is shared, but table ownership is explicit: +-- talkingq-url owns device-domain base tables; mini-program owns +-- parent/child/binding/im/location extension tables. +-- 6. During migration to the shared database, talkingq-url data in +-- `talkingq` is the source of truth. Legacy `mini_program` data is +-- not merged into `talkingq`. + +CREATE DATABASE IF NOT EXISTS `talkingq` + DEFAULT CHARACTER SET utf8mb4 + COLLATE utf8mb4_unicode_ci; + +USE `talkingq`; + +-- --------------------------------------------------------------------------- +-- talkingq-url base tables +-- --------------------------------------------------------------------------- + +CREATE TABLE IF NOT EXISTS `device_configs` ( + `id` INT NOT NULL AUTO_INCREMENT, + `device_id` VARCHAR(64) NOT NULL, + `selected_role_key` VARCHAR(64) NOT NULL, + `preferred_language` VARCHAR(10) NULL, + `volume` INT NULL, + `last_update_time` FLOAT NOT NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE INDEX `device_id_UNIQUE` (`device_id` ASC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `conversation_histories` ( + `id` INT NOT NULL AUTO_INCREMENT, + `device_id` VARCHAR(64) NOT NULL, + `role_key` VARCHAR(64) NOT NULL, + `last_interaction_time` FLOAT NOT NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + INDEX `idx_device_id` (`device_id` ASC), + INDEX `idx_role_key` (`role_key` ASC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `conversation_messages` ( + `id` INT NOT NULL AUTO_INCREMENT, + `conversation_id` INT NOT NULL, + `is_user` TINYINT(1) NOT NULL DEFAULT 0, + `content` TEXT NOT NULL, + `timestamp` FLOAT NOT NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + INDEX `idx_conversation_id` (`conversation_id` ASC), + CONSTRAINT `fk_messages_conversation` + FOREIGN KEY (`conversation_id`) + REFERENCES `conversation_histories` (`id`) + ON DELETE CASCADE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `roles` ( + `id` INT NOT NULL AUTO_INCREMENT, + `role_key` VARCHAR(64) NOT NULL, + `name` VARCHAR(128) NOT NULL, + `description` TEXT NULL, + `content` TEXT NOT NULL, + `default_language` VARCHAR(10) NULL, + `asr_provider` VARCHAR(64) NULL, + `llm_provider` VARCHAR(64) NULL, + `tts_provider` VARCHAR(64) NULL, + `competitive_llm_mode` TINYINT(1) NULL, + `volcano_model_id` VARCHAR(64) NULL, + `volcano_voice_type` VARCHAR(64) NULL, + `tencent_voice_type` VARCHAR(64) NULL, + `aliyun_voice_name` VARCHAR(64) NULL, + `minimax_voice_id` VARCHAR(64) NULL, + `url` VARCHAR(255) NULL, + `homophones` JSON NULL, + `enabled` TINYINT(1) NOT NULL DEFAULT 1, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE INDEX `role_key_UNIQUE` (`role_key` ASC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `role_languages` ( + `id` INT NOT NULL AUTO_INCREMENT, + `role_id` INT NOT NULL, + `language_code` VARCHAR(10) NOT NULL, + `name` VARCHAR(128) NULL, + `content` TEXT NULL, + `asr_provider` VARCHAR(64) NULL, + `llm_provider` VARCHAR(64) NULL, + `tts_provider` VARCHAR(64) NULL, + `volcano_voice_type` VARCHAR(64) NULL, + `tencent_voice_type` VARCHAR(64) NULL, + `aliyun_voice_name` VARCHAR(64) NULL, + `minimax_voice_id` VARCHAR(64) NULL, + `url` VARCHAR(255) NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE INDEX `uix_role_language` (`role_id`, `language_code`), + CONSTRAINT `fk_role_languages_role` + FOREIGN KEY (`role_id`) + REFERENCES `roles` (`id`) + ON DELETE CASCADE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `device_auth` ( + `id` INT NOT NULL AUTO_INCREMENT, + `device_id` VARCHAR(64) NOT NULL, + `serial_number` VARCHAR(64) NOT NULL, + `batch_id` VARCHAR(20) NULL, + `is_active` TINYINT(1) NOT NULL DEFAULT 1, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE INDEX `device_id_UNIQUE` (`device_id` ASC), + INDEX `idx_batch_id` (`batch_id` ASC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `device_firmware_update` ( + `id` INT NOT NULL AUTO_INCREMENT, + `device_id` VARCHAR(64) NOT NULL, + `serial_number` VARCHAR(64) NOT NULL, + `mac_address` VARCHAR(512) NULL, + `firmware_version` VARCHAR(64) NOT NULL, + `update_status` VARCHAR(32) NOT NULL DEFAULT 'success', + `progress` FLOAT NULL DEFAULT 0.0, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE INDEX `device_id_UNIQUE` (`device_id` ASC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `system_config` ( + `id` INT NOT NULL AUTO_INCREMENT, + `config_key` VARCHAR(128) NOT NULL, + `config_value` TEXT NULL, + `description` VARCHAR(255) NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE INDEX `config_key_UNIQUE` (`config_key` ASC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +INSERT INTO `device_configs` (`device_id`, `selected_role_key`, `preferred_language`, `last_update_time`) +VALUES ('default', 'assistant', 'zh', UNIX_TIMESTAMP()) +ON DUPLICATE KEY UPDATE `selected_role_key` = `selected_role_key`; + +INSERT INTO `system_config` (`config_key`, `config_value`, `description`) +VALUES + ('latest_firmware_version', '1.0.0', '最新固件版本'), + ('update_firmware_url', 'https://example.com/firmware/latest.bin', '固件更新URL') +ON DUPLICATE KEY UPDATE `config_value` = `config_value`; + +INSERT INTO `roles` (`role_key`, `name`, `description`, `content`, `default_language`, `enabled`) +VALUES ( + 'assistant', + '智能助手', + '默认智能助手角色', + '你是一个友好的智能助手，乐于帮助用户解答问题。', + 'zh', + 1 +) +ON DUPLICATE KEY UPDATE `role_key` = `role_key`; + +-- --------------------------------------------------------------------------- +-- mini-program extension tables +-- --------------------------------------------------------------------------- + +CREATE TABLE IF NOT EXISTS `parents` ( + `user_id` BIGINT NOT NULL AUTO_INCREMENT, + `openid` VARCHAR(64) NOT NULL, + `unionid` VARCHAR(64) NULL, + `nickname` VARCHAR(64) NULL, + `avatar_url` VARCHAR(255) NULL, + `avatar_file_key` VARCHAR(255) NULL, + `phone` VARCHAR(20) NULL, + `status` TINYINT NOT NULL DEFAULT 1, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`user_id`), + UNIQUE KEY `uq_parents_openid` (`openid`), + KEY `idx_parents_unionid` (`unionid`), + KEY `idx_parents_phone` (`phone`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `children` ( + `child_id` BIGINT NOT NULL AUTO_INCREMENT, + `child_name` VARCHAR(32) NOT NULL, + `child_gender` TINYINT NOT NULL DEFAULT 2, + `child_birthday` DATE NULL, + `status` TINYINT NOT NULL DEFAULT 1, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`child_id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `parent_child_relations` ( + `id` BIGINT NOT NULL AUTO_INCREMENT, + `user_id` BIGINT NOT NULL, + `child_id` BIGINT NOT NULL, + `relation_type` TINYINT NOT NULL DEFAULT 9, + `is_primary` TINYINT(1) NOT NULL DEFAULT 0, + `status` TINYINT NOT NULL DEFAULT 1, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + CONSTRAINT `uq_user_child` UNIQUE (`user_id`, `child_id`), + KEY `idx_pcr_user_id` (`user_id`), + KEY `idx_pcr_child_id` (`child_id`), + CONSTRAINT `fk_pcr_user` + FOREIGN KEY (`user_id`) + REFERENCES `parents` (`user_id`) + ON DELETE CASCADE, + CONSTRAINT `fk_pcr_child` + FOREIGN KEY (`child_id`) + REFERENCES `children` (`child_id`) + ON DELETE CASCADE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `device_bindings` ( + `id` BIGINT NOT NULL AUTO_INCREMENT, + `device_id` VARCHAR(64) NOT NULL, + `owner_user_id` BIGINT NOT NULL, + `child_id` BIGINT NULL, + `status` TINYINT NOT NULL DEFAULT 1, + `bound_at` DATETIME NOT NULL, + `unbound_at` DATETIME NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + CONSTRAINT `uq_device_binding_device` UNIQUE (`device_id`), + CONSTRAINT `uq_device_binding_child` UNIQUE (`child_id`), + KEY `idx_device_bindings_owner_user_id` (`owner_user_id`), + CONSTRAINT `fk_device_bindings_device` + FOREIGN KEY (`device_id`) + REFERENCES `device_auth` (`device_id`), + CONSTRAINT `fk_device_bindings_owner` + FOREIGN KEY (`owner_user_id`) + REFERENCES `parents` (`user_id`), + CONSTRAINT `fk_device_bindings_child` + FOREIGN KEY (`child_id`) + REFERENCES `children` (`child_id`) + ON DELETE SET NULL +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `device_bind_sessions` ( + `id` BIGINT NOT NULL AUTO_INCREMENT, + `bind_token` CHAR(36) NOT NULL, + `device_id` VARCHAR(64) NOT NULL, + `initiator_user_id` BIGINT NOT NULL, + `target_child_id` BIGINT NULL, + `challenge_code_hash` CHAR(64) NULL, + `challenge_set_at` DATETIME NULL, + `expires_at` DATETIME NOT NULL, + `max_attempt_count` TINYINT UNSIGNED NOT NULL DEFAULT 5, + `attempt_count` TINYINT UNSIGNED NOT NULL DEFAULT 0, + `status` TINYINT NOT NULL DEFAULT 1, + `confirmed_at` DATETIME NULL, + `consumed_at` DATETIME NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE KEY `uq_device_bind_sessions_token` (`bind_token`), + KEY `idx_device_bind_sessions_device_id` (`device_id`), + KEY `idx_device_bind_sessions_initiator_user_id` (`initiator_user_id`), + KEY `idx_device_bind_sessions_target_child_id` (`target_child_id`), + KEY `idx_device_bind_sessions_expires_at` (`expires_at`), + CONSTRAINT `fk_device_bind_sessions_device` + FOREIGN KEY (`device_id`) + REFERENCES `device_auth` (`device_id`), + CONSTRAINT `fk_device_bind_sessions_initiator` + FOREIGN KEY (`initiator_user_id`) + REFERENCES `parents` (`user_id`), + CONSTRAINT `fk_device_bind_sessions_target_child` + FOREIGN KEY (`target_child_id`) + REFERENCES `children` (`child_id`) + ON DELETE SET NULL +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `device_bind_history` ( + `id` BIGINT NOT NULL AUTO_INCREMENT, + `device_id` VARCHAR(64) NOT NULL, + `child_id` BIGINT NULL, + `bound_by_user_id` BIGINT NOT NULL, + `unbound_by_user_id` BIGINT NULL, + `bind_source` TINYINT NOT NULL DEFAULT 1, + `bound_at` DATETIME NOT NULL, + `unbound_at` DATETIME NULL, + `unbind_reason` VARCHAR(191) NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + KEY `idx_device_bind_history_device_id` (`device_id`), + KEY `idx_device_bind_history_child_id` (`child_id`), + KEY `idx_device_bind_history_bound_by_user_id` (`bound_by_user_id`), + KEY `idx_device_bind_history_unbound_by_user_id` (`unbound_by_user_id`), + KEY `idx_device_bind_history_bound_at` (`bound_at`), + CONSTRAINT `fk_device_bind_history_device` + FOREIGN KEY (`device_id`) + REFERENCES `device_auth` (`device_id`), + CONSTRAINT `fk_device_bind_history_child` + FOREIGN KEY (`child_id`) + REFERENCES `children` (`child_id`) + ON DELETE SET NULL, + CONSTRAINT `fk_device_bind_history_bound_by` + FOREIGN KEY (`bound_by_user_id`) + REFERENCES `parents` (`user_id`), + CONSTRAINT `fk_device_bind_history_unbound_by` + FOREIGN KEY (`unbound_by_user_id`) + REFERENCES `parents` (`user_id`) + ON DELETE SET NULL +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `cards` ( + `card_id` BIGINT NOT NULL AUTO_INCREMENT, + `card_uuid` VARCHAR(64) NOT NULL, + `device_id` VARCHAR(64) NULL, + `card_name` VARCHAR(64) NULL, + `status` TINYINT NOT NULL DEFAULT 0, + `total_swaps` INT NOT NULL DEFAULT 0, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`card_id`), + UNIQUE KEY `uq_cards_card_uuid` (`card_uuid`), + UNIQUE KEY `uq_cards_device_id` (`device_id`), + KEY `idx_cards_status` (`status`), + CONSTRAINT `fk_cards_device` + FOREIGN KEY (`device_id`) + REFERENCES `device_auth` (`device_id`) + ON DELETE SET NULL +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +-- Shared device volume is recommended to use device_configs.volume as source of truth. +-- device_settings.volume is retained for mini-program compatibility and UI storage. +CREATE TABLE IF NOT EXISTS `device_settings` ( + `setting_id` BIGINT NOT NULL AUTO_INCREMENT, + `device_id` VARCHAR(64) NOT NULL, + `sleep_mode` TINYINT NOT NULL DEFAULT 0, + `disable_time_start` TIME NULL, + `disable_time_end` TIME NULL, + `timezone` VARCHAR(32) NOT NULL DEFAULT 'Asia/Shanghai', + `volume` TINYINT UNSIGNED NULL, + `brightness` TINYINT UNSIGNED NULL, + `disable_weekdays` VARCHAR(32) NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`setting_id`), + UNIQUE KEY `uq_device_settings_device_id` (`device_id`), + CONSTRAINT `fk_device_settings_device` + FOREIGN KEY (`device_id`) + REFERENCES `device_auth` (`device_id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `im_conversations` ( + `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, + `conversation_type` TINYINT UNSIGNED NOT NULL, + `participant_a_type` TINYINT UNSIGNED NOT NULL, + `participant_a_id` VARCHAR(64) NOT NULL, + `participant_b_type` TINYINT UNSIGNED NOT NULL, + `participant_b_id` VARCHAR(64) NOT NULL, + `pair_key` VARCHAR(191) NOT NULL, + `status` TINYINT NOT NULL DEFAULT 1, + `last_seq` BIGINT UNSIGNED NOT NULL DEFAULT 0, + `message_count` BIGINT UNSIGNED NOT NULL DEFAULT 0, + `last_message_preview` VARCHAR(255) NULL, + `last_message_at` DATETIME NULL, + `ext_json` JSON NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + CONSTRAINT `uq_conv_type_pair` UNIQUE (`conversation_type`, `pair_key`), + KEY `idx_im_conv_participant_a` (`participant_a_type`, `participant_a_id`), + KEY `idx_im_conv_participant_b` (`participant_b_type`, `participant_b_id`), + KEY `idx_im_conv_last_message_at` (`last_message_at`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `im_messages` ( + `id` BIGINT NOT NULL AUTO_INCREMENT, + `conversation_id` BIGINT UNSIGNED NOT NULL, + `seq` BIGINT UNSIGNED NOT NULL, + `sender_type` TINYINT UNSIGNED NOT NULL, + `sender_id` VARCHAR(64) NOT NULL, + `receiver_type` TINYINT UNSIGNED NOT NULL, + `receiver_id` VARCHAR(64) NOT NULL, + `content_type` TINYINT UNSIGNED NOT NULL, + `content_text` MEDIUMTEXT NULL, + `content_json` JSON NULL, + `media_file_key` VARCHAR(255) NULL, + `media_duration_ms` INT UNSIGNED NULL, + `media_mime_type` VARCHAR(64) NULL, + `media_size_bytes` BIGINT UNSIGNED NULL, + `media_transcript_text` TEXT NULL, + `client_msg_id` VARCHAR(64) NULL, + `sender_name_snapshot` VARCHAR(64) NULL, + `sender_avatar_snapshot` VARCHAR(255) NULL, + `receiver_name_snapshot` VARCHAR(64) NULL, + `receiver_avatar_snapshot` VARCHAR(255) NULL, + `ext_json` JSON NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `deleted_at` DATETIME NULL, + PRIMARY KEY (`id`), + CONSTRAINT `uq_im_msg_conv_seq` UNIQUE (`conversation_id`, `seq`), + CONSTRAINT `uq_im_msg_client` UNIQUE (`conversation_id`, `client_msg_id`), + KEY `idx_im_msg_conversation_created_at` (`conversation_id`, `created_at`), + KEY `idx_im_msg_sender` (`sender_type`, `sender_id`, `created_at`), + KEY `idx_im_msg_receiver` (`receiver_type`, `receiver_id`, `created_at`), + CONSTRAINT `fk_im_messages_conversation` + FOREIGN KEY (`conversation_id`) + REFERENCES `im_conversations` (`id`) + ON DELETE CASCADE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `child_location_current` ( + `child_id` BIGINT NOT NULL, + `device_id` VARCHAR(64) NOT NULL, + `coord_type` VARCHAR(16) NOT NULL DEFAULT 'gcj02', + `lat` DECIMAL(10, 7) NOT NULL, + `lng` DECIMAL(10, 7) NOT NULL, + `accuracy_m` INT UNSIGNED NULL, + `altitude_m` DECIMAL(8, 2) NULL, + `speed_mps` DECIMAL(8, 2) NULL, + `heading_deg` SMALLINT UNSIGNED NULL, + `source` TINYINT UNSIGNED NOT NULL, + `battery_pct` TINYINT UNSIGNED NULL, + `device_time` DATETIME NOT NULL, + `server_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`child_id`), + KEY `idx_child_loc_current_device` (`device_id`), + KEY `idx_child_loc_current_server_time` (`server_time`), + CONSTRAINT `fk_child_loc_current_child` + FOREIGN KEY (`child_id`) + REFERENCES `children` (`child_id`) + ON DELETE CASCADE, + CONSTRAINT `fk_child_loc_current_device` + FOREIGN KEY (`device_id`) + REFERENCES `device_auth` (`device_id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `child_location_history` ( + `id` BIGINT NOT NULL AUTO_INCREMENT, + `child_id` BIGINT NOT NULL, + `device_id` VARCHAR(64) NOT NULL, + `coord_type` VARCHAR(16) NOT NULL DEFAULT 'gcj02', + `lat` DECIMAL(10, 7) NOT NULL, + `lng` DECIMAL(10, 7) NOT NULL, + `accuracy_m` INT UNSIGNED NULL, + `altitude_m` DECIMAL(8, 2) NULL, + `speed_mps` DECIMAL(8, 2) NULL, + `heading_deg` SMALLINT UNSIGNED NULL, + `source` TINYINT UNSIGNED NOT NULL, + `battery_pct` TINYINT UNSIGNED NULL, + `device_time` DATETIME NOT NULL, + `server_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + KEY `idx_child_loc_hist_child` (`child_id`, `created_at`), + KEY `idx_child_loc_hist_device` (`device_id`, `created_at`), + CONSTRAINT `fk_child_loc_hist_child` + FOREIGN KEY (`child_id`) + REFERENCES `children` (`child_id`) + ON DELETE CASCADE, + CONSTRAINT `fk_child_loc_hist_device` + FOREIGN KEY (`device_id`) + REFERENCES `device_auth` (`device_id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; diff --git a/mini-program/.env.example b/mini-program/.env.example index 1023749..65e0473 100644 --- a/mini-program/.env.example +++ b/mini-program/.env.example @@ -9,7 +9,8 @@ DB_HOST=127.0.0.1 DB_PORT=3306 DB_USER=root DB_PASSWORD=change_me -DB_NAME=mini_program +DB_NAME=talkingq +DB_AUTO_INIT_TABLES=false DB_PATH=./data.db WECHAT_APP_ID=wx_change_me WECHAT_APP_SECRET=change_me diff --git a/mini-program/app/dao/binding.py b/mini-program/app/dao/binding.py index 8d8a39d..6e18a17 100644 --- a/mini-program/app/dao/binding.py +++ b/mini-program/app/dao/binding.py @@ -13,6 +13,55 @@ logger = logging.getLogger("app.dao.binding") class BindingDAO(BaseDAO): + def get_device_auth(self, device_id: str) -> Optional[Mapping]: + return ( + self.db.execute( + text( + """ + SELECT device_id, serial_number, is_active + FROM device_auth + WHERE device_id = :device_id + LIMIT 1 + """ + ), + {"device_id": device_id}, + ) + .mappings() + .first() + ) + + def _clear_child_from_binding(self, binding_row: Mapping[str, object]) -> None: + row_id = int(binding_row["id"]) + status = int(binding_row["status"]) + + if status == 1: + self.db.execute( + text( + """ + UPDATE device_bindings + SET child_id = NULL, + status = 1, + unbound_at = NULL, + updated_at = CURRENT_TIMESTAMP + WHERE id = :id + """ + ), + {"id": row_id}, + ) + return + + self.db.execute( + text( + """ + UPDATE device_bindings + SET child_id = NULL, + updated_at = CURRENT_TIMESTAMP + WHERE id = :id + """ + ), + {"id": row_id}, + ) + def _upsert_parent_child_relation(self, user_id: int, child_id: int) -> None: updated = self.db.execute( text( @@ -73,7 +122,7 @@ class BindingDAO(BaseDAO): def _bind_device(self, device_id: str, user_id: int, child_id: Optional[int]) -> None: existing_by_device = ( self.db.execute( - text("SELECT id FROM device_bindings WHERE device_id = :device_id"), + text("SELECT id, status FROM device_bindings WHERE device_id = :device_id"), {"device_id": device_id}, ) .mappings() @@ -110,7 +159,7 @@ class BindingDAO(BaseDAO): existing_by_child = ( self.db.execute( - text("SELECT id FROM device_bindings WHERE child_id = :child_id"), + text("SELECT id, status FROM device_bindings WHERE child_id = :child_id"), {"child_id": child_id}, ) .mappings() @@ -120,20 +169,8 @@ class BindingDAO(BaseDAO): if existing_by_device: device_row_id = int(existing_by_device["id"]) if existing_by_child and int(existing_by_child["id"]) != device_row_id: - # Release child ownership from a different row before assigning to this device. - self.db.execute( - text( - """ - UPDATE device_bindings - SET child_id = NULL, - status = 0, - unbound_at = CURRENT_TIMESTAMP, - updated_at = CURRENT_TIMESTAMP - WHERE id = :id - """ - ), - {"id": existing_by_child["id"]}, - ) + # Keep the old device row visible; only release its child assignment. + self._clear_child_from_binding(existing_by_child) self.db.execute( text( @@ -146,27 +183,14 @@ class BindingDAO(BaseDAO): updated_at = CURRENT_TIMESTAMP WHERE device_id = :device_id """ - ), - {"owner_user_id": user_id, "child_id": child_id, "device_id": device_id}, - ) + ), + {"owner_user_id": user_id, "child_id": child_id, "device_id": device_id}, + ) return if existing_by_child: - self.db.execute( - text( - """ - UPDATE device_bindings - SET device_id = :device_id, - owner_user_id = :owner_user_id, - status = 1, - unbound_at = NULL, - updated_at = CURRENT_TIMESTAMP - WHERE id = :id - """ - ), - {"device_id": device_id, "owner_user_id": user_id, "id": existing_by_child["id"]}, - ) - return + # Do not overwrite the previous device row; keep it as an unbound device. + self._clear_child_from_binding(existing_by_child) self.db.execute( text( diff --git a/mini-program/app/db.py b/mini-program/app/db.py index 007a0d1..82f540e 100644 --- a/mini-program/app/db.py +++ b/mini-program/app/db.py @@ -2,7 +2,7 @@ from collections.abc import Generator import logging import time -from sqlalchemy import create_engine, event, text +from sqlalchemy import bindparam, create_engine, event, text from sqlalchemy.orm import Session, sessionmaker try: @@ -43,6 +43,29 @@ def set_sqlite_pragma(dbapi_conn, connection_record): logger = logging.getLogger("app.db") +SHARED_TALKINGQ_REQUIRED_TABLES = ( + "device_auth", + "device_configs", + "conversation_histories", + "conversation_messages", + "roles", + "role_languages", + "device_firmware_update", + "system_config", + "parents", + "children", + "parent_child_relations", + "device_bindings", + "device_bind_sessions", + "device_bind_history", + "cards", + "device_settings", + "im_conversations", + "im_messages", + "child_location_current", + "child_location_history", +) + @event.listens_for(engine, "before_cursor_execute") def before_cursor_execute( @@ -112,6 +135,46 @@ def check_db_connection() -> None: logger.info("database connection check succeeded", extra={"event": "db_check"}) +def check_shared_schema_tables() -> None: + if not settings.uses_shared_talkingq_db: + return + + query = text( + """ + SELECT table_name + FROM information_schema.tables + WHERE table_schema = :table_schema + AND table_name IN :table_names + """ + ).bindparams(bindparam("table_names", expanding=True)) + + with engine.connect() as conn: + rows = conn.execute( + query, + { + "table_schema": settings.db_name, + "table_names": list(SHARED_TALKINGQ_REQUIRED_TABLES), + }, + ) + existing_tables = {row[0] for row in rows} + + missing_tables = sorted(set(SHARED_TALKINGQ_REQUIRED_TABLES) - existing_tables) + if missing_tables: + raise RuntimeError( + "shared talkingq schema is incomplete; missing tables: " + + ", ".join(missing_tables) + ) + + logger.info( + "shared talkingq schema check succeeded", + extra={ + "event": "db_shared_schema_check", + "db_name": settings.db_name, + "table_count": len(SHARED_TALKINGQ_REQUIRED_TABLES), + }, + ) + + def close_db_engine() -> None: engine.dispose() logger.info("database engine disposed", extra={"event": "db_close"}) diff --git a/mini-program/app/main.py b/mini-program/app/main.py index 0cc8ab1..2b59702 100644 --- a/mini-program/app/main.py +++ b/mini-program/app/main.py @@ -12,17 +12,25 @@ if __package__ in (None, ""): if project_root_str not in sys.path: sys.path.insert(0, project_root_str) -from app.db import check_db_connection, close_db_engine, init_db_tables +from app.db import ( + check_db_connection, + check_shared_schema_tables, + close_db_engine, + init_db_tables, +) from app.logging_setup import configure_logging from app.middleware.auth import install_auth_middleware from app.middleware.request_log import install_request_logging_middleware from app.routers.auth import router as auth_router from app.routers.wechat_auth import router as wechat_auth_router from app.routers.health import router as health_router -from app.routers.messages import router as messages_router from app.routers.parents import router as parents_router from app.routers.children import router as children_router from app.routers.bindings import router as bindings_router +from app.routers.devices import router as devices_router +from app.routers.device_im import router as device_im_router +from app.routers.device_location import router as device_location_router +from app.routers.im import router as im_router from app.settings import settings @@ -40,7 +48,25 @@ def create_app() -> FastAPI: @app.on_event("startup") def on_startup() -> None: check_db_connection() - init_db_tables() + if settings.db_auto_init_tables: + init_db_tables() + if settings.uses_shared_talkingq_db: + logger.warning( + "database auto init only creates mini-program tables; shared deployments should use the shared schema migration script", + extra={ + "event": "db_shared_auto_init_warning", + "db_name": settings.db_name, + }, + ) + else: + check_shared_schema_tables() + logger.info( + "database auto init disabled", + extra={ + "event": "db_init_skipped", + "db_name": settings.db_name, + }, + ) logger.info("application started", extra={"event": "app_start"}) @app.on_event("shutdown") @@ -52,10 +78,13 @@ def create_app() -> FastAPI: install_request_logging_middleware(app) app.include_router(wechat_auth_router) app.include_router(health_router) - app.include_router(messages_router) app.include_router(parents_router) app.include_router(children_router) app.include_router(bindings_router) + app.include_router(devices_router) + app.include_router(device_im_router) + app.include_router(device_location_router) + app.include_router(im_router) return app diff --git a/mini-program/app/middleware/auth.py b/mini-program/app/middleware/auth.py index 278f323..f4749df 100644 --- a/mini-program/app/middleware/auth.py +++ b/mini-program/app/middleware/auth.py @@ -20,6 +20,8 @@ NO_AUTH_PATH_PREFIXES = ( "/redoc", "/openapi.json", "/auth/login", + "/device-im", + "/device-location", ) diff --git a/mini-program/app/routers/bindings.py b/mini-program/app/routers/bindings.py index 7bf2115..1961d2f 100644 --- a/mini-program/app/routers/bindings.py +++ b/mini-program/app/routers/bindings.py @@ -6,11 +6,11 @@ from pydantic import BaseModel try: from app.security import get_current_user_id - from app.service.binding import BindingService + from app.service.binding import BindingError, BindingService from app.service import get_db_session except ModuleNotFoundError: from security import get_current_user_id - from service.binding import BindingService + from service.binding import BindingError, BindingService from service import get_db_session @@ -20,6 +20,7 @@ logger = logging.getLogger("app.bindings") class BindStartRequest(BaseModel): device_id: str + serial_number: str child_id: int | None = None @@ -80,7 +81,15 @@ def start_bind( db=Depends(get_db_session), ) -> BindStartResponse: service = BindingService(db) - bind_token, expires_at = service.start_bind(current_user_id, payload.device_id, payload.child_id) + try: + bind_token, expires_at = service.start_bind( + current_user_id, + payload.device_id, + payload.serial_number, + payload.child_id, + ) + except BindingError as e: + raise HTTPException(status_code=e.status_code, detail=str(e)) return BindStartResponse(bind_token=bind_token, expires_at=expires_at.isoformat()) @@ -94,6 +103,8 @@ def confirm_bind( service = BindingService(db) try: result = service.confirm_bind(payload.bind_token, current_user_id) + except BindingError as e: + raise HTTPException(status_code=e.status_code, detail=str(e)) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) return BindConfirmResponse(**result) @@ -101,6 +112,7 @@ def confirm_bind( class DirectBindRequest(BaseModel): device_id: str + serial_number: str child_id: int | None = None @@ -121,7 +133,15 @@ def direct_bind( db=Depends(get_db_session), ) -> DirectBindResponse: service = BindingService(db) - result = service.direct_bind(payload.device_id, payload.child_id, current_user_id) + try: + result = service.direct_bind( + payload.device_id, + payload.serial_number, + payload.child_id, + current_user_id, + ) + except BindingError as e: + raise HTTPException(status_code=e.status_code, detail=str(e)) return DirectBindResponse(**result) @@ -136,6 +156,8 @@ def set_binding_child( service = BindingService(db) try: result = service.set_binding_child(device_id=device_id, child_id=payload.child_id, user_id=current_user_id) + except BindingError as e: + raise HTTPException(status_code=e.status_code, detail=str(e)) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) return DirectBindResponse(**result) diff --git a/mini-program/app/routers/device_im.py b/mini-program/app/routers/device_im.py new file mode 100644 index 0000000..0ba7dfb --- /dev/null +++ b/mini-program/app/routers/device_im.py @@ -0,0 +1,305 @@ +import logging +from typing import Any + +from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, Response, status +from sqlalchemy import text +from sqlalchemy.orm import Session + +try: + from app.db import get_db + from app.routers.im import ( + CHILD_PARTICIPANT_TYPE, + PARENT_PARTICIPANT_TYPE, + SUPPORTED_CONVERSATION_TYPES, + _fetch_child_names, + _fetch_parent_names, + _get_conversation_for_child, + _row_to_conversation_item, + _row_to_message_item, + ) + from app.schemas.im import ( + ChildConversationListResponse, + ChildConversationMessageListResponse, + ConversationMessageCreateResponse, + DeviceMessageCreateRequest, + ) + from app.service.im import authenticate_device_identity, create_device_message +except ModuleNotFoundError: + from db import get_db + from routers.im import ( + CHILD_PARTICIPANT_TYPE, + PARENT_PARTICIPANT_TYPE, + SUPPORTED_CONVERSATION_TYPES, + _fetch_child_names, + _fetch_parent_names, + _get_conversation_for_child, + _row_to_conversation_item, + _row_to_message_item, + ) + from schemas.im import ( + ChildConversationListResponse, + ChildConversationMessageListResponse, + ConversationMessageCreateResponse, + DeviceMessageCreateRequest, + ) + from service.im import authenticate_device_identity, create_device_message + + +router = APIRouter(prefix="/device-im", tags=["device-im"]) +logger = logging.getLogger("app.device_im") + + +@router.get("/{device_id}/conversations", response_model=ChildConversationListResponse) +def list_device_conversations( + device_id: str, + request: Request, + device_serial: str = Header(alias="X-Device-Serial", min_length=1), + conversation_type: int | None = Query(default=None), + cursor: int | None = Query(default=None, ge=1), + limit: int = Query(default=20, ge=1, le=100), + db: Session = Depends(get_db), +) -> ChildConversationListResponse: + if conversation_type is not None and conversation_type not in SUPPORTED_CONVERSATION_TYPES: + raise HTTPException(status_code=422, detail="unsupported conversation_type") + + device_identity = authenticate_device_identity( + db=db, + device_id=device_id, + serial_number=device_serial, + ) + child_id = device_identity.child_id + + params: dict[str, Any] = { + "child_id_str": str(child_id), + "child_participant_type": CHILD_PARTICIPANT_TYPE, + "fetch_limit": limit + 1, + } + where = """ + status = 1 + AND conversation_type IN (1, 2) + AND ( + (participant_a_type = :child_participant_type AND participant_a_id = :child_id_str) + OR (participant_b_type = :child_participant_type AND participant_b_id = :child_id_str) + ) + """ + if conversation_type is not None: + where += " AND conversation_type = :conversation_type" + params["conversation_type"] = conversation_type + if cursor is not None: + where += " AND id < :cursor" + params["cursor"] = cursor + + rows = ( + db.execute( + text( + f""" + SELECT + id, + conversation_type, + participant_a_type, + participant_a_id, + participant_b_type, + participant_b_id, + last_message_preview, + last_message_at, + message_count, + created_at + FROM im_conversations + WHERE {where} + ORDER BY COALESCE(last_message_at, created_at) DESC, id DESC + LIMIT :fetch_limit + """ + ), + params, + ) + .mappings() + .all() + ) + + has_more = len(rows) > limit + rows = rows[:limit] + next_cursor = int(rows[-1]["id"]) if has_more and rows else None + + peer_parent_ids: set[int] = set() + peer_child_ids: set[int] = set() + child_id_str = str(child_id) + for row in rows: + participant_a_type = int(row["participant_a_type"]) + participant_b_type = int(row["participant_b_type"]) + participant_a_id = str(row["participant_a_id"]) + participant_b_id = str(row["participant_b_id"]) + if participant_a_type == CHILD_PARTICIPANT_TYPE and participant_a_id == child_id_str: + peer_type = participant_b_type + peer_id = participant_b_id + else: + peer_type = participant_a_type + peer_id = participant_a_id + + if peer_id.isdigit(): + if peer_type == PARENT_PARTICIPANT_TYPE: + peer_parent_ids.add(int(peer_id)) + elif peer_type == CHILD_PARTICIPANT_TYPE: + peer_child_ids.add(int(peer_id)) + + parent_names = _fetch_parent_names(db, peer_parent_ids) + child_names = _fetch_child_names(db, peer_child_ids) + items = [ + _row_to_conversation_item( + row, + child_id=child_id, + parent_names=parent_names, + child_names=child_names, + ) + for row in rows + ] + + logger.info( + "device conversations listed", + extra={ + "event": "device_conversation_list", + "request_id": getattr(request.state, "request_id", None), + "device_id": device_id, + "child_id": child_id, + "conversation_type": conversation_type, + "count": len(items), + }, + ) + + return ChildConversationListResponse( + items=items, + total=len(items), + next_cursor=next_cursor, + ) + + +@router.get( + "/{device_id}/conversations/{conversation_id}/messages", + response_model=ChildConversationMessageListResponse, +) +def list_device_conversation_messages( + device_id: str, + conversation_id: int, + request: Request, + device_serial: str = Header(alias="X-Device-Serial", min_length=1), + cursor_seq: int | None = Query(default=None, ge=1), + limit: int = Query(default=20, ge=1, le=100), + db: Session = Depends(get_db), +) -> ChildConversationMessageListResponse: + device_identity = authenticate_device_identity( + db=db, + device_id=device_id, + serial_number=device_serial, + ) + _get_conversation_for_child( + db=db, + conversation_id=conversation_id, + child_id=device_identity.child_id, + ) + + sql = """ + SELECT + id, + conversation_id, + seq, + sender_type, + sender_id, + receiver_type, + receiver_id, + content_type, + content_text, + content_json, + media_file_key, + media_duration_ms, + media_mime_type, + media_size_bytes, + media_transcript_text, + client_msg_id, + sender_name_snapshot, + sender_avatar_snapshot, + receiver_name_snapshot, + receiver_avatar_snapshot, + ext_json, + created_at + FROM im_messages + WHERE conversation_id = :conversation_id + AND deleted_at IS NULL + """ + params: dict[str, Any] = { + "conversation_id": conversation_id, + "fetch_limit": limit + 1, + } + if cursor_seq is not None: + sql += " AND seq < :cursor_seq" + params["cursor_seq"] = cursor_seq + sql += " ORDER BY seq DESC LIMIT :fetch_limit" + + rows = db.execute(text(sql), params).mappings().all() + has_more = len(rows) > limit + rows = rows[:limit] + rows.reverse() + items = [_row_to_message_item(row) for row in rows] + next_cursor_seq = items[0].seq if has_more and items else None + + logger.info( + "device conversation messages listed", + extra={ + "event": "device_conversation_message_list", + "request_id": getattr(request.state, "request_id", None), + "device_id": device_id, + "child_id": device_identity.child_id, + "conversation_id": conversation_id, + "count": len(items), + "has_more": has_more, + }, + ) + + return ChildConversationMessageListResponse( + conversation_id=conversation_id, + has_more=has_more, + next_cursor_seq=next_cursor_seq, + items=items, + ) + + +@router.post( + "/{device_id}/messages", + response_model=ConversationMessageCreateResponse, + status_code=status.HTTP_201_CREATED, +) +def create_message_from_device( + device_id: str, + payload: DeviceMessageCreateRequest, + request: Request, + response: Response, + device_serial: str = Header(alias="X-Device-Serial", min_length=1), + db: Session = Depends(get_db), +) -> ConversationMessageCreateResponse: + device_identity, result = create_device_message( + db=db, + device_id=device_id, + serial_number=device_serial, + payload=payload, + ) + if result.idempotent: + response.status_code = status.HTTP_200_OK + + logger.info( + "device message created", + extra={ + "event": "device_message_create", + "request_id": getattr(request.state, "request_id", None), + "device_id": device_id, + "child_id": device_identity.child_id, + "conversation_id": result.conversation_id, + "conversation_type": result.conversation_type, + "idempotent": result.idempotent, + }, + ) + + return ConversationMessageCreateResponse( + idempotent=result.idempotent, + conversation_id=result.conversation_id, + conversation_type=result.conversation_type, + conversation_type_name=result.conversation_type_name, + message=result.message, + ) diff --git a/mini-program/app/routers/device_location.py b/mini-program/app/routers/device_location.py new file mode 100644 index 0000000..eafed1c --- /dev/null +++ b/mini-program/app/routers/device_location.py @@ -0,0 +1,63 @@ +import logging + +from fastapi import APIRouter, Depends, Header, Request +from sqlalchemy.orm import Session + +try: + from app.db import get_db + from app.schemas.location import DeviceLocationReportRequest, DeviceLocationReportResponse + from app.service.location import report_device_location +except ModuleNotFoundError: + from db import get_db + from schemas.location import DeviceLocationReportRequest, DeviceLocationReportResponse + from service.location import report_device_location + + +router = APIRouter(prefix="/device-location", tags=["device-location"]) +logger = logging.getLogger("app.device_location") + + +@router.post("/{device_id}/reports", response_model=DeviceLocationReportResponse) +def create_device_location_report( + device_id: str, + payload: DeviceLocationReportRequest, + request: Request, + device_serial: str = Header(alias="X-Device-Serial", min_length=1), + db: Session = Depends(get_db), +) -> DeviceLocationReportResponse: + device_identity, row = report_device_location( + db=db, + device_id=device_id, + serial_number=device_serial, + payload=payload, + ) + + logger.info( + "device location reported", + extra={ + "event": "device_location_report", + "request_id": getattr(request.state, "request_id", None), + "device_id": device_id, + "child_id": device_identity.child_id, + "lat": float(row["lat"]), + "lng": float(row["lng"]), + }, + ) + + return DeviceLocationReportResponse( + child_id=int(row["child_id"]), + child_name=device_identity.child_name, + device_id=str(row["device_id"]), + coord_type=str(row["coord_type"]), + lat=float(row["lat"]), + lng=float(row["lng"]), + accuracy_m=row["accuracy_m"], + altitude_m=float(row["altitude_m"]) if row["altitude_m"] is not None else None, + speed_mps=float(row["speed_mps"]) if row["speed_mps"] is not None else None, + heading_deg=row["heading_deg"], + source=int(row["source"]), + battery_pct=row["battery_pct"], + device_time=row["device_time"], + server_time=row["server_time"], + updated_at=row["updated_at"], + ) diff --git a/mini-program/app/routers/devices.py b/mini-program/app/routers/devices.py new file mode 100644 index 0000000..4d40402 --- /dev/null +++ b/mini-program/app/routers/devices.py @@ -0,0 +1,253 @@ +import logging +from collections.abc import Mapping +from datetime import datetime + +from fastapi import APIRouter, Depends, HTTPException, Query, Request +from pydantic import BaseModel +from sqlalchemy import text + +try: + from app.security import get_current_user_id + from app.service import get_db_session + from app.schemas.location import ( + DeviceLocationCurrentResponse, + DeviceLocationTrajectoryItem, + DeviceLocationTrajectoryResponse, + ) + from app.service.location import get_device_current_location, get_device_trajectory +except ModuleNotFoundError: + from security import get_current_user_id + from service import get_db_session + from schemas.location import ( + DeviceLocationCurrentResponse, + DeviceLocationTrajectoryItem, + DeviceLocationTrajectoryResponse, + ) + from service.location import get_device_current_location, get_device_trajectory + + +router = APIRouter(prefix="/devices", tags=["devices"]) +logger = logging.getLogger("app.devices") + + +class DeviceMessageItem(BaseModel): + id: int + conversation_id: int + role_key: str + is_user: bool + speaker: str + content: str + timestamp: float + created_at: datetime + + +class DeviceMessageListResponse(BaseModel): + items: list[DeviceMessageItem] + total: int + next_cursor: int | None = None + + +def _ensure_device_access(db, device_id: str, user_id: int) -> None: + row = ( + db.execute( + text( + """ + SELECT 1 + FROM device_bindings + WHERE device_id = :device_id + AND owner_user_id = :user_id + AND status = 1 + LIMIT 1 + """ + ), + {"device_id": device_id, "user_id": user_id}, + ) + .mappings() + .first() + ) + if row is None: + raise HTTPException(status_code=404, detail="device not found") + + +def _row_to_message_item(row: Mapping) -> DeviceMessageItem: + is_user = bool(row["is_user"]) + return DeviceMessageItem( + id=int(row["id"]), + conversation_id=int(row["conversation_id"]), + role_key=str(row["role_key"]), + is_user=is_user, + speaker="user" if is_user else "assistant", + content=str(row["content"]), + timestamp=float(row["timestamp"]), + created_at=row["created_at"], + ) + + +def _row_to_current_location_response(row: Mapping) -> DeviceLocationCurrentResponse: + return DeviceLocationCurrentResponse( + child_id=int(row["child_id"]), + child_name=row.get("child_name"), + device_id=str(row["device_id"]), + coord_type=str(row["coord_type"]), + lat=float(row["lat"]), + lng=float(row["lng"]), + accuracy_m=row["accuracy_m"], + altitude_m=float(row["altitude_m"]) if row["altitude_m"] is not None else None, + speed_mps=float(row["speed_mps"]) if row["speed_mps"] is not None else None, + heading_deg=row["heading_deg"], + source=int(row["source"]), + battery_pct=row["battery_pct"], + device_time=row["device_time"], + server_time=row["server_time"], + updated_at=row["updated_at"], + ) + + +def _row_to_trajectory_item(row: Mapping, *, child_name: str | None) -> DeviceLocationTrajectoryItem: + return DeviceLocationTrajectoryItem( + id=int(row["id"]), + child_id=int(row["child_id"]), + child_name=child_name, + device_id=str(row["device_id"]), + coord_type=str(row["coord_type"]), + lat=float(row["lat"]), + lng=float(row["lng"]), + accuracy_m=row["accuracy_m"], + altitude_m=float(row["altitude_m"]) if row["altitude_m"] is not None else None, + speed_mps=float(row["speed_mps"]) if row["speed_mps"] is not None else None, + heading_deg=row["heading_deg"], + source=int(row["source"]), + battery_pct=row["battery_pct"], + device_time=row["device_time"], + server_time=row["server_time"], + created_at=row["created_at"], + ) + + +@router.get("/{device_id}/messages", response_model=DeviceMessageListResponse) +def list_device_messages( + device_id: str, + request: Request, + cursor: int | None = Query(default=None, ge=1), + limit: int = Query(default=20, ge=1, le=100), + current_user_id: int = Depends(get_current_user_id), + db=Depends(get_db_session), +) -> DeviceMessageListResponse: + _ensure_device_access(db=db, device_id=device_id, user_id=current_user_id) + + params = {"device_id": device_id, "limit": limit + 1} + where = "ch.device_id = :device_id" + if cursor is not None: + where += " AND cm.id < :cursor" + params["cursor"] = cursor + + rows = ( + db.execute( + text( + f""" + SELECT + cm.id, + ch.id AS conversation_id, + ch.role_key, + cm.is_user, + cm.content, + cm.timestamp, + cm.created_at + FROM conversation_messages AS cm + JOIN conversation_histories AS ch + ON ch.id = cm.conversation_id + WHERE {where} + ORDER BY cm.id DESC + LIMIT :limit + """ + ), + params, + ) + .mappings() + .all() + ) + + has_more = len(rows) > limit + rows = rows[:limit] + next_cursor = int(rows[-1]["id"]) if has_more and rows else None + + logger.info( + "listed device ai messages", + extra={ + "event": "device_messages", + "request_id": getattr(request.state, "request_id", None), + "user_id": current_user_id, + "device_id": device_id, + "returned_count": len(rows), + }, + ) + + return DeviceMessageListResponse( + items=[_row_to_message_item(row) for row in rows], + total=len(rows), + next_cursor=next_cursor, + ) + + +@router.get("/{device_id}/location", response_model=DeviceLocationCurrentResponse) +def get_current_device_location( + device_id: str, + request: Request, + current_user_id: int = Depends(get_current_user_id), + db=Depends(get_db_session), +) -> DeviceLocationCurrentResponse: + row = get_device_current_location(db=db, device_id=device_id, user_id=current_user_id) + + logger.info( + "device current location fetched", + extra={ + "event": "device_current_location", + "request_id": getattr(request.state, "request_id", None), + "user_id": current_user_id, + "device_id": device_id, + "child_id": int(row["child_id"]), + }, + ) + return _row_to_current_location_response(row) + + +@router.get("/{device_id}/trajectory", response_model=DeviceLocationTrajectoryResponse) +def get_device_location_trajectory( + device_id: str, + request: Request, + start_at: datetime | None = Query(default=None), + end_at: datetime | None = Query(default=None), + limit: int = Query(default=200, ge=1, le=1000), + current_user_id: int = Depends(get_current_user_id), + db=Depends(get_db_session), +) -> DeviceLocationTrajectoryResponse: + if start_at and end_at and start_at > end_at: + raise HTTPException(status_code=422, detail="start_at must be earlier than end_at") + + access, rows = get_device_trajectory( + db=db, + device_id=device_id, + user_id=current_user_id, + start_at=start_at, + end_at=end_at, + limit=limit, + ) + + logger.info( + "device trajectory fetched", + extra={ + "event": "device_trajectory", + "request_id": getattr(request.state, "request_id", None), + "user_id": current_user_id, + "device_id": device_id, + "child_id": access.child_id, + "count": len(rows), + }, + ) + + return DeviceLocationTrajectoryResponse( + items=[_row_to_trajectory_item(row, child_name=access.child_name) for row in rows], + total=len(rows), + start_at=start_at, + end_at=end_at, + ) diff --git a/mini-program/app/routers/im.py b/mini-program/app/routers/im.py new file mode 100644 index 0000000..0b7b534 --- /dev/null +++ b/mini-program/app/routers/im.py @@ -0,0 +1,541 @@ +import json +import logging +from collections.abc import Mapping +from typing import Any + +from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status +from sqlalchemy import text +from sqlalchemy.orm import Session + +try: + from app.db import get_db + from app.security import get_current_user_id + from app.schemas.im import ( + ChildConversationItem, + ChildConversationListResponse, + ChildConversationMessageItem, + ChildConversationMessageListResponse, + ConversationMessageCreateResponse, + ParentChildMessageCreateRequest, + ) + from app.service.im import create_parent_child_message +except ModuleNotFoundError: + from db import get_db + from security import get_current_user_id + from schemas.im import ( + ChildConversationItem, + ChildConversationListResponse, + ChildConversationMessageItem, + ChildConversationMessageListResponse, + ConversationMessageCreateResponse, + ParentChildMessageCreateRequest, + ) + from service.im import create_parent_child_message + + +router = APIRouter(prefix="/children", tags=["im"]) +logger = logging.getLogger("app.im") + +PARENT_PARTICIPANT_TYPE = 1 +CHILD_PARTICIPANT_TYPE = 2 + +CHILD_PEER_CONVERSATION_TYPE = 1 +PARENT_CHILD_CONVERSATION_TYPE = 2 +SUPPORTED_CONVERSATION_TYPES = { + CHILD_PEER_CONVERSATION_TYPE, + PARENT_CHILD_CONVERSATION_TYPE, +} + +CONVERSATION_TYPE_NAMES = { + CHILD_PEER_CONVERSATION_TYPE: "child_peer", + PARENT_CHILD_CONVERSATION_TYPE: "parent_child", +} + +PARTICIPANT_TYPE_NAMES = { + PARENT_PARTICIPANT_TYPE: "parent", + CHILD_PARTICIPANT_TYPE: "child", +} + + +def _normalize_content_json(value: Any) -> dict[str, Any] | None: + if value is None: + return None + if isinstance(value, dict): + return value + if isinstance(value, str): + try: + parsed = json.loads(value) + if isinstance(parsed, dict): + return parsed + except json.JSONDecodeError: + return None + return None + + +def _participant_type_name(participant_type: int) -> str: + return PARTICIPANT_TYPE_NAMES.get(participant_type, f"unknown_{participant_type}") + + +def _conversation_type_name(conversation_type: int) -> str: + return CONVERSATION_TYPE_NAMES.get(conversation_type, f"unknown_{conversation_type}") + + +def _build_in_params(prefix: str, values: list[int]) -> tuple[str, dict[str, int]]: + placeholders: list[str] = [] + params: dict[str, int] = {} + for index, value in enumerate(values): + key = f"{prefix}_{index}" + placeholders.append(f":{key}") + params[key] = value + return ", ".join(placeholders), params + + +def _fetch_parent_names(db: Session, user_ids: set[int]) -> dict[int, str | None]: + if not user_ids: + return {} + + values = sorted(user_ids) + placeholders, params = _build_in_params("user_id", values) + rows = ( + db.execute( + text( + f""" + SELECT user_id, nickname + FROM parents + WHERE status = 1 + AND user_id IN ({placeholders}) + """ + ), + params, + ) + .mappings() + .all() + ) + return {int(row["user_id"]): row["nickname"] for row in rows} + + +def _fetch_child_names(db: Session, child_ids: set[int]) -> dict[int, str | None]: + if not child_ids: + return {} + + values = sorted(child_ids) + placeholders, params = _build_in_params("child_id", values) + rows = ( + db.execute( + text( + f""" + SELECT child_id, child_name + FROM children + WHERE status = 1 + AND child_id IN ({placeholders}) + """ + ), + params, + ) + .mappings() + .all() + ) + return {int(row["child_id"]): row["child_name"] for row in rows} + + +def _assert_child_access(db: Session, child_id: int, user_id: int) -> None: + child_row = ( + db.execute( + text( + """ + SELECT child_id + FROM children + WHERE child_id = :child_id + AND status = 1 + LIMIT 1 + """ + ), + {"child_id": child_id}, + ) + .mappings() + .first() + ) + if not child_row: + raise HTTPException(status_code=404, detail="child not found") + + has_access = ( + db.execute( + text( + """ + SELECT 1 + FROM parent_child_relations + WHERE user_id = :user_id + AND child_id = :child_id + AND status = 1 + LIMIT 1 + """ + ), + {"user_id": user_id, "child_id": child_id}, + ).scalar_one_or_none() + is not None + ) + if not has_access: + raise HTTPException(status_code=403, detail="no permission to access this child") + + +def _get_conversation_for_child( + db: Session, + *, + conversation_id: int, + child_id: int, +) -> Mapping[str, Any]: + child_id_str = str(child_id) + row = ( + db.execute( + text( + """ + SELECT + id, + conversation_type, + participant_a_type, + participant_a_id, + participant_b_type, + participant_b_id, + status + FROM im_conversations + WHERE id = :conversation_id + LIMIT 1 + """ + ), + {"conversation_id": conversation_id}, + ) + .mappings() + .first() + ) + if not row: + raise HTTPException(status_code=404, detail="conversation not found") + + conversation_type = int(row["conversation_type"]) + if conversation_type not in SUPPORTED_CONVERSATION_TYPES or int(row["status"]) != 1: + raise HTTPException(status_code=404, detail="conversation not found") + + is_child_participant = ( + ( + int(row["participant_a_type"]) == CHILD_PARTICIPANT_TYPE + and row["participant_a_id"] == child_id_str + ) + or ( + int(row["participant_b_type"]) == CHILD_PARTICIPANT_TYPE + and row["participant_b_id"] == child_id_str + ) + ) + if not is_child_participant: + raise HTTPException(status_code=404, detail="conversation not found") + + return row + + +def _row_to_conversation_item( + row: Mapping[str, Any], + *, + child_id: int, + parent_names: Mapping[int, str | None], + child_names: Mapping[int, str | None], +) -> ChildConversationItem: + child_id_str = str(child_id) + participant_a_type = int(row["participant_a_type"]) + participant_b_type = int(row["participant_b_type"]) + participant_a_id = str(row["participant_a_id"]) + participant_b_id = str(row["participant_b_id"]) + + if participant_a_type == CHILD_PARTICIPANT_TYPE and participant_a_id == child_id_str: + peer_type = participant_b_type + peer_id = participant_b_id + else: + peer_type = participant_a_type + peer_id = participant_a_id + + peer_name: str | None = None + if peer_type == PARENT_PARTICIPANT_TYPE and peer_id.isdigit(): + peer_name = parent_names.get(int(peer_id)) + elif peer_type == CHILD_PARTICIPANT_TYPE and peer_id.isdigit(): + peer_name = child_names.get(int(peer_id)) + + return ChildConversationItem( + conversation_id=int(row["id"]), + conversation_type=int(row["conversation_type"]), + conversation_type_name=_conversation_type_name(int(row["conversation_type"])), + peer_type=_participant_type_name(peer_type), + peer_id=peer_id, + peer_name=peer_name, + last_message_preview=row["last_message_preview"], + last_message_at=row["last_message_at"], + message_count=int(row["message_count"]), + ) + + +def _row_to_message_item(row: Mapping[str, Any]) -> ChildConversationMessageItem: + return ChildConversationMessageItem( + id=int(row["id"]), + conversation_id=int(row["conversation_id"]), + seq=int(row["seq"]), + sender_type=_participant_type_name(int(row["sender_type"])), + sender_id=str(row["sender_id"]), + receiver_type=_participant_type_name(int(row["receiver_type"])), + receiver_id=str(row["receiver_id"]), + content_type=int(row["content_type"]), + content_text=row["content_text"], + content_json=_normalize_content_json(row["content_json"]), + media_file_key=row["media_file_key"], + media_duration_ms=row["media_duration_ms"], + media_mime_type=row.get("media_mime_type"), + media_size_bytes=row.get("media_size_bytes"), + media_transcript_text=row.get("media_transcript_text"), + client_msg_id=row["client_msg_id"], + sender_name_snapshot=row["sender_name_snapshot"], + sender_avatar_snapshot=row.get("sender_avatar_snapshot"), + receiver_name_snapshot=row["receiver_name_snapshot"], + receiver_avatar_snapshot=row.get("receiver_avatar_snapshot"), + ext_json=_normalize_content_json(row.get("ext_json")), + created_at=row["created_at"], + ) + + +@router.get("/{child_id}/conversations", response_model=ChildConversationListResponse) +def list_child_conversations( + child_id: int, + request: Request, + conversation_type: int | None = Query(default=None), + cursor: int | None = Query(default=None, ge=1), + limit: int = Query(default=20, ge=1, le=100), + current_user_id: int = Depends(get_current_user_id), + db: Session = Depends(get_db), +) -> ChildConversationListResponse: + if conversation_type is not None and conversation_type not in SUPPORTED_CONVERSATION_TYPES: + raise HTTPException(status_code=422, detail="unsupported conversation_type") + + _assert_child_access(db=db, child_id=child_id, user_id=current_user_id) + + params: dict[str, Any] = { + "child_id_str": str(child_id), + "child_participant_type": CHILD_PARTICIPANT_TYPE, + "child_peer_conversation_type": CHILD_PEER_CONVERSATION_TYPE, + "parent_child_conversation_type": PARENT_CHILD_CONVERSATION_TYPE, + "fetch_limit": limit + 1, + } + where = """ + status = 1 + AND conversation_type IN (:child_peer_conversation_type, :parent_child_conversation_type) + AND ( + (participant_a_type = :child_participant_type AND participant_a_id = :child_id_str) + OR (participant_b_type = :child_participant_type AND participant_b_id = :child_id_str) + ) + """ + if conversation_type is not None: + where += " AND conversation_type = :conversation_type" + params["conversation_type"] = conversation_type + if cursor is not None: + where += " AND id < :cursor" + params["cursor"] = cursor + + rows = ( + db.execute( + text( + f""" + SELECT + id, + conversation_type, + participant_a_type, + participant_a_id, + participant_b_type, + participant_b_id, + last_message_preview, + last_message_at, + message_count, + created_at + FROM im_conversations + WHERE {where} + ORDER BY COALESCE(last_message_at, created_at) DESC, id DESC + LIMIT :fetch_limit + """ + ), + params, + ) + .mappings() + .all() + ) + + has_more = len(rows) > limit + rows = rows[:limit] + next_cursor = int(rows[-1]["id"]) if has_more and rows else None + + peer_parent_ids: set[int] = set() + peer_child_ids: set[int] = set() + child_id_str = str(child_id) + for row in rows: + participant_a_type = int(row["participant_a_type"]) + participant_b_type = int(row["participant_b_type"]) + participant_a_id = str(row["participant_a_id"]) + participant_b_id = str(row["participant_b_id"]) + if participant_a_type == CHILD_PARTICIPANT_TYPE and participant_a_id == child_id_str: + peer_type = participant_b_type + peer_id = participant_b_id + else: + peer_type = participant_a_type + peer_id = participant_a_id + + if peer_id.isdigit(): + if peer_type == PARENT_PARTICIPANT_TYPE: + peer_parent_ids.add(int(peer_id)) + elif peer_type == CHILD_PARTICIPANT_TYPE: + peer_child_ids.add(int(peer_id)) + + parent_names = _fetch_parent_names(db, peer_parent_ids) + child_names = _fetch_child_names(db, peer_child_ids) + + items = [ + _row_to_conversation_item( + row, + child_id=child_id, + parent_names=parent_names, + child_names=child_names, + ) + for row in rows + ] + + logger.info( + "child conversations listed", + extra={ + "event": "child_conversation_list", + "request_id": getattr(request.state, "request_id", None), + "user_id": current_user_id, + "child_id": child_id, + "conversation_type": conversation_type, + "count": len(items), + }, + ) + + return ChildConversationListResponse( + items=items, + total=len(items), + next_cursor=next_cursor, + ) + + +@router.post( + "/{child_id}/messages", + response_model=ConversationMessageCreateResponse, + status_code=status.HTTP_201_CREATED, +) +def create_child_message_for_parent( + child_id: int, + payload: ParentChildMessageCreateRequest, + request: Request, + response: Response, + current_user_id: int = Depends(get_current_user_id), + db: Session = Depends(get_db), +) -> ConversationMessageCreateResponse: + result = create_parent_child_message( + db=db, + parent_user_id=current_user_id, + child_id=child_id, + payload=payload, + ) + if result.idempotent: + response.status_code = status.HTTP_200_OK + + logger.info( + "parent child message created", + extra={ + "event": "parent_child_message_create", + "request_id": getattr(request.state, "request_id", None), + "user_id": current_user_id, + "child_id": child_id, + "conversation_id": result.conversation_id, + "idempotent": result.idempotent, + }, + ) + + return ConversationMessageCreateResponse( + idempotent=result.idempotent, + conversation_id=result.conversation_id, + conversation_type=result.conversation_type, + conversation_type_name=result.conversation_type_name, + message=result.message, + ) + + +@router.get( + "/{child_id}/conversations/{conversation_id}/messages", + response_model=ChildConversationMessageListResponse, +) +def list_child_conversation_messages( + child_id: int, + conversation_id: int, + request: Request, + cursor_seq: int | None = Query(default=None, ge=1), + limit: int = Query(default=20, ge=1, le=100), + current_user_id: int = Depends(get_current_user_id), + db: Session = Depends(get_db), +) -> ChildConversationMessageListResponse: + _assert_child_access(db=db, child_id=child_id, user_id=current_user_id) + _get_conversation_for_child(db=db, conversation_id=conversation_id, child_id=child_id) + + sql = """ + SELECT + id, + conversation_id, + seq, + sender_type, + sender_id, + receiver_type, + receiver_id, + content_type, + content_text, + content_json, + media_file_key, + media_duration_ms, + media_mime_type, + media_size_bytes, + media_transcript_text, + client_msg_id, + sender_name_snapshot, + sender_avatar_snapshot, + receiver_name_snapshot, + receiver_avatar_snapshot, + ext_json, + created_at + FROM im_messages + WHERE conversation_id = :conversation_id + AND deleted_at IS NULL + """ + params: dict[str, Any] = { + "conversation_id": conversation_id, + "fetch_limit": limit + 1, + } + if cursor_seq is not None: + sql += " AND seq < :cursor_seq" + params["cursor_seq"] = cursor_seq + sql += " ORDER BY seq DESC LIMIT :fetch_limit" + + rows = db.execute(text(sql), params).mappings().all() + has_more = len(rows) > limit + rows = rows[:limit] + rows.reverse() + items = [_row_to_message_item(row) for row in rows] + next_cursor_seq = items[0].seq if has_more and items else None + + logger.info( + "child conversation messages listed", + extra={ + "event": "child_conversation_message_list", + "request_id": getattr(request.state, "request_id", None), + "user_id": current_user_id, + "child_id": child_id, + "conversation_id": conversation_id, + "count": len(items), + "has_more": has_more, + }, + ) + + return ChildConversationMessageListResponse( + conversation_id=conversation_id, + has_more=has_more, + next_cursor_seq=next_cursor_seq, + items=items, + ) diff --git a/mini-program/app/routers/messages.py b/mini-program/app/routers/messages.py deleted file mode 100644 index 865bfc8..0000000 --- a/mini-program/app/routers/messages.py +++ /dev/null @@ -1,627 +0,0 @@ -import json -import logging -from collections.abc import Mapping -from typing import Any - -from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status -from sqlalchemy import text -from sqlalchemy.exc import IntegrityError -from sqlalchemy.orm import Session - -try: - # For module mode: `uvicorn app.main:app` - from app.db import get_db - from app.db_compat import ( - current_timestamp_sql, - get_db_dialect_name, - inserted_primary_key, - select_for_update_clause, - ) - from app.security import get_current_user_id - from app.schemas.message import ( - MessageCreateRequest, - MessageCreateResponse, - MessageItem, - MessageListResponse, - ) -except ModuleNotFoundError: - # For script mode: `python app/main.py` or VS Code "Run Python File" - from db import get_db - from db_compat import ( - current_timestamp_sql, - get_db_dialect_name, - inserted_primary_key, - select_for_update_clause, - ) - from security import get_current_user_id - from schemas.message import ( - MessageCreateRequest, - MessageCreateResponse, - MessageItem, - MessageListResponse, - ) - - -router = APIRouter(prefix="/messages", tags=["messages"]) -logger = logging.getLogger("app.messages") - -PARENT_PARTICIPANT_TYPE = 1 -PARENT_DIRECT_CONVERSATION_TYPE = 3 - - -def _build_preview(content_type: int, content_text: str | None) -> str: - if content_type == 1: - preview = f"user: {(content_text or '').strip()}" - elif content_type == 2: - preview = "user: [audio]" - elif content_type == 3: - preview = "user: [image]" - else: - preview = "user: [json]" - return preview[:255] - - -def _normalize_content_json(value: Any) -> dict[str, Any] | None: - if value is None: - return None - if isinstance(value, dict): - return value - if isinstance(value, str): - try: - parsed = json.loads(value) - if isinstance(parsed, dict): - return parsed - except json.JSONDecodeError: - return None - return None - - -def _row_to_message_item(row: Mapping[str, Any]) -> MessageItem: - sender_user_id = row.get("sender_user_id") - if sender_user_id is not None: - sender_user_id = int(sender_user_id) - - return MessageItem( - id=int(row["id"]), - conversation_id=int(row["conversation_id"]), - seq=int(row["seq"]), - sender_user_id=sender_user_id, - role=int(row.get("role", 1)), - content_type=int(row["content_type"]), - content_text=row["content_text"], - content_json=_normalize_content_json(row["content_json"]), - media_file_key=row["media_file_key"], - media_duration_ms=row["media_duration_ms"], - client_msg_id=row["client_msg_id"], - created_at=row["created_at"], - ) - - -def _get_existing_message( - db: Session, conversation_id: int, client_msg_id: str -) -> Mapping[str, Any] | None: - return ( - db.execute( - text( - """ - SELECT - id, - conversation_id, - seq, - CASE - WHEN sender_type = :parent_participant_type THEN sender_id - ELSE NULL - END AS sender_user_id, - 1 AS role, - content_type, - content_text, - content_json, - media_file_key, - media_duration_ms, - client_msg_id, - created_at - FROM im_messages - WHERE conversation_id = :conversation_id - AND client_msg_id = :client_msg_id - LIMIT 1 - """ - ), - { - "conversation_id": conversation_id, - "client_msg_id": client_msg_id, - "parent_participant_type": PARENT_PARTICIPANT_TYPE, - }, - ) - .mappings() - .first() - ) - - -def _build_parent_direct_pair(user_a_id: int, user_b_id: int) -> tuple[str, str, str]: - low_id, high_id = sorted((user_a_id, user_b_id)) - participant_a_id = str(low_id) - participant_b_id = str(high_id) - return participant_a_id, participant_b_id, f"{participant_a_id}:{participant_b_id}" - - -def _next_primary_key(db: Session, table_name: str) -> int: - if table_name not in {"im_conversations", "im_messages"}: - raise ValueError(f"unsupported table name: {table_name}") - return int( - db.execute(text(f"SELECT COALESCE(MAX(id), 0) + 1 FROM {table_name}")).scalar_one() - ) - - -def _get_direct_conversation( - db: Session, - user_low_id: int, - user_high_id: int, -) -> Mapping[str, Any] | None: - _, _, pair_key = _build_parent_direct_pair(user_low_id, user_high_id) - return ( - db.execute( - text( - """ - SELECT id, last_seq, status - FROM im_conversations - WHERE conversation_type = :conversation_type - AND pair_key = :pair_key - LIMIT 1 - """ - ), - { - "conversation_type": PARENT_DIRECT_CONVERSATION_TYPE, - "pair_key": pair_key, - }, - ) - .mappings() - .first() - ) - - -def _get_or_create_direct_conversation( - db: Session, - user_low_id: int, - user_high_id: int, -) -> int: - existing = _get_direct_conversation( - db=db, - user_low_id=user_low_id, - user_high_id=user_high_id, - ) - if existing: - return int(existing["id"]) - - now_sql = current_timestamp_sql(db) - participant_a_id, participant_b_id, pair_key = _build_parent_direct_pair(user_low_id, user_high_id) - conversation_id = None - if get_db_dialect_name(db) == "sqlite": - conversation_id = _next_primary_key(db, "im_conversations") - - try: - insert_sql = f""" - INSERT INTO im_conversations ( - {'id,' if conversation_id is not None else ''} - conversation_type, - participant_a_type, - participant_a_id, - participant_b_type, - participant_b_id, - pair_key, - status, - last_seq, - message_count, - created_at, - updated_at - ) - VALUES ( - {':id,' if conversation_id is not None else ''} - :conversation_type, - :participant_a_type, - :participant_a_id, - :participant_b_type, - :participant_b_id, - :pair_key, - 1, - 0, - 0, - {now_sql}, - {now_sql} - ) - """ - params = { - "conversation_type": PARENT_DIRECT_CONVERSATION_TYPE, - "participant_a_type": PARENT_PARTICIPANT_TYPE, - "participant_a_id": participant_a_id, - "participant_b_type": PARENT_PARTICIPANT_TYPE, - "participant_b_id": participant_b_id, - "pair_key": pair_key, - } - if conversation_id is not None: - params["id"] = conversation_id - - result = db.execute( - text( - insert_sql - ), - params, - ) - if conversation_id is not None: - return conversation_id - return inserted_primary_key(result) - except IntegrityError: - existing = _get_direct_conversation( - db=db, - user_low_id=user_low_id, - user_high_id=user_high_id, - ) - if existing: - return int(existing["id"]) - raise - - -def _get_active_parent_profiles( - db: Session, - *, - sender_user_id: int, - peer_user_id: int, -) -> dict[int, Mapping[str, Any]]: - rows = ( - db.execute( - text( - """ - SELECT user_id, nickname, avatar_url - FROM parents - WHERE user_id IN (:sender_user_id, :peer_user_id) - AND status = 1 - """ - ), - {"sender_user_id": sender_user_id, "peer_user_id": peer_user_id}, - ) - .mappings() - .all() - ) - return {int(row["user_id"]): row for row in rows} - - -def _assert_conversation_access( - db: Session, - conversation_id: int, - current_user_id: int, -) -> None: - conversation_row = ( - db.execute( - text( - """ - SELECT - id, - participant_a_type, - participant_a_id, - participant_b_type, - participant_b_id - FROM im_conversations - WHERE id = :conversation_id - LIMIT 1 - """ - ), - {"conversation_id": conversation_id}, - ) - .mappings() - .first() - ) - if not conversation_row: - raise HTTPException(status_code=404, detail="conversation not found") - - current_user_id_str = str(current_user_id) - is_participant = ( - ( - int(conversation_row["participant_a_type"]) == PARENT_PARTICIPANT_TYPE - and conversation_row["participant_a_id"] == current_user_id_str - ) - or ( - int(conversation_row["participant_b_type"]) == PARENT_PARTICIPANT_TYPE - and conversation_row["participant_b_id"] == current_user_id_str - ) - ) - if not is_participant: - raise HTTPException(status_code=403, detail="no permission for this conversation") - - -@router.post("", response_model=MessageCreateResponse, status_code=status.HTTP_201_CREATED) -def create_message( - payload: MessageCreateRequest, - request: Request, - response: Response, - current_user_id: int = Depends(get_current_user_id), - db: Session = Depends(get_db), -) -> MessageCreateResponse: - sender_user_id = current_user_id - peer_user_id = payload.peer_user_id - if sender_user_id == peer_user_id: - raise HTTPException(status_code=422, detail="peer_user_id cannot be same as current user") - - user_low_id = min(sender_user_id, peer_user_id) - user_high_id = max(sender_user_id, peer_user_id) - now_sql = current_timestamp_sql(db) - - with db.begin(): - parents = _get_active_parent_profiles( - db, - sender_user_id=sender_user_id, - peer_user_id=peer_user_id, - ) - if len(parents) != 2: - raise HTTPException(status_code=404, detail="sender or peer user not found") - - conversation_id = _get_or_create_direct_conversation( - db=db, - user_low_id=user_low_id, - user_high_id=user_high_id, - ) - - conversation = ( - db.execute( - text( - f""" - SELECT id, last_seq, status - FROM im_conversations - WHERE id = :conversation_id{select_for_update_clause(db)} - """ - ), - {"conversation_id": conversation_id}, - ) - .mappings() - .first() - ) - if not conversation: - raise HTTPException(status_code=500, detail="failed to load conversation") - - if int(conversation["status"]) != 1: - raise HTTPException(status_code=409, detail="conversation is not active") - - existing = _get_existing_message(db, conversation_id, payload.client_msg_id) - if existing: - logger.info( - "message idempotent hit", - extra={ - "event": "message_create", - "request_id": getattr(request.state, "request_id", None), - "user_id": sender_user_id, - "conversation_id": conversation_id, - "message_id": int(existing["id"]), - "seq": int(existing["seq"]), - "client_msg_id": payload.client_msg_id, - "idempotent": True, - }, - ) - response.status_code = status.HTTP_200_OK - return MessageCreateResponse( - idempotent=True, - message=_row_to_message_item(existing), - ) - - next_seq = int(conversation["last_seq"]) + 1 - preview = _build_preview(payload.content_type, payload.content_text) - message_id = None - if get_db_dialect_name(db) == "sqlite": - message_id = _next_primary_key(db, "im_messages") - - insert_sql = f""" - INSERT INTO im_messages ( - {'id,' if message_id is not None else ''} - conversation_id, - seq, - sender_type, - sender_id, - receiver_type, - receiver_id, - content_type, - content_text, - content_json, - media_file_key, - media_duration_ms, - client_msg_id, - sender_name_snapshot, - sender_avatar_snapshot, - receiver_name_snapshot, - receiver_avatar_snapshot, - created_at - ) - VALUES ( - {':id,' if message_id is not None else ''} - :conversation_id, - :seq, - :sender_type, - :sender_id, - :receiver_type, - :receiver_id, - :content_type, - :content_text, - :content_json, - :media_file_key, - :media_duration_ms, - :client_msg_id, - :sender_name_snapshot, - :sender_avatar_snapshot, - :receiver_name_snapshot, - :receiver_avatar_snapshot, - {now_sql} - ) - """ - insert_params = { - "conversation_id": conversation_id, - "seq": next_seq, - "sender_type": PARENT_PARTICIPANT_TYPE, - "sender_id": str(sender_user_id), - "receiver_type": PARENT_PARTICIPANT_TYPE, - "receiver_id": str(peer_user_id), - "content_type": payload.content_type, - "content_text": payload.content_text, - "content_json": json.dumps(payload.content_json, ensure_ascii=False) - if payload.content_json is not None - else None, - "media_file_key": payload.media_file_key, - "media_duration_ms": payload.media_duration_ms, - "client_msg_id": payload.client_msg_id, - "sender_name_snapshot": parents[sender_user_id]["nickname"], - "sender_avatar_snapshot": parents[sender_user_id]["avatar_url"], - "receiver_name_snapshot": parents[peer_user_id]["nickname"], - "receiver_avatar_snapshot": parents[peer_user_id]["avatar_url"], - } - if message_id is not None: - insert_params["id"] = message_id - - insert_result = db.execute( - text( - insert_sql - ), - insert_params, - ) - - db.execute( - text( - f""" - UPDATE im_conversations - SET - last_seq = :last_seq, - message_count = message_count + 1, - last_message_preview = :last_message_preview, - last_message_at = {now_sql}, - updated_at = {now_sql} - WHERE id = :conversation_id - """ - ), - { - "conversation_id": conversation_id, - "last_seq": next_seq, - "last_message_preview": preview, - }, - ) - - created = ( - db.execute( - text( - """ - SELECT - id, - conversation_id, - seq, - CASE - WHEN sender_type = :parent_participant_type THEN sender_id - ELSE NULL - END AS sender_user_id, - 1 AS role, - content_type, - content_text, - content_json, - media_file_key, - media_duration_ms, - client_msg_id, - created_at - FROM im_messages - WHERE id = :message_id - LIMIT 1 - """ - ), - { - "message_id": message_id if message_id is not None else inserted_primary_key(insert_result), - "parent_participant_type": PARENT_PARTICIPANT_TYPE, - }, - ) - .mappings() - .first() - ) - if not created: - raise HTTPException(status_code=500, detail="failed to load created message") - - logger.info( - "message created", - extra={ - "event": "message_create", - "request_id": getattr(request.state, "request_id", None), - "user_id": sender_user_id, - "conversation_id": conversation_id, - "message_id": int(created["id"]), - "seq": int(created["seq"]), - "client_msg_id": payload.client_msg_id, - "idempotent": False, - }, - ) - return MessageCreateResponse( - idempotent=False, - message=_row_to_message_item(created), - ) - - -@router.get("", response_model=MessageListResponse) -def list_messages( - request: Request, - conversation_id: int = Query(gt=0), - cursor_seq: int | None = Query(default=None, ge=1), - limit: int = Query(default=20, ge=1, le=100), - current_user_id: int = Depends(get_current_user_id), - db: Session = Depends(get_db), -) -> MessageListResponse: - _assert_conversation_access( - db=db, - conversation_id=conversation_id, - current_user_id=current_user_id, - ) - - sql = """ - SELECT - id, - conversation_id, - seq, - CASE - WHEN sender_type = :parent_participant_type THEN sender_id - ELSE NULL - END AS sender_user_id, - 1 AS role, - content_type, - content_text, - content_json, - media_file_key, - media_duration_ms, - client_msg_id, - created_at - FROM im_messages - WHERE conversation_id = :conversation_id - AND deleted_at IS NULL - """ - params: dict[str, Any] = { - "conversation_id": conversation_id, - "fetch_limit": limit + 1, - "parent_participant_type": PARENT_PARTICIPANT_TYPE, - } - if cursor_seq is not None: - sql += " AND seq < :cursor_seq" - params["cursor_seq"] = cursor_seq - sql += " ORDER BY seq DESC LIMIT :fetch_limit" - - rows = db.execute(text(sql), params).mappings().all() - has_more = len(rows) > limit - rows = rows[:limit] - rows.reverse() - items = [_row_to_message_item(row) for row in rows] - - next_cursor_seq = items[0].seq if has_more and items else None - logger.info( - "messages listed", - extra={ - "event": "message_list", - "request_id": getattr(request.state, "request_id", None), - "user_id": current_user_id, - "conversation_id": conversation_id, - "cursor_seq": cursor_seq, - "limit": limit, - "count": len(items), - "has_more": has_more, - }, - ) - return MessageListResponse( - conversation_id=conversation_id, - has_more=has_more, - next_cursor_seq=next_cursor_seq, - items=items, - ) diff --git a/mini-program/app/schemas/im.py b/mini-program/app/schemas/im.py new file mode 100644 index 0000000..cdab21d --- /dev/null +++ b/mini-program/app/schemas/im.py @@ -0,0 +1,111 @@ +from datetime import datetime +from typing import Any + +from pydantic import BaseModel, Field, model_validator + + +class ChildConversationItem(BaseModel): + conversation_id: int + conversation_type: int + conversation_type_name: str + peer_type: str + peer_id: str + peer_name: str | None = None + last_message_preview: str | None = None + last_message_at: datetime | None = None + message_count: int + + +class ChildConversationListResponse(BaseModel): + items: list[ChildConversationItem] + total: int + next_cursor: int | None = None + + +class ChildConversationMessageItem(BaseModel): + id: int + conversation_id: int + seq: int + sender_type: str + sender_id: str + receiver_type: str + receiver_id: str + content_type: int + content_text: str | None = None + content_json: dict[str, Any] | None = None + media_file_key: str | None = None + media_duration_ms: int | None = None + media_mime_type: str | None = None + media_size_bytes: int | None = None + media_transcript_text: str | None = None + client_msg_id: str | None = None + sender_name_snapshot: str | None = None + sender_avatar_snapshot: str | None = None + receiver_name_snapshot: str | None = None + receiver_avatar_snapshot: str | None = None + ext_json: dict[str, Any] | None = None + created_at: datetime + + +class ChildConversationMessageListResponse(BaseModel): + conversation_id: int + has_more: bool + next_cursor_seq: int | None = None + items: list[ChildConversationMessageItem] + + +class BaseConversationMessageCreateRequest(BaseModel): + content_type: int = Field(ge=1, le=4) + content_text: str | None = None + content_json: dict[str, Any] | None = None + media_file_key: str | None = None + media_duration_ms: int | None = Field(default=None, ge=0) + media_mime_type: str | None = None + media_size_bytes: int | None = Field(default=None, ge=0) + media_transcript_text: str | None = None + client_msg_id: str = Field(min_length=1, max_length=64) + ext_json: dict[str, Any] | None = None + + @model_validator(mode="after") + def validate_message_payload(self) -> "BaseConversationMessageCreateRequest": + if self.content_type == 1 and not self.content_text: + raise ValueError("content_text is required when content_type=1") + if self.content_type == 2 and not self.media_file_key: + raise ValueError("media_file_key is required when content_type=2") + if self.content_type == 3 and not self.media_file_key: + raise ValueError("media_file_key is required when content_type=3") + if self.content_type == 4 and self.content_json is None: + raise ValueError("content_json is required when content_type=4") + return self + + +class ParentChildMessageCreateRequest(BaseConversationMessageCreateRequest): + pass + + +class DeviceMessageCreateRequest(BaseConversationMessageCreateRequest): + conversation_type: int = Field(ge=1, le=2) + peer_child_id: int | None = Field(default=None, ge=1) + parent_user_id: int | None = Field(default=None, ge=1) + + @model_validator(mode="after") + def validate_target(self) -> "DeviceMessageCreateRequest": + if self.conversation_type == 1: + if self.peer_child_id is None: + raise ValueError("peer_child_id is required when conversation_type=1") + if self.parent_user_id is not None: + raise ValueError("parent_user_id must be empty when conversation_type=1") + if self.conversation_type == 2: + if self.parent_user_id is None: + raise ValueError("parent_user_id is required when conversation_type=2") + if self.peer_child_id is not None: + raise ValueError("peer_child_id must be empty when conversation_type=2") + return self + + +class ConversationMessageCreateResponse(BaseModel): + idempotent: bool + conversation_id: int + conversation_type: int + conversation_type_name: str + message: ChildConversationMessageItem diff --git a/mini-program/app/schemas/location.py b/mini-program/app/schemas/location.py new file mode 100644 index 0000000..bbb412c --- /dev/null +++ b/mini-program/app/schemas/location.py @@ -0,0 +1,53 @@ +from datetime import datetime + +from pydantic import BaseModel, Field + + +class DeviceLocationReportRequest(BaseModel): + coord_type: str = Field(default="gcj02", min_length=1, max_length=16) + lat: float = Field(ge=-90, le=90) + lng: float = Field(ge=-180, le=180) + accuracy_m: int | None = Field(default=None, ge=0) + altitude_m: float | None = None + speed_mps: float | None = None + heading_deg: int | None = Field(default=None, ge=0, le=360) + source: int = Field(ge=0) + battery_pct: int | None = Field(default=None, ge=0, le=100) + device_time: datetime + + +class DeviceLocationPoint(BaseModel): + child_id: int + child_name: str | None = None + device_id: str + coord_type: str + lat: float + lng: float + accuracy_m: int | None = None + altitude_m: float | None = None + speed_mps: float | None = None + heading_deg: int | None = None + source: int + battery_pct: int | None = None + device_time: datetime + server_time: datetime + + +class DeviceLocationCurrentResponse(DeviceLocationPoint): + updated_at: datetime + + +class DeviceLocationReportResponse(DeviceLocationCurrentResponse): + pass + + +class DeviceLocationTrajectoryItem(DeviceLocationPoint): + id: int + created_at: datetime + + +class DeviceLocationTrajectoryResponse(BaseModel): + items: list[DeviceLocationTrajectoryItem] + total: int + start_at: datetime | None = None + end_at: datetime | None = None diff --git a/mini-program/app/schemas/message.py b/mini-program/app/schemas/message.py deleted file mode 100644 index c76e7d3..0000000 --- a/mini-program/app/schemas/message.py +++ /dev/null @@ -1,57 +0,0 @@ -from datetime import datetime -from typing import Any - -from pydantic import BaseModel, Field, model_validator - - -class MessageCreateRequest(BaseModel): - peer_user_id: int = Field(gt=0) - client_msg_id: str = Field(min_length=1, max_length=64) - content_type: int = Field(description="1-text 2-audio 3-image 4-json") - content_text: str | None = Field(default=None) - content_json: dict[str, Any] | None = None - media_file_key: str | None = Field(default=None, max_length=255) - media_duration_ms: int | None = Field(default=None, ge=0) - - @model_validator(mode="after") - def validate_message_fields(self) -> "MessageCreateRequest": - if self.content_type not in {1, 2, 3, 4}: - raise ValueError("content_type must be one of 1, 2, 3, 4") - - if self.content_type == 1 and not (self.content_text and self.content_text.strip()): - raise ValueError("content_text is required when content_type is 1") - - if self.content_type in {2, 3} and not self.media_file_key: - raise ValueError("media_file_key is required when content_type is 2 or 3") - - if self.content_type == 4 and self.content_json is None: - raise ValueError("content_json is required when content_type is 4") - - return self - - -class MessageItem(BaseModel): - id: int - conversation_id: int - seq: int - sender_user_id: int | None - role: int - content_type: int - content_text: str | None - content_json: dict[str, Any] | None - media_file_key: str | None - media_duration_ms: int | None - client_msg_id: str | None - created_at: datetime - - -class MessageCreateResponse(BaseModel): - idempotent: bool - message: MessageItem - - -class MessageListResponse(BaseModel): - conversation_id: int - has_more: bool - next_cursor_seq: int | None - items: list[MessageItem] diff --git a/mini-program/app/service/binding.py b/mini-program/app/service/binding.py index da01612..3c23f79 100644 --- a/mini-program/app/service/binding.py +++ b/mini-program/app/service/binding.py @@ -5,11 +5,33 @@ from typing import Optional from app.dao.binding import BindingDAO +class BindingError(ValueError): + def __init__(self, message: str, status_code: int = 400) -> None: + super().__init__(message) + self.status_code = status_code + + class BindingService: def __init__(self, db): self.dao = BindingDAO(db) - def start_bind(self, user_id: int, device_id: str, child_id: int | None = None) -> tuple[str, datetime]: + def _ensure_bindable_device(self, device_id: str, serial_number: str) -> None: + row = self.dao.get_device_auth(device_id) + if row is None: + raise BindingError("device not found in device_auth", status_code=404) + if str(row["serial_number"]) != serial_number: + raise BindingError("serial_number does not match device_id", status_code=400) + if int(row["is_active"]) != 1: + raise BindingError("device is inactive", status_code=400) + + def start_bind( + self, + user_id: int, + device_id: str, + serial_number: str, + child_id: int | None = None, + ) -> tuple[str, datetime]: + self._ensure_bindable_device(device_id, serial_number) bind_token = self.dao.start_bind(user_id, device_id, child_id) return bind_token, datetime.utcnow() @@ -37,7 +59,14 @@ class BindingService: rows = rows[:limit] return rows, has_more - def direct_bind(self, device_id: str, child_id: int | None, user_id: int) -> Mapping: + def direct_bind( + self, + device_id: str, + serial_number: str, + child_id: int | None, + user_id: int, + ) -> Mapping: + self._ensure_bindable_device(device_id, serial_number) self.dao.direct_bind(device_id, child_id, user_id) return {"device_id": device_id, "child_id": child_id} diff --git a/mini-program/app/service/im.py b/mini-program/app/service/im.py new file mode 100644 index 0000000..dcbd5ef --- /dev/null +++ b/mini-program/app/service/im.py @@ -0,0 +1,707 @@ +from dataclasses import dataclass +import json +from collections.abc import Mapping +from typing import Any + +from fastapi import HTTPException, status +from sqlalchemy import text +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +try: + from app.db_compat import current_timestamp_sql, inserted_primary_key, select_for_update_clause + from app.schemas.im import ( + ChildConversationMessageItem, + DeviceMessageCreateRequest, + ParentChildMessageCreateRequest, + ) +except ModuleNotFoundError: + from db_compat import current_timestamp_sql, inserted_primary_key, select_for_update_clause + from schemas.im import ChildConversationMessageItem, DeviceMessageCreateRequest, ParentChildMessageCreateRequest + + +PARENT_PARTICIPANT_TYPE = 1 +CHILD_PARTICIPANT_TYPE = 2 + +CHILD_PEER_CONVERSATION_TYPE = 1 +PARENT_CHILD_CONVERSATION_TYPE = 2 + +CONVERSATION_TYPE_NAMES = { + CHILD_PEER_CONVERSATION_TYPE: "child_peer", + PARENT_CHILD_CONVERSATION_TYPE: "parent_child", +} + +PARTICIPANT_TYPE_NAMES = { + PARENT_PARTICIPANT_TYPE: "parent", + CHILD_PARTICIPANT_TYPE: "child", +} + + +@dataclass(frozen=True) +class DeviceIdentity: + device_id: str + child_id: int + child_name: str | None + + +@dataclass(frozen=True) +class ConversationMessageCreateResult: + idempotent: bool + conversation_id: int + conversation_type: int + message: ChildConversationMessageItem + + @property + def conversation_type_name(self) -> str: + return conversation_type_name(self.conversation_type) + + +def conversation_type_name(conversation_type: int) -> str: + return CONVERSATION_TYPE_NAMES.get(conversation_type, f"unknown_{conversation_type}") + + +def participant_type_name(participant_type: int) -> str: + return PARTICIPANT_TYPE_NAMES.get(participant_type, f"unknown_{participant_type}") + + +def normalize_content_json(value: Any) -> dict[str, Any] | None: + if value is None: + return None + if isinstance(value, dict): + return value + if isinstance(value, str): + try: + parsed = json.loads(value) + except json.JSONDecodeError: + return None + if isinstance(parsed, dict): + return parsed + return None + + +def row_to_message_item(row: Mapping[str, Any]) -> ChildConversationMessageItem: + return ChildConversationMessageItem( + id=int(row["id"]), + conversation_id=int(row["conversation_id"]), + seq=int(row["seq"]), + sender_type=participant_type_name(int(row["sender_type"])), + sender_id=str(row["sender_id"]), + receiver_type=participant_type_name(int(row["receiver_type"])), + receiver_id=str(row["receiver_id"]), + content_type=int(row["content_type"]), + content_text=row["content_text"], + content_json=normalize_content_json(row["content_json"]), + media_file_key=row["media_file_key"], + media_duration_ms=row["media_duration_ms"], + media_mime_type=row["media_mime_type"], + media_size_bytes=row["media_size_bytes"], + media_transcript_text=row["media_transcript_text"], + client_msg_id=row["client_msg_id"], + sender_name_snapshot=row["sender_name_snapshot"], + sender_avatar_snapshot=row["sender_avatar_snapshot"], + receiver_name_snapshot=row["receiver_name_snapshot"], + receiver_avatar_snapshot=row["receiver_avatar_snapshot"], + ext_json=normalize_content_json(row["ext_json"]), + created_at=row["created_at"], + ) + + +def assert_parent_child_access(db: Session, *, user_id: int, child_id: int) -> Mapping[str, Any]: + child_row = _get_child_row(db, child_id) + if not child_row: + raise HTTPException(status_code=404, detail="child not found") + + has_access = ( + db.execute( + text( + """ + SELECT 1 + FROM parent_child_relations + WHERE user_id = :user_id + AND child_id = :child_id + AND status = 1 + LIMIT 1 + """ + ), + {"user_id": user_id, "child_id": child_id}, + ).scalar_one_or_none() + is not None + ) + if not has_access: + raise HTTPException(status_code=403, detail="no permission to access this child") + + return child_row + + +def authenticate_device_identity(db: Session, *, device_id: str, serial_number: str) -> DeviceIdentity: + row = ( + db.execute( + text( + """ + SELECT + da.device_id, + db.child_id, + c.child_name + FROM device_auth AS da + LEFT JOIN device_bindings AS db + ON db.device_id = da.device_id + AND db.status = 1 + LEFT JOIN children AS c + ON c.child_id = db.child_id + AND c.status = 1 + WHERE da.device_id = :device_id + AND da.serial_number = :serial_number + AND da.is_active = 1 + LIMIT 1 + """ + ), + {"device_id": device_id, "serial_number": serial_number}, + ) + .mappings() + .first() + ) + if not row: + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid device credentials") + if row["child_id"] is None: + raise HTTPException(status_code=404, detail="device not bound to a child") + return DeviceIdentity( + device_id=str(row["device_id"]), + child_id=int(row["child_id"]), + child_name=row["child_name"], + ) + + +def create_parent_child_message( + db: Session, + *, + parent_user_id: int, + child_id: int, + payload: ParentChildMessageCreateRequest, +) -> ConversationMessageCreateResult: + child_row = assert_parent_child_access(db, user_id=parent_user_id, child_id=child_id) + parent_row = _get_parent_row(db, parent_user_id) + if not parent_row: + raise HTTPException(status_code=404, detail="parent not found") + + conversation_id, idempotent = _create_message( + db=db, + conversation_type=PARENT_CHILD_CONVERSATION_TYPE, + participant_a_type=CHILD_PARTICIPANT_TYPE, + participant_a_id=str(child_id), + participant_b_type=PARENT_PARTICIPANT_TYPE, + participant_b_id=str(parent_user_id), + pair_key=f"{child_id}:{parent_user_id}", + sender_type=PARENT_PARTICIPANT_TYPE, + sender_id=str(parent_user_id), + receiver_type=CHILD_PARTICIPANT_TYPE, + receiver_id=str(child_id), + sender_name_snapshot=parent_row["nickname"], + sender_avatar_snapshot=parent_row["avatar_url"], + receiver_name_snapshot=child_row["child_name"], + receiver_avatar_snapshot=None, + payload=payload, + ) + message_row = _get_message_by_conversation_client_id( + db, + conversation_id=conversation_id, + client_msg_id=payload.client_msg_id, + ) + if not message_row: + raise RuntimeError("message was not found after insert") + + return ConversationMessageCreateResult( + idempotent=idempotent, + conversation_id=conversation_id, + conversation_type=PARENT_CHILD_CONVERSATION_TYPE, + message=row_to_message_item(message_row), + ) + + +def create_device_message( + db: Session, + *, + device_id: str, + serial_number: str, + payload: DeviceMessageCreateRequest, +) -> tuple[DeviceIdentity, ConversationMessageCreateResult]: + device_identity = authenticate_device_identity( + db, + device_id=device_id, + serial_number=serial_number, + ) + + if payload.conversation_type == CHILD_PEER_CONVERSATION_TYPE: + if payload.peer_child_id == device_identity.child_id: + raise HTTPException(status_code=400, detail="peer_child_id must be different from current child") + sender_child_row = assert_child_exists(db, child_id=device_identity.child_id) + receiver_child_row = assert_child_exists(db, child_id=payload.peer_child_id) + participant_a_id, participant_b_id, pair_key = _build_child_peer_pair( + device_identity.child_id, + payload.peer_child_id, + ) + + conversation_id, idempotent = _create_message( + db=db, + conversation_type=CHILD_PEER_CONVERSATION_TYPE, + participant_a_type=CHILD_PARTICIPANT_TYPE, + participant_a_id=participant_a_id, + participant_b_type=CHILD_PARTICIPANT_TYPE, + participant_b_id=participant_b_id, + pair_key=pair_key, + sender_type=CHILD_PARTICIPANT_TYPE, + sender_id=str(device_identity.child_id), + receiver_type=CHILD_PARTICIPANT_TYPE, + receiver_id=str(payload.peer_child_id), + sender_name_snapshot=sender_child_row["child_name"], + sender_avatar_snapshot=None, + receiver_name_snapshot=receiver_child_row["child_name"], + receiver_avatar_snapshot=None, + payload=payload, + ) + else: + sender_child_row = assert_child_exists(db, child_id=device_identity.child_id) + parent_row = _get_parent_row(db, payload.parent_user_id) + if not parent_row: + raise HTTPException(status_code=404, detail="parent not found") + assert_parent_child_access(db, user_id=payload.parent_user_id, child_id=device_identity.child_id) + + conversation_id, idempotent = _create_message( + db=db, + conversation_type=PARENT_CHILD_CONVERSATION_TYPE, + participant_a_type=CHILD_PARTICIPANT_TYPE, + participant_a_id=str(device_identity.child_id), + participant_b_type=PARENT_PARTICIPANT_TYPE, + participant_b_id=str(payload.parent_user_id), + pair_key=f"{device_identity.child_id}:{payload.parent_user_id}", + sender_type=CHILD_PARTICIPANT_TYPE, + sender_id=str(device_identity.child_id), + receiver_type=PARENT_PARTICIPANT_TYPE, + receiver_id=str(payload.parent_user_id), + sender_name_snapshot=sender_child_row["child_name"], + sender_avatar_snapshot=None, + receiver_name_snapshot=parent_row["nickname"], + receiver_avatar_snapshot=parent_row["avatar_url"], + payload=payload, + ) + + message_row = _get_message_by_conversation_client_id( + db, + conversation_id=conversation_id, + client_msg_id=payload.client_msg_id, + ) + if not message_row: + raise RuntimeError("message was not found after insert") + + result = ConversationMessageCreateResult( + idempotent=idempotent, + conversation_id=conversation_id, + conversation_type=payload.conversation_type, + message=row_to_message_item(message_row), + ) + return device_identity, result + + +def assert_child_exists(db: Session, *, child_id: int) -> Mapping[str, Any]: + child_row = _get_child_row(db, child_id) + if not child_row: + raise HTTPException(status_code=404, detail="child not found") + return child_row + + +def _get_child_row(db: Session, child_id: int) -> Mapping[str, Any] | None: + return ( + db.execute( + text( + """ + SELECT child_id, child_name, status + FROM children + WHERE child_id = :child_id + AND status = 1 + LIMIT 1 + """ + ), + {"child_id": child_id}, + ) + .mappings() + .first() + ) + + +def _get_parent_row(db: Session, user_id: int) -> Mapping[str, Any] | None: + return ( + db.execute( + text( + """ + SELECT user_id, nickname, avatar_url, status + FROM parents + WHERE user_id = :user_id + AND status = 1 + LIMIT 1 + """ + ), + {"user_id": user_id}, + ) + .mappings() + .first() + ) + + +def _build_child_peer_pair(child_a_id: int, child_b_id: int) -> tuple[str, str, str]: + low_id, high_id = sorted((child_a_id, child_b_id)) + participant_a_id = str(low_id) + participant_b_id = str(high_id) + return participant_a_id, participant_b_id, f"{participant_a_id}:{participant_b_id}" + + +def _build_preview(content_type: int, content_text: str | None) -> str: + if content_type == 1: + return (content_text or "").strip()[:255] + if content_type == 2: + return "[audio]" + if content_type == 3: + return "[image]" + return "[json]" + + +def _create_message( + db: Session, + *, + conversation_type: int, + participant_a_type: int, + participant_a_id: str, + participant_b_type: int, + participant_b_id: str, + pair_key: str, + sender_type: int, + sender_id: str, + receiver_type: int, + receiver_id: str, + sender_name_snapshot: str | None, + sender_avatar_snapshot: str | None, + receiver_name_snapshot: str | None, + receiver_avatar_snapshot: str | None, + payload: ParentChildMessageCreateRequest | DeviceMessageCreateRequest, +) -> tuple[int, bool]: + conversation_id = _get_or_create_conversation( + db=db, + conversation_type=conversation_type, + participant_a_type=participant_a_type, + participant_a_id=participant_a_id, + participant_b_type=participant_b_type, + participant_b_id=participant_b_id, + pair_key=pair_key, + ) + + existing = _get_message_by_conversation_client_id( + db, + conversation_id=conversation_id, + client_msg_id=payload.client_msg_id, + ) + if existing: + return conversation_id, True + + now_sql = current_timestamp_sql(db) + preview = _build_preview(payload.content_type, payload.content_text) + + try: + conversation_row = _get_conversation_by_id(db, conversation_id=conversation_id, lock=True) + if not conversation_row: + raise HTTPException(status_code=404, detail="conversation not found") + next_seq = int(conversation_row["last_seq"]) + 1 + + message_id = _next_primary_key(db, "im_messages") + insert_sql = f""" + INSERT INTO im_messages ( + {'id,' if message_id is not None else ''} + conversation_id, + seq, + sender_type, + sender_id, + receiver_type, + receiver_id, + content_type, + content_text, + content_json, + media_file_key, + media_duration_ms, + media_mime_type, + media_size_bytes, + media_transcript_text, + client_msg_id, + sender_name_snapshot, + sender_avatar_snapshot, + receiver_name_snapshot, + receiver_avatar_snapshot, + ext_json, + created_at + ) VALUES ( + {':id,' if message_id is not None else ''} + :conversation_id, + :seq, + :sender_type, + :sender_id, + :receiver_type, + :receiver_id, + :content_type, + :content_text, + :content_json, + :media_file_key, + :media_duration_ms, + :media_mime_type, + :media_size_bytes, + :media_transcript_text, + :client_msg_id, + :sender_name_snapshot, + :sender_avatar_snapshot, + :receiver_name_snapshot, + :receiver_avatar_snapshot, + :ext_json, + {now_sql} + ) + """ + params: dict[str, Any] = { + "conversation_id": conversation_id, + "seq": next_seq, + "sender_type": sender_type, + "sender_id": sender_id, + "receiver_type": receiver_type, + "receiver_id": receiver_id, + "content_type": payload.content_type, + "content_text": payload.content_text, + "content_json": json.dumps(payload.content_json) if payload.content_json is not None else None, + "media_file_key": payload.media_file_key, + "media_duration_ms": payload.media_duration_ms, + "media_mime_type": payload.media_mime_type, + "media_size_bytes": payload.media_size_bytes, + "media_transcript_text": payload.media_transcript_text, + "client_msg_id": payload.client_msg_id, + "sender_name_snapshot": sender_name_snapshot, + "sender_avatar_snapshot": sender_avatar_snapshot, + "receiver_name_snapshot": receiver_name_snapshot, + "receiver_avatar_snapshot": receiver_avatar_snapshot, + "ext_json": json.dumps(payload.ext_json) if payload.ext_json is not None else None, + } + if message_id is not None: + params["id"] = message_id + + result = db.execute(text(insert_sql), params) + if message_id is None: + message_id = inserted_primary_key(result) + + db.execute( + text( + f""" + UPDATE im_conversations + SET status = 1, + last_seq = :last_seq, + message_count = message_count + 1, + last_message_preview = :last_message_preview, + last_message_at = {now_sql}, + updated_at = {now_sql} + WHERE id = :conversation_id + """ + ), + { + "conversation_id": conversation_id, + "last_seq": next_seq, + "last_message_preview": preview, + }, + ) + db.commit() + return conversation_id, False + except IntegrityError: + db.rollback() + existing = _get_message_by_conversation_client_id( + db, + conversation_id=conversation_id, + client_msg_id=payload.client_msg_id, + ) + if existing: + return conversation_id, True + raise + except Exception: + db.rollback() + raise + + +def _get_or_create_conversation( + db: Session, + *, + conversation_type: int, + participant_a_type: int, + participant_a_id: str, + participant_b_type: int, + participant_b_id: str, + pair_key: str, +) -> int: + row = _get_conversation_by_pair( + db, + conversation_type=conversation_type, + pair_key=pair_key, + lock=False, + ) + if row: + return int(row["id"]) + + now_sql = current_timestamp_sql(db) + conversation_id = _next_primary_key(db, "im_conversations") + insert_sql = f""" + INSERT INTO im_conversations ( + {'id,' if conversation_id is not None else ''} + conversation_type, + participant_a_type, + participant_a_id, + participant_b_type, + participant_b_id, + pair_key, + status, + last_seq, + message_count, + created_at, + updated_at + ) VALUES ( + {':id,' if conversation_id is not None else ''} + :conversation_type, + :participant_a_type, + :participant_a_id, + :participant_b_type, + :participant_b_id, + :pair_key, + 1, + 0, + 0, + {now_sql}, + {now_sql} + ) + """ + params = { + "conversation_type": conversation_type, + "participant_a_type": participant_a_type, + "participant_a_id": participant_a_id, + "participant_b_type": participant_b_type, + "participant_b_id": participant_b_id, + "pair_key": pair_key, + } + if conversation_id is not None: + params["id"] = conversation_id + + try: + result = db.execute(text(insert_sql), params) + if conversation_id is not None: + return conversation_id + return inserted_primary_key(result) + except IntegrityError: + db.rollback() + row = _get_conversation_by_pair( + db, + conversation_type=conversation_type, + pair_key=pair_key, + lock=False, + ) + if row: + return int(row["id"]) + raise + + +def _get_conversation_by_pair( + db: Session, + *, + conversation_type: int, + pair_key: str, + lock: bool, +) -> Mapping[str, Any] | None: + lock_clause = select_for_update_clause(db) if lock else "" + return ( + db.execute( + text( + f""" + SELECT id, conversation_type, last_seq, status + FROM im_conversations + WHERE conversation_type = :conversation_type + AND pair_key = :pair_key + LIMIT 1{lock_clause} + """ + ), + {"conversation_type": conversation_type, "pair_key": pair_key}, + ) + .mappings() + .first() + ) + + +def _get_conversation_by_id( + db: Session, + *, + conversation_id: int, + lock: bool, +) -> Mapping[str, Any] | None: + lock_clause = select_for_update_clause(db) if lock else "" + return ( + db.execute( + text( + f""" + SELECT id, last_seq, message_count, status + FROM im_conversations + WHERE id = :conversation_id + LIMIT 1{lock_clause} + """ + ), + {"conversation_id": conversation_id}, + ) + .mappings() + .first() + ) + + +def _get_message_by_conversation_client_id( + db: Session, + *, + conversation_id: int, + client_msg_id: str, +) -> Mapping[str, Any] | None: + return ( + db.execute( + text( + """ + SELECT + id, + conversation_id, + seq, + sender_type, + sender_id, + receiver_type, + receiver_id, + content_type, + content_text, + content_json, + media_file_key, + media_duration_ms, + media_mime_type, + media_size_bytes, + media_transcript_text, + client_msg_id, + sender_name_snapshot, + sender_avatar_snapshot, + receiver_name_snapshot, + receiver_avatar_snapshot, + ext_json, + created_at + FROM im_messages + WHERE conversation_id = :conversation_id + AND client_msg_id = :client_msg_id + LIMIT 1 + """ + ), + {"conversation_id": conversation_id, "client_msg_id": client_msg_id}, + ) + .mappings() + .first() + ) + + +def _next_primary_key(db: Session, table_name: str) -> int | None: + if db.get_bind() is None or db.get_bind().dialect.name != "sqlite": + return None + return int( + db.execute(text(f"SELECT COALESCE(MAX(id), 0) + 1 FROM {table_name}")).scalar_one() + ) diff --git a/mini-program/app/service/location.py b/mini-program/app/service/location.py new file mode 100644 index 0000000..9096ae5 --- /dev/null +++ b/mini-program/app/service/location.py @@ -0,0 +1,327 @@ +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import datetime +from typing import Any + +from fastapi import HTTPException, status +from sqlalchemy import text +from sqlalchemy.orm import Session + +try: + from app.db_compat import current_timestamp_sql, select_for_update_clause + from app.schemas.location import DeviceLocationReportRequest + from app.service.im import DeviceIdentity, authenticate_device_identity +except ModuleNotFoundError: + from db_compat import current_timestamp_sql, select_for_update_clause + from schemas.location import DeviceLocationReportRequest + from service.im import DeviceIdentity, authenticate_device_identity + + +@dataclass(frozen=True) +class ParentDeviceAccess: + device_id: str + child_id: int + child_name: str | None + + +def assert_parent_device_access(db: Session, *, device_id: str, user_id: int) -> ParentDeviceAccess: + row = ( + db.execute( + text( + """ + SELECT + db.device_id, + db.child_id, + c.child_name + FROM device_bindings AS db + LEFT JOIN children AS c + ON c.child_id = db.child_id + AND c.status = 1 + WHERE db.device_id = :device_id + AND db.owner_user_id = :user_id + AND db.status = 1 + LIMIT 1 + """ + ), + {"device_id": device_id, "user_id": user_id}, + ) + .mappings() + .first() + ) + if not row: + raise HTTPException(status_code=404, detail="device not found") + if row["child_id"] is None: + raise HTTPException(status_code=404, detail="device not bound to a child") + + return ParentDeviceAccess( + device_id=str(row["device_id"]), + child_id=int(row["child_id"]), + child_name=row["child_name"], + ) + + +def report_device_location( + db: Session, + *, + device_id: str, + serial_number: str, + payload: DeviceLocationReportRequest, +) -> tuple[DeviceIdentity, Mapping[str, Any]]: + device_identity = authenticate_device_identity( + db=db, + device_id=device_id, + serial_number=serial_number, + ) + now_sql = current_timestamp_sql(db) + child_id = device_identity.child_id + + try: + current_row = _get_current_location_row(db, child_id=child_id, lock=True) + params = { + "child_id": child_id, + "device_id": device_identity.device_id, + "coord_type": payload.coord_type, + "lat": payload.lat, + "lng": payload.lng, + "accuracy_m": payload.accuracy_m, + "altitude_m": payload.altitude_m, + "speed_mps": payload.speed_mps, + "heading_deg": payload.heading_deg, + "source": payload.source, + "battery_pct": payload.battery_pct, + "device_time": payload.device_time, + } + if current_row: + db.execute( + text( + f""" + UPDATE child_location_current + SET device_id = :device_id, + coord_type = :coord_type, + lat = :lat, + lng = :lng, + accuracy_m = :accuracy_m, + altitude_m = :altitude_m, + speed_mps = :speed_mps, + heading_deg = :heading_deg, + source = :source, + battery_pct = :battery_pct, + device_time = :device_time, + server_time = {now_sql}, + updated_at = {now_sql} + WHERE child_id = :child_id + """ + ), + params, + ) + else: + db.execute( + text( + f""" + INSERT INTO child_location_current ( + child_id, + device_id, + coord_type, + lat, + lng, + accuracy_m, + altitude_m, + speed_mps, + heading_deg, + source, + battery_pct, + device_time, + server_time, + updated_at + ) VALUES ( + :child_id, + :device_id, + :coord_type, + :lat, + :lng, + :accuracy_m, + :altitude_m, + :speed_mps, + :heading_deg, + :source, + :battery_pct, + :device_time, + {now_sql}, + {now_sql} + ) + """ + ), + params, + ) + + history_id = _next_primary_key(db, "child_location_history") + insert_history_sql = f""" + INSERT INTO child_location_history ( + {'id,' if history_id is not None else ''} + child_id, + device_id, + coord_type, + lat, + lng, + accuracy_m, + altitude_m, + speed_mps, + heading_deg, + source, + battery_pct, + device_time, + server_time, + created_at + ) VALUES ( + {':id,' if history_id is not None else ''} + :child_id, + :device_id, + :coord_type, + :lat, + :lng, + :accuracy_m, + :altitude_m, + :speed_mps, + :heading_deg, + :source, + :battery_pct, + :device_time, + {now_sql}, + {now_sql} + ) + """ + history_params = dict(params) + if history_id is not None: + history_params["id"] = history_id + db.execute(text(insert_history_sql), history_params) + db.commit() + except Exception: + db.rollback() + raise + + current_row = _get_current_location_row(db, child_id=child_id, lock=False) + if not current_row: + raise RuntimeError("current location not found after report") + return device_identity, current_row + + +def get_device_current_location( + db: Session, + *, + device_id: str, + user_id: int, +) -> Mapping[str, Any]: + access = assert_parent_device_access(db, device_id=device_id, user_id=user_id) + row = _get_current_location_row(db, child_id=access.child_id, lock=False) + if not row or str(row["device_id"]) != device_id: + raise HTTPException(status_code=404, detail="location not found") + return {**row, "child_name": access.child_name} + + +def get_device_trajectory( + db: Session, + *, + device_id: str, + user_id: int, + start_at: datetime | None, + end_at: datetime | None, + limit: int, +) -> tuple[ParentDeviceAccess, list[Mapping[str, Any]]]: + access = assert_parent_device_access(db, device_id=device_id, user_id=user_id) + params: dict[str, Any] = { + "device_id": device_id, + "child_id": access.child_id, + "fetch_limit": limit, + } + where = """ + device_id = :device_id + AND child_id = :child_id + """ + if start_at is not None: + where += " AND device_time >= :start_at" + params["start_at"] = start_at + if end_at is not None: + where += " AND device_time <= :end_at" + params["end_at"] = end_at + + rows = ( + db.execute( + text( + f""" + SELECT + id, + child_id, + device_id, + coord_type, + lat, + lng, + accuracy_m, + altitude_m, + speed_mps, + heading_deg, + source, + battery_pct, + device_time, + server_time, + created_at + FROM child_location_history + WHERE {where} + ORDER BY device_time DESC, id DESC + LIMIT :fetch_limit + """ + ), + params, + ) + .mappings() + .all() + ) + rows = list(rows) + rows.reverse() + return access, rows + + +def _get_current_location_row( + db: Session, + *, + child_id: int, + lock: bool, +) -> Mapping[str, Any] | None: + lock_clause = select_for_update_clause(db) if lock else "" + return ( + db.execute( + text( + f""" + SELECT + child_id, + device_id, + coord_type, + lat, + lng, + accuracy_m, + altitude_m, + speed_mps, + heading_deg, + source, + battery_pct, + device_time, + server_time, + updated_at + FROM child_location_current + WHERE child_id = :child_id + LIMIT 1{lock_clause} + """ + ), + {"child_id": child_id}, + ) + .mappings() + .first() + ) + + +def _next_primary_key(db: Session, table_name: str) -> int | None: + bind = db.get_bind() + if bind is None or bind.dialect.name != "sqlite": + return None + return int( + db.execute(text(f"SELECT COALESCE(MAX(id), 0) + 1 FROM {table_name}")).scalar_one() + ) diff --git a/mini-program/app/settings.py b/mini-program/app/settings.py index 2bad1e5..8ad0c74 100644 --- a/mini-program/app/settings.py +++ b/mini-program/app/settings.py @@ -5,10 +5,12 @@ from urllib.parse import quote_plus from pydantic import Field from pydantic_settings import BaseSettings, SettingsConfigDict +ENV_FILE = Path(__file__).resolve().parent.parent / ".env" + class Settings(BaseSettings): model_config = SettingsConfigDict( - env_file=".env", + env_file=str(ENV_FILE), env_file_encoding="utf-8", extra="ignore", ) @@ -27,7 +29,8 @@ class Settings(BaseSettings): db_port: int = Field(default=3306, validation_alias="DB_PORT") db_user: str = Field(default="root", validation_alias="DB_USER") db_password: str = Field(default="", validation_alias="DB_PASSWORD") - db_name: str = Field(default="mini_program", validation_alias="DB_NAME") + db_name: str = Field(default="talkingq", validation_alias="DB_NAME") + db_auto_init_tables: bool = Field(default=False, validation_alias="DB_AUTO_INIT_TABLES") db_path: str = Field(default="./data.db", validation_alias="DB_PATH") wechat_app_id: str = Field(default="", validation_alias="WECHAT_APP_ID") wechat_app_secret: str = Field(default="", validation_alias="WECHAT_APP_SECRET") @@ -89,6 +92,10 @@ class Settings(BaseSettings): return self.sqlite_dsn return self.mysql_dsn + @property + def uses_shared_talkingq_db(self) -> bool: + return self.db_type == "mysql" and self.db_name == "talkingq" + @lru_cache(maxsize=1) def get_settings() -> Settings: diff --git a/mini-program/database/init.sql b/mini-program/database/init.sql index 5df5d79..d26de6c 100644 --- a/mini-program/database/init.sql +++ b/mini-program/database/init.sql @@ -1,5 +1,12 @@ -- Current MySQL schema for mini-program +-- Deprecated for shared deployments. +-- If `mini-program` and `talkingq-url` run on the same MySQL instance, +-- use `database/talkingq_shared_schema.sql` to initialize the single +-- source-of-truth database `talkingq`. +-- +-- This file is kept only for legacy isolated development of mini-program. + CREATE DATABASE IF NOT EXISTS mini_program DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; diff --git a/mini-program/design.md b/mini-program/design.md index 4570147..c4db5b0 100644 --- a/mini-program/design.md +++ b/mini-program/design.md @@ -1,4 +1,4 @@ -# Banban 统一数据与接口设计文档（Mini Program + talkingq-url，V3） +# Banban 统一数据与接口设计文档（Mini Program + talkingq-url，V4） ## 1. 目标与边界 @@ -7,12 +7,14 @@ 1. 两个服务使用同一套设备身份（`device_id`）。 2. 保留 `talkingq-url/mysql/init/01-init.sql` 现有表结构不变。 3. 在不改动基础表的前提下，补齐家长/小孩/绑定/聊天/定位能力。 +4. 家长账号体系统一采用微信小程序登录，服务端基于 `code2Session` 换取真实 `openid/unionid`。 硬性边界： 1. `01-init.sql` 中现有表不改表名、不改字段、不改约束。 2. 所有新增表仅做增量扩展。 3. 新增表字段命名风格对齐现有风格：`snake_case`、`*_id`、`is_active`、`status`、`created_at`、`updated_at`。 +4. 微信 `AppSecret` 只能保存在服务端配置中，小程序端只持有一次性 `code`。 ## 2. 现有冻结表（来自 01-init.sql） @@ -42,10 +44,20 @@ 3. `unionid` VARCHAR(64) NULL 4. `nickname` VARCHAR(64) NULL 5. `avatar_url` VARCHAR(255) NULL -6. `phone` VARCHAR(20) NULL -7. `status` TINYINT NOT NULL DEFAULT 1 -8. `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP -9. `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP +6. `avatar_file_key` VARCHAR(255) NULL +7. `phone` VARCHAR(20) NULL +8. `status` TINYINT NOT NULL DEFAULT 1 +9. `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP +10. `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP + +说明： + +1. `avatar_file_key` 作为头像主存储值，保存 COS 对象 key。 +2. `avatar_url` 作为兼容或展示字段使用，不作为唯一主依据。 +3. `openid` 必须存储微信服务端 `code2Session` 返回的真实 OpenID；禁止将前端 `code`、测试用户名或设备标识写入该字段。 +4. `unionid` 仅在微信返回时写入；历史记录为空时，后续登录可补写，不作为首次登录成功前置条件。 +5. `nickname/avatar_url` 为展示资料，可由小程序端在登录时上传并在后续更新，但不参与身份认证。 +6. `phone` 为业务补充字段，与微信登录态解耦；如后续接入手机号授权，应基于服务端掌握的 `session_key` 解密后再写入。 索引建议： @@ -90,25 +102,33 @@ ### 3.4 设备归属表 `device_bindings` -用途：表达“当前哪块设备绑定到哪个小孩”，并与 `device_auth` 对齐。 +用途：表达“当前哪块设备归属到哪个家长，并可选关联到哪个小孩”，并与 `device_auth` 对齐。 建议字段： 1. `id` BIGINT UNSIGNED PK AUTO_INCREMENT 2. `device_id` VARCHAR(64) NOT NULL -3. `child_id` BIGINT UNSIGNED NOT NULL -4. `status` TINYINT NOT NULL DEFAULT 1 -5. `bound_at` DATETIME NOT NULL -6. `unbound_at` DATETIME NULL -7. `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP -8. `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP +3. `owner_user_id` BIGINT UNSIGNED NOT NULL +4. `child_id` BIGINT UNSIGNED NULL +5. `status` TINYINT NOT NULL DEFAULT 1 +6. `bound_at` DATETIME NOT NULL +7. `unbound_at` DATETIME NULL +8. `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP +9. `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP + +说明： + +1. 设备可以先归属到家长，再在后续流程中补充关联 `child_id`。 +2. `child_id` 为空表示当前设备已归属，但尚未绑定到具体小孩。 约束建议： 1. `UNIQUE(device_id)` 2. `UNIQUE(child_id)` -3. `FOREIGN KEY (device_id) REFERENCES device_auth(device_id)` -4. `FOREIGN KEY (child_id) REFERENCES children(child_id)` +3. `INDEX idx_owner_user_id(owner_user_id)` +4. `FOREIGN KEY (device_id) REFERENCES device_auth(device_id)` +5. `FOREIGN KEY (owner_user_id) REFERENCES parents(user_id)` +6. `FOREIGN KEY (child_id) REFERENCES children(child_id)` ### 3.5 绑定会话表 `device_bind_sessions` @@ -318,13 +338,24 @@ ## 5. 核心业务流程 -### 5.1 设备注册与鉴权 +### 5.1 家长登录流程（微信小程序） + +1. 小程序调用 `wx.login` / `Taro.login` 获取一次性 `code`。 +2. 小程序调用后端 `POST /auth/login`，上传 `code`，并可附带 `nickname/avatar_url` 作为资料字段。 +3. 后端使用服务端配置的 `wechat_app_id + wechat_app_secret` 调用微信 `code2Session`。 +4. 后端从微信响应中获取真实 `openid`、可选 `unionid`，以及仅供服务端使用的 `session_key`。 +5. 后端以 `openid` 为唯一身份键 upsert `parents`；若本次拿到 `unionid` 且库中为空，则补写 `unionid`。 +6. 后端签发本地 `access_token(JWT)` 返回给小程序，后续业务接口仍统一使用本地 Bearer Token。 +7. 若微信换码失败、`openid` 缺失、`code` 失效或被重复使用，登录直接失败，不创建本地用户。 +8. `session_key` 仅用于后续需要的微信敏感数据解密能力（如手机号）；当前阶段不作为登录态主键，也不写入 `parents` 主表。 + +### 5.2 设备注册与鉴权 1. 生产或后台调用 `talkingq-url` 现有注册接口，写入 `device_auth`。 2. 手表请求携带 `X-Device-ID` 与 `X-Device-Serial`。 3. 后端基于 `device_auth` 鉴权通过后继续处理聊天/定位/绑定。 -### 5.2 绑定流程（扫码 + 确认码） +### 5.3 绑定流程（扫码 + 确认码） 1. 家长登录后发起 `bind/start`，创建 `device_bind_sessions`。 2. 手表上报确认码，更新会话挑战信息。 @@ -335,16 +366,17 @@ 7. 追加 `device_bind_history`。 8. 更新会话状态为 `confirmed/consumed` 并提交事务。 -### 5.3 聊天流程 +### 5.4 聊天流程 1. 设备 AI 对话仍走 `conversation_histories + conversation_messages`。 2. 家长-小孩、小孩-小孩消息走 `im_conversations + im_messages`。 3. IM 消息统一使用 `sender_type/sender_id + receiver_type/receiver_id`，不按场景拆专用发送者字段。 -4. 发送接口必须使用 `client_msg_id` 幂等。 -5. 查询前必须校验 `parent_child_relations` 权限。 -6. 列表展示优先读取消息快照字段，降低跨表查询成本。 +4. `mini-program` 侧不再新增独立的 `chat_*` 消息表定义，统一以 `im_*` 为 IM 正表。 +5. 发送接口必须使用 `client_msg_id` 幂等。 +6. 查询前必须校验 `parent_child_relations` 权限。 +7. 列表展示优先读取消息快照字段，降低跨表查询成本。 -### 5.4 位置流程 +### 5.5 位置流程 1. 手表上报定位写入 `child_location_history`。 2. 同事务更新 `child_location_current`。 @@ -361,19 +393,91 @@ 6. 关键写路径（绑定确认、发消息、位置写入）必须事务化。 7. 腾讯位置服务启用前提：必须配置合法 `key`，并完成域名/来源白名单与配额设置。 8. 地图相关接口统一使用 `gcj02` 和 `lat,lng`；禁止混用 `wgs84/bd09` 直接入库。 +9. 小程序端只上送一次性 `code`，不得自行调用微信换取 `openid`，更不得持有 `AppSecret`。 +10. `code` 只用于调用微信 `code2Session`，不得入库为 `openid`、`unionid` 或任何长期身份字段。 +11. `session_key` 不返回前端，不写入 `parents` 主表；如需支持手机号等敏感数据解密，应服务端短期缓存并设置明确过期时间。 +12. `nickname/avatar_url` 仅作资料字段，不能作为登录态、账号归并或权限判断依据。 ## 7. 分阶段落地 -1. 第一阶段：落地 `parents/children/parent_child_relations/device_bind_*`，打通绑定链路。 -2. 第二阶段：落地 `im_conversations/im_messages`，打通家长与小孩 IM。 -3. 第三阶段：落地 `child_location_current/history`，打通定位链路。 -4. 第四阶段：补充审计、告警、实时推送和运营能力。 +1. 第一阶段：落地微信登录链路，打通 `Taro.login -> /auth/login -> code2Session -> parents -> JWT`。 +2. 第二阶段：落地 `parents/children/parent_child_relations/device_bind_*`，打通绑定链路。 +3. 第三阶段：落地 `im_conversations/im_messages`，打通家长与小孩 IM。 +4. 第四阶段：落地 `child_location_current/history`，打通定位链路。 +5. 第五阶段：补充审计、告警、实时推送和运营能力。 ## 8. 验收标准 -1. 不修改 `01-init.sql` 任一现有表，服务可正常启动。 -2. 设备鉴权、角色配置、OTA能力保持可用。 -3. 家长可完成扫码绑定、解绑、重绑，且历史保留。 -4. 家长可查看自己小孩与其他小孩完整双向消息。 -5. 小程序可查看小孩最新位置与历史轨迹。 +1. 家长可通过真实微信登录完成换码，并获取本地 Bearer Token。 +2. `parents.openid` 存储值来自微信返回的真实 OpenID，而不是前端 `code` 或测试标识。 +3. 不修改 `01-init.sql` 任一现有表，服务可正常启动。 +4. 设备鉴权、角色配置、OTA能力保持可用。 +5. 家长可完成扫码绑定、解绑、重绑，且历史保留。 +6. 家长可查看自己小孩与其他小孩完整双向消息。 +7. 小程序可查看小孩最新位置与历史轨迹。 + +## 9. 共享库落地规范 + +### 9.1 唯一数据库与唯一 Schema + +1. `mini-program` 与 `talkingq-url` 必须共用同一个 MySQL 数据库：`talkingq`。 +2. `database/talkingq_shared_schema.sql` 是唯一 schema 真相源。 +3. `mini-program/database/init.sql` 仅保留给历史上的“单服务独立开发”场景，不再作为共享部署的建库依据。 + +### 9.2 表归属（Table Ownership） + +`talkingq-url` 拥有以下表的写入责任： + +1. `device_auth` +2. `device_configs` +3. `conversation_histories` +4. `conversation_messages` +5. `roles` +6. `role_languages` +7. `device_firmware_update` +8. `system_config` + +`mini-program` 拥有以下表的写入责任： + +1. `parents` +2. `children` +3. `parent_child_relations` +4. `device_bindings` +5. `device_bind_sessions` +6. `device_bind_history` +7. `device_settings` +8. `cards` +9. `im_conversations` +10. `im_messages` +11. `child_location_current` +12. `child_location_history` + +约束： + +1. 两个服务允许跨服务读取共享依赖表，但不允许跨服务写对方负责的主表。 +2. `mini-program` 只读 `device_auth`，不直接承担设备注册主流程。 +3. `talkingq-url` 不写 `parents/children/device_bindings/im_* /child_location_*`。 + +### 9.3 共享字段与真相源 + +1. 设备主身份统一以 `device_auth.device_id` 为准。 +2. 设备与 AI 的历史消息统一以 `conversation_histories + conversation_messages` 为准。 +3. 家长/儿童 IM 统一以 `im_conversations + im_messages` 为准。 +4. 设备音量如需双端共享，以 `device_configs.volume` 为真相源；`device_settings.volume` 仅作小程序展示和兼容字段。 + +### 9.4 迁移原则 + +1. 目标库固定为 `talkingq`。 +2. 迁移时以 `talkingq-url` 当前使用的 `talkingq` 数据为准。 +3. 历史 `mini_program` 库中的业务数据不并入 `talkingq`，也不作为冲突解决依据。 +4. 迁移动作只做两件事： + 1. 在 `talkingq` 上补齐 `mini-program` 所需扩展表。 + 2. 将 `mini-program` 服务配置切换到 `talkingq`。 +5. `mini_program` 库可在切换验证完成后归档或删除，但不再作为运行时数据库。 + +### 9.5 启动与建表策略 + +1. 共享部署下，服务启动默认只检查数据库连通性，不自动执行 ORM `create_all`。 +2. 只有显式开启开发开关时，`mini-program` 才允许自动建表。 +3. 生产、联调、共享测试环境统一通过 `database/talkingq_shared_schema.sql` 或正式 migration 执行建表与变更。