Dish_Recognition/dao/base_sync.py

# coding=utf-8
"""从管理系统同步基地列表到 MySQL bases，并镜像到本地 SQLite。"""
import datetime

from dao.database import is_server_available, local_session_scope, server_session_scope
from dao.settings import DISH_MGMT_SYNC_ON_START
from models.Bases import BaseInfo
from service.mgmt_bases_api import MgmtApiError, fetch_bases, map_remote_base

_EXTRA_FIELDS = (
    'external_id', 'branch_name', 'province', 'city', 'district',
    'address', 'phone', 'description', 'longitude', 'latitude', 'sort_order',
)


def _apply_mapped(row, mapped, now=None):
    now = now or datetime.datetime.now()
    row.base_code = mapped['base_code']
    row.base_name = mapped['base_name']
    row.region = mapped.get('region')
    row.contact = mapped.get('contact')
    row.is_active = mapped.get('is_active', 1)
    row.last_heartbeat = now
    for key in _EXTRA_FIELDS:
        if hasattr(row, key):
            setattr(row, key, mapped.get(key))


def _find_existing(session, mapped):
    if mapped.get('external_id') is not None:
        row = (
            session.query(BaseInfo)
            .filter(BaseInfo.external_id == mapped['external_id'])
            .first()
        )
        if row:
            return row
    return (
        session.query(BaseInfo)
        .filter(BaseInfo.base_code == mapped['base_code'])
        .first()
    )


def upsert_bases_session(session, mapped_list):
    """在已有 session 中 upsert，返回 {created, updated, skipped}。"""
    now = datetime.datetime.now()
    created = updated = skipped = 0
    for mapped in mapped_list:
        if not mapped:
            skipped += 1
            continue
        row = _find_existing(session, mapped)
        if row:
            _apply_mapped(row, mapped, now=now)
            updated += 1
        else:
            row = BaseInfo(
                base_code=mapped['base_code'],
                base_name=mapped['base_name'],
                region=mapped.get('region'),
                contact=mapped.get('contact'),
                is_active=mapped.get('is_active', 1),
                register_time=now,
                last_heartbeat=now,
            )
            for key in _EXTRA_FIELDS:
                if hasattr(row, key):
                    setattr(row, key, mapped.get(key))
            session.add(row)
            created += 1
    return {'created': created, 'updated': updated, 'skipped': skipped}


def sync_bases_from_mgmt(force=False, url=None, api_key=None):
    """
    拉取管理系统基地 → 写入 MySQL → 镜像到本地 SQLite。
    MySQL 不可用时仅跳过（不抛致命错误）。
    """
    if not force and not DISH_MGMT_SYNC_ON_START:
        return {'ok': False, 'skipped': True, 'reason': 'DISH_MGMT_SYNC_ON_START=0'}

    if not is_server_available():
        return {'ok': False, 'skipped': True, 'reason': 'mysql_offline'}

    try:
        items = fetch_bases(url=url, api_key=api_key)
    except MgmtApiError as exc:
        print(f"[WARN] mgmt bases fetch failed: {exc}")
        return {'ok': False, 'error': str(exc)}

    mapped_list = []
    for item in items:
        mapped = map_remote_base(item)
        if mapped:
            mapped_list.append(mapped)

    try:
        with server_session_scope() as session:
            stats = upsert_bases_session(session, mapped_list)
            # 取出完整行供本地镜像（flush 后有 id）
            session.flush()
            rows = session.query(BaseInfo).all()
            mirror = [r.to_dict() for r in rows]
    except Exception as exc:
        print(f"[WARN] mgmt bases upsert mysql failed: {exc}")
        return {'ok': False, 'error': str(exc)}

    try:
        _mirror_bases_local(mirror)
    except Exception as exc:
        print(f"[WARN] mgmt bases mirror local failed: {exc}")

    print(
        f"[OK] mgmt bases synced: total={len(mapped_list)} "
        f"created={stats['created']} updated={stats['updated']} skipped={stats['skipped']}"
    )
    return {
        'ok': True,
        'fetched': len(items),
        'mapped': len(mapped_list),
        **stats,
    }


def _mirror_bases_local(rows):
    """把 MySQL bases 全量镜像到本地 SQLite（按 id 对齐，供 FK）。"""
    now = datetime.datetime.now()
    with local_session_scope() as session:
        for data in rows:
            bid = data.get('id')
            if bid is None:
                continue
            existing = session.get(BaseInfo, bid)
            payload = {
                'base_code': data['base_code'],
                'base_name': data['base_name'],
                'region': data.get('region'),
                'contact': data.get('contact'),
                'is_active': data.get('is_active', 1),
                'external_id': data.get('external_id'),
                'branch_name': data.get('branch_name'),
                'province': data.get('province'),
                'city': data.get('city'),
                'district': data.get('district'),
                'address': data.get('address'),
                'phone': data.get('phone'),
                'description': data.get('description'),
                'longitude': data.get('longitude'),
                'latitude': data.get('latitude'),
                'sort_order': data.get('sort_order') or 0,
            }
            if existing:
                for k, v in payload.items():
                    if hasattr(existing, k):
                        setattr(existing, k, v)
                existing.last_heartbeat = now
            else:
                row = BaseInfo(
                    id=bid,
                    register_time=now,
                    last_heartbeat=now,
                    **{k: v for k, v in payload.items() if hasattr(BaseInfo, k)},
                )
                session.add(row)