hunjing/backend/app/domains/kouxin/service.py

"""叩心应用服务：候选生成、约束裁决与派发。

铁律（PRD §6.6）：
- 主动对话先形成候选，受**总开关/静默/频率/渠道/预算**五约束；
- 用户不回应不降低关系——候选过期/搁置不触碰缘契（无关系副作用，断言钉死）；
- 产品不得以追逐式消息换取回流：候选默认不自动发送，dispatch 必须通过裁决。
"""

from __future__ import annotations

import json
from datetime import UTC, datetime, time

from sqlalchemy import func, select

from app.core.config import settings
from app.core.errors import AppError
from app.core.time import utc_now
from app.db.session import AsyncSessionLocal
from app.domains import one_qi
from app.domains.kouxin.models import KnockingCandidate
from app.models.character import Character


def _parse_hhmm(raw: str) -> time:
    hh, mm = raw.split(":")
    return time(int(hh), int(mm))


def _ruling_context() -> dict:
    """约束配置（env 级；P1 单用户档）。"""
    return {
        "enabled": settings.knocking_enabled,
        "quiet_start": _parse_hhmm(settings.knocking_quiet_start),
        "quiet_end": _parse_hhmm(settings.knocking_quiet_end),
        "daily_limit": settings.knocking_daily_limit,
        "budget_total": settings.knocking_budget_total,
        "channel": "in_app",
    }


def _in_quiet_window(now: datetime, quiet_start: time, quiet_end: time) -> bool:
    """静默窗口判定（支持跨午夜区间，如 23:00–07:00）。"""
    current = now.time()
    if quiet_start <= quiet_end:
        return quiet_start <= current < quiet_end
    return current >= quiet_start or current < quiet_end


async def rule(character_id: str, owner_scope: str) -> dict:
    """约束裁决：总开关 → 静默窗口 → 频率（24h 内已发数）→ 预算（累计已发数）。"""
    ctx = _ruling_context()
    # 静默窗口为本地时区语义（用户作息）；UTC 换算到本地再判定
    from zoneinfo import ZoneInfo

    now_utc = datetime.now(UTC)
    now = now_utc.astimezone(ZoneInfo(settings.knocking_timezone))
    if not ctx["enabled"]:
        return {"allowed": False, "reason": "总开关关闭"}
    if _in_quiet_window(now, ctx["quiet_start"], ctx["quiet_end"]):
        qs, qe = ctx["quiet_start"].strftime("%H:%M"), ctx["quiet_end"].strftime("%H:%M")
        return {"allowed": False, "reason": f"静默窗口（{qs}–{qe}）"}

    async with AsyncSessionLocal() as session:
        since = now.timestamp() - 86400
        daily = (
            await session.execute(
                select(func.count()).select_from(KnockingCandidate).where(
                    KnockingCandidate.character_id == character_id,
                    KnockingCandidate.status == "dispatched",
                    KnockingCandidate.dispatched_at.isnot(None),
                )
            )
        ).scalar_one()
        total = daily  # P1 单周期：预算=累计派发数（开发档与频率同源，分离字段留配置位）
    if daily >= ctx["daily_limit"]:
        return {"allowed": False, "reason": f"24 小时频率已达上限（{daily}/{ctx['daily_limit']}）"}
    if total >= ctx["budget_total"] > 0:
        return {"allowed": False, "reason": "预算耗尽"}
    return {"allowed": True, "reason": "通过"}


async def create_candidate(
    *,
    character_id: str,
    owner_scope: str,
    trigger: str,
    draft: str,
) -> dict:
    """登记叩心候选（candidate；不自动发送）。"""
    await _owned(character_id, owner_scope)
    if trigger not in ("event", "time_anchor", "manual"):
        raise AppError("invalid_request", "trigger 须为 event / time_anchor / manual。")
    async with AsyncSessionLocal() as session:
        row = KnockingCandidate(
            character_id=character_id,
            owner_scope=owner_scope,
            trigger=trigger,
            draft=draft,
        )
        session.add(row)
        await session.commit()
        session.expunge(row)
        return _cand_dict(row)


