#!/usr/bin/env python3

import asyncio
import hashlib
import json
import re
import statistics
import sys
from pathlib import Path

import pymysql

from telethon import TelegramClient
from telethon.errors import FloodWaitError

BASE_DIR = Path(__file__).resolve().parent.parent
DB_CONFIG = BASE_DIR / "config" / "database.php"
SESSION_DIR = BASE_DIR / "engine" / "sessions"

PUBLISHER_ID = 1

# هر Source در هر Run حداکثر این تعداد پیام قدیمی‌تر خوانده می‌شود.
BACKFILL_BATCH = 1000


def fail(message: str):
    print("")
    print("❌ ERROR")
    print(message)
    sys.exit(1)


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:
            fail(
                f"Could not read {key} "
                "from config/database.php"
            )

        result[key] = match.group(1)

    return result


def get_database():
    cfg = read_php_database_config()

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


def get_publisher(db):
    with db.cursor() as cursor:
        cursor.execute(
            """
            SELECT *
            FROM publishers
            WHERE id = %s
              AND is_active = 1
            LIMIT 1
            """,
            (PUBLISHER_ID,)
        )

        return cursor.fetchone()


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

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

    session_name = str(
        publisher.get("session_name")
        or f"publisher_{PUBLISHER_ID}"
    ).strip()

    session_name = re.sub(
        r"[^a-zA-Z0-9_-]",
        "_",
        session_name
    )

    return TelegramClient(
        str(SESSION_DIR / session_name),
        api_id,
        api_hash
    )


def get_group(db, group_key):
    with db.cursor() as cursor:
        cursor.execute(
            """
            SELECT *
            FROM content_groups
            WHERE group_key = %s
              AND is_active = 1
            LIMIT 1
            """,
            (group_key,)
        )

        return cursor.fetchone()


def get_approved_sources(db, group_id):
    with db.cursor() as cursor:
        cursor.execute(
            """
            SELECT
                s.id,
                s.name,
                s.username,
                s.telegram_chat_id
            FROM source_group_sources sgs

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

            INNER JOIN source_discovery_candidates sdc
                ON sdc.approved_source_id = s.id
               AND sdc.group_id = sgs.group_id

            LEFT JOIN content_collection_state ccs
                ON ccs.group_id = sgs.group_id
               AND ccs.source_id = s.id

            WHERE sgs.group_id = %s
              AND sgs.source_role = 'approved'
              AND sgs.is_active = 1
              AND s.is_active = 1
              AND s.source_kind = 'telegram'
              AND sdc.review_status = 'approved'
              AND sdc.join_status IN ('joined','archived')

            ORDER BY
                CASE
                    WHEN COALESCE(ccs.backfill_complete, 0) = 0
                    THEN 0
                    ELSE 1
                END ASC,
                COALESCE(ccs.last_run_at, '1970-01-01') ASC,
                s.id ASC

            LIMIT 1
            """,
            (group_id,)
        )

        return cursor.fetchall()


def get_state(db, group_id, source_id):
    with db.cursor() as cursor:

        cursor.execute(
            """
            INSERT INTO content_collection_state
            (
                group_id,
                source_id
            )
            VALUES
            (
                %s, %s
            )
            ON DUPLICATE KEY UPDATE
                source_id = VALUES(source_id)
            """,
            (
                group_id,
                source_id
            )
        )

        cursor.execute(
            """
            SELECT *
            FROM content_collection_state
            WHERE group_id = %s
              AND source_id = %s
            LIMIT 1
            """,
            (
                group_id,
                source_id
            )
        )

        return cursor.fetchone()


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

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

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

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

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

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

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

    return None


def content_type(message):
    text = (
        getattr(message, "message", None)
        or ""
    ).strip()

    media = media_type(message)

    if text and media:
        return "mixed"

    if media:
        return "media"

    return "text"


def normalize_text(text):
    text = text or ""

    text = text.strip().lower()

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

    # Source credits / whitespace differences
    # should not create different exact-text hashes.
    return text


def build_dedupe_hash(message):
    text = normalize_text(
        getattr(message, "message", None)
        or ""
    )

    mtype = media_type(message) or "text"

    if text:
        raw = (
            "text:"
            + text
        )
    else:
        # Media-only V1:
        # same-source duplicate is also prevented by source_message_id.
        raw = (
            "media:"
            + mtype
            + ":"
            + str(message.id)
        )

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


def source_reference(message, source_id):
    """
    We intentionally do NOT download media here.

    Later publisher can retrieve the original Telegram
    message by source + source_message_id and copy its media.
    """

    data = {
        "provider": "telegram",
        "source_id": source_id,
        "message_id": int(message.id),
    }

    return json.dumps(
        data,
        ensure_ascii=False
    )


