src/streamlit_app.py

import json
import os
import re
import time
from collections import defaultdict
from datetime import datetime, timezone
from typing import Any, DefaultDict, Dict, Iterable, List, Optional, Sequence, Tuple
from urllib.parse import quote

import pandas as pd
import requests
import streamlit as st


HF_TOKEN = os.environ.get("HF_TOKEN") or os.environ.get("HF_API_TOKEN")
REVIEWS_LOG_PREFIX = (os.environ.get("REVIEWS_LOG_PREFIX") or "reviews_log").strip() or "reviews_log"
HF_TIMEOUT = int(os.environ.get("HF_HTTP_TIMEOUT", "60"))
HF_RETRIES = int(os.environ.get("HF_HTTP_RETRIES", "3"))
HF_BASE = "https://huggingface.co"
SCORE_VALUES = (-2, -1, 0, 1, 2)

st.set_page_config(page_title="Экспорт результатов скоринга", layout="wide")


def hf_headers(token: Optional[str]) -> Dict[str, str]:
    headers = {"User-Agent": "hf-space-scoring-export/4.0"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    return headers


def parse_link_header(header: str) -> Dict[str, str]:
    out: Dict[str, str] = {}
    if not header:
        return out
    for url, rel in re.findall(r'<([^>]+)>;\s*rel="([^"]+)"', header):
        out[rel] = url
    return out


def parse_repo_list(raw: str) -> List[str]:
    if not raw:
        return []
    return [x.strip() for x in re.split(r"[\s,;]+", raw.strip()) if x.strip()]


def unique_preserve_order(items: Iterable[str]) -> List[str]:
    seen = set()
    out: List[str] = []
    for item in items:
        if item not in seen:
            seen.add(item)
            out.append(item)
    return out


def collect_reviews_repos() -> List[str]:
    repo_ids: List[str] = []
    repo_ids.extend(parse_repo_list(os.environ.get("REVIEWS_REPOS", "")))

    legacy_single = (os.environ.get("REVIEWS_REPO") or "").strip()
    if legacy_single:
        repo_ids.append(legacy_single)

    numbered: List[Tuple[int, str]] = []
    for key, value in os.environ.items():
        m = re.fullmatch(r"REVIEWS_REPO(\d+)", key)
        if not m:
            continue
        repo_id = (value or "").strip()
        if repo_id:
            numbered.append((int(m.group(1)), repo_id))

    for _, repo_id in sorted(numbered, key=lambda x: x[0]):
        repo_ids.append(repo_id)

    return unique_preserve_order(repo_ids)


def normalize_dir_id(dir_id: str, pad2: bool = False) -> str:
    s = str(dir_id or "").strip().upper()
    if not s.startswith("DIR"):
        return str(dir_id or "").strip()
    tail = s[3:]
    try:
        n = int(tail)
    except Exception:
        return str(dir_id or "").strip()
    return f"DIR{n:02d}" if pad2 else f"DIR{n}"


def dir_no(dir_id: str) -> int:
    s = normalize_dir_id(dir_id, pad2=False)
    try:
        return int(s[3:])
    except Exception:
        return 0


def safe_int(value: Any) -> Optional[int]:
    try:
        if value is None:
            return None
        return int(float(value))
    except Exception:
        return None


def iso_now_utc() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def request_with_retries(url: str, headers: Dict[str, str]) -> requests.Response:
    last_exc: Optional[Exception] = None
    for attempt in range(1, HF_RETRIES + 1):
        try:
            resp = requests.get(url, headers=headers, timeout=HF_TIMEOUT)
            resp.raise_for_status()
            return resp
        except Exception as e:
            last_exc = e
            if attempt < HF_RETRIES:
                time.sleep(min(2 ** (attempt - 1), 4))
    assert last_exc is not None
    raise last_exc


@st.cache_data(show_spinner=False)
def list_repo_files(repo_id: str, token: Optional[str]) -> List[str]:
    headers = hf_headers(token)
    url = f"{HF_BASE}/api/datasets/{repo_id}/tree/main?recursive=true&expand=false"
    files: List[str] = []

    while url:
        resp = request_with_retries(url, headers)
        data = resp.json()
        if not isinstance(data, list):
            raise RuntimeError("Unexpected tree response")

        for item in data:
            if isinstance(item, dict):
                path = item.get("path")
                if isinstance(path, str) and path:
                    files.append(path)

        url = parse_link_header(resp.headers.get("Link", "")).get("next")

    return sorted(files)


@st.cache_data(show_spinner=False)
def download_text_file(repo_id: str, relpath: str, token: Optional[str]) -> str:
    headers = hf_headers(token)
    quoted_path = quote(relpath, safe="/")
    url = f"{HF_BASE}/datasets/{repo_id}/resolve/main/{quoted_path}?download=true"
    resp = request_with_retries(url, headers)
    resp.encoding = resp.encoding or "utf-8"
    return resp.text


def dir_from_path(relpath: str, prefix: str) -> Optional[str]:
    parts = [p for p in relpath.split("/") if p]
    if len(parts) < 4 or parts[0] != prefix:
        return None
    return normalize_dir_id(parts[2], pad2=False)


@st.cache_data(show_spinner=False)
def discover_dirs(repo_id: str, prefix: str, token: Optional[str]) -> List[str]:
    files = list_repo_files(repo_id, token)
    dirs = set()
    for relpath in files:
        if relpath.startswith(prefix + "/") and relpath.endswith(".jsonl"):
            d = dir_from_path(relpath, prefix)
            if d:
                dirs.add(d)
    return sorted(dirs, key=dir_no)


def filter_review_files(files: Sequence[str], selected_dirs: Sequence[str], prefix: str) -> List[str]:
    paths = [p for p in files if p.startswith(prefix + "/") and p.endswith(".jsonl")]
    if not selected_dirs:
        return sorted(paths)

    needles = {f"/{normalize_dir_id(d, pad2=False)}/" for d in selected_dirs} | {
        f"/{normalize_dir_id(d, pad2=True)}/" for d in selected_dirs
    }
    return sorted([p for p in paths if any(n in p for n in needles)])


@st.cache_data(show_spinner=False)
def read_reviews_from_files(repo_id: str, review_files: Tuple[str, ...], token: Optional[str]) -> List[Dict[str, Any]]:
    out: List[Dict[str, Any]] = []
    for relpath in review_files:
        try:
            text = download_text_file(repo_id, relpath, token)
            for line in text.splitlines():
                line = line.strip()
                if not line:
                    continue
                obj = json.loads(line)
                if isinstance(obj, dict):
                    obj["__file__"] = relpath
                    out.append(obj)
        except Exception:
            continue
    return out


def build_dir_preview_df(dir_summary: List[Dict[str, Any]]) -> pd.DataFrame:
    rows: List[Dict[str, Any]] = []
    for row in dir_summary:
        rows.append(
            {
                "DIR": normalize_dir_id(row["dir_id"], pad2=True),
                "Оценено всего": row["evaluated_publications"],
                "-2": row["score_-2_count"],
                "-1": row["score_-1_count"],
                "0": row["score_0_count"],
                "1": row["score_1_count"],
                "2": row["score_2_count"],
                "1+2": row["score_1_2_count"],
            }
        )
    return pd.DataFrame(rows)


def build_results_df(items: List[Dict[str, Any]]) -> pd.DataFrame:
    rows: List[Dict[str, Any]] = []
    for item in items:
        result = item.get("result") or {}
        latest_scores = sorted(
            [str(x.get("score")) for x in (item.get("reviewers_latest") or []) if x.get("score") is not None]
        )
        rows.append(
            {
                "DIR": normalize_dir_id(str(item.get("dir_id") or ""), pad2=True),
                "openalex_work_id": item.get("openalex_work_id"),
                "итоговая_оценка": result.get("score"),
                "кто_поставил_итог": result.get("reviewer"),
                "время_итога_utc": result.get("ts_utc"),
                "оценок_в_истории": len(item.get("history") or []),
                "оценки_пользователей": ", ".join(latest_scores),
            }
        )
    return pd.DataFrame(rows)


def build_overall_totals(dir_summary: List[Dict[str, Any]]) -> Dict[str, Any]:
    total = sum(int(row.get("evaluated_publications", 0)) for row in dir_summary)
    counts = {score: sum(int(row.get(f"score_{score}_count", 0)) for row in dir_summary) for score in SCORE_VALUES}
    positive = counts[1] + counts[2]
    return {
        "evaluated_publications": total,
        "counts": counts,
        "positive_count": positive,
    }


def build_payload(selected_dirs: List[str], repo_ids: Sequence[str]) -> Dict[str, Any]:
    grouped: Dict[Tuple[str, str], Dict[str, Any]] = {}
    review_files_scanned = 0
    review_events_loaded = 0
    review_events_attached = 0
    repos_scanned_ok = 0
    repos_failed = 0
    per_repo_status: List[Dict[str, Any]] = []

    for idx, repo_id in enumerate(repo_ids, start=1):
        alias = f"База {idx}"
        try:
            repo_files = list_repo_files(repo_id, HF_TOKEN)
            review_files = filter_review_files(repo_files, selected_dirs, REVIEWS_LOG_PREFIX)
            reviews = read_reviews_from_files(repo_id, tuple(review_files), HF_TOKEN)
            repos_scanned_ok += 1
            per_repo_status.append(
                {
                    "База": alias,
                    "Статус": "OK",
                    "Файлов review": len(review_files),
                    "Событий review": len(reviews),
                    "DIR найдено": len({dir_from_path(p, REVIEWS_LOG_PREFIX) for p in review_files if dir_from_path(p, REVIEWS_LOG_PREFIX)}),
                }
            )
        except Exception as e:
            repos_failed += 1
            per_repo_status.append(
                {
                    "База": alias,
                    "Статус": "Ошибка",
                    "Файлов review": 0,
                    "Событий review": 0,
                    "DIR найдено": 0,
                    "Сообщение": str(e),
                }
            )
            continue

        review_files_scanned += len(review_files)

        for review in reviews:
            review_events_loaded += 1

            raw_dir = review.get("dir_id_canonical") or review.get("dir_id") or dir_from_path(str(review.get("__file__") or ""), REVIEWS_LOG_PREFIX)
            if not isinstance(raw_dir, str) or not raw_dir.strip():
                continue
            dir_id = normalize_dir_id(raw_dir, pad2=False)

            work_id = review.get("work_id") or review.get("openalex_work_id")
            if not isinstance(work_id, str) or not work_id.strip():
                continue
            work_id = work_id.strip()

            reviewer = review.get("reviewer")
            if not isinstance(reviewer, str) or not reviewer.strip():
                continue
            reviewer = reviewer.strip()

            score = safe_int(review.get("score"))
            if score not in SCORE_VALUES:
                continue

            ts_utc = str(review.get("ts_utc") or review.get("ts") or "")
            key = (dir_id, work_id)
            if key not in grouped:
                grouped[key] = {
                    "dir_id": dir_id,
                    "openalex_work_id": work_id,
                    "history": [],
                }

            grouped[key]["history"].append(
                {
                    "reviewer": reviewer,
                    "score": score,
                    "ts_utc": ts_utc,
                }
            )
            review_events_attached += 1

    items: List[Dict[str, Any]] = []
    score_summary_by_dir: DefaultDict[str, Dict[int, int]] = defaultdict(lambda: {s: 0 for s in SCORE_VALUES})

    for _, item in sorted(grouped.items(), key=lambda kv: (kv[0][0], kv[0][1])):
        history = item["history"]
        history.sort(key=lambda x: ((x.get("ts_utc") or ""), (x.get("reviewer") or ""), int(x.get("score") or 0)))

        latest_by_reviewer_map: Dict[str, Dict[str, Any]] = {}
        for event in history:
            latest_by_reviewer_map[event["reviewer"]] = {
                "reviewer": event["reviewer"],
                "score": int(event["score"]),
                "ts_utc": event.get("ts_utc") or "",
            }

        reviewers_latest = sorted(
            latest_by_reviewer_map.values(),
            key=lambda x: ((x.get("reviewer") or ""), (x.get("ts_utc") or "")),
        )
        result = history[-1] if history else None

        item_out = {
            "dir_id": item["dir_id"],
            "openalex_work_id": item["openalex_work_id"],
            "result": {
                "reviewer": result["reviewer"],
                "score": int(result["score"]),
                "ts_utc": result.get("ts_utc") or "",
            } if result else None,
            "reviewers_latest": reviewers_latest,
            "history": history,
        }
        items.append(item_out)

        if result and int(result["score"]) in SCORE_VALUES:
            score_summary_by_dir[item["dir_id"]][int(result["score"])] += 1

    dir_summary: List[Dict[str, Any]] = []
    for dir_id in sorted(score_summary_by_dir.keys(), key=dir_no):
        counts = score_summary_by_dir[dir_id]
        dir_summary.append(
            {
                "dir_id": dir_id,
                "evaluated_publications": int(sum(counts.values())),
                "score_-2_count": int(counts.get(-2, 0)),
                "score_-1_count": int(counts.get(-1, 0)),
                "score_0_count": int(counts.get(0, 0)),
                "score_1_count": int(counts.get(1, 0)),
                "score_2_count": int(counts.get(2, 0)),
                "score_1_2_count": int(counts.get(1, 0) + counts.get(2, 0)),
            }
        )

    meta = {
        "generated_at_utc": iso_now_utc(),
        "repositories_count": len(repo_ids),
        "repositories_scanned_ok_count": repos_scanned_ok,
        "repositories_failed_count": repos_failed,
        "reviews_log_prefix": REVIEWS_LOG_PREFIX,
        "selected_dirs": sorted(selected_dirs, key=dir_no),
        "dirs_total": len(dir_summary),
        "publications_total": len(items),
        "review_files_scanned": review_files_scanned,
        "review_events_loaded": review_events_loaded,
        "review_events_attached": review_events_attached,
        "result_definition": "latest review event per publication across all connected review repositories",
        "history_schema": ["reviewer", "score", "ts_utc"],
        "source_basis": "reviews_log only",
    }

    return {
        "meta": meta,
        "dir_summary": dir_summary,
        "items": items,
        "repo_status": per_repo_status,
    }


def discover_dirs_across_repos(repo_ids: Sequence[str], prefix: str, token: Optional[str]) -> Tuple[List[str], List[Dict[str, Any]]]:
    dirs = set()
    statuses: List[Dict[str, Any]] = []
    for idx, repo_id in enumerate(repo_ids, start=1):
        alias = f"База {idx}"
        try:
            repo_dirs = discover_dirs(repo_id, prefix, token)
            dirs.update(repo_dirs)
            statuses.append({"База": alias, "Статус": "OK", "DIR найдено": len(repo_dirs)})
        except Exception as e:
            statuses.append({"База": alias, "Статус": "Ошибка", "DIR найдено": 0, "Сообщение": str(e)})
    return sorted(dirs, key=dir_no), statuses


REVIEWS_REPOS = collect_reviews_repos()
if not REVIEWS_REPOS:
    st.error("Не задан ни один reviews repo. Используйте Secret/Variable REVIEWS_REPOS или REVIEWS_REPO1, REVIEWS_REPO2, ...")
    st.stop()

st.title("Экспорт результатов скоринга")
st.caption("Источник данных: reviews_log из одного или нескольких закрытых reviews repos.")

if not HF_TOKEN:
    st.error("Не задан Secret HF_TOKEN или HF_API_TOKEN.")
    st.stop()

dir_ids, discovery_status = discover_dirs_across_repos(REVIEWS_REPOS, REVIEWS_LOG_PREFIX, HF_TOKEN)
if not dir_ids:
    st.error("Не удалось найти ни одного DIR в подключённых базах review-логов.")
    status_df = pd.DataFrame(discovery_status)
    if not status_df.empty:
        st.dataframe(status_df, use_container_width=True, hide_index=True)
    st.stop()

ok_count = sum(1 for x in discovery_status if x.get("Статус") == "OK")
fail_count = sum(1 for x in discovery_status if x.get("Статус") != "OK")

s1, s2, s3 = st.columns(3)
s1.metric("Подключено баз", len(REVIEWS_REPOS))
s2.metric("Баз прочитано", ok_count)
s3.metric("Ошибок чтения баз", fail_count)

st.subheader("Статус чтения баз")
st.dataframe(pd.DataFrame(discovery_status), use_container_width=True, hide_index=True)

selected_dirs = st.multiselect(
    "DIR для экспорта",
    options=dir_ids,
    default=dir_ids,
    format_func=lambda d: normalize_dir_id(d, pad2=True),
)


if st.button("Собрать предпросмотр и JSON", type="primary", use_container_width=True):
    if not selected_dirs:
        st.warning("Выберите хотя бы один DIR.")
        st.stop()

    with st.spinner("Читаю review-логи и считаю статистику…"):
        payload = build_payload(selected_dirs, REVIEWS_REPOS)

    overall = build_overall_totals(payload["dir_summary"])
    c1, c2, c3 = st.columns(3)
    c1.metric("Оценено публикаций", overall["evaluated_publications"])
    c2.metric("В дальнейшую работу (1+2)", overall["positive_count"])
    c3.metric("Событий в истории", payload["meta"]["review_events_attached"])

    st.subheader("Предпросмотр по DIR")
    st.dataframe(build_dir_preview_df(payload["dir_summary"]), use_container_width=True, hide_index=True)

    st.subheader("Результаты оценивания")
    st.dataframe(build_results_df(payload["items"]), use_container_width=True, hide_index=True)


    json_text = json.dumps(
        {
            "meta": payload["meta"],
            "dir_summary": payload["dir_summary"],
            "items": payload["items"],
        },
        ensure_ascii=False,
        indent=2,
    )
    st.download_button(
        "Скачать export.json",
        data=json_text.encode("utf-8"),
        file_name="scoring_results_export.json",
        mime="application/json",
        use_container_width=True,
    )