platform/services/reader/app/api/feeds.py

"""Feed endpoints."""

import logging
import re

import feedparser
import httpx
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.api.deps import get_db_session, get_user_id
from app.models import Category, Entry, Feed

log = logging.getLogger(__name__)
router = APIRouter(prefix="/api/feeds", tags=["feeds"])


# ── Schemas ──────────────────────────────────────────────────────────────


class CategoryRef(BaseModel):
    id: int
    title: str

    class Config:
        from_attributes = True


class FeedOut(BaseModel):
    id: int
    title: str
    feed_url: str
    site_url: str | None = None
    category: CategoryRef | None = None

    class Config:
        from_attributes = True


class FeedCreate(BaseModel):
    feed_url: str
    category_id: int | None = None


class CountersOut(BaseModel):
    unreads: dict[str, int]


# ── Helpers ──────────────────────────────────────────────────────────────


def _discover_feed_url(html: str, base_url: str) -> str | None:
    """Try to find an RSS/Atom feed link in HTML."""
    patterns = [
        r'<link[^>]+type=["\']application/(?:rss|atom)\+xml["\'][^>]+href=["\']([^"\']+)["\']',
        r'<link[^>]+href=["\']([^"\']+)["\'][^>]+type=["\']application/(?:rss|atom)\+xml["\']',
    ]
    for pat in patterns:
        match = re.search(pat, html, re.IGNORECASE)
        if match:
            href = match.group(1)
            if href.startswith("/"):
                # Resolve relative URL
                from urllib.parse import urljoin
                href = urljoin(base_url, href)
            return href
    return None


async def _fetch_and_parse_feed(feed_url: str) -> tuple[str, str, str | None]:
    """
    Fetch a URL. If it's a valid feed, return (feed_url, title, site_url).
    If it's HTML, try to discover the feed link and follow it.
    """
    async with httpx.AsyncClient(timeout=30, follow_redirects=True) as client:
        resp = await client.get(feed_url, headers={"User-Agent": "Reader/1.0"})
        resp.raise_for_status()

    body = resp.text
    parsed = feedparser.parse(body)

    # Check if it's a valid feed
    if parsed.feed.get("title") or parsed.entries:
        title = parsed.feed.get("title", feed_url)
        site_url = parsed.feed.get("link")
        return feed_url, title, site_url

    # Not a feed — try to discover from HTML
    discovered = _discover_feed_url(body, feed_url)
    if not discovered:
        raise HTTPException(status_code=400, detail="No RSS/Atom feed found at this URL")

    # Fetch the discovered feed
    async with httpx.AsyncClient(timeout=30, follow_redirects=True) as client:
        resp2 = await client.get(discovered, headers={"User-Agent": "Reader/1.0"})
        resp2.raise_for_status()

    parsed2 = feedparser.parse(resp2.text)
    title = parsed2.feed.get("title", discovered)
    site_url = parsed2.feed.get("link") or feed_url
    return discovered, title, site_url


# ── Routes ───────────────────────────────────────────────────────────────


@router.get("/counters", response_model=CountersOut)
async def feed_counters(
    user_id: str = Depends(get_user_id),
    db: AsyncSession = Depends(get_db_session),
):
    result = await db.execute(
        select(Entry.feed_id, func.count(Entry.id))
        .where(Entry.user_id == user_id, Entry.status == "unread")
        .group_by(Entry.feed_id)
    )
    unreads = {str(row[0]): row[1] for row in result.all()}
    return {"unreads": unreads}


@router.get("", response_model=list[FeedOut])
async def list_feeds(
    user_id: str = Depends(get_user_id),
    db: AsyncSession = Depends(get_db_session),
):
    result = await db.execute(
        select(Feed)
        .where(Feed.user_id == user_id)
        .order_by(Feed.title)
    )
    return result.scalars().all()


@router.post("", response_model=FeedOut, status_code=201)
async def create_feed(
    body: FeedCreate,
    user_id: str = Depends(get_user_id),
    db: AsyncSession = Depends(get_db_session),
):
    # Check for duplicate
    existing = await db.execute(
        select(Feed).where(Feed.feed_url == body.feed_url)
    )
    if existing.scalar_one_or_none():
        raise HTTPException(status_code=409, detail="Feed already exists")

    # Validate category belongs to user
    if body.category_id:
        cat = await db.execute(
            select(Category).where(
                Category.id == body.category_id,
                Category.user_id == user_id,
            )
        )
        if not cat.scalar_one_or_none():
            raise HTTPException(status_code=404, detail="Category not found")

    # Fetch and discover feed
    try:
        actual_url, title, site_url = await _fetch_and_parse_feed(body.feed_url)
    except httpx.HTTPError as e:
        log.warning("Failed to fetch feed %s: %s", body.feed_url, e)
        raise HTTPException(status_code=400, detail=f"Could not fetch feed: {e}")

    # Check again with discovered URL
    if actual_url != body.feed_url:
        existing = await db.execute(
            select(Feed).where(Feed.feed_url == actual_url)
        )
        if existing.scalar_one_or_none():
            raise HTTPException(status_code=409, detail="Feed already exists")

    feed = Feed(
        user_id=user_id,
        category_id=body.category_id,
        title=title,
        feed_url=actual_url,
        site_url=site_url,
    )
    db.add(feed)
    await db.commit()
    await db.refresh(feed)
    return feed


@router.delete("/{feed_id}", status_code=204)
async def delete_feed(
    feed_id: int,
    user_id: str = Depends(get_user_id),
    db: AsyncSession = Depends(get_db_session),
):
    result = await db.execute(
        select(Feed).where(Feed.id == feed_id, Feed.user_id == user_id)
    )
    feed = result.scalar_one_or_none()
    if not feed:
        raise HTTPException(status_code=404, detail="Feed not found")
    await db.delete(feed)
    await db.commit()


@router.post("/{feed_id}/refresh")
async def refresh_feed(
    feed_id: int,
    user_id: str = Depends(get_user_id),
    db: AsyncSession = Depends(get_db_session),
):
    result = await db.execute(
        select(Feed).where(Feed.id == feed_id, Feed.user_id == user_id)
    )
    feed = result.scalar_one_or_none()
    if not feed:
        raise HTTPException(status_code=404, detail="Feed not found")

    import asyncio
    from app.worker.tasks import fetch_single_feed
    await asyncio.to_thread(fetch_single_feed, feed_id)

    return {"ok": True, "message": f"Refreshed {feed.title}"}


@router.post("/refresh-all")
async def refresh_all_feeds(
    user_id: str = Depends(get_user_id),
    db: AsyncSession = Depends(get_db_session),
):
    result = await db.execute(
        select(Feed).where(Feed.user_id == user_id)
    )
    feeds = result.scalars().all()

    import asyncio
    from app.worker.tasks import fetch_single_feed
    for feed in feeds:
        try:
            await asyncio.to_thread(fetch_single_feed, feed.id)
        except Exception:
            pass

    return {"ok": True, "message": f"Refreshed {len(feeds)} feeds"}