From 356e333c813afedf572a4dc6c22bef57b2765e7e Mon Sep 17 00:00:00 2001 From: stu2not Date: Wed, 27 May 2026 17:05:24 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=9A=E4=BD=8D=E4=BF=A1=E6=81=AF=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0=E5=9C=B0=E5=9D=80=E8=A7=A3=E6=9E=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- banban-mini/src/pages/location/index.tsx | 9 +++ banban-mini/src/services/location.ts | 3 + database/talkingq_shared_schema.sql | 3 + talkingq-url/banban/dao/location.py | 81 +++++++++++++++++-- talkingq-url/banban/routers/devices.py | 3 + talkingq-url/banban/schemas/location.py | 3 + talkingq-url/banban/service/amap_geocode.py | 53 ++++++++++++ talkingq-url/banban/service/location.py | 89 ++++++++++++++++++++- talkingq-url/config.py | 7 ++ talkingq-url/database/init_db.py | 32 ++++++++ talkingq-url/database/models.py | 3 + talkingq-url/mysql/init/02-init.sql | 3 + 12 files changed, 281 insertions(+), 8 deletions(-) create mode 100644 talkingq-url/banban/service/amap_geocode.py diff --git a/banban-mini/src/pages/location/index.tsx b/banban-mini/src/pages/location/index.tsx index e1fbe1b..6dc0f2d 100644 --- a/banban-mini/src/pages/location/index.tsx +++ b/banban-mini/src/pages/location/index.tsx @@ -53,6 +53,14 @@ function formatOptionalNumber(value?: number | null, digits = 1): string | null return Number(value).toFixed(digits) } +function getLocationAddress(point: DeviceLocation | DeviceTrajectoryPoint): string { + const address = String(point.address || '').trim() + if (address) return address + if (point.address_resolve_status === 0) return '地点解析中' + if (point.address_resolve_status === 2) return '地点暂时无法解析' + return '当前地点未知' +} + function buildLocationDetailLines( point: DeviceLocation | DeviceTrajectoryPoint, options?: { includeIdentity?: boolean } @@ -444,6 +452,7 @@ export default function Location() { 📍 + {getLocationAddress(deviceLocation)} {deviceLocation.lat.toFixed(6)}, {deviceLocation.lng.toFixed(6)} diff --git a/banban-mini/src/services/location.ts b/banban-mini/src/services/location.ts index c3cbe37..38aa38a 100644 --- a/banban-mini/src/services/location.ts +++ b/banban-mini/src/services/location.ts @@ -16,6 +16,9 @@ export interface DeviceLocation { battery_pct?: number | null device_time: string server_time: string + address?: string | null + address_resolved_at?: string | null + address_resolve_status?: number | null updated_at: string stale?: boolean realtime?: boolean diff --git a/database/talkingq_shared_schema.sql b/database/talkingq_shared_schema.sql index 48c5bb5..73afe7d 100644 --- a/database/talkingq_shared_schema.sql +++ b/database/talkingq_shared_schema.sql @@ -552,6 +552,9 @@ CREATE TABLE IF NOT EXISTS `child_location_current` ( `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, + `address` VARCHAR(255) NULL, + `address_resolved_at` DATETIME NULL, + `address_resolve_status` TINYINT NULL COMMENT '0=pending,1=success,2=failed', PRIMARY KEY (`child_id`), KEY `idx_child_loc_current_device` (`device_id`), KEY `idx_child_loc_current_server_time` (`server_time`), diff --git a/talkingq-url/banban/dao/location.py b/talkingq-url/banban/dao/location.py index d7bb3c0..137c77e 100644 --- a/talkingq-url/banban/dao/location.py +++ b/talkingq-url/banban/dao/location.py @@ -81,9 +81,16 @@ class LocationDAO(BaseDAO): payload: Any, ) -> Mapping[str, Any]: now_sql = "CURRENT_TIMESTAMP(3)" + should_resolve_address = True try: current_row = await self._get_current_location_row(child_id=child_id, lock=True) + if current_row and current_row.get("address"): + should_resolve_address = self._is_location_address_stale( + current_row, + lat=float(payload.lat), + lng=float(payload.lng), + ) params = { "child_id": child_id, "device_id": device_id, @@ -115,11 +122,14 @@ class LocationDAO(BaseDAO): battery_pct = :battery_pct, device_time = :device_time, server_time = {now_sql}, - updated_at = {now_sql} + updated_at = {now_sql}, + address = CASE WHEN :should_resolve_address THEN NULL ELSE address END, + address_resolved_at = CASE WHEN :should_resolve_address THEN NULL ELSE address_resolved_at END, + address_resolve_status = CASE WHEN :should_resolve_address THEN 0 ELSE address_resolve_status END WHERE child_id = :child_id """ ), - params, + {**params, "should_resolve_address": should_resolve_address}, ) else: await self.execute( @@ -139,7 +149,8 @@ class LocationDAO(BaseDAO): battery_pct, device_time, server_time, - updated_at + updated_at, + address_resolve_status ) VALUES ( :child_id, :device_id, @@ -154,7 +165,8 @@ class LocationDAO(BaseDAO): :battery_pct, :device_time, {now_sql}, - {now_sql} + {now_sql}, + 0 ) """ ), @@ -209,7 +221,7 @@ class LocationDAO(BaseDAO): current_row = await self._get_current_location_row(child_id=child_id, lock=False) if not current_row: raise RuntimeError("current location not found after report") - return current_row + return {**current_row, "should_resolve_address": should_resolve_address} async def get_device_current_location(self, *, device_id: str, user_id: int) -> Mapping[str, Any]: access = await self.assert_parent_device_access(device_id=device_id, user_id=user_id) @@ -300,7 +312,10 @@ class LocationDAO(BaseDAO): battery_pct, device_time, server_time, - updated_at + updated_at, + address, + address_resolved_at, + address_resolve_status FROM child_location_current WHERE child_id = :child_id LIMIT 1{lock_clause} @@ -310,6 +325,55 @@ class LocationDAO(BaseDAO): ) return result.mappings().first() + @staticmethod + def _is_location_address_stale( + row: Mapping[str, Any], + *, + lat: float, + lng: float, + ) -> bool: + try: + previous_lat = float(row["lat"]) + previous_lng = float(row["lng"]) + except (TypeError, ValueError): + return True + return abs(previous_lat - lat) >= 0.0001 or abs(previous_lng - lng) >= 0.0001 + + async def update_current_location_address( + self, + *, + child_id: int, + device_id: str, + lat: float, + lng: float, + address: str | None, + status: int, + ) -> bool: + result = await self.execute( + text( + """ + UPDATE child_location_current + SET address = :address, + address_resolve_status = :status, + address_resolved_at = CURRENT_TIMESTAMP + WHERE child_id = :child_id + AND device_id = :device_id + AND ABS(lat - :lat) < 0.0000001 + AND ABS(lng - :lng) < 0.0000001 + """ + ), + { + "child_id": child_id, + "device_id": device_id, + "lat": lat, + "lng": lng, + "address": address, + "status": status, + }, + ) + await self.commit() + return bool(result.rowcount) + async def _next_primary_key(self, table_name: str) -> int | None: return None @@ -361,7 +425,10 @@ class LocationDAO(BaseDAO): battery_pct, device_time, server_time, - updated_at + updated_at, + address, + address_resolved_at, + address_resolve_status FROM child_location_current WHERE device_id = :device_id LIMIT 1 diff --git a/talkingq-url/banban/routers/devices.py b/talkingq-url/banban/routers/devices.py index 3b6b11f..522e7e2 100644 --- a/talkingq-url/banban/routers/devices.py +++ b/talkingq-url/banban/routers/devices.py @@ -224,6 +224,9 @@ def _row_to_current_location_response( device_time=row["device_time"], server_time=row["server_time"], updated_at=row["updated_at"], + address=row.get("address"), + address_resolved_at=row.get("address_resolved_at"), + address_resolve_status=int(row["address_resolve_status"]) if row.get("address_resolve_status") is not None else None, stale=stale, realtime=realtime, ) diff --git a/talkingq-url/banban/schemas/location.py b/talkingq-url/banban/schemas/location.py index 7d79ed7..9684fc8 100644 --- a/talkingq-url/banban/schemas/location.py +++ b/talkingq-url/banban/schemas/location.py @@ -31,6 +31,9 @@ class DeviceLocationPoint(BaseModel): battery_pct: int | None = None device_time: datetime server_time: datetime + address: str | None = None + address_resolved_at: datetime | None = None + address_resolve_status: int | None = None class DeviceLocationCurrentResponse(DeviceLocationPoint): diff --git a/talkingq-url/banban/service/amap_geocode.py b/talkingq-url/banban/service/amap_geocode.py new file mode 100644 index 0000000..05fea7e --- /dev/null +++ b/talkingq-url/banban/service/amap_geocode.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +from dataclasses import dataclass + +import httpx + +from config import settings + + +class AmapGeocodeError(RuntimeError): + pass + + +@dataclass(frozen=True) +class ReverseGeocodeResult: + address: str + + +class AmapGeocodeService: + async def reverse_geocode(self, *, lng: float, lat: float) -> ReverseGeocodeResult: + api_key = settings.amap_key.strip() + if not api_key: + raise AmapGeocodeError("Amap_key is not configured") + + params = { + "key": api_key, + "location": f"{lng:.7f},{lat:.7f}", + "extensions": "base", + "output": "JSON", + } + try: + async with httpx.AsyncClient(timeout=settings.amap_http_timeout_seconds) as client: + response = await client.get(settings.amap_reverse_geocode_url, params=params) + response.raise_for_status() + payload = response.json() + except httpx.HTTPError as exc: + raise AmapGeocodeError(f"amap request failed: {exc}") from exc + except ValueError as exc: + raise AmapGeocodeError("amap response is not valid json") from exc + + if str(payload.get("status")) != "1": + raise AmapGeocodeError( + f"amap reverse geocode failed: infocode={payload.get('infocode')} info={payload.get('info')}" + ) + + address = str((payload.get("regeocode") or {}).get("formatted_address") or "").strip() + if not address: + raise AmapGeocodeError("amap reverse geocode returned empty address") + + return ReverseGeocodeResult(address=address) + + +amap_geocode_service = AmapGeocodeService() diff --git a/talkingq-url/banban/service/location.py b/talkingq-url/banban/service/location.py index 7c94057..337dfc2 100644 --- a/talkingq-url/banban/service/location.py +++ b/talkingq-url/banban/service/location.py @@ -5,16 +5,25 @@ from typing import Any from services.database_service_base import DatabaseServiceBase from database.models import ChildLocationCurrent +from services.task_manager import task_manager +from utils.logger import session_logger try: from banban.dao.location import LocationDAO, ParentDeviceAccess from banban.schemas.location import DeviceLocationReportRequest + from banban.service.amap_geocode import amap_geocode_service from banban.service.im import im_service except ModuleNotFoundError: from banban.dao.location import LocationDAO, ParentDeviceAccess from banban.schemas.location import DeviceLocationReportRequest + from banban.service.amap_geocode import amap_geocode_service from banban.service.im import im_service +ADDRESS_RESOLVE_PENDING = 0 +ADDRESS_RESOLVE_SUCCESS = 1 +ADDRESS_RESOLVE_FAILED = 2 + + class LocationService(DatabaseServiceBase): def __init__(self): super().__init__(service_name="location_service") @@ -46,6 +55,7 @@ class LocationService(DatabaseServiceBase): child_id=device_identity.child_id, payload=payload, ) + await self._queue_location_address_resolution(current_row) return device_identity, current_row finally: await db_session.close() @@ -150,11 +160,88 @@ class LocationService(DatabaseServiceBase): battery_pct=battery_pct, device_time=device_time or datetime.now(), ) - return await dao.report_device_location( + current_row = await dao.report_device_location( device_id=device_id, child_id=int(binding_row["child_id"]), payload=payload, ) + await self._queue_location_address_resolution(current_row) + return current_row + finally: + await db_session.close() + + async def _queue_location_address_resolution(self, row: Mapping[str, Any] | None) -> None: + if not row or row.get("lat") is None or row.get("lng") is None: + return + if not row.get("should_resolve_address"): + return + await task_manager.create_task( + self.resolve_current_location_address( + child_id=int(row["child_id"]), + device_id=str(row["device_id"]), + lat=float(row["lat"]), + lng=float(row["lng"]), + ), + device_id=str(row["device_id"]), + task_type="location_address", + ) + + async def resolve_current_location_address( + self, + *, + child_id: int, + device_id: str, + lat: float, + lng: float, + ) -> None: + try: + result = await amap_geocode_service.reverse_geocode(lng=lng, lat=lat) + except Exception as exc: + session_logger.warning( + device_id, + "location_address", + f"当前位置地址解析失败: child_id={child_id}, error={exc}", + ) + await self.update_current_location_address( + child_id=child_id, + device_id=device_id, + lat=lat, + lng=lng, + address=None, + status=ADDRESS_RESOLVE_FAILED, + ) + return + + await self.update_current_location_address( + child_id=child_id, + device_id=device_id, + lat=lat, + lng=lng, + address=result.address, + status=ADDRESS_RESOLVE_SUCCESS, + ) + + async def update_current_location_address( + self, + *, + child_id: int, + device_id: str, + lat: float, + lng: float, + address: str | None, + status: int, + ) -> bool: + db_session = await self.get_session() + try: + dao = LocationDAO(db_session) + return await dao.update_current_location_address( + child_id=child_id, + device_id=device_id, + lat=lat, + lng=lng, + address=address, + status=status, + ) finally: await db_session.close() diff --git a/talkingq-url/config.py b/talkingq-url/config.py index f72ca2a..e5d5329 100644 --- a/talkingq-url/config.py +++ b/talkingq-url/config.py @@ -41,6 +41,13 @@ class Settings(BaseSettings): validation_alias="MINIMAX_BASE_URL", ) + amap_key: str = Field(default="", validation_alias="Amap_key") + amap_reverse_geocode_url: str = Field( + default="https://restapi.amap.com/v3/geocode/regeo", + validation_alias="AMAP_REVERSE_GEOCODE_URL", + ) + amap_http_timeout_seconds: float = Field(default=5.0, validation_alias="AMAP_HTTP_TIMEOUT_SECONDS") + aliyun_api_key: str = Field(default="", validation_alias="ALIYUN_API_KEY") aliyun_vocabulary_id: str = Field(default="", validation_alias="ALIYUN_VOCABULARY_ID") diff --git a/talkingq-url/database/init_db.py b/talkingq-url/database/init_db.py index 214f2dc..7294b6c 100644 --- a/talkingq-url/database/init_db.py +++ b/talkingq-url/database/init_db.py @@ -59,6 +59,37 @@ async def _ensure_schedule_suppressed_until_column(conn) -> None: ) +async def _ensure_columns(conn, *, table_name: str, columns: list[tuple[str, str]]) -> None: + for column_name, definition in columns: + result = await conn.execute( + text( + """ + SELECT COUNT(*) + FROM INFORMATION_SCHEMA.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = :table_name + AND COLUMN_NAME = :column_name + """ + ), + {"table_name": table_name, "column_name": column_name}, + ) + if int(result.scalar() or 0) > 0: + continue + await conn.execute(text(f"ALTER TABLE {table_name} ADD COLUMN {definition}")) + + +async def _ensure_child_location_address_columns(conn) -> None: + await _ensure_columns( + conn, + table_name="child_location_current", + columns=[ + ("address", "address VARCHAR(255) NULL AFTER updated_at"), + ("address_resolved_at", "address_resolved_at DATETIME NULL AFTER address"), + ("address_resolve_status", "address_resolve_status TINYINT NULL AFTER address_resolved_at"), + ], + ) + + async def _ensure_device_family_tables(conn) -> None: await conn.execute( text( @@ -173,6 +204,7 @@ async def init_db(): await conn.run_sync(Base.metadata.create_all) await _ensure_manual_sleep_mode_column(conn) await _ensure_schedule_suppressed_until_column(conn) + await _ensure_child_location_address_columns(conn) await _ensure_device_family_tables(conn) await engine.dispose() session_logger.info("system", "database", "数据库表已成功创建") diff --git a/talkingq-url/database/models.py b/talkingq-url/database/models.py index b34f6e5..9b33ec9 100644 --- a/talkingq-url/database/models.py +++ b/talkingq-url/database/models.py @@ -455,6 +455,9 @@ class ChildLocationCurrent(Base): updated_at: Mapped[datetime] = mapped_column( DateTime, default=datetime.utcnow, onupdate=datetime.utcnow ) + address: Mapped[Optional[str]] = mapped_column(String(255)) + address_resolved_at: Mapped[Optional[datetime]] = mapped_column(DateTime) + address_resolve_status: Mapped[Optional[int]] = mapped_column(Integer) class ChildLocationHistory(Base): diff --git a/talkingq-url/mysql/init/02-init.sql b/talkingq-url/mysql/init/02-init.sql index 48c5bb5..73afe7d 100644 --- a/talkingq-url/mysql/init/02-init.sql +++ b/talkingq-url/mysql/init/02-init.sql @@ -552,6 +552,9 @@ CREATE TABLE IF NOT EXISTS `child_location_current` ( `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, + `address` VARCHAR(255) NULL, + `address_resolved_at` DATETIME NULL, + `address_resolve_status` TINYINT NULL COMMENT '0=pending,1=success,2=failed', PRIMARY KEY (`child_id`), KEY `idx_child_loc_current_device` (`device_id`), KEY `idx_child_loc_current_server_time` (`server_time`),