"""Isolated report service. Never calls or modifies the production MiniApp Space.""" import hmac import logging import os import threading import time import uuid from concurrent.futures import ThreadPoolExecutor, as_completed from typing import List from fastapi import Depends, FastAPI, Header, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field import legacy_parser as engine from reports import normalize, render from tgstat_api import archive_link, archived_post log = logging.getLogger(__name__) API_KEY = os.getenv("BRIDGE_API_KEY", "") app = FastAPI(title="ReportBridge", version="0.1.2") app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=False, allow_methods=["GET", "POST", "DELETE"], allow_headers=["Content-Type", "X-Bridge-Key"]) jobs, lock = {}, threading.RLock() queue = ThreadPoolExecutor(max_workers=1, thread_name_prefix="reports") def authorize(x_bridge_key: str = Header(default="")): if not API_KEY: raise HTTPException(503, "В бэкенде расширения не настроен BRIDGE_API_KEY") if not hmac.compare_digest(x_bridge_key.encode(), API_KEY.encode()): raise HTTPException(401, "Неверный ключ ReportBridge") class JobRequest(BaseModel): request_id: str = Field(min_length=16, max_length=64, pattern=r"^[a-zA-Z0-9-]+$") links: List[str] = Field(min_length=1, max_length=500) def fetch_one(record): result = dict(record) try: if record["platform"] == "telegram": title, views = engine.process_telegram_link(record["a"], record["b"], record["url"]) else: title, views = engine.process_max_link(record["a"], record["b"], record["url"]) if "ошибка" in title.lower() or views is None: raise ValueError("Источник не вернул просмотры. Попробуйте повторить позже.") # The legacy HTML parser returns 0 when it cannot find a counter. Avoid a false zero in new reports. if record["platform"] == "telegram" and views == 0: raise ValueError("Не найден счётчик просмотров Telegram; это не подтверждённый ноль") result.update(title=title, views=views) except Exception: result.update(title=record["group"], views=None, error="Не удалось получить просмотры публичного поста") if record["platform"] == "telegram": result["display_url"] = archive_link(record["a"], record["b"]) try: fallback = archived_post(record["a"], record["b"], record["group"]) if fallback: result.update(fallback, error=None) except Exception: # Never log API request URLs: TGStat credentials are query parameters. pass return result def run_job(job_id, records): result_map = {} with lock: jobs[job_id]["status"] = "running" def store_result(record): with lock: result_map[record["url"]] = record jobs[job_id]["completed"] = len(result_map) try: with ThreadPoolExecutor(max_workers=3) as pool: futures = {pool.submit(fetch_one, r): r for r in records if r["platform"] != "vk"} vk = [r for r in records if r["platform"] == "vk"] for start in range(0, len(vk), 100): batch = vk[start:start + 100] if not engine.VK_ACCESS_TOKEN: data = {} else: data = engine.process_vk_batch([(r["url"], r["a"], r["b"]) for r in batch]) for record in batch: title, views = data.get(record["url"], ("Ошибка: VK_ACCESS_TOKEN не настроен", None)) error = ("Не настроен VK_ACCESS_TOKEN в Secrets PR_Tool" if not engine.VK_ACCESS_TOKEN else "Не удалось получить просмотры VK") if "ошибка" in title.lower() else None if not error and views is None: error = "VK API не передал счётчик просмотров этого поста; это не подтверждённый ноль" store_result(dict(record, title=title if not error else record["group"], views=None if error else views, error=error)) if start + 100 < len(vk): time.sleep(0.4) for future in as_completed(futures): store_result(future.result()) output = render([result_map[r["url"]] for r in records]) with lock: jobs[job_id].update(status="complete", result=output) except Exception: log.exception("ReportBridge job failed") with lock: jobs[job_id].update(status="failed", error="Ошибка нового бэкенда. Рабочий MiniApp не затронут.") @app.get("/") def root(): return {"service": "ReportBridge", "production_miniapp": "not used", "health": "/health"} @app.get("/health") def health(): return {"status": "ok", "service": "ReportBridge", "configured": { "bridge": bool(API_KEY), "vk": bool(engine.VK_ACCESS_TOKEN), "max": bool(os.getenv("MAX_ACCESS_TOKEN")), "tgstat": bool(os.getenv("TGSTAT_API_TOKEN"))}} @app.post("/jobs", dependencies=[Depends(authorize)]) def create_job(payload: JobRequest): records, seen = [], set() for raw in payload.links: if len(raw) > 2048: raise HTTPException(422, "Слишком длинная ссылка") try: record = normalize(raw) except ValueError as exc: raise HTTPException(422, str(exc)) from None if record["url"] not in seen: seen.add(record["url"]) records.append(record) canonical = [r["url"] for r in records] with lock: now = time.time() expired = [k for k, j in jobs.items() if j["status"] not in ("queued", "running") and now - j["created"] > 3600] for key in expired: del jobs[key] for job in jobs.values(): if job["request_id"] == payload.request_id: if job["links"] != canonical: raise HTTPException(409, "Этот идентификатор уже используется для другого списка") return {"id": job["id"], "status": job["status"]} if sum(j["status"] in ("queued", "running") for j in jobs.values()) >= 5 or len(jobs) >= 100: raise HTTPException(429, "Очередь нового Space заполнена. Попробуйте позже.") job_id = str(uuid.uuid4()) jobs[job_id] = dict(id=job_id, request_id=payload.request_id, links=canonical, status="queued", completed=0, total=len(records), created=now) queue.submit(run_job, job_id, records) return {"id": job_id, "status": "queued"} @app.get("/jobs/{job_id}", dependencies=[Depends(authorize)]) def get_job(job_id: str): with lock: if job_id not in jobs: raise HTTPException(404, "Задание не найдено. Возможно, новый Space перезапустился; запустите обработку заново.") return {k: v for k, v in jobs[job_id].items() if k not in ("links", "request_id")}