삼성증권 미팅 안건별 정리

@staticmethod
def fetch_delivery_symbol_ids(session: Session) -> List[int]:
    """
    * 배송 종목(is_final)을 StockTwits 심볼 id 배열로 돌려준다
    ! 데몬 실행 시작 때 한 번만 호출한다. 쿼리 본문에서 COALESCE 문자열 비교를 없애는 것이 목적
    """
    sql = text(
        "SELECT DISTINCT s.id FROM suml.stocktwits_symbols s "
        "JOIN (SELECT DISTINCT symbol, exchange FROM suml.community_delivery_symbols "
        "      WHERE is_final = TRUE) ds "
        "  ON s.symbol = ds.symbol AND COALESCE(s.exchange, '') = COALESCE(ds.exchange, '')"
    )
    return [row[0] for row in session.execute(sql)]

@staticmethod
def fetch_new_body_messages_v2(
    session: Session, target_date: str, limit: int, logger: Logger
) -> List[Dict[str, Any]]:
    """
    * 신규 본문 조회 v2: 그날 적재분 우선 + 심볼 id 배열 판정 + EXISTS 중복 제거
    ! 반환 dict 컬럼 12개는 v1과 같다. 다운스트림 변경 없음
    """
    symbol_ids = TargetMessagesQuery.fetch_delivery_symbol_ids(session)
    sql = TargetMessagesQuery._load_sql("target_messages_new_body_only_v2.sql").format(
        limit_clause=f"LIMIT {limit}" if limit != -1 else ""
    )
    result = session.execute(
        text(sql), {"target_date": target_date, "symbol_ids": symbol_ids}
    )
    columns = result.keys()
    return [dict(zip(columns, row)) for row in result]

@staticmethod
def fetch_only_thread_cluster_messages_v2(
    session: Session, target_date: str, limit: int, logger: Logger
) -> List[Dict[str, Any]]:
    """
    * 스레드 클러스터 조회 v2: 4단계 분리 실행
    ! 재귀 CTE 한 문장은 반복마다 메시지 테이블 전체를 해시하므로(로컬 2.4~3.1초) 쓰지 않는다
    ! 반환 dict 컬럼 12개와 정렬은 v3와 같다. 결과는 로컬에서 v3와 행 단위로 같았다
    """
    symbol_ids = TargetMessagesQuery.fetch_delivery_symbol_ids(session)

    # * 1단계: 그날 답글 가운데 루트와 직접 부모가 있고 루트가 배송 종목인 후보 (created_at 인덱스)
    step1 = TargetMessagesQuery._load_sql("thread_cluster_step1_v1.sql")
    candidates = session.execute(
        text(step1), {"target_date": target_date, "symbol_ids": symbol_ids}
    ).fetchall()  # (id, root_id, in_reply_to_message_id, created_at)

    # * 클러스터를 첫 답글 시각 순으로 limit + 100개까지만 남긴다 (검증 탈락분 여유)
    first_at = {}
    for _, root_id, _, created_at in candidates:
        if root_id not in first_at or created_at < first_at[root_id]: first_at[root_id] = created_at
    kept_roots = {r for r, _ in sorted(first_at.items(), key=lambda kv: (kv[1], kv[0]))[: limit + 100]}
    limited = [c for c in candidates if c[1] in kept_roots]

    # * 2단계: 직접 부모 id를 묶어 기본키로 조회하고, 그 부모의 부모를 다시 조회하기를 루트에 닿을 때까지 반복
    parent_of = {}
    frontier = {parent for _, _, parent, _ in limited if parent is not None}
    while frontier:
        ids = [i for i in frontier if i not in parent_of]
        if not ids: break
        rows = session.execute(
            text("SELECT id, in_reply_to_message_id FROM suml.stocktwits_messages WHERE id = ANY(:ids)"),
            {"ids": ids},
        ).fetchall()
        for node_id, parent_id in rows: parent_of[node_id] = parent_id
        for node_id in ids:
            if node_id not in parent_of: parent_of[node_id] = "MISSING"
        frontier = {p for _, p in rows if p is not None and p in parent_of and p != "MISSING"} - set(parent_of)

    # * 3단계: 체인이 루트에 닿는 답글만 남기고 클러스터 limit개를 확정한다
    valid, context_by_target = [], {}
    for message_id, root_id, parent_id, created_at in limited:
        path, node, has_reached_root = [message_id], parent_id, False
        while node is not None and node != "MISSING" and node not in path:
            path.append(node)
            if node == root_id: has_reached_root = True; break
            node = parent_of.get(node)
        if has_reached_root: valid.append((message_id, root_id, created_at)); context_by_target[message_id] = path
    first_valid = {}
    for _, root_id, created_at in valid:
        if root_id not in first_valid or created_at < first_valid[root_id]: first_valid[root_id] = created_at
    roots = sorted({r for r, _ in sorted(first_valid.items(), key=lambda kv: (kv[1], kv[0]))[:limit]})
    context_ids = sorted({n for m, r, _ in valid if r in roots for n in context_by_target[m]})

    # * 4단계: 루트와 문맥 id만 넘겨 현행과 같은 12개 컬럼을 만든다. 리셰어 수도 이 id에 대해서만 센다
    context_sql = TargetMessagesQuery._load_sql("thread_cluster_context_v1.sql")
    result = session.execute(
        text(context_sql), {"roots": roots, "context_ids": context_ids, "all_ids": sorted(set(roots) | set(context_ids))}
    )
    columns = result.keys()
    return [dict(zip(columns, row)) for row in result]

def postprocess_only_thread_cluster(self, target_date: str) -> Dict[str, int]:
    # * POSTPROCESSOR_QUERY_VERSION이 "v2"면 4단계 조회, 그 밖(legacy)이면 기존 한 문장 쿼리
    is_v2 = self.config.get("POSTPROCESSOR_QUERY_VERSION") == "v2"
    fetch_fn = (
        TargetMessagesQuery.fetch_only_thread_cluster_messages_v2
        if is_v2 else TargetMessagesQuery.fetch_only_thread_cluster_messages
    )
    return self._run_postprocessing_pipeline(
        target_date=target_date,
        fetch_fn=fetch_fn,
        mode_name="MESSAGE_ONLY_THREAD_CLUSTER_BATCH",
        log_header=f"릴레이션 클러스터 후처리 시작 (target_date={target_date}, query={'v2' if is_v2 else 'legacy'})",
        batch_limit_key="POSTPROCESSOR_ONLY_THREAD_CLUSTER_BATCH_LIMIT",
        batch_unit="스레드 클러스터",
    )