app/services/wordpress_sync.py

import asyncio
import html
import ipaddress
import socket
import urllib.parse
from dataclasses import dataclass, field
from datetime import datetime, timezone
from html.parser import HTMLParser
from typing import Any, Optional
import uuid

import httpx
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models.database import SessionLocal
from app.models.smart_models import KnowledgeBase, SyncJob
from app.services.embedding_service import EmbeddingService
from app.services.kb_chunking import embed_entry_for_retrieval
from app.services.vector_store import VectorStore
from app.utils.config import config_manager
from app.utils.sync_control import check_cancel_flag, clear_cancel_flag, check_pause_flag, clear_pause_flag

MAX_POST_CHARACTERS = 100_000
WP_PREVIEW_POSTS = 3
WP_SYNC_PER_PAGE = 100
EMBEDDING_BATCH_SIZE = 50

_PRIVATE_NETWORKS = [
    ipaddress.ip_network("10.0.0.0/8"),
    ipaddress.ip_network("172.16.0.0/12"),
    ipaddress.ip_network("192.168.0.0/16"),
    ipaddress.ip_network("127.0.0.0/8"),
    ipaddress.ip_network("::1/128"),
    ipaddress.ip_network("fc00::/7"),
]


class WordPressSyncError(ValueError):
    """Raised when a WordPress sync input or upstream response is invalid."""


class WordPressPostTypeError(WordPressSyncError):
    """Raised when the requested post type does not exist on the WordPress site."""


@dataclass
class UpsertSummary:
    created: int = 0
    updated: int = 0
    skipped: int = 0
    rows_to_embed: list = field(default_factory=list)


@dataclass
class WordPressEntry:
    title: str
    content: str
    source_url: str
    warnings: list[str] = field(default_factory=list)


class _PlainTextHTMLParser(HTMLParser):
    def __init__(self) -> None:
        super().__init__()
        self._chunks: list[str] = []
        self._current_href: Optional[str] = None

    def handle_starttag(self, tag: str, attrs: list[tuple[str, Optional[str]]]) -> None:
        if tag in {"p", "div", "br", "li", "h1", "h2", "h3"}:
            self._chunks.append("\n")
        if tag == "a":
            attrs_dict = dict(attrs)
            self._current_href = attrs_dict.get("href")

    def handle_endtag(self, tag: str) -> None:
        if tag == "a":
            self._current_href = None
        if tag in {"p", "div", "li", "h1", "h2", "h3"}:
            self._chunks.append("\n")

    def handle_data(self, data: str) -> None:
        text = html.unescape(data).strip()
        if not text:
            return
        if self._current_href:
            self._chunks.append(f"{text} ({self._current_href})")
        else:
            self._chunks.append(text)
        self._chunks.append(" ")

    def get_text(self) -> str:
        lines = [" ".join(line.split()) for line in "".join(self._chunks).splitlines()]
        return "\n".join(line for line in lines if line).strip()


def extract_wordpress_plain_text(html_content: str) -> str:
    parser = _PlainTextHTMLParser()
    parser.feed(html_content)
    return parser.get_text()


def validate_public_https_url(url: str) -> str:
    parsed = urllib.parse.urlparse(url)
    if parsed.scheme != "https":
        raise WordPressSyncError("site_url must use HTTPS")

    hostname = parsed.hostname or ""
    if not hostname or hostname.lower() in {"localhost", "0.0.0.0"}:
        raise WordPressSyncError("site_url points to a local address")

    _validate_hostname_public(hostname)
    return url.rstrip("/")


async def fetch_post_type_rest_base(site_url: str, post_type: str) -> str:
    """Return the WP REST base for a post type, or raise WordPressPostTypeError."""
    api_url = f"{site_url}/wp-json/wp/v2/types"
    try:
        async with httpx.AsyncClient(timeout=15.0) as client:
            response = await client.get(api_url, follow_redirects=False)
            response.raise_for_status()
            types = response.json()
    except (httpx.HTTPError, ValueError) as exc:
        raise WordPressSyncError(f"Failed to fetch WordPress post types: {exc}") from exc
    if not isinstance(types, dict) or post_type not in types:
        raise WordPressPostTypeError(f"Post type '{post_type}' not found on this WordPress site")
    return types[post_type].get("rest_base") or post_type


