Spaces:
Sleeping
Sleeping
| """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 не затронут.") | |
| def root(): | |
| return {"service": "ReportBridge", "production_miniapp": "not used", "health": "/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"))}} | |
| 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"} | |
| 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")} | |