athena-oracle/summarize.py

#!/usr/bin/env python3
"""
AI Research Oracle — Summarization Engine (v1).

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

Output schema: {one_liner, key_technical_point, potential_use_case, confidence}

Architecture note: This v1 uses deterministic extraction rules to avoid
hallucination. When a local LLM becomes available (Ollama GPU, Hermes API),
swap in LLM mode via --llm flag. The DB schema is identical.

Usage:
    python3 summarize.py                      # summarize all pending
    python3 summarize.py --source github       # specific source
    python3 summarize.py --limit 10            # max entries
    python3 summarize.py --verify              # spot-check 2-3 summaries
"""

import argparse
import json
import os
import re
import sqlite3
import sys
import time
from datetime import datetime, timezone

sys.path.insert(0, os.path.dirname(__file__))


def extract_github_summary(title: str, content: str) -> dict:
    """Extract summary from GitHub README content.

    Strategy: Clean HTML, find the first substantive paragraph that
    describes the project (usually below the badges), extract the
    "what it does" sentence.
    """
    # Aggressive HTML cleaning
    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)
    # Remove code blocks (``` ... ```) — often ASCII art
    text = re.sub(r'```[\s\S]*?```', '', text)
    text = re.sub(r'\n\s*\n+', '\n\n', text)
    text = text.strip()

    # Confidence starts from source content quality
    source_confidence = "low"
    if len(text) > 2000:
        source_confidence = "high"
    elif len(text) > 500:
        source_confidence = "medium"

    # Find the one-liner: look for project description paragraph
    one_liner = _find_project_description(text, title)
    if not one_liner:
        one_liner = title[:200]

    # Key technical point
    key_tech = _extract_technical_point(text, source_confidence)

    # Use case
    use_case = _extract_use_case(text, title)

    # Quality-gate confidence on extraction signals, not raw length
    confidence = _assess_extraction_quality(one_liner, key_tech, use_case, source_confidence)

    # Tag security tooling if detected
    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.

    Strategy: arXiv abstracts have a predictable structure:
    1. Background/motivation
    2. "In this paper we propose..."
    3. Results
    4. Implications

    We extract the contribution statement and key finding.
    """
    text = re.sub(r'<[^>]+>', ' ', content)
    text = re.sub(r'\s+', ' ', text).strip()

    # Confidence based on abstract clarity
    confidence = "high" if len(text) > 300 else "medium"

    # One-liner: find the contribution statement
    one_liner = _find_contribution(text)
    if not one_liner:
        # Fallback: use title as base
        one_liner = f"This paper presents {title.lower()}"

    # Key technical point: look for method description
    key_tech = _extract_method(text)

    # Use case: look for application statements
    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.

    Strategy: Reddit posts vary wildly in quality. Extract the core
    question or claim, note if it's discussion vs announcement.
    """
    text = re.sub(r'<[^>]+>', ' ', content)
    text = re.sub(r'\s+', ' ', text).strip()

    # Confidence based on content length
    if len(text) > 500:
        confidence = "high"
    elif len(text) > 100:
        confidence = "medium"
    else:
        confidence = "low"

    # One-liner from title (Reddit titles are usually the summary)
    one_liner = title[:200] if title else text[:150]

    # Key technical point from content
    key_tech = text[:200] if text else "No additional content in post"

    # Use case: community relevance
    use_case = "AI community discussion"

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


# --- Extraction helpers ---

def _assess_extraction_quality(one_liner: str, key_tech: str, use_case: str, source_confidence: str) -> str:
    """Assess extraction quality based on output signals, not source length.

    A short-but-complete Reddit title should score higher confidence
    than a long README that yielded a fragment.
    """
    score = 0
    penalties = 0

    # One-liner quality
    ol = one_liner.strip()
    ol_len = len(ol)

    # Length window: 40-200 chars is a reasonable sentence
    if 40 <= ol_len <= 200:
        score += 2
    elif 20 <= ol_len < 40:
        score += 1
    elif ol_len > 200:
        penalties += 1  # too long, likely grabbed too much

    # Ends with terminal punctuation
    if ol.endswith(('.', '!', '?', '…')):
        score += 1
    else:
        penalties += 1

    # Contains subject-verb pattern (basic heuristic)
    if re.search(r'\b(?:is|are|provides|enables|implements|makes|allows|builds|creates|runs|uses)\b', ol, re.I):
        score += 1
    # Or starts with a proper noun/capitalized phrase
    elif re.match(r'^[A-Z]\w+', ol) and ol_len > 30:
        score += 0.5

    # No unmatched brackets (artifact from markdown/HTML)
    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  # likely grabbed markdown link syntax

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

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

    # Final confidence based on score - penalties
    net = score - penalties
    if net >= 3:
        return source_confidence  # extraction is good, trust source quality
    elif net >= 1:
        return "medium"
    else:
        return "low"


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


