89 lines
3.5 KiB
Python
89 lines
3.5 KiB
Python
|
"""客户画像中期记忆：MySQL 事实源 + Redis Cache-Aside。"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
from decimal import Decimal
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from config.database.redis import client as redis_client
|
||
|
|
from repositories.fin_customer_profile import FinCustomerProfileRepo
|
||
|
|
|
||
|
|
|
||
|
|
class CustomerProfileMemory:
|
||
|
|
"""提供客户画像读取、缓存失效和刷新能力。"""
|
||
|
|
|
||
|
|
CACHE_TTL = 7 * 24 * 60 * 60
|
||
|
|
|
||
|
|
def __init__(self, *, redis=None, repository_factory=FinCustomerProfileRepo):
|
||
|
|
self.redis = redis or redis_client()
|
||
|
|
self.repository_factory = repository_factory
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def cache_key(customer_id: int) -> str:
|
||
|
|
"""生成客户画像缓存 Key。"""
|
||
|
|
return f"profile:{customer_id}"
|
||
|
|
|
||
|
|
async def get(self, db, customer_id: int) -> tuple[dict[str, Any] | None, list[str]]:
|
||
|
|
"""优先读取缓存，未命中后回源 MySQL，并返回 warnings。"""
|
||
|
|
warnings: list[str] = []
|
||
|
|
key = self.cache_key(customer_id)
|
||
|
|
try:
|
||
|
|
cached = await self.redis.get(key)
|
||
|
|
if cached:
|
||
|
|
return json.loads(cached), warnings
|
||
|
|
except Exception as exc:
|
||
|
|
warnings.append(f"profile_cache_read_failed:{type(exc).__name__}")
|
||
|
|
|
||
|
|
try:
|
||
|
|
profile = await self.repository_factory(db).get_by_customer_id(customer_id)
|
||
|
|
except Exception as exc:
|
||
|
|
warnings.append(f"profile_mysql_read_failed:{type(exc).__name__}")
|
||
|
|
return None, warnings
|
||
|
|
if profile is None:
|
||
|
|
return None, warnings
|
||
|
|
|
||
|
|
payload = self._to_dict(profile)
|
||
|
|
try:
|
||
|
|
await self.redis.set(key, json.dumps(payload, ensure_ascii=False), ex=self.CACHE_TTL)
|
||
|
|
except Exception as exc:
|
||
|
|
warnings.append(f"profile_cache_write_failed:{type(exc).__name__}")
|
||
|
|
return payload, warnings
|
||
|
|
|
||
|
|
async def invalidate(self, customer_id: int) -> list[str]:
|
||
|
|
"""删除客户画像缓存，确保更新后不会长期读取旧值。"""
|
||
|
|
try:
|
||
|
|
await self.redis.delete(self.cache_key(customer_id))
|
||
|
|
return []
|
||
|
|
except Exception as exc:
|
||
|
|
return [f"profile_cache_invalidate_failed:{type(exc).__name__}"]
|
||
|
|
|
||
|
|
async def refresh(self, db, customer_id: int) -> tuple[dict[str, Any] | None, list[str]]:
|
||
|
|
"""先删除缓存，再从 MySQL 读取并重新缓存画像。"""
|
||
|
|
warnings = await self.invalidate(customer_id)
|
||
|
|
profile, read_warnings = await self.get(db, customer_id)
|
||
|
|
return profile, warnings + read_warnings
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _to_dict(profile) -> dict[str, Any]:
|
||
|
|
"""将 ORM 画像转换为可安全写入 Redis 的字典。"""
|
||
|
|
data = {
|
||
|
|
"customer_id": profile.customer_id,
|
||
|
|
"risk_level": profile.risk_level,
|
||
|
|
"risk_score": profile.risk_score,
|
||
|
|
"investment_experience": profile.investment_experience,
|
||
|
|
"annual_income_range": profile.annual_income_range,
|
||
|
|
"total_assets": profile.total_assets,
|
||
|
|
"asset_allocation": profile.asset_allocation,
|
||
|
|
"product_preference": profile.product_preference,
|
||
|
|
"customer_level": profile.customer_level,
|
||
|
|
"confidence_score": profile.confidence_score,
|
||
|
|
"profile_version": profile.profile_version,
|
||
|
|
"update_time": profile.update_time.isoformat() if profile.update_time else None,
|
||
|
|
}
|
||
|
|
return json.loads(json.dumps(data, default=lambda value: str(value), ensure_ascii=False))
|
||
|
|
|
||
|
|
|
||
|
|
__all__ = ["CustomerProfileMemory"]
|
||
|
|
|