def calculate_scores(messages):
    views = [
        int(
            getattr(message, "views", 0)
            or 0
        )
        for message in messages
        if int(
            getattr(message, "views", 0)
            or 0
        ) > 0
    ]

    median_views = (
        float(statistics.median(views))
        if views
        else 0.0
    )

    result = {}

    for message in messages:

        msg_views = int(
            getattr(message, "views", 0)
            or 0
        )

        forwards = int(
            getattr(message, "forwards", 0)
            or 0
        )

        if median_views > 0:
            relative = (
                msg_views
                / median_views
            )
        else:
            relative = 0.0

        # Relative performance is the main signal.
        score = relative * 100.0

        # Small forward bonus.
        score += min(
            forwards,
            1000
        ) * 0.15

        text = (
            getattr(message, "message", None)
            or ""
        ).strip()

        if text:
            score += 5.0

        if media_type(message):
            score += 3.0

        result[int(message.id)] = round(
            score,
            4
        )

    return result, median_views


def post_exists(
    db,
    source_id,
    source_message_id
):
    with db.cursor() as cursor:
        cursor.execute(
            """
            SELECT id
            FROM posts
            WHERE source_id = %s
              AND source_message_id = %s
            ORDER BY id ASC
            LIMIT 1
            """,
            (
                source_id,
                source_message_id
            )
        )

        row = cursor.fetchone()

        return (
            int(row["id"])
            if row
            else 0
        )


def find_post_by_hash(
    db,
    dedupe_hash
):
    with db.cursor() as cursor:
        cursor.execute(
            """
            SELECT id
            FROM posts
            WHERE dedupe_hash = %s
            ORDER BY id ASC
            LIMIT 1
            """,
            (dedupe_hash,)
        )

        row = cursor.fetchone()

        return (
            int(row["id"])
            if row
            else 0
        )


def insert_post(
    db,
    source_id,
    message,
    score,
    dedupe_hash
):
    text = (
        getattr(message, "message", None)
        or ""
    ).strip()

    published_at = None

    if getattr(
        message,
        "date",
        None
    ):
        published_at = (
            message.date
            .replace(tzinfo=None)
        )

    with db.cursor() as cursor:

        cursor.execute(
            """
            INSERT INTO posts
            (
                source_id,
                source_message_id,
                telegram_grouped_id,
                source_published_at,
                content_type,
                text_content,
                media_type,
                media_file_id,
                source_views,
                source_forwards,
                score,
                dedupe_hash,
                status
            )
            VALUES
            (
                %s, %s, %s, %s,
                %s, %s, %s, %s,
                %s, %s, %s, %s,
                'collected'
            )
            """,
            (
                source_id,
                int(message.id),
                (
                    int(message.grouped_id)
                    if getattr(
                        message,
                        "grouped_id",
                        None
                    )
                    else None
                ),
                published_at,
                content_type(message),
                text or None,
                media_type(message),
                (
                    source_reference(
                        message,
                        source_id
                    )
                    if media_type(message)
                    else None
                ),
                int(
                    getattr(
                        message,
                        "views",
                        0
                    )
                    or 0
                ),
                int(
                    getattr(
                        message,
                        "forwards",
                        0
                    )
                    or 0
                ),
                score,
                dedupe_hash,
            )
        )

        return int(
            cursor.lastrowid
        )


def attach_to_pool(
    db,
    group_id,
    post_id,
    score
):
    with db.cursor() as cursor:

        cursor.execute(
            """
            INSERT INTO content_group_posts
            (
                group_id,
                post_id,
                pool_score,
                pool_status
            )
            VALUES
            (
                %s,
                %s,
                %s,
                'candidate'
            )

            ON DUPLICATE KEY UPDATE
                pool_score =
                    GREATEST(
                        pool_score,
                        VALUES(pool_score)
                    ),
                updated_at =
                    CURRENT_TIMESTAMP
            """,
            (
                group_id,
                post_id,
                score
            )
        )


def update_source_success(
    db,
    source_id
):
    with db.cursor() as cursor:
        cursor.execute(
            """
            UPDATE sources
            SET
                last_collected_at = NOW(),
                last_success_at = NOW(),
                last_error = NULL,
                last_error_at = NULL,
                health_status = 'healthy'
            WHERE id = %s
            LIMIT 1
            """,
            (source_id,)
        )


def update_source_error(
    db,
    source_id,
    error
):
    with db.cursor() as cursor:
        cursor.execute(
            """
            UPDATE sources
            SET
                last_error_at = NOW(),
                last_error = %s,
                health_status = 'error'
            WHERE id = %s
            LIMIT 1
            """,
            (
                error[:2000],
                source_id
            )
        )


