app.py

import sqlite3
import asyncio
import aiohttp
import json
import os
import time
import threading
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  # пушить каждые 10k найденных

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(),
}

# ─── 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()
    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)
    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}')

# ─── ГЛАВНЫЙ ЦИКЛ ─────────────────────────────────────────────
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
import sqlite3
import asyncio
import aiohttp
import json
import os
import time
import threading
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  # пушить каждые 10k найденных

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(),
}

# ─── 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'))
        )
    ''')
    c.execute('''
        CREATE VIRTUAL TABLE IF NOT EXISTS posts_fts USING fts5(
            text,
            content='posts',
            content_rowid='post_id',
            tokenize='trigram'
        )
    ''')
    c.execute('''
        CREATE TRIGGER IF NOT EXISTS posts_ai AFTER INSERT ON posts BEGIN
            INSERT INTO posts_fts(rowid, text) VALUES (new.post_id, new.text);
        END
    ''')
    conn.commit()
    conn.close()

# ─── ОДНОРАЗОВАЯ МИГРАЦИЯ В FTS5 ───────────────────────────────
def migrate_to_fts():
    conn = sqlite3.connect(DB_PATH, timeout=120)
    conn.execute('PRAGMA journal_mode=WAL')
    c = conn.cursor()
    c.execute('SELECT COUNT(*) FROM posts_fts')
    fts_count = c.fetchone()[0]
    c.execute('SELECT COUNT(*) FROM posts')
    posts_count = c.fetchone()[0]

    if fts_count < posts_count:
        print(f'🔨 Строю FTS индекс: {fts_count:,} / {posts_count:,}, заполняю...')
        c.execute('''
            INSERT INTO posts_fts(rowid, text)
            SELECT post_id, text FROM posts
            WHERE post_id NOT IN (SELECT rowid FROM posts_fts)
        ''')
        conn.commit()
        print('✅ FTS индекс построен')
    else:
        print('✅ FTS индекс уже актуален')
    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)
    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}')

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

    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()

@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):
    conn = sqlite3.connect(DB_PATH, timeout=30)
    conn.execute('PRAGMA journal_mode=WAL')
    c = conn.cursor()
    safe_q = q.replace('"', '""')
    c.execute('SELECT post_id, text FROM posts_fts WHERE posts_fts MATCH ? LIMIT ?', (f'"{safe_q}"', limit))
    rows = c.fetchall()
    conn.close()
    return {
        'query': q,
        'count': len(rows),
        'results': [{'post_id': r[0], 'text': r[1]} for r in rows]
    }

# ─── СТАРТ ────────────────────────────────────────────────────
if __name__ == '__main__':
    t = threading.Thread(target=start_parser, daemon=True)
    t.start()
    uvicorn.run(app, host='0.0.0.0', port=7860)