120 lines
3.5 KiB
Python
120 lines
3.5 KiB
Python
"""
|
|
Connects to a site object's external PostgreSQL database and syncs devices.
|
|
Uses short-lived psycopg3 connections (connect → query → disconnect).
|
|
SSH tunnel support is deferred to Этап 2 (asyncssh not yet available).
|
|
"""
|
|
|
|
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)
|
|
|
|
|
|
# Maps type strings from object DB to our device categories
|
|
_CATEGORY_MAP: dict[str, str] = {
|
|
"raspberry": "raspberry",
|
|
"rpi": "raspberry",
|
|
"raspberrypi": "raspberry",
|
|
"raspberry_pi": "raspberry",
|
|
"mikrotik": "mikrotik",
|
|
"routeros": "mikrotik",
|
|
"camera": "camera",
|
|
"cam": "camera",
|
|
"ipcam": "camera",
|
|
"embedded": "embedded",
|
|
"server": "server",
|
|
}
|
|
|
|
|
|
def _map_category(raw: str) -> str:
|
|
if raw is None:
|
|
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, obj.db_col_ip]):
|
|
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:
|
|
cols = [obj.db_col_ip]
|
|
if obj.db_col_type:
|
|
cols.append(obj.db_col_type)
|
|
|
|
col_list = ", ".join(f'"{c}"' for c in cols)
|
|
rows = await conn.execute(f'SELECT {col_list} 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:
|
|
ip = str(row[0]).strip() if row[0] else None
|
|
raw_type = str(row[1]).strip() if len(row) > 1 and row[1] else None
|
|
|
|
if not ip:
|
|
result.skipped += 1
|
|
continue
|
|
|
|
category = _map_category(raw_type)
|
|
|
|
# Look for existing device by object + external source
|
|
existing = await db.execute(
|
|
select(Device).where(
|
|
Device.object_id == obj.id,
|
|
Device.source == "auto_sync",
|
|
Device.ip == ip,
|
|
)
|
|
)
|
|
device = existing.scalar_one_or_none()
|
|
|
|
if device is not None:
|
|
changed = False
|
|
if device.category != category:
|
|
device.category = category
|
|
changed = True
|
|
result.updated += 1 if changed else 0
|
|
result.skipped += 0 if changed else 1
|
|
else:
|
|
db.add(Device(
|
|
object_id=obj.id,
|
|
ip=ip,
|
|
category=category,
|
|
source="auto_sync",
|
|
external_id=ip,
|
|
))
|
|
result.created += 1
|
|
|
|
return result
|