def update_state_success(
    db,
    group_id,
    source_id,
    oldest_message_id,
    newest_message_id,
    scanned,
    collected,
    complete
):
    with db.cursor() as cursor:
        cursor.execute(
            """
            UPDATE content_collection_state
            SET
                oldest_message_id = %s,

                newest_message_id =
                    CASE
                        WHEN newest_message_id IS NULL
                        THEN %s
                        ELSE GREATEST(
                            newest_message_id,
                            %s
                        )
                    END,

                total_scanned =
                    total_scanned + %s,

                total_collected =
                    total_collected + %s,

                backfill_complete = %s,

                last_run_at = NOW(),
                last_success_at = NOW(),

                last_error = NULL,
                last_error_at = NULL

            WHERE group_id = %s
              AND source_id = %s
            LIMIT 1
            """,
            (
                oldest_message_id,
                newest_message_id,
                newest_message_id,
                scanned,
                collected,
                1 if complete else 0,
                group_id,
                source_id
            )
        )


def update_state_error(
    db,
    group_id,
    source_id,
    error
):
    with db.cursor() as cursor:
        cursor.execute(
            """
            UPDATE content_collection_state
            SET
                last_run_at = NOW(),
                last_error_at = NOW(),
                last_error = %s
            WHERE group_id = %s
              AND source_id = %s
            LIMIT 1
            """,
            (
                error[:2000],
                group_id,
                source_id
            )
        )


async def collect_source(
    client,
    db,
    group_id,
    source
):
    source_id = int(
        source["id"]
    )

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

    if not username:
        return {
            "scanned": 0,
            "collected": 0,
            "duplicate": 0,
        }

    state = get_state(
        db,
        group_id,
        source_id
    )

    max_id = int(
        state.get(
            "oldest_message_id"
        )
        or 0
    )

    print("")
    print(
        f"[SOURCE #{source_id}] "
        f"@{username}"
    )

    backfill_complete = int(
        state.get("backfill_complete") or 0
    ) == 1

    entity = await client.get_entity(
        username
    )

    messages = []
    raw_scanned = 0

    kwargs = {
        "limit": BACKFILL_BATCH
    }

    if backfill_complete:
        newest_id = int(
            state.get("newest_message_id") or 0
        )

        if newest_id > 0:
            kwargs["min_id"] = newest_id

        print("  Mode:          INCREMENTAL")

    else:
        if max_id > 0:
            kwargs["max_id"] = max_id

        print("  Mode:          BACKFILL")

    async for message in client.iter_messages(
        entity,
        **kwargs
    ):
        if not message:
            continue

        raw_scanned += 1

        text = (
            getattr(
                message,
                "message",
                None
            )
            or ""
        ).strip()

        mtype = media_type(
            message
        )

        # Ignore service/empty messages, but still count
        # them for detecting the real end of history.
        if not text and not mtype:
            continue

        messages.append(
            message
        )

    if not messages:

        if backfill_complete:

            update_source_success(
                db,
                source_id
            )

            print("  No new messages.")

            return {
                "scanned": raw_scanned,
                "collected": 0,
                "duplicate": 0,
            }

        update_state_success(
            db,
            group_id,
            source_id,
            max_id or None,
            state.get("newest_message_id"),
            raw_scanned,
            0,
            True
        )

        update_source_success(
            db,
            source_id
        )

        print(
            "  No older messages. "
            "Backfill COMPLETE."
        )

        return {
            "scanned": raw_scanned,
            "collected": 0,
            "duplicate": 0,
        }

    scores, median_views = (
        calculate_scores(
            messages
        )
    )

    collected = 0
    duplicate = 0

    message_ids = [
        int(message.id)
        for message in messages
    ]

    oldest_id = min(
        message_ids
    )

    newest_id = max(
        message_ids
    )

    for message in messages:

        message_id = int(
            message.id
        )

        score = scores.get(
            message_id,
            0.0
        )

        # -------------------------------------------------
        # Group-specific pool eligibility
        # -------------------------------------------------
        #
        # Music Group ID 7:
        # only real Telegram audio/music posts may enter
        # the Music pool.
        #
        # Mixed music sources are allowed. Their text,
        # photos, videos, voices, documents, etc. can still
        # exist in the general posts archive, but they must
        # NOT enter content_group_posts for Music.
        #
        message_media_type = media_type(
            message
        )

        pool_eligible = True

        if int(group_id) == 7:
            pool_eligible = (
                message_media_type == "audio"
            )

        existing_id = post_exists(
            db,
            source_id,
            message_id
        )

        if existing_id:

            if pool_eligible:
                attach_to_pool(
                    db,
                    group_id,
                    existing_id,
                    score
                )

            duplicate += 1
            continue

        dedupe_hash = (
            build_dedupe_hash(
                message
            )
        )

        duplicate_post_id = (
            find_post_by_hash(
                db,
                dedupe_hash
            )
        )

        if duplicate_post_id:

            if pool_eligible:
                attach_to_pool(
                    db,
                    group_id,
                    duplicate_post_id,
                    score
                )

            duplicate += 1
            continue

        post_id = insert_post(
            db,
            source_id,
            message,
            score,
            dedupe_hash
        )

        if pool_eligible:
            attach_to_pool(
                db,
                group_id,
                post_id,
                score
            )

        collected += 1

    if backfill_complete:
        complete = True
        state_oldest_id = (
            int(state.get("oldest_message_id") or 0)
            or oldest_id
        )
    else:
        complete = (
            raw_scanned < BACKFILL_BATCH
        )
        state_oldest_id = oldest_id

    update_state_success(
        db,
        group_id,
        source_id,
        state_oldest_id,
        newest_id,
        raw_scanned,
        collected,
        complete
    )

    update_source_success(
        db,
        source_id
    )

    print(
        f"  Scanned:       "
        f"{raw_scanned}"
    )

    print(
        f"  New Posts:     "
        f"{collected}"
    )

    print(
        f"  Duplicates:    "
        f"{duplicate}"
    )

    print(
        f"  Median Views:  "
        f"{median_views:.0f}"
    )

    print(
        f"  Oldest ID:     "
        f"{oldest_id}"
    )

    print(
        "  Backfill:      "
        + (
            "COMPLETE"
            if complete
            else "MORE"
        )
    )

    return {
        "scanned": raw_scanned,
        "collected": collected,
        "duplicate": duplicate,
    }