def _validate_hostname_public(hostname: str) -> None:
    addresses: set[str] = set()
    try:
        addresses.add(str(ipaddress.ip_address(hostname)))
    except ValueError:
        try:
            for family, _, _, _, sockaddr in socket.getaddrinfo(hostname, None):
                if family in {socket.AF_INET, socket.AF_INET6}:
                    addresses.add(sockaddr[0])
        except socket.gaierror as exc:
            raise WordPressSyncError("site_url hostname could not be resolved") from exc

    for raw_address in addresses:
        address = ipaddress.ip_address(raw_address)
        if any(address in network for network in _PRIVATE_NETWORKS):
            raise WordPressSyncError("site_url points to a private network")


async def fetch_single_wordpress_post(
    site_url: str, source_url: str, rest_base: str
) -> WordPressEntry:
    """Fetch one WordPress post by permalink from the given rest_base endpoint."""
    api_url = f"{site_url}/wp-json/wp/v2/{rest_base}"
    try:
        async with httpx.AsyncClient(timeout=15.0) as client:
            response = await client.get(
                api_url,
                params={"link": source_url, "per_page": 1, "_fields": "id,title,content,link,guid"},
                follow_redirects=False,
            )
            response.raise_for_status()
            posts = response.json()
    except (httpx.HTTPError, ValueError) as exc:
        raise WordPressSyncError(f"Failed to fetch WordPress post: {exc}") from exc

    if not isinstance(posts, list) or len(posts) == 0:
        raise WordPressSyncError(f"No WordPress post found at {source_url}")

    if posts[0]["link"].rstrip("/") != source_url.rstrip("/"):
        raise WordPressSyncError(f"No WordPress post found at {source_url}")

    return build_wordpress_entry(posts[0])


async def fetch_single_wordpress_post_by_url(source_url: str) -> WordPressEntry:
    """Fetch a single WordPress post by its public permalink via the WP REST API."""
    parsed = urllib.parse.urlparse(source_url)
    site_url = f"{parsed.scheme}://{parsed.netloc}"
    api_url = f"{site_url}/wp-json/wp/v2/posts"
    try:
        async with httpx.AsyncClient(timeout=15.0) as client:
            response = await client.get(
                api_url,
                params={"link": source_url, "per_page": 1, "_fields": "id,title,content,link,guid"},
                follow_redirects=False,
            )
            response.raise_for_status()
            posts = response.json()
    except (httpx.HTTPError, ValueError) as exc:
        raise WordPressSyncError(f"Failed to fetch WordPress post: {exc}") from exc

    if not isinstance(posts, list) or len(posts) == 0:
        raise WordPressSyncError(f"No WordPress post found at {source_url}")

    if posts[0]["link"].rstrip("/") != source_url.rstrip("/"):
        raise WordPressSyncError(f"No WordPress post found at {source_url}")

    return build_wordpress_entry(posts[0])


async def fetch_wordpress_posts(
    site_url: str, page: int, per_page: int, rest_base: str = "posts"
) -> tuple[list[dict[str, Any]], int]:
    api_url = f"{site_url}/wp-json/wp/v2/{rest_base}"
    try:
        async with httpx.AsyncClient(timeout=15.0) as client:
            response = await client.get(
                api_url,
                params={"page": page, "per_page": per_page},
                follow_redirects=False,
            )
            response.raise_for_status()
            posts = response.json()
    except (httpx.HTTPError, ValueError) as exc:
        raise WordPressSyncError(f"Failed to fetch WordPress posts: {exc}") from exc

    if not isinstance(posts, list):
        raise WordPressSyncError("Expected a list of posts from WordPress API")

    total_pages = int(response.headers.get("X-WP-TotalPages", "1"))
    return posts, total_pages


