main.py

"""InhumanScience — automated AI/ML content curation and publishing pipeline."""

from __future__ import annotations

import logging
import os
import signal
import sys
import time
from pathlib import Path

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
import pytz

import config
from sources.alphaxiv import fetch_trending_papers
from sources.blogs import fetch_blog_posts, fetch_full_blog_content
from sources.twitter_feed import fetch_ai_leader_tweets
from oracle.oracle import evaluate_content, verify_content, is_duplicate
from processors.pdf import download_pdf, extract_text
from processors.images import extract_best_figure
from processors.post_generator import (
    generate_paper_post_ru,
    generate_paper_post_en,
    generate_blog_post_ru,
    generate_blog_post_en,
    generate_tweet_summary_ru,
)
from publishers.telegram import send_post_with_image, send_error, send_status
from publishers.twitter import post_tweet, retweet
from storage.state import (
    is_paper_posted,
    mark_paper_posted,
    is_blog_posted,
    mark_blog_posted,
    is_tweet_posted,
    mark_tweet_posted,
)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
)
logger = logging.getLogger("main")


def _parse_cron(expr: str) -> dict:
    """Parse '0 10 * * *' into CronTrigger kwargs."""
    parts = expr.strip().split()
    if len(parts) != 5:
        raise ValueError(f"Invalid cron expression: {expr}")
    return dict(
        minute=parts[0], hour=parts[1], day=parts[2],
        month=parts[3], day_of_week=parts[4],
    )


# ---------------------------------------------------------------------------
# Pipeline: Papers
# ---------------------------------------------------------------------------

def run_papers_pipeline() -> None:
    logger.info("=== Papers pipeline started ===")
    try:
        papers = fetch_trending_papers(max_papers=config.ORACLE_MAX_PAPERS_PER_RUN * 3)
        logger.info("Fetched %d candidate papers from AlphaRxiv", len(papers))

        published = 0
        for item in papers:
            if published >= config.ORACLE_MAX_PAPERS_PER_RUN:
                break
            if is_paper_posted(item.content_id):
                logger.debug("Already posted: %s", item.content_id)
                continue

            score, should_publish, reason = evaluate_content(item)
            if not should_publish:
                logger.info("Skipping (score=%.1f): %s", score, item.title[:60])
                continue

            dup, dup_of = is_duplicate(item)
            if dup:
                logger.info("Skipping duplicate paper: %s ~ %s", item.title[:60], dup_of)
                continue

            try:
                pdf_path = download_pdf(item.content_id, item.pdf_url)
                paper_text = extract_text(pdf_path)
                figure_path = extract_best_figure(pdf_path)

                authors_str = ", ".join(item.organizations or item.authors)
                post_ru = generate_paper_post_ru(paper_text, item.title, authors_str)
                post_en = generate_paper_post_en(paper_text, item.title, authors_str)

                tg_msg_id = send_post_with_image(post_ru, figure_path, item.url)
                tweet_id = post_tweet(post_en, figure_path, item.url)

                mark_paper_posted(
                    item.content_id, item.source_name, item.title,
                    tg_msg_id=tg_msg_id or "", tweet_id=tweet_id or "",
                )
                published += 1
                logger.info("Published paper: %s", item.title[:60])

            except Exception:
                logger.exception("Failed to process paper %s", item.content_id)
                send_error(f"Paper pipeline error: {item.content_id}")

    except Exception:
        logger.exception("Papers pipeline crashed")
        send_error("Papers pipeline crashed")

    n = published if "published" in dir() else 0
    logger.info("=== Papers pipeline done (%d published) ===", n)
    send_status(f"Papers pipeline done: {n} published")


# ---------------------------------------------------------------------------
# Pipeline: Blogs
# ---------------------------------------------------------------------------

def run_blogs_pipeline() -> None:
    logger.info("=== Blogs pipeline started ===")
    try:
        posts = fetch_blog_posts(max_age_days=3)
        logger.info("Fetched %d blog posts", len(posts))

        published = 0
        for item in posts:
            if published >= config.ORACLE_MAX_BLOGS_PER_RUN:
                break
            if is_blog_posted(item.content_id):
                continue

            full_content = fetch_full_blog_content(item.url)
            if full_content:
                item.full_text = full_content
                item.summary = full_content[:2000]

            score, should_publish, reason = evaluate_content(item)
            if not should_publish:
                logger.info("Skipping blog (score=%.1f): %s", score, item.title[:60])
                continue

            verified, confidence, issues = verify_content(item)
            if not verified and confidence > 0.6:
                logger.warning("Blog fact-check failed: %s — %s", item.title[:60], issues)
                continue

            dup, dup_of = is_duplicate(item)
            if dup:
                logger.info("Skipping duplicate blog: %s ~ %s", item.title[:60], dup_of)
                continue

            try:
                source_label = item.source_name.replace("_", " ").title()
                content = item.full_text or item.summary
                post_ru = generate_blog_post_ru(item.title, source_label, content)
                post_en = generate_blog_post_en(item.title, source_label, content)

                tg_msg_id = send_post_with_image(post_ru, link=item.url)
                tweet_id = post_tweet(post_en, link=item.url)

                mark_blog_posted(
                    item.content_id, item.source_name, item.title,
                    tg_msg_id=tg_msg_id or "", tweet_id=tweet_id or "",
                )
                published += 1
                logger.info("Published blog: %s", item.title[:60])

            except Exception:
                logger.exception("Failed to process blog %s", item.url)
                send_error(f"Blog pipeline error: {item.url}")

    except Exception:
        logger.exception("Blogs pipeline crashed")
        send_error("Blogs pipeline crashed")

    n = published if "published" in dir() else 0
    logger.info("=== Blogs pipeline done (%d published) ===", n)
    send_status(f"Blogs pipeline done: {n} published")