def _find_project_description(text: str, title: str) -> str | None:
    """Find the project description paragraph in a README."""
    paras = text.split('\n\n')
    proj_name = title.split(':')[0].split('/')[0].strip().lower()

    for para in paras:
        para = para.strip()
        if not para or para.startswith('##') or len(para) < 20:
            continue
        # Skip badges, stats lines, separator lines
        if 'img' in para.lower() or 'badge' in para.lower() or 'shields' in para.lower():
            continue
        # Skip lines that start with stats (~54%, etc.)
        if re.match(r'^[~$#€£¥*»\d]', para):
            continue
        # Skip ASCII art (high ratio of special chars)
        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
        if len(para) < 40:
            continue
        # Good paragraph — extract first sentence
        sentence = re.split(r'[.!?]', para)[0].strip()
        if len(sentence) > 30:
            return sentence + '.'

    # Fallback: look for "is a" pattern anywhere
    patterns = [
        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+[^.]+\.?',
    ]
    for pattern in patterns:
        match = re.search(pattern, text, re.I)
        if match:
            return match.group(0)[:200]

    return None


def _find_what_sentence(text: str, title: str) -> str | None:
    """Find the 'X is a...' sentence that describes what the project does."""
    patterns = [
        rf'{re.escape(title[:30])}\s+(?:is|enables|provides|implements)\s+[^.]+\.?',
        r'(?:This\s+)?(?:project|library|framework|tool|package)\s+(?:is|enables|provides|implements)\s+[^.]+\.?',
        r'(?:makes|allows)\s+[^\s]+\s+(?:to|can)\s+[^.]+\.?',
        r'(?:\w+\s+(?:is|provides|enables|implements|delivers))\s+[a-z].{10,100}\.',
    ]

    for pattern in patterns:
        match = re.search(pattern, text, re.I)
        if match:
            return match.group(0)[:200]

    # Fallback: first meaningful paragraph
    for para in text.split('\n\n'):
        para = para.strip()
        if len(para) > 30 and not para.startswith('#'):
            return para[:200]
    return None


def _find_contribution(text: str) -> str | None:
    """Find the 'we propose/introduce/present' statement in an abstract."""
    patterns = [
        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}\.',
        r'(?:we\s+(?:introduce|present|propose))\s+(?:a|an|our)\s+\w+\s+[^.]{5,150}\.',
    ]

    for pattern in patterns:
        match = re.search(pattern, text, re.I)
        if match:
            return match.group(0)[:200]

    # Fallback: first sentence
    first = re.split(r'[.!?]', text)[0].strip()
    return first if first else None


def _extract_technical_point(text: str, confidence: str) -> str:
    """Extract the main technical approach or innovation."""
    patterns = [
        r'architecture(?:\s+designed)?\s+(?:for|to|that)\s+[^.]+\.?',
        r'(?:using|via|based\s+on|through)\s+[a-z][^.]{10,100}\.',
        r'(?:novel|new|unique|innovative)\s+\w+\s+[^.]{5,80}\.',
    ]

    for pattern in patterns:
        match = re.search(pattern, text, re.I)
        if match:
            return match.group(0)[:200]

    # Fallback: confidence-based
    if confidence == "low":
        return "Technical details not available in extracted content"
    return "See README for technical details"


def _extract_method(text: str) -> str:
    """Extract the method/approach from an arXiv abstract."""
    patterns = [
        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}\.',
        r'(?:learn|train|optimize|generate)\s+[a-z][^.]{10,120}\.',
    ]

    for pattern in patterns:
        match = re.search(pattern, text, re.I)
        if match:
            return match.group(0)[:200]

    # Fallback: core contribution
    for pattern in [
        r'(?:propose|introduce)\s+(?:a|an)\s+[^.]{10,100}\.',
    ]:
        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:
    """Extract potential use case from README content."""
    patterns = [
        r'(?:for|to)\s+(?:developers|engineers|researchers|teams)\s+who?\s+[^.]{5,80}\.',
        r'(?:enables|allows|helps)\s+[^\s]+\s+to\s+[^.]{10,80}\.',
        r'(?:use\s+case|application|target\s+user)\s*:\s*[^.]{10,80}\.',
    ]

    for pattern in patterns:
        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:
    """Extract application/use case from arXiv abstract."""
    patterns = [
        r'(?:application|use\s+case|can\s+be\s+used|could\s+be\s+applied)\s+(?:for|in|to)\s+[a-z][^.]{10,80}\.',
        r'(?:improve|enhance|advance)\s+[a-z][^.]{10,80}\.',
    ]

    for pattern in patterns:
        match = re.search(pattern, text, re.I)
        if match:
            return match.group(0)[:200]

    # Generic fallback based on title keywords
    title_lower = text[:200].lower()
    if any(k in title_lower for k in ["agent", "agentic"]):
        return "Building AI agent systems"
    elif any(k in title_lower for k in ["verification", "verify"]):
        return "LLM output verification and reliability"
    elif any(k in title_lower for k in ["embodied", "robot"]):
        return "Embodied AI and robotics applications"
    elif any(k in title_lower for k in ["distill"]):
        return "Model distillation and knowledge transfer"
    return "See paper for specific applications"


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

    # Source-specific extraction
    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)  # fallback

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