async def dispatch(candidate_id: str, owner_scope: str, *, draft_override: str | None) -> dict:
    """派发候选：先过约束裁决，未过被拒且裁决快照留痕。"""
    async with AsyncSessionLocal() as session:
        row = await session.get(KnockingCandidate, candidate_id)
        if row is None or row.owner_scope != owner_scope:
            raise AppError("not_found", "叩心候选不存在。", status_code=404)
        if row.status != "candidate":
            raise AppError("invalid_request", "候选不在可派发状态。", status_code=409)
        character_id = row.character_id
        session.expunge(row)

    ruling = await rule(character_id, owner_scope)
    if not ruling["allowed"]:
        async with AsyncSessionLocal() as session:
            row = await session.get(KnockingCandidate, candidate_id)
            row.ruling_json = json.dumps(ruling, ensure_ascii=False)
            await session.commit()
        raise AppError("invalid_request", f"叩心被拒：{ruling['reason']}", status_code=429)

    async with AsyncSessionLocal() as session:
        row = await session.get(KnockingCandidate, candidate_id)
        if draft_override is not None:
            row.draft = draft_override
        row.status = "dispatched"
        row.dispatched_at = utc_now()
        row.ruling_json = json.dumps(ruling, ensure_ascii=False)
        await session.commit()
    await one_qi.emit(
        "knocking.dispatched", character_id=character_id,
        payload={"candidate_id": candidate_id, "channel": "in_app"}, source="kouxin",
    )
    return {"candidate_id": candidate_id, "status": "dispatched", "channel": "in_app"}


async def dismiss(candidate_id: str, owner_scope: str) -> dict:
    """搁置候选（用户决定；不触碰缘契——无关系副作用）。"""
    await _get(candidate_id, owner_scope)
    async with AsyncSessionLocal() as session:
        row = await session.get(KnockingCandidate, candidate_id)
        row.status = "dismissed"
        await session.commit()
    return {"candidate_id": candidate_id, "status": "dismissed"}


async def expire_stale(max_age_hours: int = 72) -> int:
    """过期未响应候选（expired）：只是状态推进，不产生任何关系影响。"""
    async with AsyncSessionLocal() as session:
        rows = (
            await session.execute(
                select(KnockingCandidate).where(KnockingCandidate.status == "dispatched")
            )
        ).scalars().all()
        n = 0
        now = utc_now()
        for row in rows:
            if row.dispatched_at is not None:
                dispatched = row.dispatched_at
                if dispatched.tzinfo is None:  # SQLite 回读 naive（开发档）
                    dispatched = dispatched.replace(tzinfo=UTC)
                age = (now - dispatched).total_seconds() / 3600
                if age >= max_age_hours:
                    row.status = "expired"
                    n += 1
        if n:
            await session.commit()
    return n


async def list_candidates(owner_scope: str) -> list[dict]:
    async with AsyncSessionLocal() as session:
        rows = (
            await session.execute(
                select(KnockingCandidate)
                .where(KnockingCandidate.owner_scope == owner_scope)
                .order_by(KnockingCandidate.created_at.desc())
            )
        ).scalars().all()
        return [_cand_dict(r) for r in rows]


async def _owned(character_id: str, owner_scope: str) -> None:
    async with AsyncSessionLocal() as session:
        character = await session.get(Character, character_id)
        if character is None or character.owner_scope != owner_scope:
            raise AppError("not_found", "人物不存在。", status_code=404)


async def _get(candidate_id: str, owner_scope: str) -> dict:
    async with AsyncSessionLocal() as session:
        row = await session.get(KnockingCandidate, candidate_id)
        if row is None or row.owner_scope != owner_scope:
            raise AppError("not_found", "叩心候选不存在。", status_code=404)
        session.expunge(row)
        return _cand_dict(row)


def _cand_dict(c: KnockingCandidate) -> dict:
    return {
        "id": c.id,
        "character_id": c.character_id,
        "trigger": c.trigger,
        "draft": c.draft,
        "channel": c.channel,
        "status": c.status,
        "ruling": json.loads(c.ruling_json or "{}"),
        "created_at": c.created_at.isoformat() if c.created_at else None,
        "dispatched_at": c.dispatched_at.isoformat() if c.dispatched_at else None,
    }