def build_wordpress_entry(post: dict[str, Any]) -> WordPressEntry:
    title = post.get("title", {}).get("rendered", "Untitled WP Post")
    source_url = post.get("link") or post.get("guid", {}).get("rendered")
    if not source_url:
        raise WordPressSyncError(f"Post '{title}': missing source URL")

    body = extract_wordpress_plain_text(post.get("content", {}).get("rendered", ""))
    if not body:
        raise WordPressSyncError(f"Post '{title}': empty content after HTML extraction")

    warnings = []
    if len(body) > MAX_POST_CHARACTERS:
        body = body[:MAX_POST_CHARACTERS]
        warnings.append("Content truncated to 100000 characters")

    published = post.get("date") or "unknown"
    content = f"Published: {published}\nSource: {source_url}\n\n{body}"
    return WordPressEntry(title=title, content=content, source_url=source_url, warnings=warnings)


async def preview_wordpress_entries(
    site_url: str, post_type: str = "post"
) -> tuple[str, list[WordPressEntry], list[str]]:
    safe_site_url = validate_public_https_url(site_url)
    rest_base = await fetch_post_type_rest_base(safe_site_url, post_type)
    posts, _ = await fetch_wordpress_posts(safe_site_url, page=1, per_page=WP_PREVIEW_POSTS, rest_base=rest_base)
    entries, errors = _build_entries(posts)
    return safe_site_url, entries, errors


def _build_entries(posts: list[dict[str, Any]]) -> tuple[list[WordPressEntry], list[str]]:
    entries: list[WordPressEntry] = []
    errors: list[str] = []
    for post in posts:
        try:
            entries.append(build_wordpress_entry(post))
        except WordPressSyncError as exc:
            errors.append(str(exc))
    return entries, errors


def process_wordpress_sync_job(
    tenant_id: str, job_id: str, site_url: str, rest_base: str = "posts", force: bool = False
) -> None:
    asyncio.run(process_wordpress_sync_job_async(tenant_id, job_id, site_url, rest_base=rest_base, force=force))


async def process_wordpress_sync_job_async(
    tenant_id: str, job_id: str, site_url: str, rest_base: str = "posts", force: bool = False
) -> None:
    safe_site_url = validate_public_https_url(site_url)
    tenant_uuid = uuid.UUID(tenant_id)
    job_uuid = uuid.UUID(job_id)
    config = config_manager.get_config()
    embedding_service = EmbeddingService(
        config.embedding_model,
        api_key=config.gemini_embedding_api_key,
    )
    vector_store = VectorStore(config.chroma_persist_path)

    with SessionLocal() as db:
        job = _get_owned_job(db, tenant_uuid, job_uuid)
        if job is None:
            return
        if job.status in ("cancelled", "paused"):  # job was stopped before Celery picked it up
            return
        job.status = "running"
        job.error_message = None
        job.updated_at = datetime.now(timezone.utc)
        db.commit()

        try:
            first_posts, total_pages = await fetch_wordpress_posts(
                safe_site_url, page=1, per_page=WP_SYNC_PER_PAGE, rest_base=rest_base
            )
            all_posts = list(first_posts)
            for page in range(2, total_pages + 1):
                if _handle_cancel(db, job, job_id) or _handle_pause(db, job, job_id):
                    return
                page_posts, _ = await fetch_wordpress_posts(
                    safe_site_url, page=page, per_page=WP_SYNC_PER_PAGE, rest_base=rest_base
                )
                all_posts.extend(page_posts)

            if _handle_cancel(db, job, job_id) or _handle_pause(db, job, job_id):
                return

            job.total = len(all_posts)
            db.commit()
            entries, errors = _build_entries(all_posts)
            job.failed = len(errors)
            if errors:
                job.error_message = "; ".join(errors[:5])

            summary = _upsert_synced_entries(db, tenant_uuid, entries, force=force)

            if _handle_cancel(db, job, job_id) or _handle_pause(db, job, job_id):
                return

            _embed_rows(vector_store, embedding_service, tenant_uuid, summary.rows_to_embed, db)
            job.processed = len(entries)
            job.created = summary.created
            job.updated = summary.updated
            job.skipped = summary.skipped
            # Read current status directly from DB to detect a concurrent cancel without
            # overwriting the stats we just set on the in-memory object.
            db_status = db.execute(
                select(SyncJob.status).where(SyncJob.id == job_uuid)
            ).scalar_one()
            if db_status not in ("cancelled", "paused"):
                job.status = "completed"
            job.updated_at = datetime.now(timezone.utc)
            db.commit()
        except Exception as exc:  # pylint: disable=broad-exception-caught
            db.rollback()
            failed_job = _get_owned_job(db, tenant_uuid, job_uuid)
            if failed_job is not None:
                failed_job.status = "failed"
                failed_job.error_message = f"{type(exc).__name__}: {exc}"
                failed_job.updated_at = datetime.now(timezone.utc)
                db.commit()
            raise


