gahusb
c6b969443f
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EqCYBhvTcdeCTUDX3RhWx9
84 lines
3.5 KiB
Python
84 lines
3.5 KiB
Python
"""매물 알림 텔레그램 포맷+전송 (본인+아내 각각). realestate-lab notify_new_listings 수신.
|
|
|
|
telegram_trade.py(매매알람)와 대칭 구조: send_raw 저수준 전송 + chat_id 리스트 순회.
|
|
"""
|
|
import logging
|
|
from typing import Any, Dict, List
|
|
|
|
from ..telegram.messaging import send_raw
|
|
from ..config import TELEGRAM_CHAT_ID, TELEGRAM_WIFE_CHAT_ID
|
|
|
|
logger = logging.getLogger("agent-office")
|
|
|
|
_TIER_EMOJI = {
|
|
"안전": "🟢 안전", "주의": "🟡 주의", "위험": "🔴 위험", "보류": "⚪ 보류(표본부족)",
|
|
"저평가": "🟢 저평가", "시세": "🟡 시세 수준", "고가": "🔴 고가",
|
|
}
|
|
|
|
|
|
def _manwon(v) -> str:
|
|
if not v:
|
|
return "-"
|
|
return f"{v / 10000:.1f}억" if v >= 10000 else f"{v:,}만"
|
|
|
|
|
|
def format_listing_alert(a: Dict[str, Any]) -> str:
|
|
cat = a.get("category") or ("매매" if a.get("deal_type") == "매매" else "임차")
|
|
if cat == "임차":
|
|
price = _manwon(a.get("deposit"))
|
|
tier = _TIER_EMOJI.get(a.get("safety_tier"), a.get("safety_tier") or "")
|
|
ratio = a.get("jeonse_ratio")
|
|
if ratio is not None:
|
|
judge = (f"{tier} (전세가율 {int(ratio * 100)}% · 시세 {_manwon(a.get('market_median'))}, "
|
|
f"실거래 {a.get('sample_size', 0)}건)")
|
|
else:
|
|
judge = f"{tier} (실거래 표본 부족 — 수동 확인)"
|
|
warn = "⚠️ 등기부 선순위 근저당 수동 확인 필수(인터넷등기소)"
|
|
else:
|
|
price = _manwon(a.get("sale_price"))
|
|
tier = _TIER_EMOJI.get(a.get("valuation_tier"), a.get("valuation_tier") or "")
|
|
ratio = a.get("price_ratio")
|
|
budget = " · 예산 내 ✅" if a.get("budget_ok") else ""
|
|
if ratio is not None:
|
|
judge = (f"{tier} (호가율 {int(ratio * 100)}% · 실거래 median {_manwon(a.get('market_median'))}, "
|
|
f"{a.get('sample_size', 0)}건){budget}")
|
|
else:
|
|
judge = f"{tier} (실거래 표본 부족 — 수동 확인)"
|
|
flags = a.get("regulation_flags") or []
|
|
warn = "⚠️ " + (" · ".join(flags) if flags else "비토허") + " · 등기부 선순위 수동 확인 필수"
|
|
|
|
area = f"전용 {a.get('area_exclusive')}㎡" if a.get("area_exclusive") else ""
|
|
return (
|
|
f"🏠 [{a.get('deal_type') or ''}] {a.get('complex_name') or ''} · {price} · {area} · {a.get('floor') or ''}\n"
|
|
f"{judge}\n"
|
|
f"📍 {a.get('dong') or ''} · {warn}\n"
|
|
f"🔗 {a.get('url') or ''}"
|
|
)
|
|
|
|
|
|
async def send_listing_alerts(listings: List[Dict[str, Any]]) -> dict:
|
|
"""매물마다 본인+아내 chat_id 각각으로 send_raw. 1건 이상 성공하면 delivered→sent_ids에 id 수집.
|
|
실패해도 나머지 계속 진행(per-send try/except)."""
|
|
sent = 0
|
|
sent_ids: List[Any] = []
|
|
all_ok = True
|
|
chat_ids = [c for c in (TELEGRAM_CHAT_ID, TELEGRAM_WIFE_CHAT_ID) if c]
|
|
for a in listings:
|
|
text = format_listing_alert(a)
|
|
delivered = False
|
|
for cid in chat_ids:
|
|
try:
|
|
r = await send_raw(text, chat_id=cid)
|
|
except Exception as e:
|
|
logger.warning(f"[telegram_realestate_listing] send failed (chat_id={cid}): {e}")
|
|
all_ok = False
|
|
continue
|
|
if r.get("ok"):
|
|
sent += 1
|
|
delivered = True
|
|
else:
|
|
all_ok = False
|
|
if delivered and a.get("id") is not None:
|
|
sent_ids.append(a["id"])
|
|
return {"sent": sent, "sent_ids": sent_ids, "ok": all_ok}
|