async def main():

    group_key = (
        str(sys.argv[1]).strip().lower()
        if len(sys.argv) > 1
        else "deep"
    )

    db = get_database()
    client = None

    try:

        group = get_group(
            db,
            group_key
        )

        if not group:
            fail(
                f"Content group "
                f"'{group_key}' not found."
            )

        group_id = int(
            group["id"]
        )

        sources = (
            get_approved_sources(
                db,
                group_id
            )
        )

        publisher = get_publisher(
            db
        )

        if not publisher:
            fail(
                "Active Telegram "
                "publisher not found."
            )

        print("")
        print(
            "CONTENT COLLECTOR V1.0"
        )
        print(
            "=============================================="
        )
        print(
            f"Group:          "
            f"{group_key}"
        )
        print(
            f"Approved Sources:"
            f" {len(sources)}"
        )
        print(
            f"Batch/Source:   "
            f"{BACKFILL_BATCH}"
        )
        print(
            "Mode:           BACKFILL"
        )
        print(
            "=============================================="
        )

        if not sources:
            print(
                "No approved sources."
            )
            return

        client = build_client(
            publisher
        )

        await client.connect()

        if not await client.is_user_authorized():
            fail(
                "Telegram session "
                "is not authorized."
            )

        total_scanned = 0
        total_collected = 0
        total_duplicate = 0
        errors = 0

        for source in sources:

            try:

                result = await collect_source(
                    client,
                    db,
                    group_id,
                    source
                )

                total_scanned += (
                    result["scanned"]
                )

                total_collected += (
                    result["collected"]
                )

                total_duplicate += (
                    result["duplicate"]
                )

            except FloodWaitError as exc:

                seconds = int(
                    exc.seconds
                )

                update_state_error(
                    db,
                    group_id,
                    int(source["id"]),
                    (
                        f"FloodWait: "
                        f"{seconds} seconds"
                    )
                )

                print("")
                print(
                    f"⚠️ FloodWait: "
                    f"{seconds}s"
                )

                print(
                    "Collector stopped safely."
                )

                break

            except Exception as exc:

                errors += 1

                error_text = str(
                    exc
                )

                update_state_error(
                    db,
                    group_id,
                    int(source["id"]),
                    error_text
                )

                update_source_error(
                    db,
                    int(source["id"]),
                    error_text
                )

                print(
                    "  ❌ ERROR: "
                    + error_text
                )

        print("")
        print(
            "=============================================="
        )
        print(
            "COLLECTOR COMPLETE ✅"
        )
        print(
            f"Scanned:        "
            f"{total_scanned}"
        )
        print(
            f"New Posts:      "
            f"{total_collected}"
        )
        print(
            f"Duplicates:     "
            f"{total_duplicate}"
        )
        print(
            f"Errors:         "
            f"{errors}"
        )
        print(
            "=============================================="
        )

    finally:

        if client is not None:
            try:
                if client.is_connected():
                    await client.disconnect()
            except Exception:
                pass

        db.close()


if __name__ == "__main__":
    asyncio.run(main())