def _handle_cancel(db: Session, job: SyncJob, job_id: uuid.UUID) -> bool:
    """Return True and mark job cancelled if the cancel flag is set. Clears the flag."""
    if check_cancel_flag(str(job_id)):
        clear_cancel_flag(str(job_id))
        job.status = "cancelled"
        job.updated_at = datetime.now(timezone.utc)
        db.commit()
        return True
    return False


def _handle_pause(db: Session, job: SyncJob, job_id: uuid.UUID) -> bool:
    """Return True and mark job paused if the pause flag is set. Clears the flag."""
    if check_pause_flag(str(job_id)):
        clear_pause_flag(str(job_id))
        job.status = "paused"
        job.updated_at = datetime.now(timezone.utc)
        db.commit()
        return True
    return False


def _get_owned_job(db: Session, tenant_id: uuid.UUID, job_id: uuid.UUID) -> Optional[SyncJob]:
    return db.scalars(
        select(SyncJob).where(SyncJob.id == job_id, SyncJob.tenant_id == tenant_id)
    ).first()


def _upsert_synced_entries(
    db: Session, tenant_id: uuid.UUID, entries: list[WordPressEntry], force: bool = False
) -> UpsertSummary:
    summary = UpsertSummary()
    for entry in entries:
        kb = db.scalars(
            select(KnowledgeBase).where(
                KnowledgeBase.tenant_id == tenant_id,
                KnowledgeBase.source_url == entry.source_url,
            )
        ).first()
        if kb is None:
            kb = KnowledgeBase(
                tenant_id=tenant_id,
                title=entry.title,
                content=entry.content,
                source_url=entry.source_url,
                origin="synced",
            )
            db.add(kb)
            db.flush()
            summary.created += 1
            summary.rows_to_embed.append(kb)
            continue

        if not force:
            summary.skipped += 1
            continue

        if kb.title != entry.title or kb.content != entry.content:
            kb.title = entry.title
            kb.content = entry.content
            kb.origin = "synced"
        kb.is_indexed = False
        summary.updated += 1
        summary.rows_to_embed.append(kb)

    db.commit()
    for row in summary.rows_to_embed:
        db.refresh(row)
    return summary


def _embed_rows(
    vector_store: VectorStore,
    embedding_service: EmbeddingService,
    tenant_id: uuid.UUID,
    rows: list[KnowledgeBase],
    db: Session,
) -> None:
    for start in range(0, len(rows), EMBEDDING_BATCH_SIZE):
        for row in rows[start : start + EMBEDDING_BATCH_SIZE]:
            embed_entry_for_retrieval(
                embedding_service=embedding_service,
                vector_store=vector_store,
                tenant_id=str(tenant_id),
                entry_id=str(row.id),
                text=row.content,
                metadata={
                    "origin": row.origin,
                    "source_url": row.source_url,
                },
            )
            row.is_indexed = True