#!/usr/bin/env python3

import asyncio
import hashlib
import re
import signal
from datetime import datetime
from pathlib import Path

import pymysql

from telethon import TelegramClient
from telethon.errors import FloodWaitError


# ============================================================
# CONFIG
# ============================================================

BASE_DIR = Path(__file__).resolve().parent.parent

DB_CONFIG = (
    BASE_DIR
    / "config"
    / "database.php"
)

SESSION_PATH = (
    BASE_DIR
    / "engine"
    / "sessions"
    / "news_reader"
)

PUBLISHER_ID = 1

POLL_SECONDS = 30

#
# Safety cap:
# If a source suddenly has thousands of unseen messages,
# do not consume everything in one cycle.
#
MAX_MESSAGES_PER_SOURCE_PER_CYCLE = 100


# ============================================================
# GLOBAL STATE
# ============================================================

stop_event = None


# ============================================================
# HELPERS
# ============================================================

def log(message):

    now = datetime.now().strftime(
        "%Y-%m-%d %H:%M:%S"
    )

    print(
        f"[{now}] {message}",
        flush=True
    )


def read_php_database_config():

    content = DB_CONFIG.read_text(
        encoding="utf-8",
        errors="ignore"
    )

    patterns = {
        "host":
            r"\$dbHost\s*=\s*['\"]([^'\"]+)['\"]",

        "name":
            r"\$dbName\s*=\s*['\"]([^'\"]+)['\"]",

        "user":
            r"\$dbUser\s*=\s*['\"]([^'\"]+)['\"]",

        "password":
            r"\$dbPass\s*=\s*['\"]([^'\"]*)['\"]",
    }

    result = {}

    for key, pattern in patterns.items():

        match = re.search(
            pattern,
            content
        )

        if not match:
            raise RuntimeError(
                f"Could not read DB config: {key}"
            )

        result[key] = match.group(1)

    return result


DB_SETTINGS = read_php_database_config()


def get_db():

    return pymysql.connect(
        host=DB_SETTINGS["host"],
        user=DB_SETTINGS["user"],
        password=DB_SETTINGS["password"],
        database=DB_SETTINGS["name"],
        charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor,
        autocommit=True,
        connect_timeout=10
    )


def get_publisher():

    db = get_db()

    try:

        with db.cursor() as cursor:

            cursor.execute("""
                SELECT
                    id,
                    telegram_api_id,
                    telegram_api_hash,
                    is_active,
                    health_status,
                    flood_status,
                    flood_until

                FROM publishers

                WHERE id = %s
                LIMIT 1
            """, (
                PUBLISHER_ID,
            ))

            return cursor.fetchone()

    finally:
        db.close()


def get_sources():

    db = get_db()

    try:

        with db.cursor() as cursor:

            cursor.execute("""
                SELECT
                    ns.id AS news_source_id,
                    ns.source_id,
                    ns.priority,
                    ns.last_message_id,

                    s.name,
                    s.username,
                    s.telegram_chat_id

                FROM news_sources ns

                INNER JOIN sources s
                    ON s.id = ns.source_id

                WHERE ns.is_active = 1
                  AND s.is_active = 1
                  AND s.source_kind = 'telegram'

                ORDER BY
                    ns.priority DESC,
                    ns.id ASC
            """)

            return cursor.fetchall()

    finally:
        db.close()


def normalize_text(text):

    text = str(
        text or ""
    ).strip()

    return re.sub(
        r"\s+",
        " ",
        text
    )


def make_hash(text):

    normalized = (
        normalize_text(text)
        .lower()
    )

    if not normalized:
        return None

    return hashlib.sha256(
        normalized.encode("utf-8")
    ).hexdigest()


def detect_media_type(message):

    if getattr(
        message,
        "photo",
        None
    ):
        return "photo"

    if getattr(
        message,
        "voice",
        None
    ):
        return "voice"

    if getattr(
        message,
        "video",
        None
    ):
        return "video"

    if getattr(
        message,
        "audio",
        None
    ):
        return "audio"

    document = getattr(
        message,
        "document",
        None
    )

    if document:

        mime = str(
            getattr(
                document,
                "mime_type",
                ""
            )
            or ""
        ).lower()

        if mime.startswith("video/"):
            return "video"

        if mime.startswith("audio/"):
            return "audio"

        return "document"

    return None


# ============================================================
# DB STATE
# ============================================================

