athena-oracle/oracle/summarize.py

"""Summarization engine for the AI Research Oracle.

Generates structured summaries for entries where summary IS NULL.
Uses source-specific extraction logic (no LLM required).

Output schema: {one_liner, key_technical_point, potential_use_case, confidence}
"""
import json
import re
import sqlite3
from typing import Optional

from oracle.config import DB_PATH


def extract_github_summary(title: str, content: str) -> dict:
    """Extract summary from GitHub README content."""
    text = re.sub(r'<p[^>]*>', '\n', content)
    text = re.sub(r'</p>', '\n', content)
    text = re.sub(r'<h[1-6][^>]*>', '\n## ', text)
    text = re.sub(r'</h[1-6]>', '\n', text)
    text = re.sub(r'<[^>]+>', '', text)
    text = re.sub(r'&amp;', '&', text)
    text = re.sub(r'&mdash;', '—', text)
    text = re.sub(r'&#39;', "'", text)
    text = re.sub(r'&middot;', '·', text)
    text = re.sub(r'```[\s\S]*?```', '', text)
    text = re.sub(r'\n\s*\n+', '\n\n', text)
    text = text.strip()

    source_confidence = "low"
    if len(text) > 2000:
        source_confidence = "high"
    elif len(text) > 500:
        source_confidence = "medium"

    one_liner = _find_project_description(text, title) or title[:200]
    key_tech = _extract_technical_point(text, source_confidence)
    use_case = _extract_use_case(text, title)
    confidence = _assess_extraction_quality(one_liner, key_tech, use_case, source_confidence)

    if _is_security_tooling(title, one_liner, key_tech):
        use_case = use_case + " [security:dual-use]"

    return {
        "one_liner": one_liner[:200],
        "key_technical_point": key_tech[:200],
        "potential_use_case": use_case[:200],
        "confidence": confidence,
    }


def extract_arxiv_summary(title: str, content: str) -> dict:
    """Extract summary from arXiv abstract."""
    text = re.sub(r'<[^>]+>', ' ', content)
    text = re.sub(r'\s+', ' ', text).strip()

    confidence = "high" if len(text) > 300 else "medium"
    one_liner = _find_contribution(text) or f"This paper presents {title.lower()}"
    key_tech = _extract_method(text)
    use_case = _extract_application(text)

    return {
        "one_liner": one_liner[:200],
        "key_technical_point": key_tech[:200],
        "potential_use_case": use_case[:200],
        "confidence": confidence,
    }


def extract_reddit_summary(title: str, content: str) -> dict:
    """Extract summary from Reddit post."""
    text = re.sub(r'<[^>]+>', ' ', content)
    text = re.sub(r'\s+', ' ', text).strip()

    if len(text) > 500:
        confidence = "high"
    elif len(text) > 100:
        confidence = "medium"
    else:
        confidence = "low"

    return {
        "one_liner": (title or text[:150])[:200],
        "key_technical_point": (text or "No additional content in post")[:200],
        "potential_use_case": "AI community discussion",
        "confidence": confidence,
    }


# ── Extraction helpers ─────────────────────────────────────────────────────
def _assess_extraction_quality(one_liner, key_tech, use_case, source_confidence) -> str:
    score = 0
    penalties = 0
    ol = one_liner.strip()
    ol_len = len(ol)

    if 40 <= ol_len <= 200:
        score += 2
    elif 20 <= ol_len < 40:
        score += 1
    elif ol_len > 200:
        penalties += 1

    if ol.endswith(('.', '!', '?', '…')):
        score += 1
    else:
        penalties += 1

    if re.search(r'\b(?:is|are|provides|enables|implements|makes|allows|builds|creates|runs|uses)\b', ol, re.I):
        score += 1
    elif re.match(r'^[A-Z]\w+', ol) and ol_len > 30:
        score += 0.5

    open_brackets = ol.count('[') + ol.count('(')
    close_brackets = ol.count(']') + ol.count(')')
    if abs(open_brackets - close_brackets) > 0:
        penalties += 1
    if open_brackets > 2:
        penalties += 1

    kt = key_tech.strip()
    if kt and len(kt) > 20 and not kt.startswith('See '):
        score += 1
    else:
        penalties += 0.5

    uc = use_case.strip()
    if uc and len(uc) > 10 and not uc.startswith('Relevant for'):
        score += 1
    else:
        penalties += 0.5

    net = score - penalties
    if net >= 3:
        return source_confidence
    elif net >= 1:
        return "medium"
    return "low"


def _is_security_tooling(title: str, one_liner: str, key_tech: str) -> bool:
    combined = f"{title} {one_liner} {key_tech}".lower()
    return any(sig in combined for sig in [
        "offensive", "pentest", "red team", "exploit", "kill chain",
        "attack surface", "vulnerability scan", "zero-day",
        "reverse engineer", "c2", "command and control",
    ])