def verify_summaries(conn: sqlite3.Connection, source: str, sample_size: int = 3):
    """Spot-check summaries against source text.

    Look for hallucinated specifics: numbers, claims, features not
    present in the original extracted_text.

    NOTE: Rule-based extraction v1 is inherently lower-risk for
    hallucination since it extracts actual text, not generates new claims.
    But we still verify the extraction logic is working correctly.
    """
    cur = conn.cursor()
    cur.execute("""
        SELECT id, title, extracted_text, summary
        FROM entries WHERE source = ? AND summary IS NOT NULL
        ORDER BY RANDOM()
        LIMIT ?
    """, (source, sample_size))

    rows = cur.fetchall()
    if not rows:
        print(f"  No summaries to verify for {source}")
        return

    for eid, title, source_text, summary_json in rows:
        summary = json.loads(summary_json)
        one_liner = summary.get("one_liner", "")
        confidence = summary.get("confidence", "?")

        issues = []

        # Check: does the one-liner contain text actually present in source?
        # (For rule-based extraction, this should always be true)
        words = one_liner.split()[:5]
        found = sum(1 for w in words if w.lower() in source_text.lower())
        if found < 3:
            issues.append(f"Low overlap: {found}/5 words from source")

        # Check: confidence matches content length
        if confidence == "high" and len(source_text) < 500:
            issues.append("High confidence on short source")
        elif confidence == "low" and len(source_text) > 2000:
            issues.append("Low confidence on long source")

        if issues:
            print(f"  ⚠ [{eid}] {title[:50]}... issues: {'; '.join(issues)}")
            print(f"     Summary: {one_liner[:80]}...")
        else:
            print(f"  ✓ [{eid}] {title[:50]}... confidence={confidence}")

        time.sleep(0.3)


def main():
    parser = argparse.ArgumentParser(description="AI Research Oracle — Summarization")
    parser.add_argument("--source", default=None, help="Filter by source (github/arxiv/reddit)")
    parser.add_argument("--limit", type=int, default=0, help="Max entries (0=all)")
    parser.add_argument("--verify", action="store_true", help="Spot-check summaries")
    args = parser.parse_args()

    db_path = os.path.join(os.path.dirname(__file__), "oracle.db")
    conn = sqlite3.connect(db_path)
    cur = conn.cursor()

    # Find pending entries
    where = "summary IS NULL"
    params = []
    if args.source:
        where += " AND source = ?"
        params.append(args.source)

    cur.execute(f"SELECT COUNT(*) FROM entries WHERE {where}", params)
    total_pending = cur.fetchone()[0]
    print(f"=== Summarization Engine (Rule-based v1) ===")
    print(f"  Pending entries: {total_pending}")

    if total_pending == 0:
        print("  Nothing to summarize.")
        conn.close()
        return

    # Fetch entries
    limit_clause = " LIMIT ?" if args.limit > 0 else ""
    limit_params = params + [args.limit] if args.limit > 0 else params

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

    entries = [{"id": r[0], "source": r[1], "title": r[2], "extracted_text": r[3]} for r in cur.fetchall()]
    print(f"  Processing: {len(entries)} entries")
    print()

    success = 0
    failed = 0
    for entry in entries:
        try:
            if summarize_entry(entry, conn):
                success += 1
                print(f"  ✓ [{entry['id']}] {entry['title'][:60]}... ({entry['source']})")
            else:
                failed += 1
                print(f"  ⚠ [{entry['id']}] Skipped: {entry['title'][:40]}... (too short)")
        except Exception as e:
            print(f"  ✗ [{entry['id']}] Error: {e}")
            failed += 1

    if args.verify:
        print(f"\n  [Verification]")
        sources = [args.source] if args.source else ["github", "arxiv", "reddit"]
        for src in sources:
            print(f"    Checking {src}...")
            verify_summaries(conn, src)
        print()

    print(f"  Results: {success} summarized, {failed} failed")
    conn.close()
    print(f"\n  Done.")


if __name__ == "__main__":
    main()