def source_success(
    news_source_id,
    last_message_id
):

    db = get_db()

    try:

        with db.cursor() as cursor:

            cursor.execute("""
                UPDATE news_sources

                SET
                    last_message_id = GREATEST(
                        COALESCE(last_message_id, 0),
                        %s
                    ),

                    last_collected_at = NOW(),
                    last_error = NULL,
                    consecutive_errors = 0

                WHERE id = %s
                LIMIT 1
            """, (
                int(last_message_id),
                int(news_source_id)
            ))

    finally:
        db.close()


def source_error(
    news_source_id,
    error
):

    db = get_db()

    try:

        with db.cursor() as cursor:

            cursor.execute("""
                UPDATE news_sources

                SET
                    last_error = %s,
                    consecutive_errors =
                        consecutive_errors + 1

                WHERE id = %s
                LIMIT 1
            """, (
                str(error)[:4000],
                int(news_source_id)
            ))

    finally:
        db.close()


def update_source_identity(
    source_id,
    entity
):

    telegram_id = getattr(
        entity,
        "id",
        None
    )

    if telegram_id is None:
        return

    #
    # Telethon gives raw channel ID.
    # Existing project may use its own Telegram ID
    # convention, so we only populate NULL values.
    #
    db = get_db()

    try:

        with db.cursor() as cursor:

            cursor.execute("""
                UPDATE sources

                SET
                    telegram_chat_id =
                        COALESCE(
                            telegram_chat_id,
                            %s
                        ),

                    health_status = 'healthy',
                    last_error = NULL

                WHERE id = %s
                LIMIT 1
            """, (
                str(telegram_id),
                int(source_id)
            ))

    finally:
        db.close()


# ============================================================
# INSERT FEED
# ============================================================

def insert_feed_item(
    source,
    message
):

    text = str(
        getattr(
            message,
            "message",
            ""
        )
        or ""
    )

    media_type = detect_media_type(
        message
    )

    if text and media_type:
        content_type = "mixed"

    elif media_type:
        content_type = "media"

    else:
        content_type = "text"

    grouped_id = getattr(
        message,
        "grouped_id",
        None
    )

    views = getattr(
        message,
        "views",
        None
    )

    forwards = getattr(
        message,
        "forwards",
        None
    )

    published_at = None

    if getattr(
        message,
        "date",
        None
    ):

        published_at = (
            message.date
            .replace(tzinfo=None)
            .strftime(
                "%Y-%m-%d %H:%M:%S"
            )
        )

    content_hash = make_hash(
        text
    )

    db = get_db()

    try:

        with db.cursor() as cursor:

            cursor.execute("""
                INSERT IGNORE INTO news_feed_items
                (
                    news_source_id,
                    source_id,
                    source_message_id,
                    telegram_grouped_id,

                    published_at,

                    content_type,
                    text_content,
                    media_type,

                    source_views,
                    source_forwards,

                    content_hash,
                    feed_status
                )

                VALUES
                (
                    %s,
                    %s,
                    %s,
                    %s,

                    %s,

                    %s,
                    %s,
                    %s,

                    %s,
                    %s,

                    %s,
                    'raw'
                )
            """, (
                source["news_source_id"],
                source["source_id"],
                int(message.id),
                grouped_id,

                published_at,

                content_type,
                text,
                media_type,

                views,
                forwards,

                content_hash
            ))

            return cursor.rowcount == 1

    finally:
        db.close()


# ============================================================
# SOURCE COLLECTION
# ============================================================

async def collect_source(
    client,
    source
):

    username = str(
        source.get("username")
        or ""
    ).strip()

    if not username:

        raise RuntimeError(
            "Source has no Telegram username."
        )

    target = (
        "@"
        + username.lstrip("@")
    )

    last_message_id = int(
        source.get("last_message_id")
        or 0
    )

    entity = await client.get_entity(
        target
    )

    update_source_identity(
        source["source_id"],
        entity
    )

    messages = []

    #
    # min_id means Telegram returns only messages
    # newer than our stored state.
    #
    async for message in client.iter_messages(
        entity,
        min_id=last_message_id,
        limit=MAX_MESSAGES_PER_SOURCE_PER_CYCLE
    ):

        if not message:
            continue

        #
        # Ignore Telegram service events.
        #
        if (
            not getattr(
                message,
                "message",
                None
            )
            and
            not getattr(
                message,
                "media",
                None
            )
        ):
            continue

        messages.append(
            message
        )

    if not messages:

        return {
            "target": target,
            "new": 0,
            "inserted": 0,
            "duplicates": 0
        }

    #
    # Telegram returns newest first.
    # Process oldest → newest.
    #
    messages.sort(
        key=lambda item: item.id
    )

    inserted = 0
    duplicates = 0
    highest_message_id = last_message_id

    for message in messages:

        highest_message_id = max(
            highest_message_id,
            int(message.id)
        )

        if insert_feed_item(
            source,
            message
        ):
            inserted += 1
        else:
            duplicates += 1

    #
    # IMPORTANT:
    # State moves only AFTER messages were processed.
    #
    source_success(
        source["news_source_id"],
        highest_message_id
    )

    return {
        "target": target,
        "new": len(messages),
        "inserted": inserted,
        "duplicates": duplicates
    }