# ---------------------------------------------------------------------------
# Pipeline: Twitter monitoring
# ---------------------------------------------------------------------------

def run_twitter_pipeline() -> None:
    logger.info("=== Twitter monitoring pipeline started ===")
    try:
        tweets = fetch_ai_leader_tweets(max_age_days=2)
        logger.info("Fetched %d tweets from AI leaders", len(tweets))

        for item in tweets:
            if is_tweet_posted(item.content_id):
                continue

            score, should_publish, reason = evaluate_content(item)
            if not should_publish:
                continue

            verified, confidence, issues = verify_content(item)
            if not verified and confidence > 0.6:
                logger.warning("Tweet fact-check failed: %s", item.title[:60])
                continue

            dup, dup_of = is_duplicate(item)
            if dup:
                logger.info("Skipping duplicate tweet: %s ~ %s", item.title[:60], dup_of)
                continue

            try:
                author = item.authors[0] if item.authors else item.source_name
                post_ru = generate_tweet_summary_ru(author, item.summary)

                tg_msg_id = send_post_with_image(post_ru, link=item.url)
                rt_id = retweet(item.url) if item.url else None

                mark_tweet_posted(
                    item.content_id, author,
                    tg_msg_id=tg_msg_id or "",
                    our_tweet_id=rt_id or "",
                )
                logger.info("Published tweet summary: %s (rt=%s)", item.title[:60], rt_id)

            except Exception:
                logger.exception("Failed to process tweet %s", item.content_id)
                send_error(f"Tweet pipeline error: {item.content_id}")

    except Exception:
        logger.exception("Twitter pipeline crashed")
        send_error("Twitter pipeline crashed")

    logger.info("=== Twitter monitoring pipeline done ===")
    send_status("Twitter monitoring pipeline done")


# ---------------------------------------------------------------------------
# CLI entry point
# ---------------------------------------------------------------------------

def main() -> None:
    os.makedirs(config.PDF_DIR, exist_ok=True)
    os.makedirs(config.IMG_DIR, exist_ok=True)

    if len(sys.argv) > 1:
        cmd = sys.argv[1]
        if cmd == "papers":
            run_papers_pipeline()
        elif cmd == "blogs":
            run_blogs_pipeline()
        elif cmd == "twitter":
            run_twitter_pipeline()
        elif cmd == "all":
            run_papers_pipeline()
            run_blogs_pipeline()
            run_twitter_pipeline()
        else:
            print(f"Unknown command: {cmd}")
            print("Usage: python main.py [papers|blogs|twitter|all|serve]")
            sys.exit(1)
        return

    tz = pytz.timezone(config.TIMEZONE)
    scheduler = BackgroundScheduler()

    papers_cron = _parse_cron(config.SCHEDULE_PAPERS_CRON)
    blogs_cron = _parse_cron(config.SCHEDULE_BLOGS_CRON)
    twitter_cron = _parse_cron(config.SCHEDULE_TWITTER_CRON)

    scheduler.add_job(run_papers_pipeline, CronTrigger(timezone=tz, **papers_cron), id="papers")
    scheduler.add_job(run_blogs_pipeline, CronTrigger(timezone=tz, **blogs_cron), id="blogs")
    scheduler.add_job(run_twitter_pipeline, CronTrigger(timezone=tz, **twitter_cron), id="twitter")

    scheduler.start()
    logger.info(
        "Scheduler running (papers=%s, blogs=%s, twitter=%s, tz=%s)",
        config.SCHEDULE_PAPERS_CRON, config.SCHEDULE_BLOGS_CRON,
        config.SCHEDULE_TWITTER_CRON, config.TIMEZONE,
    )

    def _shutdown(signum, frame):
        logger.info("Shutting down scheduler...")
        scheduler.shutdown(wait=False)
        sys.exit(0)

    signal.signal(signal.SIGINT, _shutdown)
    signal.signal(signal.SIGTERM, _shutdown)

    while True:
        time.sleep(60)


if __name__ == "__main__":
    main()