133 lines
3.9 KiB
Python
133 lines
3.9 KiB
Python
"""
|
|
Connects to a site object's external PostgreSQL database and syncs devices.
|
|
Uses short-lived psycopg3 connections (connect → query → disconnect).
|
|
|
|
Expected table schema (racks):
|
|
id TEXT — external identifier, e.g. "lane-1-1"
|
|
name TEXT — short label, e.g. "01"
|
|
type TEXT — "rack.lane" | "rack.payment" | ...
|
|
data JSONB — contains "host" (IP address) and other device-specific fields
|
|
|
|
SSH tunnel support is deferred to Этап 2.
|
|
"""
|
|
|
|
from dataclasses import dataclass, field
|
|
|
|
import psycopg
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.device import Device
|
|
from app.models.object import Object
|
|
from app.services.crypto import decrypt
|
|
|
|
|
|
@dataclass
|
|
class SyncResult:
|
|
created: int = 0
|
|
updated: int = 0
|
|
skipped: int = 0
|
|
errors: list[str] = field(default_factory=list)
|
|
|
|
|
|
# rack.* type → our internal category
|
|
_CATEGORY_MAP: dict[str, str] = {
|
|
"rack.lane": "embedded",
|
|
"rack.payment": "other",
|
|
"rack.server": "server",
|
|
"rack.camera": "camera",
|
|
"raspberry": "raspberry",
|
|
"rpi": "raspberry",
|
|
"mikrotik": "mikrotik",
|
|
"camera": "camera",
|
|
"server": "server",
|
|
"embedded": "embedded",
|
|
}
|
|
|
|
|
|
def _map_category(raw: str | None) -> str:
|
|
if not raw:
|
|
return "other"
|
|
return _CATEGORY_MAP.get(raw.strip().lower(), "other")
|
|
|
|
|
|
async def sync_from_object_db(db: AsyncSession, obj: Object) -> SyncResult:
|
|
result = SyncResult()
|
|
|
|
if not all([obj.db_host, obj.db_name, obj.db_user, obj.db_pass_enc, obj.db_table]):
|
|
result.errors.append("Object DB connection is not fully configured")
|
|
return result
|
|
|
|
if obj.db_via_tunnel:
|
|
result.errors.append("SSH tunnel sync is not yet supported (coming in Этап 2)")
|
|
return result
|
|
|
|
try:
|
|
password = decrypt(obj.db_pass_enc)
|
|
except Exception:
|
|
result.errors.append("Failed to decrypt object DB password")
|
|
return result
|
|
|
|
dsn = (
|
|
f"host={obj.db_host} port={obj.db_port} dbname={obj.db_name} "
|
|
f"user={obj.db_user} password={password} connect_timeout=10"
|
|
)
|
|
|
|
try:
|
|
async with await psycopg.AsyncConnection.connect(dsn) as conn:
|
|
# Pull id, name, type and host from the JSONB data column
|
|
rows = await conn.execute(
|
|
f'SELECT id, name, type, data->>\'host\' AS host FROM "{obj.db_table}"'
|
|
)
|
|
remote_rows = await rows.fetchall()
|
|
except Exception as exc:
|
|
result.errors.append(f"Cannot connect to object DB: {exc}")
|
|
return result
|
|
|
|
for row in remote_rows:
|
|
external_id, name, rack_type, host = row[0], row[1], row[2], row[3]
|
|
|
|
ip = str(host).strip() if host else None
|
|
if not ip:
|
|
result.skipped += 1
|
|
continue
|
|
|
|
category = _map_category(rack_type)
|
|
hostname = str(name).strip() if name else None
|
|
|
|
existing = await db.execute(
|
|
select(Device).where(
|
|
Device.object_id == obj.id,
|
|
Device.external_id == str(external_id),
|
|
)
|
|
)
|
|
device = existing.scalar_one_or_none()
|
|
|
|
if device is not None:
|
|
changed = False
|
|
if device.ip != ip:
|
|
device.ip = ip
|
|
changed = True
|
|
if device.hostname != hostname:
|
|
device.hostname = hostname
|
|
changed = True
|
|
if device.category != category:
|
|
device.category = category
|
|
changed = True
|
|
if changed:
|
|
result.updated += 1
|
|
else:
|
|
result.skipped += 1
|
|
else:
|
|
db.add(Device(
|
|
object_id=obj.id,
|
|
ip=ip,
|
|
hostname=hostname,
|
|
category=category,
|
|
source="auto_sync",
|
|
external_id=str(external_id),
|
|
role=rack_type or "unknown",
|
|
))
|
|
result.created += 1
|
|
|
|
return result
|