"""
商品资料 SQLite 数据库 + 商品记忆库

记忆库功能：
- 处理每步后自动学习商品数据（置信度+一致性加速）
- OCR 字段缺失时用记忆库补全 (conf > 50 直接采用)
- 价格异常检测：偏差 > 2倍触发补全，偏差 > 50% 记录预警
- 批量预加载 → 内存操作 → 批量写回，保障性能
"""

import os
import json
import sqlite3
from datetime import datetime
from typing import Dict, List, Optional, Tuple, Callable

import pandas as pd

from app.core.utils.log_utils import get_logger
from app.core.utils.file_utils import smart_read_excel
from app.core.handlers.column_mapper import ColumnMapper

logger = get_logger(__name__)


class ProductDatabase:
    """商品资料 SQLite 数据库 + 商品记忆库"""

    SCHEMA = """
    CREATE TABLE IF NOT EXISTS products (
        barcode TEXT PRIMARY KEY,
        name TEXT DEFAULT '',
        price REAL DEFAULT 0.0,
        unit TEXT DEFAULT '',
        updated_at TEXT,
        specification TEXT DEFAULT '',
        source TEXT DEFAULT 'template',
        confidence INTEGER DEFAULT 0,
        usage_count INTEGER DEFAULT 0,
        last_seen TEXT,
        avg_price REAL DEFAULT 0.0,
        min_price REAL DEFAULT 0.0,
        max_price REAL DEFAULT 0.0,
        price_count INTEGER DEFAULT 0
    );
    CREATE TABLE IF NOT EXISTS missing_barcodes (
        barcode     TEXT PRIMARY KEY,
        name        TEXT DEFAULT '',
        last_seen   TEXT,
        source_file TEXT,
        count       INTEGER DEFAULT 1
    );
    """

    _NEW_COLUMNS = {
        'specification': "TEXT DEFAULT ''",
        'source': "TEXT DEFAULT 'template'",
        'confidence': 'INTEGER DEFAULT 0',
        'usage_count': 'INTEGER DEFAULT 0',
        'last_seen': 'TEXT',
        'avg_price': 'REAL DEFAULT 0.0',
        'min_price': 'REAL DEFAULT 0.0',
        'max_price': 'REAL DEFAULT 0.0',
        'price_count': 'INTEGER DEFAULT 0',
    }

    def __init__(self, db_path: str, excel_source: str):
        self.db_path = db_path
        self.excel_source = excel_source
        self._ensure_db()

    def _connect(self) -> sqlite3.Connection:
        return sqlite3.connect(self.db_path)

    def _ensure_db(self):
        if os.path.exists(self.db_path):
            self._migrate_schema()
            return
        if not os.path.exists(self.excel_source):
            logger.warning(f"商品资料 Excel 不存在: {self.excel_source}")
            self._create_empty_db()
            return
        logger.info(f"首次运行，从 Excel 导入商品资料: {self.excel_source}")
        os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
        self._create_empty_db()
        count = self.import_from_excel(self.excel_source)
        logger.info(f"商品资料导入完成: {count} 条记录")

    def _create_empty_db(self):
        conn = self._connect()
        try:
            conn.executescript(self.SCHEMA)
            conn.commit()
        finally:
            conn.close()

    def _migrate_schema(self):
        conn = self._connect()
        try:
            # 迁移 products 表
            cursor = conn.execute("PRAGMA table_info(products)")
            existing_cols = {row[1] for row in cursor.fetchall()}
            for col_name, col_type in self._NEW_COLUMNS.items():
                if col_name not in existing_cols:
                    conn.execute(f"ALTER TABLE products ADD COLUMN {col_name} {col_type}")
                    logger.info(f"数据库迁移: 添加列 {col_name}")
            
            # 确保 missing_barcodes 表存在
            conn.execute("""
                CREATE TABLE IF NOT EXISTS missing_barcodes (
                    barcode     TEXT PRIMARY KEY,
                    name        TEXT DEFAULT '',
                    last_seen   TEXT,
                    source_file TEXT,
                    count       INTEGER DEFAULT 1
                )
            """)
            conn.commit()
        finally:
            conn.close()

    # ══════════════════════════════════════════════════════════════
    # 缺失条码记录
    # ══════════════════════════════════════════════════════════════

    def record_missing_barcode(self, barcode: str, name: str = '', source_file: str = ''):
        """记录缺失条码。"""
        barcode = str(barcode).strip()
        if not barcode:
            return
        now = datetime.now().isoformat(timespec='seconds')
        conn = self._connect()
        try:
            conn.execute("""
                INSERT INTO missing_barcodes (barcode, name, last_seen, source_file, count)
                VALUES (?, ?, ?, ?, 1)
                ON CONFLICT(barcode) DO UPDATE SET
                    name = CASE WHEN excluded.name != '' THEN excluded.name ELSE name END,
                    last_seen = excluded.last_seen,
                    source_file = excluded.source_file,
                    count = count + 1
            """, (barcode, name, now, os.path.basename(source_file)))
            conn.commit()
            logger.info(f"已记录缺失条码: {barcode} ({name})")
        except Exception as e:
            logger.error(f"记录缺失条码失败: {e}")
        finally:
            conn.close()

    def get_missing_barcodes(self, limit: int = 100) -> List[Dict]:
        """获取缺失条码列表。"""
        conn = self._connect()
        conn.row_factory = sqlite3.Row
        try:
            rows = conn.execute(
                "SELECT * FROM missing_barcodes ORDER BY last_seen DESC LIMIT ?",
                (limit,)).fetchall()
            return [dict(r) for r in rows]
        finally:
            conn.close()

    # ══════════════════════════════════════════════════════════════
    # 导入
    # ══════════════════════════════════════════════════════════════

    def import_from_excel(self, excel_path: str) -> int:
        df = smart_read_excel(excel_path)
        if df is None or df.empty:
            return 0
        barcode_col = ColumnMapper.find_column(list(df.columns), 'barcode')
        if not barcode_col:
            return 0
        price_col = ColumnMapper.find_column(list(df.columns), 'unit_price')
        if not price_col:
            for col in df.columns:
                if '进货价' in str(col).strip():
                    price_col = col
                    break
        name_col = ColumnMapper.find_column(list(df.columns), 'name')
        unit_col = ColumnMapper.find_column(list(df.columns), 'unit')
        spec_col = ColumnMapper.find_column(list(df.columns), 'specification')

        now = datetime.now().isoformat()
        rows = []
        for _, row in df.iterrows():
            barcode = str(row.get(barcode_col, '')).strip()
            if not barcode or barcode == 'nan':
                continue
            price = 0.0
            if price_col:
                try:
                    p = row.get(price_col)
                    if p is not None and str(p).strip() not in ('', 'nan', 'None'):
                        price = float(p)
                except (ValueError, TypeError):
                    pass
            name = str(row.get(name_col, '')).strip() if name_col else ''
            if name == 'nan': name = ''
            unit = str(row.get(unit_col, '')).strip() if unit_col else ''
            if unit == 'nan': unit = ''
            spec = str(row.get(spec_col, '')).strip() if spec_col else ''
            if spec == 'nan': spec = ''
            # template 源置信度 50
            rows.append((barcode, name, price, unit, now, spec, 'template', 50, 0, now,
                         price, price, price, 1 if price > 0 else 0))

        if not rows:
            return 0
        conn = self._connect()
        try:
            conn.executemany(
                "INSERT OR REPLACE INTO products "
                "(barcode, name, price, unit, updated_at, specification, source, confidence, "
                "usage_count, last_seen, avg_price, min_price, max_price, price_count) "
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                rows)
            conn.commit()
        finally:
            conn.close()
        return len(rows)

    def reimport(self) -> int:
        conn = self._connect()
        try:
            conn.execute("DELETE FROM products")
            conn.commit()
        finally:
            conn.close()
        return self.import_from_excel(self.excel_source)

    # ══════════════════════════════════════════════════════════════
    # 查询
    # ══════════════════════════════════════════════════════════════

    def get_price(self, barcode: str) -> Optional[float]:
        conn = self._connect()
        try:
            row = conn.execute("SELECT avg_price FROM products WHERE barcode=?",
                               (str(barcode).strip(),)).fetchone()
            if not row:
                return None
            return row[0] if row[0] is not None else 0.0
        finally:
            conn.close()

    def get_prices(self, barcodes: List[str], column: str = 'price') -> Dict[str, float]:
        """批量查进货价。

        Args:
            barcodes: 条码列表
            column: 价字段，默认 'price'（Excel 导入的进货价，按商家主单位），
                    可选 'avg_price'（OCR 学来的历史均价，调试用）。
        """
        if not barcodes:
            return {}
        valid = {'price', 'avg_price', 'min_price', 'max_price'}
        if column not in valid:
            raise ValueError(f"column must be one of {valid}")
        conn = self._connect()
        try:
            placeholders = ','.join('?' * len(barcodes))
            rows = conn.execute(
                f"SELECT barcode, {column} FROM products WHERE barcode IN ({placeholders})",
                [str(b).strip() for b in barcodes]).fetchall()
            return {r[0]: r[1] for r in rows if r[1]}
        finally:
            conn.close()

    def count(self) -> int:
        conn = self._connect()
        try:
            return conn.execute("SELECT COUNT(*) FROM products").fetchone()[0]
        finally:
            conn.close()

    def get_memory(self, barcode: str) -> Optional[Dict]:
        conn = self._connect()
        conn.row_factory = sqlite3.Row
        try:
            row = conn.execute("SELECT * FROM products WHERE barcode=?",
                               (str(barcode).strip(),)).fetchone()
            return dict(row) if row else None
        finally:
            conn.close()

    def get_memories(self, barcodes: List[str]) -> Dict[str, Dict]:
        if not barcodes:
            return {}
        conn = self._connect()
        conn.row_factory = sqlite3.Row
        try:
            placeholders = ','.join('?' * len(barcodes))
            rows = conn.execute(
                f"SELECT * FROM products WHERE barcode IN ({placeholders})",
                [str(b).strip() for b in barcodes]).fetchall()
            return {r['barcode']: dict(r) for r in rows}
        finally:
            conn.close()

    def get_all_memories(self) -> List[Dict]:
        conn = self._connect()
        conn.row_factory = sqlite3.Row
        try:
            return [dict(row) for row in
                    conn.execute("SELECT * FROM products ORDER BY usage_count DESC, barcode").fetchall()]
        finally:
            conn.close()

    # ══════════════════════════════════════════════════════════════
    # 批量预加载 — 性能核心
    # ══════════════════════════════════════════════════════════════

    def load_batch(self, barcodes: List[str]) -> Dict[str, Dict]:
        """批量预加载条码记忆到 dict — 单次 SQL，后续纯内存操作"""
        if not barcodes:
            return {}
        conn = self._connect()
        conn.row_factory = sqlite3.Row
        try:
            placeholders = ','.join('?' * len(barcodes))
            rows = conn.execute(
                f"SELECT * FROM products WHERE barcode IN ({placeholders})",
                [str(b).strip() for b in barcodes]).fetchall()
            return {r['barcode']: dict(r) for r in rows}
        finally:
            conn.close()

    # ══════════════════════════════════════════════════════════════
    # 学习逻辑 — 一致性加速 + 价格区间
    # ══════════════════════════════════════════════════════════════

    def learn_from_product(self, product: Dict, source: str = 'ocr',
                           memory: Dict[str, Dict] = None,
                           add_log: Callable = None) -> Optional[str]:
        """
        从处理结果中学习，返回日志字符串。
        memory: 可选的预加载批量内存，传入则零 DB 查询。
        """
        barcode = str(product.get('barcode', '')).strip()
        if not barcode:
            return None

        name = str(product.get('name', ''))
        spec = str(product.get('specification', ''))
        unit = str(product.get('unit', ''))
        price = float(product.get('price', 0))
        now = datetime.now().isoformat()

        # 查现有记录（优先从内存查）
        if memory is not None and barcode in memory:
            row = memory[barcode]
            old_name = row.get('name', '')
            old_spec = row.get('specification', '')
            old_unit = row.get('unit', '')
            old_conf = row.get('confidence', 0)
            old_count = row.get('usage_count', 0)
            old_avg = row.get('avg_price', 0) or 0
            old_min = row.get('min_price') or price
            old_max = row.get('max_price') or price
            pc = row.get('price_count', 0) or 0
            exists = True
        else:
            conn = self._connect()
            try:
                cursor = conn.execute(
                    "SELECT name, specification, unit, confidence, usage_count, "
                    "avg_price, min_price, max_price, price_count FROM products WHERE barcode=?",
                    (barcode,)).fetchone()
            finally:
                conn.close()
            if cursor is None:
                exists = False
            else:
                old_name, old_spec, old_unit, old_conf, old_count, old_avg, old_min, old_max, pc = cursor
                old_avg = old_avg or 0
                pc = pc or 0
                old_min = old_min if old_min is not None else price
                old_max = old_max if old_max is not None else price
                exists = True

        new_count = old_count + 1 if exists else 1

        # ── 置信度 ──
        if source == 'user_confirmed':
            new_conf = 90
        elif source == 'template':
            new_conf = 50
        elif exists and old_conf < 50:
            # 一致性加速
            spec_match = bool(spec and old_spec and spec == old_spec)
            unit_match = bool(unit and old_unit and unit == old_unit)
            if spec_match and unit_match:
                boost = 10
            elif unit_match:
                boost = 5
            else:
                boost = 3
            new_conf = min(50, old_conf + boost)
        elif exists:
            new_conf = old_conf  # > 50 稳定不变
        else:
            new_conf = 10  # 新 OCR 记录

        # ── 价格区间 ──
        if price > 0:
            new_pc = (pc if exists else 0) + 1
            new_avg = ((old_avg * (new_pc - 1)) + price) / new_pc if exists else price
            new_min = min(old_min, price) if exists else price
            new_max = max(old_max, price) if exists else price
        else:
            new_avg = old_avg if exists else 0
            new_min = old_min if exists else 0
            new_max = old_max if exists else 0
            new_pc = pc if exists else 0

        # ── 写入 ──
        conn = self._connect()
        try:
            if not exists:
                conn.execute(
                    "INSERT INTO products (barcode, name, specification, unit, price, "
                    "source, confidence, usage_count, last_seen, updated_at, "
                    "avg_price, min_price, max_price, price_count) "
                    "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                    (barcode, name, spec, unit, price, source, new_conf, 1, now, now,
                     new_avg, new_min, new_max, new_pc))
                log = f"记忆库新增: {barcode} {name} 源={source} 可信度={new_conf}"
            else:
                # 高可信度源全字段覆盖；低可信度仅填空
                if source in ('template', 'user_confirmed') or new_conf > 50:
                    # OCR 来源不覆盖 price（price 是商品资料 Excel 导入的进货价，应作为基准不被 OCR 学习的"按箱单价"覆盖）
                    if source == 'ocr':
                        conn.execute(
                            "UPDATE products SET name=?, specification=?, unit=?, "
                            "source=?, confidence=?, usage_count=?, last_seen=?, updated_at=?, "
                            "avg_price=?, min_price=?, max_price=?, price_count=? WHERE barcode=?",
                            (name or old_name, spec or old_spec, unit or old_unit,
                             source, new_conf, new_count, now, now,
                             new_avg, new_min, new_max, new_pc, barcode))
                    else:
                        conn.execute(
                            "UPDATE products SET name=?, specification=?, unit=?, price=?, "
                            "source=?, confidence=?, usage_count=?, last_seen=?, updated_at=?, "
                            "avg_price=?, min_price=?, max_price=?, price_count=? WHERE barcode=?",
                            (name or old_name, spec or old_spec, unit or old_unit, price,
                             source, new_conf, new_count, now, now,
                             new_avg, new_min, new_max, new_pc, barcode))
                else:
                    conn.execute(
                        "UPDATE products SET "
                        "name=CASE WHEN name='' THEN ? ELSE name END, "
                        "specification=CASE WHEN specification='' THEN ? ELSE specification END, "
                        "unit=CASE WHEN unit='' THEN ? ELSE unit END, "
                        "source=?, confidence=?, usage_count=?, last_seen=?, updated_at=?, "
                        "avg_price=?, min_price=?, max_price=?, price_count=? WHERE barcode=?",
                        (name, spec, unit, source, new_conf, new_count, now, now,
                         new_avg, new_min, new_max, new_pc, barcode))
                log = f"记忆库更新: {barcode} 可信度{old_conf if exists else 0}→{new_conf}"
                if price > 0:
                    log += f" 均价{new_avg:.4f}({new_pc}次)"

            conn.commit()

            # 更新内存 dict（如果传入了）
            if memory is not None and barcode in memory:
                memory[barcode].update({
                    'confidence': new_conf, 'usage_count': new_count,
                    'avg_price': new_avg, 'min_price': new_min,
                    'max_price': new_max, 'price_count': new_pc,
                    'name': name or old_name,
                    'specification': spec or old_spec,
                    'unit': unit or old_unit,
                })

            if add_log:
                add_log(log)
            return log
        finally:
            conn.close()

    def learn_from_products(self, products: List[Dict], source: str = 'ocr',
                            add_log: Callable = None) -> int:
        """批量学习 — 先批量预加载，再逐条处理，返回更新条数"""
        barcodes = [str(p.get('barcode', '')) for p in products if p.get('barcode')]
        memory = self.load_batch(barcodes)
        count = 0
        for p in products:
            try:
                result = self.learn_from_product(p, source, memory=memory, add_log=add_log)
                if result:
                    count += 1
            except Exception as e:
                logger.warning(f"学习商品记忆失败: {e}")
        return count

    # ══════════════════════════════════════════════════════════════
    # 记忆辅助 — OCR 补全
    # ══════════════════════════════════════════════════════════════

    def _price_anomaly(self, product: Dict, mem: Dict) -> bool:
        """价格异常：> 2倍偏差"""
        price = float(product.get('price', 0))
        avg = mem.get('avg_price', 0)
        if not price or not avg:
            return False
        return price > avg * 2 or price < avg * 0.5

    def fill_from_memory(self, barcode: str, ocr_result: Dict,
                         memory: Dict[str, Dict] = None) -> Tuple[Dict, str]:
        """用记忆库补全 OCR 缺失字段。返回 (补全后的dict, 日志字符串)"""
        if memory:
            mem = memory.get(barcode)
        else:
            mem = self.get_memory(barcode)

        if not mem or mem.get('confidence', 0) < 10:
            return ocr_result, ""

        logs = []
        result = dict(ocr_result)
        conf = mem.get('confidence', 0)

        has_spec = result.get('specification')
        has_unit = result.get('unit')
        price = float(result.get('price', 0))

        if conf > 50 and not has_spec and mem.get('specification'):
            result['specification'] = mem['specification']
            logs.append(f"规格补全(可信{conf}): {barcode} → {mem['specification']}")
        elif not has_spec and mem.get('specification') and self._price_anomaly(result, mem):
            result['specification'] = mem['specification']
            logs.append(f"价格异常→规格补全: {barcode} 本次{price:.2f} vs 均价{mem['avg_price']:.2f} → {mem['specification']}")

        if conf > 50 and not has_unit and mem.get('unit'):
            result['unit'] = mem['unit']
            logs.append(f"单位补全(可信{conf}): {barcode} → {mem['unit']}")
        elif not has_unit and mem.get('unit') and self._price_anomaly(result, mem):
            result['unit'] = mem['unit']
            logs.append(f"价格异常→单位补全: {barcode} → {mem['unit']}")

        return result, "; ".join(logs)

    def price_warning(self, barcode: str, price: float,
                      memory: Dict[str, Dict] = None) -> Optional[str]:
        """价格预警。> 50% 偏差告警"""
        if memory:
            mem = memory.get(barcode)
        else:
            mem = self.get_memory(barcode)
        if not mem or not mem.get('avg_price'):
            return None
        avg = mem['avg_price']
        min_p = mem.get('min_price', avg)
        max_p = mem.get('max_price', avg)
        pc = mem.get('price_count', 0)
        if price > avg * 1.5 or price < avg * 0.5:
            return (f"单价预警: {barcode} 本次{price:.4f}元 vs "
                    f"历史均价{avg:.4f} (范围{min_p:.4f}~{max_p:.4f}, {pc}次)")
        return None

    # ══════════════════════════════════════════════════════════════
    # 手动编辑
    # ══════════════════════════════════════════════════════════════

    def update_memory(self, barcode: str, fields: Dict) -> bool:
        barcode = str(barcode).strip()
        if not barcode:
            return False
        allowed = {'name', 'specification', 'unit', 'price', 'confidence'}
        updates = {k: v for k, v in fields.items() if k in allowed}
        if not updates:
            return False
        now = datetime.now().isoformat()
        set_clause = ', '.join(f"{k}=?" for k in updates)
        values = list(updates.values())
        extra_sql = ", source='user_confirmed'"
        if 'confidence' not in updates:
            extra_sql += ", confidence=90"
        conn = self._connect()
        try:
            conn.execute(
                f"UPDATE products SET {set_clause}{extra_sql}, updated_at=? WHERE barcode=?",
                values + [now, barcode])
            conn.commit()
            return conn.total_changes > 0
        finally:
            conn.close()

    def delete_memory(self, barcode: str) -> bool:
        conn = self._connect()
        try:
            conn.execute("DELETE FROM products WHERE barcode=?", (str(barcode).strip(),))
            conn.commit()
            return conn.total_changes > 0
        finally:
            conn.close()

    # ══════════════════════════════════════════════════════════════
    # 云端同步
    # ══════════════════════════════════════════════════════════════

    def export_for_sync(self) -> Dict:
        conn = self._connect()
        try:
            cursor = conn.execute(
                "SELECT barcode, name, specification, unit, price, source, "
                "confidence, usage_count, last_seen, avg_price, min_price, max_price, price_count "
                "FROM products")
            result = {}
            for row in cursor.fetchall():
                result[row[0]] = {
                    'name': row[1], 'specification': row[2], 'unit': row[3],
                    'price': row[4], 'source': row[5], 'confidence': row[6],
                    'usage_count': row[7], 'last_seen': row[8],
                    'avg_price': row[9], 'min_price': row[10],
                    'max_price': row[11], 'price_count': row[12],
                }
            return result
        finally:
            conn.close()

    def import_from_sync(self, data: Dict) -> int:
        now = datetime.now().isoformat()
        count = 0
        conn = self._connect()
        try:
            for barcode, info in data.items():
                barcode = str(barcode).strip()
                if not barcode:
                    continue
                name = str(info.get('name', ''))
                spec = str(info.get('specification', ''))
                unit = str(info.get('unit', ''))
                price = float(info.get('price', 0))
                remote_source = str(info.get('source', 'ocr'))
                remote_conf = int(info.get('confidence', 50))
                remote_count = int(info.get('usage_count', 1))
                remote_seen = str(info.get('last_seen', now))
                remote_avg = float(info.get('avg_price', price))
                remote_min = float(info.get('min_price', price))
                remote_max = float(info.get('max_price', price))
                remote_pc = int(info.get('price_count', 1))

                row = conn.execute("SELECT confidence FROM products WHERE barcode=?",
                                   (barcode,)).fetchone()
                if row is None:
                    conn.execute(
                        "INSERT INTO products (barcode, name, specification, unit, price, "
                        "source, confidence, usage_count, last_seen, updated_at, "
                        "avg_price, min_price, max_price, price_count) "
                        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                        (barcode, name, spec, unit, price, remote_source, remote_conf,
                         remote_count, remote_seen, now,
                         remote_avg, remote_min, remote_max, remote_pc))
                    count += 1
                else:
                    local_conf = row[0]
                    if remote_conf > local_conf:
                        conn.execute(
                            "UPDATE products SET name=?, specification=?, unit=?, price=?, "
                            "source=?, confidence=?, usage_count=?, last_seen=?, updated_at=?, "
                            "avg_price=?, min_price=?, max_price=?, price_count=? WHERE barcode=?",
                            (name, spec, unit, price, remote_source, remote_conf,
                             remote_count, remote_seen, now,
                             remote_avg, remote_min, remote_max, remote_pc, barcode))
                        count += 1
                    elif remote_conf == local_conf:
                        conn.execute(
                            "UPDATE products SET "
                            "name=CASE WHEN name='' THEN ? ELSE name END, "
                            "specification=CASE WHEN specification='' THEN ? ELSE specification END, "
                            "unit=CASE WHEN unit='' THEN ? ELSE unit END, "
                            "usage_count=MAX(usage_count, ?), updated_at=? WHERE barcode=?",
                            (name, spec, unit, remote_count, now, barcode))
                        count += 1
            conn.commit()
        finally:
            conn.close()
        return count

    def _export_memory_json(self, json_path=None):
        """导出记忆库为 JSON（兼容旧代码调用）"""
        import os as _os
        if json_path is None:
            json_path = _os.path.join(_os.path.dirname(self.db_path), 'product_memory.json')
        data = self.export_for_sync()
        _os.makedirs(_os.path.dirname(json_path), exist_ok=True)
        with open(json_path, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        return json_path
