devplacepy/devplacepy/database/content.py

# retoor <retoor@molodetz.nl>

from collections import Counter

from devplacepy.cache import TTLCache

from .core import db, get_table, or_

_daily_topic_cache = TTLCache(ttl=60, max_size=1)
_trending_cache = TTLCache(ttl=15, max_size=1)


def resolve_by_slug(table, slug, include_deleted=False):
    has_soft_delete = table.has_column("deleted_at")
    flt = {} if include_deleted or not has_soft_delete else {"deleted_at": None}
    entry = table.find_one(slug=slug, **flt)
    if not entry:
        entry = table.find_one(uid=slug, **flt)
    return entry


def resolve_object_url(target_type: str, target_uid: str) -> str:
    if target_type == "post":
        post = resolve_by_slug(get_table("posts"), target_uid)
        return f"/posts/{post['slug'] or post['uid']}" if post else "/feed"
    if target_type == "project":
        project = resolve_by_slug(get_table("projects"), target_uid)
        return (
            f"/projects/{project['slug'] or project['uid']}" if project else "/projects"
        )
    if target_type == "news":
        article = resolve_by_slug(get_table("news"), target_uid)
        if article:
            return f"/news/{article.get('slug', '') or article['uid']}"
        return "/news"
    if target_type == "issue":
        return f"/issues?highlight={target_uid}"
    if target_type == "gist":
        gist = resolve_by_slug(get_table("gists"), target_uid)
        return f"/gists/{gist['slug'] or gist['uid']}" if gist else "/gists"
    if target_type == "quiz":
        quiz = resolve_by_slug(get_table("quizzes"), target_uid)
        return f"/quizzes/{quiz['slug'] or quiz['uid']}" if quiz else "/quizzes"
    if target_type == "comment":
        comment = get_table("comments").find_one(uid=target_uid, deleted_at=None)
        if not comment:
            return "/feed"
        parent_url = resolve_object_url(
            comment.get("target_type", "post"),
            comment.get("target_uid") or comment.get("post_uid", ""),
        )
        return f"{parent_url}#comment-{target_uid}"
    if target_type == "award":
        award = resolve_by_slug(get_table("awards"), target_uid)
        if award:
            receiver = get_table("users").find_one(uid=award.get("receiver_uid", ""))
            if receiver:
                return f"/profile/{receiver['username']}?tab=awards#award-{award.get('slug', '')}"
        return "/feed"
    if target_type == "user":
        person = get_table("users").find_one(uid=target_uid)
        return f"/profile/{person['username']}" if person else "/feed"
    if target_type == "project_file":
        node = get_table("project_files").find_one(uid=target_uid)
        if not node:
            return "/projects"
        project = get_table("projects").find_one(uid=node.get("project_uid", ""))
        if not project:
            return "/projects"
        slug = project.get("slug") or project["uid"]
        return f"/projects/{slug}/files?path={node.get('path', '')}"
    if target_type == "attachment":
        attachment = get_table("attachments").find_one(uid=target_uid)
        if not attachment:
            return "/feed"
        parent_type = attachment.get("target_type") or ""
        parent_uid = attachment.get("target_uid") or ""
        if parent_type and parent_uid:
            return resolve_object_url(parent_type, parent_uid)
        owner = get_table("users").find_one(uid=attachment.get("user_uid", ""))
        return f"/profile/{owner['username']}?tab=media" if owner else "/feed"
    if target_type == "message":
        message = get_table("messages").find_one(uid=target_uid)
        if not message:
            return "/messages"
        return f"/messages?with_uid={message.get('sender_uid', '')}"
    if target_type == "poll":
        poll = get_table("polls").find_one(uid=target_uid)
        if not poll:
            return "/feed"
        return resolve_object_url("post", poll.get("post_uid", ""))
    if target_type == "workspace":
        instance = get_table("instances").find_one(uid=target_uid)
        return f"/admin/containers/{instance['uid']}" if instance else "/admin/containers"
    if target_type == "devii_output":
        return "/devii"
    return "/feed"


def get_uids_by_username_match(search, limit=200):
    term = (search or "").strip()
    if not term or "users" not in db.tables:
        return []
    rows = db.query(
        "SELECT uid FROM users WHERE username LIKE :q LIMIT :limit",
        q=f"%{term}%",
        limit=limit,
    )
    return [row["uid"] for row in rows]


def text_search_clause(
    table, search, fields=("title", "description"), author_field=None
):
    if not search or not search.strip() or not table.exists:
        return None
    columns = table.table.columns
    like = f"%{search.strip()}%"
    matches = [columns[field].ilike(like) for field in fields if field in columns]
    if author_field and author_field in columns:
        author_uids = get_uids_by_username_match(search)
        if author_uids:
            matches.append(columns[author_field].in_(author_uids))
    return or_(*matches) if matches else None


def get_daily_topic():
    cached = _daily_topic_cache.get("topic")
    if cached is not None:
        return cached
    topic = _load_daily_topic()
    _daily_topic_cache.set("topic", topic)
    return topic


def _load_daily_topic():
    if "news" in db.tables:
        article = db["news"].find_one(
            status="published", deleted_at=None, order_by=["-synced_at"]
        )
        if article:
            desc = (article.get("description") or "")[:200] or (
                article.get("content") or ""
            )[:200]
            return {
                "title": article.get("title", ""),
                "summary": desc,
                "slug": article.get("slug", ""),
                "url": article.get("url", ""),
                "image_url": article.get("image_url", ""),
            }
    return {
        "title": "Welcome to DevPlace",
        "summary": "Stay tuned for the latest dev news.",
    }


def get_featured_news(limit=5):
    if "news" not in db.tables:
        return []
    from devplacepy.utils import time_ago

    rows = list(
        db["news"].find(
            show_on_landing=1, deleted_at=None, order_by=["-synced_at"], _limit=limit
        )
    )
    articles = []
    for article in rows:
        summary = (article.get("description") or "")[:120] or (
            article.get("content") or ""
        )[:120]
        articles.append(
            {
                "title": article.get("title", ""),
                "summary": summary,
                "slug": article.get("slug", ""),
                "url": article.get("url", ""),
                "source_name": article.get("source_name", ""),
                "featured": article.get("featured", 0),
                "image_url": article.get("image_url", "") or "",
                "time_ago": time_ago(article["synced_at"])
                if article.get("synced_at")
                else "",
            }
        )
    return articles


def get_trending_topics(limit: int = 6) -> list[dict]:
    cached = _trending_cache.get("topics")
    if cached is not None:
        return cached[:limit]
    if "posts" not in db.tables or "topic" not in db["posts"].columns:
        return []
    rows = db.query(
        "SELECT topic FROM posts WHERE deleted_at IS NULL "
        "AND topic IS NOT NULL AND topic != '' "
        "ORDER BY created_at DESC LIMIT 200"
    )
    counter: Counter[str] = Counter()
    for row in rows:
        topic = (row["topic"] or "").strip()
        if topic:
            counter[topic] += 1
    topics = [{"topic": t, "count": c} for t, c in counter.most_common(limit)]
    _trending_cache.set("topics", topics)
    return topics