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)