app.py

import sqlite3
import asyncio
import aiohttp
import json
import os
import time
import threading
import urllib.request
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeout
from bs4 import BeautifulSoup
from fastapi import FastAPI
from huggingface_hub import HfApi, hf_hub_download
import uvicorn

# ─── НАСТРОЙКИ ────────────────────────────────────────────────
DB_PATH         = '/app/linkm_posts.db'
PROGRESS_FILE   = '/app/linkm_progress.json'
BASE_URL        = 'https://linkm.me/posts/{}'
START_ID        = 213470273
END_ID          = START_ID
WORKERS         = 150
BATCH_SIZE      = 500
REQUEST_TIMEOUT = 10
HF_TOKEN        = os.environ.get('HF_TOKEN')
HF_DATASET      = os.environ.get('HF_DATASET')
PUSH_EVERY      = 10000

HEADERS = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
    'Accept-Language': 'en-US,en;q=0.5',
}

# ─── СОСТОЯНИЕ ────────────────────────────────────────────────
state = {
    'status': 'starting',
    'current_id': START_ID,
    'found': 0,
    'processed': 0,
    'speed': 0,
    'last_push': 0,
    'started_at': time.time(),
}

# ─── ПУЛ ПОТОКОВ ДЛЯ ПОИСКА ──────────────────────────────────
executor = ThreadPoolExecutor(max_workers=3)

# ─── PULL С HF ПРИ СТАРТЕ ─────────────────────────────────────
def pull_from_hf():
    if not HF_TOKEN or not HF_DATASET:
        return
    try:
        hf_hub_download(repo_id=HF_DATASET, filename='linkm_posts.db',
                        repo_type='dataset', local_dir='/app', token=HF_TOKEN)
        print('✅ База скачана с HF')
    except Exception as e:
        print(f'⚠️ Не удалось скачать базу: {e}')
    try:
        hf_hub_download(repo_id=HF_DATASET, filename='linkm_progress.json',
                        repo_type='dataset', local_dir='/app', token=HF_TOKEN)
        print('✅ Прогресс скачан с HF')
    except Exception as e:
        print(f'⚠️ Прогресс не найден на HF: {e}')

# ─── БАЗА ДАННЫХ ──────────────────────────────────────────────
def init_db():
    conn = sqlite3.connect(DB_PATH)
    conn.execute('PRAGMA journal_mode=WAL')
    c = conn.cursor()
    c.execute('''
        CREATE TABLE IF NOT EXISTS posts (
            post_id   INTEGER PRIMARY KEY,
            text      TEXT NOT NULL,
            parsed_at TEXT DEFAULT (datetime('now'))
        )
    ''')
    conn.commit()
    # Прогрев
    c.execute('SELECT COUNT(*) FROM posts')
    count = c.fetchone()[0]
    print(f'✅ База прогрета: {count:,} постов')
    conn.close()

# ─── ПРОГРЕСС ─────────────────────────────────────────────────
def load_progress():
    if os.path.exists(PROGRESS_FILE):
        try:
            with open(PROGRESS_FILE, 'r') as f:
                data = json.load(f)
            print(f'▶️  Продолжаем с ID: {data["next_id"]} (найдено: {data["found"]:,})')
            return data
        except Exception:
            pass
    return {'next_id': START_ID, 'processed': 0, 'found': 0}

def save_progress(data):
    with open(PROGRESS_FILE, 'w') as f:
        json.dump(data, f)

# ─── ПАРСИНГ ──────────────────────────────────────────────────
def extract_text(html: str):
    soup = BeautifulSoup(html, 'html.parser')

    main = soup.find('main', class_='deeplink_main')
    if main:
        p = main.find('p')
        if p:
            parts = [node.strip() for node in p.descendants if isinstance(node, str) and node.strip()]
            result = ' '.join(parts).strip()
            if ' on Link. ' in result:
                result = result.split(' on Link. ', 1)[1].strip()
            elif result.endswith(' on Link.'):
                return None
            if result:
                return result

    og = soup.find('meta', property='og:description')
    if og:
        content = og.get('content', '').strip()
        if ' on Link. ' in content:
            content = content.split(' on Link. ', 1)[1].strip()
        elif content.endswith(' on Link.'):
            return None
        if content:
            return content

    return None

async def fetch_post(session, post_id):
    try:
        async with session.get(
            BASE_URL.format(post_id),
            timeout=aiohttp.ClientTimeout(total=REQUEST_TIMEOUT)
        ) as resp:
            if resp.status != 200:
                return post_id, None
            html = await resp.text(encoding='utf-8', errors='replace')
            return post_id, extract_text(html)
    except Exception:
        return post_id, None

def save_batch(rows):
    if not rows:
        return
    conn = sqlite3.connect(DB_PATH, timeout=60)
    conn.execute('PRAGMA journal_mode=WAL')
    conn.execute('PRAGMA busy_timeout=60000')
    conn.executemany('INSERT OR IGNORE INTO posts (post_id, text) VALUES (?, ?)', rows)
    conn.commit()
    conn.close()

# ─── PUSH В HUGGINGFACE ───────────────────────────────────────
def push_to_hf(progress):
    if not HF_TOKEN or not HF_DATASET:
        return
    try:
        api = HfApi(token=HF_TOKEN)
        api.upload_file(path_or_fileobj=DB_PATH, path_in_repo='linkm_posts.db',
                        repo_id=HF_DATASET, repo_type='dataset')
        api.upload_file(path_or_fileobj=PROGRESS_FILE, path_in_repo='linkm_progress.json',
                        repo_id=HF_DATASET, repo_type='dataset')
        print(f'✅ Запушено в HF (найдено: {progress["found"]:,})')
        state['last_push'] = progress['found']
    except Exception as e:
        print(f'❌ Ошибка пуша: {e}')