# ============================================================
# CYCLE
# ============================================================

async def run_cycle(
    client
):

    sources = get_sources()

    total_new = 0
    total_inserted = 0
    total_errors = 0

    for source in sources:

        if stop_event.is_set():
            break

        try:

            result = await collect_source(
                client,
                source
            )

            total_new += result["new"]
            total_inserted += result["inserted"]

            if result["new"] > 0:

                log(
                    f"{result['target']} | "
                    f"new={result['new']} | "
                    f"inserted={result['inserted']} | "
                    f"duplicates={result['duplicates']}"
                )

        except FloodWaitError as exc:

            seconds = int(
                exc.seconds
            )

            source_error(
                source["news_source_id"],
                f"FloodWait {seconds}s"
            )

            log(
                f"⚠️ FloodWait {seconds}s "
                f"on @{source.get('username')}"
            )

            #
            # Respect Telegram's requested wait.
            #
            try:

                await asyncio.wait_for(
                    stop_event.wait(),
                    timeout=seconds
                )

            except asyncio.TimeoutError:
                pass

        except Exception as exc:

            total_errors += 1

            source_error(
                source["news_source_id"],
                exc
            )

            log(
                f"❌ @{source.get('username')} | "
                f"{type(exc).__name__}: {exc}"
            )

    return {
        "sources": len(sources),
        "new": total_new,
        "inserted": total_inserted,
        "errors": total_errors
    }


# ============================================================
# MAIN
# ============================================================

async def main():

    publisher = get_publisher()

    if not publisher:
        raise RuntimeError(
            "Publisher #1 not found."
        )

    if int(
        publisher.get("is_active")
        or 0
    ) != 1:
        raise RuntimeError(
            "Publisher #1 is inactive."
        )

    api_id = int(
        str(
            publisher.get(
                "telegram_api_id"
            )
            or ""
        ).strip()
    )

    api_hash = str(
        publisher.get(
            "telegram_api_hash"
        )
        or ""
    ).strip()

    if not api_id or not api_hash:
        raise RuntimeError(
            "Telegram API credentials missing."
        )

    client = TelegramClient(
        str(SESSION_PATH),
        api_id,
        api_hash
    )

    await client.connect()

    try:

        if not await client.is_user_authorized():

            raise RuntimeError(
                "news_reader.session "
                "is not authorized."
            )

        log(
            "NEWS COLLECTOR STARTED ✅"
        )

        log(
            f"Poll interval: {POLL_SECONDS}s"
        )

        while not stop_event.is_set():

            try:

                result = await run_cycle(
                    client
                )

                if (
                    result["new"] > 0
                    or result["errors"] > 0
                ):

                    log(
                        "Cycle | "
                        f"sources={result['sources']} | "
                        f"new={result['new']} | "
                        f"inserted={result['inserted']} | "
                        f"errors={result['errors']}"
                    )

            except Exception as exc:

                log(
                    "❌ Cycle Error | "
                    f"{type(exc).__name__}: {exc}"
                )

            try:

                await asyncio.wait_for(
                    stop_event.wait(),
                    timeout=POLL_SECONDS
                )

            except asyncio.TimeoutError:
                pass

    finally:

        await client.disconnect()

        log(
            "NEWS COLLECTOR STOPPED"
        )


def request_stop():

    if not stop_event.is_set():
        stop_event.set()


async def runner():

    global stop_event

    stop_event = asyncio.Event()

    loop = asyncio.get_running_loop()

    for sig in (
        signal.SIGTERM,
        signal.SIGINT
    ):

        try:

            loop.add_signal_handler(
                sig,
                request_stop
            )

        except (
            NotImplementedError,
            RuntimeError
        ):
            pass

    await main()


if __name__ == "__main__":

    asyncio.run(
        runner()
    )