from __future__ import annotations import logging from datetime import datetime import math import random logger = logging.getLogger(__name__) from sqlalchemy import or_, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import aliased from sqlalchemy.ext.asyncio import AsyncSession from app.models.match import Match, MatchLike from app.models.user import User from app.services.admin_service import get_system_config_value from app.utils.media import sanitize_user_media_payload WEIGHTS = { "age": 0.20, "hobbies": 0.18, "values": 0.18, "education": 0.14, "activity": 0.12, "location": 0.10, "height": 0.08, } def _birth_year_range(birth_year: int | None) -> str | None: if not birth_year: return None start = birth_year - 2 end = birth_year + 2 return f"{start}-{end}" def _level1_info(user: User, *, score: float | None = None, reasons: list[str] | None = None) -> dict: personality_tags = _safe_personality_tags(user.personality_tags) return sanitize_user_media_payload({ "user_id": user.id, "nickname": user.nickname, "real_name": user.real_name, "birth_year": user.birth_year, "birth_year_range": _birth_year_range(user.birth_year), "city": user.city, "education": user.education, "personality_tags": personality_tags, "hobbies": user.hobbies if isinstance(user.hobbies, list) else [], "job_industry": user.job_industry, "job_company": user.job_company, "self_intro": user.self_intro, "profile_images": user.profile_images if isinstance(user.profile_images, list) else [], "avatar_url": user.avatar_url, "avatar_blur_url": user.avatar_blur_url, "match_score": score, "match_reasons": reasons or [], }) def _level2_info(user: User) -> dict: return sanitize_user_media_payload({ "nickname": user.nickname, "avatar_url": user.avatar_url, "profile_images": user.profile_images if isinstance(user.profile_images, list) else [], "education": user.education, "hobbies": list(_safe_hobbies(user.hobbies)), }) def _level3_info(user: User) -> dict: return sanitize_user_media_payload({ "user_id": user.id, "nickname": user.nickname, "real_name": user.real_name, "avatar_url": user.avatar_url, "profile_images": user.profile_images if isinstance(user.profile_images, list) else [], "education": user.education, "birth_year": user.birth_year, "city": user.city, "hobbies": list(_safe_hobbies(user.hobbies)), "job_industry": user.job_industry, "job_company": user.job_company, "self_intro": user.self_intro, "income_range": user.income_range, }) def _age_score(user_a: User, user_b: User) -> float: if not user_b.birth_year: return 50.0 age_b = datetime.now().year - user_b.birth_year if user_a.prefer_age_min and age_b < user_a.prefer_age_min: return 0.0 if user_a.prefer_age_max and age_b > user_a.prefer_age_max: return 0.0 if user_a.prefer_age_min and user_a.prefer_age_max: mid = (user_a.prefer_age_min + user_a.prefer_age_max) / 2 diff = abs(age_b - mid) half_range = max((user_a.prefer_age_max - user_a.prefer_age_min) / 2, 1) return max(0.0, 100 - (diff / half_range) * 50) return 80.0 def _hobby_score(user_a: User, user_b: User) -> tuple[float, list[str]]: hobbies_a = _safe_hobbies(user_a.hobbies) hobbies_b = _safe_hobbies(user_b.hobbies) if not hobbies_a or not hobbies_b: return 30.0, [] intersection = hobbies_a & hobbies_b union = hobbies_a | hobbies_b score = len(intersection) / len(union) * 100 if union else 0.0 return round(score, 1), list(intersection) def _value_score(user_a: User, user_b: User) -> tuple[float, str | None]: answers_a = _safe_dict(getattr(user_a, "value_answers", None)) answers_b = _safe_dict(getattr(user_b, "value_answers", None)) if not answers_a or not answers_b: return 50.0, None common_keys = set(answers_a.keys()) & set(answers_b.keys()) if not common_keys: return 50.0, None matched = sum(1 for key in common_keys if answers_a.get(key) == answers_b.get(key)) total = len(common_keys) score = matched / total * 100 if total > 0 else 0.0 ratio = matched / total if total > 0 else 0.0 if ratio >= 0.75: return round(score, 1), "生活节奏相近" if ratio >= 0.5: return round(score, 1), "价值观较为一致" return round(score, 1), None def _education_score(user_a: User, user_b: User) -> float: scores = [] if user_a.prefer_education and user_b.education: if user_b.education >= user_a.prefer_education: scores.append(100.0) else: scores.append(max(0.0, 100 - (user_a.prefer_education - user_b.education) * 25)) elif user_a.prefer_education: scores.append(40.0) if user_b.prefer_education and user_a.education: if user_a.education >= user_b.prefer_education: scores.append(100.0) else: scores.append(max(0.0, 100 - (user_b.prefer_education - user_a.education) * 25)) elif user_b.prefer_education: scores.append(40.0) if not scores: return 80.0 return sum(scores) / len(scores) def _activity_score(user_b: User) -> float: if not user_b.updated_at: return 30.0 updated_at = user_b.updated_at if updated_at.tzinfo is not None: updated_at = updated_at.replace(tzinfo=None) days_inactive = (datetime.now() - updated_at).days if days_inactive <= 3: return 100.0 if days_inactive <= 7: return 80.0 if days_inactive <= 30: return 50.0 return 20.0 def _location_score(user_a: User, user_b: User) -> float: if not user_a.city or not user_b.city: return 50.0 if user_a.city == user_b.city: return 100.0 if user_a.city[:2] == user_b.city[:2]: return 60.0 return 20.0 def _height_score(user_a: User, user_b: User) -> float: if not user_b.height: return 50.0 if user_a.gender == 2 and user_b.gender == 1: if user_b.height >= 175: return 100.0 if user_b.height >= 170: return 70.0 return 40.0 if user_a.gender == 1 and user_b.gender == 2: if user_b.height >= 160: return 100.0 if user_b.height >= 155: return 70.0 return 40.0 return 70.0 def _safe_personality_tags(value: object) -> list[str]: return value if isinstance(value, list) else [] def _safe_hobbies(value: object) -> set[str]: return set(value) if isinstance(value, list) else set() def _safe_dict(value: object) -> dict: return value if isinstance(value, dict) else {} def calculate_match_score(user_a: User, user_b: User) -> dict: scores = {} reasons: list[str] = [] scores["age"] = _age_score(user_a, user_b) hobby_score, hobby_overlap = _hobby_score(user_a, user_b) scores["hobbies"] = hobby_score if hobby_overlap: reasons.append(f"都喜欢{hobby_overlap[0]}") value_score, value_reason = _value_score(user_a, user_b) scores["values"] = value_score if value_reason: reasons.append(value_reason) scores["education"] = _education_score(user_a, user_b) scores["activity"] = _activity_score(user_b) location_score = _location_score(user_a, user_b) scores["location"] = location_score if location_score == 100: reasons.append("同城") scores["height"] = _height_score(user_a, user_b) total = sum(scores[key] * WEIGHTS[key] for key in scores) return {"score": round(total, 1), "reasons": reasons[:3]} async def _mutual_match_exists(session: AsyncSession, user_id: int, other_user_id: int) -> bool: user_a_id, user_b_id = sorted([user_id, other_user_id]) result = await session.execute( select(Match.id).where(Match.user_a_id == user_a_id, Match.user_b_id == user_b_id) ) return result.scalar_one_or_none() is not None async def _has_liked(session: AsyncSession, from_user_id: int, to_user_id: int) -> bool: result = await session.execute( select(MatchLike.id).where( MatchLike.from_user_id == from_user_id, MatchLike.to_user_id == to_user_id, ) ) return result.scalar_one_or_none() is not None async def get_candidates(session: AsyncSession, current_user: User, source: str) -> list[dict]: liked_rows = await session.execute( select(MatchLike.to_user_id).where(MatchLike.from_user_id == current_user.id) ) liked_user_ids = {row[0] for row in liked_rows.all()} matched_rows = await session.execute( select(Match.user_a_id, Match.user_b_id).where( or_(Match.user_a_id == current_user.id, Match.user_b_id == current_user.id) ) ) matched_user_ids: set[int] = set() for row in matched_rows.all(): matched_user_ids.add(row.user_a_id) matched_user_ids.add(row.user_b_id) matched_user_ids.discard(current_user.id) excluded_user_ids = liked_user_ids | matched_user_ids | {current_user.id} stmt = select(User).where( User.audit_status == 2, User.is_active == 1, ) if current_user.gender in {1, 2}: stmt = stmt.where(User.gender == (2 if current_user.gender == 1 else 1)) if excluded_user_ids: stmt = stmt.where(~User.id.in_(excluded_user_ids)) stmt = stmt.limit(10) result = await session.execute(stmt) users = result.scalars().all() items = [] for user in users: try: score_result = calculate_match_score(current_user, user) reasons = score_result["reasons"] if source == "ai" else None items.append(_level1_info(user, score=score_result["score"], reasons=reasons)) except Exception as e: logger.warning("Failed to score candidate user %s: %s", user.id, e) continue return items async def _get_activity_match_limit(session: AsyncSession) -> int: value = await get_system_config_value(session, "matches_per_activity_limit", 3) try: return max(int(value), 1) except (TypeError, ValueError): return 3 async def _activity_match_count(session: AsyncSession, activity_id: int, user_id: int) -> int: result = await session.execute( select(Match.id).where( Match.source_activity_id == activity_id, or_(Match.user_a_id == user_id, Match.user_b_id == user_id), ) ) return len(result.scalars().all()) async def like_user( session: AsyncSession, current_user: User, to_user: User, source: str, activity_id: int | None = None, ) -> dict: if source not in {"manual", "ai", "activity"}: raise ValueError("无效的来源参数") if current_user.id == to_user.id: raise ValueError("不能对自己表示感兴趣") if current_user.gender in {1, 2} and to_user.gender in {1, 2} and current_user.gender == to_user.gender: raise ValueError("当前仅支持异性匹配") if source == "activity": if activity_id is None: raise ValueError("活动匹配缺少活动标识") limit = await _get_activity_match_limit(session) current_count = await _activity_match_count(session, activity_id, current_user.id) if current_count >= limit: raise ValueError("该活动下你的可匹配次数已达上限") existing_like = await session.execute( select(MatchLike).where( MatchLike.from_user_id == current_user.id, MatchLike.to_user_id == to_user.id, ) ) current_like = existing_like.scalar_one_or_none() if current_like is None: current_like = MatchLike( from_user_id=current_user.id, to_user_id=to_user.id, source=source, created_at=datetime.now(), ) session.add(current_like) await session.flush() reverse_like_exists = await _has_liked(session, to_user.id, current_user.id) if not reverse_like_exists: await session.commit() return {"is_mutual": False, "match_id": None, "match_info": None} user_a_id, user_b_id = sorted([current_user.id, to_user.id]) result = await session.execute( select(Match).where(Match.user_a_id == user_a_id, Match.user_b_id == user_b_id) ) match = result.scalar_one_or_none() if match is None: try: score_result = calculate_match_score(current_user, to_user) match = Match( user_a_id=user_a_id, user_b_id=user_b_id, match_score=score_result["score"], match_type=source, source_activity_id=activity_id if source == "activity" else None, matched_at=datetime.now(), ) session.add(match) await session.flush() except IntegrityError: await session.rollback() result = await session.execute( select(Match).where(Match.user_a_id == user_a_id, Match.user_b_id == user_b_id) ) match = result.scalar_one() elif source == "activity" and match.source_activity_id is None: match.source_activity_id = activity_id session.add(match) await session.commit() await session.refresh(match) return {"is_mutual": True, "match_id": match.id, "match_info": _level2_info(to_user)} async def list_my_matches(session: AsyncSession, current_user: User) -> list[dict]: other_user = aliased(User) match_stmt = ( select(Match, other_user) .join( other_user, or_( other_user.id == Match.user_a_id, other_user.id == Match.user_b_id, ), ) .where(or_(Match.user_a_id == current_user.id, Match.user_b_id == current_user.id)) ) match_result = await session.execute(match_stmt) match_rows = match_result.all() like_stmt = ( select(MatchLike, other_user) .join(other_user, other_user.id == MatchLike.to_user_id) .where(MatchLike.from_user_id == current_user.id) ) like_result = await session.execute(like_stmt) like_rows = like_result.all() items: list[dict] = [] matched_user_ids: set[int] = set() for match, joined_user in match_rows: if joined_user.id == current_user.id: continue matched_user_ids.add(joined_user.id) items.append( { "match_id": match.id, "matched_at": match.matched_at.isoformat(), "match_status": "success", "match_status_text": "匹配成功", "fail_reason": None, "other_user": sanitize_user_media_payload({ "user_id": joined_user.id, "nickname": joined_user.nickname, "real_name": joined_user.real_name, "avatar_url": joined_user.avatar_url, "education": joined_user.education, "city": joined_user.city, "hobbies": list(_safe_hobbies(joined_user.hobbies)), }), } ) for like, joined_user in like_rows: if joined_user.id == current_user.id or joined_user.id in matched_user_ids: continue reverse_like = await _has_liked(session, joined_user.id, current_user.id) items.append( { "match_id": f"like-{like.id}", "matched_at": like.created_at.isoformat(), "match_status": "success" if reverse_like else "pending", "match_status_text": "匹配成功" if reverse_like else "待回应", "fail_reason": None if reverse_like else "等待对方回应", "other_user": sanitize_user_media_payload({ "user_id": joined_user.id, "nickname": joined_user.nickname, "real_name": joined_user.real_name, "avatar_url": joined_user.avatar_url, "education": joined_user.education, "city": joined_user.city, "hobbies": list(_safe_hobbies(joined_user.hobbies)), }), } ) items.sort(key=lambda item: item["matched_at"], reverse=True) return items async def get_match_detail(session: AsyncSession, current_user: User, match_id: int) -> dict | None: match = await session.get(Match, match_id) if match is None: return None if current_user.id not in {match.user_a_id, match.user_b_id}: raise PermissionError("无权限查看该匹配记录") other_user_id = match.user_b_id if match.user_a_id == current_user.id else match.user_a_id other_user = await session.get(User, other_user_id) if other_user is None: return None return { "match_id": match.id, "matched_at": match.matched_at.isoformat(), "other_user": _level3_info(other_user), } async def get_public_user_info(session: AsyncSession, current_user: User, target_user: User) -> dict: if await _mutual_match_exists(session, current_user.id, target_user.id): return _level3_info(target_user) if await _has_liked(session, current_user.id, target_user.id): level2 = _level1_info(target_user) level2.update(_level2_info(target_user)) return level2 return _level1_info(target_user) async def ai_suggest_candidates(session: AsyncSession, current_user: User, limit: int = 5) -> list[dict]: raw_candidates = await get_candidates(session, current_user, "ai") raw_candidates.sort(key=lambda item: item.get("match_score") or 0, reverse=True) main_count = math.ceil(limit * 0.7) main_list = raw_candidates[:main_count] diverse_pool = raw_candidates[main_count:main_count + 10] diverse_count = limit - len(main_list) diverse_list = random.sample(diverse_pool, min(diverse_count, len(diverse_pool))) if diverse_pool and diverse_count > 0 else [] return main_list + diverse_list async def get_match_users(session: AsyncSession, match: Match) -> tuple[User | None, User | None]: result = await session.execute( select(User).where(User.id.in_([match.user_a_id, match.user_b_id])) ) users = {user.id: user for user in result.scalars().all()} return users.get(match.user_a_id), users.get(match.user_b_id)