# ─── KEEPALIVE ────────────────────────────────────────────────
def keepalive():
    time.sleep(30)
    while True:
        try:
            urllib.request.urlopen('http://localhost:7860/status', timeout=5)
            print('💓 Keepalive ping')
        except Exception:
            pass
        time.sleep(240)

# ─── ГЛАВНЫЙ ЦИКЛ ─────────────────────────────────────────────
async def run_parser():
    pull_from_hf()
    init_db()

    progress = load_progress()
    current_id = progress['next_id']
    start_time = time.time()

    state['status'] = 'running'
    state['current_id'] = current_id
    state['found'] = progress['found']
    state['processed'] = progress['processed']
    state['last_push'] = progress['found']

    connector = aiohttp.TCPConnector(limit=WORKERS, ssl=False)

    async with aiohttp.ClientSession(connector=connector, headers=HEADERS) as session:
        while current_id > END_ID:
            batch_start = max(current_id - BATCH_SIZE, END_ID)
            batch_ids = list(range(current_id, batch_start, -1))
            results = await asyncio.gather(*[fetch_post(session, pid) for pid in batch_ids])

            found_rows = [(pid, txt) for pid, txt in results if txt]
            save_batch(found_rows)

            progress['processed'] += len(batch_ids)
            progress['found'] += len(found_rows)
            progress['next_id'] = batch_start
            save_progress(progress)

            elapsed = time.time() - start_time
            speed = progress['processed'] / max(elapsed, 1)

            state['current_id'] = current_id
            state['found'] = progress['found']
            state['processed'] = progress['processed']
            state['speed'] = round(speed)

            print(f'ID: {current_id:,} | Найдено: {progress["found"]:,} | {speed:.0f} ID/сек')

            if progress['found'] - state['last_push'] >= PUSH_EVERY:
                push_to_hf(progress)

            current_id = batch_start

    state['status'] = 'done'
    push_to_hf(progress)

def start_parser():
    asyncio.run(run_parser())

# ─── FASTAPI ──────────────────────────────────────────────────
app = FastAPI()

def get_db():
    conn = sqlite3.connect(DB_PATH, timeout=30, check_same_thread=False)
    conn.execute('PRAGMA journal_mode=WAL')
    conn.execute('PRAGMA busy_timeout=30000')
    conn.execute('PRAGMA cache_size=-65536')  # 64MB кэш
    conn.execute('PRAGMA temp_store=MEMORY')
    return conn

def _do_search(q, limit):
    conn = get_db()
    c = conn.cursor()
    c.execute(
        'SELECT post_id, text FROM posts WHERE text LIKE ? LIMIT ?',
        (f'%{q}%', limit)
    )
    rows = c.fetchall()
    conn.close()
    return rows

@app.get('/status')
def get_status():
    elapsed = time.time() - state['started_at']
    remaining_ids = state['current_id'] - END_ID
    remaining_sec = remaining_ids / max(state['speed'], 1)
    return {
        'status': state['status'],
        'current_id': state['current_id'],
        'end_id': END_ID,
        'percent': round((START_ID - state['current_id']) / START_ID * 100, 2),
        'found': state['found'] + 13000000,
        'processed': state['processed'],
        'speed_per_sec': state['speed'],
        'elapsed_min': round(elapsed / 60, 1),
        'remaining_min': round(remaining_sec / 60, 1),
        'last_push_at': state['last_push'],
    }

@app.get('/search')
def search(q: str, limit: int = 50):
    if len(q.strip()) < 2:
        return {'error': 'Минимум 2 символа', 'query': q, 'count': 0, 'results': []}
    try:
        future = executor.submit(_do_search, q, limit)
        rows = future.result(timeout=25)
        return {
            'query': q,
            'count': len(rows),
            'results': [{'post_id': r[0], 'text': r[1]} for r in rows]
        }
    except FuturesTimeout:
        future.cancel()
        return {'query': q, 'count': 0, 'results': []}
    except Exception as e:
        return {'error': str(e), 'query': q, 'count': 0, 'results': []}

@app.get('/debug')
def debug(q: str = 'salam'):
    try:
        conn = get_db()
        c = conn.cursor()
        c.execute('SELECT COUNT(*) FROM posts')
        posts_count = c.fetchone()[0]
        c.execute('SELECT COUNT(*) FROM posts WHERE text LIKE ?', (f'%{q}%',))
        like_count = c.fetchone()[0]
        c.execute('SELECT post_id, text FROM posts WHERE text LIKE ? LIMIT 3', (f'%{q}%',))
        samples = c.fetchall()
        conn.close()
        return {
            'posts_table_total': posts_count,
            'like_search_count': like_count,
            'samples': [{'post_id': r[0], 'text': r[1][:100]} for r in samples],
        }
    except Exception as e:
        return {'error': str(e)}

# ─── СТАРТ ────────────────────────────────────────────────────
if __name__ == '__main__':
    t = threading.Thread(target=start_parser, daemon=True)
    t.start()

    k = threading.Thread(target=keepalive, daemon=True)
    k.start()

    uvicorn.run(app, host='0.0.0.0', port=7860)