Blame for service.py

"""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")}