定位信息增加地址解析

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() {
                     <Text>📍</Text>
                   </View>
                   <View className='location-info'>
+                    <Text className='location-address'>{getLocationAddress(deviceLocation)}</Text>
                     <Text className='location-coordinate'>
                       {deviceLocation.lat.toFixed(6)}, {deviceLocation.lng.toFixed(6)}
                     </Text>

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

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`),

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

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,
     )

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):

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()

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()
 

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")
 

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", "数据库表已成功创建")

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):

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`),