def _find_project_description(text: str, title: str) -> Optional[str]:
    proj_name = title.split(':')[0].split('/')[0].strip().lower()
    for para in text.split('\n\n'):
        para = para.strip()
        if not para or para.startswith('##') or len(para) < 20:
            continue
        if 'img' in para.lower() or 'badge' in para.lower() or 'shields' in para.lower():
            continue
        if re.match(r'^[~$#€£¥*»\d]', para):
            continue
        special_chars = sum(1 for c in para if not c.isalnum() and not c.isspace() and c not in ',.!?;:\'\"-()[]')
        if special_chars / max(len(para), 1) > 0.4:
            continue
        sentence = re.split(r'[.!?]', para)[0].strip()
        if len(sentence) > 30:
            return sentence + '.'

    for pattern in [
        rf'{re.escape(proj_name[:20])}\s+(?:is|enables|provides|implements)\s+[^.]+\.?',
        r'(?:This\s+)?(?:project|library|framework|tool|package)\s+(?:is|enables|provides)\s+[^.]+\.?',
    ]:
        match = re.search(pattern, text, re.I)
        if match:
            return match.group(0)[:200]
    return None


def _find_contribution(text: str) -> Optional[str]:
    for pattern in [
        r'(?:we|this\s+paper)\s+(?:propose|introduce|present|propose and evaluate)\s+[^.]{10,150}\.',
        r'(?:we\s+(?:show|demonstrate|find|discover|observe))\s+[^.]{10,150}\.',
    ]:
        match = re.search(pattern, text, re.I)
        if match:
            return match.group(0)[:200]
    first = re.split(r'[.!?]', text)[0].strip()
    return first if first else None


def _extract_technical_point(text: str, confidence: str) -> str:
    for pattern in [
        r'architecture(?:\s+designed)?\s+(?:for|to|that)\s+[^.]+\.?',
        r'(?:using|via|based\s+on|through)\s+[a-z][^.]{10,100}\.',
    ]:
        match = re.search(pattern, text, re.I)
        if match:
            return match.group(0)[:200]
    if confidence == "low":
        return "Technical details not available in extracted content"
    return "See README for technical details"


def _extract_method(text: str) -> str:
    for pattern in [
        r'(?:method|approach|framework|technique|model|system)\s+(?:based|using|via|through|with)\s+[a-z][^.]{10,120}\.',
        r'(?:combining|leveraging|exploiting)\s+[a-z][^.]{10,120}\.',
    ]:
        match = re.search(pattern, text, re.I)
        if match:
            return match.group(0)[:200]
    return "See full paper for methodology"


def _extract_use_case(text: str, title: str) -> str:
    for pattern in [
        r'(?:for|to)\s+(?:developers|engineers|researchers|teams)\s+who?\s+[^.]{5,80}\.',
        r'(?:enables|allows|helps)\s+[^\s]+\s+to\s+[^.]{10,80}\.',
    ]:
        match = re.search(pattern, text, re.I)
        if match:
            return match.group(0)[:200]
    return f"Relevant for {title.lower()[:50]} developers and users"


def _extract_application(text: str) -> str:
    title_lower = text[:200].lower()
    if any(k in title_lower for k in ["agent", "agentic"]):
        return "Building AI agent systems"
    if any(k in title_lower for k in ["verification", "verify"]):
        return "LLM output verification and reliability"
    if any(k in title_lower for k in ["embodied", "robot"]):
        return "Embodied AI and robotics applications"
    if any(k in title_lower for k in ["distill"]):
        return "Model distillation and knowledge transfer"
    return "See paper for specific applications"


# ── Pipeline functions ─────────────────────────────────────────────────────
def summarize_entry(entry: dict, conn: sqlite3.Connection) -> bool:
    """Summarize a single entry using rule-based extraction."""
    source = entry["source"]
    title = entry["title"]
    content = entry.get("extracted_text", "")
    eid = entry["id"]

    if not content or len(content) < 50:
        return False

    if source == "github":
        summary = extract_github_summary(title, content)
    elif source == "arxiv":
        summary = extract_arxiv_summary(title, content)
    elif source == "reddit":
        summary = extract_reddit_summary(title, content)
    else:
        summary = extract_reddit_summary(title, content)

    conn.execute("UPDATE entries SET summary = ? WHERE id = ?",
                 (json.dumps(summary), eid))
    conn.commit()
    return True


def run_summarization(source: Optional[str] = None, limit: int = 0) -> None:
    """Summarize all pending entries."""
    conn = sqlite3.connect(str(DB_PATH))
    cur = conn.cursor()

    where = "summary IS NULL"
    params = []
    if source:
        where += " AND source = ?"
        params.append(source)

    cur.execute(f"SELECT COUNT(*) FROM entries WHERE {where}", params)
    total_pending = cur.fetchone()[0]
    print(f"[summarize] {total_pending} pending entries")

    if limit:
        limit_clause = f"LIMIT {limit}"
    else:
        limit_clause = ""

    cur.execute(f"""
        SELECT id, source, title, extracted_text, summary
        FROM entries WHERE {where}
        ORDER BY first_seen DESC
        {limit_clause}
    """, params)

    summarized = 0
    for row in cur.fetchall():
        entry = {
            "id": row[0], "source": row[1], "title": row[2],
            "extracted_text": row[3], "summary": row[4],
        }
        if summarize_entry(entry, conn):
            summarized += 1

    conn.close()
    print(f"[summarize] done — {summarized} entries summarized")