Function get_latest_push

skills/storage-helper/scripts/db_ops.py:1300–1368

Get the latest push information for a user Args: user_id: User ID Returns: Push information with papers list

(user_id: str)

def get_latest_push(user_id: str) -> Optional[Dict]:
    """
    Get the latest push information for a user

    Args:
        user_id: User ID

    Returns:
        Push information with papers list
    """
    import json
    conn = get_connection()
    cursor = conn.cursor()

    # Get the latest real push_id from behavior logs
    cursor.execute("""
        SELECT push_id, MAX(timestamp) as timestamp
        FROM behavior_logs
        WHERE user_id = ?
          AND action IN ('pushed', 'push_empty')
        GROUP BY push_id
        ORDER BY timestamp DESC, push_id DESC
        LIMIT 1
    """, (user_id,))

    row = cursor.fetchone()
    if not row:
        conn.close()
        return None

    push_id = row['push_id']
    push_time = row['timestamp']

    # Get all papers in this push with their category and rank from metadata.
    cursor.execute("""
        SELECT p.*, bl.id as behavior_log_id, bl.metadata as bl_metadata FROM papers p
        JOIN behavior_logs bl ON p.id = bl.paper_id
        WHERE bl.user_id = ? AND bl.push_id = ? AND bl.action = 'pushed'
        ORDER BY bl.id ASC
    """, (user_id, push_id))

    papers = [_build_paper_dict(row, metadata_key="bl_metadata") for row in cursor.fetchall()]
    papers.sort(key=lambda paper: (paper.get("rank", 10**9), paper.get("behavior_log_id", 10**9)))

    cursor.execute(
        """
        SELECT metadata
        FROM behavior_logs
        WHERE user_id = ?
          AND push_id = ?
          AND action = 'push_empty'
        ORDER BY id DESC
        LIMIT 1
        """,
        (user_id, push_id),
    )
    metadata_row = cursor.fetchone()
    push_metadata = _load_json_metadata(metadata_row["metadata"]) if metadata_row else {}